"""The digest's long memory — every ingested item, kept across runs.

A run is otherwise amnesiac: it fetches the last six hours, writes a digest, and
forgets. That makes some of the most valuable things this system could say
impossible to say at all. "This signature has fired every night for a week",
"unemployment is up for the third month", "merchant traffic through this
chokepoint has halved since May", "this story first appeared four days ago and
has not moved" — none of them are visible in one window, and all of them are the
kind of finding the political and network prompts are otherwise told they may
not make (see the "one snapshot is not a trend" rules).

This module is that memory: a SQLite database in the same `/data` volume the
Telegram session and IDSconf.json already live in, written at the end of
ingestion and read back at the start of the next run.

TWO TABLES, BECAUSE THERE ARE TWO KINDS OF THING
------------------------------------------------
`items` — discrete things that happen once and are then referred to again: an
article, a message, a video, a calendar event. Deduplicated on a fingerprint, so
an article that appears in four consecutive runs is ONE row that has been seen
four times, not four rows. That is what makes `first_seen_at` meaningful, and
`first_seen_at` is what tells the prompt whether it is looking at a new
development or the same story it showed you yesterday.

`observations` — numbers that are re-measured every run and only mean anything
as a series: an unemployment rate, a share price, an aircraft count in a region,
the number of IDS alerts on a signature. Stored one row per measurement so a
trend is a query rather than a guess.

WHAT THIS IS NOT
----------------
It is not a cache — nothing reads it to avoid fetching. It is not a second
source of truth: every claim in a digest still has to trace to this run's
context, and history enters the prompt as its own clearly-labelled block with
its own dates attached, so "in the last 30 days" can never be presented as
something that happened today.

It is also not free of consequence: this file persists the household's mail and
messages for as long as its retention window, where before them only the last
few runs' `context.json` did. `DIGEST_ARCHIVE_EXCLUDE_SOURCES` exists for
exactly that reason, and README.md says so plainly rather than burying it.

Every operation here is best-effort. A corrupt or unwritable database logs a
warning and the run continues without history — a digest with no memory is worth
far more than no digest, the same rule every ingestion module follows.
"""

import hashlib
import json
import logging
import os
import sqlite3
from datetime import datetime, timedelta, timezone
from pathlib import Path

LOG = logging.getLogger(__name__)

DEFAULT_DB_PATH = "/data/digest-archive.db"
DEFAULT_RETENTION_DAYS = 400

# Sources whose items are discrete things worth deduplicating and dating. Anything
# not listed contributes observations only (see _record_observations) or nothing.
ITEM_SOURCES = (
    "news", "rci_social", "email", "signal", "telegram", "discord", "whatsapp",
    "calendar", "grocy",
)

# How much of an item's own text takes part in its fingerprint. Long enough that two
# genuinely different messages don't collide, short enough that an article whose
# summary gets re-edited between runs is still recognised as the same article.
FINGERPRINT_TEXT_CHARS = 200

MAX_SERIES_POINTS = 60


def _now_iso():
    return datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def _db_path():
    return Path(os.environ.get("DIGEST_ARCHIVE_DB_PATH", DEFAULT_DB_PATH))


def _excluded_sources():
    raw = os.environ.get("DIGEST_ARCHIVE_EXCLUDE_SOURCES", "")
    return {part.strip() for part in raw.split(",") if part.strip()}


