"""Capture live legacy API data without changing either backend.

An export covers only the explicitly listed endpoints. It is not evidence that
an unknown app has no other stores. Preserve raw rows for later schema/ID mapping.
"""

from __future__ import annotations

import hashlib
import json
import os
import re
import tempfile
import zipfile
from datetime import UTC, datetime
from pathlib import Path
from urllib.parse import urlsplit
from uuid import UUID, uuid4

import httpx


def _digest(value: object) -> str:
    return hashlib.sha256(
        json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode()
    ).hexdigest()


def validate_export_plan(plan: dict, project_id: str) -> dict:
    if not isinstance(plan, dict) or plan.get("version") != 1:
        raise ValueError("Expected a version 1 export plan")
    if str(UUID(plan["project_id"])) != str(UUID(project_id)):
        raise ValueError("Export plan belongs to a different project")
    origin = urlsplit(plan.get("origin", ""))
    if (
        origin.scheme != "https"
        or not origin.netloc
        or origin.username
        or origin.password
        or origin.path not in {"", "/"}
        or origin.query
        or origin.fragment
    ):
        raise ValueError("Export origin must be an HTTPS origin without credentials")
    collections = plan.get("collections")
    if not isinstance(collections, list) or not collections:
        raise ValueError("List the live collection endpoints to export")
    names = set()
    entities = set()
    for item in collections:
        if not isinstance(item, dict) or not re.fullmatch(r"[a-z][a-z0-9_]*", item.get("key", "")):
            raise ValueError("Invalid collection key")
        if item["key"] in names:
            raise ValueError("Duplicate collection key")
        names.add(item["key"])
        endpoint = item.get("path", "")
        if not re.fullmatch(r"/api/[A-Za-z0-9_/-]+", endpoint) or ".." in endpoint:
            raise ValueError("Export endpoints must be literal /api/ paths on the given origin")
        method = str(item.get("method") or "GET").upper()
        pagination = item.get("pagination")
        if method == "GET" and pagination == "page":
            pass
        elif method == "POST" and pagination == "offset":
            pass
        else:
            raise ValueError("Export pagination must be GET page or POST offset")
        if item["key"] == "users":
            if item.get("entity"):
                raise ValueError("Users are Cloud auth identities, not a record entity")
        else:
            entity = item.get("entity")
            if not isinstance(entity, str) or not re.fullmatch(r"[a-z][a-z0-9_]*", entity):
                raise ValueError("Record collections require a Cloud entity mapping")
            if entity in entities:
                raise ValueError("Duplicate Cloud entity mapping")
            entities.add(entity)
        params = item.get("params", {})
        if not isinstance(params, dict) or set(params) & {"page", "limit", "offset"}:
            raise ValueError("Collection parameters cannot override pagination")
        if any(re.search(r"token|password|secret|authorization|api.?key", k, re.I) for k in params):
            raise ValueError("Supply export authentication through an environment variable")
        body = item.get("body", {})
        if body and (method != "POST" or not isinstance(body, dict)):
            raise ValueError("Only POST collections may include a JSON body")
        if isinstance(body, dict) and set(body) & {"page", "limit", "offset"}:
            raise ValueError("Collection body cannot override pagination")
    return plan


_ENTITY_KEY = re.compile(r"^[a-z][a-z0-9_]*$")
_RECORD_IMPORT_ORDER = (
    "universities",
    "programs",
    "pipeline_stages",
    "students",
    "applications",
    "documents",
    "tasks",
    "messages",
    "appointments",
    "notifications",
    "activities",
    "communication_logs",
    "application_status_history",
    "audit_logs",
)


def preview_origin_for_project(project) -> str:
    """HTTPS origin of the live preview. Prefer the stored hostname."""
    host = str(getattr(project, "preview_hostname", None) or "").strip()
    if host:
        origin = host if host.startswith("https://") else f"https://{host}"
        parsed = urlsplit(origin)
        if parsed.scheme != "https" or not parsed.netloc or parsed.username or parsed.password:
            raise ValueError("Preview hostname is not a valid HTTPS origin")
        return f"https://{parsed.netloc}"
    ref = str(getattr(project, "project_ref", None) or "").strip()
    if not ref:
        raise ValueError("Project has no preview hostname or project_ref")
    from src.config import get_settings
    from src.domain.hosting.hostnames import canonical_preview_url

    apex = get_settings().hosting_public_apex or "sites.p2hs.com"
    parsed = urlsplit(canonical_preview_url(ref, apex))
    return f"https://{parsed.netloc}"


