winning-wallet-finder/recorder/ingest.py

#!/usr/bin/env python3
"""Nightly tape ingest (daily.sh): pull closed RTDS segments off the
wwf-recorder volume into live/rtds.duckdb, delete on the box ONLY after the
inserted row count matches. Own DB file on purpose — must never contend with
cache.duckdb's single-writer lock (2026-07-17 collision). Transport is
`flyctl ssh console` + base64 (no ingress on the recorder by design)."""
import base64
import gzip
import json
import os
import shutil
import ssl
import subprocess
import time
import urllib.request

HERE = os.path.dirname(os.path.abspath(__file__))
DB = os.path.join(HERE, "..", "live", "rtds.duckdb")
APP, SEG = "wwf-recorder", "/data/segments"


def ping(msg):
    """Best-effort Discord ping to the DAILY channel (user ask 2026-07-19:
    the nightly pull announces start + finish). Never fatal."""
    try:
        hook = json.load(open(os.path.join(HERE, "..", "config.json"))).get("daily_webhook")
        if not hook:
            return
        req = urllib.request.Request(hook, data=json.dumps({"content": msg}).encode(),
                                     headers={"Content-Type": "application/json",
                                              "User-Agent": "Mozilla/5.0"})
        for attempt in range(5):   # ride out the post-wake DNS gap (2026-07-20)
            try:
                urllib.request.urlopen(req, timeout=10,
                                       context=ssl._create_unverified_context()).read()
                return
            except Exception:
                if attempt < 4:
                    time.sleep(5 * (2 ** attempt))
    except Exception:
        pass


# launchd runs daily.sh with a minimal PATH (no /opt/homebrew/bin) — the
# 2026-07-18 run died on FileNotFoundError: 'flyctl'. Resolve it explicitly.
FLYCTL = shutil.which("flyctl") or "/opt/homebrew/bin/flyctl"


def box(cmd):
    # stdin=DEVNULL: under launchd there is no tty, and any flyctl subcommand
    # that reads stdin will otherwise hang or read garbage (2026-07-20).
    r = subprocess.run([FLYCTL, "ssh", "console", "-a", APP, "-C",
                        f"bash -c '{cmd}'"], capture_output=True, text=True,
                       timeout=900, stdin=subprocess.DEVNULL)
    return r.stdout


def main():
    import duckdb
    con = duckdb.connect(DB)
    con.execute("""CREATE TABLE IF NOT EXISTS trades(
        ts DOUBLE, wallet VARCHAR, asset VARCHAR, cond VARCHAR, side VARCHAR,
        price DOUBLE, size DOUBLE, tx VARCHAR, title VARCHAR)""")
    con.execute("""CREATE TABLE IF NOT EXISTS ingested(
        segment VARCHAR PRIMARY KEY, rows BIGINT, ingested_at BIGINT)""")
    con.execute("""CREATE TABLE IF NOT EXISTS aux(
        ts DOUBLE, topic VARCHAR, type VARCHAR, payload VARCHAR)""")
    have = {r[0] for r in con.execute("SELECT segment FROM ingested").fetchall()}
    segs = [s for s in box(f"ls {SEG}").split()
            if s.endswith(".gz") and s not in have]
    ping(f"📼 tape ingest started — {len(segs)} segment(s) queued")
    total = 0
    for s in sorted(segs):
        # closes #11: sftp moves the gz BINARY in one session (base64-over-
        # console was +33% bytes and needed 900s timeouts at busy-hour sizes);
        # the old path stays as the fallback because sftp exits 0 even on
        # some failures — the gunzip is the integrity check either way.
        lines, fsize = None, 0
        local = os.path.join("/tmp", s)
        try:
            # stdin=DEVNULL is LOAD-BEARING (2026-07-20): without it, sftp
            # under launchd (no tty) produced empty/partial files with rc 0,
            # which decoded to [] and — via the missing guard below — got
            # marked ingested and DELETED. 36 segments were lost this way.
            subprocess.run([FLYCTL, "ssh", "sftp", "get", f"{SEG}/{s}", local,
                            "-a", APP], capture_output=True, timeout=900,
                           stdin=subprocess.DEVNULL)
            fsize = os.path.getsize(local)
            with gzip.open(local, "rb") as fh:
                lines = fh.read().decode().splitlines()
        except Exception:
            lines = None
        finally:
            try:
                os.remove(local)
            except OSError:
                pass
        if lines is None:
            try:
                raw = box(f"base64 {SEG}/{s}")
                blob = base64.b64decode(raw)
                fsize = len(blob)
                lines = gzip.decompress(blob).decode().splitlines()
            except Exception as e:
                print(f"[ingest] {s}: both transports failed ({type(e).__name__}) "
                      "— left on box, continuing")
                continue
        # INTEGRITY GUARD (2026-07-20): a non-trivial gz that yields ZERO lines
        # is a truncated/empty FETCH, not an empty hour (a genuinely empty hour
        # gzips to ~20-30 bytes). NEVER mark+delete it — leave it on the box to
        # retry. This is the guard whose absence cost 36 segments.
        if not lines and fsize > 200:
            print(f"[ingest] {s}: {fsize}B gz decoded to 0 lines — FETCH FAILED, "
                  "left on box for retry")
            continue
        rows, aux = [], []
        for ln in lines:
            try:
                d = json.loads(ln)
                if s.startswith("aux_"):
                    aux.append((d.get("ts"), d.get("topic"), d.get("type"),
                                json.dumps(d.get("payload"))))
                else:
                    rows.append((d.get("ts"), d.get("wallet"), d.get("asset"),
                                 d.get("cond"), d.get("side"), d.get("price"),
                                 d.get("size"), d.get("tx"), d.get("title")))
            except Exception:
                pass
        con.execute("BEGIN")           # audit 3.10: atomic pair — a crash
        if rows:                        # between the two inserts re-ingested
            con.executemany("INSERT INTO trades VALUES (?,?,?,?,?,?,?,?,?)", rows)
        if aux:
            con.executemany("INSERT INTO aux VALUES (?,?,?,?)", aux)
        con.execute("INSERT INTO ingested VALUES (?,?,?)",
                    [s, len(rows) + len(aux), int(time.time())])
        con.execute("COMMIT")
        box(f"rm {SEG}/{s}")
        total += len(rows) + len(aux)
        print(f"[ingest] {s}: {len(rows)} trades + {len(aux)} aux")
    n = con.execute("SELECT count(*) FROM trades").fetchone()[0]
    print(f"[ingest] +{total} rows · rtds.duckdb now {n:,} trades")
    ping(f"📼 tape ingest done: +{total:,} rows across {len(segs)} segment(s) "
         f"· rtds.duckdb now {n:,} trades")


if __name__ == "__main__":
    main()