def _connect():
    path = _db_path()
    path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(path, timeout=15)
    conn.row_factory = sqlite3.Row
    conn.executescript(
        """
        CREATE TABLE IF NOT EXISTS items (
            id INTEGER PRIMARY KEY,
            -- What makes two sightings the same thing. See _fingerprint().
            fingerprint TEXT NOT NULL UNIQUE,
            source TEXT NOT NULL,
            category TEXT,
            -- The item's own timestamp (when it was published/sent), which is not the
            -- same as when this system first saw it — a feed can be hours behind.
            occurred_at TEXT,
            first_seen_at TEXT NOT NULL,
            last_seen_at TEXT NOT NULL,
            times_seen INTEGER NOT NULL DEFAULT 1,
            first_run_id TEXT NOT NULL,
            last_run_id TEXT NOT NULL,
            title TEXT,
            url TEXT,
            payload TEXT NOT NULL
        );
        CREATE INDEX IF NOT EXISTS items_source_seen ON items (source, last_seen_at);
        CREATE INDEX IF NOT EXISTS items_occurred ON items (occurred_at);

        CREATE TABLE IF NOT EXISTS observations (
            id INTEGER PRIMARY KEY,
            -- Stable machine key ("fred:UNRATE", "ids:signature:2013028"), plus a human
            -- label kept alongside it so a series can be read without a lookup table.
            series TEXT NOT NULL,
            label TEXT,
            metric TEXT NOT NULL,
            value REAL NOT NULL,
            observed_at TEXT NOT NULL,
            run_id TEXT NOT NULL,
            -- One measurement per series per metric per timestamp: a re-run of the same
            -- hour must not double-count, and FRED re-serving last month's figure must
            -- not look like a fresh reading.
            UNIQUE (series, metric, observed_at)
        );
        CREATE INDEX IF NOT EXISTS observations_series ON observations (series, observed_at);
        """
    )
    return conn


def _fingerprint(source, item):
    """What makes two sightings of a thing the same thing.

    A URL when there is one — that is what an article, a video or a post actually
    is. Otherwise the source plus the item's own timestamp and the head of its
    text, which is stable for a message (they don't get edited under you) and
    stable enough for a calendar event.
    """
    url = str(item.get("url") or item.get("link") or "").strip()
    if url:
        return hashlib.sha256(f"{source}\n{url}".encode("utf-8")).hexdigest()

    text = " ".join(
        str(item.get(key) or "")
        for key in ("title", "summary", "subject", "body", "message", "name", "chat")
    )[:FINGERPRINT_TEXT_CHARS]
    stamp = str(item.get("timestamp") or item.get("start") or item.get("occurred_at") or "")
    return hashlib.sha256(f"{source}\n{stamp}\n{text}".encode("utf-8")).hexdigest()


def _item_title(item):
    for key in ("title", "subject", "summary", "name", "chat"):
        value = item.get(key)
        if value:
            return str(value)[:300]
    body = item.get("body") or item.get("message") or ""
    return str(body)[:300] or None


def _record_items(conn, run_id, seen_at, collected, excluded):
    """Upserts this run's items and annotates them in place with what the archive
    already knew. The annotation is the whole point: every entry the prompt sees
    carries `first_seen_at` and `times_seen`, so "new this run" and "the same story
    for four days" are distinguishable without the model having to infer it."""
    known = 0
    for source in ITEM_SOURCES:
        if source in excluded:
            continue
        for item in collected.get(source) or []:
            if not isinstance(item, dict):
                continue
            fingerprint = _fingerprint(source, item)
            row = conn.execute(
                "SELECT first_seen_at, times_seen FROM items WHERE fingerprint = ?",
                (fingerprint,),
            ).fetchone()

            if row:
                times_seen = row["times_seen"] + 1
                conn.execute(
                    "UPDATE items SET last_seen_at = ?, last_run_id = ?, times_seen = ? "
                    "WHERE fingerprint = ?",
                    (seen_at, run_id, times_seen, fingerprint),
                )
                item["first_seen_at"] = row["first_seen_at"]
                item["times_seen"] = times_seen
                known += 1
            else:
                conn.execute(
                    "INSERT INTO items (fingerprint, source, category, occurred_at, "
                    "first_seen_at, last_seen_at, times_seen, first_run_id, last_run_id, "
                    "title, url, payload) VALUES (?, ?, ?, ?, ?, ?, 1, ?, ?, ?, ?, ?)",
                    (
                        fingerprint,
                        source,
                        item.get("category"),
                        item.get("timestamp") or item.get("start"),
                        seen_at,
                        seen_at,
                        run_id,
                        run_id,
                        _item_title(item),
                        item.get("url") or item.get("link"),
                        json.dumps(item, ensure_ascii=False, default=str),
                    ),
                )
                item["first_seen_at"] = seen_at
                item["times_seen"] = 1
    return known


