db.py

"""Kith Local — SQLite storage layer.

Local file storage is the product story: your conversations never leave the
machine. One connection per call (handlers may run concurrently on ZeroGPU).
"""

import os
import sqlite3
from datetime import date, datetime, timedelta

DB_PATH = os.environ.get("KITH_DB", os.path.join(os.path.dirname(__file__), "data", "kith.db"))

SCHEMA = """
CREATE TABLE IF NOT EXISTS conversations (
  id INTEGER PRIMARY KEY,
  title TEXT,
  transcript TEXT NOT NULL,
  occurred_at TEXT NOT NULL,
  processed_at TEXT
);
CREATE TABLE IF NOT EXISTS people (
  id INTEGER PRIMARY KEY,
  name TEXT NOT NULL,
  name_key TEXT NOT NULL UNIQUE,
  relationship TEXT NOT NULL DEFAULT 'other',
  summary TEXT NOT NULL DEFAULT '',
  last_contact TEXT,
  talk_count INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS signals (
  id INTEGER PRIMARY KEY,
  person_id INTEGER NOT NULL REFERENCES people(id),
  type TEXT NOT NULL,
  text TEXT NOT NULL,
  event_on TEXT,
  quote TEXT,
  source_id INTEGER REFERENCES conversations(id)
);
CREATE TABLE IF NOT EXISTS commitments (
  id INTEGER PRIMARY KEY,
  person_id INTEGER NOT NULL REFERENCES people(id),
  owner TEXT NOT NULL CHECK (owner IN ('me','them')),
  text TEXT NOT NULL,
  promised_on TEXT,
  due TEXT,
  status TEXT NOT NULL DEFAULT 'open',
  quote TEXT,
  source_id INTEGER REFERENCES conversations(id)
);
CREATE TABLE IF NOT EXISTS decisions (
  id INTEGER PRIMARY KEY,
  text TEXT NOT NULL,
  rationale TEXT,
  decided_on TEXT,
  source_id INTEGER REFERENCES conversations(id)
);
-- A dismissed card hides the suggestion without forgetting the underlying
-- memory: the feed query filters these out, the person/signal row stays.
CREATE TABLE IF NOT EXISTS dismissals (
  card_id TEXT PRIMARY KEY,
  dismissed_at TEXT NOT NULL
);
"""


def connect() -> sqlite3.Connection:
    os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
    conn = sqlite3.connect(DB_PATH, timeout=15)
    conn.row_factory = sqlite3.Row
    conn.executescript(SCHEMA)
    _migrate(conn)
    return conn


def _migrate(conn: sqlite3.Connection) -> None:
    """Add columns that may be missing on a DB created by an earlier version."""
    for table in ("signals", "commitments"):
        cols = {r["name"] for r in conn.execute(f"PRAGMA table_info({table})")}
        if "quote" not in cols:
            conn.execute(f"ALTER TABLE {table} ADD COLUMN quote TEXT")


def _name_key(name: str) -> str:
    return " ".join(name.lower().split())


def insert_conversation(transcript: str, title: str | None = None) -> int:
    with connect() as conn:
        cur = conn.execute(
            "INSERT INTO conversations (title, transcript, occurred_at) VALUES (?, ?, ?)",
            (title or "Conversation", transcript, datetime.now().isoformat(timespec="seconds")),
        )
        return cur.lastrowid


