"""Deadlock timeline markers from the match record.

Deadlock has no Game State Integration. The HUD scanner reads events off the
recording instead, and for marking a timeline that is not good enough: its
kill pass is gated on a template of the player's own name, so it finds
nothing at all for whole matches — on one night it produced no clips for two
of four matches the record says were 13 and 8 kills. It also cannot be
shipped, because that name template only matches one person.

The community mirror at deadlock-api.com answers the same questions exactly,
keyless, over HTTP:

    GET /v1/players/<account_id>/match-history   every match, exact numbers
    GET /v1/matches/<match_id>/metadata          full detail, ~900 KB

Matches are tied to a recording by wall-clock overlap — the same containment
idea the CS2 path uses — and then the match clock is pinned to the recording
clock, after which every kill and objective in the record has an exact
position on the timeline.

Pinning is the hard half and it is load-bearing: an unpinned match
contributes NOTHING, deliberately, because a match whose clock is a minute
out is worse than the scanner it replaces. It is also the only part that
looks at pixels — reading the match clock off a handful of frames — using
ten 20x12 digit glyphs of the game's own clock font. Those are the same for
everyone, so unlike the scanner's name templates they ship.

This is timestamping, not statistics: there is no stats page here, and the
numbers exist only to label a marker.
"""
import gzip
import json
import math
import os
import sqlite3
import statistics
import time
import urllib.error
import urllib.request
from pathlib import Path

from clipshelf import config

API = os.getenv("DEADLOCK_API", "https://api.deadlock-api.com")
ASSETS_API = os.getenv("DEADLOCK_ASSETS", "https://assets.deadlock-api.com")
UA = "Clipshelf (VOD timestamping)"
STEAM64_BASE = 76561197960265728

# game_mode from the history rows: 4 carries brawl_score_* and averages
# ~14min, 1 is a full ~33min match.
GAME_MODES = {1: "Normal", 4: "Street Brawl"}
MODE_SLUG = {1: "deadlock", 4: "street_brawl"}

# team_objective_id -> (label, kind). Derived from the data, not documented:
# across every full-loss match only the LOSING team ever loses 0 and 9, and
# 10/11 only fall in a base loss. Lane ids run 1,3,4 / 5,7,8 / 12,14,15 —
# the 4-lane enum kept its slots when the map went to three lanes.
OBJECTIVES = {
    0: ("Core", "core"), 9: ("Patron", "patron"),
    10: ("Shrine", "shrine"), 11: ("Shrine", "shrine"),
    **{i: ("Guardian", "guardian") for i in (1, 2, 3, 4)},
    **{i: ("Walker", "walker") for i in (5, 6, 7, 8)},
    **{i: ("Base Guardian", "base") for i in (12, 13, 14, 15)},
}

# The library, not the install directory — these are caches of someone
# else's data and belong with the rest of the user's, where an uninstall
# does not take them and a reinstall does not have to refetch them.
RAW_DIR = config.DATA / "deadlock_raw"      # gzipped /metadata blobs
HERO_CACHE = config.DATA / "deadlock_heroes.json"
ASSET_TTL = 7 * 86400

# A community API carries no obligation to serve us. One call at a time,
# paced, and a hard stop when it pushes back.
REQ_GAP = float(os.getenv("DEADLOCK_REQ_GAP", "10.0"))
BLOCK_BACKOFF = float(os.getenv("DEADLOCK_BACKOFF", "3600"))
API_KEY = os.getenv("DEADLOCK_API_KEY", "")
_last_req = [0.0]
_backoff_until = [0.0]

# clock alignment
CAL_WINDOW = 6          # seconds — a vision kill this close is the same kill
CAL_MIN_HITS = 2
CAL_MIN_FRAC = 0.6
OFFSET0_RANGE = (-120.0, 300.0)

_SCHEMA_READY = False
class RateLimited(RuntimeError):
    """The API asked us to stop. Transient — never cache this as a failure.

    `per_resource` means only THIS match was refused (it isn't archived yet, so
    it needs a live Steam pull, capped at 3/h) — other matches that ARE archived
    still answer instantly from cache. Blocking everything on that would waste
    the fast path entirely."""

    def __init__(self, msg, quota=None, per_resource=False):
        super().__init__(msg)
        self.quota = quota or {}
        self.per_resource = per_resource


def backoff_left() -> int:
    return max(0, int(_backoff_until[0] - time.time()))