def _observation_rows(collected):
    """The numeric series worth trending, flattened out of this run's summaries.

    Each source's shape is different and each is handled explicitly rather than by
    a generic "find the numbers" walk — a wrong series is worse than a missing one,
    because it would be trended and reported with a straight face.
    """
    rows = []

    for entry in collected.get("financial") or []:
        if not isinstance(entry, dict):
            continue
        if entry.get("source") == "fred" and entry.get("series_id"):
            try:
                rows.append((
                    f"fred:{entry['series_id']}", entry["series_id"], "value",
                    float(entry["latest_value"]), str(entry.get("latest_date") or ""),
                ))
            except (TypeError, ValueError):
                pass
        elif entry.get("source") == "stooq" and entry.get("symbol"):
            try:
                rows.append((
                    f"stooq:{entry['symbol']}", entry["symbol"], "close",
                    float(entry["close"]), str(entry.get("date") or ""),
                ))
            except (TypeError, ValueError):
                pass

    for entry in collected.get("flight_traffic") or []:
        if not isinstance(entry, dict) or not entry.get("region"):
            continue
        observed = str(entry.get("observed_at") or "")
        region = entry["region"]
        rows.append((f"flight:{region}", region, "aircraft_total",
                     float(entry.get("aircraft_total") or 0), observed))
        rows.append((f"flight:{region}", region, "military_callsign_matches",
                     float(entry.get("military_callsign_match_count") or 0), observed))

    for entry in collected.get("naval_traffic") or []:
        if not isinstance(entry, dict) or not entry.get("region"):
            continue
        region = entry["region"]
        rows.append((f"naval:{region}", region, "distinct_vessels",
                     float(entry.get("distinct_vessels") or 0),
                     str(entry.get("observed_at") or "")))

    for entry in collected.get("opnsense_ids") or []:
        if not isinstance(entry, dict):
            continue
        observed = str(entry.get("observed_at") or "")
        rows.append(("ids:total", "IDS alerts", "alerts",
                     float(entry.get("alert_count") or 0), observed))
        for signature in entry.get("top_signatures") or []:
            if not isinstance(signature, dict) or signature.get("sid") is None:
                continue
            rows.append((
                f"ids:signature:{signature['sid']}",
                str(signature.get("signature") or signature["sid"])[:300],
                "alerts", float(signature.get("count") or 0), observed,
            ))
        for host in entry.get("top_local_hosts") or []:
            if isinstance(host, dict) and host.get("ip"):
                rows.append((f"ids:host:{host['ip']}", host["ip"], "alerts",
                             float(host.get("alerts") or 0), observed))

    return [row for row in rows if row[4]]


def _prune(conn, retention_days):
    cutoff = (datetime.now(timezone.utc) - timedelta(days=retention_days)).replace(
        microsecond=0).isoformat().replace("+00:00", "Z")
    conn.execute("DELETE FROM items WHERE last_seen_at < ?", (cutoff,))
    conn.execute("DELETE FROM observations WHERE observed_at < ?", (cutoff,))


def record(run_id, collected):
    """Writes this run into the archive and annotates `collected` in place with what
    was already known. Returns False if the archive is unavailable, in which case the
    run simply proceeds without memory."""
    if os.environ.get("ENABLE_DIGEST_ARCHIVE", "true").strip().lower() != "true":
        LOG.info("archive: disabled, this run will neither read nor write history")
        return False

    excluded = _excluded_sources()
    if excluded:
        LOG.info("archive: not storing %s", ", ".join(sorted(excluded)))

    seen_at = _now_iso()
    try:
        retention_days = int(os.environ.get("DIGEST_ARCHIVE_RETENTION_DAYS", DEFAULT_RETENTION_DAYS))
    except ValueError:
        retention_days = DEFAULT_RETENTION_DAYS

    try:
        with _connect() as conn:
            known = _record_items(conn, run_id, seen_at, collected, excluded)
            observations = [
                row for row in _observation_rows(collected)
                if row[0].split(":", 1)[0] not in excluded
            ]
            for series, label, metric, value, observed_at in observations:
                conn.execute(
                    "INSERT OR IGNORE INTO observations "
                    "(series, label, metric, value, observed_at, run_id) VALUES (?, ?, ?, ?, ?, ?)",
                    (series, label, metric, value, observed_at, run_id),
                )
            _prune(conn, retention_days)
        LOG.info(
            "archive: recorded run %s (%d item(s) already known, %d observation(s))",
            run_id, known, len(observations),
        )
        return True
    except Exception:
        LOG.warning("archive: could not record this run, continuing without it", exc_info=True)
        return False


