upbeatBytes/goodnews/db.py

from __future__ import annotations

import sqlite3
from pathlib import Path


SCHEMA = """
PRAGMA foreign_keys = ON;

CREATE TABLE IF NOT EXISTS sources (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL UNIQUE,
    homepage_url TEXT,
    feed_url TEXT NOT NULL UNIQUE,
    source_type TEXT NOT NULL DEFAULT 'rss',
    default_category TEXT,
    trust_score INTEGER NOT NULL DEFAULT 5,
    pr_risk_score INTEGER NOT NULL DEFAULT 3,
    active INTEGER NOT NULL DEFAULT 1,
    poll_interval_minutes INTEGER NOT NULL DEFAULT 60,
    notes TEXT,
    last_success_at TEXT,
    last_error_at TEXT,
    last_error TEXT,
    consecutive_failures INTEGER NOT NULL DEFAULT 0,
    review_flag INTEGER NOT NULL DEFAULT 0,
    review_reason TEXT,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS articles (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    source_id INTEGER NOT NULL REFERENCES sources(id) ON DELETE CASCADE,
    canonical_url TEXT NOT NULL,
    title TEXT NOT NULL,
    description TEXT,
    author TEXT,
    published_at TEXT,
    discovered_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    image_url TEXT,
    language TEXT,
    raw_guid TEXT,
    url_hash TEXT NOT NULL UNIQUE,
    title_hash TEXT,
    duplicate_of INTEGER REFERENCES articles(id) ON DELETE SET NULL,
    image_checked_at TEXT,
    FOREIGN KEY (source_id) REFERENCES sources(id)
);

CREATE INDEX IF NOT EXISTS idx_articles_published_at ON articles(published_at);
CREATE INDEX IF NOT EXISTS idx_articles_source_id ON articles(source_id);
CREATE INDEX IF NOT EXISTS idx_articles_title_hash ON articles(title_hash);

CREATE TABLE IF NOT EXISTS article_scores (
    article_id INTEGER PRIMARY KEY REFERENCES articles(id) ON DELETE CASCADE,
    constructive_score INTEGER,
    cortisol_score INTEGER,
    ragebait_score INTEGER,
    agency_score INTEGER,
    human_benefit_score INTEGER,
    novelty_score INTEGER,
    pr_risk_score INTEGER,
    accepted INTEGER,
    reason_code TEXT,
    reason_text TEXT,
    topic TEXT,
    flavor TEXT,
    model_name TEXT,
    scored_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS article_tags (
    article_id INTEGER NOT NULL REFERENCES articles(id) ON DELETE CASCADE,
    tag TEXT NOT NULL,
    PRIMARY KEY (article_id, tag)
);

CREATE INDEX IF NOT EXISTS idx_article_tags_tag ON article_tags(tag);

CREATE TABLE IF NOT EXISTS article_embeddings (
    article_id INTEGER PRIMARY KEY REFERENCES articles(id) ON DELETE CASCADE,
    vector BLOB NOT NULL,
    dim INTEGER NOT NULL,
    model TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS ingest_runs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    source_id INTEGER REFERENCES sources(id) ON DELETE SET NULL,
    started_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    finished_at TEXT,
    status TEXT NOT NULL DEFAULT 'running',
    items_seen INTEGER NOT NULL DEFAULT 0,
    items_inserted INTEGER NOT NULL DEFAULT 0,
    items_duplicate INTEGER NOT NULL DEFAULT 0,
    error TEXT
);

CREATE TABLE IF NOT EXISTS source_candidates (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    feed_url TEXT NOT NULL UNIQUE,
    homepage_url TEXT,
    name TEXT,
    status TEXT NOT NULL DEFAULT 'suggested',
    preview_json TEXT,
    notes TEXT,
    last_previewed_at TEXT,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE IF NOT EXISTS daily_briefs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    brief_date TEXT NOT NULL UNIQUE,
    title TEXT NOT NULL,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    notes TEXT
);

CREATE TABLE IF NOT EXISTS daily_brief_items (
    brief_id INTEGER NOT NULL REFERENCES daily_briefs(id) ON DELETE CASCADE,
    article_id INTEGER NOT NULL REFERENCES articles(id) ON DELETE CASCADE,
    rank INTEGER NOT NULL,
    selection_reason TEXT,
    PRIMARY KEY (brief_id, article_id),
    UNIQUE (brief_id, rank)
);

-- ---- Accounts ----------------------------------------------------------------
-- Self-hosted, minimal-PII. The host ingestion owns the content tables above;
-- the API owns these (writes happen via the API, so the DB runs in WAL mode).

CREATE TABLE IF NOT EXISTS users (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    email TEXT NOT NULL UNIQUE,
    display_name TEXT,
    avatar_url TEXT,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);

-- One row per sign-in method linked to a user; lets Google + magic-link
-- (same verified email) resolve to a single account.
CREATE TABLE IF NOT EXISTS identities (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    provider TEXT NOT NULL,           -- 'email' | 'google' | 'apple'
    provider_subject TEXT NOT NULL,   -- email address, or the provider's stable user id
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    UNIQUE (provider, provider_subject)
);
CREATE INDEX IF NOT EXISTS idx_identities_user ON identities(user_id);

-- Single-use, short-lived magic-link tokens (stored hashed).
CREATE TABLE IF NOT EXISTS login_tokens (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    email TEXT NOT NULL,
    token_hash TEXT NOT NULL UNIQUE,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    expires_at TEXT NOT NULL,
    consumed_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_login_tokens_email ON login_tokens(email);

-- Active sessions (opaque token stored hashed); validated for cookie or bearer.
CREATE TABLE IF NOT EXISTS sessions (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    token_hash TEXT NOT NULL UNIQUE,
    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    last_seen_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    expires_at TEXT NOT NULL,
    user_agent TEXT
);
CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);

CREATE TABLE IF NOT EXISTS saved_articles (
    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    article_id INTEGER NOT NULL REFERENCES articles(id) ON DELETE CASCADE,
    saved_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (user_id, article_id)
);

CREATE TABLE IF NOT EXISTS user_history (
    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
    article_id INTEGER NOT NULL REFERENCES articles(id) ON DELETE CASCADE,
    event TEXT NOT NULL DEFAULT 'seen',   -- 'seen' | 'dismissed'
    at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
    PRIMARY KEY (user_id, article_id, event)
);

CREATE TABLE IF NOT EXISTS user_prefs (
    user_id INTEGER PRIMARY KEY REFERENCES users(id) ON DELETE CASCADE,
    prefs_json TEXT NOT NULL DEFAULT '{}',
    updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
);
"""


