Process SAC balance entry changes and store SAC balances

cmd/ingest.go

@@ -35,8 +35,6 @@ func (c *ingestCmd) Command() *cobra.Command {
 		utils.IngestServerPortOption(&cfg.ServerPort),
 		utils.AdminPortOption(&cfg.AdminPort),
 		utils.GetLedgersLimitOption(&cfg.GetLedgersLimit),
-		utils.RedisHostOption(&cfg.RedisHost),
-		utils.RedisPortOption(&cfg.RedisPort),
 		utils.EnableParticipantFilteringOption(&cfg.EnableParticipantFiltering),
 		{
 			Name:        "ingestion-mode",

cmd/serve.go

@@ -37,8 +37,6 @@ func (c *serveCmd) Command() *cobra.Command {
 		utils.GraphQLComplexityLimitOption(&cfg.GraphQLComplexityLimit),
 		utils.MaxAccountsPerBalancesQueryOption(&cfg.MaxAccountsPerBalancesQuery),
 		utils.MaxGraphQLWorkerPoolSizeOption(&cfg.MaxGraphQLWorkerPoolSize),
-		utils.RedisHostOption(&cfg.RedisHost),
-		utils.RedisPortOption(&cfg.RedisPort),
 		utils.EnableParticipantFilteringOption(&cfg.EnableParticipantFiltering),
 		{
 			Name:        "port",

cmd/utils/global_options.go

@@ -10,26 +10,6 @@ import (
 	"github.com/stellar/wallet-backend/internal/signing"
 )
 
-func RedisHostOption(configKey *string) *config.ConfigOption {
-	return &config.ConfigOption{
-		Name:        "redis-host",
-		Usage:       "The host of the Redis server.",
-		OptType:     types.String,
-		ConfigKey:   configKey,
-		FlagDefault: "localhost",
-	}
-}
-
-func RedisPortOption(configKey *int) *config.ConfigOption {
-	return &config.ConfigOption{
-		Name:        "redis-port",
-		Usage:       "The port of the Redis server.",
-		OptType:     types.Int,
-		ConfigKey:   configKey,
-		FlagDefault: 6379,
-	}
-}
-
 func IngestServerPortOption(configKey *int) *config.ConfigOption {
 	return &config.ConfigOption{
 		Name:        "ingest-server-port",

docker-compose.yaml

@@ -27,24 +27,6 @@ services:
       - source: postgres_init
         target: /docker-entrypoint-initdb.d/init-schemas.sql
 
-  redis:
-    container_name: redis
-    image: redis:7-alpine
-    healthcheck:
-      test: ["CMD", "redis-cli", "ping"]
-      interval: 10s
-      timeout: 5s
-      retries: 5
-      start_period: 10s
-    ports:
-      - 6379:6379
-    volumes:
-      - redis-data:/data
-    command: >
-      redis-server
-      --hash-max-listpack-entries 1000
-      --hash-max-listpack-value 8000
-
   stellar-rpc-mainnet:
     container_name: stellar-rpc-mainnet
     image: stellar/stellar-rpc:stable
@@ -208,8 +190,6 @@ services:
         condition: service_completed_successfully
       db:
         condition: service_healthy
-      redis:
-        condition: service_healthy
     entrypoint: ""
     command:
       - sh
@@ -238,9 +218,6 @@ services:
       GET_LEDGERS_LIMIT: 150
       TRACKER_DSN: ${TRACKER_DSN}
       STELLAR_ENVIRONMENT: ${STELLAR_ENVIRONMENT:-development}
-      REDIS_HOST: ${REDIS_HOST:-redis}
-      REDIS_PORT: ${REDIS_PORT:-6379}
-      REDIS_PASSWORD: ${REDIS_PASSWORD:-}
       # ADMIN_PORT: 9094  # Optional: Port for pprof endpoints at /debug/pprof. WARNING: Do NOT expose publicly!
 
   ingest-mainnet:
@@ -256,8 +233,6 @@ services:
         condition: service_completed_successfully
       db:
         condition: service_healthy
-      redis:
-        condition: service_healthy
     entrypoint: ""
     command:
       - sh
@@ -285,15 +260,10 @@ services:
       GET_LEDGERS_LIMIT: 20
       TRACKER_DSN: ${TRACKER_DSN}
       STELLAR_ENVIRONMENT: ${STELLAR_ENVIRONMENT:-development}
-      REDIS_HOST: ${REDIS_HOST:-redis}
-      REDIS_PORT: ${REDIS_PORT:-6379}
-      REDIS_PASSWORD: ${REDIS_PASSWORD:-}
       # ADMIN_PORT: 9095  # Optional: Port for pprof endpoints at /debug/pprof. WARNING: Do NOT expose publicly!
 volumes:
   postgres-db:
     driver: local
-  redis-data:
-    driver: local
 
 configs:
   postgres_init:

go.mod

@@ -6,7 +6,6 @@ toolchain go1.24.2
 
 require (
 	github.com/99designs/gqlgen v0.17.76
-	github.com/alicebob/miniredis/v2 v2.35.0
 	github.com/alitto/pond/v2 v2.5.0
 	github.com/avast/retry-go/v4 v4.6.1
 	github.com/aws/aws-sdk-go v1.55.7
@@ -18,6 +17,7 @@ require (
 	github.com/go-chi/chi v4.1.2+incompatible
 	github.com/go-playground/validator/v10 v10.27.0
 	github.com/golang-jwt/jwt/v5 v5.2.3
+	github.com/google/uuid v1.6.0
 	github.com/guregu/null v4.0.0+incompatible
 	github.com/jackc/pgx/v5 v5.7.6
 	github.com/jmoiron/sqlx v1.4.0
@@ -26,7 +26,6 @@ require (
 	github.com/pelletier/go-toml v1.9.5
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.22.0
-	github.com/redis/go-redis/v9 v9.16.0
 	github.com/rubenv/sql-migrate v1.8.0
 	github.com/sirupsen/logrus v1.9.3
 	github.com/spf13/cobra v1.9.1
@@ -91,7 +90,6 @@ require (
 	github.com/creachadair/jrpc2 v1.2.0 // indirect
 	github.com/creachadair/mds v0.13.4 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
-	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/distribution/reference v0.6.0 // indirect
 	github.com/djherbis/fscache v0.10.1 // indirect
 	github.com/docker/docker v28.0.1+incompatible // indirect
@@ -115,7 +113,6 @@ require (
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
 	github.com/google/go-cmp v0.7.0 // indirect
 	github.com/google/s2a-go v0.1.8 // indirect
-	github.com/google/uuid v1.6.0 // indirect
 	github.com/googleapis/enterprise-certificate-proxy v0.3.4 // indirect
 	github.com/googleapis/gax-go/v2 v2.14.1 // indirect
 	github.com/gorilla/schema v1.4.1 // indirect
@@ -165,7 +162,6 @@ require (
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/tklauser/go-sysconf v0.3.12 // indirect
 	github.com/tklauser/numcpus v0.6.1 // indirect
-	github.com/yuin/gopher-lua v1.1.1 // indirect
 	github.com/yusufpapurcu/wmi v1.2.4 // indirect
 	github.com/zeebo/errs v1.4.0 // indirect
 	go.opencensus.io v0.24.0 // indirect

go.sum

@@ -56,8 +56,6 @@ github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuy
 github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc=
 github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
 github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0=
-github.com/alicebob/miniredis/v2 v2.35.0 h1:QwLphYqCEAo1eu1TqPRN2jgVMPBweeQcR21jeqDCONI=
-github.com/alicebob/miniredis/v2 v2.35.0/go.mod h1:TcL7YfarKPGDAthEtl5NBeHZfeUQj6OXMm/+iu5cLMM=
 github.com/alitto/pond/v2 v2.5.0 h1:vPzS5GnvSDRhWQidmj2djHllOmjFExVFbDGCw1jdqDw=
 github.com/alitto/pond/v2 v2.5.0/go.mod h1:xkjYEgQ05RSpWdfSd1nM3OVv7TBhLdy7rMp3+2Nq+yE=
 github.com/andreyvit/diff v0.0.0-20170406064948-c7f18ee00883 h1:bvNMNQO63//z+xNgfBlViaCIJKLlCJ6/fmUseuG0wVQ=
@@ -116,10 +114,6 @@ github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24
 github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
-github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
-github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
-github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
-github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8=
 github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE=
 github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
@@ -149,8 +143,6 @@ github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/deckarep/golang-set/v2 v2.8.0 h1:swm0rlPCmdWn9mESxKOjWk8hXSqoxOp+ZlfuyaAdFlQ=
 github.com/deckarep/golang-set/v2 v2.8.0/go.mod h1:VAky9rY/yGXJOLEDv3OMci+7wtDpOF4IN+y82NBOac4=
-github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
-github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/dgryski/trifles v0.0.0-20230903005119-f50d829f2e54 h1:SG7nF6SRlWhcT7cNTs5R6Hk4V2lcmLz2NsG2VnInyNo=
 github.com/dgryski/trifles v0.0.0-20230903005119-f50d829f2e54/go.mod h1:if7Fbed8SFyPtHLHbg49SI7NAdJiC5WIA09pe59rfAA=
 github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
@@ -420,8 +412,6 @@ github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsT
 github.com/prometheus/procfs v0.0.8/go.mod h1:7Qr8sr6344vo1JqZ6HhLceV9o3AJ1Ff+GxbHq6oeK9A=
 github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
 github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
-github.com/redis/go-redis/v9 v9.16.0 h1:OotgqgLSRCmzfqChbQyG1PHC3tLNR89DG4jdOERSEP4=
-github.com/redis/go-redis/v9 v9.16.0/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370=
 github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII=
 github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o=
 github.com/rs/cors v1.11.0 h1:0B9GE/r9Bc2UxRMMtymBkHTenPkHDv0CW4Y98GBY+po=
@@ -513,8 +503,6 @@ github.com/yudai/golcs v0.0.0-20150405163532-d1c525dea8ce h1:888GrqRxabUce7lj4Oa
 github.com/yudai/golcs v0.0.0-20150405163532-d1c525dea8ce/go.mod h1:lgjkn3NuSvDfVJdfcVVdX+jpBxNmX4rDAzaS45IcYoM=
 github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
 github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
-github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
-github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
 github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
 github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
 github.com/zeebo/errs v1.4.0 h1:XNdoD/RRMKP7HD0UhJnIzUy74ISdGGxURlYG8HSWSfM=

internal/data/account_contract_tokens.go

@@ -0,0 +1,109 @@
+// Package data provides data access layer for account-contract token mappings.
+// This file handles PostgreSQL storage of account-to-contract relationships.
+package data
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/jackc/pgx/v5"
+
+	"github.com/stellar/wallet-backend/internal/db"
+	"github.com/stellar/wallet-backend/internal/metrics"
+)
+
+// AccountContractTokensModelInterface defines the interface for account-contract mapping operations.
+type AccountContractTokensModelInterface interface {
+	// Read operations (for API/balances queries)
+	GetByAccount(ctx context.Context, accountAddress string) ([]*Contract, error)
+
+	// Write operations (for initial population and live ingestion)
+	BatchInsert(ctx context.Context, dbTx pgx.Tx, contractsByAccount map[string][]uuid.UUID) error
+}
+
+// AccountContractTokensModel implements AccountContractTokensModelInterface.
+type AccountContractTokensModel struct {
+	DB             db.ConnectionPool
+	MetricsService metrics.MetricsService
+}
+
+var _ AccountContractTokensModelInterface = (*AccountContractTokensModel)(nil)
+
+// GetByAccount retrieves all contract tokens for an account with full metadata via JOIN.
+func (m *AccountContractTokensModel) GetByAccount(ctx context.Context, accountAddress string) ([]*Contract, error) {
+	if accountAddress == "" {
+		return nil, fmt.Errorf("empty account address")
+	}
+
+	const query = `
+		SELECT ct.id, ct.contract_id, ct.type, ct.code, ct.issuer,
+		       ct.name, ct.symbol, ct.decimals, ct.created_at, ct.updated_at
+		FROM account_contract_tokens ac
+		INNER JOIN contract_tokens ct ON ct.id = ac.contract_id
+		WHERE ac.account_address = $1`
+
+	start := time.Now()
+	rows, err := m.DB.PgxPool().Query(ctx, query, accountAddress)
+	if err != nil {
+		m.MetricsService.IncDBQueryError("GetByAccount", "account_contract_tokens", "query_error")
+		return nil, fmt.Errorf("querying contracts for %s: %w", accountAddress, err)
+	}
+	defer rows.Close()
+
+	var contracts []*Contract
+	for rows.Next() {
+		var c Contract
+		if err := rows.Scan(&c.ID, &c.ContractID, &c.Type, &c.Code, &c.Issuer,
+			&c.Name, &c.Symbol, &c.Decimals, &c.CreatedAt, &c.UpdatedAt); err != nil {
+			return nil, fmt.Errorf("scanning contract: %w", err)
+		}
+		contracts = append(contracts, &c)
+	}
+
+	if err := rows.Err(); err != nil {
+		return nil, fmt.Errorf("iterating contracts: %w", err)
+	}
+
+	m.MetricsService.ObserveDBQueryDuration("GetByAccount", "account_contract_tokens", time.Since(start).Seconds())
+	m.MetricsService.IncDBQuery("GetByAccount", "account_contract_tokens")
+	return contracts, nil
+}
+
+// BatchInsert inserts contract IDs for multiple accounts (contracts are never removed).
+func (m *AccountContractTokensModel) BatchInsert(ctx context.Context, dbTx pgx.Tx, contractsByAccount map[string][]uuid.UUID) error {
+	if len(contractsByAccount) == 0 {
+		return nil
+	}
+
+	start := time.Now()
+
+	// Flatten to parallel arrays for UNNEST
+	var addresses []string
+	var contractIDs []uuid.UUID
+	for accountAddress, ids := range contractsByAccount {
+		for _, id := range ids {
+			addresses = append(addresses, accountAddress)
+			contractIDs = append(contractIDs, id)
+		}
+	}
+
+	if len(addresses) == 0 {
+		return nil
+	}
+
+	const query = `
+		INSERT INTO account_contract_tokens (account_address, contract_id)
+		SELECT unnest($1::text[]), unnest($2::uuid[])
+		ON CONFLICT DO NOTHING`
+
+	_, err := dbTx.Exec(ctx, query, addresses, contractIDs)
+	if err != nil {
+		return fmt.Errorf("batch inserting contract tokens: %w", err)
+	}
+
+	m.MetricsService.ObserveDBQueryDuration("BatchInsert", "account_contract_tokens", time.Since(start).Seconds())
+	m.MetricsService.IncDBQuery("BatchInsert", "account_contract_tokens")
+	return nil
+}