"""Provisional (盘中观察) serving: quotes, index quotes, intraday points.

Free sources only. Never writes official eod_* tables. Uses rt_cache + LKG.
"""

from __future__ import annotations

import hashlib
import json
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
from typing import Any

from datahub.adapters.base import AdapterError
from datahub.adapters.eastmoney import EastmoneyAdapter
from datahub.adapters.tencent import TencentAdapter
from datahub.codes import resolve_code
from datahub.db import HubDB
from datahub.governance.lkg import LastKnownGood
from datahub.timeutil import isoformat, now_shanghai, yyyymmdd

QUOTE_TTL = 60
INDEX_TTL = 60
INTRADAY_TTL = 20
QUOTE_BATCH = 60


class RealtimeApiError(RuntimeError):
    def __init__(self, code: str, message: str) -> None:
        super().__init__(message)
        self.code = code
        self.message = message


def _envelope(data: Any, meta: dict[str, Any]) -> dict[str, Any]:
    from datahub import SCHEMA_VERSION

    return {"schema_version": SCHEMA_VERSION, "data": data, "meta": meta}


def fetch_index_quotes(db: HubDB) -> dict[str, Any]:
    cache_key = "indexes:quotes"
    cached = _read_cache(db, cache_key)
    if cached is not None:
        return cached
    eastmoney = EastmoneyAdapter()
    try:
        rows = eastmoney.fetch_indices()
        source = "eastmoney:ulist"
    except Exception:
        rows = TencentAdapter().fetch_indices()
        source = "tencent:qt"
    if len(rows) < 3:
        raise RealtimeApiError("SOURCE_UNAVAILABLE", "index quotes incomplete")
    payload = _envelope(
        rows,
        {
            "tier": "provisional",
            "trade_date": yyyymmdd(now_shanghai()),
            "source": source,
            "stale": False,
            "staleness_seconds": 0,
            "published_at": isoformat(now_shanghai()),
        },
    )
    _write_cache(db, cache_key, payload, INDEX_TTL, source)
    return payload


def fetch_market_quotes(db: HubDB) -> dict[str, Any]:
    cache_key = "quotes:market"
    cached = _read_cache(db, cache_key)
    if cached is not None:
        return cached
    errors: list[str] = []
    rows: list[dict[str, Any]] = []
    source = ""
    try:
        rows = EastmoneyAdapter().fetch_market_quotes()
        source = "eastmoney:clist"
    except Exception as exc:
        errors.append(f"eastmoney:{exc}")
        try:
            listed = _listed_ts_codes(db)
            if not listed:
                raise AdapterError("no local stock master for tencent market snapshot")
            rows = _tencent_named_quotes(listed)
            if len(rows) < 200:
                raise AdapterError(f"Tencent market snapshot too small: {len(rows)}")
            source = "tencent:qt"
        except Exception as backup_exc:
            errors.append(f"tencent:{backup_exc}")
            recovered = _load_quotes_lkg(db, cache_key, yyyymmdd(now_shanghai()))
            if recovered is not None:
                return recovered
            raise RealtimeApiError(
                "SOURCE_UNAVAILABLE",
                "market quotes unavailable: " + "；".join(errors),
            ) from backup_exc
    payload = _quote_payload(rows, source, scope="market")
    _write_cache(db, cache_key, payload, QUOTE_TTL, source)
    return payload


