Add remaining state changes

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/indexer/indexer.go

@@ -45,6 +45,7 @@ type Indexer struct {
 	tokenTransferProcessor  TokenTransferProcessorInterface
 	effectsProcessor        OperationProcessorInterface
 	contractDeployProcessor OperationProcessorInterface
+	contractEventsProcessor OperationProcessorInterface
 }
 
 func NewIndexer(networkPassphrase string) *Indexer {
@@ -54,6 +55,7 @@ func NewIndexer(networkPassphrase string) *Indexer {
 		tokenTransferProcessor:  processors.NewTokenTransferProcessor(networkPassphrase),
 		effectsProcessor:        processors.NewEffectsProcessor(networkPassphrase),
 		contractDeployProcessor: processors.NewContractDeployProcessor(networkPassphrase),
+		contractEventsProcessor: processors.NewEventsProcessor(networkPassphrase),
 	}
 }
 
@@ -80,7 +82,7 @@ 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
+	var effectsStateChanges, contractDeployStateChanges, contractEventsStateChanges []types.StateChange
 	for opID, opParticipants := range opsParticipants {
 		dataOp, err = processors.ConvertOperation(&transaction, &opParticipants.OpWrapper.Operation, opID)
 		if err != nil {
@@ -104,6 +106,13 @@ func (i *Indexer) ProcessTransaction(ctx context.Context, transaction ingest.Led
 			return fmt.Errorf("processing contract deploy state changes: %w", err)
 		}
 		i.Buffer.PushStateChanges(contractDeployStateChanges)
+
+		// 2.3 Index contract events state changes
+		contractEventsStateChanges, err = i.contractEventsProcessor.ProcessOperation(ctx, opParticipants.OpWrapper)
+		if err != nil && !errors.Is(err, processors.ErrInvalidOpType) {
+			return fmt.Errorf("processing contract events state changes: %w", err)
+		}
+		i.Buffer.PushStateChanges(contractEventsStateChanges)
 	}
 
 	// 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, mockEventsProcessor *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
 				participants := set.NewSet("alice", "bob")
 				mockParticipants.On("GetTransactionParticipants", mock.Anything).Return(participants, nil)
 
@@ -121,6 +121,9 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 				contractDeployStateChanges := []types.StateChange{{ToID: 3, StateChangeOrder: 1}}
 				mockContractDeploy.On("ProcessOperation", mock.Anything, mock.Anything).Return(contractDeployStateChanges, nil)
 
+				contractEventsStateChanges := []types.StateChange{{ToID: 4, StateChangeOrder: 1}}
+				mockEventsProcessor.On("ProcessOperation", mock.Anything, mock.Anything).Return(contractEventsStateChanges, 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, mockEventsProcessor *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, mockEventsProcessor *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, mockEventsProcessor *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, mockEventsProcessor *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
 				participants := set.NewSet[string]()
 				mockParticipants.On("GetTransactionParticipants", mock.Anything).Return(participants, nil)
 
@@ -236,7 +243,7 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 		},
 		{
 			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, mockEventsProcessor *MockOperationProcessor, mockBuffer *MockIndexerBuffer) {
 				participants := set.NewSet[string]()
 				mockParticipants.On("GetTransactionParticipants", mock.Anything).Return(participants, nil)
 
@@ -256,10 +263,11 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 			mockTokenTransfer := &MockTokenTransferProcessor{}
 			mockEffects := &MockOperationProcessor{}
 			mockContractDeploy := &MockOperationProcessor{}
+			mockEventsProcessor := &MockOperationProcessor{}
 			mockBuffer := &MockIndexerBuffer{}
 
 			// Setup mock expectations
-			tt.setupMocks(mockParticipants, mockTokenTransfer, mockEffects, mockContractDeploy, mockBuffer)
+			tt.setupMocks(mockParticipants, mockTokenTransfer, mockEffects, mockContractDeploy, mockEventsProcessor, mockBuffer)
 
 			// Create testable indexer with mocked dependencies
 			indexer := &Indexer{
@@ -268,6 +276,7 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 				tokenTransferProcessor:  mockTokenTransfer,
 				effectsProcessor:        mockEffects,
 				contractDeployProcessor: mockContractDeploy,
+				contractEventsProcessor: mockEventsProcessor,
 			}
 
 			err := indexer.ProcessTransaction(context.Background(), testTx)
@@ -305,6 +314,8 @@ func TestIndexer_ProcessTransaction(t *testing.T) {
 			mockParticipants.AssertExpectations(t)
 			mockTokenTransfer.AssertExpectations(t)
 			mockEffects.AssertExpectations(t)
+			mockContractDeploy.AssertExpectations(t)
+			mockEventsProcessor.AssertExpectations(t)
 			mockBuffer.AssertExpectations(t)
 		})
 	}

internal/indexer/processors/effects.go