def discover_entity_keys(repo_path: str) -> list[str]:
    keys: set[str] = set()
    root = Path(repo_path)
    for relative in ("p2hs/entities.json", "p2hs/provision.json"):
        path = root / relative
        if not path.is_file():
            continue
        try:
            data = json.loads(path.read_text(encoding="utf-8-sig"))
        except (OSError, json.JSONDecodeError):
            continue
        for entity in data.get("entities") or []:
            if not isinstance(entity, dict):
                continue
            key = entity.get("slug") or entity.get("key")
            if isinstance(key, str) and _ENTITY_KEY.fullmatch(key):
                keys.add(key)
    records = root / "server/api/records"
    if records.is_dir():
        for child in records.iterdir():
            if child.is_dir() and _ENTITY_KEY.fullmatch(child.name):
                keys.add(child.name)
    ordered = [key for key in _RECORD_IMPORT_ORDER if key in keys]
    ordered.extend(sorted(keys.difference(ordered)))
    return ordered


def build_live_export_plan(project_id: str, origin: str, repo_path: str) -> dict:
    collections = [
        {
            "key": "users",
            "path": "/api/admin/users",
            "pagination": "page",
            # Preview may already have the Cloud adapter without Worker credentials.
            "skip_unconfigured": True,
        },
    ]
    for entity in discover_entity_keys(repo_path):
        collections.append(
            {
                "key": entity,
                "path": f"/api/records/{entity}/query",
                "method": "POST",
                "pagination": "offset",
                "entity": entity,
                "optional": True,
            }
        )
    return validate_export_plan(
        {
            "version": 1,
            "project_id": str(UUID(project_id)),
            "origin": origin,
            "collections": collections,
        },
        project_id,
    )


def default_export_dir(project_id: str) -> str:
    path = (
        Path(tempfile.gettempdir()) / "p2hs-migration-exports" / str(UUID(project_id))
    ).resolve()
    return str(path)


async def export_project_live_data(
    *,
    project_id: str,
    origin: str,
    repo_path: str,
    client: httpx.AsyncClient,
    export_dir: str | None = None,
) -> tuple[Path, dict]:
    """Build the live plan from the checkout, capture every listed collection, persist it."""
    plan = build_live_export_plan(project_id, origin, repo_path)
    snapshot = await export_live_data(plan, project_id, client=client)
    path = save_export(snapshot, export_dir or default_export_dir(project_id))
    return path, snapshot


def _page_rows(payload: dict, spec: dict) -> tuple[list, int, int, int]:
    if spec.get("pagination") == "offset":
        rows = payload.get("records", payload.get("data"))
        if isinstance(rows, dict):
            rows = rows.get("records", rows.get("data"))
        total = payload.get("total", (payload.get("pagination") or {}).get("total"))
        if not isinstance(rows, list) or type(total) is not int:
            raise ValueError(f"Export {spec['key']} has an unsupported response shape")
        return rows, total, int(spec.get("_offset", 0)), 100
    if payload.get("success") is not True:
        raise ValueError(f"Export {spec['key']} did not report success")
    data, pagination = payload.get("data"), payload.get("pagination")
    if not isinstance(data, list) or not isinstance(pagination, dict):
        raise ValueError(f"Export {spec['key']} has an unsupported response shape")
    count, current, limit = (pagination.get(k) for k in ("total", "page", "limit"))
    if any(type(x) is not int for x in (count, current, limit)):
        raise ValueError(f"Export {spec['key']} has invalid pagination")
    return data, count, current, limit


def _empty_collection(spec: dict) -> dict:
    return {
        "path": spec["path"],
        "count": 0,
        "rows": [],
        "pages": [],
        "kind": "users" if spec["key"] == "users" else "records",
        **({"entity": spec["entity"]} if spec.get("entity") else {}),
    }


