backend/app/db.py

"""Application database (SQLite): documents, audit log, and managed prompts.

This is the real persistence layer for the retail IT deployment — every processed
document, every audit event, and every prompt version is stored here. (Metrics/KPIs
live in app/metrics.py; the vector index lives in app/rag_store.py.)

Production swap: point APP_DB_PATH at Postgres-compatible storage, or replace the
thin DAO with SQLAlchemy — the call sites only use the methods below.
"""
from __future__ import annotations

import json
import sqlite3
import threading
import time
from pathlib import Path


class Database:
    def __init__(self, db_path: str | Path) -> None:
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        self._lock = threading.Lock()
        self._conn = sqlite3.connect(str(self.db_path), check_same_thread=False)
        self._conn.row_factory = sqlite3.Row
        self._init()
        self._seed_prompts()

    def _init(self):
        with self._lock:
            self._conn.executescript(
                """
                CREATE TABLE IF NOT EXISTS documents (
                    run_id TEXT PRIMARY KEY, doc_id TEXT, doc_type TEXT, category TEXT,
                    channel TEXT, ocr_backend TEXT, confidence REAL, posted INTEGER,
                    requires_review INTEGER, total_cost_usd REAL, total_tokens INTEGER,
                    latency_ms REAL, flags TEXT, extracted TEXT, created_at REAL);
                CREATE INDEX IF NOT EXISTS idx_docs_cat ON documents(category);
                CREATE TABLE IF NOT EXISTS audit_log (
                    id INTEGER PRIMARY KEY AUTOINCREMENT, ts REAL, action TEXT,
                    run_id TEXT, actor TEXT, detail TEXT);
                CREATE TABLE IF NOT EXISTS prompts (
                    name TEXT PRIMARY KEY, version INTEGER, content TEXT,
                    updated_at REAL, updated_by TEXT);
                CREATE TABLE IF NOT EXISTS prompt_history (
                    id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT, version INTEGER,
                    content TEXT, ts REAL, actor TEXT);
                """
            )
            self._conn.commit()

    # --- documents -----------------------------------------------------------
    def save_document(self, run: dict) -> None:
        res = run.get("result", {}) or {}
        with self._lock:
            self._conn.execute(
                """INSERT OR REPLACE INTO documents (run_id, doc_id, doc_type, category,
                   channel, ocr_backend, confidence, posted, requires_review,
                   total_cost_usd, total_tokens, latency_ms, flags, extracted, created_at)
                   VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
                (run.get("run_id"), run.get("doc_id"), run.get("doc_type"), run.get("category"),
                 run.get("channel"), run.get("ocr_backend"), run.get("confidence"),
                 int(bool(res.get("posted"))), int(bool(run.get("hitl"))),
                 run.get("total_cost_usd", 0.0),
                 (run.get("total_input_tokens", 0) + run.get("total_output_tokens", 0)),
                 run.get("latency_ms", 0.0), json.dumps(run.get("flags", [])),
                 json.dumps(res.get("extracted")), time.time()),
            )
            self._conn.commit()

    def list_documents(self, category: str | None = None, limit: int = 100) -> list[dict]:
        with self._lock:
            if category and category != "all":
                rows = self._conn.execute(
                    "SELECT * FROM documents WHERE category=? ORDER BY created_at DESC LIMIT ?",
                    (category, limit)).fetchall()
            else:
                rows = self._conn.execute(
                    "SELECT * FROM documents ORDER BY created_at DESC LIMIT ?", (limit,)).fetchall()
        return [self._doc_row(r) for r in rows]

    def get_document(self, run_id: str) -> dict | None:
        with self._lock:
            r = self._conn.execute("SELECT * FROM documents WHERE run_id=?", (run_id,)).fetchone()
        return self._doc_row(r) if r else None

    def category_counts(self) -> dict:
        with self._lock:
            rows = self._conn.execute(
                "SELECT category, COUNT(*) c FROM documents GROUP BY category").fetchall()
        return {r["category"]: r["c"] for r in rows}

    @staticmethod
    def _doc_row(r: sqlite3.Row) -> dict:
        d = dict(r)
        for k in ("flags", "extracted"):
            if d.get(k):
                try:
                    d[k] = json.loads(d[k])
                except (json.JSONDecodeError, TypeError):
                    pass
        d["posted"] = bool(d.get("posted"))
        d["requires_review"] = bool(d.get("requires_review"))
        return d

    # --- audit ---------------------------------------------------------------
    def audit(self, action: str, run_id: str | None = None, actor: str = "system",
              detail: dict | None = None) -> None:
        with self._lock:
            self._conn.execute(
                "INSERT INTO audit_log (ts, action, run_id, actor, detail) VALUES (?,?,?,?,?)",
                (time.time(), action, run_id, actor, json.dumps(detail or {})))
            self._conn.commit()

    def recent_audit(self, limit: int = 100) -> list[dict]:
        with self._lock:
            rows = self._conn.execute(
                "SELECT * FROM audit_log ORDER BY ts DESC LIMIT ?", (limit,)).fetchall()
        out = []
        for r in rows:
            d = dict(r)
            try:
                d["detail"] = json.loads(d["detail"] or "{}")
            except (json.JSONDecodeError, TypeError):
                pass
            out.append(d)
        return out

    # --- prompt management ---------------------------------------------------
    def _seed_prompts(self):
        from .prompts import (
            BROWSER_AGENT_SYSTEM, CLASSIFY_SYSTEM, COMPLEX_REFINE_SYSTEM,
            EXTRACT_SYSTEM, NORMALIZE_SYSTEM,
        )
        defaults = {
            "classify": CLASSIFY_SYSTEM, "extract": EXTRACT_SYSTEM,
            "normalize": NORMALIZE_SYSTEM, "browser_agent": BROWSER_AGENT_SYSTEM,
            "complex_refine": COMPLEX_REFINE_SYSTEM,
        }
        with self._lock:
            for name, content in defaults.items():
                exists = self._conn.execute(
                    "SELECT 1 FROM prompts WHERE name=?", (name,)).fetchone()
                if not exists:
                    self._conn.execute(
                        "INSERT INTO prompts (name, version, content, updated_at, updated_by) "
                        "VALUES (?,?,?,?,?)", (name, 1, content, time.time(), "seed"))
            self._conn.commit()

    def list_prompts(self) -> list[dict]:
        with self._lock:
            rows = self._conn.execute(
                "SELECT name, version, length(content) AS chars, updated_at, updated_by "
                "FROM prompts ORDER BY name").fetchall()
        return [dict(r) for r in rows]

    def get_prompt(self, name: str) -> dict | None:
        with self._lock:
            r = self._conn.execute("SELECT * FROM prompts WHERE name=?", (name,)).fetchone()
        return dict(r) if r else None

    def set_prompt(self, name: str, content: str, actor: str = "admin") -> dict:
        with self._lock:
            cur = self._conn.execute("SELECT version FROM prompts WHERE name=?", (name,)).fetchone()
            version = (cur["version"] + 1) if cur else 1
            self._conn.execute(
                "INSERT OR REPLACE INTO prompts (name, version, content, updated_at, updated_by) "
                "VALUES (?,?,?,?,?)", (name, version, content, time.time(), actor))
            self._conn.execute(
                "INSERT INTO prompt_history (name, version, content, ts, actor) VALUES (?,?,?,?,?)",
                (name, version, content, time.time(), actor))
            self._conn.commit()
        return {"name": name, "version": version}

    def prompt_history(self, name: str, limit: int = 20) -> list[dict]:
        with self._lock:
            rows = self._conn.execute(
                "SELECT version, ts, actor, length(content) chars FROM prompt_history "
                "WHERE name=? ORDER BY version DESC LIMIT ?", (name, limit)).fetchall()
        return [dict(r) for r in rows]