From c4cf8f57f41f8149fc6552f974a732513ea69151 Mon Sep 17 00:00:00 2001 From: Marcus Pasell <3690498+rickyrombo@users.noreply.github.com> Date: Mon, 27 Jul 2026 22:47:10 -0700 Subject: [PATCH] feat(jobs): derive artist coin stats on-chain (shadow of Birdeye CoinStatsJob) Adds CoinStatsOnchainJob, which computes price, holder count, liquidity (pool TVL), market cap, all-time volume, and 24h price change from on-chain data (Meteora DBC/DAMM v2 pools, token-account balances, and an RPC getTokenSupply) instead of Birdeye. It writes a new shadow table artist_coin_stats_onchain and runs alongside the existing CoinStatsJob so the two can be compared via the artist_coin_stats_comparison view before cutover. Also adds artist_coin_price_history (24h change) and artist_coin_volume_accumulator (running volume) plus seed base rows for the new tables. AudioPriceJob (the AUDIO->USD anchor) is unchanged. Co-Authored-By: Claude Opus 4.8 --- database/seed.go | 44 ++ .../0229_artist_coin_stats_onchain.sql | 14 + .../0230_artist_coin_price_history.sql | 20 + .../0231_artist_coin_volume_accumulator.sql | 18 + ddl/views/artist_coin_stats_comparison.sql | 42 ++ jobs/coin_stats_onchain.go | 520 ++++++++++++++++++ jobs/coin_stats_onchain_test.go | 148 +++++ solana/indexer/solana_indexer.go | 8 + sql/01_schema.sql | 174 ++++++ sql/03_migration_tracker.sql | 4 + 10 files changed, 992 insertions(+) create mode 100644 ddl/migrations/0229_artist_coin_stats_onchain.sql create mode 100644 ddl/migrations/0230_artist_coin_price_history.sql create mode 100644 ddl/migrations/0231_artist_coin_volume_accumulator.sql create mode 100644 ddl/views/artist_coin_stats_comparison.sql create mode 100644 jobs/coin_stats_onchain.go create mode 100644 jobs/coin_stats_onchain_test.go diff --git a/database/seed.go b/database/seed.go index c6ead2ee..5f32b52c 100644 --- a/database/seed.go +++ b/database/seed.go @@ -524,6 +524,50 @@ var ( "created_at": time.Now(), "updated_at": time.Now(), }, + "artist_coin_stats_onchain": { + "mint": nil, + "created_at": time.Now(), + "updated_at": time.Now(), + }, + "artist_coin_price_history": { + "mint": nil, + "timestamp": nil, + "price": 0.0, + }, + "artist_coin_volume_accumulator": { + "mint": nil, + "last_processed_slot": 0, + "total_volume": 0.0, + "total_volume_usd": 0.0, + }, + "sol_meteora_dbc_pools": { + "account": nil, + "slot": 1, + "config": "config-placeholder", + "creator": "creator-placeholder", + "base_mint": nil, + "base_vault": "base-vault-placeholder", + "quote_vault": "quote-vault-placeholder", + "base_reserve": 0, + "quote_reserve": 0, + "protocol_base_fee": 0, + "protocol_quote_fee": 0, + "partner_base_fee": 0, + "partner_quote_fee": 0, + "sqrt_price": 0, + "activation_point": 0, + "pool_type": 0, + "is_migrated": 0, + "is_partner_withdraw_surplus": 0, + "is_protocol_withdraw_surplus": 0, + "migration_progress": 0, + "is_withdraw_leftover": 0, + "is_creator_withdraw_surplus": 0, + "migration_fee_withdraw_status": 0, + "finish_curve_timestamp": 0, + "creator_base_fee": 0, + "creator_quote_fee": 0, + }, "sol_token_account_balances": { "account": nil, "owner": "owner-acc", diff --git a/ddl/migrations/0229_artist_coin_stats_onchain.sql b/ddl/migrations/0229_artist_coin_stats_onchain.sql new file mode 100644 index 00000000..6704eb21 --- /dev/null +++ b/ddl/migrations/0229_artist_coin_stats_onchain.sql @@ -0,0 +1,14 @@ +-- Shadow table for the on-chain-derived replacement of CoinStatsJob. +-- Mirrors artist_coin_stats exactly so the on-chain job can be validated against +-- the Birdeye-populated table before cutover (and so cutover is a repoint/rename +-- with no consumer-struct changes). Populated by CoinStatsOnchainJob. +BEGIN; + +CREATE TABLE IF NOT EXISTS artist_coin_stats_onchain ( + LIKE artist_coin_stats INCLUDING ALL +); + +COMMENT ON TABLE artist_coin_stats_onchain IS + 'On-chain-derived shadow of artist_coin_stats, written by CoinStatsOnchainJob. Used to validate against the Birdeye-populated artist_coin_stats before replacing it.'; + +COMMIT; diff --git a/ddl/migrations/0230_artist_coin_price_history.sql b/ddl/migrations/0230_artist_coin_price_history.sql new file mode 100644 index 00000000..69ecc1f1 --- /dev/null +++ b/ddl/migrations/0230_artist_coin_price_history.sql @@ -0,0 +1,20 @@ +-- Hourly USD price snapshots per artist coin, used to compute the 24h price change +-- (history_24h_price / price_change_24h_percent) on-chain, replacing Birdeye. +-- Written by CoinStatsOnchainJob; kept small via hourly binning + 7-day retention. +BEGIN; + +CREATE TABLE IF NOT EXISTS artist_coin_price_history ( + mint TEXT NOT NULL, + timestamp TIMESTAMP NOT NULL, + price DOUBLE PRECISION NOT NULL, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (mint, timestamp) +); + +CREATE INDEX IF NOT EXISTS artist_coin_price_history_mint_ts_idx + ON artist_coin_price_history (mint, timestamp DESC); + +COMMENT ON TABLE artist_coin_price_history IS + 'Hourly USD price snapshots per artist coin, used to compute 24h price change.'; + +COMMIT; diff --git a/ddl/migrations/0231_artist_coin_volume_accumulator.sql b/ddl/migrations/0231_artist_coin_volume_accumulator.sql new file mode 100644 index 00000000..30af88b0 --- /dev/null +++ b/ddl/migrations/0231_artist_coin_volume_accumulator.sql @@ -0,0 +1,18 @@ +-- Running all-time trading-volume accumulator per artist coin, replacing Birdeye's +-- all-time stats. Each CoinStatsOnchainJob run adds only trades in new slots +-- (slot > last_processed_slot), valued in USD at that interval's AUDIO price, so +-- historical USD isn't repriced at today's rate. total_volume is AUDIO-denominated. +BEGIN; + +CREATE TABLE IF NOT EXISTS artist_coin_volume_accumulator ( + mint TEXT PRIMARY KEY, + last_processed_slot BIGINT NOT NULL DEFAULT 0, + total_volume DOUBLE PRECISION NOT NULL DEFAULT 0, + total_volume_usd DOUBLE PRECISION NOT NULL DEFAULT 0, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP +); + +COMMENT ON TABLE artist_coin_volume_accumulator IS + 'Running per-coin trading volume (AUDIO + USD) accumulated from pool quote-vault balance changes, watermarked by last_processed_slot. Written by CoinStatsOnchainJob.'; + +COMMIT; diff --git a/ddl/views/artist_coin_stats_comparison.sql b/ddl/views/artist_coin_stats_comparison.sql new file mode 100644 index 00000000..66578178 --- /dev/null +++ b/ddl/views/artist_coin_stats_comparison.sql @@ -0,0 +1,42 @@ +-- Side-by-side comparison of the Birdeye-populated artist_coin_stats against the +-- on-chain-derived artist_coin_stats_onchain, for validating CoinStatsOnchainJob +-- before cutover. `*_pct_diff` = (onchain - birdeye) / birdeye * 100. +-- A large diff (or NULL onchain where birdeye has a value) flags a missed pool, +-- external market, or migration lag to investigate. +DROP VIEW IF EXISTS artist_coin_stats_comparison; +CREATE VIEW artist_coin_stats_comparison AS + SELECT + ac.mint, + ac.ticker, + + b.price AS birdeye_price, + o.price AS onchain_price, + ROUND((((o.price - b.price) / NULLIF(b.price, 0)) * 100)::numeric, 2) AS price_pct_diff, + + b.market_cap AS birdeye_market_cap, + o.market_cap AS onchain_market_cap, + ROUND((((o.market_cap - b.market_cap) / NULLIF(b.market_cap, 0)) * 100)::numeric, 2) AS market_cap_pct_diff, + + b.holder AS birdeye_holder, + o.holder AS onchain_holder, + (o.holder - b.holder) AS holder_diff, + + b.liquidity AS birdeye_liquidity, + o.liquidity AS onchain_liquidity, + ROUND((((o.liquidity - b.liquidity) / NULLIF(b.liquidity, 0)) * 100)::numeric, 2) AS liquidity_pct_diff, + + b.total_volume_usd AS birdeye_total_volume_usd, + o.total_volume_usd AS onchain_total_volume_usd, + ROUND((((o.total_volume_usd - b.total_volume_usd) / NULLIF(b.total_volume_usd, 0)) * 100)::numeric, 2) AS total_volume_usd_pct_diff, + + b.price_change_24h_percent AS birdeye_price_change_24h_percent, + o.price_change_24h_percent AS onchain_price_change_24h_percent, + + o.updated_at AS onchain_updated_at, + b.updated_at AS birdeye_updated_at + FROM artist_coins ac + LEFT JOIN artist_coin_stats b ON b.mint = ac.mint + LEFT JOIN artist_coin_stats_onchain o ON o.mint = ac.mint; + +COMMENT ON VIEW artist_coin_stats_comparison IS + 'Compares Birdeye artist_coin_stats vs on-chain artist_coin_stats_onchain per coin (values + % diff) to validate CoinStatsOnchainJob before cutover.'; diff --git a/jobs/coin_stats_onchain.go b/jobs/coin_stats_onchain.go new file mode 100644 index 00000000..8812992e --- /dev/null +++ b/jobs/coin_stats_onchain.go @@ -0,0 +1,520 @@ +package jobs + +import ( + "context" + "fmt" + "strconv" + "sync" + "time" + + "api.audius.co/config" + "api.audius.co/database" + "api.audius.co/logging" + "github.com/gagliardetto/solana-go" + "github.com/gagliardetto/solana-go/rpc" + "github.com/jackc/pgx/v5" + "go.uber.org/zap" + "golang.org/x/sync/errgroup" +) + +// supplyFetchConcurrency bounds concurrent GetTokenSupply RPC calls. +const supplyFetchConcurrency = 8 + +// supplyFetchTimeout bounds a single GetTokenSupply RPC call. +const supplyFetchTimeout = 10 * time.Second + +// TokenSupplyFetcher is the subset of the Solana RPC client the job needs to +// read SPL mint supply. Abstracted so tests can inject a fake without a live RPC. +type TokenSupplyFetcher interface { + GetTokenSupply(ctx context.Context, tokenMint solana.PublicKey, commitment rpc.CommitmentType) (*rpc.GetTokenSupplyResult, error) +} + +// CoinStatsOnchainJob derives artist-coin market stats from on-chain data +// (Meteora DBC/DAMM v2 pools, token-account balances, mint supply) and writes +// them to the artist_coin_stats_onchain shadow table. It is the on-chain +// replacement for the Birdeye-backed CoinStatsJob; both run side by side until +// the on-chain values are validated (see artist_coin_stats_comparison). +type CoinStatsOnchainJob struct { + pool database.DbPool + rpcClient TokenSupplyFetcher + audioMint string + logger *zap.Logger + + mutex sync.Mutex + isRunning bool +} + +func NewCoinStatsOnchainJob(cfg config.Config, pool database.DbPool) *CoinStatsOnchainJob { + logger := logging.NewZapLogger(cfg).Named("CoinStatsOnchainJob") + + var rpcClient TokenSupplyFetcher + if len(cfg.SolanaConfig.RpcProviders) > 0 { + rpcClient = rpc.New(cfg.SolanaConfig.RpcProviders[0]) + } + + return &CoinStatsOnchainJob{ + pool: pool, + rpcClient: rpcClient, + audioMint: cfg.SolanaConfig.MintAudio.String(), + logger: logger, + } +} + +// ScheduleEvery runs the job every `duration` until the context is cancelled. +func (j *CoinStatsOnchainJob) ScheduleEvery(ctx context.Context, duration time.Duration) *CoinStatsOnchainJob { + go func() { + ticker := time.NewTicker(duration) + defer ticker.Stop() + for { + select { + case <-ticker.C: + j.logger.Info("Job started") + j.Run(ctx) + case <-ctx.Done(): + j.logger.Info("Job shutting down") + return + } + } + }() + return j +} + +// Run executes the job once +func (j *CoinStatsOnchainJob) Run(ctx context.Context) { + if err := j.run(ctx); err != nil { + j.logger.Error("Job run failed", zap.Error(err)) + } else { + j.logger.Info("Job completed successfully") + } +} + +type coinAgg struct { + Mint string `db:"mint"` + Price *float64 `db:"price"` + Holder int64 `db:"holder"` + Liquidity float64 `db:"liquidity"` +} + +type volumeRow struct { + Mint string `db:"mint"` + TotalVolume float64 `db:"total_volume"` + TotalVolumeUsd float64 `db:"total_volume_usd"` +} + +type priceRow struct { + Mint string `db:"mint"` + Price float64 `db:"price"` +} + +func (j *CoinStatsOnchainJob) run(ctx context.Context) error { + j.mutex.Lock() + if j.isRunning { + j.mutex.Unlock() + return fmt.Errorf("job is already running") + } + j.isRunning = true + j.mutex.Unlock() + defer func() { + j.mutex.Lock() + j.isRunning = false + j.mutex.Unlock() + }() + + now := time.Now() + + // AUDIO USD anchor (maintained by AudioPriceJob). Used to convert all + // AUDIO-denominated reserves/volume to USD. + audioPrice, err := j.getAudioPrice(ctx) + if err != nil { + return fmt.Errorf("error reading AUDIO price: %w", err) + } + if audioPrice == 0 { + j.logger.Warn("AUDIO price is 0; USD-denominated stats will be 0 this run") + } + + // 1. Set-based aggregates: on-chain price, holder count, liquidity (TVL). + aggs, err := j.queryAggregates(ctx, audioPrice) + if err != nil { + return fmt.Errorf("error computing aggregates: %w", err) + } + + // 2. Advance the all-time volume accumulator with trades in new slots. + if err := j.updateVolumeAccumulator(ctx, audioPrice); err != nil { + return fmt.Errorf("error updating volume accumulator: %w", err) + } + volumes, err := j.readVolumeAccumulator(ctx) + if err != nil { + return fmt.Errorf("error reading volume accumulator: %w", err) + } + + // 3. Snapshot current on-chain prices (hourly bin) and read the ~24h-ago price. + if err := j.snapshotPrices(ctx, now); err != nil { + return fmt.Errorf("error snapshotting prices: %w", err) + } + history24h, err := j.read24hAgoPrices(ctx, now) + if err != nil { + return fmt.Errorf("error reading 24h price history: %w", err) + } + + // 4. Mint supply via RPC (bounded concurrency, skip-on-error). + mints := make([]string, 0, len(aggs)) + for mint := range aggs { + mints = append(mints, mint) + } + supplies := j.fetchSupplies(ctx, mints) + + // 5. Upsert per coin. + for mint, agg := range aggs { + // AUDIO has no artist pool; its on-chain price is NULL. Use the anchor + // price for market cap and never write NULL over its maintained price. + effectivePrice := agg.Price + if mint == j.audioMint { + ap := audioPrice + effectivePrice = &ap + } + + var totalSupply *float64 + if s, ok := supplies[mint]; ok { + totalSupply = &s + } + + var marketCap *float64 + if effectivePrice != nil && totalSupply != nil { + mc := *effectivePrice * *totalSupply + marketCap = &mc + } + + var history *float64 + var priceChange *float64 + if h, ok := history24h[mint]; ok && h > 0 { + history = &h + if agg.Price != nil { + pc := ((*agg.Price - h) / h) * 100 + priceChange = &pc + } + } + + vol := volumes[mint] // zero value if absent + + if err := j.upsertStats(ctx, mint, agg, marketCap, totalSupply, vol, history, priceChange); err != nil { + j.logger.Error("error upserting onchain stats", zap.String("mint", mint), zap.Error(err)) + } + } + + // 6. Retention: keep the price-history table small. + if _, err := j.pool.Exec(ctx, + `DELETE FROM artist_coin_price_history WHERE timestamp < $1`, + now.Add(-7*24*time.Hour), + ); err != nil { + j.logger.Error("error pruning price history", zap.Error(err)) + } + + return nil +} + +func (j *CoinStatsOnchainJob) getAudioPrice(ctx context.Context) (float64, error) { + var price *float64 + err := j.pool.QueryRow(ctx, + `SELECT price FROM artist_coin_stats WHERE mint = $1`, + j.audioMint, + ).Scan(&price) + if err != nil { + if err == pgx.ErrNoRows { + return 0, nil + } + return 0, err + } + if price == nil { + return 0, nil + } + return *price, nil +} + +// queryAggregates computes per-coin on-chain price, holder count, and liquidity +// (pool TVL = base-side USD + quote-side USD) for every artist coin. +func (j *CoinStatsOnchainJob) queryAggregates(ctx context.Context, audioPrice float64) (map[string]coinAgg, error) { + sql := ` + WITH onchain_price AS ( + SELECT mint, COALESCE(damm_v2_price, dbc_price, pools_price_usd) AS price + FROM artist_coin_prices + ), + holders AS ( + SELECT mint, COUNT(DISTINCT owner) AS holder + FROM sol_token_account_balances + WHERE balance > 0 + GROUP BY mint + ), + liq_dbc AS ( + -- Active bonding curve: value both real vault reserves in USD. + SELECT + p.base_mint AS mint, + (p.base_reserve::numeric / POWER(10, ac.decimals)) * COALESCE(op.price, 0) + + (p.quote_reserve::numeric / POWER(10, 8)) * @audio_price AS liquidity + FROM sol_meteora_dbc_pools p + JOIN artist_coins ac ON ac.mint = p.base_mint + LEFT JOIN onchain_price op ON op.mint = p.base_mint + WHERE p.is_migrated = 0 + ), + liq_damm AS ( + -- Migrated pool: value both vault balances in USD (token_a=coin, token_b=AUDIO). + SELECT + p.token_a_mint AS mint, + (COALESCE(ba.balance, 0)::numeric / POWER(10, ac.decimals)) * COALESCE(op.price, 0) + + (COALESCE(bb.balance, 0)::numeric / POWER(10, 8)) * @audio_price AS liquidity + FROM sol_meteora_damm_v2_pools p + JOIN artist_coins ac ON ac.mint = p.token_a_mint + LEFT JOIN onchain_price op ON op.mint = p.token_a_mint + LEFT JOIN sol_token_account_balances ba ON ba.account = p.token_a_vault + LEFT JOIN sol_token_account_balances bb ON bb.account = p.token_b_vault + ), + liquidity AS ( + SELECT mint, SUM(liquidity) AS liquidity + FROM (SELECT * FROM liq_dbc UNION ALL SELECT * FROM liq_damm) x + GROUP BY mint + ) + SELECT + ac.mint, + op.price::double precision AS price, + COALESCE(h.holder, 0) AS holder, + COALESCE(l.liquidity, 0)::double precision AS liquidity + FROM artist_coins ac + LEFT JOIN onchain_price op ON op.mint = ac.mint + LEFT JOIN holders h ON h.mint = ac.mint + LEFT JOIN liquidity l ON l.mint = ac.mint + ` + rows, err := j.pool.Query(ctx, sql, pgx.NamedArgs{"audio_price": audioPrice}) + if err != nil { + return nil, err + } + list, err := pgx.CollectRows(rows, pgx.RowToStructByName[coinAgg]) + if err != nil { + return nil, err + } + out := make(map[string]coinAgg, len(list)) + for _, r := range list { + out[r.Mint] = r + } + return out, nil +} + +// updateVolumeAccumulator adds trading volume from balance changes in new slots +// (per-coin watermark) to the running accumulator, valued in USD at the current +// AUDIO price so historical USD isn't repriced at today's rate. +func (j *CoinStatsOnchainJob) updateVolumeAccumulator(ctx context.Context, audioPrice float64) error { + sql := ` + WITH vaults AS ( + SELECT base_mint AS mint, quote_vault AS vault FROM sol_meteora_dbc_pools + UNION + SELECT token_a_mint AS mint, token_b_vault AS vault FROM sol_meteora_damm_v2_pools + ), + delta AS ( + SELECT + v.mint, + SUM(ABS(c.change))::numeric / POWER(10, 8) AS vol_audio, + MAX(c.slot) AS max_slot + FROM vaults v + JOIN sol_token_account_balance_changes c + ON c.account = v.vault AND c.mint = @audio_mint + WHERE c.slot > COALESCE( + (SELECT last_processed_slot FROM artist_coin_volume_accumulator acc WHERE acc.mint = v.mint), + 0 + ) + GROUP BY v.mint + ) + INSERT INTO artist_coin_volume_accumulator (mint, last_processed_slot, total_volume, total_volume_usd, updated_at) + SELECT d.mint, d.max_slot, d.vol_audio, d.vol_audio * @audio_price, NOW() + FROM delta d + WHERE d.max_slot IS NOT NULL + ON CONFLICT (mint) DO UPDATE SET + total_volume = artist_coin_volume_accumulator.total_volume + EXCLUDED.total_volume, + total_volume_usd = artist_coin_volume_accumulator.total_volume_usd + EXCLUDED.total_volume_usd, + last_processed_slot = EXCLUDED.last_processed_slot, + updated_at = NOW() + ` + _, err := j.pool.Exec(ctx, sql, pgx.NamedArgs{ + "audio_mint": j.audioMint, + "audio_price": audioPrice, + }) + return err +} + +func (j *CoinStatsOnchainJob) readVolumeAccumulator(ctx context.Context) (map[string]volumeRow, error) { + rows, err := j.pool.Query(ctx, + `SELECT mint, total_volume, total_volume_usd FROM artist_coin_volume_accumulator`, + ) + if err != nil { + return nil, err + } + list, err := pgx.CollectRows(rows, pgx.RowToStructByName[volumeRow]) + if err != nil { + return nil, err + } + out := make(map[string]volumeRow, len(list)) + for _, r := range list { + out[r.Mint] = r + } + return out, nil +} + +// snapshotPrices records the current on-chain USD price for each coin in an +// hourly bin, used to compute the 24h price change. +func (j *CoinStatsOnchainJob) snapshotPrices(ctx context.Context, now time.Time) error { + sql := ` + INSERT INTO artist_coin_price_history (mint, timestamp, price) + SELECT + mint, + date_trunc('hour', @now::timestamp), + COALESCE(damm_v2_price, dbc_price, pools_price_usd) + FROM artist_coin_prices + WHERE COALESCE(damm_v2_price, dbc_price, pools_price_usd) IS NOT NULL + ON CONFLICT (mint, timestamp) DO UPDATE SET price = EXCLUDED.price + ` + _, err := j.pool.Exec(ctx, sql, pgx.NamedArgs{"now": now}) + return err +} + +// read24hAgoPrices returns, per coin, the most recent price snapshot that is at +// least 24h old (and no older than 30h, to avoid stale baselines across gaps). +func (j *CoinStatsOnchainJob) read24hAgoPrices(ctx context.Context, now time.Time) (map[string]float64, error) { + sql := ` + SELECT DISTINCT ON (mint) mint, price + FROM artist_coin_price_history + WHERE timestamp <= @cutoff AND timestamp >= @floor + ORDER BY mint, timestamp DESC + ` + rows, err := j.pool.Query(ctx, sql, pgx.NamedArgs{ + "cutoff": now.Add(-24 * time.Hour), + "floor": now.Add(-30 * time.Hour), + }) + if err != nil { + return nil, err + } + list, err := pgx.CollectRows(rows, pgx.RowToStructByName[priceRow]) + if err != nil { + return nil, err + } + out := make(map[string]float64, len(list)) + for _, r := range list { + out[r.Mint] = r.Price + } + return out, nil +} + +// fetchSupplies reads SPL mint total supply (human units) per mint over RPC with +// bounded concurrency. Mints that error are omitted (caller keeps last known). +func (j *CoinStatsOnchainJob) fetchSupplies(ctx context.Context, mints []string) map[string]float64 { + out := make(map[string]float64, len(mints)) + if j.rpcClient == nil { + j.logger.Warn("no RPC client configured; skipping supply fetch") + return out + } + + var mu sync.Mutex + g, gctx := errgroup.WithContext(ctx) + g.SetLimit(supplyFetchConcurrency) + + for _, mint := range mints { + g.Go(func() error { + pubkey, err := solana.PublicKeyFromBase58(mint) + if err != nil { + j.logger.Error("invalid mint address", zap.String("mint", mint), zap.Error(err)) + return nil + } + callCtx, cancel := context.WithTimeout(gctx, supplyFetchTimeout) + defer cancel() + + res, err := j.rpcClient.GetTokenSupply(callCtx, pubkey, rpc.CommitmentConfirmed) + if err != nil { + j.logger.Warn("failed to fetch token supply", zap.String("mint", mint), zap.Error(err)) + return nil + } + supply, ok := uiAmount(res) + if !ok { + return nil + } + mu.Lock() + out[mint] = supply + mu.Unlock() + return nil + }) + } + // Errors are swallowed per-mint; Wait returns nil. + _ = g.Wait() + return out +} + +// uiAmount extracts the human-unit supply from a GetTokenSupply result. +func uiAmount(res *rpc.GetTokenSupplyResult) (float64, bool) { + if res == nil || res.Value == nil { + return 0, false + } + if res.Value.UiAmountString != "" { + if v, err := strconv.ParseFloat(res.Value.UiAmountString, 64); err == nil { + return v, true + } + } + if res.Value.UiAmount != nil { + return *res.Value.UiAmount, true + } + // Fall back to raw amount / 10^decimals. + if raw, err := strconv.ParseFloat(res.Value.Amount, 64); err == nil { + return raw / pow10(int(res.Value.Decimals)), true + } + return 0, false +} + +func pow10(n int) float64 { + v := 1.0 + for range n { + v *= 10 + } + return v +} + +func (j *CoinStatsOnchainJob) upsertStats( + ctx context.Context, + mint string, + agg coinAgg, + marketCap *float64, + totalSupply *float64, + vol volumeRow, + history24h *float64, + priceChange *float64, +) error { + sql := ` + INSERT INTO artist_coin_stats_onchain ( + mint, price, market_cap, liquidity, holder, total_supply, + total_volume, total_volume_usd, history_24h_price, price_change_24h_percent, + created_at, updated_at + ) VALUES ( + @mint, @price, @market_cap, @liquidity, @holder, @total_supply, + @total_volume, @total_volume_usd, @history_24h_price, @price_change_24h_percent, + NOW(), NOW() + ) + ON CONFLICT (mint) DO UPDATE SET + -- Never overwrite a maintained price with NULL (e.g. AUDIO / pool-less coins). + price = COALESCE(EXCLUDED.price, artist_coin_stats_onchain.price), + market_cap = COALESCE(EXCLUDED.market_cap, artist_coin_stats_onchain.market_cap), + liquidity = EXCLUDED.liquidity, + holder = EXCLUDED.holder, + total_supply = COALESCE(EXCLUDED.total_supply, artist_coin_stats_onchain.total_supply), + total_volume = EXCLUDED.total_volume, + total_volume_usd = EXCLUDED.total_volume_usd, + history_24h_price = COALESCE(EXCLUDED.history_24h_price, artist_coin_stats_onchain.history_24h_price), + price_change_24h_percent = COALESCE(EXCLUDED.price_change_24h_percent, artist_coin_stats_onchain.price_change_24h_percent), + updated_at = NOW() + ` + _, err := j.pool.Exec(ctx, sql, pgx.NamedArgs{ + "mint": mint, + "price": agg.Price, + "market_cap": marketCap, + "liquidity": agg.Liquidity, + "holder": agg.Holder, + "total_supply": totalSupply, + "total_volume": vol.TotalVolume, + "total_volume_usd": vol.TotalVolumeUsd, + "history_24h_price": history24h, + "price_change_24h_percent": priceChange, + }) + return err +} diff --git a/jobs/coin_stats_onchain_test.go b/jobs/coin_stats_onchain_test.go new file mode 100644 index 00000000..e237db8f --- /dev/null +++ b/jobs/coin_stats_onchain_test.go @@ -0,0 +1,148 @@ +package jobs + +import ( + "context" + "fmt" + "strconv" + "testing" + "time" + + "api.audius.co/database" + "github.com/gagliardetto/solana-go" + "github.com/gagliardetto/solana-go/rpc" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.uber.org/zap" +) + +// fakeSupplyFetcher returns canned SPL mint supplies (human units) keyed by mint. +type fakeSupplyFetcher struct { + supplies map[string]float64 +} + +func (f fakeSupplyFetcher) GetTokenSupply( + _ context.Context, + mint solana.PublicKey, + _ rpc.CommitmentType, +) (*rpc.GetTokenSupplyResult, error) { + v, ok := f.supplies[mint.String()] + if !ok { + return nil, fmt.Errorf("no supply for %s", mint) + } + return &rpc.GetTokenSupplyResult{ + Value: &rpc.UiTokenAmount{ + Amount: "0", + Decimals: 6, + UiAmountString: strconv.FormatFloat(v, 'f', -1, 64), + }, + }, nil +} + +func TestCoinStatsOnchainJob(t *testing.T) { + pool := database.CreateTestDatabase(t, "test_jobs") + defer pool.Close() + + ctx := context.Background() + cfg := newTestConfig() + audioMint := cfg.SolanaConfig.MintAudio.String() + + // A valid base58 SPL mint for the artist coin under test (GetTokenSupply parses it). + const coinMint = "So11111111111111111111111111111111111111112" + const quoteVault = "QVau1t1111111111111111111111111111111111111" + + // ~25h-old snapshot timestamp, within the job's [now-30h, now-24h] lookup window. + snapshot24hAgo := time.Now().Add(-25 * time.Hour) + + fixtures := database.FixtureMap{ + "users": { + {"user_id": 1, "wallet": "0x1234567890123456789012345678901234567890"}, + }, + "artist_coins": { + {"ticker": "COIN1", "decimals": 6, "user_id": 1, "mint": coinMint}, + {"ticker": "AUDIO", "decimals": 8, "user_id": 1, "mint": audioMint}, + }, + // AUDIO USD anchor (normally maintained by AudioPriceJob). + "artist_coin_stats": { + {"mint": audioMint, "price": 2.0}, + }, + // On-chain USD price for the coin comes via artist_coin_prices -> pools_price_usd. + "artist_coin_pools": { + {"address": "pool1", "base_mint": coinMint, "price_usd": 4.0}, + }, + // Holders: 3 distinct owners with balance > 0, one with 0 (excluded). + "sol_token_account_balances": { + {"account": "acct1", "mint": coinMint, "owner": "owner1", "balance": 10, "slot": 1}, + {"account": "acct2", "mint": coinMint, "owner": "owner2", "balance": 5, "slot": 1}, + {"account": "acct3", "mint": coinMint, "owner": "owner3", "balance": 1, "slot": 1}, + {"account": "acct4", "mint": coinMint, "owner": "owner4", "balance": 0, "slot": 1}, + }, + // Volume through the DBC quote vault (AUDIO leg): |1e8| + |-2e8| = 3e8 -> 3 AUDIO. + "sol_token_account_balance_changes": { + {"signature": "sig1", "mint": audioMint, "owner": "trader1", "account": quoteVault, + "change": 100_000_000, "balance": 100_000_000, "slot": 10, "block_timestamp": "2024-01-01 00:00:00"}, + {"signature": "sig2", "mint": audioMint, "owner": "trader2", "account": quoteVault, + "change": -200_000_000, "balance": 0, "slot": 11, "block_timestamp": "2024-01-01 00:01:00"}, + }, + // Active DBC pool -> liquidity = base_reserve*price + quote_reserve*audioPrice + // = (1e9/1e6)*4.0 + (5e10/1e8)*2.0 = 1000*4 + 500*2 = 5000. + "sol_meteora_dbc_pools": { + {"account": "dbcpool1", "base_mint": coinMint, "quote_vault": quoteVault, + "base_reserve": 1_000_000_000, "quote_reserve": 50_000_000_000, "is_migrated": 0}, + }, + // Pre-existing on-chain AUDIO row to verify its price is not clobbered. + "artist_coin_stats_onchain": { + {"mint": audioMint, "price": 2.0}, + }, + // A ~25h-old price snapshot so the 24h change is computable: (4.0-2.0)/2.0*100 = 100%. + "artist_coin_price_history": { + {"mint": coinMint, "timestamp": snapshot24hAgo, "price": 2.0}, + }, + } + database.Seed(pool, fixtures) + + job := &CoinStatsOnchainJob{ + pool: pool, + rpcClient: fakeSupplyFetcher{supplies: map[string]float64{coinMint: 1000, audioMint: 1_000_000}}, + audioMint: audioMint, + logger: zap.NewNop(), + } + + require.NoError(t, job.run(ctx)) + + // Assert the coin's derived stats. + var ( + price float64 + holder int + liquidity float64 + totalSupply float64 + marketCap float64 + totalVolume float64 + totalVolUSD float64 + history24h float64 + priceChange24 float64 + ) + err := pool.QueryRow(ctx, ` + SELECT price, holder, liquidity, total_supply, market_cap, + total_volume, total_volume_usd, history_24h_price, price_change_24h_percent + FROM artist_coin_stats_onchain WHERE mint = $1`, coinMint). + Scan(&price, &holder, &liquidity, &totalSupply, &marketCap, + &totalVolume, &totalVolUSD, &history24h, &priceChange24) + require.NoError(t, err) + + assert.InDelta(t, 4.0, price, 1e-9, "price from pools_price_usd") + assert.Equal(t, 3, holder, "distinct owners with balance > 0") + assert.InDelta(t, 5000.0, liquidity, 1e-6, "TVL = base_usd + quote_usd") + assert.InDelta(t, 1000.0, totalSupply, 1e-9, "supply from RPC") + assert.InDelta(t, 4000.0, marketCap, 1e-6, "price * supply") + assert.InDelta(t, 3.0, totalVolume, 1e-9, "AUDIO volume through vault") + assert.InDelta(t, 6.0, totalVolUSD, 1e-9, "volume * audioPrice") + assert.InDelta(t, 2.0, history24h, 1e-9, "24h-ago snapshot") + assert.InDelta(t, 100.0, priceChange24, 1e-6, "24h percent change") + + // AUDIO's maintained price must not be clobbered (on-chain price is NULL for it). + var audioPrice float64 + err = pool.QueryRow(ctx, + `SELECT price FROM artist_coin_stats_onchain WHERE mint = $1`, audioMint).Scan(&audioPrice) + require.NoError(t, err) + assert.InDelta(t, 2.0, audioPrice, 1e-9, "AUDIO price preserved") +} diff --git a/solana/indexer/solana_indexer.go b/solana/indexer/solana_indexer.go index 59c25f06..4df0cf56 100644 --- a/solana/indexer/solana_indexer.go +++ b/solana/indexer/solana_indexer.go @@ -123,6 +123,14 @@ func (s *SolanaIndexer) Start(ctx context.Context) error { statsJob.ScheduleEvery(statsCtx, 15*time.Minute) go statsJob.Run(statsCtx) + // On-chain replacement for CoinStatsJob, running in shadow: it writes + // artist_coin_stats_onchain so its values can be validated against Birdeye + // (see artist_coin_stats_comparison) before cutover. + statsOnchainJob := jobs.NewCoinStatsOnchainJob(s.config, s.pool) + statsOnchainCtx := context.WithoutCancel(ctx) + statsOnchainJob.ScheduleEvery(statsOnchainCtx, 15*time.Minute) + go statsOnchainJob.Run(statsOnchainCtx) + audioPriceJob := jobs.NewAudioPriceJob(s.config, s.pool) priceCtx := context.WithoutCancel(ctx) audioPriceJob.ScheduleEvery(priceCtx, 5*time.Minute) diff --git a/sql/01_schema.sql b/sql/01_schema.sql index 3e656dc9..cdc8008f 100644 --- a/sql/01_schema.sql +++ b/sql/01_schema.sql @@ -7348,6 +7348,25 @@ CREATE TABLE public.artist_coin_pools ( ); +-- +-- Name: artist_coin_price_history; Type: TABLE; Schema: public; Owner: - +-- + +CREATE TABLE public.artist_coin_price_history ( + mint text NOT NULL, + "timestamp" timestamp without time zone NOT NULL, + price double precision NOT NULL, + created_at timestamp without time zone DEFAULT CURRENT_TIMESTAMP NOT NULL +); + + +-- +-- Name: TABLE artist_coin_price_history; Type: COMMENT; Schema: public; Owner: - +-- + +COMMENT ON TABLE public.artist_coin_price_history IS 'Hourly USD price snapshots per artist coin, used to compute 24h price change.'; + + -- -- Name: artist_coin_stats; Type: TABLE; Schema: public; Owner: - -- @@ -7612,6 +7631,130 @@ CREATE VIEW public.artist_coin_prices AS COMMENT ON VIEW public.artist_coin_prices IS 'View that provides artist coin prices using DAMM V2 pool if available, DBC pools if not and still applicable, artist_coin_pools.price_usd as fallback, and artist_coin_stats.price as final fallback (primarily for AUDIO and other tokens without pools). Makes use of the price of the quote token (AUDIO) from Birdeye if using a pool.'; +-- +-- Name: artist_coin_stats_onchain; Type: TABLE; Schema: public; Owner: - +-- + +CREATE TABLE public.artist_coin_stats_onchain ( + mint text NOT NULL, + market_cap double precision, + fdv double precision, + liquidity double precision, + last_trade_unix_time bigint, + last_trade_human_time text, + price double precision, + history_24h_price double precision, + price_change_24h_percent double precision, + unique_wallet_24h integer, + unique_wallet_history_24h integer, + unique_wallet_24h_change_percent double precision, + total_supply double precision, + circulating_supply double precision, + holder integer, + trade_24h integer, + trade_history_24h integer, + trade_24h_change_percent double precision, + sell_24h integer, + sell_history_24h integer, + sell_24h_change_percent double precision, + buy_24h integer, + buy_history_24h integer, + buy_24h_change_percent double precision, + v_24h double precision, + v_24h_usd double precision, + v_history_24h double precision, + v_history_24h_usd double precision, + v_24h_change_percent double precision, + v_buy_24h double precision, + v_buy_24h_usd double precision, + v_buy_history_24h double precision, + v_buy_history_24h_usd double precision, + v_buy_24h_change_percent double precision, + v_sell_24h double precision, + v_sell_24h_usd double precision, + v_sell_history_24h double precision, + v_sell_history_24h_usd double precision, + v_sell_24h_change_percent double precision, + number_markets integer, + created_at timestamp without time zone DEFAULT CURRENT_TIMESTAMP NOT NULL, + updated_at timestamp without time zone DEFAULT CURRENT_TIMESTAMP NOT NULL, + total_volume double precision, + total_volume_usd double precision, + volume_buy double precision, + volume_buy_usd double precision, + volume_sell double precision, + volume_sell_usd double precision, + buy integer, + sell integer, + total_trade integer +); + + +-- +-- Name: TABLE artist_coin_stats_onchain; Type: COMMENT; Schema: public; Owner: - +-- + +COMMENT ON TABLE public.artist_coin_stats_onchain IS 'On-chain-derived shadow of artist_coin_stats, written by CoinStatsOnchainJob. Used to validate against the Birdeye-populated artist_coin_stats before replacing it.'; + + +-- +-- Name: artist_coin_stats_comparison; Type: VIEW; Schema: public; Owner: - +-- + +CREATE VIEW public.artist_coin_stats_comparison AS + SELECT ac.mint, + ac.ticker, + b.price AS birdeye_price, + o.price AS onchain_price, + round(((((o.price - b.price) / NULLIF(b.price, (0)::double precision)) * (100)::double precision))::numeric, 2) AS price_pct_diff, + b.market_cap AS birdeye_market_cap, + o.market_cap AS onchain_market_cap, + round(((((o.market_cap - b.market_cap) / NULLIF(b.market_cap, (0)::double precision)) * (100)::double precision))::numeric, 2) AS market_cap_pct_diff, + b.holder AS birdeye_holder, + o.holder AS onchain_holder, + (o.holder - b.holder) AS holder_diff, + b.liquidity AS birdeye_liquidity, + o.liquidity AS onchain_liquidity, + round(((((o.liquidity - b.liquidity) / NULLIF(b.liquidity, (0)::double precision)) * (100)::double precision))::numeric, 2) AS liquidity_pct_diff, + b.total_volume_usd AS birdeye_total_volume_usd, + o.total_volume_usd AS onchain_total_volume_usd, + round(((((o.total_volume_usd - b.total_volume_usd) / NULLIF(b.total_volume_usd, (0)::double precision)) * (100)::double precision))::numeric, 2) AS total_volume_usd_pct_diff, + b.price_change_24h_percent AS birdeye_price_change_24h_percent, + o.price_change_24h_percent AS onchain_price_change_24h_percent, + o.updated_at AS onchain_updated_at, + b.updated_at AS birdeye_updated_at + FROM ((public.artist_coins ac + LEFT JOIN public.artist_coin_stats b ON ((b.mint = (ac.mint)::text))) + LEFT JOIN public.artist_coin_stats_onchain o ON ((o.mint = (ac.mint)::text))); + + +-- +-- Name: VIEW artist_coin_stats_comparison; Type: COMMENT; Schema: public; Owner: - +-- + +COMMENT ON VIEW public.artist_coin_stats_comparison IS 'Compares Birdeye artist_coin_stats vs on-chain artist_coin_stats_onchain per coin (values + % diff) to validate CoinStatsOnchainJob before cutover.'; + + +-- +-- Name: artist_coin_volume_accumulator; Type: TABLE; Schema: public; Owner: - +-- + +CREATE TABLE public.artist_coin_volume_accumulator ( + mint text NOT NULL, + last_processed_slot bigint DEFAULT 0 NOT NULL, + total_volume double precision DEFAULT 0 NOT NULL, + total_volume_usd double precision DEFAULT 0 NOT NULL, + updated_at timestamp without time zone DEFAULT CURRENT_TIMESTAMP NOT NULL +); + + +-- +-- Name: TABLE artist_coin_volume_accumulator; Type: COMMENT; Schema: public; Owner: - +-- + +COMMENT ON TABLE public.artist_coin_volume_accumulator IS 'Running per-coin trading volume (AUDIO + USD) accumulated from pool quote-vault balance changes, watermarked by last_processed_slot. Written by CoinStatsOnchainJob.'; + + -- -- Name: associated_wallets; Type: TABLE; Schema: public; Owner: - -- @@ -11220,6 +11363,22 @@ ALTER TABLE ONLY public.artist_coin_pools ADD CONSTRAINT artist_coin_pools_pkey PRIMARY KEY (address); +-- +-- Name: artist_coin_price_history artist_coin_price_history_pkey; Type: CONSTRAINT; Schema: public; Owner: - +-- + +ALTER TABLE ONLY public.artist_coin_price_history + ADD CONSTRAINT artist_coin_price_history_pkey PRIMARY KEY (mint, "timestamp"); + + +-- +-- Name: artist_coin_stats_onchain artist_coin_stats_onchain_pkey; Type: CONSTRAINT; Schema: public; Owner: - +-- + +ALTER TABLE ONLY public.artist_coin_stats_onchain + ADD CONSTRAINT artist_coin_stats_onchain_pkey PRIMARY KEY (mint); + + -- -- Name: artist_coin_stats artist_coin_stats_pkey; Type: CONSTRAINT; Schema: public; Owner: - -- @@ -11228,6 +11387,14 @@ ALTER TABLE ONLY public.artist_coin_stats ADD CONSTRAINT artist_coin_stats_pkey PRIMARY KEY (mint); +-- +-- Name: artist_coin_volume_accumulator artist_coin_volume_accumulator_pkey; Type: CONSTRAINT; Schema: public; Owner: - +-- + +ALTER TABLE ONLY public.artist_coin_volume_accumulator + ADD CONSTRAINT artist_coin_volume_accumulator_pkey PRIMARY KEY (mint); + + -- -- Name: artist_coins artist_coins_pkey; Type: CONSTRAINT; Schema: public; Owner: - -- @@ -12419,6 +12586,13 @@ ALTER TABLE ONLY public.volume_leader_exclusions CREATE INDEX agg_user_has_tracks_idx ON public.aggregate_user USING btree (user_id) WHERE (total_track_count > 0); +-- +-- Name: artist_coin_price_history_mint_ts_idx; Type: INDEX; Schema: public; Owner: - +-- + +CREATE INDEX artist_coin_price_history_mint_ts_idx ON public.artist_coin_price_history USING btree (mint, "timestamp" DESC); + + -- -- Name: artist_coins_ticker_idx; Type: INDEX; Schema: public; Owner: - -- diff --git a/sql/03_migration_tracker.sql b/sql/03_migration_tracker.sql index 358035df..216d8300 100644 --- a/sql/03_migration_tracker.sql +++ b/sql/03_migration_tracker.sql @@ -204,6 +204,10 @@ migrations/0227_notification_single_recipient_user_timestamp_idx.sql f50ed8c909f migrations/0228_notification_multi_recipient_user_ids_idx.sql 43c58f2b7875cd09428fc8a9b01e1677 2026-07-28 05:42:28.131348+00 functions/get_user_score.sql 20f79f8ffe7b3b2582c5a7e29a9db45a 2026-07-28 05:42:28.312865+00 functions/handle_track_collaborator.sql 781dd365ffdf7619e234f5aff2544ba6 2026-07-28 05:42:29.217485+00 +migrations/0229_artist_coin_stats_onchain.sql 3e469a25759d4598603e4dc230466be3 2026-07-28 05:45:37.995068+00 +migrations/0230_artist_coin_price_history.sql 3f8e262d95c76c2b58d4eea8f2135840 2026-07-28 05:45:38.06593+00 +migrations/0231_artist_coin_volume_accumulator.sql 08ad700f8e10b84ee2c1b1e3eecbe033 2026-07-28 05:45:38.134036+00 +views/artist_coin_stats_comparison.sql e323f13c890cf86ce06d164523c1ec5b 2026-07-28 05:45:38.857194+00 \.