flowdeck/app/migrations.py

"""FlowDeck — versioned schema migrations (lightweight, no Alembic).

This replaces the previous "ad-hoc" approach where every new schema change was
appended directly to `app/db.py::init_db()` with no tracking. A `schema_version`
table now records the highest applied migration; the full baseline schema
(created idempotently by `init_db`) is treated as version 1, and any incremental
change is expressed as an ordered, versioned step below and applied exactly once.

Each migration function receives a raw ``sqlite3.Connection`` (WAL + foreign keys
already enabled) and must be written idempotently (``IF NOT EXISTS`` / guarded
``ALTER TABLE``) so it is safe even if partially re-run.
"""
from __future__ import annotations

import logging
import sqlite3
from typing import Callable, List, Tuple

logger = logging.getLogger(__name__)

# The full baseline schema created by `app.db::init_db()` is "version 1".
BASELINE_VERSION = 1

# (version, name, apply_fn). Kept sorted by version at registration time.
MIGRATIONS: List[Tuple[int, str, Callable[[sqlite3.Connection], None]]] = []


def register(version: int, name: str) -> Callable:
    """Decorator registering a migration in the ordered registry."""
    if any(v == version for v, _, _ in MIGRATIONS):
        raise ValueError(f"Duplicate migration version {version}")

    def decorator(fn: Callable[[sqlite3.Connection], None]):
        MIGRATIONS.append((version, name, fn))
        MIGRATIONS.sort(key=lambda item: item[0])
        return fn

    return decorator


def _ensure_table(conn: sqlite3.Connection) -> None:
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS schema_version (
            version INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """
    )


def current_version(conn: sqlite3.Connection) -> int:
    _ensure_table(conn)
    row = conn.execute(
        "SELECT COALESCE(MAX(version), 0) AS v FROM schema_version"
    ).fetchone()
    return int(row[0])


def fts5_available() -> bool:
    """True when the bundled SQLite ships the FTS5 extension."""
    probe = sqlite3.connect(":memory:")
    try:
        probe.execute("CREATE VIRTUAL TABLE _fts5_probe USING fts5(x)")
        return True
    except sqlite3.OperationalError:
        return False
    finally:
        probe.close()


def apply_migrations(conn: sqlite3.Connection) -> int:
    """Seal the baseline schema (version 1) and apply pending migrations.

    Returns the resulting schema version.
    """
    _ensure_table(conn)
    applied = current_version(conn)

    if applied < BASELINE_VERSION:
        # The pre-existing schema (already created by init_db) is our baseline.
        conn.execute(
            "INSERT OR IGNORE INTO schema_version (version, name) VALUES (?, ?)",
            (BASELINE_VERSION, "baseline"),
        )
        conn.commit()
        applied = BASELINE_VERSION

    for version, name, fn in MIGRATIONS:
        if version <= applied:
            continue
        fn(conn)
        conn.execute(
            "INSERT INTO schema_version (version, name) VALUES (?, ?)",
            (version, name),
        )
        conn.commit()
        applied = version
        logger.info("Applied migration %d: %s", version, name)

    return applied


# ═══════════════════════════════════════════════════════════════════════════
# Migrations
# ═══════════════════════════════════════════════════════════════════════════


@register(2, "missing indexes")
def _migration_missing_indexes(conn: sqlite3.Connection) -> None:
    """Add the indexes flagged in the roadmap (fast lookups by email, forge user)."""
    for ddl in (
        "CREATE INDEX IF NOT EXISTS idx_users_email ON users(email)",
        "CREATE INDEX IF NOT EXISTS idx_user_oauth_tokens_user ON user_oauth_tokens(user_id, provider)",
        "CREATE INDEX IF NOT EXISTS idx_collections_workspace ON collections(workspace_id)",
        "CREATE INDEX IF NOT EXISTS idx_pages_workspace ON pages(workspace_id)",
        "CREATE INDEX IF NOT EXISTS idx_pages_deleted ON pages(deleted_at)",
    ):
        conn.execute(ddl)


@register(3, "full-text search (FTS5)")
def _migration_fts5(conn: sqlite3.Connection) -> None:
    """Create a full-text index over pages (title + content) for the command palette.

    Kept in sync via row-level triggers on the ``pages`` table so page
    insert/update/delete are reflected immediately. Skips gracefully if the
    bundled SQLite lacks FTS5 (search then falls back to LIKE).
    """
    if not fts5_available():
        logger.warning("FTS5 unavailable — skipping full-text index (LIKE fallback active)")
        return

    conn.execute("CREATE VIRTUAL TABLE IF NOT EXISTS pages_fts USING fts5(title, body)")

    conn.execute(
        """
        CREATE TRIGGER IF NOT EXISTS pages_fts_ai AFTER INSERT ON pages BEGIN
            INSERT INTO pages_fts(rowid, title, body)
            VALUES (new.id, COALESCE(new.title, ''), COALESCE(new.content, ''));
        END
        """
    )
    # `pages_fts` is a standalone FTS5 table (it stores its own content), so deletes
    # use a plain DELETE by rowid (NOT the special 'delete' insert that only applies
    # to external-content/contentless FTS5 tables).
    conn.execute(
        """
        CREATE TRIGGER IF NOT EXISTS pages_fts_ad AFTER DELETE ON pages BEGIN
            DELETE FROM pages_fts WHERE rowid = old.id;
        END
        """
    )
    conn.execute(
        """
        CREATE TRIGGER IF NOT EXISTS pages_fts_au AFTER UPDATE ON pages BEGIN
            DELETE FROM pages_fts WHERE rowid = old.id;
            INSERT INTO pages_fts(rowid, title, body)
            VALUES (new.id, COALESCE(new.title, ''), COALESCE(new.content, ''));
        END
        """
    )

    # Backfill the index from any rows that already exist.
    conn.execute(
        """
        INSERT INTO pages_fts(rowid, title, body)
        SELECT id, COALESCE(title, ''), COALESCE(content, '') FROM pages
        WHERE deleted_at IS NULL
        """
    )