Index meteora Locker VestingEscrows

ddl/migrations/0177_sol_locker_vesting_escrows.sql

@@ -0,0 +1,26 @@
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS sol_locker_vesting_escrows (
+    account TEXT PRIMARY KEY,
+    slot BIGINT NOT NULL,
+    recipient TEXT NOT NULL,
+    token_mint TEXT NOT NULL,
+    creator TEXT NOT NULL,
+    base TEXT NOT NULL,
+    escrow_bump SMALLINT NOT NULL,
+    update_recipient_mode SMALLINT NOT NULL,
+    cancel_mode SMALLINT NOT NULL,
+    token_program_flag SMALLINT NOT NULL,
+    cliff_time BIGINT NOT NULL,
+    frequency BIGINT NOT NULL,
+    cliff_unlock_amount BIGINT NOT NULL,
+    amount_per_period BIGINT NOT NULL,
+    number_of_period BIGINT NOT NULL,
+    total_claimed_amount BIGINT NOT NULL,
+    vesting_start_time BIGINT NOT NULL,
+    cancelled_at BIGINT NOT NULL,
+    created_at TIMESTAMP DEFAULT NOW(),
+    updated_at TIMESTAMP DEFAULT NOW()
+);
+
+COMMIT;

solana/indexer/dbc/indexer.go

@@ -12,6 +12,7 @@ import (
 	"api.audius.co/database"
 	"api.audius.co/solana/indexer/common"
 	"api.audius.co/solana/spl/programs/meteora_dbc"
+	"api.audius.co/solana/spl/programs/meteora_locker"
 	bin "github.com/gagliardetto/binary"
 	"github.com/gagliardetto/solana-go"
 	"github.com/gagliardetto/solana-go/rpc"
@@ -229,6 +230,25 @@ func (d *Indexer) HandleUpdate(ctx context.Context, msg *pb.SubscribeUpdate) err
 				err = d.processTransaction(ctx, txRes.Slot, tx)
 			}
 		}
+
+		// Handle Vesting Escrow updates
+		if len(accountUpdate.Account.Data) > 8 && bytes.Equal(accountUpdate.Account.Data[:8], meteora_locker.Account_VestingEscrow[:]) {
+			var escrow meteora_locker.VestingEscrow
+			err := bin.NewBorshDecoder(accountUpdate.Account.Data).Decode(&escrow)
+			if err != nil {
+				return fmt.Errorf("failed to decode Vesting Escrow account: %w", err)
+			}
+			account := solana.PublicKeyFromBytes(accountUpdate.Account.Pubkey)
+
+			err = processVestingEscrowUpdate(ctx, d.pool, accountUpdate.Slot, account, &escrow)
+			if err != nil {
+				return fmt.Errorf("failed to process Vesting Escrow update: %w", err)
+			}
+			d.logger.Debug("processed Vesting Escrow update",
+				zap.String("account", account.String()),
+				zap.String("recipient", escrow.Recipient.String()),
+			)
+		}
 	}
 	return nil
 }
@@ -320,6 +340,34 @@ func (d *Indexer) makeSubscriptionRequest(ctx context.Context, mints []string) *
 			},
 		}
 		subscription.Accounts[mint] = &poolFilter