def _series_history(conn, prefixes, points=MAX_SERIES_POINTS):
    series = {}
    for prefix in prefixes:
        rows = conn.execute(
            "SELECT series, label, metric, value, observed_at FROM observations "
            "WHERE series LIKE ? ORDER BY observed_at DESC LIMIT ?",
            (f"{prefix}%", points * 12),
        ).fetchall()
        for row in rows:
            key = f"{row['series']}|{row['metric']}"
            entry = series.setdefault(key, {
                "series": row["series"], "label": row["label"],
                "metric": row["metric"], "readings": [],
            })
            if len(entry["readings"]) < points:
                entry["readings"].append({"at": row["observed_at"], "value": row["value"]})
    # Oldest-first reads like a series rather than a stack.
    for entry in series.values():
        entry["readings"].reverse()
    return sorted(series.values(), key=lambda entry: entry["series"])


def _span_days(conn):
    row = conn.execute(
        "SELECT MIN(first_seen_at) AS oldest FROM items"
    ).fetchone()
    if not row or not row["oldest"]:
        return 0
    try:
        oldest = datetime.fromisoformat(str(row["oldest"]).replace("Z", "+00:00"))
    except ValueError:
        return 0
    return max(0, (datetime.now(timezone.utc) - oldest).days)


def history(sections):
    """The history blocks for the sections this run is generating.

    Deliberately shaped per section rather than handed over whole: the network
    prompt has no use for share prices, and every token of history in a prompt is a
    token not spent on this run's actual material.
    """
    if os.environ.get("ENABLE_DIGEST_ARCHIVE", "true").strip().lower() != "true":
        return {}

    try:
        with _connect() as conn:
            span = _span_days(conn)
            blocks = {}

            if "political" in sections:
                blocks["political"] = {
                    "archive_span_days": span,
                    "note": (
                        "Earlier readings of the same series, oldest first, each with its "
                        "own timestamp. This is the only basis on which this section may "
                        "describe anything as rising, falling or unchanged."
                    ),
                    "series": _series_history(conn, ("fred:", "stooq:", "flight:", "naval:")),
                }

            if "network" in sections:
                # Recurrence is the whole question for an IDS: one alert is noise, the
                # same signature every night on the same host is a fact about the house.
                signatures = conn.execute(
                    "SELECT series, label, SUM(value) AS alerts, COUNT(*) AS runs_seen, "
                    "MIN(observed_at) AS first_seen, MAX(observed_at) AS last_seen "
                    "FROM observations WHERE series LIKE 'ids:signature:%' "
                    "GROUP BY series ORDER BY alerts DESC LIMIT 12"
                ).fetchall()
                hosts = conn.execute(
                    "SELECT label AS host, SUM(value) AS alerts, COUNT(*) AS runs_seen, "
                    "MIN(observed_at) AS first_seen, MAX(observed_at) AS last_seen "
                    "FROM observations WHERE series LIKE 'ids:host:%' "
                    "GROUP BY series ORDER BY alerts DESC LIMIT 10"
                ).fetchall()
                blocks["network"] = {
                    "archive_span_days": span,
                    "note": (
                        "Alert history across every run in the archive, not just this "
                        "window. `runs_seen` is how many digest runs this has appeared in "
                        "— that is what tells a one-off apart from something recurring."
                    ),
                    "alert_totals": _series_history(conn, ("ids:total",), points=30),
                    "recurring_signatures": [dict(row) for row in signatures],
                    "recurring_hosts": [dict(row) for row in hosts],
                }

            return blocks
    except Exception:
        LOG.warning("archive: could not read history, continuing without it", exc_info=True)
        return {}