def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]:
    if not codes:
        return fetch_market_quotes(db)
    resolved: list[str] = []
    seen: set[str] = set()
    for code in codes:
        item = resolve_code(db, code) or _guess_ts_code(code)
        if item and item not in seen:
            seen.add(item)
            resolved.append(item)
    if not resolved:
        raise RealtimeApiError("INVALID_ARGUMENT", "no resolvable codes")
    digest = hashlib.sha1(",".join(sorted(resolved)).encode("utf-8")).hexdigest()
    cache_key = f"quotes:{digest}:{len(resolved)}"
    cached = _read_cache(db, cache_key)
    if cached is not None and _quote_codes(cached.get("data") or []) >= set(resolved):
        return cached
    minimum = max(1, int(len(resolved) * 0.5))
    errors: list[str] = []
    by_code = _quote_map((cached or {}).get("data") or [], resolved)
    sources: list[str] = []
    cached_source = str(((cached or {}).get("meta") or {}).get("source") or "")
    if by_code and cached_source:
        sources.append(cached_source)

    missing = [code for code in resolved if code not in by_code]
    try:
        rows = _eastmoney_named_quotes(missing)
        by_code.update(_quote_map(rows, missing))
        if rows:
            sources.append("eastmoney:ulist")
    except Exception as exc:
        errors.append(f"eastmoney:{exc}")

    missing = [code for code in resolved if code not in by_code]
    if missing:
        try:
            rows = _tencent_named_quotes(missing)
            by_code.update(_quote_map(rows, missing))
            if rows:
                sources.append("tencent:qt")
        except Exception as backup_exc:
            errors.append(f"tencent:{backup_exc}")

    fresh_rows = [by_code[code] for code in resolved if code in by_code]
    source = "+".join(dict.fromkeys(sources)) or "unavailable"
    if fresh_rows:
        _store_quote_rows_lkg(db, fresh_rows, source)

    missing = [code for code in resolved if code not in by_code]
    today = yyyymmdd(now_shanghai())
    recovered_rows, stale_age = _recover_quote_rows(db, missing, today)
    by_code.update(_quote_map(recovered_rows, missing))
    missing = [code for code in resolved if code not in by_code]
    if missing:
        group_lkg = _load_quotes_lkg(db, cache_key, today)
        group_rows = _quote_map((group_lkg or {}).get("data") or [], missing)
        by_code.update(group_rows)
        if group_rows:
            recovered_rows.extend(group_rows.values())
            stale_age = max(
                stale_age,
                int(((group_lkg or {}).get("meta") or {}).get("staleness_seconds") or 0),
            )
    rows = [by_code[code] for code in resolved if code in by_code]
    if len(rows) < minimum:
        detail = "；".join(errors) or f"only {len(rows)}/{len(resolved)} quotes returned"
        raise RealtimeApiError("SOURCE_UNAVAILABLE", "quotes unavailable: " + detail)

    if recovered_rows:
        source = "+".join(dict.fromkeys([*sources, "same-day-lkg"]))
    payload = _quote_payload(rows, source)
    payload["meta"].update({
        "requested_count": len(resolved),
        "returned_count": len(rows),
        "complete": len(rows) == len(resolved),
        "missing_codes": [code for code in resolved if code not in by_code],
    })
    if recovered_rows:
        payload["meta"].update({
            "stale": True,
            "staleness_seconds": stale_age,
            "delay_notice": f"主备免费行情暂不完整，已用当天 {stale_age} 秒前的真实快照补齐",
        })
    else:
        _write_cache(db, cache_key, payload, QUOTE_TTL, source)
    return payload


def fetch_sector_quote(db: HubDB, code: str, expected_date: str = "") -> dict[str, Any]:
    ts_code = str(code or "").strip().upper()
    if ts_code.isdigit():
        ts_code = f"{ts_code}.SI"
    cache_key = f"sector:{ts_code}"
    canonical_name = _canonical_sw_name(db, ts_code)
    cached = _read_cache(db, cache_key)
    if cached is not None and _sector_payload_matches(cached, canonical_name):
        return cached
    errors: list[str] = []
    try:
        row = EastmoneyAdapter().fetch_shenwan_quote(ts_code)
        if not _sector_row_matches(row, canonical_name):
            raise AdapterError(
                f"industry name mismatch: expected {canonical_name}, got {row.get('name') or '--'}"
            )
        source = str(row.get("source") or "eastmoney_sw")
    except Exception as exc:
        errors.append(f"eastmoney:{exc}")
        recovered = _load_quotes_lkg(db, cache_key, expected_date)
        if recovered is not None and _sector_payload_matches(recovered, canonical_name):
            return recovered
        raise RealtimeApiError(
            "SOURCE_UNAVAILABLE",
            "sector quote unavailable: " + "；".join(errors),
        ) from exc
    want = str(expected_date or "").replace("-", "")[:8]
    quote_date = str(row.get("quote_date") or "")
    if want and quote_date and quote_date != want:
        recovered = _load_quotes_lkg(db, cache_key, want)
        if recovered is not None and _sector_payload_matches(recovered, canonical_name):
            return recovered
        raise RealtimeApiError("SOURCE_UNAVAILABLE", f"sector quote date {quote_date} != {want}")
    payload = _envelope(
        row,
        {
            "tier": "provisional",
            "trade_date": quote_date or yyyymmdd(now_shanghai()),
            "source": source,
            "stale": False,
            "staleness_seconds": 0,
            "published_at": isoformat(now_shanghai()),
        },
    )
    _write_cache(db, cache_key, payload, INDEX_TTL, source)
    return payload