+
+		lockFilter := pb.SubscribeRequestFilterAccounts{
+			Owner: []string{meteora_locker.ProgramID.String()},
+			Filters: []*pb.SubscribeRequestFilterAccountsFilter{
+				{
+					Filter: &pb.SubscribeRequestFilterAccountsFilter_Memcmp{
+						Memcmp: &pb.SubscribeRequestFilterAccountsFilterMemcmp{
+							Offset: 0,
+							Data: &pb.SubscribeRequestFilterAccountsFilterMemcmp_Bytes{
+								Bytes: meteora_locker.Account_VestingEscrow[:],
+							},
+						},
+					},
+				},
+				{
+					Filter: &pb.SubscribeRequestFilterAccountsFilter_Memcmp{
+						Memcmp: &pb.SubscribeRequestFilterAccountsFilterMemcmp{
+							// Pool mint is after discriminator and recipient
+							Offset: 8 + 32,
+							Data: &pb.SubscribeRequestFilterAccountsFilterMemcmp_Base58{
+								Base58: mint,
+							},
+						},
+					},
+				},
+			},
+		}
+		subscription.Accounts["lock_"+mint] = &lockFilter
 	}
 
 	configFilter := pb.SubscribeRequestFilterAccounts{
@@ -430,6 +478,30 @@ func processDbcConfigUpdate(
 	return nil
 }
 
+func processVestingEscrowUpdate(
+	ctx context.Context,
+	db database.DbPool,
+	slot uint64,
+	account solana.PublicKey,
+	escrow *meteora_locker.VestingEscrow,
+) error {
+	sqlTx, err := db.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to begin transaction: %w", err)
+	}
+	defer sqlTx.Rollback(ctx)
+
+	err = upsertVestingEscrow(ctx, sqlTx, slot, account, escrow)
+	if err != nil {
+		return fmt.Errorf("failed to upsert vesting escrow: %w", err)
+	}
+	err = sqlTx.Commit(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to commit transaction: %w", err)
+	}
+	return nil
+}
+
 func (i *Indexer) processTransaction(ctx context.Context, slot uint64, tx *solana.Transaction) error {
 	signature := tx.Signatures[0].String()
 	logger := i.logger.With(

solana/indexer/dbc/indexer_test.go

@@ -403,3 +403,82 @@ func TestHandleUpdate_Config(t *testing.T) {
 	require.NoError(t, err, "failed to query for dbc config vestings")
 	assert.True(t, exists, "dbc config vestings should exist after indexing")
 }
+
+func TestHandleUpdate_VestingEscrow(t *testing.T) {
+	pool := database.CreateTestDatabase(t, "test_solana_indexer_dbc")
+	rpcClient := fake_rpc_client.FakeRpcClient{}
+	logger := zap.NewNop()
+
+	indexer := New(common.GrpcConfig{}, &rpcClient, pool, config.Cfg, nil, logger)
+
+	escrowAddress := solana.MustPublicKeyFromBase58("7fXoYtLh1bG7q3Yh3b3Y9T5oX5nU5Y1Z6L8v1K5vU6Lm")
+	escrowBase64 := "9He3BEl0h8Pma7+K8q1+Y5zeZOJijiDxit+NwzCo6fSDuaAK1rtuUHcaLxqg1AjgXzqUrEhTyula0B3153h+c/pj9a/R+ETB2mNoH3KGvMgGcZ4sK1CiAVckO/yWaLEVILxThD7c2wiAwc02YNTKuGhqgHA0AyospRPz5Pq9NNsMGdSqUKuLgf4CAQAAAAAAwZvyaAAAAACAUQEAAAAAAAAAAAAAAAAADhyqNy35AAAhBwAAAAAAAAYAAAAAAAAAwZvyaAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="
+	escrowData, err := base64.StdEncoding.DecodeString(escrowBase64)
+	require.NoError(t, err)
+
+	update := pb.SubscribeUpdate{
+		UpdateOneof: &pb.SubscribeUpdate_Account{
+			Account: &pb.SubscribeUpdateAccount{
+				Slot: 600000000,
+				Account: &pb.SubscribeUpdateAccountInfo{
+					Pubkey:       escrowAddress.Bytes(),
+					Data:         escrowData,
+					TxnSignature: nil,
+				},
+			},
+		},
+	}
+
+	err = indexer.HandleUpdate(t.Context(), &update)
+	require.NoError(t, err)
+
+	// Verify that the vesting escrow was inserted
+	sql := `
+		SELECT EXISTS (
+			SELECT 1
+			FROM sol_locker_vesting_escrows
+			WHERE account = @account
+				AND slot = @slot
+				AND recipient = @recipient
+				AND token_mint = @token_mint
+				AND creator = @creator
+				AND base = @base
+				AND escrow_bump = @escrow_bump
+				AND update_recipient_mode = @update_recipient_mode
+				AND cancel_mode = @cancel_mode
+				AND token_program_flag = @token_program_flag
+				AND cliff_time = @cliff_time
+				AND frequency = @frequency
+				AND cliff_unlock_amount = @cliff_unlock_amount
+				AND amount_per_period = @amount_per_period
+				AND number_of_period = @number_of_period
+				AND total_claimed_amount = @total_claimed_amount
+				AND vesting_start_time = @vesting_start_time
+				AND cancelled_at = @cancelled_at
+			LIMIT 1
+		)
+	`
+	var exists bool
+	err = pool.QueryRow(t.Context(), sql, pgx.NamedArgs{
+		"account":               escrowAddress.String(),
+		"slot":                  int64(600000000),
+		"recipient":             "GWU4gnhaGPdhh4mcXudMpBXRUVJxqRPPXaSc8UkU2D3u",
+		"token_mint":            "91vg3y8HsmcShJARjpEMZBu5z2W5BwY4fdCj46QZCCnk",
+		"creator":               "FhVo3mqL8PW5pH5U2CN4XE33DokiyZnUwuGpH2hmHLuM",
+		"base":                  "9fcau4PNu4JGuS5J8dqP2qJiQbzxu5KeV12a94khdTma",
+		"escrow_bump":           uint8(254),
+		"update_recipient_mode": int8(2),
+		"cancel_mode":           int8(1),
+		"token_program_flag":    int8(0),
+		"cliff_time":            int64(1760730049),
+		"frequency":             int64(86400),
+		"cliff_unlock_amount":   uint64(0),
+		"amount_per_period":     uint64(273972602739726),
+		"number_of_period":      uint64(1825),
+		"total_claimed_amount":  uint64(6),
+		"vesting_start_time":    int64(1760730049),
+		"cancelled_at":          int64(0),
+	}).Scan(&exists)
+	require.NoError(t, err, "failed to query for vesting escrow")
+	assert.True(t, exists, "vesting escrow should exist after indexing")
+}

solana/indexer/dbc/locker.go

@@ -0,0 +1,105 @@
+package dbc
+
+import (
+	"context"
+
+	"api.audius.co/database"
+	"api.audius.co/solana/spl/programs/meteora_locker"
+	"github.com/gagliardetto/solana-go"
+	"github.com/jackc/pgx/v5"
+)
+
+func upsertVestingEscrow(
+	ctx context.Context,
+	db database.DBTX,
+	slot uint64,
+	account solana.PublicKey,
+	escrow *meteora_locker.VestingEscrow,
+) error {
+	sql := `
+		INSERT INTO sol_locker_vesting_escrows (
+			account,
+			slot,
+			recipient,
+			token_mint,
+			creator,
+			base,
+			escrow_bump,
+			update_recipient_mode,
+			cancel_mode,
+			token_program_flag,
+			cliff_time,
+			frequency,
+			cliff_unlock_amount,
+			amount_per_period,
+			number_of_period,
+			total_claimed_amount,
+			vesting_start_time,
+			cancelled_at,
+			created_at,
+			updated_at
+		) VALUES (
+			@account,
+			@slot,
+			@recipient,
+			@tokenMint,
+			@creator,
+			@base,
+			@escrowBump,
+			@updateRecipientMode,
+			@cancelMode,
+			@tokenProgramFlag,
+			@cliffTime,
+			@frequency,
+			@cliffUnlockAmount,
+			@amountPerPeriod,
+			@numberOfPeriod,
+			@totalClaimedAmount,
+			@vestingStartTime,
+			@cancelledAt,
+			NOW(),
+			NOW()
+		)
+		ON CONFLICT (account) DO UPDATE SET
+			slot = EXCLUDED.slot,
+			recipient = EXCLUDED.recipient,
+			token_mint = EXCLUDED.token_mint,
+			creator = EXCLUDED.creator,
+			base = EXCLUDED.base,
+			escrow_bump = EXCLUDED.escrow_bump,
+			update_recipient_mode = EXCLUDED.update_recipient_mode,
+			cancel_mode = EXCLUDED.cancel_mode,
+			token_program_flag = EXCLUDED.token_program_flag,
+			cliff_time = EXCLUDED.cliff_time,
+			frequency = EXCLUDED.frequency,
+			cliff_unlock_amount = EXCLUDED.cliff_unlock_amount,
+			amount_per_period = EXCLUDED.amount_per_period,
+			number_of_period = EXCLUDED.number_of_period,
+			total_claimed_amount = EXCLUDED.total_claimed_amount,
+			vesting_start_time = EXCLUDED.vesting_start_time,
+			cancelled_at = EXCLUDED.cancelled_at,
+			updated_at = NOW()
+		WHERE sol_locker_vesting_escrows.slot > EXCLUDED.slot
+	;`
+	_, err := db.Exec(ctx, sql, pgx.NamedArgs{
+		"account":             account.String(),
+		"slot":                slot,
+		"recipient":           escrow.Recipient.String(),
+		"tokenMint":           escrow.TokenMint.String(),
+		"creator":             escrow.Creator.String(),
+		"base":                escrow.Base.String(),
+		"escrowBump":          escrow.EscrowBump,
+		"updateRecipientMode": int8(escrow.UpdateRecipientMode),
+		"cancelMode":          int8(escrow.CancelMode),
+		"tokenProgramFlag":    int8(escrow.TokenProgramFlag),
+		"cliffTime":           escrow.CliffTime,
+		"frequency":           escrow.Frequency,
+		"cliffUnlockAmount":   escrow.CliffUnlockAmount,
+		"amountPerPeriod":     escrow.AmountPerPeriod,
+		"numberOfPeriod":      escrow.NumberOfPeriod,
+		"totalClaimedAmount":  escrow.TotalClaimedAmount,
+		"vestingStartTime":    escrow.VestingStartTime,
+		"cancelledAt":         escrow.CancelledAt,
+	})
+	return err
+}