async def _read_collection(client: httpx.AsyncClient, origin: str, spec: dict) -> dict:
    rows, pages, seen = [], [], set()
    total = None
    page = 1
    while True:
        if spec.get("method", "GET").upper() == "POST":
            offset = (page - 1) * 100
            spec["_offset"] = offset
            response = await client.post(
                origin.rstrip("/") + spec["path"],
                json={**spec.get("body", {}), "limit": 100, "offset": offset},
            )
        else:
            response = await client.get(
                origin.rstrip("/") + spec["path"],
                params={**spec.get("params", {}), "page": page, "limit": 100},
            )
        if response.status_code != 200:
            if (
                spec.get("optional")
                and response.status_code in {404, 405, 503}
                and page == 1
                and not rows
            ):
                return _empty_collection(spec)
            if (
                spec.get("skip_unconfigured")
                and response.status_code == 503
                and page == 1
                and not rows
            ):
                return _empty_collection(spec)
            raise ValueError(f"Export {spec['key']} failed: HTTP {response.status_code}")
        try:
            payload = response.json()
        except ValueError:
            raise ValueError(f"Export {spec['key']} returned invalid JSON") from None
        if not isinstance(payload, dict):
            raise ValueError(f"Export {spec['key']} returned invalid JSON")
        data, count, current, limit = _page_rows(payload, spec)
        if (
            count < 0
            or limit < 1
            or limit > 100
            or (spec.get("pagination") == "page" and current != page)
        ):
            raise ValueError(f"Export {spec['key']} has invalid pagination")
        if total is not None and count != total:
            raise ValueError(
                f"Export {spec['key']} changed during pagination; freeze writes and retry"
            )
        total = count
        if len(data) > limit or (not data and len(rows) < total):
            raise ValueError(f"Export {spec['key']} was truncated")
        for row in data:
            if not isinstance(row, dict) or not isinstance(row.get("id"), str) or not row["id"]:
                raise ValueError(f"Export {spec['key']} contains a row without a stable ID")
            if row["id"] in seen:
                raise ValueError(f"Export {spec['key']} repeated an ID; pagination is unstable")
            seen.add(row["id"])
            rows.append(row)
        pages.append(payload)
        if len(rows) == total:
            break
        if len(rows) > total or page >= 100000:
            raise ValueError(f"Export {spec['key']} has inconsistent totals")
        page += 1
    exported = {
        "path": spec["path"],
        "count": total,
        "rows": rows,
        "pages": pages,
        "kind": "users" if spec["key"] == "users" else "records",
    }
    if spec.get("entity"):
        exported["entity"] = spec["entity"]
    return exported


async def export_live_data(plan: dict, project_id: str, *, client: httpx.AsyncClient) -> dict:
    plan = validate_export_plan(plan, project_id)
    collections = {}
    for spec in plan["collections"]:
        collections[spec["key"]] = await _read_collection(client, plan["origin"], spec)
    # A second pass catches same-count edits and ephemeral instances serving different data.
    # A write freeze is still required for a real cutover across multiple stores.
    for spec in plan["collections"]:
        verify = await _read_collection(client, plan["origin"], spec)
        if _digest(sorted(verify["rows"], key=lambda row: row["id"])) != _digest(
            sorted(collections[spec["key"]]["rows"], key=lambda row: row["id"])
        ):
            raise ValueError(f"Export {spec['key']} changed between reads; freeze writes and retry")
    result = {
        "version": 1,
        "project_id": str(UUID(project_id)),
        "origin": plan["origin"],
        "exported_at": datetime.now(UTC).isoformat(),
        "coverage": "listed_endpoints_only",
        "collections": collections,
    }
    result["sha256"] = _digest(result)
    return result


def save_export(snapshot: dict, directory: str) -> Path:
    """Keep raw data out of Git, create exclusively, never overwrite an earlier export."""
    root = Path(directory).resolve()
    if any((parent / ".git").exists() for parent in [root, *root.parents]):
        raise ValueError("Raw migration exports must be stored outside Git repositories")
    root.mkdir(parents=True, exist_ok=True, mode=0o700)
    path = root / f"{UUID(snapshot['project_id'])}-{uuid4().hex}.json"
    encoded = json.dumps(snapshot, ensure_ascii=False, indent=2).encode("utf-8")
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    with os.fdopen(fd, "wb") as stream:
        stream.write(encoded)
        stream.flush()
        os.fsync(stream.fileno())
    if json.loads(path.read_text(encoding="utf-8")) != snapshot:
        raise ValueError("Export read-back failed")
    return path


def _contains_secret(value) -> bool:
    if isinstance(value, dict):
        return any(
            re.search(r"password|secret|token|hash", key, re.I) or _contains_secret(child)
            for key, child in value.items()
        )
    return isinstance(value, list) and any(_contains_secret(child) for child in value)