def _canonical_sw_name(db: HubDB, ts_code: str) -> str:
    row = db.fetchone(
        "SELECT name FROM sector_master WHERE ts_code = ? AND family = 'sw'",
        (ts_code,),
    )
    if not row:
        row = db.fetchone(
            "SELECT name FROM eod_sector_daily WHERE ts_code = ? AND family = 'sw' "
            "ORDER BY trade_date DESC LIMIT 1",
            (ts_code,),
        )
    return str((row or {}).get("name") or "").strip()


def _sector_row_matches(row: dict[str, Any], canonical_name: str) -> bool:
    return not canonical_name or str(row.get("name") or "").strip() == canonical_name


def _sector_payload_matches(payload: dict[str, Any], canonical_name: str) -> bool:
    data = payload.get("data") if isinstance(payload, dict) else None
    return isinstance(data, dict) and _sector_row_matches(data, canonical_name)


def fetch_limit_pool(db: HubDB, trade_date: str = "") -> dict[str, Any]:
    day = yyyymmdd(trade_date or now_shanghai())
    cache_key = f"limit-pool:{day}"
    cached = _read_cache(db, cache_key)
    if cached is not None:
        return cached
    try:
        rows = EastmoneyAdapter().fetch_limit_pool(day)
        source = "eastmoney:zt_pool"
    except Exception as exc:
        recovered = _load_quotes_lkg(db, cache_key)
        if recovered is not None:
            return recovered
        raise RealtimeApiError("SOURCE_UNAVAILABLE", f"limit pool unavailable: {exc}") from exc
    payload = _envelope(
        rows,
        {
            "tier": "provisional",
            "trade_date": day,
            "source": source,
            "stale": False,
            "staleness_seconds": 0,
            "published_at": isoformat(now_shanghai()),
        },
    )
    _write_cache(db, cache_key, payload, QUOTE_TTL, source)
    return payload


def _eastmoney_named_quotes(codes: list[str]) -> list[dict[str, Any]]:
    return _parallel_named_quotes(EastmoneyAdapter, codes)


def _tencent_named_quotes(codes: list[str]) -> list[dict[str, Any]]:
    return _parallel_named_quotes(TencentAdapter, codes)


def _parallel_named_quotes(adapter_factory, codes: list[str]) -> list[dict[str, Any]]:
    chunks = [codes[index:index + QUOTE_BATCH] for index in range(0, len(codes), QUOTE_BATCH)]
    if not chunks:
        return []
    rows: list[dict[str, Any]] = []
    errors: list[str] = []
    with ThreadPoolExecutor(max_workers=min(8, len(chunks))) as executor:
        futures = [executor.submit(adapter_factory().fetch_quotes, chunk) for chunk in chunks]
        for future in as_completed(futures):
            try:
                rows.extend(future.result())
            except Exception as exc:
                errors.append(str(exc))
    if not rows and errors:
        raise AdapterError("; ".join(errors[:3]))
    return rows


def _listed_ts_codes(db: HubDB) -> list[str]:
    try:
        rows = db.fetchall(
            "SELECT ts_code FROM stock_master WHERE list_status = 'L' ORDER BY ts_code"
        )
    except Exception:
        return []
    return [str(row.get("ts_code") or "") for row in rows if row.get("ts_code")]


def _quote_payload(
    rows: list[dict[str, Any]],
    source: str,
    scope: str = "",
) -> dict[str, Any]:
    meta: dict[str, Any] = {
        "tier": "provisional",
        "trade_date": yyyymmdd(now_shanghai()),
        "source": source,
        "stale": False,
        "staleness_seconds": 0,
        "published_at": isoformat(now_shanghai()),
        "failover": "tencent" in source,
        "delay_notice": "",
    }
    if scope:
        meta["scope"] = scope
    return _envelope(rows, meta)


