feat: IPRPC over IBC

app/app.go

@@ -284,6 +284,7 @@ var (
 		string(rewardsmoduletypes.ProvidersRewardsAllocationPool):        {authtypes.Minter, authtypes.Staking},
 		dualstakingmoduletypes.ModuleName:                                {authtypes.Burner, authtypes.Staking},
 		string(rewardsmoduletypes.IprpcPoolName):                         nil,
+		string(rewardsmoduletypes.PendingIprpcPoolName):                  nil,
 		// this line is used by starport scaffolding # stargate/app/maccPerms
 	}
 )
@@ -560,6 +561,7 @@ func New(
 		authtypes.FeeCollectorName,
 		app.TimerStoreKeeper,
 		authtypes.NewModuleAddress(govtypes.ModuleName).String(),
+		app.IBCKeeper.ChannelKeeper,
 	)
 	rewardsModule := rewardsmodule.NewAppModule(appCodec, app.RewardsKeeper, app.AccountKeeper, app.BankKeeper)
 
@@ -715,6 +717,7 @@ func New(
 		packetforwardkeeper.DefaultForwardTransferPacketTimeoutTimestamp, // forward timeout
 		packetforwardkeeper.DefaultRefundTransferPacketTimeoutTimestamp,  // refund timeout
 	)
+	transferStack = rewardsmodule.NewIBCMiddleware(transferStack, app.RewardsKeeper)
 	ibcRouter.AddRoute(ibctransfertypes.ModuleName, transferStack)
 
 	// this line is used by starport scaffolding # ibc/app/router

proto/lavanet/lava/rewards/iprpc.proto

@@ -17,4 +17,10 @@ message Specfund {
         (gogoproto.castrepeated) = "github.com/cosmos/cosmos-sdk/types.Coins",
         (gogoproto.nullable)     = false
     ];
+}
+
+message IprpcMemo {
+    string creator = 1;
+    string spec = 2;
+    uint64 duration = 3; // Iprpc fund period in months
 }

testutil/keeper/keepers_init.go