def _get(url: str, timeout: int = 30, pace: bool = True):
    left = backoff_left()
    if left:
        raise RateLimited(f"backing off for another {left}s")
    if pace:
        gap = REQ_GAP - (time.time() - _last_req[0])
        if gap > 0:
            time.sleep(gap)
    hdr = {"User-Agent": UA, "Accept": "application/json"}
    if API_KEY:
        hdr["X-API-KEY"] = API_KEY
    req = urllib.request.Request(url, headers=hdr)
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return json.loads(r.read().decode())
    except urllib.error.HTTPError as e:
        if e.code in (403, 429, 503):
            # the API returns a precise budget on 429, e.g.
            # {"status":429,"error":{"type":"IP","quota":{"limit":1,"period":3600},
            #  "requests":1,"remaining":0,"next_request_in":3565}}
            # — honour that instead of guessing an hour.
            wait, detail = None, ""
            try:
                body = json.loads(e.read().decode() or "{}")
                err = body.get("error") or {}
                if err.get("next_request_in") is not None:
                    wait = float(err["next_request_in"]) + 5
                    q = err.get("quota") or {}
                    detail = (f" ({err.get('type','?')} quota "
                              f"{q.get('limit')}/{q.get('period')}s)")
            except Exception:  # noqa: BLE001
                pass
            if wait is None:
                try:
                    wait = float(e.headers.get("Retry-After") or 0) or BLOCK_BACKOFF
                except Exception:
                    wait = BLOCK_BACKOFF
            # The Steam tier (3/h) refuses one unarchived match; it must not
            # pause requests for matches already in their cache.
            q = (locals().get("err") or {}).get("quota") or {}
            steam_tier = q.get("period") == 3600 and (q.get("limit") or 99) <= 5
            if not steam_tier:
                _backoff_until[0] = time.time() + wait
            raise RateLimited(f"HTTP {e.code}{detail} — "
                              + ("not archived yet, skipping" if steam_tier
                                 else f"backing off {int(wait)}s"),
                              quota=q, per_resource=steam_tier) from e
        raise
    finally:
        _last_req[0] = time.time()


def account_id_of(steamid: str | int) -> int:
    """A steamID64, a bare account_id (steamID3), or the profile URL those
    are buried in — because what a person actually has to hand is the page
    they are looking at, and telling them to go convert it themselves is
    how a setup field goes unused.

    A vanity URL (/id/name) cannot be resolved without a Steam API key, so
    it is refused by name rather than parsed into a wrong number."""
    t = str(steamid).strip().strip("/")
    if "/id/" in t:
        raise ValueError("that is a custom URL — open the profile, and use "
                         "the steamcommunity.com/profiles/<number> form "
                         "(or the number itself)")
    if "/" in t:
        t = t.rsplit("/", 1)[-1]
    n = int(t)
    return n - STEAM64_BASE if n > STEAM64_BASE else n


def heroes() -> dict:
    """{hero_id: {name, img}} — cached on disk, the asset list is ~1MB."""
    try:
        if HERO_CACHE.exists() and time.time() - HERO_CACHE.stat().st_mtime < ASSET_TTL:
            return {int(k): v for k, v in json.loads(HERO_CACHE.read_text(encoding="utf-8")).items()}
    except Exception:
        pass
    out = {}
    try:
        for h in _get(f"{ASSETS_API}/v2/heroes?only_active=true", timeout=40):
            imgs = h.get("images") or {}
            out[int(h["id"])] = {
                "name": h.get("name") or f"hero {h['id']}",
                "img": imgs.get("icon_image_small_webp") or imgs.get("icon_image_small")
                       or imgs.get("icon_hero_card_webp") or imgs.get("icon_hero_card"),
            }
        HERO_CACHE.parent.mkdir(parents=True, exist_ok=True)
        HERO_CACHE.write_text(json.dumps(out), encoding="utf-8")
    except Exception:
        # stale cache beats no names at all
        try:
            return {int(k): v for k, v in json.loads(HERO_CACHE.read_text(encoding="utf-8")).items()}
        except Exception:
            return {}
    return out


def ensure_schema(con: sqlite3.Connection):
    """The tables are created by migrations.py like every other table here.

    In the archive this function built them itself with executescript on
    every sync, which took a schema lock and lost the race against a busy
    writer often enough to abort whole syncs with "database is locked".
    It stays as a name because the call sites read better with it, and so
    a fresh connection in a worker thread cannot outrun app startup."""
    global _SCHEMA_READY
    if _SCHEMA_READY:
        return
    from clipshelf.migrations import migrate
    migrate(con)
    _mark_schema_ready()


def _mark_schema_ready():
    global _SCHEMA_READY
    _SCHEMA_READY = True


def seed_accounts(con: sqlite3.Connection) -> list[int]:
    """Whose matches to look up.

    The table is the record; settings seed it. A Steam profile URL, a
    steam64 or a bare account id all work — account_id_of takes whichever
    the user pasted, because nobody knows which one they have."""
    ensure_schema(con)
    rows = [r[0] for r in con.execute(
        "SELECT account_id FROM deadlock_accounts WHERE enabled=1")]
    if rows:
        return rows
    seeds = []
    raw = os.getenv("DEADLOCK_ACCOUNT_IDS", "") or config.CFG.get("deadlock_accounts", "")
    for part in str(raw).replace(",", " ").split():
        try:
            seeds.append(account_id_of(part))
        except Exception:
            pass
    now = int(time.time())
    for a in seeds:
        con.execute("INSERT OR IGNORE INTO deadlock_accounts(account_id,name,enabled,created_at)"
                    " VALUES(?,?,1,?)", (a, None, now))
    con.commit()
    return seeds


def fetch_history(account_id: int) -> list[dict]:
    return _get(f"{API}/v1/players/{account_id}/match-history", timeout=45)