def _load_quotes_lkg(
    db: HubDB,
    cache_key: str,
    expected_date: str = "",
) -> dict[str, Any] | None:
    store = LastKnownGood(db)
    item = store.load(cache_key)
    payload = item.get("payload") if item else None
    if not isinstance(payload, dict):
        return None
    data = payload.get("data")
    if not isinstance(data, (list, dict)) or not data:
        return None
    day = yyyymmdd(expected_date) if expected_date else ""
    if day:
        payload_day = yyyymmdd((payload.get("meta") or {}).get("trade_date"))
        if isinstance(data, list):
            dated = [
                row for row in data
                if isinstance(row, dict) and _row_quote_date(row, payload_day) == day
            ]
            if not dated:
                return None
            payload = {**payload, "data": dated}
        elif _row_quote_date(data, payload_day) != day:
            return None
    stamped = dict(payload)
    meta = dict(stamped.get("meta") or {})
    stored = str((item or {}).get("stored_at") or "")
    try:
        age = max(0, int(time.time() - datetime.fromisoformat(stored).timestamp()))
    except Exception:
        age = QUOTE_TTL
    meta["stale"] = True
    meta["staleness_seconds"] = age
    meta["delay_notice"] = f"主备免费行情均暂不可用，显示 {age} 秒前的真实快照"
    meta["lkg_source"] = str((item or {}).get("source") or meta.get("source") or "")
    stamped["meta"] = meta
    return stamped


def _quote_map(rows: list[Any], wanted: list[str]) -> dict[str, dict[str, Any]]:
    allowed = set(wanted)
    result: dict[str, dict[str, Any]] = {}
    for raw in rows:
        if not isinstance(raw, dict):
            continue
        code = str(raw.get("ts_code") or "").upper()
        if code in allowed and float(raw.get("close") or raw.get("price") or 0) > 0:
            result[code] = dict(raw)
    return result


def _quote_codes(rows: list[Any]) -> set[str]:
    return {
        str(row.get("ts_code") or "").upper()
        for row in rows
        if isinstance(row, dict) and row.get("ts_code")
    }


def _row_quote_date(row: dict[str, Any], fallback: str = "") -> str:
    raw_day = str(row.get("quote_date") or "").strip()
    if raw_day:
        try:
            return yyyymmdd(raw_day)
        except ValueError:
            pass
    stamp = str(row.get("quote_time") or row.get("trade_time") or "")
    raw_fallback = stamp[:10] if stamp else str(fallback or "")
    if not raw_fallback:
        return ""
    try:
        return yyyymmdd(raw_fallback)
    except ValueError:
        return ""


def _store_quote_rows_lkg(db: HubDB, rows: list[dict[str, Any]], source: str) -> None:
    now = now_shanghai()
    today = yyyymmdd(now)
    stored = isoformat(now)
    values = []
    for row in rows:
        code = str(row.get("ts_code") or "").upper()
        if not code or _row_quote_date(row, today) != today:
            continue
        payload = _envelope(
            dict(row),
            {"tier": "provisional", "trade_date": today, "source": source, "stale": False},
        )
        values.append((f"quote:{code}", json.dumps(payload, ensure_ascii=False), source, stored))
    if not values:
        return
    with db.write() as connection:
        connection.executemany(
            """
            INSERT INTO last_known_good(cache_key, payload, source, stored_at)
            VALUES (?,?,?,?)
            ON CONFLICT(cache_key) DO UPDATE SET
              payload=excluded.payload, source=excluded.source, stored_at=excluded.stored_at
            """,
            values,
        )


def _recover_quote_rows(
    db: HubDB,
    codes: list[str],
    expected_date: str,
) -> tuple[list[dict[str, Any]], int]:
    if not codes:
        return [], 0
    today = yyyymmdd(expected_date)
    recovered: dict[str, dict[str, Any]] = {}
    oldest_age = 0
    placeholders = ",".join("?" for _ in codes)
    keys = [f"quote:{code}" for code in codes]
    rows = db.fetchall(
        f"SELECT cache_key,payload,stored_at FROM last_known_good WHERE cache_key IN ({placeholders})",
        tuple(keys),
    )
    now_epoch = time.time()
    for item in rows:
        try:
            payload = json.loads(item["payload"])
        except (json.JSONDecodeError, TypeError):
            continue
        quote = payload.get("data") if isinstance(payload, dict) else None
        if not isinstance(quote, dict) or _row_quote_date(quote, (payload.get("meta") or {}).get("trade_date")) != today:
            continue
        code = str(quote.get("ts_code") or "").upper()
        if code not in codes:
            continue
        recovered[code] = dict(quote)
        try:
            oldest_age = max(oldest_age, int(now_epoch - datetime.fromisoformat(item["stored_at"]).timestamp()))
        except (TypeError, ValueError):
            oldest_age = max(oldest_age, QUOTE_TTL)

    remaining = [code for code in codes if code not in recovered]
    if remaining:
        market = _load_quotes_lkg(db, "quotes:market", today)
        market_rows = _quote_map((market or {}).get("data") or [], remaining)
        recovered.update(market_rows)
        oldest_age = max(oldest_age, int(((market or {}).get("meta") or {}).get("staleness_seconds") or 0))
    return [recovered[code] for code in codes if code in recovered], oldest_age


