# Author: Simon-Pierre Boucher
# Mail: contact@spboucher.ai
"""Storage layer — SQLite (dev, zero-config) or PostgreSQL (production).

Backend selection: DATABASE_URL env set → PostgreSQL via psycopg2;
otherwise SQLite at the given path (WAL mode so indexer threads write
while the API reads).

The DDL below is written in the dialect intersection both engines accept.
amount is TEXT: token amounts are uint256 and overflow both engines'
64-bit integers (e.g. any DAI transfer over ~9.2M). Aggregations CAST to
DOUBLE PRECISION (REAL affinity on SQLite) — fine for dashboards, not
accounting.
"""

import os
import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS chains (
    chain    TEXT PRIMARY KEY,
    family   TEXT,
    chain_id BIGINT
);

CREATE TABLE IF NOT EXISTS transfers (
    chain      TEXT    NOT NULL,
    block      BIGINT  NOT NULL,
    block_hash TEXT,
    tx_hash    TEXT    NOT NULL,
    log_index  INTEGER NOT NULL,
    timestamp  BIGINT,
    token      TEXT    NOT NULL,
    symbol     TEXT,
    "from"     TEXT,
    "to"       TEXT,
    amount     TEXT,
    decimals   INTEGER,
    PRIMARY KEY (chain, tx_hash, log_index)
);
CREATE INDEX IF NOT EXISTS idx_transfers_token_block ON transfers (chain, token, block);
CREATE INDEX IF NOT EXISTS idx_transfers_ts          ON transfers (symbol, timestamp);
CREATE INDEX IF NOT EXISTS idx_transfers_ts_only     ON transfers (timestamp);
CREATE INDEX IF NOT EXISTS idx_transfers_from        ON transfers ("from");
CREATE INDEX IF NOT EXISTS idx_transfers_to          ON transfers ("to");

