src/densefeed/schema.py

"""DuckDB schema — DDL for all DenseFeed tables."""

from __future__ import annotations

import duckdb

SCHEMA_SQL = """
-- Run metadata
CREATE TABLE IF NOT EXISTS runs (
    run_id        TEXT PRIMARY KEY,
    date          TEXT NOT NULL,
    seq           INTEGER NOT NULL,
    user_id       TEXT DEFAULT '',
    status        TEXT NOT NULL DEFAULT 'running',
    last_completed_stage INTEGER DEFAULT 0,
    started_at    TEXT NOT NULL,
    finished_at   TEXT,
    config_json   TEXT DEFAULT '{}',
    UNIQUE(date, seq)
);

-- Fetched stories (Stage 1)
CREATE SEQUENCE IF NOT EXISTS items_id_seq;
CREATE TABLE IF NOT EXISTS items (
    id            INTEGER PRIMARY KEY DEFAULT nextval('items_id_seq'),
    run_id        TEXT NOT NULL,
    source        TEXT NOT NULL,
    title         TEXT NOT NULL,
    url           TEXT NOT NULL,
    summary       TEXT DEFAULT '',
    published_ts  DOUBLE DEFAULT 0.0,
    score         DOUBLE,
    reason        TEXT
);

-- Extracted articles (Stage 3)
CREATE SEQUENCE IF NOT EXISTS articles_id_seq;
CREATE TABLE IF NOT EXISTS articles (
    id            INTEGER PRIMARY KEY DEFAULT nextval('articles_id_seq'),
    run_id        TEXT NOT NULL,
    url           TEXT NOT NULL,
    title         TEXT NOT NULL,
    body          TEXT DEFAULT '',
    author        TEXT DEFAULT '',
    published     TEXT DEFAULT ''
);

-- Podcast dialogue lines (Stage 4)
CREATE SEQUENCE IF NOT EXISTS script_lines_id_seq;
CREATE TABLE IF NOT EXISTS script_lines (
    id            INTEGER PRIMARY KEY DEFAULT nextval('script_lines_id_seq'),
    run_id        TEXT NOT NULL,
    line_idx      INTEGER NOT NULL,
    speaker       TEXT NOT NULL CHECK (speaker IN ('A', 'B')),
    text          TEXT NOT NULL,
    emotion       TEXT DEFAULT 'neutral'
);

-- Audio metadata (Stage 5)
CREATE SEQUENCE IF NOT EXISTS audio_id_seq;
CREATE TABLE IF NOT EXISTS audio (
    id              INTEGER PRIMARY KEY DEFAULT nextval('audio_id_seq'),
    run_id          TEXT NOT NULL UNIQUE,
    hf_repo_id      TEXT NOT NULL,
    hf_path         TEXT NOT NULL,
    public_url      TEXT NOT NULL,
    duration_s      DOUBLE DEFAULT 0.0,
    file_size_bytes INTEGER DEFAULT 0
);

-- Publication records (Stage 6)
CREATE SEQUENCE IF NOT EXISTS publications_id_seq;
CREATE TABLE IF NOT EXISTS publications (
    id                INTEGER PRIMARY KEY DEFAULT nextval('publications_id_seq'),
    run_id            TEXT NOT NULL UNIQUE,
    published_at      TEXT NOT NULL,
    notification_sent BOOLEAN DEFAULT FALSE
);

-- Per-stage execution log
CREATE SEQUENCE IF NOT EXISTS stage_log_id_seq;
CREATE TABLE IF NOT EXISTS stage_log (
    id          INTEGER PRIMARY KEY DEFAULT nextval('stage_log_id_seq'),
    run_id      TEXT NOT NULL,
    stage       INTEGER NOT NULL,
    name        TEXT NOT NULL,
    success     BOOLEAN NOT NULL,
    duration_s  DOUBLE DEFAULT 0.0,
    error       TEXT,
    items_in    INTEGER DEFAULT 0,
    items_out   INTEGER DEFAULT 0
);

-- ZeroGPU call metrics
CREATE SEQUENCE IF NOT EXISTS gpu_calls_id_seq;
CREATE TABLE IF NOT EXISTS gpu_calls (
    id            INTEGER PRIMARY KEY DEFAULT nextval('gpu_calls_id_seq'),
    run_id        TEXT NOT NULL,
    call_type     TEXT NOT NULL,
    queue_wait_s  DOUBLE DEFAULT 0.0,
    load_s        DOUBLE DEFAULT 0.0,
    first_token_s DOUBLE DEFAULT 0.0,
    decode_s      DOUBLE DEFAULT 0.0,
    tokens_in     INTEGER DEFAULT 0,
    tokens_out    INTEGER DEFAULT 0,
    reserved_s    DOUBLE DEFAULT 0.0,
    deadhead_s    DOUBLE DEFAULT 0.0,
    created_at    TEXT NOT NULL
);

-- URL dedup tracking
CREATE TABLE IF NOT EXISTS seen_urls (
    url               TEXT PRIMARY KEY,
    first_seen_run_id TEXT NOT NULL,
    first_seen_at     TEXT NOT NULL
);

-- Content-hash cache for rank calls
CREATE TABLE IF NOT EXISTS rank_cache (
    content_hash TEXT PRIMARY KEY,
    output_json  TEXT NOT NULL,
    created_at   TEXT NOT NULL,
    tokens_in    INTEGER DEFAULT 0,
    tokens_out   INTEGER DEFAULT 0
);

-- Per-user preferences
CREATE TABLE IF NOT EXISTS user_preferences (
    user_id                 TEXT PRIMARY KEY,
    topics                  TEXT DEFAULT '[]',
    excluded_sources        TEXT DEFAULT '[]',
    speaker_a_voice         TEXT DEFAULT 'en-US-AndrewMultilingualNeural',
    speaker_b_voice         TEXT DEFAULT 'en-US-AvaMultilingualNeural',
    speaker_a_rate          TEXT DEFAULT '+10%',
    speaker_b_rate          TEXT DEFAULT '+10%',
    cross_speaker_pause_ms  INTEGER DEFAULT 550,
    same_speaker_pause_ms   INTEGER DEFAULT 250,
    min_relevance           INTEGER DEFAULT 6,
    top_n_items             INTEGER DEFAULT 12,
    hf_dataset_repo         TEXT DEFAULT '',
    hf_audio_repo           TEXT DEFAULT '',
    updated_at              TEXT NOT NULL
);
"""