def apply_extraction(conversation_id: int, extraction: dict, today: str | None = None) -> dict:
    """Upsert extracted people, then attach signals/commitments/decisions.

    Returns counts per entity type for the UI.
    """
    today = today or date.today().isoformat()
    counts = {"people": 0, "signals": 0, "commitments": 0, "decisions": 0}
    with connect() as conn:
        ids_by_key: dict[str, int] = {}
        for p in extraction.get("people", []):
            key = _name_key(p["name"])
            conn.execute(
                """INSERT INTO people (name, name_key, relationship, summary, last_contact, talk_count)
                   VALUES (?, ?, ?, ?, ?, 1)
                   ON CONFLICT(name_key) DO UPDATE SET
                     relationship = excluded.relationship,
                     summary = excluded.summary,
                     last_contact = excluded.last_contact,
                     talk_count = people.talk_count + 1""",
                (p["name"], key, p.get("relationship", "other"), p.get("summary", ""), today),
            )
            counts["people"] += 1
        for row in conn.execute("SELECT id, name_key FROM people"):
            ids_by_key[row["name_key"]] = row["id"]

        def pid(name: str) -> int | None:
            return ids_by_key.get(_name_key(name or ""))

        for s in extraction.get("signals", []):
            person_id = pid(s.get("person"))
            if person_id is None or not s.get("text"):
                continue
            dup = conn.execute(
                "SELECT 1 FROM signals WHERE person_id = ? AND type = ? AND text = ?",
                (person_id, s.get("type", "preference"), s["text"]),
            ).fetchone()
            if dup:
                continue
            conn.execute(
                "INSERT INTO signals (person_id, type, text, event_on, quote, source_id) VALUES (?, ?, ?, ?, ?, ?)",
                (person_id, s.get("type", "preference"), s["text"], s.get("event_on"), s.get("quote"), conversation_id),
            )
            counts["signals"] += 1

        for c in extraction.get("commitments", []):
            person_id = pid(c.get("person"))
            if person_id is None or not c.get("text"):
                continue
            dup = conn.execute(
                "SELECT 1 FROM commitments WHERE person_id = ? AND owner = ? AND text = ? AND status = 'open'",
                (person_id, c.get("owner", "me"), c["text"]),
            ).fetchone()
            if dup:
                continue
            conn.execute(
                """INSERT INTO commitments (person_id, owner, text, promised_on, due, status, quote, source_id)
                   VALUES (?, ?, ?, ?, ?, 'open', ?, ?)""",
                (person_id, c.get("owner", "me"), c["text"], c.get("promised_on", today), c.get("due"), c.get("quote"), conversation_id),
            )
            counts["commitments"] += 1

        for d in extraction.get("decisions", []):
            if not d.get("text"):
                continue
            if conn.execute("SELECT 1 FROM decisions WHERE text = ?", (d["text"],)).fetchone():
                continue
            conn.execute(
                "INSERT INTO decisions (text, rationale, decided_on, source_id) VALUES (?, ?, ?, ?)",
                (d["text"], d.get("rationale"), d.get("decided_on", today), conversation_id),
            )
            counts["decisions"] += 1

        conn.execute(
            "UPDATE conversations SET processed_at = ? WHERE id = ?",
            (datetime.now().isoformat(timespec="seconds"), conversation_id),
        )
    return counts


def today_feed(today: str | None = None, limit: int = 6) -> list[dict]:
    """The Today feed. Four tiers, most-actionable first:
      thoughtful — a dated event coming up (birthday, appointment) within 14 days
      owed       — an open promise YOU made (overdue ones first)
      reconnect  — a personal contact you haven't spoken to in 21+ days
      remember   — something meaningful you just learned (undated signal)
    The 'remember' tier is what guarantees a fresh capture always produces
    cards: right after recording, nothing is yet overdue or in the future, but
    you've still learned things worth surfacing.
    """
    today_d = date.fromisoformat(today) if today else date.today()
    today_s = today_d.isoformat()
    horizon = (today_d + timedelta(days=14)).isoformat()
    reconnect_before = (today_d - timedelta(days=21)).isoformat()
    cards: list[dict] = []
    with connect() as conn:
        dismissed = {r["card_id"] for r in conn.execute("SELECT card_id FROM dismissals")}

        # 1. thoughtful — upcoming dated events
        rows = conn.execute(
            """SELECT s.*, p.name FROM signals s JOIN people p ON p.id = s.person_id
               WHERE s.type IN ('date','want','life_event') AND s.event_on IS NOT NULL
                 AND s.event_on >= ? AND s.event_on <= ?
               ORDER BY s.event_on, (s.type = 'date') DESC LIMIT ?""",
            (today_s, horizon, limit),
        ).fetchall()
        seen_events: set[tuple[int, str]] = set()
        for r in rows:
            event_key = (r["person_id"], r["event_on"])
            if event_key in seen_events:
                continue  # one card per person per dated event
            seen_events.add(event_key)
            days = (date.fromisoformat(r["event_on"]) - today_d).days
            when = "today" if days == 0 else ("tomorrow" if days == 1 else f"in {days} days")
            want = conn.execute(
                "SELECT text FROM signals WHERE person_id = ? AND type = 'want' AND id != ? ORDER BY id DESC",
                (r["person_id"], r["id"]),
            ).fetchone()
            cards.append({
                "id": f"thoughtful:{r['id']}",
                "kind": "thoughtful",
                "personName": r["name"],
                "title": f"{r['text']} — {when}",
                "why": want["text"] if want else f"You talked about this with {r['name']} recently.",
            })

        # 2. owed — every open promise I made; overdue first, newest of the rest next
        rows = conn.execute(
            """SELECT c.*, p.name FROM commitments c JOIN people p ON p.id = c.person_id
               WHERE c.owner = 'me' AND c.status = 'open'
               ORDER BY (c.due IS NOT NULL AND c.due <= ?) DESC, c.due, c.id DESC
               LIMIT ?""",
            (today_s, limit),
        ).fetchall()
        for r in rows:
            overdue = r["due"] and r["due"] <= today_s
            if overdue:
                why = f"Was due {r['due']}"
            elif r["due"]:
                why = f"Due {r['due']}"
            elif r["promised_on"]:
                why = f"You offered this on {r['promised_on']}"
            else:
                why = f"Something you offered {r['name']}"
            cards.append({
                "id": f"owed:{r['id']}",
                "kind": "owed",
                "personName": r["name"],
                "title": f"You owe {r['name']}: {r['text']}",
                "why": why,
            })

        # 3. reconnect — personal contacts gone quiet
        rows = conn.execute(
            """SELECT * FROM people
               WHERE relationship = 'personal' AND last_contact IS NOT NULL AND last_contact <= ?
               ORDER BY last_contact LIMIT ?""",
            (reconnect_before, limit),
        ).fetchall()
        for r in rows:
            cards.append({
                "id": f"reconnect:{r['id']}",
                "kind": "reconnect",
                "personName": r["name"],
                "title": f"It's been a while since you talked to {r['name']}",
                "why": r["summary"] or f"Last contact {r['last_contact']}.",
            })

        # 4. remember — meaningful undated things you just learned (fills a fresh feed)
        rows = conn.execute(
            """SELECT s.*, p.name FROM signals s JOIN people p ON p.id = s.person_id
               WHERE s.type IN ('want','life_event','health','preference') AND s.event_on IS NULL
               ORDER BY s.id DESC LIMIT ?""",
            (limit,),
        ).fetchall()
        verb = {"want": "wants", "life_event": "", "health": "", "preference": "mentioned"}
        for r in rows:
            cards.append({
                "id": f"remember:{r['id']}",
                "kind": "remember",
                "personName": r["name"],
                "title": r["text"],
                "why": f"From your conversation with {r['name']}",
            })

    # de-dupe, drop dismissed, cap. Tiers are already in priority order.
    seen_ids: set[str] = set()
    out: list[dict] = []
    for c in cards:
        if c["id"] in dismissed or c["id"] in seen_ids:
            continue
        seen_ids.add(c["id"])
        out.append(c)
    return out[:limit]