@@ -264,7 +264,7 @@ func InitAllKeepers(t testing.TB) (*Servers, *Keepers, context.Context) {
 	ks.Projects = *projectskeeper.NewKeeper(cdc, projectsStoreKey, projectsMemStoreKey, projectsparamsSubspace, ks.Epochstorage, ks.FixationStoreKeeper)
 	ks.Protocol = *protocolkeeper.NewKeeper(cdc, protocolStoreKey, protocolMemStoreKey, protocolparamsSubspace, authtypes.NewModuleAddress(govtypes.ModuleName).String())
 	ks.Downtime = downtimekeeper.NewKeeper(cdc, downtimeKey, downtimeParamsSubspace, ks.Epochstorage)
-	ks.Rewards = *rewardskeeper.NewKeeper(cdc, rewardsStoreKey, rewardsMemStoreKey, rewardsparamsSubspace, ks.BankKeeper, ks.AccountKeeper, ks.Spec, ks.Epochstorage, ks.Downtime, ks.StakingKeeper, ks.Dualstaking, ks.Distribution, authtypes.FeeCollectorName, ks.TimerStoreKeeper, authtypes.NewModuleAddress(govtypes.ModuleName).String())
+	ks.Rewards = *rewardskeeper.NewKeeper(cdc, rewardsStoreKey, rewardsMemStoreKey, rewardsparamsSubspace, ks.BankKeeper, ks.AccountKeeper, ks.Spec, ks.Epochstorage, ks.Downtime, ks.StakingKeeper, ks.Dualstaking, ks.Distribution, authtypes.FeeCollectorName, ks.TimerStoreKeeper, authtypes.NewModuleAddress(govtypes.ModuleName).String(), nil)
 	ks.Subscription = *subscriptionkeeper.NewKeeper(cdc, subscriptionStoreKey, subscriptionMemStoreKey, subscriptionparamsSubspace, &ks.BankKeeper, &ks.AccountKeeper, &ks.Epochstorage, ks.Projects, ks.Plans, ks.Dualstaking, ks.Rewards, ks.FixationStoreKeeper, ks.TimerStoreKeeper, ks.StakingKeeper)
 	ks.Pairing = *pairingkeeper.NewKeeper(cdc, pairingStoreKey, pairingMemStoreKey, pairingparamsSubspace, &ks.BankKeeper, &ks.AccountKeeper, ks.Spec, &ks.Epochstorage, ks.Projects, ks.Subscription, ks.Plans, ks.Downtime, ks.Dualstaking, &ks.StakingKeeper, ks.FixationStoreKeeper, ks.TimerStoreKeeper)
 	ks.ParamsKeeper = paramsKeeper

testutil/keeper/rewards.go

@@ -94,6 +94,7 @@ func RewardsKeeper(t testing.TB) (*keeper.Keeper, sdk.Context) {
 		authtypes.FeeCollectorName,
 		timerstorekeeper.NewKeeper(cdc),
 		authtypes.NewModuleAddress(govtypes.ModuleName).String(),
+		nil,
 	)
 
 	// Initialize params

testutil/sample/sample.go

@@ -13,6 +13,13 @@ func AccAddress() string {
 	return sdk.AccAddress(addr).String()
 }
 
+// AccAddress returns a sample account address of type sdk.AccAddress
+func AccAddressObject() sdk.AccAddress {
+	pk := ed25519.GenPrivKey().PubKey()
+	addr := pk.Address()
+	return sdk.AccAddress(addr)
+}
+
 // ValAddress returns a sample validator account address
 func ValAddress() string {
 	pk := ed25519.GenPrivKey().PubKey()

x/rewards/ibc_middleware.go

@@ -0,0 +1,236 @@
+package rewards
+
+import (
+	"errors"
+	"fmt"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	capabilitytypes "github.com/cosmos/cosmos-sdk/x/capability/types"
+	transfertypes "github.com/cosmos/ibc-go/v7/modules/apps/transfer/types"
+	clienttypes "github.com/cosmos/ibc-go/v7/modules/core/02-client/types"
+	channeltypes "github.com/cosmos/ibc-go/v7/modules/core/04-channel/types"
+	porttypes "github.com/cosmos/ibc-go/v7/modules/core/05-port/types"
+	"github.com/cosmos/ibc-go/v7/modules/core/exported"
+	"github.com/lavanet/lava/utils"
+	"github.com/lavanet/lava/x/rewards/keeper"
+	"github.com/lavanet/lava/x/rewards/types"
+)
+
+var _ porttypes.Middleware = &IBCMiddleware{}
+
+// IBCMiddleware implements the ICS26 callbacks for the transfer middleware given
+// the rewards keeper and the underlying application.
+type IBCMiddleware struct {
+	app    porttypes.IBCModule // transfer stack
+	keeper keeper.Keeper
+}
+
+// NewIBCMiddleware creates a new IBCMiddleware given the keeper and underlying application
+func NewIBCMiddleware(app porttypes.IBCModule, k keeper.Keeper) IBCMiddleware {
+	return IBCMiddleware{
+		app:    app,
+		keeper: k,
+	}
+}
+
+// IBCModule interface implementation. Only OnRecvPacket() calls a callback from the keeper. The rest have default implementations
+
+func (im IBCMiddleware) OnChanOpenInit(ctx sdk.Context, order channeltypes.Order, connectionHops []string, portID string,
+	channelID string,
+	chanCap *capabilitytypes.Capability,
+	counterparty channeltypes.Counterparty,
+	version string,
+) (string, error) {
+	return im.app.OnChanOpenInit(ctx, order, connectionHops, portID, channelID, chanCap, counterparty, version)
+}
+
+func (im IBCMiddleware) OnChanOpenTry(
+	ctx sdk.Context,
+	order channeltypes.Order,
+	connectionHops []string,
+	portID, channelID string,
+	chanCap *capabilitytypes.Capability,
+	counterparty channeltypes.Counterparty,
+	counterpartyVersion string,
+) (version string, err error) {
+	return im.app.OnChanOpenTry(ctx, order, connectionHops, portID, channelID, chanCap, counterparty, counterpartyVersion)
+}
+
+func (im IBCMiddleware) OnChanOpenAck(
+	ctx sdk.Context,
+	portID, channelID string,
+	counterpartyChannelID string,
+	counterpartyVersion string,
+) error {
+	return im.app.OnChanOpenAck(ctx, portID, channelID, counterpartyChannelID, counterpartyVersion)
+}
+
+func (im IBCMiddleware) OnChanOpenConfirm(ctx sdk.Context, portID, channelID string) error {
+	return im.app.OnChanOpenConfirm(ctx, portID, channelID)
+}
+
+func (im IBCMiddleware) OnChanCloseInit(ctx sdk.Context, portID, channelID string) error {
+	return im.app.OnChanCloseInit(ctx, portID, channelID)
+}
+
+func (im IBCMiddleware) OnChanCloseConfirm(ctx sdk.Context, portID, channelID string) error {
+	return im.app.OnChanCloseConfirm(ctx, portID, channelID)
+}
+
+// OnRecvPacket checks the packet's memo and funds the IPRPC pool accordingly. If the memo is not the expected JSON,
+// the packet is transferred normally to the next IBC module in the transfer stack
+func (im IBCMiddleware) OnRecvPacket(ctx sdk.Context, packet channeltypes.Packet, relayer sdk.AccAddress) exported.Acknowledgement {
+	// unmarshal the packet's data with the transfer module codec (expect an ibc-transfer packet)
+	var data transfertypes.FungibleTokenPacketData
+	if err := transfertypes.ModuleCdc.UnmarshalJSON(packet.GetData(), &data); err != nil {
+		return channeltypes.NewErrorAcknowledgement(err)
+	}
+
+	// sanity check: validate data
+	if err := data.ValidateBasic(); err != nil {
+		detaliedErr := utils.LavaFormatError("rewards module IBC middleware processing failed", err,
+			utils.LogAttr("data", data),
+		)
+		return channeltypes.NewErrorAcknowledgement(detaliedErr)
+	}
+
+	// extract the packet's memo
+	memo, err := im.keeper.ExtractIprpcMemoFromPacket(ctx, data)
+	if errors.Is(err, types.ErrMemoNotIprpcOverIbc) {
+		// not a packet that should be handled as IPRPC over IBC (not considered as error)
+		utils.LavaFormatDebug("rewards module IBC middleware processing skipped, memo is invalid for IPRPC over IBC funding",
+			utils.LogAttr("memo", memo),
+		)
+		return im.app.OnRecvPacket(ctx, packet, relayer)
+	} else if errors.Is(err, types.ErrIprpcMemoInvalid) {
+		// memo is in the right format of IPRPC over IBC but the data is invalid
+		utils.LavaFormatWarning("rewards module IBC middleware processing failed, memo data is invalid", err,
+			utils.LogAttr("memo", memo))
+		return channeltypes.NewErrorAcknowledgement(err)
+	}
+
+	// send the IBC transfer to get the tokens
+	ack := im.sendIbcTransfer(ctx, packet, relayer, data.Sender, types.IbcIprpcReceiverAddress().String())
+	if ack == nil || !ack.Success() {
+		// we check for ack == nil because it means that IBC transfer module did not return an acknowledgement.
+		// This isn't necessarily an error, but it could indicate unexpected behavior or asynchronous processing
+		// on the IBC transfer module's side (which returns a non-nil ack when executed without errors). Asynchronous
+		// processing can be queued processing of packets, interacting with external APIs and more. These can cause
+		// delays in the IBC-transfer's processing which will make the module return a nil ack until the processing is done.
+		// Asyncronous acks are handled in the WriteAcknowledgement middleware method.
+		return ack
+	}
+
+	// get the IBC tokens that were transferred to the IbcIprpcReceiverAddress
+	amount, ok := sdk.NewIntFromString(data.Amount)
+	if !ok {
+		detailedErr := utils.LavaFormatError("rewards module IBC middleware processing failed", fmt.Errorf("invalid amount in ibc-transfer data"),
+			utils.LogAttr("data", data),
+		)
+		return channeltypes.NewErrorAcknowledgement(detailedErr)
+	}
+	ibcTokens := transfertypes.GetTransferCoin(packet.DestinationPort, packet.DestinationChannel, data.Denom, amount)
+
+	// send the IBC tokens from IbcIprpcReceiver to PendingIprpcPool
+	err = im.keeper.SendIbcIprpcReceiverTokensToPendingIprpcPool(ctx, ibcTokens)
+	if err != nil {
+		detailedErr := utils.LavaFormatError("sending token from IbcIprpcReceiver to the PendingIprpcPool failed", err,
+			utils.LogAttr("sender", data.Sender),
+		)
+		return channeltypes.NewErrorAcknowledgement(detailedErr)
+	}
+
+	// TODO: define a PendingIbcIprpcFund store to keep track on pending requests. Also create an event that
+	// a new pending request has been created
+
+	return channeltypes.NewResultAcknowledgement([]byte{byte(1)})
+}
+
+func (im IBCMiddleware) OnAcknowledgementPacket(
+	ctx sdk.Context,
+	packet channeltypes.Packet,
+	acknowledgement []byte,
+	relayer sdk.AccAddress,
+) error {
+	return im.app.OnAcknowledgementPacket(ctx, packet, acknowledgement, relayer)
+}
+
+func (im IBCMiddleware) OnTimeoutPacket(
+	ctx sdk.Context,
+	packet channeltypes.Packet,
+	relayer sdk.AccAddress,
+) error {
+	return im.app.OnTimeoutPacket(ctx, packet, relayer)
+}
+
+// ICS4Wrapper interface (default implementations)
+func (im IBCMiddleware) SendPacket(ctx sdk.Context, chanCap *capabilitytypes.Capability, sourcePort string, sourceChannel string,
+	timeoutHeight clienttypes.Height, timeoutTimestamp uint64, data []byte,
+) (sequence uint64, err error) {
+	return im.keeper.SendPacket(ctx, chanCap, sourcePort, sourceChannel, timeoutHeight, timeoutTimestamp, data)
+}
+
+// WriteAcknowledgement is called when handling async acks. Since the OnRecvPacket code returns on a nil ack (which indicates
+// that an async ack will occur), funds can stay stuck in the IbcIprpcReceiver account (which is a temp account that should
+// not hold funds). This code simply does the missing functionaliy that OnRecvPacket would do if the ack was not nil.
+func (im IBCMiddleware) WriteAcknowledgement(ctx sdk.Context, chanCap *capabilitytypes.Capability, packet exported.PacketI,
+	ack exported.Acknowledgement,
+) error {
+	err := im.keeper.WriteAcknowledgement(ctx, chanCap, packet, ack)
+	if err != nil {
+		return err
+	}
+
+	// unmarshal the packet's data with the transfer module codec (expect an ibc-transfer packet)
+	var data transfertypes.FungibleTokenPacketData
+	if err := transfertypes.ModuleCdc.UnmarshalJSON(packet.GetData(), &data); err != nil {
+		return err
+	}
+
+	// sanity check: validate data
+	if err := data.ValidateBasic(); err != nil {
+		return utils.LavaFormatError("handling async transfer packet failed", err,
+			utils.LogAttr("data", data),
+		)
+	}
+
+	// get the IBC tokens that were transferred to the IbcIprpcReceiverAddress
+	amount, ok := sdk.NewIntFromString(data.Amount)
+	if !ok {
+		return utils.LavaFormatError("handling async transfer packet failed", fmt.Errorf("invalid amount in ibc-transfer data"),
+			utils.LogAttr("data", data),
+		)
+	}
+	ibcTokens := transfertypes.GetTransferCoin(packet.GetDestPort(), packet.GetDestChannel(), data.Denom, amount)
+
+	// send the tokens from the IbcIprpcReceiver to the PendingIprpcPool
+	return im.keeper.SendIbcIprpcReceiverTokensToPendingIprpcPool(ctx, ibcTokens)
+}
+
+func (im IBCMiddleware) GetAppVersion(ctx sdk.Context, portID, channelID string) (string, bool) {
+	return im.keeper.GetAppVersion(ctx, portID, channelID)
+}
+
+// sendIbcTransfer sends the ibc-transfer packet to the IBC Transfer module
+func (im IBCMiddleware) sendIbcTransfer(ctx sdk.Context, packet channeltypes.Packet, relayer sdk.AccAddress, sender string, receiver string) exported.Acknowledgement {
+	// unmarshal the packet's data with the transfer module codec (expect an ibc-transfer packet)
+	var data transfertypes.FungibleTokenPacketData
+	if err := transfertypes.ModuleCdc.UnmarshalJSON(packet.GetData(), &data); err != nil {
+		return channeltypes.NewErrorAcknowledgement(err)
+	}
+
+	// change the ibc-transfer packet data with sender and receiver
+	data.Sender = sender
+	data.Receiver = receiver
+	data.Memo = ""
+	marshelledData, err := transfertypes.ModuleCdc.MarshalJSON(&data)
+	if err != nil {
+		utils.LavaFormatError("rewards module IBC middleware processing failed, cannot marshal packet data", err,
+			utils.LogAttr("data", data))
+		return channeltypes.NewErrorAcknowledgement(err)
+	}
+	packet.Data = marshelledData
+
+	// call the next OnRecvPacket() of the transfer stack to make the IBC Transfer module's OnRecvPacket send the IBC tokens
+	return im.app.OnRecvPacket(ctx, packet, relayer)
+}

x/rewards/keeper/grpc_query_pools.go

@@ -37,6 +37,10 @@ func (k Keeper) Pools(goCtx context.Context, req *types.QueryPoolsRequest) (*typ
 			Name:    string(types.IprpcPoolName),
 			Balance: k.TotalPoolTokens(ctx, types.IprpcPoolName),
 		},
+		{
+			Name:    string(types.PendingIprpcPoolName),
+			Balance: k.TotalPoolTokens(ctx, types.PendingIprpcPoolName),
+		},
 	}
 
 	estimatedBlocksToRefill := k.BlocksToNextTimerExpiry(ctx)

x/rewards/keeper/helpers_test.go

@@ -9,8 +9,10 @@ import (
 	authtypes "github.com/cosmos/cosmos-sdk/x/auth/types"
 	govtypes "github.com/cosmos/cosmos-sdk/x/gov/types"
 	stakingtypes "github.com/cosmos/cosmos-sdk/x/staking/types"
+	transfertypes "github.com/cosmos/ibc-go/v7/modules/apps/transfer/types"
 	"github.com/lavanet/lava/testutil/common"
 	testkeeper "github.com/lavanet/lava/testutil/keeper"
+	"github.com/lavanet/lava/testutil/sample"
 	commontypes "github.com/lavanet/lava/utils/common/types"
 	"github.com/lavanet/lava/utils/sigs"
 	planstypes "github.com/lavanet/lava/x/plans/types"
@@ -140,6 +142,10 @@ func (ts *tester) setupForIprpcTests(fundIprpcPool bool) {
 	}
 }
 
+func (ts *tester) createIbcTransferPacketData(memo string) transfertypes.FungibleTokenPacketData {
+	return transfertypes.NewFungibleTokenPacketData(ts.TokenDenom(), "100000", sample.AccAddress(), sample.AccAddress(), memo)
+}
+
 // getConsumersForIprpcSubTest is a helper function specifically for the TestIprpcEligibleSubscriptions unit test
 // this function returns two consumer addresses to test depending on the input mode:
 //