def _load_export_snapshot(path: str, project_id: str) -> dict:
    snapshot = json.loads(Path(path).read_text(encoding="utf-8-sig"))
    checksum = snapshot.pop("sha256", None)
    if checksum != _digest(snapshot) or snapshot.get("version") != 1:
        raise ValueError("Invalid or modified live data export")
    if str(UUID(snapshot["project_id"])) != str(UUID(project_id)):
        raise ValueError("Data export belongs to a different project")
    collections = snapshot.get("collections")
    if not isinstance(collections, dict) or not collections:
        raise ValueError("Incomplete data export")
    return snapshot


def load_record_export(path: str, project_id: str) -> dict[str, list[dict]]:
    """Rows keyed by Cloud entity. Empty when the snapshot is users-only."""
    snapshot = _load_export_snapshot(path, project_id)
    records: dict[str, list[dict]] = {}
    for key, section in snapshot["collections"].items():
        if not isinstance(section, dict):
            raise ValueError("Incomplete data export")
        kind = section.get("kind") or ("users" if key == "users" else None)
        if kind == "users":
            continue
        entity = section.get("entity")
        if not isinstance(entity, str) or not re.fullmatch(r"[a-z][a-z0-9_]*", entity):
            raise ValueError(
                "Non-user collections require Cloud entity and route mappings before cutover"
            )
        rows = section.get("rows")
        if not isinstance(rows, list) or section.get("count") != len(rows):
            raise ValueError("Incomplete record export")
        seen = set()
        for row in rows:
            if not isinstance(row, dict) or not row.get("id") or row["id"] in seen:
                raise ValueError("Record export needs unique stable IDs")
            if _contains_secret(row):
                raise ValueError("Credentials need a compatible auth import; cannot copy into metadata")
            seen.add(row["id"])
        if entity in records:
            raise ValueError("Duplicate Cloud entity mapping")
        records[entity] = rows
    return records


def load_user_export(path: str, project_id: str):
    """Never silently discard collections that need project-specific mappings."""
    from src.domain.codegen.legacy_user_store import DiscoveredUser, normalize_role

    snapshot = _load_export_snapshot(path, project_id)
    load_record_export(path, project_id)
    collections = snapshot["collections"]
    if "users" not in collections:
        return []
    section = collections["users"]
    rows = section.get("rows")
    if not isinstance(rows, list) or section.get("count") != len(rows):
        raise ValueError("Incomplete user export")
    seen, ids, users = set(), set(), []
    for row in rows:
        if not isinstance(row, dict):
            raise ValueError("Invalid user export row")
        email = str(row.get("email", "")).strip().lower()
        if not email or "@" not in email or email in seen or not row.get("id") or row["id"] in ids:
            raise ValueError("User export needs unique emails and stable IDs")
        role = normalize_role(str(row.get("role") or ""))
        if type(row.get("is_active")) is not bool:
            raise ValueError("User export must preserve each account's active status")
        if _contains_secret(row):
            raise ValueError("Credentials need a compatible auth import; cannot copy into metadata")
        seen.add(email)
        ids.add(row["id"])
        users.append(
            DiscoveredUser(email, role, str(row.get("name") or ""), "live_export", row)
        )
    return users


def archive_legacy_checkout(repo_path: str) -> Path:
    """Snapshot on-disk source/data before cutover; never reset or publish the archive."""
    source = Path(repo_path).resolve()
    if not source.is_dir():
        raise ValueError("Migration checkout is missing")
    root = source.parent / "project_meta" / source.name / "migration-backups" / uuid4().hex
    if any((parent / ".git").exists() for parent in [root, *root.parents]):
        raise ValueError("Migration backup must be outside Git repositories")
    root.mkdir(parents=True, mode=0o700)
    path = root / "legacy-checkout.zip"
    with os.fdopen(os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600), "wb") as output:
        with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as archive:
            for directory, dirs, files in os.walk(source, followlinks=False):
                dirs[:] = sorted(
                    d for d in dirs if d not in {".git", "node_modules", "__pycache__"}
                )
                if any((Path(directory) / d).is_symlink() for d in dirs):
                    raise ValueError("Inspect source symlinks before migrating")
                for name in sorted(files):
                    entry = Path(directory) / name
                    if entry.is_symlink():
                        raise ValueError("Inspect source symlinks before migrating")
                    archive.write(entry, entry.relative_to(source).as_posix())
        output.flush()
        os.fsync(output.fileno())
    with zipfile.ZipFile(path) as archive:
        if archive.testzip() is not None:
            raise ValueError("Legacy backup verification failed")
    return path