Convert indexer tables to TimescaleDB hypertables

.github/workflows/go.yaml

@@ -87,12 +87,12 @@ jobs:
     runs-on: ubuntu-latest
     services:
       postgres:
-        image: postgres:12-alpine
+        image: timescale/timescaledb:2.25.0-pg17
         env:
           POSTGRES_USER: postgres
           POSTGRES_DB: postgres
           POSTGRES_PASSWORD: postgres
-          PGHOST: localhost
+          PGHOST: /var/run/postgresql
         options: >-
           --health-cmd pg_isready
           --health-interval 10s

.gitignore

@@ -9,3 +9,4 @@ __debug_bin*
 *.out
 CLAUDE.md
 .claude/
+.serena

cmd/ingest.go

@@ -140,6 +140,38 @@ func (c *ingestCmd) Command() *cobra.Command {
 			FlagDefault: true,
 			Required:    false,
 		},
+		{
+			Name:        "chunk-interval",
+			Usage:       "TimescaleDB chunk time interval for hypertables. Only affects future chunks. Uses PostgreSQL INTERVAL syntax.",
+			OptType:     types.String,
+			ConfigKey:   &cfg.ChunkInterval,
+			FlagDefault: "1 day",
+			Required:    false,
+		},
+		{
+			Name:        "retention-period",
+			Usage:       "TimescaleDB data retention period. Chunks older than this are automatically dropped. Empty disables retention. Uses PostgreSQL INTERVAL syntax.",
+			OptType:     types.String,
+			ConfigKey:   &cfg.RetentionPeriod,
+			FlagDefault: "",
+			Required:    false,
+		},
+		{
+			Name:        "compression-schedule-interval",
+			Usage:       "How frequently the TimescaleDB compression policy job checks for chunks to compress. Does not change which chunks are eligible (that's controlled by compress_after). Empty skips configuration. Uses PostgreSQL INTERVAL syntax.",
+			OptType:     types.String,
+			ConfigKey:   &cfg.CompressionScheduleInterval,
+			FlagDefault: "",
+			Required:    false,
+		},
+		{
+			Name:        "compression-compress-after",
+			Usage:       "How long after a chunk is closed before it becomes eligible for compression. Lower values reduce the number of uncompressed chunks. Empty skips configuration. Uses PostgreSQL INTERVAL syntax.",
+			OptType:     types.String,
+			ConfigKey:   &cfg.CompressAfter,
+			FlagDefault: "",
+			Required:    false,
+		},
 	}
 
 	cmd := &cobra.Command{

docker-compose.yaml

@@ -7,7 +7,8 @@ services:
 
   db:
     container_name: db
-    image: postgres:14-alpine
+    image: timescale/timescaledb:2.25.0-pg17
+    command: ["postgres", "-c", "timescaledb.enable_chunk_skipping=on", "-c", "timescaledb.enable_sparse_index_bloom=on"]
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -U postgres -d wallet-backend"]
       interval: 10s
@@ -268,5 +269,4 @@ volumes:
 configs:
   postgres_init:
     content: |
-      CREATE SCHEMA IF NOT EXISTS wallet_backend_mainnet;
-      CREATE SCHEMA IF NOT EXISTS wallet_backend_testnet;
+      CREATE EXTENSION IF NOT EXISTS timescaledb;

internal/data/accounts_test.go

@@ -273,7 +273,7 @@ func TestAccountModelBatchGetByToIDs(t *testing.T) {
 	require.NoError(t, err)
 
 	// Insert test transactions_accounts links
-	_, err = m.DB.ExecContext(ctx, "INSERT INTO transactions_accounts (tx_to_id, account_id) VALUES ($1, $2), ($3, $4)",
+	_, err = m.DB.ExecContext(ctx, "INSERT INTO transactions_accounts (ledger_created_at, tx_to_id, account_id) VALUES (NOW(), $1, $2), (NOW(), $3, $4)",
 		toID1, types.AddressBytea(address1), toID2, types.AddressBytea(address2))
 	require.NoError(t, err)
 
@@ -333,7 +333,7 @@ func TestAccountModelBatchGetByOperationIDs(t *testing.T) {
 	require.NoError(t, err)
 
 	// Insert test operations_accounts links (account_id is BYTEA)
-	_, err = m.DB.ExecContext(ctx, "INSERT INTO operations_accounts (operation_id, account_id) VALUES ($1, $2), ($3, $4)",
+	_, err = m.DB.ExecContext(ctx, "INSERT INTO operations_accounts (ledger_created_at, operation_id, account_id) VALUES (NOW(), $1, $2), (NOW(), $3, $4)",
 		operationID1, types.AddressBytea(address1), operationID2, types.AddressBytea(address2))
 	require.NoError(t, err)
 

internal/data/ingest_store.go

@@ -98,3 +98,18 @@ func (m *IngestStoreModel) GetLedgerGaps(ctx context.Context) ([]LedgerRange, er
 	m.MetricsService.IncDBQuery("GetLedgerGaps", "transactions")
 	return ledgerGaps, nil
 }
+
+func (m *IngestStoreModel) GetOldestLedger(ctx context.Context) (uint32, error) {
+	oldest := uint32(0)
+	start := time.Now()
+	err := m.DB.GetContext(ctx, &oldest,
+		`SELECT ledger_number FROM transactions ORDER BY ledger_created_at ASC, to_id ASC LIMIT 1`)
+	duration := time.Since(start).Seconds()
+	m.MetricsService.ObserveDBQueryDuration("GetOldestLedger", "transactions", duration)
+	if err != nil && !errors.Is(err, sql.ErrNoRows) {
+		m.MetricsService.IncDBQueryError("GetOldestLedger", "transactions", utils.GetDBErrorType(err))
+		return 0, fmt.Errorf("getting actual oldest ledger from transactions: %w", err)
+	}
+	m.MetricsService.IncDBQuery("GetOldestLedger", "transactions")
+	return oldest, nil
+}

internal/data/ingest_store_test.go

@@ -316,3 +316,65 @@ func Test_IngestStoreModel_GetLedgerGaps(t *testing.T) {
 		})
 	}
 }
+
+func Test_IngestStoreModel_GetOldestLedger(t *testing.T) {
+	dbt := dbtest.Open(t)
+	defer dbt.Close()
+	dbConnectionPool, err := db.OpenDBConnectionPool(dbt.DSN)
+	require.NoError(t, err)
+	defer dbConnectionPool.Close()
+
+	ctx := context.Background()
+
+	testCases := []struct {
+		name           string
+		setupDB        func(t *testing.T)
+		expectedLedger uint32
+	}{
+		{
+			name:           "returns_zero_when_table_is_empty",
+			expectedLedger: 0,
+		},
+		{
+			name: "returns_oldest_ledger_when_data_exists",
+			setupDB: func(t *testing.T) {
+				// Insert ledgers with distinct timestamps so ORDER BY ledger_created_at
+				// returns the correct oldest regardless of to_id ordering.
+				for i, ledger := range []uint32{150, 100, 200} {
+					_, err := dbConnectionPool.ExecContext(ctx,
+						`INSERT INTO transactions (hash, to_id, envelope_xdr, fee_charged, result_code, meta_xdr, ledger_number, ledger_created_at)
+						VALUES ($1, $2, 'env', 100, 'TransactionResultCodeTxSuccess', 'meta', $3, NOW() - INTERVAL '1 day' * $4)`,
+						fmt.Sprintf("hash%d", i), i+1, ledger, 300-int(ledger))
+					require.NoError(t, err)
+				}
+			},
+			expectedLedger: 100,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := dbConnectionPool.ExecContext(ctx, "DELETE FROM transactions")
+			require.NoError(t, err)
+
+			mockMetricsService := metrics.NewMockMetricsService()
+			mockMetricsService.
+				On("ObserveDBQueryDuration", "GetOldestLedger", "transactions", mock.Anything).Return().
+				On("IncDBQuery", "GetOldestLedger", "transactions").Return()
+			defer mockMetricsService.AssertExpectations(t)
+
+			m := &IngestStoreModel{
+				DB:             dbConnectionPool,
+				MetricsService: mockMetricsService,
+			}
+
+			if tc.setupDB != nil {
+				tc.setupDB(t)
+			}
+
+			oldest, err := m.GetOldestLedger(ctx)
+			require.NoError(t, err)
+			assert.Equal(t, tc.expectedLedger, oldest)
+		})
+	}
+}

internal/data/native_balances.go

@@ -129,23 +129,21 @@ func (m *NativeBalanceModel) BatchCopy(ctx context.Context, dbTx pgx.Tx, balance
 
 	start := time.Now()
 
-	rows := make([][]any, len(balances))
-	for i, nb := range balances {
-		rows[i] = []any{nb.AccountAddress, nb.Balance, nb.MinimumBalance, nb.BuyingLiabilities, nb.SellingLiabilities, nb.LedgerNumber}
-	}
-
 	copyCount, err := dbTx.CopyFrom(
 		ctx,
 		pgx.Identifier{"native_balances"},
 		[]string{"account_address", "balance", "minimum_balance", "buying_liabilities", "selling_liabilities", "last_modified_ledger"},
-		pgx.CopyFromRows(rows),
+		pgx.CopyFromSlice(len(balances), func(i int) ([]any, error) {
+			nb := balances[i]
+			return []any{nb.AccountAddress, nb.Balance, nb.MinimumBalance, nb.BuyingLiabilities, nb.SellingLiabilities, nb.LedgerNumber}, nil
+		}),
 	)
 	if err != nil {
 		return fmt.Errorf("bulk inserting native balances via COPY: %w", err)
 	}
 
-	if int(copyCount) != len(rows) {
-		return fmt.Errorf("expected %d rows copied, got %d", len(rows), copyCount)
+	if int(copyCount) != len(balances) {
+		return fmt.Errorf("expected %d rows copied, got %d", len(balances), copyCount)
 	}
 
 	m.MetricsService.ObserveDBQueryDuration("BatchCopy", "native_balances", time.Since(start).Seconds())

internal/data/operations.go

@@ -269,137 +269,13 @@ func (m *OperationModel) BatchGetByStateChangeIDs(ctx context.Context, scToIDs [
 	return operationsWithStateChanges, nil
 }
 
-// BatchInsert inserts the operations and the operations_accounts links.
-// It returns the IDs of the successfully inserted operations.
-func (m *OperationModel) BatchInsert(
-	ctx context.Context,
-	sqlExecuter db.SQLExecuter,
-	operations []*types.Operation,
-	stellarAddressesByOpID map[int64]set.Set[string],
-) ([]int64, error) {
-	if sqlExecuter == nil {
-		sqlExecuter = m.DB
-	}
-
-	// 1. Flatten the operations into parallel slices
-	ids := make([]int64, len(operations))
-	operationTypes := make([]string, len(operations))
-	operationXDRs := make([][]byte, len(operations))
-	resultCodes := make([]string, len(operations))
-	successfulFlags := make([]bool, len(operations))
-	ledgerNumbers := make([]uint32, len(operations))
-	ledgerCreatedAts := make([]time.Time, len(operations))
-
-	for i, op := range operations {
-		ids[i] = op.ID
-		operationTypes[i] = string(op.OperationType)
-		operationXDRs[i] = []byte(op.OperationXDR)
-		resultCodes[i] = op.ResultCode
-		successfulFlags[i] = op.Successful
-		ledgerNumbers[i] = op.LedgerNumber
-		ledgerCreatedAts[i] = op.LedgerCreatedAt
-	}
-
-	// 2. Flatten the stellarAddressesByOpID into parallel slices, converting to BYTEA
-	var opIDs []int64
-	var stellarAddressBytes [][]byte
-	for opID, addresses := range stellarAddressesByOpID {
-		for address := range addresses.Iter() {
-			opIDs = append(opIDs, opID)
-			addrBytesValue, err := types.AddressBytea(address).Value()
-			if err != nil {
-				return nil, fmt.Errorf("converting address %s to bytes: %w", address, err)
-			}
-			addrBytes, ok := addrBytesValue.([]byte)
-			if !ok || addrBytes == nil {
-				return nil, fmt.Errorf("converting address %s to bytes: unexpected value %T", address, addrBytesValue)
-			}
-			stellarAddressBytes = append(stellarAddressBytes, addrBytes)
-		}
-	}
-
-	// Insert operations and operations_accounts links.
-	const insertQuery = `
-	WITH
-	-- Insert operations
-	inserted_operations AS (
-		INSERT INTO operations
-			(id, operation_type, operation_xdr, result_code, successful, ledger_number, ledger_created_at)
-		SELECT
-			o.id, o.operation_type, o.operation_xdr, o.result_code, o.successful, o.ledger_number, o.ledger_created_at
-		FROM (
-			SELECT
-				UNNEST($1::bigint[]) AS id,
-				UNNEST($2::text[]) AS operation_type,
-				UNNEST($3::bytea[]) AS operation_xdr,
-				UNNEST($4::text[]) AS result_code,
-				UNNEST($5::boolean[]) AS successful,
-				UNNEST($6::bigint[]) AS ledger_number,
-				UNNEST($7::timestamptz[]) AS ledger_created_at
-		) o
-		ON CONFLICT (id) DO NOTHING
-		RETURNING id
-	),
-
-	-- Insert operations_accounts links
-	inserted_operations_accounts AS (
-		INSERT INTO operations_accounts
-			(operation_id, account_id)
-		SELECT
-			oa.op_id, oa.account_id
-		FROM (
-			SELECT
-				UNNEST($8::bigint[]) AS op_id,
-				UNNEST($9::bytea[]) AS account_id
-		) oa
-		ON CONFLICT DO NOTHING
-	)
-
-	-- Return the IDs of successfully inserted operations
-	SELECT id FROM inserted_operations;
-    `
-
-	start := time.Now()
-	var insertedIDs []int64
-	err := sqlExecuter.SelectContext(ctx, &insertedIDs, insertQuery,
-		pq.Array(ids),
-		pq.Array(operationTypes),
-		pq.Array(operationXDRs),
-		pq.Array(resultCodes),
-		pq.Array(successfulFlags),
-		pq.Array(ledgerNumbers),
-		pq.Array(ledgerCreatedAts),
-		pq.Array(opIDs),
-		pq.Array(stellarAddressBytes),
-	)
-	duration := time.Since(start).Seconds()
-	for _, dbTableName := range []string{"operations", "operations_accounts"} {
-		m.MetricsService.ObserveDBQueryDuration("BatchInsert", dbTableName, duration)
-		if dbTableName == "operations" {
-			m.MetricsService.ObserveDBBatchSize("BatchInsert", dbTableName, len(operations))
-		}
-		if err == nil {
-			m.MetricsService.IncDBQuery("BatchInsert", dbTableName)
-		}
-	}
-	if err != nil {
-		for _, dbTableName := range []string{"operations", "operations_accounts"} {
-			m.MetricsService.IncDBQueryError("BatchInsert", dbTableName, utils.GetDBErrorType(err))
-		}
-		return nil, fmt.Errorf("batch inserting operations and operations_accounts: %w", err)
-	}
-
-	return insertedIDs, nil
-}
-
 // BatchCopy inserts operations using pgx's binary COPY protocol.
 // Uses pgx.Tx for binary format which is faster than lib/pq's text format.
 // Uses native pgtype types for optimal performance (see https://github.com/jackc/pgx/issues/763).
 //
-// IMPORTANT: Unlike BatchInsert which uses ON CONFLICT DO NOTHING, BatchCopy will FAIL
-// if any duplicate records exist. The PostgreSQL COPY protocol does not support conflict
-// handling. Callers must ensure no duplicates exist before calling this method, or handle
-// the unique constraint violation error appropriately.
+// IMPORTANT: BatchCopy will FAIL if any duplicate records exist. The PostgreSQL COPY
+// protocol does not support conflict handling. Callers must ensure no duplicates exist
+// before calling this method, or handle the unique constraint violation error appropriately.
 func (m *OperationModel) BatchCopy(
 	ctx context.Context,
 	pgxTx pgx.Tx,
@@ -440,23 +316,34 @@ func (m *OperationModel) BatchCopy(
 
 	// COPY operations_accounts using pgx binary format with native pgtype types
 	if len(stellarAddressesByOpID) > 0 {
+		// Build OpID -> LedgerCreatedAt lookup from operations
+		ledgerCreatedAtByOpID := make(map[int64]time.Time, len(operations))
+		for _, op := range operations {
+			ledgerCreatedAtByOpID[op.ID] = op.LedgerCreatedAt
+		}
+
 		var oaRows [][]any
 		for opID, addresses := range stellarAddressesByOpID {
+			ledgerCreatedAt := ledgerCreatedAtByOpID[opID]
+			ledgerCreatedAtPgtype := pgtype.Timestamptz{Time: ledgerCreatedAt, Valid: true}
 			opIDPgtype := pgtype.Int8{Int64: opID, Valid: true}
 			for _, addr := range addresses.ToSlice() {
-				var addrBytes any
-				addrBytes, err = types.AddressBytea(addr).Value()
-				if err != nil {
-					return 0, fmt.Errorf("converting address %s to bytes: %w", addr, err)
+				addrBytes, addrErr := types.AddressBytea(addr).Value()
+				if addrErr != nil {
+					return 0, fmt.Errorf("converting address %s to bytes: %w", addr, addrErr)
 				}
-				oaRows = append(oaRows, []any{opIDPgtype, addrBytes})
+				oaRows = append(oaRows, []any{
+					ledgerCreatedAtPgtype,
+					opIDPgtype,
+					addrBytes,
+				})
 			}
 		}
 
 		_, err = pgxTx.CopyFrom(
 			ctx,
 			pgx.Identifier{"operations_accounts"},
-			[]string{"operation_id", "account_id"},
+			[]string{"ledger_created_at", "operation_id", "account_id"},
 			pgx.CopyFromRows(oaRows),
 		)
 		if err != nil {