feat(x/protocolpool): Stream funds from community pool

proto/cosmos/protocolpool/v1/budget.proto

@@ -0,0 +1,46 @@
+// Since: cosmos-sdk 0.50
+syntax = "proto3";
+package cosmos.protocolpool.v1;
+
+option go_package = "cosmossdk.io/x/protocolpool/types";
+
+import "gogoproto/gogo.proto";
+import "cosmos/base/v1beta1/coin.proto";
+import "cosmos_proto/cosmos.proto";
+import "cosmos/msg/v1/msg.proto";
+
+// Budget defines budget proposal type.
+//
+// Since: cosmos-sdk 0.50
+message BudgetProposal {
+  option (cosmos.msg.v1.signer)               = "recipient_address";
+  string                   recipient_address  = 1 [(cosmos_proto.scalar) = "cosmos.AddressString"];
+  cosmos.base.v1beta1.Coin total_budget       = 2;
+  int64                    start_time         = 3;
+  int64                    remaining_tranches = 4;
+  int64                    period             = 5;
+}
+
+// MsgBudgetProposalResponse defines the response to executing a
+// MsgBudgetProposal message.
+//
+// Since: cosmos-sdk 0.50
+message MsgBudgetProposalResponse {}
+
+// MsgClaimBudget defines a message for claiming the distributed budget.
+//
+// Since: cosmos-sdk 0.50
+message MsgClaimBudget {
+  option (cosmos.msg.v1.signer) = "recipient_address";
+  string recipient_address      = 1 [(cosmos_proto.scalar) = "cosmos.AddressString"];
+}
+
+// MsgClaimBudgetResponse defines the response to executing a
+// MsgClaimBudget message.
+//
+// Since: cosmos-sdk 0.50
+message MsgClaimBudgetResponse {
+  cosmos.base.v1beta1.Coin amount = 1
+      [(gogoproto.nullable) = false, (gogoproto.castrepeated) = "github.com/cosmos/cosmos-sdk/types.Coins"];
+  ;
+}

simapp/app.go

@@ -437,6 +437,7 @@ func NewSimApp(
 		genutiltypes.ModuleName,
 		feegrant.ModuleName,
 		group.ModuleName,
+		pooltypes.ModuleName,
 	)
 
 	// NOTE: The genutils module must occur after staking so that pools are

simapp/app_config.go