def _load_vods(con: sqlite3.Connection) -> list:
    """All recorded nights, once. Querying per match instead meant hundreds of
    SELECTs interleaved with the UPSERTs below, which forces SQLite to upgrade
    a read transaction to a write one — and that returns SQLITE_BUSY
    IMMEDIATELY when another writer intervened, ignoring busy_timeout. Reading
    up front keeps the write phase a single short transaction."""
    return con.execute(
        "SELECT id, start_epoch, duration_ms FROM vods "
        "WHERE start_epoch > 0 AND duration_ms > 0").fetchall()


def _link_vod(vods: list, start: int, dur: int):
    """Biggest wall-clock overlap wins; returns (vod_id, ms_into_vod)."""
    end = start + max(0, dur)
    best = None
    for vid, ve, vd in vods:
        if ve > end + 600 or ve + vd / 1000 < start - 600:
            continue
        ov = min(end, ve + vd / 1000) - max(start, ve)
        if ov > 0 and (best is None or ov > best[1]):
            best = ((vid, max(0, int((start - ve) * 1000))), ov)
    return best[0] if best else (None, None)


def sync(con: sqlite3.Connection, accounts=None, verbose=False) -> dict:
    ensure_schema(con)
    accounts = accounts or seed_accounts(con)
    added = updated = 0
    for acct in accounts:
        try:
            hist = fetch_history(acct)
        except Exception as e:  # noqa: BLE001
            if verbose:
                print(f"[!] history failed for {acct}: {e}")
            continue
        vods = _load_vods(con)
        existing = {r[0] for r in con.execute(
            "SELECT match_id FROM deadlock_match_stats WHERE account_id=?", (acct,))}
        con.execute("BEGIN IMMEDIATE")      # take the write lock up front, once
        for m in hist:
            start, dur = int(m["start_time"]), int(m.get("match_duration_s") or 0)
            vod_id, vod_ms = _link_vod(vods, start, dur)
            win = 1 if m.get("player_team") == m.get("match_result") else 0
            mine, theirs = None, None
            if m.get("brawl_score_team0") is not None:
                a, b = m.get("brawl_score_team0"), m.get("brawl_score_team1")
                mine, theirs = (a, b) if m.get("player_team") == 0 else (b, a)
            row = (int(m["match_id"]), int(acct), m.get("hero_id"), start, dur,
                   m.get("game_mode"), m.get("match_mode"), m.get("player_team"),
                   m.get("match_result"), win, m.get("player_kills"),
                   m.get("player_deaths"), m.get("player_assists"),
                   m.get("net_worth"), m.get("last_hits"), m.get("denies"),
                   m.get("hero_level"), mine, theirs,
                   m.get("objectives_mask_team0"), m.get("objectives_mask_team1"),
                   1 if m.get("team_abandoned") else 0, vod_id, vod_ms,
                   int(time.time()))
            exists = row[0] in existing
            con.execute("""
                INSERT INTO deadlock_match_stats
                  (match_id,account_id,hero_id,start_time,duration_s,game_mode,
                   match_mode,player_team,match_result,win,kills,deaths,assists,
                   net_worth,last_hits,denies,hero_level,brawl_score_us,
                   brawl_score_them,objectives_mask_team0,objectives_mask_team1,
                   team_abandoned,vod_id,vod_time_ms,created_at)
                VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                ON CONFLICT(match_id,account_id) DO UPDATE SET
                   vod_id=excluded.vod_id, vod_time_ms=excluded.vod_time_ms,
                   win=excluded.win, kills=excluded.kills, deaths=excluded.deaths,
                   assists=excluded.assists, net_worth=excluded.net_worth,
                   last_hits=excluded.last_hits, denies=excluded.denies
            """, row)
            added += 0 if exists else 1
            updated += 1 if exists else 0
        con.commit()
        if verbose:
            print(f"[*] account {acct}: {len(hist)} matches seen")
    return {"added": added, "updated": updated, "accounts": list(accounts)}


def fetch_metadata(match_id: int, refresh: bool = False) -> dict:
    """Raw /metadata, gzip-cached on disk. ~880KB in, ~165KB stored."""
    import gzip
    RAW_DIR.mkdir(parents=True, exist_ok=True)
    f = RAW_DIR / f"{int(match_id)}.json.gz"
    if f.exists() and not refresh:
        try:
            with gzip.open(f, "rt") as fh:
                return json.load(fh)
        except Exception:
            f.unlink(missing_ok=True)
    d = _get(f"{API}/v1/matches/{int(match_id)}/metadata", timeout=90)
    try:
        tmp = f.with_suffix(".tmp")
        with gzip.open(tmp, "wt") as fh:
            json.dump(d, fh)
        tmp.replace(f)
    except Exception:
        pass
    return d


def pauses_of(match_id: int) -> list:
    """[(game_time_s, duration_s)] from the match record. A paused match keeps
    running in the VOD while its clock stands still, so every event after a
    pause sits that much later in the footage."""
    import gzip
    f = RAW_DIR / f"{int(match_id)}.json.gz"
    if not f.exists():
        return []
    try:
        with gzip.open(f, "rt") as fh:
            mi = (json.load(fh) or {}).get("match_info") or {}
    except Exception:
        return []
    return sorted((int(p["game_time_s"]), int(p["pause_duration_s"]))
                  for p in (mi.get("match_pauses") or [])
                  if p.get("pause_duration_s"))


