spb/coinexplorer Public MIT

1# Author: Simon-Pierre Boucher2# Mail: contact@spboucher.ai3"""Storage layer — SQLite (dev, zero-config) or PostgreSQL (production).45Backend selection: DATABASE_URL env set → PostgreSQL via psycopg2;6otherwise SQLite at the given path (WAL mode so indexer threads write7while the API reads).89The DDL below is written in the dialect intersection both engines accept.10amount is TEXT: token amounts are uint256 and overflow both engines'1164-bit integers (e.g. any DAI transfer over ~9.2M). Aggregations CAST to12DOUBLE PRECISION (REAL affinity on SQLite) — fine for dashboards, not13accounting.14"""1516import os17import sqlite31819SCHEMA = """20CREATE TABLE IF NOT EXISTS chains (21    chain    TEXT PRIMARY KEY,22    family   TEXT,23    chain_id BIGINT24);2526CREATE TABLE IF NOT EXISTS transfers (27    chain      TEXT    NOT NULL,28    block      BIGINT  NOT NULL,29    block_hash TEXT,30    tx_hash    TEXT    NOT NULL,31    log_index  INTEGER NOT NULL,32    timestamp  BIGINT,33    token      TEXT    NOT NULL,34    symbol     TEXT,35    "from"     TEXT,36    "to"       TEXT,37    amount     TEXT,38    decimals   INTEGER,39    PRIMARY KEY (chain, tx_hash, log_index)40);41CREATE INDEX IF NOT EXISTS idx_transfers_token_block ON transfers (chain, token, block);42CREATE INDEX IF NOT EXISTS idx_transfers_ts          ON transfers (symbol, timestamp);43CREATE INDEX IF NOT EXISTS idx_transfers_ts_only     ON transfers (timestamp);44CREATE INDEX IF NOT EXISTS idx_transfers_from        ON transfers ("from");45CREATE INDEX IF NOT EXISTS idx_transfers_to          ON transfers ("to");4647-- rolling aggregates, maintained by StatsWorker (query-time scans over the48-- full window don't scale past a few million rows on free-tier hardware)49CREATE TABLE IF NOT EXISTS agg_volume (50    window    TEXT   NOT NULL,51    symbol    TEXT   NOT NULL,52    chain     TEXT   NOT NULL,53    volume    REAL,54    transfers INTEGER,55    senders   INTEGER,56    receivers INTEGER,57    updated   BIGINT,58    PRIMARY KEY (window, symbol, chain)59);6061CREATE TABLE IF NOT EXISTS agg_series (62    window    TEXT   NOT NULL,63    symbol    TEXT   NOT NULL,64    chain     TEXT   NOT NULL,65    t         BIGINT NOT NULL,66    volume    REAL,67    transfers INTEGER,68    updated   BIGINT,69    PRIMARY KEY (window, symbol, chain, t)70);7172-- pre-extracted large transfers (>= WHALE_TABLE_MIN_USD), scanned73-- incrementally so /whales never rescans the transfers table74CREATE TABLE IF NOT EXISTS whale_events (75    chain     TEXT    NOT NULL,76    block     BIGINT,77    tx_hash   TEXT    NOT NULL,78    log_index INTEGER NOT NULL,79    timestamp BIGINT,80    token     TEXT,81    symbol    TEXT,82    "from"    TEXT,83    "to"      TEXT,84    amount    TEXT,85    decimals  INTEGER,86    usd       REAL,87    PRIMARY KEY (chain, tx_hash, log_index)88);89CREATE INDEX IF NOT EXISTS idx_whales_ts  ON whale_events (timestamp);90CREATE INDEX IF NOT EXISTS idx_whales_usd ON whale_events (usd);9192CREATE TABLE IF NOT EXISTS cursors (93    chain          TEXT PRIMARY KEY,94    last_block     BIGINT NOT NULL,95    last_hash      TEXT,96    backfill_block BIGINT,97    head_block     BIGINT98);99100CREATE TABLE IF NOT EXISTS rpc_health (101    chain      TEXT NOT NULL,102    url        TEXT NOT NULL,103    score      REAL,104    ok         INTEGER,105    fail       INTEGER,106    latency_ms REAL,107    cooldown_s REAL,108    updated    BIGINT,109    PRIMARY KEY (chain, url)110);111112CREATE TABLE IF NOT EXISTS tokens (113    chain    TEXT NOT NULL,114    address  TEXT NOT NULL,115    symbol   TEXT,116    decimals INTEGER,117    native   INTEGER,118    PRIMARY KEY (chain, address)119);120121CREATE TABLE IF NOT EXISTS prices (122    symbol  TEXT PRIMARY KEY,123    usd     REAL,124    updated BIGINT125);126127CREATE TABLE IF NOT EXISTS supply_snapshots (128    chain     TEXT   NOT NULL,129    token     TEXT   NOT NULL,130    symbol    TEXT,131    supply    TEXT,132    decimals  INTEGER,133    timestamp BIGINT NOT NULL,134    PRIMARY KEY (chain, token, timestamp)135);136CREATE INDEX IF NOT EXISTS idx_supply_ts ON supply_snapshots (symbol, timestamp);137"""138139COLUMNS = [140    "chain", "block", "block_hash", "tx_hash", "log_index", "timestamp",141    "token", "symbol", "from", "to", "amount", "decimals",142]143144# shared SQL fragments: face-value USD (stablecoins ≈ $1) scaled by the145# latest known price for crypto assets; CASE avoids relying on POW()146USD_EXPR = (147    "CAST(amount AS DOUBLE PRECISION) / (CASE decimals "148    + " ".join(f"WHEN {d} THEN 1e{d}" for d in range(19))149    + " ELSE 1e6 END)"150)151PRICE_JOIN = "LEFT JOIN prices ON prices.symbol = transfers.symbol"152USD_PRICED = f"(({USD_EXPR}) * COALESCE(prices.usd, 1.0))"153154155def is_postgres():156    return bool(os.environ.get("DATABASE_URL"))157158159class _PgConn:160    """Thin psycopg2 wrapper exposing the sqlite3 surface our code uses.161    Translates '?' placeholders to '%s' (no literal '?' appears in queries).162    Rows come from DictCursor: indexable by position AND by column name."""163164    def __init__(self, dsn):165        import psycopg2166        import psycopg2.extras167        self._x = psycopg2.extras168        self.raw = psycopg2.connect(dsn)169170    def execute(self, sql, params=()):171        cur = self.raw.cursor(cursor_factory=self._x.DictCursor)172        cur.execute(sql.replace("?", "%s"), params)173        return cur174175    def executemany(self, sql, seq):176        cur = self.raw.cursor()177        cur.executemany(sql.replace("?", "%s"), list(seq))178        return cur179180    def executescript(self, script):181        cur = self.raw.cursor()182        cur.execute(script)183        self.raw.commit()184185    def commit(self):186        self.raw.commit()187188    def close(self):189        self.raw.close()190191192def connect(path=None, readonly=False):193    dsn = os.environ.get("DATABASE_URL")194    if dsn:195        conn = _PgConn(dsn)196        if not readonly:197            conn.executescript(SCHEMA)198            _migrate(conn)199        return conn200    if readonly:201        conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=10)202        conn.row_factory = sqlite3.Row203        return conn204    conn = sqlite3.connect(path, timeout=30)205    conn.row_factory = sqlite3.Row  # rows usable by name AND position206    conn.execute("PRAGMA journal_mode=WAL")207    conn.execute("PRAGMA synchronous=NORMAL")208    conn.executescript(SCHEMA)209    _migrate(conn)210    return conn211212213def _migrate(conn):214    """Additive column migrations for DBs created by earlier versions."""215    for col in ("backfill_block", "head_block"):216        try:217            conn.execute(f"ALTER TABLE cursors ADD COLUMN {col} BIGINT")218            conn.commit()219        except Exception:220            try:  # psycopg2 aborts the tx on error — reset it221                conn.raw.rollback()222            except AttributeError:223                pass224225226def insert_transfers(conn, rows):227    if not rows:228        return229    cols = ", ".join(f'"{c}"' for c in COLUMNS)230    ph = ", ".join("?" for _ in COLUMNS)231    conn.executemany(232        f"INSERT INTO transfers ({cols}) VALUES ({ph}) ON CONFLICT DO NOTHING",233        [tuple(r[c] for c in COLUMNS) for r in rows],234    )235    conn.commit()236237238def get_cursor(conn, chain):239    row = conn.execute(240        "SELECT last_block, last_hash, backfill_block FROM cursors WHERE chain = ?",241        (chain,),242    ).fetchone()243    return (row[0], row[1], row[2]) if row else None244245246def set_cursor(conn, chain, block, block_hash, head=None):247    conn.execute(248        "INSERT INTO cursors (chain, last_block, last_hash, head_block) VALUES (?, ?, ?, ?) "249        "ON CONFLICT (chain) DO UPDATE SET last_block = ?, last_hash = ?, "250        "head_block = COALESCE(?, cursors.head_block)",251        (chain, block, block_hash, head, block, block_hash, head),252    )253    conn.commit()254255256def set_backfill(conn, chain, block):257    conn.execute(258        "UPDATE cursors SET backfill_block = ? WHERE chain = ?", (block, chain)259    )260    conn.commit()261262263def rollback(conn, chain, to_block):264    """Reorg: drop everything above to_block and rewind the cursor."""265    conn.execute(266        "DELETE FROM transfers WHERE chain = ? AND block > ?", (chain, to_block)267    )268    conn.execute(269        "UPDATE cursors SET last_block = ?, last_hash = NULL WHERE chain = ?",270        (to_block, chain),271    )272    conn.commit()273274275def save_rpc_health(conn, chain, stats, now):276    conn.executemany(277        "INSERT INTO rpc_health (chain, url, score, ok, fail, latency_ms, cooldown_s, updated) "278        "VALUES (?, ?, ?, ?, ?, ?, ?, ?) "279        "ON CONFLICT (chain, url) DO UPDATE SET score = excluded.score, ok = excluded.ok, "280        "fail = excluded.fail, latency_ms = excluded.latency_ms, "281        "cooldown_s = excluded.cooldown_s, updated = excluded.updated",282        [283            (chain, s["url"], s["score"], s["ok"], s["fail"], s["latency_ms"], s["cooldown_s"], now)284            for s in stats285        ],286    )287    conn.commit()288289290def upsert_chain(conn, chain, family, chain_id):291    conn.execute(292        "INSERT INTO chains (chain, family, chain_id) VALUES (?, ?, ?) "293        "ON CONFLICT (chain) DO UPDATE SET family = ?, chain_id = ?",294        (chain, family, chain_id, family, chain_id),295    )296    conn.commit()297298299def upsert_tokens(conn, chain, tokens):300    conn.executemany(301        "INSERT INTO tokens (chain, address, symbol, decimals, native) "302        "VALUES (?, ?, ?, ?, ?) "303        "ON CONFLICT (chain, address) DO UPDATE SET "304        "symbol = excluded.symbol, decimals = excluded.decimals, native = excluded.native",305        [306            (chain, token_key(t), t["symbol"], t.get("decimals"), int(t.get("native", False)))307            for t in tokens308        ],309    )310    conn.commit()311312313def insert_supply(conn, rows):314    if not rows:315        return316    conn.executemany(317        "INSERT INTO supply_snapshots (chain, token, symbol, supply, decimals, timestamp) "318        "VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING",319        rows,320    )321    conn.commit()322323324def upsert_prices(conn, rows):325    """rows: [(symbol, usd, updated)]"""326    if not rows:327        return328    conn.executemany(329        "INSERT INTO prices (symbol, usd, updated) VALUES (?, ?, ?) "330        "ON CONFLICT (symbol) DO UPDATE SET usd = excluded.usd, updated = excluded.updated",331        rows,332    )333    conn.commit()334335336def token_key(t):337    """Canonical stored identifier: EVM addresses lowercase; everything else338    (Base58, coin types, denoms…) is case-sensitive and kept verbatim."""339    key = t.get("address") or t.get("id")340    return key.lower() if key.startswith("0x") else key341