@@ -128,6 +128,7 @@ var (
 						stakingtypes.ModuleName,
 						feegrant.ModuleName,
 						group.ModuleName,
+						pooltypes.ModuleName,
 					},
 					OverrideStoreKeys: []*runtimev1alpha1.StoreKeyConfig{
 						{

testutil/configurator/configurator.go

@@ -81,6 +81,7 @@ func defaultConfig() *Config {
 			"upgrade",
 			"vesting",
 			"circuit",
+			"protocolpool",
 		},
 		InitGenesisOrder: []string{
 			"auth",

x/gov/client/cli/prompt.go

@@ -44,6 +44,10 @@ var suggestedProposalTypes = []proposalType{
 		Name:    "cancel-software-upgrade",
 		MsgType: "/cosmos.upgrade.v1beta1.MsgCancelUpgrade",
 	},
+	{
+		Name:    "set-budget",
+		MsgType: "/cosmos.protocolpool.v1.BudgetProposal",
+	},
 	{
 		Name:    proposalOther,
 		MsgType: "", // user will input the message type

x/protocolpool/abci.go

@@ -0,0 +1,73 @@
+package protocolpool
+
+import (
+	"fmt"
+
+	"cosmossdk.io/x/protocolpool/keeper"
+	"cosmossdk.io/x/protocolpool/types"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+)
+
+func EndBlocker(ctx sdk.Context, k keeper.Keeper, authKeeper types.AccountKeeper) error {
+	err := k.BudgetProposal.Walk(ctx, nil, func(key sdk.AccAddress, budget types.BudgetProposal) (stop bool, err error) {
+		// check if the distribution is completed
+		if budget.RemainingTranches <= 0 {
+			// Log the end of the budget
+			k.Logger(ctx).Info(fmt.Sprintf("Budget ended for recipient: %s", key.String()))
+			return false, nil // Continue iterating
+		}
+
+		currentTime := ctx.BlockTime().Unix()
+
+		// Check if the start time is reached
+		if currentTime < budget.StartTime {
+			return false, fmt.Errorf("distribution has not started yet")
+		}
+
+		// Calculate the number of blocks elapsed since the start time
+		blocksElapsed := ctx.BlockHeight() - budget.Period
+
+		// Check if its time to distribute funds based on period intervals
+		if blocksElapsed > 0 && blocksElapsed%budget.Period == 0 {
+			// Calculate the amount to distribute in each tranche
+			amountPerTranche := budget.TotalBudget.Amount.QuoRaw(budget.RemainingTranches)
+			amount := sdk.NewCoin(budget.TotalBudget.Denom, amountPerTranche)
+
+			recipient, err := authKeeper.AddressCodec().StringToBytes(budget.RecipientAddress)
+			if err != nil {
+				return false, err
+			}
+
+			distributionInfo := types.DistributionInfo{
+				Address: recipient,
+				Amount:  amount,
+			}
+
+			// store funds in the DistributionInfo until a claim tx
+			err = k.AppendDistributionInfo(ctx, distributionInfo)
+			if err != nil {
+				return false, err
+			}
+
+			// update the budget's remaining tranches
+			budget.RemainingTranches--
+
+			// update the TotalBudget amount
+			budget.TotalBudget.Amount.Sub(amountPerTranche)
+
+			k.Logger(ctx).Info(fmt.Sprintf("Processing budget for recipient: %s. Amount: %s", budget.RecipientAddress, amountPerTranche.String()))
+
+			// Save the updated budget in the state
+			err = k.BudgetProposal.Set(ctx, recipient, budget)
+			if err != nil {
+				k.Logger(ctx).Error(fmt.Sprintf("Error while updating the budget for recipient %s", budget.RecipientAddress))
+				return false, err
+			}
+
+		}
+		return false, nil
+	})
+
+	return err
+}

x/protocolpool/go.mod

@@ -4,6 +4,7 @@ go 1.21
 
 require (
 	cosmossdk.io/api v0.7.2-0.20230927090904-9dd34510e273
+	cosmossdk.io/collections v0.4.0
 	cosmossdk.io/core v0.12.0
 	cosmossdk.io/depinject v1.0.0-alpha.4
 	cosmossdk.io/errors v1.0.0
@@ -22,7 +23,6 @@ require (
 )
 
 require (
-	cosmossdk.io/collections v0.4.0 // indirect
 	cosmossdk.io/math v1.1.3-rc.1 // indirect
 	cosmossdk.io/store v1.0.0-rc.0 // indirect
 	cosmossdk.io/x/tx v0.10.0 // indirect

x/protocolpool/keeper/keeper.go

@@ -2,8 +2,10 @@ package keeper
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
+	"cosmossdk.io/collections"
 	storetypes "cosmossdk.io/core/store"
 	errorsmod "cosmossdk.io/errors"
 	"cosmossdk.io/log"
@@ -19,7 +21,14 @@ type Keeper struct {
 	authKeeper   types.AccountKeeper
 	bankKeeper   types.BankKeeper
 
+	cdc codec.BinaryCodec
+
 	authority string
+
+	// State
+	Schema         collections.Schema
+	BudgetProposal collections.Map[sdk.AccAddress, types.BudgetProposal]
+	DistrInfo      collections.Map[sdk.AccAddress, sdk.Coin]
 }
 
 func NewKeeper(cdc codec.BinaryCodec, storeService storetypes.KVStoreService,
@@ -29,12 +38,25 @@ func NewKeeper(cdc codec.BinaryCodec, storeService storetypes.KVStoreService,
 	if addr := ak.GetModuleAddress(types.ModuleName); addr == nil {
 		panic(fmt.Sprintf("%s module account has not been set", types.ModuleName))
 	}
-	return Keeper{
-		storeService: storeService,
-		authKeeper:   ak,
-		bankKeeper:   bk,
-		authority:    authority,
+	sb := collections.NewSchemaBuilder(storeService)
+
+	keeper := Keeper{
+		storeService:   storeService,
+		authKeeper:     ak,
+		bankKeeper:     bk,
+		cdc:            cdc,
+		authority:      authority,
+		BudgetProposal: collections.NewMap(sb, types.BudgetKey, "budget", sdk.AccAddressKey, codec.CollValue[types.BudgetProposal](cdc)),
+		DistrInfo:      collections.NewMap(sb, types.DistrInfoKey, "distribution_info", sdk.AccAddressKey, codec.CollValue[sdk.Coin](cdc)),
+	}
+
+	schema, err := sb.Build()
+	if err != nil {
+		panic(err)
 	}
+	keeper.Schema = schema
+
+	return keeper
 }
 
 // GetAuthority returns the x/protocolpool module's authority.
@@ -67,3 +89,81 @@ func (k Keeper) GetCommunityPool(ctx context.Context) (sdk.Coins, error) {
 	}
 	return k.bankKeeper.GetAllBalances(ctx, moduleAccount.GetAddress()), nil
 }
+
+func (k Keeper) AppendDistributionInfo(ctx sdk.Context, distributionInfo types.DistributionInfo) error {
+	amount, err := k.DistrInfo.Get(ctx, distributionInfo.Address)
+	if err != nil {
+		if errors.Is(err, collections.ErrNotFound) {
+			err := k.DistrInfo.Set(ctx, distributionInfo.Address, distributionInfo.Amount)
+			if err != nil {
+				return err
+			}
+		} else {
+			return err
+		}
+	}
+	updatedAmount := distributionInfo.Amount.Add(amount)
+	return k.DistrInfo.Set(ctx, distributionInfo.Address, updatedAmount)
+}
+
+func (k Keeper) ClaimFunds(ctx context.Context, recipient sdk.AccAddress) (amount sdk.Coin, err error) {
+	// get claimable funds from distribution info
+	amount, err = k.getClaimableFunds(ctx, recipient)
+	if err != nil {
+		return sdk.Coin{}, err
+	}
+
+	// distribute amount from feepool
+	err = k.DistributeFromFeePool(ctx, sdk.NewCoins(amount), recipient)
+	if err != nil {
+		return sdk.Coin{}, err
+	}
+
+	// remove the recipient from the DistributionInfo
+	err = k.DistrInfo.Remove(ctx, recipient)
+	if err != nil {
+		return sdk.Coin{}, err
+	}
+
+	return amount, nil
+}
+
+func (k Keeper) getClaimableFunds(ctx context.Context, recipient sdk.AccAddress) (amount sdk.Coin, err error) {
+	amount, err = k.DistrInfo.Get(ctx, recipient)
+	if err != nil {
+		if errors.Is(err, collections.ErrNotFound) {
+			return sdk.Coin{}, fmt.Errorf("no claimable funds are present for recipient: %s", recipient.String())
+		}
+		return sdk.Coin{}, err
+	}
+	return amount, nil
+}
+
+func (k Keeper) validateBudgetProposal(ctx context.Context, bp types.BudgetProposal) error {
+	account := k.authKeeper.GetAccount(ctx, sdk.AccAddress(bp.RecipientAddress))
+	if account == nil {
+		return fmt.Errorf("account not found: %s", bp.RecipientAddress)
+	}
+
+	if bp.TotalBudget.IsZero() {
+		return fmt.Errorf("total budget cannot be zero")
+	}
+
+	if err := validateAmount(sdk.NewCoins(*bp.TotalBudget)); err != nil {
+		return err
+	}
+
+	if bp.StartTime <= 0 {
+		return fmt.Errorf("start time must be a positive value")
+	}
+
+	if bp.RemainingTranches <= 0 {
+		return fmt.Errorf("remaining tranches must be a positive value")
+	}
+
+	if bp.Period <= 0 {
+		return fmt.Errorf("period should be a positive value")
+	}
+
+	return nil
+}

x/protocolpool/keeper/msg_server.go

@@ -22,6 +22,38 @@ func NewMsgServerImpl(keeper Keeper) types.MsgServer {
 	return &MsgServer{Keeper: keeper}
 }
 
+func (k MsgServer) MsgClaimBudget(ctx context.Context, msg *types.MsgClaimBudget) (*types.MsgClaimBudgetResponse, error) {
+	recipient, err := k.Keeper.authKeeper.AddressCodec().StringToBytes(msg.RecipientAddress)
+	if err != nil {
+		return nil, sdkerrors.ErrInvalidAddress.Wrapf("invalid recipient address: %s", err)
+	}
+
+	amount, err := k.ClaimFunds(ctx, recipient)
+	if err != nil {
+		return nil, err
+	}
+
+	return &types.MsgClaimBudgetResponse{Amount: amount}, nil
+}
+
+func (k MsgServer) SubmitBudgetProposal(ctx context.Context, msg *types.BudgetProposal) (*types.MsgBudgetProposalResponse, error) {
+	recipient, err := k.Keeper.authKeeper.AddressCodec().StringToBytes(msg.RecipientAddress)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := k.validateBudgetProposal(ctx, *msg); err != nil {
+		return nil, err
+	}
+
+	// set budget proposal in state
+	err = k.BudgetProposal.Set(ctx, recipient, *msg)
+	if err != nil {
+		return nil, err
+	}
+	return &types.MsgBudgetProposalResponse{}, nil
+}
+
 func (k MsgServer) FundCommunityPool(ctx context.Context, msg *types.MsgFundCommunityPool) (*types.MsgFundCommunityPoolResponse, error) {
 	depositor, err := k.authKeeper.AddressCodec().StringToBytes(msg.Depositor)
 	if err != nil {

x/protocolpool/module.go

@@ -16,6 +16,7 @@ import (
 	"github.com/cosmos/cosmos-sdk/client"
 	"github.com/cosmos/cosmos-sdk/codec"
 	codectypes "github.com/cosmos/cosmos-sdk/codec/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
 	"github.com/cosmos/cosmos-sdk/types/module"
 	simtypes "github.com/cosmos/cosmos-sdk/types/simulation"
 	authtypes "github.com/cosmos/cosmos-sdk/x/auth/types"
@@ -29,6 +30,7 @@ var (
 
 	_ module.AppModule           = AppModule{}
 	_ module.AppModuleSimulation = AppModule{}
+	_ appmodule.HasEndBlocker    = AppModule{}
 )
 
 // AppModuleBasic defines the basic application module used by the pool module.
@@ -92,6 +94,12 @@ func NewAppModule(cdc codec.Codec, keeper keeper.Keeper,
 // ConsensusVersion implements AppModule/ConsensusVersion.
 func (AppModule) ConsensusVersion() uint64 { return ConsensusVersion }
 
+// EndBlock implements the protocolpool module's EndBlock.
+func (am AppModule) EndBlock(ctx context.Context) error {
+	c := sdk.UnwrapSDKContext(ctx)
+	return EndBlocker(c, am.keeper, am.accountKeeper)
+}
+
 //
 // App Wiring Setup
 //