def game_to_vod_ms(base_ms, offset0_s, pauses, game_t_s):
    """The whole alignment, analytically. `offset0` is the pre-game gap between
    match creation and the clock starting; the pause list supplies the rest, so
    nothing has to be estimated across the match."""
    if base_ms is None or offset0_s is None:
        return None
    held = sum(d for t, d in (pauses or []) if t < game_t_s)
    return int(base_ms + (offset0_s + game_t_s + held) * 1000)


def calibrate_match(con: sqlite3.Connection, row: dict):
    """-> (offset_s, confidence, n_matched) or (None, 0, 0) if it can't be pinned."""
    f = RAW_DIR / f"{row['match_id']}.json.gz"
    if not f.exists() or not row.get("vod_id") or row.get("vod_time_ms") is None:
        return None, 0.0, 0
    import gzip
    try:
        with gzip.open(f, "rt") as fh:
            mi = (json.load(fh) or {}).get("match_info") or {}
    except Exception:
        return None, 0.0, 0
    mine = my_kill_times(mi, row["account_id"])
    if not mine:
        return None, 0.0, 0
    base = row["vod_time_ms"] / 1000.0
    dur = row.get("duration_s") or 0
    vis = [r[0] / 1000.0 for r in con.execute(
        "SELECT vod_time_ms FROM gsi_events WHERE vod_id=? AND game='deadlock' "
        "AND event_type='kill' AND vod_time_ms BETWEEN ? AND ?",
        (row["vod_id"], row["vod_time_ms"] - 180000,
         row["vod_time_ms"] + (dur + 300) * 1000))]
    if not vis:
        return None, 0.0, 0
    best, hits = None, -1
    for lag in range(-90, 301):
        h = sum(1 for t in mine if any(abs(base + t + lag - v) <= CAL_WINDOW for v in vis))
        if h > hits:
            best, hits = lag, h
    if hits < CAL_MIN_HITS or hits < CAL_MIN_FRAC * len(mine):
        return None, 0.0, 0
    # The 1s grid search only finds the right neighbourhood. Re-centre on the
    # median residual of the pairs it matched: halves the error (~4s -> ~2s,
    # often 0) at no cost.
    #
    # NB: anchor on MY kills only. Using all 12 players' deaths gives ~5x more
    # candidates and a flatteringly small residual, but with 60-86 candidates
    # against ~20 scanner detections it locks onto coincidences — measured
    # disagreements of 64s and 157s versus this fit, at hit ratios like 4/60.
    deltas = []
    for t in mine:
        near = [v - (base + t + best) for v in vis
                if abs(base + t + best - v) <= CAL_WINDOW]
        if near:
            deltas.append(min(near, key=abs))
    if deltas:
        best = best + statistics.median(deltas)
    return round(best, 1), round(hits / len(mine), 2), hits


