vignette/scripts/check-deploy-preflight.py

#!/usr/bin/env python3
"""Deployment preflight checks for portable Vignette installs.

This script is intentionally read-only.  Owner-run DB migrations/init scripts
must happen before the API starts; this preflight only proves that the runtime
app role and packaged files are ready enough for startup.
"""

from __future__ import annotations

import argparse
import asyncio
import json
import os
import re
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Any


REPO_ROOT = Path(__file__).resolve().parents[1]
REQUIRED_ENV_KEYS = (
    "ENVIRONMENT",
    "POSTGRES_PASSWORD",
    "APP_DB_PASSWORD",
    "OPENAI_API_KEY",
    "SESSION_SECRET",
    "OAUTH_GOOGLE_CLIENT_ID",
    "OAUTH_GOOGLE_CLIENT_SECRET",
)
PLACEHOLDER_VALUES = {
    "",
    "change-me",
    "change-me-app",
    "change-me-random",
    "dummy",
    "dummy-openai-key",
    "dummy-session-secret-not-real-12345",
    "dummy-client",
    "dummy-secret",
}
FORBIDDEN_PROD_TRUE_FLAGS = (
    "AUTH_DEV_LOGIN_ENABLED",
    "AUTO_SEED_PERSONAS",
    "ALLOW_SEED_PERSONA_FALLBACK",
    "VIGNETTE_VOICE_POC_SAMPLE_TTS",
)
VALID_ENGINE_MODES = {"claude_api", "claude_cli", "openai", "solar"}
REQUIREMENT_RE = re.compile(r"^[A-Za-z0-9_.-]+(?:\[[A-Za-z0-9_,.-]+\])?==[^\s#]+$")


@dataclass(frozen=True)
class Check:
    name: str
    ok: bool
    detail: str


def _rel(path: Path) -> str:
    try:
        return path.relative_to(REPO_ROOT).as_posix()
    except ValueError:
        return str(path)


def _iter_requirement_lines(path: Path) -> list[tuple[int, str]]:
    rows: list[tuple[int, str]] = []
    for lineno, raw in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1):
        line = raw.strip()
        if not line or line.startswith("#"):
            continue
        rows.append((lineno, line))
    return rows


def check_pinned_requirements() -> Check:
    paths = [
        REPO_ROOT / "apps" / "api" / "requirements.txt",
        REPO_ROOT / "apps" / "api" / "requirements-rag.txt",
    ]
    offenders: list[str] = []
    for path in paths:
        if not path.exists():
            offenders.append(f"{_rel(path)} missing")
            continue
        for lineno, line in _iter_requirement_lines(path):
            if not REQUIREMENT_RE.match(line):
                offenders.append(f"{_rel(path)}:{lineno} {line}")
    if offenders:
        return Check("requirements_pinned", False, "; ".join(offenders))
    return Check("requirements_pinned", True, "all active API requirement lines use exact ==")


def _load_json(path: Path, errors: list[str]) -> Any:
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception as exc:  # pragma: no cover - detail is reported in CLI output
        errors.append(f"{_rel(path)} JSON load failed: {exc}")
        return None


def check_live_coach_kb() -> Check:
    root = REPO_ROOT / "data" / "kb"
    workbook = root / "live_coaching_workbook_0615.json"
    source_dir = root / "live_coaching_sources"
    errors: list[str] = []
    if not workbook.exists():
        errors.append(f"{_rel(workbook)} missing")
    if not source_dir.is_dir():
        errors.append(f"{_rel(source_dir)} missing")
    if errors:
        return Check("live_coach_kb", False, "; ".join(errors))

    payload = _load_json(workbook, errors)
    if isinstance(payload, dict):
        _validate_source_pack(path=workbook, payload=payload, errors=errors)
    source_paths = sorted(source_dir.glob("*.json"))
    if not source_paths:
        errors.append(f"{_rel(source_dir)} has no json source files")
    for path in source_paths:
        data = _load_json(path, errors)
        if isinstance(data, dict):
            _validate_source_pack(path=path, payload=data, errors=errors)
    if errors:
        return Check("live_coach_kb", False, "; ".join(errors))
    return Check(
        "live_coach_kb",
        True,
        f"workbook present and {len(source_paths)} source json files parse",
    )