def connect(db_path: Path | str) -> sqlite3.Connection:
    path = Path(db_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    conn = sqlite3.connect(path, check_same_thread=False)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA foreign_keys = ON")
    # WAL lets the API write account data while the ingestion cycle writes content
    # concurrently (readers never block the writer). busy_timeout rides out the
    # brief moments the single writer lock is held. Both are no-ops if already set.
    conn.execute("PRAGMA busy_timeout = 5000")
    if str(path) != ":memory:":
        conn.execute("PRAGMA journal_mode = WAL")
        conn.execute("PRAGMA synchronous = NORMAL")
    return conn


def init_db(conn: sqlite3.Connection) -> None:
    conn.executescript(SCHEMA)
    _migrate(conn)
    conn.commit()


def _migrate(conn: sqlite3.Connection) -> None:
    """Add columns introduced after the initial schema to existing databases.

    CREATE TABLE IF NOT EXISTS never alters an existing table, so new columns
    need an explicit, idempotent ALTER guarded by the current column set.
    """
    score_cols = {row["name"] for row in conn.execute("PRAGMA table_info(article_scores)")}
    for column in ("topic", "flavor"):
        if column not in score_cols:
            conn.execute(f"ALTER TABLE article_scores ADD COLUMN {column} TEXT")

    # users.avatar_url added for Google profile pictures.
    user_tbl = {row["name"] for row in conn.execute("PRAGMA table_info(users)")}
    if user_tbl and "avatar_url" not in user_tbl:
        conn.execute("ALTER TABLE users ADD COLUMN avatar_url TEXT")

    article_cols = {row["name"] for row in conn.execute("PRAGMA table_info(articles)")}
    if "duplicate_of" not in article_cols:
        conn.execute(
            "ALTER TABLE articles ADD COLUMN duplicate_of INTEGER REFERENCES articles(id)"
        )
    if "image_checked_at" not in article_cols:
        conn.execute("ALTER TABLE articles ADD COLUMN image_checked_at TEXT")
    # Created here (not in SCHEMA) so it runs after the column exists on upgrades.
    conn.execute("CREATE INDEX IF NOT EXISTS idx_articles_duplicate_of ON articles(duplicate_of)")

    source_cols = {row["name"] for row in conn.execute("PRAGMA table_info(sources)")}
    health_columns = {
        "last_success_at": "TEXT",
        "last_error_at": "TEXT",
        "last_error": "TEXT",
        "consecutive_failures": "INTEGER NOT NULL DEFAULT 0",
        "review_flag": "INTEGER NOT NULL DEFAULT 0",
        "review_reason": "TEXT",
    }
    for column, decl in health_columns.items():
        if column not in source_cols:
            conn.execute(f"ALTER TABLE sources ADD COLUMN {column} {decl}")