def warm_realtime(db: HubDB) -> dict[str, Any]:
    """Proactively keep same-day market snapshots warm during trading hours."""
    today = yyyymmdd(now_shanghai())
    result: dict[str, Any] = {"trade_date": today, "rows": 0, "errors": []}
    try:
        indexes = fetch_index_quotes(db)
        result["indexes"] = len(indexes.get("data") or [])
    except Exception as exc:
        result["errors"].append(f"indexes:{exc}")
    try:
        market = fetch_market_quotes(db)
        result["market"] = len(market.get("data") or [])
        result["rows"] += result["market"]
    except Exception as exc:
        result["errors"].append(f"market:{exc}")

    sector_rows: list[dict[str, Any]] = []
    try:
        masters = db.fetchall(
            """
            SELECT ts_code, name, 0 AS priority FROM sector_master WHERE family = 'sw'
            UNION ALL
            SELECT ts_code, name, 1 AS priority FROM eod_sector_daily WHERE family = 'sw'
            ORDER BY priority, ts_code
            """
        )
        canonical_names: dict[str, str] = {}
        for master in masters:
            master_code = str(master.get("ts_code") or "")
            master_name = str(master.get("name") or "").strip()
            if master_code and master_name:
                canonical_names.setdefault(master_code, master_name)
        codes = list(canonical_names)
        fetched_sector_rows = _eastmoney_sector_quotes(codes)
        for row in fetched_sector_rows:
            if _row_quote_date(row, today) != today:
                continue
            code = str(row.get("ts_code") or "").upper()
            if not _sector_row_matches(row, canonical_names.get(code, "")):
                result["errors"].append(
                    f"sector:{code}:industry name mismatch"
                )
                continue
            payload = _envelope(
                row,
                {
                    "tier": "provisional",
                    "trade_date": today,
                    "source": str(row.get("source") or "eastmoney_sw"),
                    "stale": False,
                    "staleness_seconds": 0,
                    "published_at": isoformat(now_shanghai()),
                },
            )
            _write_cache(db, f"sector:{code}", payload, INDEX_TTL, "eastmoney_sw")
            sector_rows.append(row)
        result["sectors"] = len(sector_rows)
        result["rows"] += len(sector_rows)
    except Exception as exc:
        result["errors"].append(f"sectors:{exc}")
    return result


def _eastmoney_sector_quotes(codes: list[str]) -> list[dict[str, Any]]:
    chunks = [codes[index:index + QUOTE_BATCH] for index in range(0, len(codes), QUOTE_BATCH)]
    if not chunks:
        return []
    rows: list[dict[str, Any]] = []
    errors: list[str] = []
    with ThreadPoolExecutor(max_workers=min(8, len(chunks))) as executor:
        futures = [executor.submit(EastmoneyAdapter().fetch_shenwan_quotes, chunk) for chunk in chunks]
        for future in as_completed(futures):
            try:
                rows.extend(future.result())
            except Exception as exc:
                errors.append(str(exc))
    if not rows and errors:
        raise AdapterError("; ".join(errors[:3]))
    return rows


