Add remaining state changes (#313)

Author: aditya1702

go.mod

@@ -22,7 +22,7 @@ require (
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/cobra v1.9.1
 	github.com/spf13/viper v1.20.1
-	github.com/stellar/go v0.0.0-20250822224526-9397ce4b6da2
+	github.com/stellar/go v0.0.0-20250903085211-00c0b06cd7cc
 	github.com/stellar/go-xdr v0.0.0-20231122183749-b53fb00bcac2
 	github.com/stellar/stellar-rpc v0.9.6-0.20250523162628-6bb9d7a387d5
 	github.com/stretchr/testify v1.10.0

go.sum

@@ -358,8 +358,8 @@ github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o=
 github.com/spf13/pflag v1.0.6/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
 github.com/spf13/viper v1.20.1 h1:ZMi+z/lvLyPSCoNtFCpqjy0S4kPbirhpTMwl8BkW9X4=
 github.com/spf13/viper v1.20.1/go.mod h1:P9Mdzt1zoHIG8m2eZQinpiBjo6kCmZSKBClNNqjJvu4=
-github.com/stellar/go v0.0.0-20250822224526-9397ce4b6da2 h1:nZEjpYUZ+0cYqZXhYO+QiJcwPQnwTrEs2bWsbRv/Tz8=
-github.com/stellar/go v0.0.0-20250822224526-9397ce4b6da2/go.mod h1:ac8hwpljbFXC3Sf9nGfqBXXEvAEdnNRqQHGqP7QN8oY=
+github.com/stellar/go v0.0.0-20250903085211-00c0b06cd7cc h1:iRveajcdqBdOlp5U6ZCts9dr1J4VM3LJ8sVnpHYkoVE=
+github.com/stellar/go v0.0.0-20250903085211-00c0b06cd7cc/go.mod h1:ac8hwpljbFXC3Sf9nGfqBXXEvAEdnNRqQHGqP7QN8oY=
 github.com/stellar/go-xdr v0.0.0-20231122183749-b53fb00bcac2 h1:OzCVd0SV5qE3ZcDeSFCmOWLZfEWZ3Oe8KtmSOYKEVWE=
 github.com/stellar/go-xdr v0.0.0-20231122183749-b53fb00bcac2/go.mod h1:yoxyU/M8nl9LKeWIoBrbDPQ7Cy+4jxRcWcOayZ4BMps=
 github.com/stellar/stellar-rpc v0.9.6-0.20250523162628-6bb9d7a387d5 h1:V+XezRLVHuk6c1nMkXkWjCwtoHN7F+DK86dK2kkNSZo=

internal/db/migrations/2025-06-10.4-create_indexer_table_state_changes.sql

@@ -25,6 +25,7 @@ CREATE TABLE state_changes (
     deployer_account_id TEXT,
     funder_account_id TEXT,
     thresholds JSONB,
+    trustline_limit JSONB,
     PRIMARY KEY (to_id, state_change_order)
 );
 

internal/indexer/indexer.go

@@ -10,6 +10,7 @@ import (
 	operation_processor "github.com/stellar/go/processors/operation"
 
 	"github.com/stellar/wallet-backend/internal/indexer/processors"
+	contract_processors "github.com/stellar/wallet-backend/internal/indexer/processors/contracts"
 	"github.com/stellar/wallet-backend/internal/indexer/types"
 )
 
@@ -37,23 +38,26 @@ type ParticipantsProcessorInterface interface {
 
 type OperationProcessorInterface interface {
 	ProcessOperation(ctx context.Context, opWrapper *operation_processor.TransactionOperationWrapper) ([]types.StateChange, error)
+	Name() string
 }
 
 type Indexer struct {
-	Buffer                  IndexerBufferInterface
-	participantsProcessor   ParticipantsProcessorInterface
-	tokenTransferProcessor  TokenTransferProcessorInterface
-	effectsProcessor        OperationProcessorInterface
-	contractDeployProcessor OperationProcessorInterface
+	Buffer                 IndexerBufferInterface
+	participantsProcessor  ParticipantsProcessorInterface
+	tokenTransferProcessor TokenTransferProcessorInterface
+	processors             []OperationProcessorInterface
 }
 
-func NewIndexer(networkPassphrase string) *Indexer {
+func NewIndexer(networkPassphrase string, ledgerEntryProvider processors.LedgerEntryProvider) *Indexer {
 	return &Indexer{
-		Buffer:                  NewIndexerBuffer(),
-		participantsProcessor:   processors.NewParticipantsProcessor(networkPassphrase),
-		tokenTransferProcessor:  processors.NewTokenTransferProcessor(networkPassphrase),
-		effectsProcessor:        processors.NewEffectsProcessor(networkPassphrase),
-		contractDeployProcessor: processors.NewContractDeployProcessor(networkPassphrase),
+		Buffer:                 NewIndexerBuffer(),
+		participantsProcessor:  processors.NewParticipantsProcessor(networkPassphrase),
+		tokenTransferProcessor: processors.NewTokenTransferProcessor(networkPassphrase),
+		processors: []OperationProcessorInterface{
+			processors.NewEffectsProcessor(networkPassphrase, ledgerEntryProvider),
+			processors.NewContractDeployProcessor(networkPassphrase),
+			contract_processors.NewSACEventsProcessor(networkPassphrase),
+		},
 	}
 }
 
@@ -80,7 +84,6 @@ func (i *Indexer) ProcessTransaction(ctx context.Context, transaction ingest.Led
 		return fmt.Errorf("getting operations participants: %w", err)
 	}
 	var dataOp *types.Operation
-	var effectsStateChanges, contractDeployStateChanges []types.StateChange
 	for opID, opParticipants := range opsParticipants {
 		dataOp, err = processors.ConvertOperation(&transaction, &opParticipants.OpWrapper.Operation, opID)
 		if err != nil {
@@ -91,19 +94,13 @@ func (i *Indexer) ProcessTransaction(ctx context.Context, transaction ingest.Led
 			i.Buffer.PushParticipantOperation(participant, *dataOp, *dataTx)
 		}
 
-		// 2.1. Index effects state changes
-		effectsStateChanges, err = i.effectsProcessor.ProcessOperation(ctx, opParticipants.OpWrapper)
-		if err != nil {
-			return fmt.Errorf("processing effects state changes: %w", err)
-		}
-		i.Buffer.PushStateChanges(effectsStateChanges)
-
-		// 2.2. Index contract deploy state changes
-		contractDeployStateChanges, err = i.contractDeployProcessor.ProcessOperation(ctx, opParticipants.OpWrapper)
-		if err != nil && !errors.Is(err, processors.ErrInvalidOpType) {
-			return fmt.Errorf("processing contract deploy state changes: %w", err)
+		for _, processor := range i.processors {
+			stateChanges, processorErr := processor.ProcessOperation(ctx, opParticipants.OpWrapper)
+			if processorErr != nil && !errors.Is(processorErr, processors.ErrInvalidOpType) {
+				return fmt.Errorf("processing %s state changes: %w", processor.Name(), processorErr)
+			}
+			i.Buffer.PushStateChanges(stateChanges)
 		}
-		i.Buffer.PushStateChanges(contractDeployStateChanges)
 	}
 
 	// 3. Index token transfer state changes

internal/indexer/indexer_test.go

@@ -88,14 +88,14 @@ var (
 func TestIndexer_ProcessTransaction(t *testing.T) {
 	tests := []struct {
 		name            string
-		setupMocks      func(*MockParticipantsProcessor, *MockTokenTransferProcessor, *MockOperationProcessor, *MockOperationProcessor, *MockIndexerBuffer)
+		setupMocks      func(*MockParticipantsProcessor, *MockTokenTransferProcessor, *MockOperationProcessor, *MockOperationProcessor, *MockOperationProcessor, *MockIndexerBuffer)
 		wantError       string
 		txParticipants  set.Set[string]
 		opsParticipants map[int64]processors.OperationParticipants
 	}{
 		{
 			name: "🟢 successful processing with participants",
-			setupMocks: func(mockParticipants *MockParticipantsProcessor, mockTokenTransfer *MockTokenTransferProcessor, mockEffects *MockOperationProcessor, mockContractDeploy *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
+			setupMocks: func(mockParticipants *MockParticipantsProcessor, mockTokenTransfer *MockTokenTransferProcessor, mockEffects *MockOperationProcessor, mockContractDeploy *MockOperationProcessor, mockSACEventsProcessor *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
 				participants := set.NewSet("alice", "bob")
 				mockParticipants.On("GetTransactionParticipants", mock.Anything).Return(participants, nil)
 
@@ -116,11 +116,14 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 				mockTokenTransfer.On("ProcessTransaction", mock.Anything, mock.Anything).Return(tokenStateChanges, nil)
 
 				effectsStateChanges := []types.StateChange{{ToID: 2, StateChangeOrder: 1}}
-				mockEffects.On("ProcessOperation", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(effectsStateChanges, nil)
+				mockEffects.On("ProcessOperation", mock.Anything, mock.Anything).Return(effectsStateChanges, nil)
 
 				contractDeployStateChanges := []types.StateChange{{ToID: 3, StateChangeOrder: 1}}
 				mockContractDeploy.On("ProcessOperation", mock.Anything, mock.Anything).Return(contractDeployStateChanges, nil)
 
+				contractSACEventsStateChanges := []types.StateChange{{ToID: 4, StateChangeOrder: 1}}
+				mockSACEventsProcessor.On("ProcessOperation", mock.Anything, mock.Anything).Return(contractSACEventsStateChanges, nil)
+
 				// Verify transaction was pushed to buffer with correct participants
 				// PushParticipantTransaction is called once for each participant
 				mockBuffer.On("PushParticipantTransaction", "alice",
@@ -157,6 +160,10 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 					mock.MatchedBy(func(stateChanges []types.StateChange) bool {
 						return len(stateChanges) == 1 && stateChanges[0].ToID == 3 && stateChanges[0].StateChangeOrder == 1
 					})).Return()
+				mockBuffer.On("PushStateChanges",
+					mock.MatchedBy(func(stateChanges []types.StateChange) bool {
+						return len(stateChanges) == 1 && stateChanges[0].ToID == 4 && stateChanges[0].StateChangeOrder == 1
+					})).Return()
 			},
 			txParticipants: set.NewSet("alice", "bob"),
 			opsParticipants: map[int64]processors.OperationParticipants{
@@ -173,7 +180,7 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 		},
 		{
 			name: "🟢 successful processing without participants",
-			setupMocks: func(mockParticipants *MockParticipantsProcessor, mockTokenTransfer *MockTokenTransferProcessor, mockEffects *MockOperationProcessor, mockContractDeploy *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
+			setupMocks: func(mockParticipants *MockParticipantsProcessor, mockTokenTransfer *MockTokenTransferProcessor, mockEffects *MockOperationProcessor, mockContractDeploy *MockOperationProcessor, mockSACEventsProcessor *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
 				participants := set.NewSet[string]()
 				mockParticipants.On("GetTransactionParticipants", mock.Anything).Return(participants, nil)
 
@@ -196,14 +203,14 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 		},
 		{
 			name: "🔴 error getting transaction participants",
-			setupMocks: func(mockParticipants *MockParticipantsProcessor, mockTokenTransfer *MockTokenTransferProcessor, mockEffects *MockOperationProcessor, mockContractDeploy *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
+			setupMocks: func(mockParticipants *MockParticipantsProcessor, mockTokenTransfer *MockTokenTransferProcessor, mockEffects *MockOperationProcessor, mockContractDeploy *MockOperationProcessor, mockSACEventsProcessor *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
 				mockParticipants.On("GetTransactionParticipants", mock.Anything).Return(set.NewSet[string](), errors.New("participant error"))
 			},
 			wantError: "getting transaction participants: participant error",
 		},
 		{
 			name: "🔴 error getting operations participants",
-			setupMocks: func(mockParticipants *MockParticipantsProcessor, mockTokenTransfer *MockTokenTransferProcessor, mockEffects *MockOperationProcessor, mockContractDeploy *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
+			setupMocks: func(mockParticipants *MockParticipantsProcessor, mockTokenTransfer *MockTokenTransferProcessor, mockEffects *MockOperationProcessor, mockContractDeploy *MockOperationProcessor, mockSACEventsProcessor *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
 				participants := set.NewSet[string]()
 				mockParticipants.On("GetTransactionParticipants", mock.Anything).Return(participants, nil)
 				mockParticipants.On("GetOperationsParticipants", mock.Anything).Return(map[int64]processors.OperationParticipants{}, errors.New("operations error"))
@@ -212,7 +219,7 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 		},
 		{
 			name: "🔴 error processing effects state changes",
-			setupMocks: func(mockParticipants *MockParticipantsProcessor, mockTokenTransfer *MockTokenTransferProcessor, mockEffects *MockOperationProcessor, mockContractDeploy *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
+			setupMocks: func(mockParticipants *MockParticipantsProcessor, mockTokenTransfer *MockTokenTransferProcessor, mockEffects *MockOperationProcessor, mockContractDeploy *MockOperationProcessor, mockSACEventsProcessor *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
 				participants := set.NewSet[string]()
 				mockParticipants.On("GetTransactionParticipants", mock.Anything).Return(participants, nil)
 
@@ -231,12 +238,13 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 
 				mockBuffer.On("PushParticipantOperation", mock.Anything, mock.Anything, mock.Anything).Return()
 				mockEffects.On("ProcessOperation", mock.Anything, mock.Anything).Return([]types.StateChange{}, errors.New("effects error"))
+				mockEffects.On("Name").Return("effects")
 			},
 			wantError: "processing effects state changes: effects error",
 		},
 		{
 			name: "🔴 error processing token transfer state changes",
-			setupMocks: func(mockParticipants *MockParticipantsProcessor, mockTokenTransfer *MockTokenTransferProcessor, mockEffects *MockOperationProcessor, mockContractDeploy *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
+			setupMocks: func(mockParticipants *MockParticipantsProcessor, mockTokenTransfer *MockTokenTransferProcessor, mockEffects *MockOperationProcessor, mockContractDeploy *MockOperationProcessor, mockSACEventsProcessor *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
 				participants := set.NewSet[string]()
 				mockParticipants.On("GetTransactionParticipants", mock.Anything).Return(participants, nil)
 
@@ -256,18 +264,18 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 			mockTokenTransfer := &MockTokenTransferProcessor{}
 			mockEffects := &MockOperationProcessor{}
 			mockContractDeploy := &MockOperationProcessor{}
+			mockSACEventsProcessor := &MockOperationProcessor{}
 			mockBuffer := &MockIndexerBuffer{}
 
 			// Setup mock expectations
-			tt.setupMocks(mockParticipants, mockTokenTransfer, mockEffects, mockContractDeploy, mockBuffer)
+			tt.setupMocks(mockParticipants, mockTokenTransfer, mockEffects, mockContractDeploy, mockSACEventsProcessor, mockBuffer)
 
 			// Create testable indexer with mocked dependencies
 			indexer := &Indexer{
-				Buffer:                  mockBuffer,
-				participantsProcessor:   mockParticipants,
-				tokenTransferProcessor:  mockTokenTransfer,
-				effectsProcessor:        mockEffects,
-				contractDeployProcessor: mockContractDeploy,
+				Buffer:                 mockBuffer,
+				participantsProcessor:  mockParticipants,
+				tokenTransferProcessor: mockTokenTransfer,
+				processors:             []OperationProcessorInterface{mockEffects, mockContractDeploy, mockSACEventsProcessor},
 			}
 
 			err := indexer.ProcessTransaction(context.Background(), testTx)
@@ -305,6 +313,8 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 			mockParticipants.AssertExpectations(t)
 			mockTokenTransfer.AssertExpectations(t)
 			mockEffects.AssertExpectations(t)
+			mockContractDeploy.AssertExpectations(t)
+			mockSACEventsProcessor.AssertExpectations(t)
 			mockBuffer.AssertExpectations(t)
 		})
 	}

internal/indexer/mocks.go

@@ -45,6 +45,11 @@ func (m *MockOperationProcessor) ProcessOperation(ctx context.Context, opWrapper
 	return args.Get(0).([]types.StateChange), args.Error(1)
 }
 
+func (m *MockOperationProcessor) Name() string {
+	args := m.Called()
+	return args.String(0)
+}
+
 type MockIndexerBuffer struct {
 	mock.Mock
 }

internal/indexer/processors/contract_deploy.go

@@ -19,6 +19,10 @@ func NewContractDeployProcessor(networkPassphrase string) *ContractDeployProcess
 	return &ContractDeployProcessor{networkPassphrase: networkPassphrase}
 }
 
+func (p *ContractDeployProcessor) Name() string {
+	return "contract_deploy"
+}
+
 // ProcessOperation emits a state change for each contract deployment (including subinvocations).
 func (p *ContractDeployProcessor) ProcessOperation(_ context.Context, op *operation_processor.TransactionOperationWrapper) ([]types.StateChange, error) {
 	if op.OperationType() != xdr.OperationTypeInvokeHostFunction {