@@ -11,6 +11,7 @@ import (
 	"github.com/stellar/go/ingest"
 	effects "github.com/stellar/go/processors/effects"
 	operation_processor "github.com/stellar/go/processors/operation"
+	"github.com/stellar/go/strkey"
 	"github.com/stellar/go/xdr"
 
 	"github.com/stellar/go/support/log"
@@ -100,6 +101,11 @@ func (p *EffectsProcessor) ProcessOperation(_ context.Context, opWrapper *operat
 		// Process different types of effects based on what account property they modify
 		//exhaustive:ignore
 		switch effectType {
+		case effects.EffectSequenceBumped:
+			stateChanges = append(stateChanges, changeBuilder.
+				WithCategory(types.StateChangeCategorySequence).
+				WithReason(types.StateChangeReasonSequenceBump).
+				Build())
 		// Signer effects: track changes to account signers (add/remove/update signer keys and weights)
 		case effects.EffectSignerCreated, effects.EffectSignerRemoved, effects.EffectSignerUpdated:
 			changeBuilder = changeBuilder.
@@ -135,6 +141,15 @@ func (p *EffectsProcessor) ProcessOperation(_ context.Context, opWrapper *operat
 			changeBuilder = changeBuilder.WithCategory(types.StateChangeCategoryTrustlineFlags)
 			stateChanges = append(stateChanges, p.parseFlags(trustlineFlags, changeBuilder, &effect)...)
 
+		// Change trust effects
+		case effects.EffectTrustlineCreated, effects.EffectTrustlineRemoved, effects.EffectTrustlineUpdated:
+			changeBuilder = changeBuilder.WithCategory(types.StateChangeCategoryTrustline)
+			trustlineChange, err := p.parseTrustline(changeBuilder, &effect, effectType, changes)
+			if err != nil {
+				return nil, fmt.Errorf("parsing trustline effect: effectType: %s, address: %s, txHash: %s, opID: %d, err: %w", effect.TypeString, effect.Address, txHash, opWrapper.ID(), err)
+			}
+			stateChanges = append(stateChanges, trustlineChange)
+
 		// Data entry effects: track changes to account data entries (key-value storage)
 		case effects.EffectDataCreated, effects.EffectDataRemoved, effects.EffectDataUpdated:
 			keyValueMap := p.parseKeyValue(&effect, effectType, changes, "name")
@@ -249,6 +264,59 @@ func (p *EffectsProcessor) createTargetSponsorshipChange(reason types.StateChang
 	return builder.Build()
 }
 
+func (p *EffectsProcessor) parseTrustline(baseBuilder *StateChangeBuilder, effect *effects.EffectOutput, effectType effects.EffectType, changes []ingest.Change) (types.StateChange, error) {
+	parseAsset := func(assetType, assetCode, assetIssuer string) (string, error) {
+		var asset xdr.Asset
+
+		switch assetType {
+		case "native":
+			asset = xdr.Asset{
+				Type: xdr.AssetTypeAssetTypeNative,
+			}
+		case "credit_alphanum4", "credit_alphanum12":
+			asset = xdr.MustNewCreditAsset(assetCode, assetIssuer)
+		default:
+			return "", fmt.Errorf("invalid asset type: %s", assetType)
+		}
+
+		contractID, err := asset.ContractID(p.networkPassphrase)
+		if err != nil {
+			return "", fmt.Errorf("getting asset contract ID: %w", err)
+		}
+
+		return strkey.MustEncode(strkey.VersionByteContract, contractID[:]), nil
+	}
+
+	var change types.StateChange
+	assetStr, err := parseAsset(effect.Details["asset_type"].(string), effect.Details["asset_code"].(string), effect.Details["asset_issuer"].(string))
+	if err != nil {
+		return types.StateChange{}, fmt.Errorf("parsing asset: %w", err)
+	}
+	//exhaustive:ignore
+	switch effectType {
+	case effects.EffectTrustlineCreated:
+		change = baseBuilder.WithReason(types.StateChangeReasonSet).WithToken(assetStr).WithKeyValue(
+			map[string]any{
+				"limit": effect.Details["limit"],
+			},
+		).Build()
+	case effects.EffectTrustlineRemoved:
+		change = baseBuilder.WithReason(types.StateChangeReasonRemove).WithToken(assetStr).Build()
+	case effects.EffectTrustlineUpdated:
+		prevLedgerEntryState := p.getPrevLedgerEntryState(effect, xdr.LedgerEntryTypeTrustline, changes)
+		prevTrustline := prevLedgerEntryState.Data.MustTrustLine()
+		change = baseBuilder.WithReason(types.StateChangeReasonUpdate).WithToken(assetStr).WithKeyValue(
+			map[string]any{
+				"limit": map[string]any{
+					"old": strconv.FormatInt(int64(prevTrustline.Limit), 10),
+					"new": effect.Details["limit"],
+				},
+			},
+		).Build()
+	}
+	return change, nil
+}
+
 // parseKeyValue extracts specified key-value pairs from effect details.
 // This is used to capture metadata like home domain values or data entry names and values.
 func (p *EffectsProcessor) parseKeyValue(effect *effects.EffectOutput, effectType effects.EffectType, changes []ingest.Change, key string) map[string]any {