def _validate_source_pack(path: Path, payload: dict[str, Any], errors: list[str]) -> None:
    source = payload.get("source")
    if not isinstance(source, dict):
        errors.append(f"{_rel(path)} missing object source")
        source = {}
    for key in ("source_id", "title"):
        if not str(source.get(key) or "").strip():
            errors.append(f"{_rel(path)} source.{key} missing")
    chunks = payload.get("chunks")
    if not isinstance(chunks, list) or not chunks:
        errors.append(f"{_rel(path)} chunks missing or empty")
        return
    for index, chunk in enumerate(chunks):
        if not isinstance(chunk, dict):
            errors.append(f"{_rel(path)} chunks[{index}] is not an object")
            continue
        if not str(chunk.get("summary") or "").strip():
            errors.append(f"{_rel(path)} chunks[{index}].summary missing")


def _parse_env_file(path: Path) -> dict[str, str]:
    values: dict[str, str] = {}
    for raw in path.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, value = line.split("=", 1)
        values[key.strip()] = value.strip().strip('"').strip("'")
    return values


def check_env_file(path: Path, *, allow_placeholder_secrets: bool) -> Check:
    if not path.exists():
        return Check("env_file", False, f"{_rel(path)} missing")
    values = _parse_env_file(path)
    errors: list[str] = []
    for key in REQUIRED_ENV_KEYS:
        value = values.get(key, "")
        if key not in values:
            errors.append(f"{key} missing")
        elif not allow_placeholder_secrets and value in PLACEHOLDER_VALUES:
            errors.append(f"{key} is placeholder or empty")
    engine_mode = values.get("ENGINE_MODE", "claude_cli")
    if engine_mode not in VALID_ENGINE_MODES:
        errors.append(f"ENGINE_MODE invalid: {engine_mode}")
    if values.get("ENVIRONMENT") in {"prod", "staging"}:
        for key in FORBIDDEN_PROD_TRUE_FLAGS:
            if values.get(key, "").lower() == "true":
                errors.append(f"{key}=true is forbidden outside dev")
    if errors:
        return Check("env_file", False, "; ".join(errors))
    return Check("env_file", True, f"{_rel(path)} has required deployment keys")