def dismiss_card(card_id: str) -> None:
    """Hide one suggestion. The underlying memory is kept."""
    with connect() as conn:
        conn.execute(
            "INSERT OR IGNORE INTO dismissals (card_id, dismissed_at) VALUES (?, ?)",
            (card_id, datetime.now().isoformat(timespec="seconds")),
        )


# How each kind of card maps back to the row that produced it, so we can show
# its source quote and find the person it's about.
_CARD_SOURCE = {
    "thoughtful": ("signals", "person_id"),
    "remember": ("signals", "person_id"),
    "owed": ("commitments", "person_id"),
    "reconnect": ("people", "id"),
}


def card_detail(card_id: str) -> dict:
    """Everything behind one card: the source quote + the rest of what we
    remember about that person. This is what makes a card auditable."""
    kind, _, rid = card_id.partition(":")
    if kind not in _CARD_SOURCE or not rid.isdigit():
        return {}
    table, person_col = _CARD_SOURCE[kind]
    with connect() as conn:
        row = conn.execute(f"SELECT * FROM {table} WHERE id = ?", (int(rid),)).fetchone()
        if row is None:
            return {}
        person_id = row[person_col]
        person = conn.execute("SELECT * FROM people WHERE id = ?", (person_id,)).fetchone()
        if person is None:
            return {}

        quote = row["quote"] if kind in ("thoughtful", "owed", "remember") else None
        # the conversation this fact came from (for "from your conversation on …")
        occurred = None
        if kind in ("thoughtful", "owed", "remember") and row["source_id"]:
            conv = conn.execute(
                "SELECT occurred_at FROM conversations WHERE id = ?", (row["source_id"],)
            ).fetchone()
            occurred = conv["occurred_at"][:10] if conv else None

        memories = []
        for s in conn.execute(
            "SELECT type, text, event_on FROM signals WHERE person_id = ? ORDER BY id", (person_id,)
        ):
            memories.append({"kind": s["type"], "text": s["text"], "date": s["event_on"]})
        for c in conn.execute(
            "SELECT owner, text, due, status FROM commitments WHERE person_id = ? ORDER BY id", (person_id,)
        ):
            label = ("You owe" if c["owner"] == "me" else f"{person['name']} owes you")
            memories.append({"kind": "commitment", "text": f"{label}: {c['text']}", "date": c["due"]})

        return {
            "kind": kind,
            "person": {
                "name": person["name"],
                "relationship": person["relationship"],
                "summary": person["summary"],
                "talkCount": person["talk_count"],
                "lastContact": person["last_contact"],
            },
            "quote": quote,
            "occurredOn": occurred,
            "memories": memories,
        }


