#!/usr/bin/env python3
"""Host-side legacy mirror: live Supabase -> sidecar ``legacy`` schema.

The nightly snapshot (scripts/nightly_snapshot.py) runs pg_dump 17
INSIDE the app container. Host-side tooling can't (host pg_dump is v16),
so this script mirrors the same tables with plain COPY streams over
psycopg — version-agnostic — for host-side transform/harness work:

  - the ~18 ``public`` app tables (schema introspected, data COPYed)
  - ``auth.users`` scrubbed BY CONSTRUCTION: only id/email/created_at are
    ever SELECTed — credential columns never leave the source

Same destination contract as the snapshot: drop+recreate schema
``legacy`` and load in ONE transaction (a failed run leaves the previous
snapshot intact). Constraints/indexes/defaults are NOT mirrored — the
landed copy is read-only extractor/transform input, exactly like the
pg_dump path (pre-data + data, no post-data).

    python -m scripts.sync_legacy_host [--source-env .supabase-export.env]
                                       [--dest-dsn DSN]

Read-only on the source (forced); secrets never printed.
"""

from __future__ import annotations

import argparse
import json
import sys
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:  # `python scripts/sync_legacy_host.py` support
    sys.path.insert(0, str(ROOT))

from harness.common import load_env_file  # noqa: E402
from harness.extract_legacy import TABLES, conninfo_from_env  # noqa: E402

#: format_type() base types we mirror as-is; anything else (enums like
#: public.org_role, extension types) lands as text — COPY's text format
#: round-trips the labels/values byte-identically.
BUILTIN_TYPES = {
    "text", "uuid", "boolean", "integer", "bigint", "smallint", "numeric",
    "date", "timestamp with time zone", "timestamp without time zone",
    "jsonb", "json", "double precision", "real", "interval", "bytea",
    "character varying", "character",
}

AUTH_USERS_COLUMNS = ("id", "email", "created_at")  # scrub by construction
AUTH_USERS_TYPES = ("uuid", "text", "timestamp with time zone")


def _mirror_type(format_type: str) -> str:
    base = format_type.split("(")[0].strip()
    return format_type if base in BUILTIN_TYPES else "text"


def introspect(src_cur, schema: str, table: str) -> list[tuple[str, str]]:
    src_cur.execute(
        """
        SELECT a.attname, pg_catalog.format_type(a.atttypid, a.atttypmod)
        FROM pg_attribute a
        JOIN pg_class c ON c.oid = a.attrelid
        JOIN pg_namespace n ON n.oid = c.relnamespace
        WHERE n.nspname = %s AND c.relname = %s
          AND a.attnum > 0 AND NOT a.attisdropped
        ORDER BY a.attnum
        """,
        (schema, table),
    )
    return [(name, _mirror_type(ftype)) for name, ftype in src_cur.fetchall()]


def copy_table(src_conn, dst_conn, schema: str, table: str,
               columns: list[tuple[str, str]]) -> int:
    quoted = ", ".join(f'"{name}"' for name, _ in columns)
    ddl_cols = ", ".join(f'"{name}" {ftype}' for name, ftype in columns)
    with dst_conn.cursor() as dst_cur:
        dst_cur.execute(f'CREATE TABLE legacy."{table}" ({ddl_cols})')
        with (
            src_conn.cursor() as src_cur,
            src_cur.copy(
                f'COPY (SELECT {quoted} FROM "{schema}"."{table}") TO STDOUT'
            ) as out,
            dst_cur.copy(f'COPY legacy."{table}" ({quoted}) FROM STDIN') as into,
        ):
            for data in out:
                into.write(data)
        dst_cur.execute(f'SELECT count(*) FROM legacy."{table}"')
        return dst_cur.fetchone()[0]


def run_sync(source_env: str, dest_dsn: str) -> dict:
    import psycopg

    env = load_env_file(source_env)
    counts: dict[str, int] = {}
    t0 = time.monotonic()
    with (
        psycopg.connect(
            conninfo_from_env(env),
            options="-c default_transaction_read_only=on",
            application_name="with-legacy-sync",
        ) as src,
        psycopg.connect(dest_dsn, application_name="with-legacy-sync") as dst,
    ):
        src.read_only = True
        with dst.cursor() as cur:
            cur.execute("DROP SCHEMA IF EXISTS legacy CASCADE")
            cur.execute("CREATE SCHEMA legacy")
        for table in TABLES:
            counts[table] = copy_table(
                src, dst, "public", table, introspect(src.cursor(), "public", table)
            )
        # auth.users: ONLY the identity-mapping columns — the credential
        # scrub is by construction (nothing else is ever selected).
        counts["users"] = copy_table(
            src, dst, "auth", "users",
            list(zip(AUTH_USERS_COLUMNS, AUTH_USERS_TYPES, strict=True)),
        )
        dst.commit()
    return {
        "ok": True,
        "duration_s": round(time.monotonic() - t0, 2),
        "row_counts": counts,
    }


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Mirror the live legacy tables into the sidecar (host-side).")
    parser.add_argument(
        "--source-env", default=str(ROOT / ".supabase-export.env"),
        help="KEY=VALUE file with SUPABASE_DB_* (default: .supabase-export.env)")
    parser.add_argument("--dest-dsn", help="sidecar DSN (else WITH_DB_URL_HOST from .env)")
    args = parser.parse_args(argv)

    dest = args.dest_dsn
    if not dest:
        from app.config import settings

        dest = (settings.with_db_url_host or settings.with_db_url).replace(
            "postgresql+psycopg://", "postgresql://", 1
        )
    if not dest:
        print("no destination DSN (--dest-dsn or WITH_DB_URL_HOST)", file=sys.stderr)
        return 2

    result = run_sync(args.source_env, dest)
    print(json.dumps(result, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())