async def check_database(database_url: str, *, require_app_role: bool) -> Check:
    try:
        import asyncpg  # type: ignore
    except Exception as exc:
        return Check("database", False, f"asyncpg unavailable: {exc}")

    try:
        conn = await asyncpg.connect(database_url)
    except Exception as exc:
        return Check("database", False, f"connect failed: {exc}")
    try:
        row = await conn.fetchrow(
            """
            SELECT
              current_user AS current_user,
              (SELECT rolbypassrls FROM pg_roles WHERE rolname = current_user) AS bypassrls,
              to_regclass('app.app_user') IS NOT NULL AS has_app_user,
              to_regclass('app.auth_session') IS NOT NULL AS has_auth_session,
              to_regclass('app.admin_engine_config') IS NOT NULL AS has_engine_config,
              to_regclass('app.session_evaluation') IS NOT NULL AS has_session_evaluation,
              to_regclass('app.live_coach_events') IS NOT NULL AS has_live_coach_events,
              to_regclass('app.session_share_link') IS NOT NULL AS has_session_share_link,
              to_regclass('app.notification_event') IS NOT NULL AS has_notification_event,
              to_regclass('app.notification_delivery') IS NOT NULL AS has_notification_delivery,
              to_regclass('app.sessions') IS NOT NULL AS has_sessions,
              to_regclass('app.turns') IS NOT NULL AS has_turns,
              to_regclass('app.session_review_status') IS NOT NULL AS has_session_review_status,
              EXISTS (
                SELECT 1 FROM information_schema.columns
                WHERE table_schema = 'app'
                  AND table_name = 'persona_card'
                  AND column_name = 'triggers'
              ) AS has_persona_triggers,
              EXISTS (
                SELECT 1 FROM information_schema.columns
                WHERE table_schema = 'app'
                  AND table_name = 'turns'
                  AND column_name IN (
                    'audio_ref',
                    'silence_ms',
                    'speech_rate',
                    'barge_in',
                    'provider_events'
                  )
                GROUP BY table_schema, table_name
                HAVING count(*) = 5
              ) AS has_turn_voice_metadata_columns,
              EXISTS (
                SELECT 1 FROM information_schema.columns
                WHERE table_schema = 'app'
                  AND table_name = 'session_review_status'
                  AND column_name IN (
                    'worksheet_status',
                    'worksheet_note',
                    'worksheet_reviewed_at'
                  )
                GROUP BY table_schema, table_name
                HAVING count(*) = 3
              ) AS has_session_review_worksheet_columns,
              EXISTS (
                SELECT 1 FROM pg_indexes
                WHERE schemaname = 'app'
                  AND tablename = 'notification_delivery'
                  AND indexname = 'idx_notification_delivery_queue'
              ) AS has_notification_delivery_queue_index,
              EXISTS (
                SELECT 1 FROM pg_policies
                WHERE schemaname = 'app'
                  AND tablename = 'sessions'
                  AND policyname IN ('p_sessions_insert','p_sessions_update','p_sessions_delete')
                GROUP BY schemaname, tablename
                HAVING count(*) = 3
              ) AS has_session_write_policies,
              EXISTS (
                SELECT 1
                FROM pg_class c
                JOIN pg_namespace n ON n.oid = c.relnamespace
                WHERE n.nspname = 'app'
                  AND c.relname = 'persona_card'
                  AND pg_has_role(current_user, c.relowner, 'MEMBER')
              ) AS owns_persona_card
            """
        )
    except Exception as exc:
        return Check("database", False, f"readiness query failed: {exc}")
    finally:
        await conn.close()

    missing = [
        key
        for key in (
            "has_app_user",
            "has_auth_session",
            "has_engine_config",
            "has_session_evaluation",
            "has_live_coach_events",
            "has_session_share_link",
            "has_notification_event",
            "has_notification_delivery",
            "has_sessions",
            "has_turns",
            "has_session_review_status",
            "has_persona_triggers",
            "has_turn_voice_metadata_columns",
            "has_session_review_worksheet_columns",
            "has_notification_delivery_queue_index",
            "has_session_write_policies",
        )
        if not row[key]
    ]
    errors: list[str] = []
    if missing:
        errors.append("missing DB readiness items: " + ", ".join(missing))
    if require_app_role:
        if row["bypassrls"]:
            errors.append("runtime role has BYPASSRLS")
        if row["owns_persona_card"]:
            errors.append("runtime role owns app.persona_card; use app role DSN for API")
    if errors:
        return Check("database", False, "; ".join(errors))
    role_note = f"current_user={row['current_user']}"
    if require_app_role:
        role_note += ", non-owner app role"
    return Check("database", True, role_note)


def print_results(results: list[Check]) -> int:
    failed = False
    for result in results:
        status = "OK" if result.ok else "FAIL"
        print(f"{status} {result.name}: {result.detail}")
        failed = failed or not result.ok
    return 1 if failed else 0


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--env-file",
        default=None,
        help="Optional deployment env file to validate, for example infra/.env.",
    )
    parser.add_argument(
        "--allow-placeholder-secrets",
        action="store_true",
        help="Allow example placeholder secrets when validating .env.example.",
    )
    parser.add_argument(
        "--database-url",
        default=os.getenv("DATABASE_URL", ""),
        help="Optional runtime app-role DATABASE_URL for read-only DB readiness checks.",
    )
    parser.add_argument(
        "--skip-db",
        action="store_true",
        help="Skip DB readiness checks even when DATABASE_URL is set.",
    )
    parser.add_argument(
        "--require-app-role",
        action="store_true",
        help="Require the supplied DATABASE_URL to be a non-owner, NOBYPASSRLS runtime app role.",
    )
    return parser


async def main_async(argv: list[str]) -> int:
    args = build_parser().parse_args(argv)
    results = [
        check_pinned_requirements(),
        check_live_coach_kb(),
    ]
    if args.env_file:
        results.append(
            check_env_file(
                Path(args.env_file),
                allow_placeholder_secrets=args.allow_placeholder_secrets,
            )
        )
    if args.database_url and not args.skip_db:
        results.append(
            await check_database(
                args.database_url,
                require_app_role=args.require_app_role,
            )
        )
    return print_results(results)


def main() -> int:
    return asyncio.run(main_async(sys.argv[1:]))


if __name__ == "__main__":
    raise SystemExit(main())