def fetch_intraday(db: HubDB, code: str, date: str = "") -> dict[str, Any]:
    ts_code = resolve_code(db, code) or _guess_ts_code(code)
    if not ts_code:
        raise RealtimeApiError("INVALID_ARGUMENT", f"ambiguous code: {code}")
    cache_key = f"intraday:{ts_code}:{date or 'today'}"
    cached = _read_cache(db, cache_key)
    if cached is not None:
        return cached
    adapter = EastmoneyAdapter()
    try:
        payload_data = adapter.fetch_intraday(ts_code, date)
        source = "eastmoney:trends2"
    except Exception as exc:
        recovered = _load_intraday_lkg(db, ts_code, date)
        if recovered is None:
            raise RealtimeApiError("SOURCE_UNAVAILABLE", f"intraday unavailable: {exc}") from exc
        return recovered
    payload = _envelope(
        payload_data,
        {
            "tier": "provisional",
            "trade_date": yyyymmdd(payload_data.get("trade_date") or date or now_shanghai()),
            "source": source,
            "stale": False,
            "staleness_seconds": 0,
            "published_at": isoformat(now_shanghai()),
        },
    )
    _write_cache(db, cache_key, payload, INTRADAY_TTL, source)
    return payload


def _load_intraday_lkg(db: HubDB, ts_code: str, date: str = "") -> dict[str, Any] | None:
    store = LastKnownGood(db)
    keys = [f"intraday:{ts_code}:{date or 'today'}"]
    if date:
        keys.append(f"intraday:{ts_code}:today")
    for key in keys:
        item = store.load(key)
        payload = _lkg_payload(item)
        if payload is not None:
            return payload
    row = db.fetchone(
        "SELECT * FROM last_known_good WHERE cache_key LIKE ? ORDER BY stored_at DESC LIMIT 1",
        (f"intraday:{ts_code}:%",),
    )
    if not row:
        return None
    try:
        raw = json.loads(row["payload"])
    except json.JSONDecodeError:
        return None
    return _mark_stale(raw) if isinstance(raw, dict) else None


def _lkg_payload(item: dict[str, Any] | None) -> dict[str, Any] | None:
    if not item:
        return None
    payload = item.get("payload")
    return _mark_stale(payload) if isinstance(payload, dict) else None


def _mark_stale(payload: dict[str, Any]) -> dict[str, Any] | None:
    data = payload.get("data")
    if not isinstance(data, dict) or not data.get("points"):
        return None
    stamped = dict(payload)
    meta = dict(stamped.get("meta") or {})
    meta["stale"] = True
    stamped["meta"] = meta
    return stamped


def _guess_ts_code(code: str) -> str | None:
    raw = str(code or "").strip().upper()
    if "." in raw:
        return raw
    if len(raw) == 6 and raw.isdigit():
        if raw.startswith(("4", "8", "92")):
            return f"{raw}.BJ"
        if raw.startswith(("5", "6", "9")):
            return f"{raw}.SH"
        return f"{raw}.SZ"
    return None


def _read_cache(db: HubDB, cache_key: str) -> dict[str, Any] | None:
    row = db.fetchone("SELECT * FROM rt_cache WHERE cache_key = ?", (cache_key,))
    if not row:
        return None
    expires = str(row.get("expires_at") or "")
    now = isoformat(now_shanghai())
    if expires and expires < now:
        return None
    try:
        payload = json.loads(row["payload"])
    except json.JSONDecodeError:
        return None
    if isinstance(payload, dict) and isinstance(payload.get("meta"), dict):
        stored = str(row.get("stored_at") or "")
        try:
            age = max(0, int(time.time() - datetime.fromisoformat(stored).timestamp()))
        except Exception:
            age = 0
        payload["meta"]["staleness_seconds"] = age
        payload["meta"]["stale"] = age > QUOTE_TTL
    return payload


def _write_cache(db: HubDB, cache_key: str, payload: dict[str, Any], ttl: int, source: str) -> None:
    from datetime import timedelta

    now = now_shanghai()
    stored = isoformat(now)
    expires = isoformat(now + timedelta(seconds=ttl))
    db.execute(
        """
        INSERT INTO rt_cache(cache_key, payload, source, stored_at, expires_at)
        VALUES (?,?,?,?,?)
        ON CONFLICT(cache_key) DO UPDATE SET
          payload=excluded.payload, source=excluded.source,
          stored_at=excluded.stored_at, expires_at=excluded.expires_at
        """,
        (cache_key, json.dumps(payload, ensure_ascii=False), source, stored, expires),
    )
    db.execute(
        """
        INSERT INTO last_known_good(cache_key, payload, source, stored_at)
        VALUES (?,?,?,?)
        ON CONFLICT(cache_key) DO UPDATE SET
          payload=excluded.payload, source=excluded.source, stored_at=excluded.stored_at
        """,
        (cache_key, json.dumps(payload, ensure_ascii=False), source, stored),
    )