def _conversation_title(conn: sqlite3.Connection, conv_id: int) -> str:
    """A human title for a conversation: the people it's about, else a snippet."""
    names = [
        r["name"]
        for r in conn.execute(
            """SELECT DISTINCT p.name FROM people p
               WHERE p.id IN (SELECT person_id FROM signals WHERE source_id = ?)
                  OR p.id IN (SELECT person_id FROM commitments WHERE source_id = ?)
               ORDER BY p.name""",
            (conv_id, conv_id),
        )
    ]
    if names:
        if len(names) == 1:
            return names[0]
        if len(names) == 2:
            return f"{names[0]} & {names[1]}"
        return f"{names[0]}, {names[1]} +{len(names) - 2}"
    row = conn.execute("SELECT transcript FROM conversations WHERE id = ?", (conv_id,)).fetchone()
    snippet = (row["transcript"][:38].strip() + "…") if row and row["transcript"] else "Conversation"
    return snippet


def list_conversations(limit: int = 50) -> list[dict]:
    """Newest-first history for the sidebar, with per-conversation counts."""
    out = []
    with connect() as conn:
        rows = conn.execute(
            "SELECT id, occurred_at FROM conversations ORDER BY id DESC LIMIT ?", (limit,)
        ).fetchall()
        for r in rows:
            cid = r["id"]
            people = conn.execute(
                """SELECT COUNT(DISTINCT person_id) AS n FROM (
                       SELECT person_id FROM signals WHERE source_id = ?
                       UNION SELECT person_id FROM commitments WHERE source_id = ?
                   )""",
                (cid, cid),
            ).fetchone()["n"]
            signals = conn.execute("SELECT COUNT(*) AS n FROM signals WHERE source_id = ?", (cid,)).fetchone()["n"]
            commitments = conn.execute("SELECT COUNT(*) AS n FROM commitments WHERE source_id = ?", (cid,)).fetchone()["n"]
            out.append({
                "id": cid,
                "title": _conversation_title(conn, cid),
                "occurredOn": r["occurred_at"][:10],
                "counts": {"people": people, "signals": signals, "commitments": commitments},
            })
    return out


def conversation_detail(conversation_id: int) -> dict:
    """Full record of one past conversation: transcript + what was extracted."""
    with connect() as conn:
        conv = conn.execute(
            "SELECT * FROM conversations WHERE id = ?", (conversation_id,)
        ).fetchone()
        if conv is None:
            return {}
        memories = []
        for s in conn.execute(
            """SELECT s.type, s.text, s.event_on, p.name FROM signals s
               JOIN people p ON p.id = s.person_id WHERE s.source_id = ? ORDER BY s.id""",
            (conversation_id,),
        ):
            memories.append({"kind": s["type"], "person": s["name"], "text": s["text"], "date": s["event_on"]})
        for c in conn.execute(
            """SELECT c.owner, c.text, c.due, p.name FROM commitments c
               JOIN people p ON p.id = c.person_id WHERE c.source_id = ? ORDER BY c.id""",
            (conversation_id,),
        ):
            label = "You owe" if c["owner"] == "me" else f"{c['name']} owes you"
            memories.append({"kind": "commitment", "person": c["name"], "text": f"{label}: {c['text']}", "date": c["due"]})
        for d in conn.execute(
            "SELECT text FROM decisions WHERE source_id = ? ORDER BY id", (conversation_id,)
        ):
            memories.append({"kind": "decision", "person": None, "text": d["text"], "date": None})
        return {
            "id": conv["id"],
            "title": _conversation_title(conn, conversation_id),
            "occurredOn": conv["occurred_at"][:10],
            "transcript": conv["transcript"],
            "memories": memories,
        }


def reset_all() -> dict:
    """Forget everything — the honest 'it's your local file' wipe."""
    with connect() as conn:
        for table in ("dismissals", "signals", "commitments", "decisions", "people", "conversations"):
            conn.execute(f"DELETE FROM {table}")
    return stats()


def stats() -> dict:
    with connect() as conn:
        out = {}
        for table in ("conversations", "people", "signals", "commitments", "decisions"):
            out[table] = conn.execute(f"SELECT COUNT(*) AS n FROM {table}").fetchone()["n"]
        return out