Uh oh!
There was an error while loading. Please reload this page.
Add new job to save token insights to the database#352
Uh oh!
There was an error while loading. Please reload this page.
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
File renamed without changes.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,18 @@ | ||
| package database | ||
| import ( | ||
| "context" | ||
| "github.com/jackc/pgx/v5" | ||
| "github.com/jackc/pgx/v5/pgconn" | ||
| "github.com/jackc/pgx/v5/pgxpool" | ||
| ) | ||
| type DbPool interface { | ||
| Acquire(context.Context) (*pgxpool.Conn, error) | ||
| Begin(context.Context) (pgx.Tx, error) | ||
| Exec(context.Context, string, ...any) (pgconn.CommandTag, error) | ||
| Query(context.Context, string, ...any) (pgx.Rows, error) | ||
| QueryRow(context.Context, string, ...any) pgx.Row | ||
| Close() | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,2 +1,2 @@ | ||
| ALTER TABLE artist_coins | ||
| ADD COLUMN IF NOT EXISTS name TEXT NOT NULL DEFAULT ''; | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,2 +1,2 @@ | ||
| ALTER TABLE artist_coins | ||
| ADD COLUMN IF NOT EXISTS dbc_pool TEXT DEFAULT NULL; |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,44 @@ | ||
| CREATE TABLE IF NOT EXISTS artist_coin_stats ( | ||
| mint TEXT PRIMARY KEY, | ||
| 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 NOT NULL DEFAULT CURRENT_TIMESTAMP, | ||
| updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP | ||
| ); |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,243 @@ | ||
| package jobs | ||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "sync" | ||
| "time" | ||
| "bridgerton.audius.co/birdeye" | ||
| "bridgerton.audius.co/config" | ||
| "bridgerton.audius.co/database" | ||
| "bridgerton.audius.co/logging" | ||
| "github.com/jackc/pgx/v5" | ||
| "go.uber.org/zap" | ||
| ) | ||
| // Birdeye rate limit is 15 requests per second. | ||
| // To be safe, we use 10 requests per second. | ||
| const birdeyeDelay = 100 * time.Millisecond | ||
| // How many tokens to fetch from the database in one batch. | ||
| const tokenPageSize = 1000 | ||
| type CoinStatsJob struct { | ||
| birdeyeClient *birdeye.Client | ||
| pool database.DbPool | ||
| logger *zap.Logger | ||
| mutex sync.Mutex | ||
| isRunning bool | ||
| } | ||
| func NewCoinStatsJob(config config.Config, pool database.DbPool) *CoinStatsJob { | ||
| logger := logging.NewZapLogger(config).Named("CoinStatsJob") | ||
| birdeyeClient := birdeye.New(config.BirdeyeToken) | ||
| return &CoinStatsJob{ | ||
| birdeyeClient: birdeyeClient, | ||
| logger: logger, | ||
| pool: pool, | ||
| } | ||
| } | ||
| // ScheduleEvery runs the job every `duration` until the context is cancelled. | ||
| func (j *CoinStatsJob) ScheduleEvery(ctx context.Context, duration time.Duration) error { | ||
| ticker := time.NewTicker(duration) | ||
| defer ticker.Stop() | ||
| for { | ||
| select { | ||
| case <-ticker.C: | ||
| j.logger.Info("Job started") | ||
| if err := j.run(ctx); err != nil { | ||
| j.logger.Error("Job run failed", zap.Error(err)) | ||
| } else { | ||
| j.logger.Info("Job completed successfully") | ||
| } | ||
| case <-ctx.Done(): | ||
| j.logger.Info("Job shutting down") | ||
| return nil | ||
| } | ||
| } | ||
| } | ||
| // run executes the job once. Ensures only one instance runs at a time. | ||
| // For each artist coin in the database, fetches the latest stats from Birdeye and | ||
| // updates the artist_coin_stats table. | ||
| func (j *CoinStatsJob) 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() | ||
| }() | ||
| count, err := j.getTokenCount(ctx) | ||
| if err != nil { | ||
| return fmt.Errorf("error getting token count: %w", err) | ||
| } | ||
| for offset := 0; offset < count; offset += tokenPageSize { | ||
| batch, err := j.getTokenBatch(ctx, tokenPageSize, offset) | ||
There was a problem hiding this comment. I think batch is kinda overkill, could just paginate 1 by 1 and fetch i think. but this works | ||
| if err != nil { | ||
| return fmt.Errorf("error getting token batch: %w", err) | ||
| } | ||
| for _, mint := range batch { | ||
| overview, err := j.birdeyeClient.GetTokenOverview(ctx, mint, "24h") | ||
| if err != nil { | ||
| return fmt.Errorf("error getting token overview for mint %s: %w", mint, err) | ||
| } | ||
| err = j.insertArtistCoinStats(ctx, mint, overview) | ||
| if err != nil { | ||
| return fmt.Errorf("error inserting artist coin stats for mint %s: %w", mint, err) | ||
| } | ||
| // Prevent rate limiting | ||
| time.Sleep(birdeyeDelay) | ||
| } | ||
| j.logger.Info("Processed batch", zap.Int("offset", offset), zap.Int("batch_size", len(batch))) | ||
| } | ||
| return nil | ||
| } | ||
| func (j *CoinStatsJob) getTokenCount(ctx context.Context) (int, error) { | ||
| var count int | ||
| err := j.pool.QueryRow(ctx, "SELECT COUNT(*) FROM artist_coins").Scan(&count) | ||
| if err != nil { | ||
| return 0, err | ||
| } | ||
| return count, nil | ||
| } | ||
| func (j *CoinStatsJob) getTokenBatch(ctx context.Context, limit, offset int) ([]string, error) { | ||
| rows, err := j.pool.Query(ctx, "SELECT mint FROM artist_coins ORDER BY mint LIMIT $1 OFFSET $2", limit, offset) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| defer rows.Close() | ||
| mints := make([]string, 0, limit) | ||
| for rows.Next() { | ||
| var mint string | ||
| if err := rows.Scan(&mint); err != nil { | ||
| return nil, err | ||
| } | ||
| mints = append(mints, mint) | ||
| } | ||
| return mints, nil | ||
| } | ||
| func (j *CoinStatsJob) insertArtistCoinStats(ctx context.Context, mint string, overview *birdeye.TokenOverview) error { | ||
| _, err := j.pool.Exec(ctx, ` | ||
| INSERT INTO artist_coin_stats ( | ||
| mint, market_cap, fdv, liquidity, last_trade_unix_time, last_trade_human_time, price, history_24h_price, | ||
| price_change_24h_percent, unique_wallet_24h, unique_wallet_history_24h, unique_wallet_24h_change_percent, | ||
| total_supply, circulating_supply, holder, trade_24h, trade_history_24h, trade_24h_change_percent, | ||
| sell_24h, sell_history_24h, sell_24h_change_percent, buy_24h, buy_history_24h, buy_24h_change_percent, | ||
| v_24h, v_24h_usd, v_history_24h, v_history_24h_usd, v_24h_change_percent, | ||
| v_buy_24h, v_buy_24h_usd, v_buy_history_24h, v_buy_history_24h_usd, v_buy_24h_change_percent, | ||
| v_sell_24h, v_sell_24h_usd, v_sell_history_24h, v_sell_history_24h_usd, v_sell_24h_change_percent, | ||
| number_markets, created_at, updated_at | ||
| ) VALUES ( | ||
| @mint, @market_cap, @fdv, @liquidity, @last_trade_unix_time, @last_trade_human_time, @price, @history_24h_price, | ||
| @price_change_24h_percent, @unique_wallet_24h, @unique_wallet_history_24h, @unique_wallet_24h_change_percent, | ||
| @total_supply, @circulating_supply, @holder, @trade_24h, @trade_history_24h, @trade_24h_change_percent, | ||
| @sell_24h, @sell_history_24h, @sell_24h_change_percent, @buy_24h, @buy_history_24h, @buy_24h_change_percent, | ||
| @v_24h, @v_24h_usd, @v_history_24h, @v_history_24h_usd, @v_24h_change_percent, | ||
| @v_buy_24h, @v_buy_24h_usd, @v_buy_history_24h, @v_buy_history_24h_usd, @v_buy_24h_change_percent, | ||
| @v_sell_24h, @v_sell_24h_usd, @v_sell_history_24h, @v_sell_history_24h_usd, @v_sell_24h_change_percent, | ||
| @number_markets, NOW(), NOW() | ||
| ) | ||
| ON CONFLICT (mint) DO UPDATE SET | ||
| market_cap = EXCLUDED.market_cap, | ||
| fdv = EXCLUDED.fdv, | ||
| liquidity = EXCLUDED.liquidity, | ||
| last_trade_unix_time = EXCLUDED.last_trade_unix_time, | ||
| last_trade_human_time = EXCLUDED.last_trade_human_time, | ||
| price = EXCLUDED.price, | ||
| history_24h_price = EXCLUDED.history_24h_price, | ||
| price_change_24h_percent = EXCLUDED.price_change_24h_percent, | ||
| unique_wallet_24h = EXCLUDED.unique_wallet_24h, | ||
| unique_wallet_history_24h = EXCLUDED.unique_wallet_history_24h, | ||
| unique_wallet_24h_change_percent = EXCLUDED.unique_wallet_24h_change_percent, | ||
| total_supply = EXCLUDED.total_supply, | ||
| circulating_supply = EXCLUDED.circulating_supply, | ||
| holder = EXCLUDED.holder, | ||
| trade_24h = EXCLUDED.trade_24h, | ||
| trade_history_24h = EXCLUDED.trade_history_24h, | ||
| trade_24h_change_percent = EXCLUDED.trade_24h_change_percent, | ||
| sell_24h = EXCLUDED.sell_24h, | ||
| sell_history_24h = EXCLUDED.sell_history_24h, | ||
| sell_24h_change_percent = EXCLUDED.sell_24h_change_percent, | ||
| buy_24h = EXCLUDED.buy_24h, | ||
| buy_history_24h = EXCLUDED.buy_history_24h, | ||
| buy_24h_change_percent = EXCLUDED.buy_24h_change_percent, | ||
| v_24h = EXCLUDED.v_24h, | ||
| v_24h_usd = EXCLUDED.v_24h_usd, | ||
| v_history_24h = EXCLUDED.v_history_24h, | ||
| v_history_24h_usd = EXCLUDED.v_history_24h_usd, | ||
| v_24h_change_percent = EXCLUDED.v_24h_change_percent, | ||
| v_buy_24h = EXCLUDED.v_buy_24h, | ||
| v_buy_24h_usd = EXCLUDED.v_buy_24h_usd, | ||
| v_buy_history_24h = EXCLUDED.v_buy_history_24h, | ||
| v_buy_history_24h_usd = EXCLUDED.v_buy_history_24h_usd, | ||
| v_buy_24h_change_percent = EXCLUDED.v_buy_24h_change_percent, | ||
| v_sell_24h = EXCLUDED.v_sell_24h, | ||
| v_sell_24h_usd = EXCLUDED.v_sell_24h_usd, | ||
| v_sell_history_24h = EXCLUDED.v_sell_history_24h, | ||
| v_sell_history_24h_usd = EXCLUDED.v_sell_history_24h_usd, | ||
| v_sell_24h_change_percent = EXCLUDED.v_sell_24h_change_percent, | ||
| number_markets = EXCLUDED.number_markets, | ||
| updated_at = NOW(); | ||
| `, pgx.NamedArgs{ | ||
| "mint": mint, | ||
| "market_cap": overview.MarketCap, | ||
| "fdv": overview.FDV, | ||
| "liquidity": overview.Liquidity, | ||
| "last_trade_unix_time": overview.LastTradeUnixTime, | ||
| "last_trade_human_time": overview.LastTradeHumanTime, | ||
| "price": overview.Price, | ||
| "history_24h_price": overview.History24hPrice, | ||
| "price_change_24h_percent": overview.PriceChange24hPercent, | ||
| "unique_wallet_24h": overview.UniqueWallet24h, | ||
| "unique_wallet_history_24h": overview.UniqueWalletHistory24h, | ||
| "unique_wallet_24h_change_percent": overview.UniqueWallet24hChangePercent, | ||
| "total_supply": overview.TotalSupply, | ||
| "circulating_supply": overview.CirculatingSupply, | ||
| "holder": overview.Holder, | ||
| "trade_24h": overview.Trade24h, | ||
| "trade_history_24h": overview.TradeHistory24h, | ||
| "trade_24h_change_percent": overview.Trade24hChangePercent, | ||
| "sell_24h": overview.Sell24h, | ||
| "sell_history_24h": overview.SellHistory24h, | ||
| "sell_24h_change_percent": overview.Sell24hChangePercent, | ||
| "buy_24h": overview.Buy24h, | ||
| "buy_history_24h": overview.BuyHistory24h, | ||
| "buy_24h_change_percent": overview.Buy24hChangePercent, | ||
| "v_24h": overview.V24h, | ||
| "v_24h_usd": overview.V24hUSD, | ||
| "v_history_24h": overview.VHistory24h, | ||
| "v_history_24h_usd": overview.VHistory24hUSD, | ||
| "v_24h_change_percent": overview.V24hChangePercent, | ||
| "v_buy_24h": overview.VBuy24h, | ||
| "v_buy_24h_usd": overview.VBuy24hUSD, | ||
| "v_buy_history_24h": overview.VBuyHistory24h, | ||
| "v_buy_history_24h_usd": overview.VBuyHistory24hUSD, | ||
| "v_buy_24h_change_percent": overview.VBuy24hChangePercent, | ||
| "v_sell_24h": overview.VSell24h, | ||
| "v_sell_24h_usd": overview.VSell24hUSD, | ||
| "v_sell_history_24h": overview.VSellHistory24h, | ||
| "v_sell_history_24h_usd": overview.VSellHistory24hUSD, | ||
| "v_sell_24h_change_percent": overview.VSell24hChangePercent, | ||
| "number_markets": overview.NumberMarkets, | ||
| }) | ||
| return err | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Uh oh!
There was an error while loading. Please reload this page.
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
👍