feat(blend): periodic SEP-40 oracle price snapshot with Comet BLND/LP leg

cmd/ingest.go

@@ -35,6 +35,8 @@ func (c *ingestCmd) Command() *cobra.Command {
 		utils.IngestServerPortOption(&cfg.ServerPort),
 		utils.AdminPortOption(&cfg.AdminPort),
 		utils.GetLedgersLimitOption(&cfg.GetLedgersLimit),
+		utils.BlendPriceIntervalOption(&cfg.BlendPriceInterval),
+		utils.BlendBackstopLPContractIDOption(&cfg.BlendBackstopLPContractID),
 		{
 			Name:        "ingestion-mode",
 			Usage:       "What mode to run ingestion in - live or backfill",

cmd/utils/global_options.go

@@ -153,6 +153,34 @@ func NetworkOption(configKey *string) *config.ConfigOption {
 	}
 }
 
+// BlendPriceIntervalOption returns the config option for the wait between Blend v2 oracle
+// price snapshot passes. Setting it to 0 disables the snapshot task entirely.
+func BlendPriceIntervalOption(configKey *time.Duration) *config.ConfigOption {
+	return &config.ConfigOption{
+		Name:           "blend-price-interval",
+		Usage:          `Interval between Blend v2 oracle price snapshot passes (Go duration string, e.g. "60s"). 0 disables the snapshot task.`,
+		OptType:        types.String,
+		CustomSetValue: SetConfigOptionDuration,
+		ConfigKey:      configKey,
+		FlagDefault:    "60s",
+		Required:       false,
+	}
+}
+
+// BlendBackstopLPContractIDOption returns the config option for the Blend v2 backstop's
+// Comet BLND:USDC weighted pool. Its C-address enables the price snapshot task's BLND/LP-share
+// derived-pricing leg; leaving it empty disables that leg.
+func BlendBackstopLPContractIDOption(configKey *string) *config.ConfigOption {
+	return &config.ConfigOption{
+		Name:        "blend-backstop-lp-contract-id",
+		Usage:       "C-address of the Blend v2 backstop's Comet BLND:USDC weighted pool, enabling the BLND/LP-share price leg. Empty disables it.",
+		OptType:     types.String,
+		ConfigKey:   configKey,
+		FlagDefault: "",
+		Required:    false,
+	}
+}
+
 func GraphQLComplexityLimitOption(configKey *int) *config.ConfigOption {
 	return &config.ConfigOption{
 		Name:        "graphql-complexity-limit",

docker-compose.yaml

@@ -175,6 +175,8 @@ services:
       TRACKER_DSN: ${TRACKER_DSN}
       STELLAR_ENVIRONMENT: ${STELLAR_ENVIRONMENT:-development}
       # ADMIN_PORT: 9094  # Optional: Port for pprof endpoints at /debug/pprof. WARNING: Do NOT expose publicly!
+      # BLEND_PRICE_INTERVAL: 60s  # Optional: interval between Blend v2 oracle price snapshot passes. 0 disables it.
+      # BLEND_BACKSTOP_LP_CONTRACT_ID: ""  # Optional: Comet BLND:USDC LP C-address enabling the BLND/LP price leg.
 
   ingest-mainnet:
     container_name: ingest-mainnet
@@ -217,6 +219,8 @@ services:
       TRACKER_DSN: ${TRACKER_DSN}
       STELLAR_ENVIRONMENT: ${STELLAR_ENVIRONMENT:-development}
       # ADMIN_PORT: 9095  # Optional: Port for pprof endpoints at /debug/pprof. WARNING: Do NOT expose publicly!
+      # BLEND_PRICE_INTERVAL: 60s  # Optional: interval between Blend v2 oracle price snapshot passes. 0 disables it.
+      # BLEND_BACKSTOP_LP_CONTRACT_ID: ""  # Optional: Comet BLND:USDC LP C-address enabling the BLND/LP price leg.
 volumes:
   postgres-db:
     driver: local

internal/data/blend/models.go

@@ -19,7 +19,7 @@ type Models struct {
 	PoolClaimed       *PoolClaimedModel
 	BackstopClaimed   *BackstopClaimedModel
 	Auctions          *AuctionModel
-	// OraclePrices *OraclePriceModel is added in PR4 alongside oracle_prices.go.
+	OraclePrices      *OraclePriceModel
 }
 
 // NewModels constructs the Blend Models aggregate wired to the given pool/metrics.
@@ -35,5 +35,6 @@ func NewModels(pool *pgxpool.Pool, dbMetrics *metrics.DBMetrics) Models {
 		PoolClaimed:       &PoolClaimedModel{DB: pool, Metrics: dbMetrics},
 		BackstopClaimed:   &BackstopClaimedModel{DB: pool, Metrics: dbMetrics},
 		Auctions:          &AuctionModel{DB: pool, Metrics: dbMetrics},
+		OraclePrices:      &OraclePriceModel{DB: pool, Metrics: dbMetrics},
 	}
 }

internal/data/blend/oracle_prices.go

@@ -0,0 +1,151 @@
+// Oracle price snapshot storage for Blend v2 (blend_oracle_prices).
+package blend
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	"github.com/stellar/wallet-backend/internal/indexer/types"
+	"github.com/stellar/wallet-backend/internal/metrics"
+	"github.com/stellar/wallet-backend/internal/utils"
+)
+
+// oraclePricesTable is the metrics label for blend_oracle_prices queries.
+const oraclePricesTable = "blend_oracle_prices"
+
+// PriceTarget identifies one (oracle, asset) pair that some pool's reserve
+// prices against — the input list the SEP-40 lastprice snapshot task queries
+// on each tick.
+type PriceTarget struct {
+	OracleContractID types.AddressBytea
+	AssetContractID  types.AddressBytea
+}
+
+// OraclePrice is a full row of blend_oracle_prices: the current-only (not
+// historical) latest known SEP-40 price for one (oracle, asset) pair.
+type OraclePrice struct {
+	OracleContractID types.AddressBytea
+	AssetContractID  types.AddressBytea
+	Price            string // raw i128 fixed-point value at PriceDecimals
+	PriceDecimals    int32
+	PriceTimestamp   int64 // oracle-reported timestamp
+	// UpdatedAt is populated when scanning an existing row. BatchUpsert ignores
+	// it on write and always sets updated_at to NOW() itself.
+	UpdatedAt time.Time
+}
+
+// OraclePriceModelInterface exposes Blend v2 oracle price snapshot storage
+// operations.
+//
+// Unlike sibling Blend models, both methods here operate directly on the
+// pgxpool rather than a caller-supplied pgx.Tx: the price snapshot task runs
+// on its own schedule, independent of ledger ingestion, so there is no
+// enclosing ingest transaction for it to join.
+type OraclePriceModelInterface interface {
+	// GetPriceTargets returns the deduplicated set of (oracle, asset) pairs
+	// that some pool's reserve prices against — every blend_reserves row
+	// joined to its pool's oracle. Pools with no oracle wired yet (NULL
+	// oracle_contract_id) are excluded.
+	GetPriceTargets(ctx context.Context) ([]PriceTarget, error)
+	// BatchUpsert inserts or fully replaces price rows keyed by
+	// (oracle_contract_id, asset_contract_id). updated_at is always set to
+	// NOW(), whether the row is freshly inserted or replaced.
+	BatchUpsert(ctx context.Context, rows []OraclePrice) error
+}
+
+// OraclePriceModel implements OraclePriceModelInterface against blend_oracle_prices.
+type OraclePriceModel struct {
+	DB      *pgxpool.Pool
+	Metrics *metrics.DBMetrics
+}
+
+var _ OraclePriceModelInterface = (*OraclePriceModel)(nil)
+
+// GetPriceTargets returns the deduplicated (oracle, asset) pairs derived from
+// blend_reserves joined to their pool's oracle. See OraclePriceModelInterface.
+func (m *OraclePriceModel) GetPriceTargets(ctx context.Context) ([]PriceTarget, error) {
+	start := time.Now()
+
+	const query = `
+		SELECT DISTINCT p.oracle_contract_id, r.asset_contract_id
+		FROM blend_pools p
+		JOIN blend_reserves r ON r.pool_contract_id = p.pool_contract_id
+		WHERE p.oracle_contract_id IS NOT NULL
+		ORDER BY p.oracle_contract_id, r.asset_contract_id`
+	rows, err := m.DB.Query(ctx, query)
+	if err != nil {
+		m.Metrics.QueryErrors.WithLabelValues("GetPriceTargets", oraclePricesTable, utils.GetDBErrorType(err)).Inc()
+		return nil, fmt.Errorf("querying blend oracle price targets: %w", err)
+	}
+	defer rows.Close()
+
+	var targets []PriceTarget
+	for rows.Next() {
+		var t PriceTarget
+		if err := rows.Scan(&t.OracleContractID, &t.AssetContractID); err != nil {
+			m.Metrics.QueryErrors.WithLabelValues("GetPriceTargets", oraclePricesTable, utils.GetDBErrorType(err)).Inc()
+			return nil, fmt.Errorf("scanning blend oracle price target row: %w", err)
+		}
+		targets = append(targets, t)
+	}
+	if err := rows.Err(); err != nil {
+		m.Metrics.QueryErrors.WithLabelValues("GetPriceTargets", oraclePricesTable, utils.GetDBErrorType(err)).Inc()
+		return nil, fmt.Errorf("iterating blend oracle price target rows: %w", err)
+	}
+
+	duration := time.Since(start).Seconds()
+	m.Metrics.QueryDuration.WithLabelValues("GetPriceTargets", oraclePricesTable).Observe(duration)
+	m.Metrics.QueriesTotal.WithLabelValues("GetPriceTargets", oraclePricesTable).Inc()
+	return targets, nil
+}
+
+// BatchUpsert inserts or fully replaces blend_oracle_prices rows. See OraclePriceModelInterface.
+func (m *OraclePriceModel) BatchUpsert(ctx context.Context, rows []OraclePrice) error {
+	if len(rows) == 0 {
+		return nil
+	}
+
+	start := time.Now()
+
+	oracles := make([][]byte, len(rows))
+	assets := make([][]byte, len(rows))
+	prices := make([]string, len(rows))
+	decimals := make([]int32, len(rows))
+	timestamps := make([]int64, len(rows))
+	for i, r := range rows {
+		oracleBytes, err := addressToBytes(string(r.OracleContractID))
+		if err != nil {
+			return fmt.Errorf("converting oracle address for price upsert: %w", err)
+		}
+		assetBytes, err := addressToBytes(string(r.AssetContractID))
+		if err != nil {
+			return fmt.Errorf("converting asset address for price upsert: %w", err)
+		}
+		oracles[i] = oracleBytes
+		assets[i] = assetBytes
+		prices[i] = r.Price
+		decimals[i] = r.PriceDecimals
+		timestamps[i] = r.PriceTimestamp
+	}
+
+	const upsertQuery = `
+		INSERT INTO blend_oracle_prices (oracle_contract_id, asset_contract_id, price, price_decimals, price_timestamp, updated_at)
+		SELECT u.*, NOW() FROM UNNEST($1::bytea[], $2::bytea[], $3::text[], $4::integer[], $5::bigint[])
+			AS u(oracle_contract_id, asset_contract_id, price, price_decimals, price_timestamp)
+		ON CONFLICT (oracle_contract_id, asset_contract_id) DO UPDATE SET
+			price = EXCLUDED.price, price_decimals = EXCLUDED.price_decimals,
+			price_timestamp = EXCLUDED.price_timestamp, updated_at = NOW()`
+	if _, err := m.DB.Exec(ctx, upsertQuery, oracles, assets, prices, decimals, timestamps); err != nil {
+		m.Metrics.QueryErrors.WithLabelValues("BatchUpsert", oraclePricesTable, utils.GetDBErrorType(err)).Inc()
+		return fmt.Errorf("upserting blend oracle prices: %w", err)
+	}
+
+	duration := time.Since(start).Seconds()
+	m.Metrics.QueryDuration.WithLabelValues("BatchUpsert", oraclePricesTable).Observe(duration)
+	m.Metrics.QueriesTotal.WithLabelValues("BatchUpsert", oraclePricesTable).Inc()
+	m.Metrics.BatchSize.WithLabelValues("BatchUpsert", oraclePricesTable).Observe(float64(len(rows)))
+	return nil
+}