-- rolling aggregates, maintained by StatsWorker (query-time scans over the
-- full window don't scale past a few million rows on free-tier hardware)
CREATE TABLE IF NOT EXISTS agg_volume (
    window    TEXT   NOT NULL,
    symbol    TEXT   NOT NULL,
    chain     TEXT   NOT NULL,
    volume    REAL,
    transfers INTEGER,
    senders   INTEGER,
    receivers INTEGER,
    updated   BIGINT,
    PRIMARY KEY (window, symbol, chain)
);

CREATE TABLE IF NOT EXISTS agg_series (
    window    TEXT   NOT NULL,
    symbol    TEXT   NOT NULL,
    chain     TEXT   NOT NULL,
    t         BIGINT NOT NULL,
    volume    REAL,
    transfers INTEGER,
    updated   BIGINT,
    PRIMARY KEY (window, symbol, chain, t)
);

-- pre-extracted large transfers (>= WHALE_TABLE_MIN_USD), scanned
-- incrementally so /whales never rescans the transfers table
CREATE TABLE IF NOT EXISTS whale_events (
    chain     TEXT    NOT NULL,
    block     BIGINT,
    tx_hash   TEXT    NOT NULL,
    log_index INTEGER NOT NULL,
    timestamp BIGINT,
    token     TEXT,
    symbol    TEXT,
    "from"    TEXT,
    "to"      TEXT,
    amount    TEXT,
    decimals  INTEGER,
    usd       REAL,
    PRIMARY KEY (chain, tx_hash, log_index)
);
CREATE INDEX IF NOT EXISTS idx_whales_ts  ON whale_events (timestamp);
CREATE INDEX IF NOT EXISTS idx_whales_usd ON whale_events (usd);

CREATE TABLE IF NOT EXISTS cursors (
    chain          TEXT PRIMARY KEY,
    last_block     BIGINT NOT NULL,
    last_hash      TEXT,
    backfill_block BIGINT,
    head_block     BIGINT
);

CREATE TABLE IF NOT EXISTS rpc_health (
    chain      TEXT NOT NULL,
    url        TEXT NOT NULL,
    score      REAL,
    ok         INTEGER,
    fail       INTEGER,
    latency_ms REAL,
    cooldown_s REAL,
    updated    BIGINT,
    PRIMARY KEY (chain, url)
);

CREATE TABLE IF NOT EXISTS tokens (
    chain    TEXT NOT NULL,
    address  TEXT NOT NULL,
    symbol   TEXT,
    decimals INTEGER,
    native   INTEGER,
    PRIMARY KEY (chain, address)
);

CREATE TABLE IF NOT EXISTS prices (
    symbol  TEXT PRIMARY KEY,
    usd     REAL,
    updated BIGINT
);

CREATE TABLE IF NOT EXISTS supply_snapshots (
    chain     TEXT   NOT NULL,
    token     TEXT   NOT NULL,
    symbol    TEXT,
    supply    TEXT,
    decimals  INTEGER,
    timestamp BIGINT NOT NULL,
    PRIMARY KEY (chain, token, timestamp)
);
CREATE INDEX IF NOT EXISTS idx_supply_ts ON supply_snapshots (symbol, timestamp);
"""

COLUMNS = [
    "chain", "block", "block_hash", "tx_hash", "log_index", "timestamp",
    "token", "symbol", "from", "to", "amount", "decimals",
]

# shared SQL fragments: face-value USD (stablecoins ≈ $1) scaled by the
# latest known price for crypto assets; CASE avoids relying on POW()
USD_EXPR = (
    "CAST(amount AS DOUBLE PRECISION) / (CASE decimals "
    + " ".join(f"WHEN {d} THEN 1e{d}" for d in range(19))
    + " ELSE 1e6 END)"
)
PRICE_JOIN = "LEFT JOIN prices ON prices.symbol = transfers.symbol"
USD_PRICED = f"(({USD_EXPR}) * COALESCE(prices.usd, 1.0))"


def is_postgres():
    return bool(os.environ.get("DATABASE_URL"))


class _PgConn:
    """Thin psycopg2 wrapper exposing the sqlite3 surface our code uses.
    Translates '?' placeholders to '%s' (no literal '?' appears in queries).
    Rows come from DictCursor: indexable by position AND by column name."""

    def __init__(self, dsn):
        import psycopg2
        import psycopg2.extras
        self._x = psycopg2.extras
        self.raw = psycopg2.connect(dsn)

    def execute(self, sql, params=()):
        cur = self.raw.cursor(cursor_factory=self._x.DictCursor)
        cur.execute(sql.replace("?", "%s"), params)
        return cur

    def executemany(self, sql, seq):
        cur = self.raw.cursor()
        cur.executemany(sql.replace("?", "%s"), list(seq))
        return cur

    def executescript(self, script):
        cur = self.raw.cursor()
        cur.execute(script)
        self.raw.commit()

    def commit(self):
        self.raw.commit()

    def close(self):
        self.raw.close()


def connect(path=None, readonly=False):
    dsn = os.environ.get("DATABASE_URL")
    if dsn:
        conn = _PgConn(dsn)
        if not readonly:
            conn.executescript(SCHEMA)
            _migrate(conn)
        return conn
    if readonly:
        conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=10)
        conn.row_factory = sqlite3.Row
        return conn
    conn = sqlite3.connect(path, timeout=30)
    conn.row_factory = sqlite3.Row  # rows usable by name AND position
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute("PRAGMA synchronous=NORMAL")
    conn.executescript(SCHEMA)
    _migrate(conn)
    return conn


def _migrate(conn):
    """Additive column migrations for DBs created by earlier versions."""
    for col in ("backfill_block", "head_block"):
        try:
            conn.execute(f"ALTER TABLE cursors ADD COLUMN {col} BIGINT")
            conn.commit()
        except Exception:
            try:  # psycopg2 aborts the tx on error — reset it
                conn.raw.rollback()
            except AttributeError:
                pass


def insert_transfers(conn, rows):
    if not rows:
        return
    cols = ", ".join(f'"{c}"' for c in COLUMNS)
    ph = ", ".join("?" for _ in COLUMNS)
    conn.executemany(
        f"INSERT INTO transfers ({cols}) VALUES ({ph}) ON CONFLICT DO NOTHING",
        [tuple(r[c] for c in COLUMNS) for r in rows],
    )
    conn.commit()


def get_cursor(conn, chain):
    row = conn.execute(
        "SELECT last_block, last_hash, backfill_block FROM cursors WHERE chain = ?",
        (chain,),
    ).fetchone()
    return (row[0], row[1], row[2]) if row else None


def set_cursor(conn, chain, block, block_hash, head=None):
    conn.execute(
        "INSERT INTO cursors (chain, last_block, last_hash, head_block) VALUES (?, ?, ?, ?) "
        "ON CONFLICT (chain) DO UPDATE SET last_block = ?, last_hash = ?, "
        "head_block = COALESCE(?, cursors.head_block)",
        (chain, block, block_hash, head, block, block_hash, head),
    )
    conn.commit()


def set_backfill(conn, chain, block):
    conn.execute(
        "UPDATE cursors SET backfill_block = ? WHERE chain = ?", (block, chain)
    )
    conn.commit()


def rollback(conn, chain, to_block):
    """Reorg: drop everything above to_block and rewind the cursor."""
    conn.execute(
        "DELETE FROM transfers WHERE chain = ? AND block > ?", (chain, to_block)
    )
    conn.execute(
        "UPDATE cursors SET last_block = ?, last_hash = NULL WHERE chain = ?",
        (to_block, chain),
    )
    conn.commit()


def save_rpc_health(conn, chain, stats, now):
    conn.executemany(
        "INSERT INTO rpc_health (chain, url, score, ok, fail, latency_ms, cooldown_s, updated) "
        "VALUES (?, ?, ?, ?, ?, ?, ?, ?) "
        "ON CONFLICT (chain, url) DO UPDATE SET score = excluded.score, ok = excluded.ok, "
        "fail = excluded.fail, latency_ms = excluded.latency_ms, "
        "cooldown_s = excluded.cooldown_s, updated = excluded.updated",
        [
            (chain, s["url"], s["score"], s["ok"], s["fail"], s["latency_ms"], s["cooldown_s"], now)
            for s in stats
        ],
    )
    conn.commit()


def upsert_chain(conn, chain, family, chain_id):
    conn.execute(
        "INSERT INTO chains (chain, family, chain_id) VALUES (?, ?, ?) "
        "ON CONFLICT (chain) DO UPDATE SET family = ?, chain_id = ?",
        (chain, family, chain_id, family, chain_id),
    )
    conn.commit()


def upsert_tokens(conn, chain, tokens):
    conn.executemany(
        "INSERT INTO tokens (chain, address, symbol, decimals, native) "
        "VALUES (?, ?, ?, ?, ?) "
        "ON CONFLICT (chain, address) DO UPDATE SET "
        "symbol = excluded.symbol, decimals = excluded.decimals, native = excluded.native",
        [
            (chain, token_key(t), t["symbol"], t.get("decimals"), int(t.get("native", False)))
            for t in tokens
        ],
    )
    conn.commit()


def insert_supply(conn, rows):
    if not rows:
        return
    conn.executemany(
        "INSERT INTO supply_snapshots (chain, token, symbol, supply, decimals, timestamp) "
        "VALUES (?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING",
        rows,
    )
    conn.commit()


def upsert_prices(conn, rows):
    """rows: [(symbol, usd, updated)]"""
    if not rows:
        return
    conn.executemany(
        "INSERT INTO prices (symbol, usd, updated) VALUES (?, ?, ?) "
        "ON CONFLICT (symbol) DO UPDATE SET usd = excluded.usd, updated = excluded.updated",
        rows,
    )
    conn.commit()


def token_key(t):
    """Canonical stored identifier: EVM addresses lowercase; everything else
    (Base58, coin types, denoms…) is case-sensitive and kept verbatim."""
    key = t.get("address") or t.get("id")
    return key.lower() if key.startswith("0x") else key