def calibrate_via_clock(vod_path, base_s, duration_s, pauses, samples=6,
                        verbose=False, game_mode=1):
    """offset0 by reading the in-game clock. Because the pause list is known,
    EVERY sample can be corrected back to offset0 and they must all agree —
    which is a far stronger check than the old cross-correlation, where an
    estimator could land between two pauses and report 8/8 confidence on a
    number that was wrong at both ends of the match."""
    if game_mode == 4:
        # Street Brawl's top-centre is a PER-ROUND timer over "ROUND N", not a
        # cumulative match clock — it resets every round, so a read there means
        # nothing. Two resets can coincidentally agree and mint a confident,
        # completely wrong pin (measured: +277s, +327s, +349s, +573s).
        return None, 0, 0
    try:
        from clipshelf.games import deadlock_clock as dclock
    except Exception:
        return None, 0, 0
    tpl = dclock.load_templates()
    if not tpl:
        return None, 0, 0
    votes = []
    for i in range(samples):
        frac = 0.08 + 0.84 * (i / max(1, samples - 1))
        vt = base_s + duration_s * frac
        img = dclock._grab(vod_path, vt)
        if img is None:
            continue
        c = dclock.read_clock(img, tpl)
        if c is None:
            continue
        held = sum(d for t, d in pauses if t < c)
        votes.append(vt - base_s - c - held)
    if not votes:
        return None, 0, 0
    votes.sort()
    med = votes[len(votes) // 2]
    agree = [v for v in votes if abs(v - med) <= 2.5]
    if verbose:
        print(f"      clock votes {[round(v) for v in votes]} -> {len(agree)} agree")
    if len(agree) < 2:
        return None, len(agree), len(votes)
    off0 = round(sum(agree) / len(agree), 1)
    if not (OFFSET0_RANGE[0] <= off0 <= OFFSET0_RANGE[1]):
        if verbose:
            print(f"      rejecting implausible offset0 {off0:+.0f}s")
        return None, 0, len(votes)
    return off0, len(agree), len(votes)


def brawl_rounds_of(match_id: int):
    """[round_duration_s] from the record, in order."""
    import gzip
    f = RAW_DIR / f"{int(match_id)}.json.gz"
    if not f.exists():
        return []
    try:
        with gzip.open(f, "rt") as fh:
            mi = (json.load(fh) or {}).get("match_info") or {}
    except Exception:
        return []
    return [int(x["round_duration_s"]) for x in (mi.get("street_brawl_rounds") or [])
            if x.get("round_duration_s")]


def calibrate_brawl(vod_path, base_s, duration_s, rounds, pauses, samples=None,
                    step_s=15.0, verbose=False, steps=(15.0, 7.0)):
    """offset0 for Street Brawl, where the HUD has no cumulative match clock.

    The top-centre timer counts UP during a round and DOWN during the setup
    phase between rounds. So `vod_t - timer` is CONSTANT for every sample
    inside a round (it is that round's start in VOD time) and drifts during a
    countdown — which makes the rounds fall out of the data without ever
    reading the round number.

    With round starts S1..Sn measured and the durations d1..dn from the record:

        g(k+1) = (S(k+1) - S(k)) - d(k)        each setup gap, MEASURED
        g1     = duration - sum(d) - sum(g2..gn)   closure, no assumption
        round_start_match(k) = sum(g1..gk) + sum(d1..d(k-1))
        offset0 = S(k) - base - round_start_match(k)

    Every round yields an independent offset0 and they must agree — on the
    frame-verified match all four gave +32 against a ground truth of +30.5.
    An earlier version assumed a UNIFORM gap instead of measuring each one;
    that drifted by round 3-4 and spread candidates over +-25s."""
    if not rounds:
        return None, 0, 0, []
    try:
        from clipshelf.games import deadlock_clock as dclock
    except Exception:
        return None, 0, 0, []
    tpl = dclock.load_templates()
    if not tpl:
        return None, 0, 0, []

    def _detect(step):
        seen = []
        vt = base_s
        while vt < base_s + duration_s:
            img = dclock._grab(vod_path, vt)
            t = dclock.read_clock(img, tpl) if img is not None else None
            if t is not None and t <= max(rounds) + 30:
                seen.append((vt, vt - t))
            vt += step
        # a round = a run of samples agreeing on their implied start; a
        # countdown walks its implied start forward every sample, never a run
        starts, run = [], []
        for _, st in seen:
            if run and abs(st - run[-1]) <= 2.0:
                run.append(st)
            else:
                if len(run) >= 2:
                    starts.append(sum(run) / len(run))
                run = [st]
        if len(run) >= 2:
            starts.append(sum(run) / len(run))
        out = []
        for st in starts:                  # collapse a round split by a bad read
            if out and abs(st - out[-1]) <= 6.0:
                out[-1] = (out[-1] + st) / 2
            else:
                out.append(st)
        return out

    # A round is only seen if two consecutive samples inside it both read, so a
    # patch of unreadable frames can swallow one whole round and the count
    # check then throws the match away. Rounds run 87-180s against a 15s step,
    # so the first pass is not short of chances — when it still comes up short,
    # resample finer rather than give up.
    merged = []
    for step in steps:
        merged = _detect(step)
        if verbose:
            print(f"      brawl round starts @{step:.0f}s (vod): "
                  f"{[round(x) for x in merged]} vs {len(rounds)} in the record")
        if len(merged) == len(rounds):
            break
    if len(merged) != len(rounds):
        return None, len(merged), len(rounds), []

    gaps = [0.0] * len(rounds)
    for k in range(1, len(rounds)):
        gaps[k] = (merged[k] - merged[k - 1]) - rounds[k - 1]
    gaps[0] = duration_s - sum(rounds) - sum(gaps[1:])
    if any(g < -5 or g > 200 for g in gaps):
        return None, 0, len(rounds), []         # structure doesn't hold together
    votes = []
    for k in range(len(rounds)):
        rs = sum(gaps[:k + 1]) + sum(rounds[:k])
        votes.append(merged[k] - base_s - rs
                     - sum(d for pt, d in pauses if pt < rs))
    votes.sort()
    med = votes[len(votes) // 2]
    agree = [v for v in votes if abs(v - med) <= 4.0]
    if verbose:
        print(f"      gaps {[round(g) for g in gaps]}  offset0 votes "
              f"{[round(v, 1) for v in votes]}")
    if len(agree) < 2 or not (OFFSET0_RANGE[0] <= med <= OFFSET0_RANGE[1]):
        return None, len(agree), len(votes), []
    wins = []
    for k in range(len(rounds)):
        rs = sum(gaps[:k + 1]) + sum(rounds[:k])
        wins.append([round(rs, 1), round(rs + rounds[k], 1)])
    return round(sum(agree) / len(agree), 1), len(agree), len(votes), wins


def calibrate_all(con: sqlite3.Connection, verbose=False, vod_dir=None,
                  only_missing=True) -> dict:
    """Pin every on-tape match: offset0 from the in-game clock where possible,
    else the kill cross-correlation. The pause list comes from the record."""
    ensure_schema(con)
    con.row_factory = sqlite3.Row
    vod_dir = vod_dir or os.getenv("VOD_DIR", "/mnt/holyfootage/OBS_Raw")
    q = ("SELECT m.*, v.filename FROM deadlock_match_stats m "
         "JOIN vods v ON v.id = m.vod_id WHERE m.vod_id IS NOT NULL "
         "AND m.detail IS NOT NULL")
    if only_missing:
        # Also re-pin matches that were pinned while their pause list was
        # unknown. `clock_pauses` only gets filled from the record, so a match
        # enriched AFTER it was calibrated keeps an empty list forever, and
        # `only_missing` would never look at it again. See the pz check below.
        q += (" AND (m.clock_offset0 IS NULL"
              "      OR m.clock_pauses IS NULL OR m.clock_pauses = '[]')")
    rows = [dict(r) for r in con.execute(q + " ORDER BY m.start_time DESC")]
    by_clock = by_scan = failed = repinned = 0
    pending = []                      # (…, match_id, account_id) written at the end
    for r in rows:
        # `detail IS NOT NULL` is NOT the same as "we have the record" — a
        # match whose metadata 404'd carries an {"error": ...} blob, which
        # satisfies that test while `pauses_of()` returns nothing. Calibrating
        # against an unknown pause list produced a single averaged offset that
        # looked fine (it even got saved with a confidence) but was 64s out for
        # every kill after the pause. Measured on 97741755, 2026-08-06.
        try:
            det = json.loads(r.get("detail") or "null")
        except Exception:               # noqa: BLE001
            det = None
        if not isinstance(det, dict) or det.get("error"):
            if verbose:
                print(f"    {r['match_id']}: no usable record yet — not pinning")
            continue
        pz = pauses_of(r["match_id"])
        if r.get("clock_offset0") is not None:
            # Only here because the stored pause list is empty. If the record
            # agrees it has none, the existing pin is fine — leave it alone.
            if not pz:
                continue
            repinned += 1
            if verbose:
                print(f"    {r['match_id']}: re-pinning — record has "
                      f"{len(pz)} pause(s) totalling {sum(d for _, d in pz)}s "
                      f"that the old pin ({r['clock_offset0']:+.1f}s) never saw")
        base_s = (r["vod_time_ms"] or 0) / 1000.0
        path = os.path.join(vod_dir, r["filename"] or "")
        off0, src, conf, n, rwins = None, None, 0.0, 0, None
        if os.path.exists(path):
            if (r.get("game_mode") or 1) == 4:
                # brawl: round starts fall out of the timer, gaps get measured
                rounds = brawl_rounds_of(r["match_id"])
                off0, agree, tot, wins = calibrate_brawl(
                    path, base_s, r["duration_s"] or 0, rounds, pz,
                    verbose=verbose)
                if off0 is not None:
                    src, conf, n = "rounds", round(agree / max(1, tot), 2), agree
                    rwins = wins
            else:
                off0, agree, tot = calibrate_via_clock(
                    path, base_s, r["duration_s"] or 0, pz, verbose=verbose,
                    game_mode=r.get("game_mode") or 1)
                if off0 is not None:
                    src, conf, n = "clock", round(agree / max(1, tot), 2), agree
        if off0 is None:                       # fall back to the old anchor
            o2, c2, n2 = calibrate_match(con, r)
            if o2 is not None:
                off0, src, conf, n = o2, "scan", c2, n2
        # Queue the write instead of doing it here. Pinning a match means
        # decoding and template-matching VOD frames — minutes of CPU per
        # match — and the old code wrote inside that loop, so one implicit
        # transaction stayed open across the whole run. On a clips.db this
        # busy that reliably ends in `database is locked` (it did, mid-run,
        # 2026-08-06). Same rule as sync(): hoist the reads, take the write
        # lock once, never mix.
        pending.append((off0, json.dumps(pz), src, conf, n, off0,
                        json.dumps(rwins) if rwins else None,
                        r["match_id"], r["account_id"]))
        if src in ("clock", "rounds"):
            by_clock += 1
        elif src == "scan":
            by_scan += 1
        else:
            failed += 1
        if verbose and off0 is not None:
            print(f"    {r['match_id']} vod {r['vod_id']}: offset0 {off0:+.1f}s "
                  f"via {src}, {len(pz)} pause(s) totalling {sum(d for _, d in pz)}s")

    if pending:
        con.execute("BEGIN IMMEDIATE")
        con.executemany(
            "UPDATE deadlock_match_stats SET clock_offset0=?, clock_pauses=?, "
            "clock_src=?, clock_conf=?, clock_n=?, clock_offset_s=?, "
            "clock_rounds=? WHERE match_id=? AND account_id=?", pending)
        con.commit()
    return {"clock": by_clock, "scan": by_scan, "unpinned": failed,
            "repinned": repinned, "total": len(rows)}


def vod_events(con: sqlite3.Connection, vod_id: int) -> dict:
    """Exact game events for one night, stamped in VOD time. Only matches whose
    clock could be pinned contribute events — an unaligned match would drop its
    ticks up to a minute off, which is worse than the scanner it replaces."""
    import gzip
    ensure_schema(con)
    con.row_factory = sqlite3.Row
    rows = [dict(r) for r in con.execute(
        "SELECT * FROM deadlock_match_stats WHERE vod_id=? ORDER BY start_time", (vod_id,))]
    H = heroes()
    matches, events = [], []
    for r in rows:
        off = r.get("clock_offset0") if r.get("clock_offset0") is not None \
            else r.get("clock_offset_s")
        conf = r.get("clock_conf") or 0
        aligned = off is not None
        pz = pauses_of(r["match_id"])
        # every event time goes through the pause-aware map, so a match that
        # paused doesn't drift out of sync partway through
        gv = lambda g: game_to_vod_ms(r["vod_time_ms"], off, pz, g)  # noqa: E731
        base_ms = gv(0) if aligned else (r["vod_time_ms"] or 0)
        hero = H.get(r.get("hero_id")) or {}
        matches.append({
            "match_id": r["match_id"], "hero_id": r.get("hero_id"),
            "hero": hero.get("name"), "img": hero.get("img"),
            "mode": GAME_MODES.get(r.get("game_mode"), "Other"),
            "win": bool(r.get("win")), "kills": r.get("kills"),
            "deaths": r.get("deaths"), "assists": r.get("assists"),
            "net_worth": r.get("net_worth"), "damage": r.get("damage"),
            "start_ms": base_ms, "duration_s": r.get("duration_s"),
            "end_ms": gv(r.get("duration_s") or 0) if aligned
                      else base_ms + (r.get("duration_s") or 0) * 1000,
            "aligned": aligned, "offset_s": off, "confidence": conf,
            "src": r.get("clock_src"), "paused_s": sum(d for _, d in pz),
        })
        f = RAW_DIR / f"{r['match_id']}.json.gz"
        if not aligned or not f.exists():
            continue
        try:
            with gzip.open(f, "rt") as fh:
                mi = (json.load(fh) or {}).get("match_info") or {}
        except Exception:
            continue
        me = next((p for p in (mi.get("players") or [])
                   if p.get("account_id") == r["account_id"]), None)
        if not me:
            continue
        slot = me.get("player_slot")
        by_slot = {p.get("player_slot"): p for p in (mi.get("players") or [])}
        add = lambda **kw: events.append(  # noqa: E731
            dict(match_id=r["match_id"], **kw))
        for p in (mi.get("players") or []):
            for d in (p.get("death_details") or []):
                t = d.get("game_time_s")
                if t is None:
                    continue
                if d.get("killer_player_slot") == slot and p.get("player_slot") != slot:
                    add(t_ms=gv(t), kind="kill",
                        label=f"killed {(H.get(p.get('hero_id')) or {}).get('name', '?')}",
                        hero=(H.get(p.get("hero_id")) or {}).get("name"), notable=True)
                elif p.get("player_slot") == slot:
                    k = by_slot.get(d.get("killer_player_slot"))
                    add(t_ms=gv(t), kind="death",
                        label=f"died to {(H.get((k or {}).get('hero_id')) or {}).get('name', '?')}",
                        hero=(H.get((k or {}).get("hero_id")) or {}).get("name"), notable=False)
        if (mi.get("game_mode")) != 4:
            for o in (mi.get("objectives") or []):
                t = o.get("destroyed_time_s") or 0
                if t <= 0:
                    continue
                label, kind = OBJECTIVES.get(o.get("team_objective_id"), ("Objective", "obj"))
                ours = o.get("team") != me.get("team")
                add(t_ms=gv(t), kind="objective", obj=kind,
                    label=f"{label} {'taken' if ours else 'lost'}", ours=ours,
                    notable=kind in ("patron", "core", "walker", "shrine"))
        for mb in (mi.get("mid_boss") or []):
            t = mb.get("destroyed_time_s")
            if t:
                add(t_ms=gv(t), kind="midboss",
                    label="Rejuvenator " + ("ours" if mb.get("team_claimed") == me.get("team") else "theirs"),
                    ours=mb.get("team_claimed") == me.get("team"), notable=True)
    events.sort(key=lambda e: e["t_ms"])
    return {"vod_id": vod_id, "matches": matches, "events": events,
            "aligned_matches": sum(1 for m in matches if m["aligned"])}


def my_kill_times(mi: dict, account_id: int) -> list:
    """Exact game-times of the player's own takedowns, ascending. These live in
    the VICTIMS' `death_details`, not the player's own (his own list is the
    times he died)."""
    ps = mi.get("players") or []
    me = next((p for p in ps if p.get("account_id") == account_id), None)
    if not me:
        return []
    slot = me.get("player_slot")
    return sorted(d["game_time_s"] for p in ps
                  for d in (p.get("death_details") or [])
                  if d.get("game_time_s") is not None
                  and d.get("killer_player_slot") == slot
                  and p.get("player_slot") != slot)


def vod_kills(con: sqlite3.Connection, vod_id: int, max_gap_s: float = 10.0,
              min_kills: int = 1) -> list:
    """The player's kill chains for a night, from the RECORD, stamped in VOD
    time. Same chaining model as the scanner's kill cutter (group while the gap
    to the next kill is <= max_gap_s) so the clips are the same shape — only
    the times are exact instead of vision-detected.

    This exists because the HUD scanner's kill pass is gated on a name-template
    match and silently finds nothing for whole matches: on vod 1640 it produced
    0 clips for two of four matches that the record says were 13 and 8 kills.
    Pinned matches only — an unaligned match would cut the wrong minute."""
    import gzip
    ensure_schema(con)
    con.row_factory = sqlite3.Row
    rows = [dict(r) for r in con.execute(
        "SELECT * FROM deadlock_match_stats WHERE vod_id=? "
        "AND COALESCE(clock_offset0, clock_offset_s) IS NOT NULL "
        "ORDER BY start_time", (vod_id,))]
    H, out = heroes(), []
    for r in rows:
        f = RAW_DIR / f"{r['match_id']}.json.gz"
        if not f.exists():
            continue
        try:
            with gzip.open(f, "rt") as fh:
                mi = (json.load(fh) or {}).get("match_info") or {}
        except Exception:
            continue
        kt = my_kill_times(mi, r["account_id"])
        if not kt:
            continue
        off = r["clock_offset0"] if r["clock_offset0"] is not None \
            else r["clock_offset_s"]
        pz = pauses_of(r["match_id"])
        chains = []
        for t in kt:
            if chains and t - chains[-1][-1] <= max_gap_s:
                chains[-1].append(t)
            else:
                chains.append([t])
        for ch in chains:
            if len(ch) < min_kills:
                continue
            out.append({
                "match_id": r["match_id"], "vod_id": vod_id,
                "hero": (H.get(r.get("hero_id")) or {}).get("name"),
                "mode": GAME_MODES.get(r.get("game_mode"), "Other"),
                "mode_slug": MODE_SLUG.get(r.get("game_mode"), "deadlock"),
                "start_ms": game_to_vod_ms(r["vod_time_ms"], off, pz, ch[0]),
                "end_ms": game_to_vod_ms(r["vod_time_ms"], off, pz, ch[-1]),
                "match_t_start": ch[0], "match_t_end": ch[-1],
                "kills": len(ch),
                # every takedown in VOD ms, for per-kill overlays downstream
                "kill_times": [game_to_vod_ms(r["vod_time_ms"], off, pz, t)
                               for t in ch],
            })
    out.sort(key=lambda x: x["start_ms"])
    return out


# ------------------------------------------------------- into the timeline
# Everything above is the archive's code, which answered questions. This is
# the only part Clipshelf needs: putting the answers on the timeline it
# already draws. Deadlock events land in gsi_events beside CS2's, so the
# scrubber, the chapter brackets and the event pins need no special case —
# a marker is a marker regardless of how it was found.

MARKER_SOURCE = "record"          # vs the HUD scanner's "vision"


def materialize(con: sqlite3.Connection, vod_id: int) -> dict:
    """Write one night's aligned Deadlock events into the shared tables.

    Idempotent by deletion: every run clears what a previous run wrote for
    this recording first. Re-running after a better clock pin must correct
    the timeline, not double it — and the alignment genuinely does improve
    on a second pass, so this will be re-run.

    Only ever touches rows it wrote (game='deadlock' AND the record tag),
    so a hand-placed mark or a scanner event on the same night survives."""
    ensure_schema(con)
    payload = vod_events(con, vod_id)
    aligned = [m for m in payload["matches"] if m["aligned"]]

    row = con.execute("SELECT start_epoch FROM vods WHERE id=?", (vod_id,)).fetchone()
    if row is None:
        return {"vod_id": vod_id, "chapters": 0, "events": 0, "skipped": "no such recording"}
    start_epoch = row[0]

    con.execute("BEGIN IMMEDIATE")
    try:
        con.execute("DELETE FROM gsi_events WHERE vod_id=? AND game='deadlock'"
                    " AND detail LIKE ?", (vod_id, f'%"src": "{MARKER_SOURCE}"%'))
        con.execute("DELETE FROM vod_chapters WHERE vod_id=? AND game_tag IN"
                    " ('deadlock','street_brawl')", (vod_id,))

        for m in aligned:
            tag = MODE_SLUG.get(m.get("mode") == "Street Brawl" and 4 or 1, "deadlock")
            con.execute(
                "INSERT INTO vod_chapters (vod_id, game_tag, start_time_ms, epoch_ms)"
                " VALUES (?,?,?,?)",
                (vod_id, tag, m["start_ms"], start_epoch * 1000 + (m["start_ms"] or 0)))

        n = 0
        for e in payload["events"]:
            t = e.get("t_ms")
            if t is None:
                continue
            detail = dict(e)
            detail["src"] = MARKER_SOURCE
            con.execute(
                "INSERT INTO gsi_events (vod_id, game, event_type, title, detail,"
                " epoch_ms, vod_time_ms) VALUES (?,?,?,?,?,?,?)",
                (vod_id, "deadlock", e.get("kind") or "event", e.get("label") or "",
                 json.dumps(detail), start_epoch * 1000 + t, t))
            n += 1
        con.commit()
    except Exception:
        con.rollback()
        raise
    return {"vod_id": vod_id, "chapters": len(aligned), "events": n,
            "unaligned": len(payload["matches"]) - len(aligned)}


def ensure_metadata(con: sqlite3.Connection, vod_id: int, limit: int = 12) -> int:
    """Pull the /metadata blobs a night's markers need, and no others.

    The archive fetched these as part of building stats. Here they exist
    only because kill and objective times live in them — so fetch exactly
    the matches on this recording, newest first, and stop."""
    ensure_schema(con)
    rows = con.execute(
        "SELECT match_id FROM deadlock_match_stats WHERE vod_id=?"
        " ORDER BY start_time DESC LIMIT ?", (vod_id, limit)).fetchall()
    got = 0
    for (mid,) in rows:
        if (RAW_DIR / f"{int(mid)}.json.gz").exists():
            continue
        try:
            fetch_metadata(mid)
            got += 1
        except RateLimited:
            break              # the API asked us to stop; try again next cycle
        except Exception:
            continue
    return got
