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
        """
    )


@register(5, "automations (v5.1.0 rules engine)")
def _migration_automations(conn: sqlite3.Connection) -> None:
    """v5.1.0: database automations — if-this-then-that rule engine (trigger +
    condition + action) and clickable buttons that trigger actions.

    ``automations``      — the rules (event/cron/button trigger, optional
                           condition JSON, actions JSON, run counters).
    ``automation_runs``  — execution history for auditing and the Settings UI.
    """
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS automations (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            workspace TEXT NOT NULL DEFAULT '',
            name TEXT NOT NULL,
            trigger_type TEXT NOT NULL DEFAULT 'event',  -- event | cron | button
            event TEXT NOT NULL DEFAULT 'page.created',  -- for trigger_type='event'
            cron_expression TEXT NOT NULL DEFAULT '',    -- for trigger_type='cron'
            collection_id INTEGER,                       -- optional scope (event triggers)
            condition_json TEXT NOT NULL DEFAULT '[]',   -- list of condition clauses
            actions_json TEXT NOT NULL DEFAULT '[]',     -- list of action descriptors
            enabled BOOLEAN NOT NULL DEFAULT 1,
            created_by INTEGER,
            last_run_at TIMESTAMP,
            run_count INTEGER NOT NULL DEFAULT 0,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_automations_trigger ON automations(trigger_type, event, enabled)"
    )
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS automation_runs (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            automation_id INTEGER NOT NULL REFERENCES automations(id) ON DELETE CASCADE,
            trigger_source TEXT NOT NULL DEFAULT 'event',
            status TEXT NOT NULL DEFAULT 'fired',   -- fired | skipped | error
            detail TEXT NOT NULL DEFAULT '',
            collection_id INTEGER,
            page_id INTEGER,
            created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        )
        """
    )
    conn.execute(
        "CREATE INDEX IF NOT EXISTS idx_automation_runs_auto ON automation_runs(automation_id, created_at)"
    )


@register(4, "database templates (icon) + property validation")
def _migration_db_templates_validation(conn: sqlite3.Connection) -> None:
    """v5.3.0: database templates get an icon, properties a validation config,
    and the built-in database templates are seeded (idempotently)."""
    _cols = {r[1] for r in conn.execute("PRAGMA table_info(database_templates)").fetchall()}
    if "icon" not in _cols:
        conn.execute("ALTER TABLE database_templates ADD COLUMN icon TEXT NOT NULL DEFAULT '📋'")

    _pcols = {r[1] for r in conn.execute("PRAGMA table_info(collection_properties)").fetchall()}
    if "validation_json" not in _pcols:
        conn.execute("ALTER TABLE collection_properties ADD COLUMN validation_json TEXT NOT NULL DEFAULT '{}'")

    # Seed built-in templates (idempotent: only missing names are inserted).
    from app.services.db_templates import SEED_TEMPLATES
    for tpl in SEED_TEMPLATES:
        conn.execute(
            """INSERT OR IGNORE INTO database_templates (name, icon, description, schema_json)
               VALUES (?, ?, ?, ?)""",
            (tpl["name"], tpl.get("icon", "📋"), tpl.get("description", ""),
             __import__("json").dumps(tpl.get("schema", []))),
        )