INDEX_SQL = """
CREATE INDEX IF NOT EXISTS idx_items_run_id    ON items(run_id);
CREATE INDEX IF NOT EXISTS idx_items_url       ON items(url);
CREATE INDEX IF NOT EXISTS idx_items_source    ON items(source);
CREATE INDEX IF NOT EXISTS idx_articles_run_id ON articles(run_id);
CREATE INDEX IF NOT EXISTS idx_script_run_id   ON script_lines(run_id);
CREATE INDEX IF NOT EXISTS idx_audio_run_id    ON audio(run_id);
CREATE INDEX IF NOT EXISTS idx_stage_log_run   ON stage_log(run_id);
CREATE INDEX IF NOT EXISTS idx_runs_date       ON runs(date);
CREATE INDEX IF NOT EXISTS idx_runs_user       ON runs(user_id);
CREATE INDEX IF NOT EXISTS idx_seen_urls_at    ON seen_urls(first_seen_at);
CREATE INDEX IF NOT EXISTS idx_gpu_calls_run   ON gpu_calls(run_id);
CREATE INDEX IF NOT EXISTS idx_gpu_calls_type  ON gpu_calls(call_type);
"""


def init_schema(conn: duckdb.DuckDBPyConnection) -> None:
    """Create all tables and indexes if they don't exist."""
    conn.execute(SCHEMA_SQL)
    conn.execute(INDEX_SQL)
    # Ensure columns exist in user_preferences for existing DB files
    for col, default in [("hf_dataset_repo", "''"), ("hf_audio_repo", "''")]:
        try:
            conn.execute(
                f"ALTER TABLE user_preferences ADD COLUMN {col} TEXT DEFAULT {default}"
            )
        except duckdb.CatalogException:
            pass  # Column already exists — expected
    try:
        conn.execute(
            "ALTER TABLE runs ADD COLUMN last_completed_stage INTEGER DEFAULT 0"
        )
    except duckdb.CatalogException:
        pass  # Column already exists — expected