//go:build !nosqlite package storage import ( "context" "database/sql" "encoding/json" "fmt" "math" "os" "path/filepath" "strconv" "strings" "sync" // ONE sqlite C library in this binary, under a name nothing else claims. // csqlite is the house build of the amalgamation and registers "sqlite3". // Upstream mattn compiles a second copy of the same C, so linking both // defines every sqlite3_* symbol twice — a link failure on darwin, and on // linux survivable only because that linker is laxer about duplicates. // // The name is the other half. hanzoai/sqlite wraps this same C but takes // the name "sqlite", which modernc.org/sqlite also takes — and modernc is // in every binary that stores here, by way of hanzoai/replicate next door // in this package. Two packages registering one name is a panic in init, // before main runs. _ "github.com/hanzoai/csqlite" ) // Store is the unified storage backend backed by SQLite WAL. type Store struct { dataDir string mu sync.RWMutex db *sql.DB } // New creates a store rooted at dataDir with SQLite WAL. func New(dataDir string) (*Store, error) { if err := os.MkdirAll(dataDir, 0755); err != nil { return nil, fmt.Errorf("storage: mkdir %s: %w", dataDir, err) } dbPath := filepath.Join(dataDir, "graph.db") // One SQLite driver, the same one the indexer and the explorer load. It // carries a copy of the C amalgamation, so a second package bringing another // copy — a dependency two hops away is enough — collides at link time and // takes every test in the repo down with it. That is worth knowing before // adding a dependency here; it is what an unused ORM did to this module. db, err := sql.Open("sqlite3", dbPath+ "?_journal_mode=WAL&_busy_timeout=5000&_synchronous=NORMAL&_cache_size=-64000") if err != nil { return nil, fmt.Errorf("storage: open %s: %w", dbPath, err) } // Write-ahead logging is a property of the file, not of a connection, so it // is set once here rather than handed to every connection that opens. It is // what lets readers keep reading while a pass writes. if _, err := db.Exec("PRAGMA journal_mode=WAL"); err != nil { db.Close() return nil, fmt.Errorf("storage: enable WAL on %s: %w", dbPath, err) } db.SetMaxOpenConns(8) return &Store{dataDir: dataDir, db: db}, nil } // Init creates tables and indexes. func (s *Store) Init(_ context.Context) error { schema := ` CREATE TABLE IF NOT EXISTS factories (id TEXT PRIMARY KEY, data JSON); CREATE TABLE IF NOT EXISTS bundles (id TEXT PRIMARY KEY, data JSON); CREATE TABLE IF NOT EXISTS tokens (id TEXT PRIMARY KEY, data JSON); CREATE TABLE IF NOT EXISTS pools (id TEXT PRIMARY KEY, data JSON); CREATE TABLE IF NOT EXISTS swaps (id TEXT PRIMARY KEY, data JSON, timestamp INTEGER, pool TEXT, amountUSD REAL NOT NULL DEFAULT 0); CREATE TABLE IF NOT EXISTS entities (type TEXT, id TEXT, data JSON, PRIMARY KEY(type, id)); CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT); CREATE INDEX IF NOT EXISTS idx_swaps_timestamp ON swaps(timestamp); CREATE INDEX IF NOT EXISTS idx_entities_type ON entities(type); -- GetSwaps compares pool COLLATE NOCASE and orders by timestamp, and an -- index is used only when its collation matches the comparison. The -- earlier index on swaps(pool) was BINARY, so a pool's history was a full -- scan of every swap on the chain — seconds per pool, and a token page -- asks once per pool it trades in. DROP INDEX IF EXISTS idx_swaps_pool; CREATE INDEX IF NOT EXISTS idx_swaps_pool_time ON swaps(pool COLLATE NOCASE, timestamp); ` if _, err := s.db.Exec(schema); err != nil { return err } // A swap's columns are the queryable projection of its document. Timestamp // and pool were promoted so a window and a pool's history could be asked of // the database instead of of Go; amountUSD is promoted for the same reason. // What the protocol has traded for all time is a SUM over this table, and a // figure that exists only inside a JSON document cannot be summed — this // driver's amalgamation is built without JSON1, so json_extract is not an // option here either (see ValueSwaps). // // A database written before the column gets it now, filled once from the // rows themselves. Every write since keeps the two in step, so the fill runs // exactly once per database and the ALTER failing IS how it knows. if _, err := s.db.Exec(`ALTER TABLE swaps ADD COLUMN amountUSD REAL NOT NULL DEFAULT 0`); err == nil { return s.projectAmounts() } return nil } // projectAmounts copies each stored swap's dollar value out of its document and // into the column beside it. func (s *Store) projectAmounts() error { rows, err := s.db.Query(`SELECT id, data FROM swaps`) if err != nil { return fmt.Errorf("storage: read swaps to project: %w", err) } type valued struct { id string usd float64 } // Drained whole before the write opens, because a read held open across a // write on the same file is how SQLite is asked to deadlock with itself. var priced []valued for rows.Next() { var id, raw string if err := rows.Scan(&id, &raw); err != nil { continue } var d SeedSwapData if json.Unmarshal([]byte(raw), &d) != nil { continue } if usd := dollars(d.AmountUSD); usd != 0 { priced = append(priced, valued{id, usd}) } } rows.Close() if len(priced) == 0 { return nil } tx, err := s.db.Begin() if err != nil { return fmt.Errorf("storage: project amounts: %w", err) } stmt, err := tx.Prepare(`UPDATE swaps SET amountUSD = ? WHERE id = ?`) if err != nil { tx.Rollback() return fmt.Errorf("storage: project amounts: %w", err) } defer stmt.Close() for _, v := range priced { if _, err := stmt.Exec(v.usd, v.id); err != nil { tx.Rollback() return fmt.Errorf("storage: project %s: %w", v.id, err) } } return tx.Commit() } // dollars reads a formatted dollar figure back as a number. Anything unreadable // is worth nothing, which is what an unpriced swap already says. func dollars(s string) float64 { v, err := strconv.ParseFloat(s, 64) if err != nil || math.IsNaN(v) || math.IsInf(v, 0) { return 0 } return v } // SwapsAfter returns up to limit swaps stored later than ts, oldest first. // // The mirror of RecentSwapsRaw: one reads the head of the table, this reads // forward from wherever a caller last got to, so a history longer than any // window can be walked a batch at a time and finish. func (s *Store) SwapsAfter(ts int64, limit int) map[string]*SeedSwapData { s.mu.RLock() defer s.mu.RUnlock() out := map[string]*SeedSwapData{} rows, err := s.db.Query( "SELECT id, data FROM swaps WHERE timestamp > ? ORDER BY timestamp ASC LIMIT ?", ts, limit) if err != nil { return out } defer rows.Close() for rows.Next() { var id, raw string if err := rows.Scan(&id, &raw); err != nil { continue } var sw SeedSwapData if json.Unmarshal([]byte(raw), &sw) != nil { continue } out[id] = &sw } return out } // PricedThrough is the newest trade time a valuation pass has reached walking // forward from the start of the table, or 0 for a store that has never walked. // // A mark, not a test of the rows: a trade whose tokens reach no stablecoin // cannot be priced at all, and asking "which rows still have no value" would // hand back the same unpriceable trades every pass forever. Having LOOKED is // monotone where having VALUED is not, so the mark is what is remembered. func (s *Store) PricedThrough() int64 { s.mu.RLock() defer s.mu.RUnlock() var v string if err := s.db.QueryRow("SELECT value FROM meta WHERE key='pricedThrough'").Scan(&v); err != nil { return 0 } n, _ := strconv.ParseInt(v, 10, 64) return n } // SetPricedThrough records how far forward the pass has now valued. func (s *Store) SetPricedThrough(ts int64) { s.mu.Lock() defer s.mu.Unlock() s.db.Exec("INSERT OR REPLACE INTO meta(key, value) VALUES('pricedThrough', ?)", strconv.FormatInt(ts, 10)) } // Traded is a count of trades and what they were worth. type Traded struct { Trades int64 VolumeUSD float64 } // TradedByPool reports what each pool has traded, over the whole table. // // One question sliced one way, and the protocol's own total is the fold of it — // asked separately they drifted apart by fifty times: the protocol summed every // trade ever indexed while each pool summed the window one pass had just read, // so a landing page claimed $206,718 above a list of pools claiming $4,159. // // Over the whole table, not over a window: a pass values a bounded slice because // the table grows forever, and that bound is right for the pass and wrong for a // figure the wire calls all-time. Grouping costs one scan and there are as many // rows out as there are pools. func (s *Store) TradedByPool() (map[string]Traded, error) { s.mu.RLock() defer s.mu.RUnlock() rows, err := s.db.Query( `SELECT pool, COUNT(*), COALESCE(SUM(amountUSD), 0) FROM swaps GROUP BY pool COLLATE NOCASE`) if err != nil { return nil, fmt.Errorf("storage: traded by pool: %w", err) } defer rows.Close() out := map[string]Traded{} for rows.Next() { var pool string var t Traded if err := rows.Scan(&pool, &t.Trades, &t.VolumeUSD); err != nil { continue } out[strings.ToLower(pool)] = t } return out, rows.Err() } // Close closes the database. func (s *Store) Close() error { if s.db != nil { return s.db.Close() } return nil } // DataDir returns the store's data directory. func (s *Store) DataDir() string { return s.dataDir } // SetLastBlock persists the indexer's last processed block. func (s *Store) SetLastBlock(block uint64) { s.mu.Lock() defer s.mu.Unlock() s.db.Exec("INSERT OR REPLACE INTO meta(key, value) VALUES('lastBlock', ?)", strconv.FormatUint(block, 10)) } // SetSource records which addresses the rows below the cursor were built from. // Called once the indexer has committed to a set, so a later start can tell // whether the cursor it inherits describes the same book. func (s *Store) SetSource(fingerprint string) { s.mu.Lock() defer s.mu.Unlock() s.db.Exec("INSERT OR REPLACE INTO meta(key, value) VALUES('source', ?)", fingerprint) } // Source returns the fingerprint the rows were built from, or "" for a store // written before this was recorded — which is indistinguishable from a store // built by another set, and is treated as one. func (s *Store) Source() string { s.mu.RLock() defer s.mu.RUnlock() var v string if err := s.db.QueryRow("SELECT value FROM meta WHERE key='source'").Scan(&v); err != nil { return "" } return v } // DropDerived removes everything read off the chain, leaving the file and its // schema. Used when the addresses changed under a store: those rows describe a // different book, and mixing them with the new one gives an answer that was // never true anywhere. func (s *Store) DropDerived() error { s.mu.Lock() defer s.mu.Unlock() for _, t := range []string{"factories", "bundles", "tokens", "pools", "swaps", "entities"} { if _, err := s.db.Exec("DELETE FROM " + t); err != nil { return err } } _, err := s.db.Exec("DELETE FROM meta WHERE key='lastBlock'") return err } // GetLastBlock returns the indexer's last processed block. func (s *Store) GetLastBlock() uint64 { s.mu.RLock() defer s.mu.RUnlock() var v string if err := s.db.QueryRow("SELECT value FROM meta WHERE key='lastBlock'").Scan(&v); err != nil { return 0 } n, _ := strconv.ParseUint(v, 10, 64) return n } // --- Seed methods --- func (s *Store) SeedFactory(id string, d *SeedFactoryData) { s.mu.Lock() defer s.mu.Unlock() data, _ := json.Marshal(d) s.db.Exec("INSERT OR REPLACE INTO factories(id, data) VALUES(?, ?)", id, string(data)) } func (s *Store) SeedBundle(id string, d *SeedBundleData) { s.mu.Lock() defer s.mu.Unlock() data, _ := json.Marshal(d) s.db.Exec("INSERT OR REPLACE INTO bundles(id, data) VALUES(?, ?)", id, string(data)) } func (s *Store) SeedToken(id string, d *SeedTokenData) { s.mu.Lock() defer s.mu.Unlock() data, _ := json.Marshal(d) s.db.Exec("INSERT OR REPLACE INTO tokens(id, data) VALUES(?, ?)", id, string(data)) } func (s *Store) SeedPool(id string, d *SeedPoolData) { s.mu.Lock() defer s.mu.Unlock() data, _ := json.Marshal(d) s.db.Exec("INSERT OR REPLACE INTO pools(id, data) VALUES(?, ?)", id, string(data)) } func (s *Store) SeedSwap(id string, d *SeedSwapData) { s.mu.Lock() defer s.mu.Unlock() data, _ := json.Marshal(d) s.db.Exec("INSERT OR REPLACE INTO swaps(id, data, timestamp, pool, amountUSD) VALUES(?, ?, ?, ?, ?)", id, string(data), d.Timestamp, d.Pool, dollars(d.AmountUSD)) } // ValueSwaps writes what each trade was worth onto the stored swap. // // One transaction and one prepared statement for the whole batch. Sent through // the plain Exec path instead, every row commits on its own and the interval is // spent waiting on the disk; a chain with a real trade history never finishes // the pass. That cost is why the value used to be dropped on the floor and // every swap read back a dollar amount of zero. // // Callers pass values already formatted, so there is exactly one place that // decides how a dollar figure is written. // It reports how many rows it actually changed. A write that quietly does // nothing is the failure this had first: the pass ran, said nothing, and every // swap kept reading zero under a pool claiming thousands in volume. func (s *Store) ValueSwaps(usd map[string]string) (int64, error) { if len(usd) == 0 { return 0, nil } s.mu.Lock() defer s.mu.Unlock() tx, err := s.db.Begin() if err != nil { return 0, fmt.Errorf("begin: %w", err) } // The field is edited here rather than by SQLite's json_set: that function // belongs to the JSON1 extension, which this driver's amalgamation is not // always built with. It is present on a developer's machine and absent in // the image, so the statement prepared cleanly in a test and failed on every // pass in production. read, err := tx.Prepare(`SELECT data FROM swaps WHERE id = ?`) if err != nil { tx.Rollback() return 0, fmt.Errorf("prepare read: %w", err) } defer read.Close() write, err := tx.Prepare(`UPDATE swaps SET data = ?, amountUSD = ? WHERE id = ?`) if err != nil { tx.Rollback() return 0, fmt.Errorf("prepare write: %w", err) } defer write.Close() var rows int64 for id, v := range usd { var raw string if err := read.QueryRow(id).Scan(&raw); err != nil { if err == sql.ErrNoRows { continue // the window moved on; nothing to price } tx.Rollback() return 0, fmt.Errorf("read %s: %w", id, err) } var d SeedSwapData if err := json.Unmarshal([]byte(raw), &d); err != nil { tx.Rollback() return 0, fmt.Errorf("decode %s: %w", id, err) } d.AmountUSD = v next, err := json.Marshal(&d) if err != nil { tx.Rollback() return 0, fmt.Errorf("encode %s: %w", id, err) } res, err := write.Exec(string(next), dollars(v), id) if err != nil { tx.Rollback() return 0, fmt.Errorf("write %s: %w", id, err) } n, _ := res.RowsAffected() rows += n } if err := tx.Commit(); err != nil { return 0, fmt.Errorf("commit: %w", err) } return rows, nil } // --- Generic entity storage --- func (s *Store) SetEntity(entityType, id string, data interface{}) { s.mu.Lock() defer s.mu.Unlock() j, _ := json.Marshal(data) s.db.Exec("INSERT OR REPLACE INTO entities(type, id, data) VALUES(?, ?, ?)", entityType, id, string(j)) } func (s *Store) GetByType(entityType, id string) (interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() var raw string if err := s.db.QueryRow("SELECT data FROM entities WHERE type=? AND id=?", entityType, id).Scan(&raw); err != nil { if err == sql.ErrNoRows { return nil, nil } return nil, err } var v interface{} json.Unmarshal([]byte(raw), &v) return v, nil } // ListByType returns every entity of a type, up to limit. The result is always // a list — an unpopulated type yields an empty one. // // It must never be a nil slice: nil encodes as JSON `null`, and a GraphQL list // field answering `null` is the wire signature of a subgraph that is not // deployed or is erroring, not of one that has simply seen no events yet. The // `dex` subgraph answered `{"markets":null}` for exactly this reason while it // was deployed, subscribed to 0x9999 and fully caught up to the chain head. func (s *Store) ListByType(entityType string, limit int) (interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() rows, err := s.db.Query("SELECT data FROM entities WHERE type=? LIMIT ?", entityType, limit) if err != nil { return nil, err } defer rows.Close() result := []interface{}{} for rows.Next() { var raw string if err := rows.Scan(&raw); err != nil { continue } var v interface{} json.Unmarshal([]byte(raw), &v) result = append(result, v) } return result, nil } // Entities returns the stored entities of a type, filtered, ordered and cut to // limit. // // It reads the whole type where ListByType reads the first `limit` rows: a // filter and an order are properties of the WHOLE collection, so a SQL LIMIT // applied before them would answer with the first N rows of the table rather // than the first N matches — a chart asking for the newest 90 days would get // the oldest 90 rows of whatever happened to be stored. func (s *Store) Entities(entityType string, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() rows, err := s.db.Query("SELECT data FROM entities WHERE type=?", entityType) if err != nil { return nil, err } defer rows.Close() result := []interface{}{} for rows.Next() { var raw string if err := rows.Scan(&raw); err != nil { continue } var v interface{} if json.Unmarshal([]byte(raw), &v) == nil { result = append(result, v) } } return page(result, limit, orderBy, orderDirection, where), nil } // --- Block queries (stubs) --- func (s *Store) GetBlock(_ context.Context, id string) (interface{}, error) { return nil, nil } func (s *Store) GetBlockByNumber(_ context.Context, num string) (interface{}, error) { return nil, nil } func (s *Store) GetLatestBlock(_ context.Context) (interface{}, error) { return nil, nil } func (s *Store) GetBlocks(_ context.Context, limit int) (interface{}, error) { return []interface{}{}, nil } // --- Transaction queries (stubs) --- func (s *Store) GetTransaction(_ context.Context, hash string) (interface{}, error) { return nil, nil } func (s *Store) GetTransactions(_ context.Context, limit int) (interface{}, error) { return []interface{}{}, nil } // --- Token queries --- func (s *Store) GetToken(_ context.Context, addr string) (interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() t, err := s.loadToken(addr) if err != nil || t == nil { return nil, err } return map[string]interface{}{ "id": addr, "symbol": t.Symbol, "name": t.Name, "decimals": t.Decimals, "volumeUSD": t.VolumeUSD, "totalValueLockedUSD": t.TotalValueLockedUSD, "derivedETH": t.DerivedETH, "txCount": t.TxCount, "totalSupply": t.TotalSupply, "staked": t.Staked, }, nil } func (s *Store) GetTokens(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() rows, err := s.db.Query("SELECT id, data FROM tokens") if err != nil { return nil, err } defer rows.Close() result := []interface{}{} for rows.Next() { var id, raw string if err := rows.Scan(&id, &raw); err != nil { continue } var t SeedTokenData json.Unmarshal([]byte(raw), &t) result = append(result, map[string]interface{}{ "id": id, "symbol": t.Symbol, "name": t.Name, "decimals": t.Decimals, "volumeUSD": t.VolumeUSD, "totalValueLockedUSD": t.TotalValueLockedUSD, "derivedETH": t.DerivedETH, "txCount": t.TxCount, "totalSupply": t.TotalSupply, "staked": t.Staked, }) } result = FilterResults(result, where) sortResults(result, orderBy, orderDirection) if len(result) > limit { result = result[:limit] } return result, nil } // --- DEX queries --- func (s *Store) GetFactory(_ context.Context, id string) (interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() f, err := s.loadFactory(id) if err != nil || f == nil { return nil, err } return map[string]interface{}{ "id": id, "poolCount": f.PoolCount, "txCount": f.TxCount, "totalVolumeUSD": f.TotalVolumeUSD, "totalValueLockedUSD": f.TotalValueLockedUSD, }, nil } func (s *Store) GetFactories(_ context.Context) (interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() rows, err := s.db.Query("SELECT id, data FROM factories") if err != nil { return nil, err } defer rows.Close() result := []interface{}{} for rows.Next() { var id, raw string if err := rows.Scan(&id, &raw); err != nil { continue } var f SeedFactoryData json.Unmarshal([]byte(raw), &f) result = append(result, map[string]interface{}{ "id": id, "poolCount": f.PoolCount, "txCount": f.TxCount, "totalVolumeUSD": f.TotalVolumeUSD, "totalValueLockedUSD": f.TotalValueLockedUSD, }) } return result, nil } func (s *Store) GetBundle(_ context.Context, id string) (interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() var raw string if err := s.db.QueryRow("SELECT data FROM bundles WHERE id=?", id).Scan(&raw); err != nil { if err == sql.ErrNoRows { return nil, nil } return nil, err } var b SeedBundleData json.Unmarshal([]byte(raw), &b) return map[string]interface{}{ "id": id, "ethPriceUSD": b.EthPriceUSD, "ethPrice": b.EthPriceUSD, }, nil } func (s *Store) GetPool(_ context.Context, id string) (interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() p, err := s.loadPool(id) if err != nil || p == nil { return nil, err } return s.poolToMap(id, p), nil } func (s *Store) GetPools(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() // Load all pools first, then close rows before resolving tokens (avoids conn deadlock). type idPool struct { id string p SeedPoolData } rows, err := s.db.Query("SELECT id, data FROM pools") if err != nil { return nil, err } var pools []idPool for rows.Next() { var id, raw string if err := rows.Scan(&id, &raw); err != nil { continue } var p SeedPoolData json.Unmarshal([]byte(raw), &p) pools = append(pools, idPool{id, p}) } rows.Close() result := []interface{}{} for _, pp := range pools { p := pp.p result = append(result, s.poolToMap(pp.id, &p)) } result = FilterResults(result, where) sortResults(result, orderBy, orderDirection) if len(result) > limit { result = result[:limit] } return result, nil } func (s *Store) poolToMap(id string, p *SeedPoolData) map[string]interface{} { var t0, t1 interface{} if tok, _ := s.loadToken(p.Token0); tok != nil { t0 = map[string]interface{}{"id": p.Token0, "symbol": tok.Symbol, "name": tok.Name, "decimals": tok.Decimals} } if tok, _ := s.loadToken(p.Token1); tok != nil { t1 = map[string]interface{}{"id": p.Token1, "symbol": tok.Symbol, "name": tok.Name, "decimals": tok.Decimals} } return map[string]interface{}{ "id": id, "token0": t0, "token1": t1, "feeTier": p.FeeTier, "totalValueLockedUSD": p.TotalValueLockedUSD, "volumeUSD": p.VolumeUSD, "token0Price": p.Token0Price, "token1Price": p.Token1Price, "txCount": p.TxCount, } } func (s *Store) GetPair(_ context.Context, id string) (interface{}, error) { return s.GetPool(nil, id) } func (s *Store) GetPairs(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return s.GetPools(nil, limit, orderBy, orderDirection, where) } func (s *Store) GetSwap(_ context.Context, id string) (interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() var raw string if err := s.db.QueryRow("SELECT data FROM swaps WHERE id=?", id).Scan(&raw); err != nil { if err == sql.ErrNoRows { return nil, nil } return nil, err } var sw SeedSwapData json.Unmarshal([]byte(raw), &sw) return s.swapToMap(id, &sw), nil } func (s *Store) GetSwaps(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { s.mu.RLock() defer s.mu.RUnlock() if orderBy == "" { orderBy = "timestamp" } if orderDirection == "" { orderDirection = "desc" } // A filter on the pool goes into the SQL, because the LIMIT is applied by // the database and a limit that runs before the filter answers with the // first N rows rather than the first N matches. // // One pair does most of the trading on a live chain, so it fills any recent // window by itself: asking for one quiet pool's last 1000 trades read the // 1000 newest chain-wide, kept the handful that matched, and usually // answered with none at all. The pool column is indexed for exactly this. args := []interface{}{} conds := []string{} if pool, ok := whereString(where, "pool"); ok { conds = append(conds, "pool = ? COLLATE NOCASE") args = append(args, pool) } // A page of history is asked for as "the N before this time", and that bound // has to sit beside the LIMIT for the same reason the pool does: applied // after it, the query returns the newest N and then throws them all away. if before, ok := whereNumber(where, "timestamp_lt"); ok { conds = append(conds, "timestamp < ?") args = append(args, before) } query := "SELECT id, data FROM swaps" if len(conds) > 0 { query += " WHERE " + strings.Join(conds, " AND ") } if orderBy == "timestamp" { if strings.EqualFold(orderDirection, "desc") { query += " ORDER BY timestamp DESC" } else { query += " ORDER BY timestamp ASC" } } query += " LIMIT ?" args = append(args, limit) // Load all swaps first, close rows, then resolve pool/token refs. type idSwap struct { id string sw SeedSwapData } rows, err := s.db.Query(query, args...) if err != nil { return nil, err } var swaps []idSwap for rows.Next() { var id, raw string if err := rows.Scan(&id, &raw); err != nil { continue } var sw SeedSwapData json.Unmarshal([]byte(raw), &sw) swaps = append(swaps, idSwap{id, sw}) } rows.Close() result := []interface{}{} for _, ss := range swaps { sw := ss.sw result = append(result, s.swapToMap(ss.id, &sw)) } result = FilterResults(result, where) if orderBy != "timestamp" { sortResults(result, orderBy, orderDirection) } if len(result) > limit { result = result[:limit] } return result, nil } // whereString reads a plain string equality out of a `where` map — the shape a // filter naming one pool arrives in. Anything richer is left to FilterResults, // which still runs over whatever the query returned. func whereString(where map[string]interface{}, key string) (string, bool) { v, ok := where[key] if !ok { return "", false } s, ok := v.(string) if !ok || s == "" { return "", false } return s, true } // whereNumber reads a numeric bound out of a `where` map, however the client // wrote it — GraphQL variables arrive as float64, literals may arrive as strings. func whereNumber(where map[string]interface{}, key string) (float64, bool) { v, ok := where[key] if !ok || v == nil { return 0, false } f, err := strconv.ParseFloat(fmt.Sprint(v), 64) if err != nil { return 0, false } return f, true } func (s *Store) swapToMap(id string, sw *SeedSwapData) map[string]interface{} { var pool interface{} = sw.Pool if p, _ := s.loadPool(sw.Pool); p != nil { pool = s.poolToMap(sw.Pool, p) } return map[string]interface{}{ "id": id, "timestamp": sw.Timestamp, "pool": pool, "amount0": sw.Amount0, "amount1": sw.Amount1, "amountUSD": sw.AmountUSD, "sender": sw.Sender, } } // Stubs for unimplemented entity types func (s *Store) GetMint(_ context.Context, id string) (interface{}, error) { return nil, nil } func (s *Store) GetMints(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return []interface{}{}, nil } func (s *Store) GetBurn(_ context.Context, id string) (interface{}, error) { return nil, nil } func (s *Store) GetBurns(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return []interface{}{}, nil } func (s *Store) GetTick(_ context.Context, id string) (interface{}, error) { return nil, nil } func (s *Store) GetTicks(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return []interface{}{}, nil } func (s *Store) GetPosition(_ context.Context, id string) (interface{}, error) { return nil, nil } func (s *Store) GetPositions(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return []interface{}{}, nil } func (s *Store) GetCollect(_ context.Context, id string) (interface{}, error) { return nil, nil } func (s *Store) GetCollects(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return []interface{}{}, nil } func (s *Store) GetFlash(_ context.Context, id string) (interface{}, error) { return nil, nil } func (s *Store) GetFlashes(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return []interface{}{}, nil } // --- Day series --- // // The indexer writes one row per (subject, UTC day) as an entity; these read // them back. Three subjects have readers because three are asked for: a token's // candles, a pool's candles, and the protocol total behind the explore tiles. // PairDayData and PairHourData are the v2 spellings and no client asks for // them. func (s *Store) GetTokenDayDatas(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return s.Entities(TokenDay, limit, orderBy, orderDirection, where) } func (s *Store) GetPoolDayDatas(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return s.Entities(PoolDay, limit, orderBy, orderDirection, where) } func (s *Store) GetFactoryDayDatas(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return s.Entities(FactoryDay, limit, orderBy, orderDirection, where) } func (s *Store) GetPairDayDatas(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return []interface{}{}, nil } func (s *Store) GetPairHourDatas(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return []interface{}{}, nil } func (s *Store) GetTransfer(_ context.Context, id string) (interface{}, error) { return nil, nil } func (s *Store) GetTransfers(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return []interface{}{}, nil } func (s *Store) GetNFT(_ context.Context, id string) (interface{}, error) { return nil, nil } func (s *Store) GetNFTs(_ context.Context, limit int, orderBy, orderDirection string, where map[string]interface{}) (interface{}, error) { return []interface{}{}, nil } // --- V4 storage methods --- func (s *Store) GetModifyLiquidity(_ context.Context, id string) (interface{}, error) { return nil, nil } func (s *Store) GetModifyLiquiditys(_ context.Context, limit int, orderBy, orderDirection string) (interface{}, error) { return []interface{}{}, nil } func (s *Store) GetSubscribe(_ context.Context, id string) (interface{}, error) { return nil, nil } func (s *Store) GetSubscribes(_ context.Context, limit int, orderBy, orderDirection string) (interface{}, error) { return []interface{}{}, nil } func (s *Store) GetUnsubscribe(_ context.Context, id string) (interface{}, error) { return nil, nil } func (s *Store) GetUnsubscribes(_ context.Context, limit int, orderBy, orderDirection string) (interface{}, error) { return []interface{}{}, nil } func (s *Store) GetPoolHourDatas(_ context.Context, limit int, orderBy, orderDirection string) (interface{}, error) { return []interface{}{}, nil } func (s *Store) GetTokenHourDatas(_ context.Context, limit int, orderBy, orderDirection string) (interface{}, error) { return []interface{}{}, nil } // --- Internal helpers --- func (s *Store) loadFactory(id string) (*SeedFactoryData, error) { var raw string if err := s.db.QueryRow("SELECT data FROM factories WHERE id=?", id).Scan(&raw); err != nil { if err == sql.ErrNoRows { return nil, nil } return nil, err } var f SeedFactoryData json.Unmarshal([]byte(raw), &f) return &f, nil } func (s *Store) loadToken(addr string) (*SeedTokenData, error) { var raw string if err := s.db.QueryRow("SELECT data FROM tokens WHERE id=?", addr).Scan(&raw); err != nil { if err == sql.ErrNoRows { return nil, nil } return nil, err } var t SeedTokenData json.Unmarshal([]byte(raw), &t) return &t, nil } func (s *Store) loadPool(id string) (*SeedPoolData, error) { var raw string if err := s.db.QueryRow("SELECT data FROM pools WHERE id=?", id).Scan(&raw); err != nil { if err == sql.ErrNoRows { return nil, nil } return nil, err } var p SeedPoolData json.Unmarshal([]byte(raw), &p) return &p, nil } // --- Raw accessors (valuation) --- // // The valuation pass (indexer/valuation.go) is the one owner of the derived USD // aggregates. It must read a pool/swap row, compute, and write the SAME row back // — and SeedPool/SeedSwap are INSERT OR REPLACE (full overwrite). Round-tripping // through GetPools/GetSwaps would be lossy: those return a *presentation* map // (tokens nested and dropped entirely when the token row is missing). These // accessors hand back the stored value itself so a read-modify-write preserves // every field the writer does not own. // PoolsRaw returns every stored pool keyed by id. func (s *Store) PoolsRaw() map[string]*SeedPoolData { s.mu.RLock() defer s.mu.RUnlock() out := map[string]*SeedPoolData{} rows, err := s.db.Query("SELECT id, data FROM pools") if err != nil { return out } defer rows.Close() for rows.Next() { var id, raw string if err := rows.Scan(&id, &raw); err != nil { continue } var p SeedPoolData if json.Unmarshal([]byte(raw), &p) == nil { out[id] = &p } } return out } // RecentSwapsRaw returns the `limit` most recent stored swaps keyed by id, // newest first by the indexed timestamp column. // // The bound is not an optimisation. A live chain's swap table grows without // limit — the Lux C-Chain's already spans a million blocks — so an unbounded // read would make every caller's cost grow forever with chain age. Callers that // summarise recent activity take a window; nothing needs the whole table in // memory at once. func (s *Store) RecentSwapsRaw(limit int) map[string]*SeedSwapData { s.mu.RLock() defer s.mu.RUnlock() out := map[string]*SeedSwapData{} rows, err := s.db.Query("SELECT id, data FROM swaps ORDER BY timestamp DESC LIMIT ?", limit) if err != nil { return out } defer rows.Close() for rows.Next() { var id, raw string if err := rows.Scan(&id, &raw); err != nil { continue } var sw SeedSwapData if json.Unmarshal([]byte(raw), &sw) == nil { out[id] = &sw } } return out } // UndatedSwapsRaw returns up to `limit` swaps that still hold a block number // where a time belongs, keyed by id. // // It asks for them by the property that identifies them — no swap is mined in // block 0, so an unset block is exactly the older build's row — rather than by // taking a window of the newest. The newest is the one window they are never // in: an undated row holds a block number, block numbers are small beside unix // seconds, so ordering by timestamp descending sorts every row needing repair // to the far end of the table, behind every row already correct. func (s *Store) UndatedSwapsRaw(limit int) map[string]*SeedSwapData { s.mu.RLock() defer s.mu.RUnlock() out := map[string]*SeedSwapData{} // A row's block lives in the JSON document; only id, timestamp and pool are // columns. Ascending timestamp puts the undated rows first for the same // reason descending puts them last, so the scan reaches them immediately and // the Dated test below decides. rows, err := s.db.Query("SELECT id, data FROM swaps ORDER BY timestamp ASC LIMIT ?", limit) if err != nil { return out } defer rows.Close() for rows.Next() { var id, raw string if err := rows.Scan(&id, &raw); err != nil { continue } var sw SeedSwapData if json.Unmarshal([]byte(raw), &sw) == nil && !sw.Dated() && sw.Timestamp > 0 { out[id] = &sw } } return out } // TokensRaw returns every stored token keyed by address. func (s *Store) TokensRaw() map[string]*SeedTokenData { s.mu.RLock() defer s.mu.RUnlock() out := map[string]*SeedTokenData{} rows, err := s.db.Query("SELECT id, data FROM tokens") if err != nil { return out } defer rows.Close() for rows.Next() { var id, raw string if err := rows.Scan(&id, &raw); err != nil { continue } var t SeedTokenData if json.Unmarshal([]byte(raw), &t) == nil { out[id] = &t } } return out }