"""SQLite persistence for history + memory (SPEC-009).

One database per deployment, stdlib sqlite3 only (D-010). Callers run
the sync methods in a worker thread (`asyncio.to_thread`) so the
event loop never blocks on disk (PER-05 / D6).
"""

import logging
import os
import pickle
import sqlite3
from contextlib import closing
from pathlib import Path
from typing import Any, Dict, List, Optional

SCHEMA_VERSION = 6


class PersistentStore:
    def __init__(self, db_path: Path) -> None:
        self.db_path = Path(db_path)
        self._init_db()

    def _connect(self) -> sqlite3.Connection:
        conn = sqlite3.connect(self.db_path)
        conn.execute("PRAGMA journal_mode=WAL")
        return conn

    def _init_db(self) -> None:
        # Forward-only migrations keyed on user_version (PER-06)
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        with closing(self._connect()) as conn, conn:
            version = conn.execute("PRAGMA user_version").fetchone()[0]
            if version < 1:
                conn.execute(
                    "CREATE TABLE IF NOT EXISTS history (id INTEGER PRIMARY KEY, channel TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL)"
                )
                conn.execute("CREATE INDEX IF NOT EXISTS history_channel ON history (channel)")
                conn.execute("CREATE TABLE IF NOT EXISTS memory (channel TEXT PRIMARY KEY, content TEXT NOT NULL)")
            if version < 2:
                conn.execute(
                    "CREATE TABLE IF NOT EXISTS usage (day TEXT NOT NULL, key TEXT NOT NULL, value REAL NOT NULL, PRIMARY KEY (day, key))"
                )
            if version < 3:
                conn.execute(
                    "CREATE TABLE IF NOT EXISTS user_facts (id INTEGER PRIMARY KEY, user TEXT NOT NULL, fact TEXT NOT NULL,"
                    " source TEXT NOT NULL DEFAULT 'self', updated_at TEXT NOT NULL DEFAULT (datetime('now')))"
                )
                conn.execute(
                    "CREATE TABLE IF NOT EXISTS pinned_facts (id INTEGER PRIMARY KEY, channel TEXT, fact TEXT NOT NULL,"
                    " created_at TEXT NOT NULL DEFAULT (datetime('now')))"
                )
                conn.execute(
                    "CREATE TABLE IF NOT EXISTS episodes (id INTEGER PRIMARY KEY, channel TEXT NOT NULL, summary TEXT NOT NULL,"
                    " created_at TEXT NOT NULL DEFAULT (datetime('now')))"
                )
                conn.execute(
                    "CREATE TABLE IF NOT EXISTS observations (id INTEGER PRIMARY KEY, channel TEXT NOT NULL, user TEXT NOT NULL,"
                    " kind TEXT NOT NULL, content TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT (datetime('now')))"
                )
                # Legacy single-string memories carry over as one episode each (MEM-08)
                conn.execute("INSERT INTO episodes (channel, summary) SELECT channel, content FROM memory")
            if version < 4:
                conn.execute(
                    "CREATE TABLE IF NOT EXISTS images (id INTEGER PRIMARY KEY, sha256 TEXT UNIQUE NOT NULL, channel TEXT NOT NULL,"
                    " user TEXT NOT NULL, message_id TEXT, ext TEXT NOT NULL, bytes INTEGER NOT NULL,"
                    " created_at TEXT NOT NULL DEFAULT (datetime('now')))"
                )
            if version < 5:
                conn.execute(
                    "CREATE TABLE IF NOT EXISTS tasks (id INTEGER PRIMARY KEY, kind TEXT NOT NULL, channel TEXT NOT NULL,"
                    " due_at TEXT NOT NULL, payload TEXT NOT NULL, state TEXT NOT NULL DEFAULT 'queued',"
                    " created_at TEXT NOT NULL DEFAULT (datetime('now')), executed_at TEXT)"
                )
            if version < 6:
                # News memory (SPEC-013 NEWS-09): deduped rolling store of fetched items
                conn.execute(
                    "CREATE TABLE IF NOT EXISTS news (id INTEGER PRIMARY KEY, dedup_key TEXT UNIQUE NOT NULL,"
                    " source TEXT NOT NULL DEFAULT '', title TEXT NOT NULL, link TEXT NOT NULL DEFAULT '',"
                    " summary TEXT NOT NULL DEFAULT '', first_seen TEXT NOT NULL DEFAULT (datetime('now')))"
                )
            if version < SCHEMA_VERSION:
                conn.execute(f"PRAGMA user_version = {SCHEMA_VERSION}")
        os.chmod(self.db_path, 0o600)  # conversation data (PER-04)

    def load_history(self, channel: str) -> List[Dict[str, Any]]:
        with closing(self._connect()) as conn:
            rows = conn.execute("SELECT role, content FROM history WHERE channel = ? ORDER BY id", (channel,)).fetchall()
        return [{"role": role, "content": content} for role, content in rows]

    def save_history(self, channel: str, history: List[Dict[str, Any]]) -> None:
        # Full replace per save: histories are small (<= history-limit)
        # and trims must be reflected (D-011)
        with closing(self._connect()) as conn, conn:
            conn.execute("DELETE FROM history WHERE channel = ?", (channel,))
            conn.executemany(
                "INSERT INTO history (channel, role, content) VALUES (?, ?, ?)",
                [(channel, str(entry.get("role", "user")), str(entry.get("content", ""))) for entry in history],
            )

    def load_memory(self, channel: str) -> Optional[str]:
        with closing(self._connect()) as conn:
            row = conn.execute("SELECT content FROM memory WHERE channel = ?", (channel,)).fetchone()
        return row[0] if row else None

    def save_memory(self, channel: str, content: str) -> None:
        with closing(self._connect()) as conn, conn:
            conn.execute(
                "INSERT INTO memory (channel, content) VALUES (?, ?) ON CONFLICT(channel) DO UPDATE SET content = excluded.content",
                (channel, content),
            )

    def usage_add(self, day: str, key: str, amount: float) -> None:
        with closing(self._connect()) as conn, conn:
            conn.execute(
                "INSERT INTO usage (day, key, value) VALUES (?, ?, ?) ON CONFLICT(day, key) DO UPDATE SET value = value + excluded.value",
                (day, key, amount),
            )

    def usage_get(self, day: str, key: str) -> float:
        with closing(self._connect()) as conn:
            row = conn.execute("SELECT value FROM usage WHERE day = ? AND key = ?", (day, key)).fetchone()
        return float(row[0]) if row else 0.0

    # --- news memory (SPEC-013 NEWS-09..12) ---

    def add_news_items(self, items: List[Dict[str, Any]]) -> int:
        """Insert deduped news rows (by link or title); returns how many were new (NEWS-09)."""
        added = 0
        with closing(self._connect()) as conn, conn:
            for item in items:
                title = str(item.get("title") or "").strip()
                key = (str(item.get("link") or "").strip()) or title
                if not title or not key:
                    continue
                cursor = conn.execute(
                    "INSERT OR IGNORE INTO news (dedup_key, source, title, link, summary) VALUES (?, ?, ?, ?, ?)",
                    (key, str(item.get("source") or ""), title, str(item.get("link") or ""), str(item.get("summary") or "")),
                )
                added += cursor.rowcount
        return added

    def recent_news(self, limit: int = 20, source: Optional[str] = None) -> List[Dict[str, Any]]:
        sql = "SELECT source, title, link, summary FROM news"
        params: List[Any] = []
        if source:
            sql += " WHERE source = ?"
            params.append(source)
        sql += " ORDER BY id DESC LIMIT ?"
        params.append(int(limit))
        with closing(self._connect()) as conn:
            rows = conn.execute(sql, params).fetchall()
        return [{"source": r[0], "title": r[1], "link": r[2], "summary": r[3]} for r in rows]

    def search_news(self, terms: List[str], limit: int = 20, source: Optional[str] = None) -> List[Dict[str, Any]]:
        """Rows where every term appears in title or summary; optional source filter (NEWS-11)."""
        params: List[Any] = []
        clauses = []
        for term in terms:
            clauses.append("(title LIKE ? OR summary LIKE ? OR source LIKE ?)")
            like = f"%{term}%"
            params += [like, like, like]
        where = " AND ".join(clauses) if clauses else "1=1"
        if source:
            where = f"({where}) AND source = ?"
            params.append(source)
        params.append(int(limit))
        sql = f"SELECT source, title, link, summary FROM news WHERE {where} ORDER BY id DESC LIMIT ?"  # nosec B608 - fixed templates; values parameterised
        with closing(self._connect()) as conn:
            rows = conn.execute(sql, params).fetchall()
        return [{"source": r[0], "title": r[1], "link": r[2], "summary": r[3]} for r in rows]

    def prune_news(self, keep: int) -> int:
        """Keep the newest `keep` rows, delete the rest (rolling window, NEWS-09)."""
        with closing(self._connect()) as conn, conn:
            cursor = conn.execute("DELETE FROM news WHERE id NOT IN (SELECT id FROM news ORDER BY id DESC LIMIT ?)", (int(keep),))
        return cursor.rowcount

    def news_count(self) -> int:
        with closing(self._connect()) as conn:
            return int(conn.execute("SELECT COUNT(*) FROM news").fetchone()[0])

    def delete_history_of_user(self, user: str) -> int:
        """Remove persisted rows carrying this user's messages (SAF-08)."""
        with closing(self._connect()) as conn, conn:
            cursor = conn.execute("DELETE FROM history WHERE content LIKE ?", (f'%"user": "{user}"%',))
        return cursor.rowcount

    # --- structured memory (SPEC-002) ---

    def add_observation(self, channel: str, user: str, kind: str, content: str) -> None:
        with closing(self._connect()) as conn, conn:
            conn.execute("INSERT INTO observations (channel, user, kind, content) VALUES (?, ?, ?, ?)", (channel, user, kind, content))

    def unconsumed_observations(self, channel: str) -> int:
        with closing(self._connect()) as conn:
            return int(conn.execute("SELECT COUNT(*) FROM observations WHERE channel = ?", (channel,)).fetchone()[0])

    def peek_observations(self, channel: str) -> List[Dict[str, Any]]:
        with closing(self._connect()) as conn:
            rows = conn.execute("SELECT id, user, kind, content FROM observations WHERE channel = ? ORDER BY id", (channel,)).fetchall()
        return [{"id": row[0], "user": row[1], "kind": row[2], "content": row[3]} for row in rows]

    def consume_observations(self, channel: str, up_to_id: int) -> None:
        with closing(self._connect()) as conn, conn:
            conn.execute("DELETE FROM observations WHERE channel = ? AND id <= ?", (channel, up_to_id))

    def add_user_fact(self, user: str, fact: str, source: str = "self") -> None:
        with closing(self._connect()) as conn, conn:
            conn.execute("INSERT INTO user_facts (user, fact, source) VALUES (?, ?, ?)", (user, fact, source))

    def facts_for(self, users: List[str]) -> List[Dict[str, Any]]:
        if not users:
            return []
        marks = ",".join("?" for _ in users)
        with closing(self._connect()) as conn:
            # marks is only "?" placeholders; user values stay parameterized
            rows = conn.execute(
                f"SELECT id, user, fact FROM user_facts WHERE user IN ({marks}) ORDER BY id", tuple(users)  # nosec B608
            ).fetchall()
        return [{"id": row[0], "user": row[1], "fact": row[2]} for row in rows]

    def delete_fact(self, fact_id: int) -> int:
        with closing(self._connect()) as conn, conn:
            return conn.execute("DELETE FROM user_facts WHERE id = ?", (fact_id,)).rowcount

    def purge_old_facts(self, retention_days: int) -> int:
        with closing(self._connect()) as conn, conn:
            cursor = conn.execute("DELETE FROM user_facts WHERE updated_at < datetime('now', ?)", (f"-{int(retention_days)} days",))
        return cursor.rowcount

    def add_pinned(self, channel: Optional[str], fact: str) -> None:
        with closing(self._connect()) as conn, conn:
            conn.execute("INSERT INTO pinned_facts (channel, fact) VALUES (?, ?)", (channel, fact))

    def pinned_for(self, channel: str) -> List[Dict[str, Any]]:
        with closing(self._connect()) as conn:
            rows = conn.execute(
                "SELECT id, channel, fact FROM pinned_facts WHERE channel IS NULL OR channel = ? ORDER BY id", (channel,)
            ).fetchall()
        return [{"id": row[0], "channel": row[1], "fact": row[2]} for row in rows]

    def pinned_all(self) -> List[Dict[str, Any]]:
        with closing(self._connect()) as conn:
            rows = conn.execute("SELECT id, channel, fact FROM pinned_facts ORDER BY id").fetchall()
        return [{"id": row[0], "channel": row[1], "fact": row[2]} for row in rows]

    def delete_pinned(self, pin_id: int) -> int:
        with closing(self._connect()) as conn, conn:
            return conn.execute("DELETE FROM pinned_facts WHERE id = ?", (pin_id,)).rowcount

    def add_episode(self, channel: str, summary: str) -> None:
        with closing(self._connect()) as conn, conn:
            conn.execute("INSERT INTO episodes (channel, summary) VALUES (?, ?)", (channel, summary))

    def recent_episodes(self, channel: str, count: int) -> List[str]:
        with closing(self._connect()) as conn:
            rows = conn.execute("SELECT summary FROM episodes WHERE channel = ? ORDER BY id DESC LIMIT ?", (channel, count)).fetchall()
        return [row[0] for row in reversed(rows)]

    def trim_episodes(self, channel: str, keep: int) -> int:
        with closing(self._connect()) as conn, conn:
            cursor = conn.execute(
                "DELETE FROM episodes WHERE channel = ? AND id NOT IN (SELECT id FROM episodes WHERE channel = ? ORDER BY id DESC LIMIT ?)",
                (channel, channel, keep),
            )
        return cursor.rowcount

    # --- task queue (SPEC-005, FDB-011) ---

    def task_add(self, kind: str, channel: str, due_at: str, payload: str, state: str = "queued") -> int:
        with closing(self._connect()) as conn, conn:
            cursor = conn.execute(
                "INSERT INTO tasks (kind, channel, due_at, payload, state) VALUES (?, ?, ?, ?, ?)",
                (kind, channel, due_at, payload, state),
            )
        return int(cursor.lastrowid or 0)

    def tasks_due(self) -> List[Dict[str, Any]]:
        with closing(self._connect()) as conn:
            rows = conn.execute(
                "SELECT id, kind, channel, payload FROM tasks WHERE state = 'queued' AND due_at <= datetime('now') ORDER BY due_at"
            ).fetchall()
        return [{"id": row[0], "kind": row[1], "channel": row[2], "payload": row[3]} for row in rows]

    def tasks_open(self) -> List[Dict[str, Any]]:
        with closing(self._connect()) as conn:
            rows = conn.execute(
                "SELECT id, kind, channel, due_at, state FROM tasks WHERE state IN ('queued', 'approval') ORDER BY id"
            ).fetchall()
        return [{"id": row[0], "kind": row[1], "channel": row[2], "due_at": row[3], "state": row[4]} for row in rows]

    def task_set_state(self, task_id: int, state: str) -> int:
        with closing(self._connect()) as conn, conn:
            return conn.execute("UPDATE tasks SET state = ?, executed_at = datetime('now') WHERE id = ?", (state, task_id)).rowcount

    def tasks_pending_of_kind(self, kind: str) -> int:
        with closing(self._connect()) as conn:
            row = conn.execute("SELECT COUNT(*) FROM tasks WHERE kind = ? AND state IN ('queued', 'approval')", (kind,)).fetchone()
        return int(row[0])

    # --- image cache index (SPEC-004, FDB-010) ---

    def image_add(self, sha256: str, channel: str, user: str, message_id: Optional[str], ext: str, nbytes: int) -> None:
        with closing(self._connect()) as conn, conn:
            conn.execute(
                "INSERT OR IGNORE INTO images (sha256, channel, user, message_id, ext, bytes) VALUES (?, ?, ?, ?, ?, ?)",
                (sha256, channel, user, message_id, ext, nbytes),
            )

    def images_recent(self, channel: str, count: int) -> List[Dict[str, Any]]:
        with closing(self._connect()) as conn:
            rows = conn.execute(
                "SELECT sha256, user, ext FROM images WHERE channel = ? ORDER BY id DESC LIMIT ?", (channel, count)
            ).fetchall()
        return [{"sha256": row[0], "user": row[1], "ext": row[2]} for row in rows]

    def images_total_bytes(self) -> int:
        with closing(self._connect()) as conn:
            row = conn.execute("SELECT COALESCE(SUM(bytes), 0) FROM images").fetchone()
        return int(row[0])

    def images_oldest(self, count: int) -> List[Dict[str, Any]]:
        with closing(self._connect()) as conn:
            rows = conn.execute("SELECT sha256, ext, bytes FROM images ORDER BY id LIMIT ?", (count,)).fetchall()
        return [{"sha256": row[0], "ext": row[1], "bytes": row[2]} for row in rows]

    def images_expired(self, ttl_days: int) -> List[Dict[str, Any]]:
        with closing(self._connect()) as conn:
            rows = conn.execute(
                "SELECT sha256, ext FROM images WHERE created_at < datetime('now', ?)", (f"-{int(ttl_days)} days",)
            ).fetchall()
        return [{"sha256": row[0], "ext": row[1]} for row in rows]

    def images_delete(self, sha256: str) -> None:
        with closing(self._connect()) as conn, conn:
            conn.execute("DELETE FROM images WHERE sha256 = ?", (sha256,))

    def images_for_user(self, user: str) -> List[Dict[str, Any]]:
        with closing(self._connect()) as conn:
            rows = conn.execute("SELECT sha256, ext FROM images WHERE user = ?", (user,)).fetchall()
        return [{"sha256": row[0], "ext": row[1]} for row in rows]

    def images_for_message(self, message_id: str) -> List[Dict[str, Any]]:
        with closing(self._connect()) as conn:
            rows = conn.execute("SELECT sha256, ext FROM images WHERE message_id = ?", (message_id,)).fetchall()
        return [{"sha256": row[0], "ext": row[1]} for row in rows]

    def purge_user_memory(self, user: str) -> int:
        """Facts, observations and episode traces of one user (MEM-09)."""
        removed = 0
        with closing(self._connect()) as conn, conn:
            removed += conn.execute("DELETE FROM user_facts WHERE user = ?", (user,)).rowcount
            removed += conn.execute("DELETE FROM observations WHERE user = ?", (user,)).rowcount
            removed += conn.execute("DELETE FROM episodes WHERE summary LIKE ?", (f"%{user}%",)).rowcount
        return removed

    def migrate_pickles(self, channel: str, history_file: Path, memory_file: Path) -> None:
        """Import legacy pickles once; rename them *.migrated (PER-03)."""
        if self.load_history(channel) or self.load_memory(channel) is not None:
            return
        if history_file.exists():
            try:
                with open(history_file, "rb") as fd:
                    self.save_history(channel, pickle.load(fd))
                history_file.rename(history_file.with_name(history_file.name + ".migrated"))
                logging.info(f"migrated legacy history pickle for {channel}")
            except Exception as err:
                logging.error(f"failed to migrate history pickle {history_file}: {err!r}")
        if memory_file.exists():
            try:
                with open(memory_file, "rb") as fd:
                    legacy_memory = str(pickle.load(fd))
                self.save_memory(channel, legacy_memory)
                self.add_episode(channel, legacy_memory)  # MEM-08
                memory_file.rename(memory_file.with_name(memory_file.name + ".migrated"))
                logging.info(f"migrated legacy memory pickle for {channel}")
            except Exception as err:
                logging.error(f"failed to migrate memory pickle {memory_file}: {err!r}")