#!/usr/bin/env python3
"""Manifest-backed patch replay for the serviceified Test D3 deployer."""

from __future__ import annotations

import json
import os
import re
import shutil
import sys
from pathlib import Path
from typing import Any, Callable, Mapping

_REPO_ROOT = Path(__file__).resolve().parent.parent
if str(_REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(_REPO_ROOT))

from scripts.serviceification_d3_engine import DeployError, D3Context


def runtime_patch_modules() -> dict[str, Any]:
    runtime_dir = str(_REPO_ROOT / "deployments" / "offline" / "runtime")
    if runtime_dir not in sys.path:
        sys.path.insert(0, runtime_dir)
    from common import (  # type: ignore[import-not-found]
        load_patch_bundle_definition,
        require_patch_bundle_environment,
        resolve_patch_sql_files,
        validate_patch_sql_idempotency,
    )
    from deployment_patch_replay import (  # type: ignore[import-not-found]
        execute_sql_file,
        patch_bundle_tree_digest,
    )

    return {
        "load_patch_bundle_definition": load_patch_bundle_definition,
        "require_patch_bundle_environment": require_patch_bundle_environment,
        "resolve_patch_sql_files": resolve_patch_sql_files,
        "validate_patch_sql_idempotency": validate_patch_sql_idempotency,
        "execute_sql_file": execute_sql_file,
        "patch_bundle_tree_digest": patch_bundle_tree_digest,
    }


OPERATOR_PATCH_BUNDLE_ROOT = Path("docs") / "migration" / "patch-bundles"
SEAL_TREE_DIGEST_NAME = "tree-digest"
_TREE_DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
_COMMIT_RE = re.compile(r"^[0-9a-f]{40}$")
_ENV_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.-]*$")
# Staging apply-patch runs before Catalog apply. D3 matches that mutation
# window after stopped-upgrade quiescence and immediately before leftover
# preconvergence; catalog-ready is a same-process fallback. Catalog
# operations are immutable once begun.
PATCH_SQL_READY_PHASES = frozenset(
    {
        "quiesced",
        "runtime-apply-in-progress",
        "runtime-candidate-applied",
        "catalog-ready",
    }
)
PATCH_SQL_BLOCKED_PHASES = frozenset(
    {
        "catalog-in-progress",
        "catalog-applied",
        "governed-reference-data-in-progress",
        "governed-reference-data-applied",
        "serving-inputs-ready",
        "worker-activation-in-progress",
        "worker-owned",
        "serving-start-in-progress",
        "serving",
        "worker-image-replace-in-progress",
        "worker-runtime-config-in-progress",
        "frontend-image-replace-in-progress",
        "service-image-replace-in-progress",
    }
)


def require_patch_bundle_declares_no_assertions(definition: Mapping[str, Any], patch_dir: Path) -> None:
    """Refuse an assertion-bearing bundle on the D3 path, which cannot enforce assertions.

    The staging apply-patch stage evaluates a manifest's `assertions` and fails the deploy when
    the expected state is not met. This replay path executes `sql_files` only, so accepting such
    a bundle here would silently drop its verification and record the run as passed. Failing is
    the fail-closed choice; wiring assertion evaluation into D3 is a separate change.
    """
    manifest_path = Path(str(definition.get("manifest_path", patch_dir)))
    try:
        raw = json.loads(manifest_path.read_text())
    except (OSError, ValueError):
        return
    # Presence, not truthiness. parse_patch_assertions rejects a present-but-malformed block
    # such as `assertions: []` or `assertions: 0`; accepting those here would replay a manifest
    # that the enforcing path considers invalid.
    if isinstance(raw, dict) and "assertions" in raw:
        raise DeployError(
            "D3 patch replay does not evaluate manifest assertions; "
            f"refusing assertion-bearing bundle: {patch_dir}"
        )


def execute_patch_sql_file(envv: Mapping[str, str], sql_file: Path) -> None:
    """Run one patch SQL file without putting query output in the exception."""
    try:
        runtime_patch_modules()["execute_sql_file"](dict(envv), sql_file)
    except Exception as exc:
        # The shared patch runner includes stdout before stderr. Never forward
        # that payload because an arbitrary operator bundle may select customer
        # data. Preserve only the database error number and input line, which
        # are sufficient to distinguish syntax from an intentional fail-closed
        # assertion and map the failure back to reviewed SQL.
        match = re.search(
            r"\bERROR\s+(\d+)(?:\s+\([^\r\n)]*\))?\s+at\s+line\s+(\d+)\s*:",
            str(exc),
            re.IGNORECASE,
        )
        already_sanitized = re.search(
            r"\bdatabase error \d+ at input line \d+\b",
            str(exc),
            re.IGNORECASE,
        )
        if match is not None:
            detail = (
                f" database error {match.group(1)} at input line {match.group(2)}"
            )
        elif already_sanitized is not None:
            detail = " " + already_sanitized.group(0).lower()
        else:
            detail = ""
        raise DeployError(f"patch sql failed: {sql_file.name}{detail}") from None


def active_run_phase(
    context: D3Context,
    *,
    read_state: Callable[[Path], Mapping[str, Any]],
) -> str:
    state_file = getattr(context, "state_file", None)
    if state_file is None:
        return ""
    path = Path(state_file)
    if not (path.exists() or path.is_symlink()):
        return ""
    state = read_state(path)
    if state.get("runId") != getattr(context, "run_id", None):
        return ""
    return str(state.get("phase", ""))


def require_active_run_can_replay_patch(
    context: D3Context,
    requested_patch_dir: str,
    *,
    read_state: Callable[[Path], Mapping[str, Any]],
) -> dict[str, Any] | None:
    """Admit pre-Catalog replay or recover proof of an already completed replay.

    A post-Catalog retry must never execute SQL again. Its existing run-owned
    receipt and sealed tree must agree exactly; callers still perform normal
    source/environment/promoted-digest preflight before any convergence.
    """
    if not requested_patch_dir.strip():
        return None
    phase = active_run_phase(context, read_state=read_state)
    if phase in PATCH_SQL_BLOCKED_PHASES:
        return require_completed_patch_replay(context, requested_patch_dir)
    return None


def require_completed_patch_replay(
    context: D3Context, requested_patch_dir: str,
) -> dict[str, Any]:
    """Read, never create or replay, the exact run's durable successful receipt."""
    require_operator_patch_request(requested_patch_dir)
    # Replay persists the trimmed input, including accepted trailing slashes.
    # Validate the path without rewriting existing receipt identity on retry.
    requested = requested_patch_dir.strip()
    sealed = patch_seal_dir(context)
    digest = require_sealed_patch_bundle(sealed)
    modules = runtime_patch_modules()
    definition = modules["load_patch_bundle_definition"](sealed)
    require_patch_bundle_declares_no_assertions(definition, sealed)
    sql_files = modules["resolve_patch_sql_files"](sealed, definition)
    names = [path.name for path in sql_files]
    summary_path = context.run_root / "patch-replay" / "patch-summary.json"
    if (
        summary_path.parent.is_symlink()
        or summary_path.is_symlink()
        or not summary_path.is_file()
    ):
        raise DeployError(
            "requested --patch-dir requires a pre-Catalog D3 run or an existing "
            "successful same-run patch replay receipt"
        )
    try:
        summary = json.loads(summary_path.read_text(encoding="utf-8"))
    except (OSError, ValueError) as exc:
        raise DeployError("completed D3 patch replay receipt cannot be read") from exc
    expected = {
        "schema": "nexuscrm.serviceification-d3-patch-replay/v1",
        "patchDir": str(sealed),
        "requestedPatchDir": requested,
        "manifestPath": str(definition["manifest_path"]),
        "treeDigest": digest,
        "sqlFiles": names,
        "status": "passed",
        "results": [{"file": name, "status": "passed"} for name in names],
    }
    if not isinstance(summary, dict) or any(
        summary.get(key) != value for key, value in expected.items()
    ):
        raise DeployError(
            "completed D3 patch replay receipt disagrees with the exact run, "
            "requested bundle, sealed digest, or complete SQL results"
        )
    return summary


def patch_tree_digest(root: Path) -> str:
    return str(runtime_patch_modules()["patch_bundle_tree_digest"](root))


def patch_seal_digest_path(sealed: Path) -> Path:
    return sealed.parent / SEAL_TREE_DIGEST_NAME


def write_seal_tree_digest(sealed: Path, digest: str) -> None:
    if not _TREE_DIGEST_RE.fullmatch(digest):
        raise DeployError("sealed D3 patch digest is malformed")
    path = patch_seal_digest_path(sealed)
    pending = path.with_name(path.name + ".pending")
    if pending.exists() or pending.is_symlink():
        pending.unlink()
    pending.write_text(digest + "\n")
    os.chmod(pending, 0o600)
    os.replace(pending, path)


def read_seal_tree_digest(sealed: Path) -> str:
    path = patch_seal_digest_path(sealed)
    if path.is_symlink() or not path.is_file():
        raise DeployError(f"sealed D3 patch bundle has no tree digest: {path}")
    value = path.read_text().strip()
    if not _TREE_DIGEST_RE.fullmatch(value):
        raise DeployError("sealed D3 patch digest is malformed")
    return value


def require_patch_bundle_does_not_override_refs(definition: Mapping[str, Any]) -> None:
    refs = definition.get("refs")
    if isinstance(refs, dict) and any(str(value or "").strip() for value in refs.values()):
        raise DeployError(
            "D3 patch bundles may not override deploy refs; the signed BOM remains authority"
        )


def require_patch_bundle_matches_d3_context(
    context: D3Context,
    definition: Mapping[str, Any],
) -> None:
    """Bind a constrained patch to one environment and one signed-BOM L2 input."""
    manifest_path = Path(str(definition.get("manifest_path", "")))
    try:
        manifest = json.loads(manifest_path.read_text())
    except (OSError, ValueError) as exc:
        raise DeployError("D3 patch context manifest cannot be read") from exc
    if not isinstance(manifest, dict):
        raise DeployError("D3 patch context manifest is malformed")
    constraint = manifest.get("d3_context")
    if constraint is None:
        return
    if not isinstance(constraint, Mapping):
        raise DeployError("D3 patch context constraint is malformed")
    if set(constraint) != {"env_name", "l2_metadata"}:
        raise DeployError(
            "D3 patch context must contain exactly env_name and l2_metadata"
        )
    expected_env_name = str(constraint.get("env_name", "")).strip()
    if not _ENV_NAME_RE.fullmatch(expected_env_name):
        raise DeployError("D3 patch context environment constraint is malformed")
    actual_env_name = str(context.envv.get("ENV_NAME", "")).strip()
    if actual_env_name != expected_env_name:
        raise DeployError(
            "D3 patch bundle environment constraint disagrees with the selected slot"
        )
    expected_l2 = constraint.get("l2_metadata")
    actual_l2 = context.bom.get("l2_metadata")
    if not isinstance(expected_l2, Mapping) or not isinstance(actual_l2, Mapping):
        raise DeployError("D3 patch bundle requires exact signed-BOM L2 metadata")
    fields = ("repository", "commit", "path", "tree_sha256")
    if set(expected_l2) != set(fields):
        raise DeployError(
            "D3 patch context L2 constraint must contain exactly repository, "
            "commit, path, and tree_sha256"
        )
    if (
        not str(expected_l2.get("repository", "")).strip()
        or not str(expected_l2.get("path", "")).strip()
        or not _COMMIT_RE.fullmatch(str(expected_l2.get("commit", "")).strip())
        or not _TREE_DIGEST_RE.fullmatch(
            str(expected_l2.get("tree_sha256", "")).strip()
        )
    ):
        raise DeployError("D3 patch context L2 constraint is malformed")
    if any(
        str(actual_l2.get(field, "")).strip()
        != str(expected_l2.get(field, "")).strip()
        for field in fields
    ):
        raise DeployError(
            "D3 patch bundle L2 constraint disagrees with the selected signed BOM"
        )


def require_operator_patch_request(raw: str) -> Path:
    requested = raw.strip()
    if not requested:
        raise DeployError("--patch-dir requires a directory")
    if requested == "__diagnose_search__":
        raise DeployError("D3 does not implement the monolith diagnose-search path")
    if any(character in requested for character in ("\n", "\r", "\0")):
        raise DeployError("D3 --patch-dir must be a single path")
    relative = Path(requested)
    if relative.is_absolute() or relative.anchor:
        raise DeployError(
            "D3 --patch-dir must be repo-relative under docs/migration/patch-bundles"
        )
    if any(part in {"", ".", ".."} for part in relative.parts):
        raise DeployError("D3 --patch-dir must not contain . or ..")
    try:
        nested = relative.relative_to(OPERATOR_PATCH_BUNDLE_ROOT)
    except ValueError as exc:
        raise DeployError(
            "D3 --patch-dir must stay under docs/migration/patch-bundles"
        ) from exc
    if not nested.parts:
        raise DeployError("D3 --patch-dir must name one patch bundle directory")
    return relative


def resolve_operator_patch_dir(raw: str, *, repo_root: Path | None = None) -> Path:
    relative = require_operator_patch_request(raw)
    root = (repo_root or _REPO_ROOT).resolve()
    candidate = root / relative
    if candidate.is_symlink():
        raise DeployError("D3 --patch-dir refuses a symlink")
    if not candidate.is_dir():
        raise DeployError(f"patch dir not found: {relative}")
    path = candidate.resolve()
    bundle_root = (root / OPERATOR_PATCH_BUNDLE_ROOT).resolve()
    try:
        nested = path.relative_to(bundle_root)
        path.relative_to(root)
    except ValueError as exc:
        raise DeployError(
            "D3 --patch-dir escaped docs/migration/patch-bundles"
        ) from exc
    if not nested.parts:
        raise DeployError("D3 --patch-dir must name one patch bundle directory")
    modules = runtime_patch_modules()
    definition = modules["load_patch_bundle_definition"](path)
    require_patch_bundle_does_not_override_refs(definition)
    require_patch_bundle_declares_no_assertions(definition, path)
    sql_files = modules["resolve_patch_sql_files"](path, definition)
    if not sql_files:
        raise DeployError(f"patch bundle has no sql files: {path}")
    for sql_file in sql_files:
        modules["validate_patch_sql_idempotency"](sql_file)
    return path


def require_patch_bundle_environment(
    raw: str,
    *,
    environment: str,
    repo_root: Path | None = None,
) -> Path:
    """Reject an explicitly scoped bundle outside its reviewed environment."""
    path = resolve_operator_patch_dir(raw, repo_root=repo_root)
    definition = runtime_patch_modules()["load_patch_bundle_definition"](path)
    require_patch_bundle_definition_environment(
        definition,
        environment=environment,
        patch_dir=path,
    )
    return path


def require_patch_bundle_definition_environment(
    definition: Mapping[str, Any],
    *,
    environment: str | None,
    patch_dir: Path,
) -> None:
    """Enforce an environment scope from the definition being authorized."""
    runtime_patch_modules()["require_patch_bundle_environment"](
        dict(definition),
        environment=environment,
        patch_dir=patch_dir,
    )


def copy_patch_tree(source: Path, dest: Path) -> None:
    if dest.exists() or dest.is_symlink():
        raise DeployError(f"sealed D3 patch bundle already exists: {dest}")
    dest.mkdir(parents=True, mode=0o700)
    for path in sorted(source.rglob("*")):
        if path.is_symlink():
            raise DeployError(f"patch bundle refuses a symlink: {path}")
        relative = path.relative_to(source)
        target = dest / relative
        if path.is_dir():
            target.mkdir(parents=True, exist_ok=True)
            os.chmod(target, 0o700)
            continue
        if not path.is_file():
            raise DeployError(f"patch bundle contains an unsupported path: {path}")
        target.parent.mkdir(parents=True, exist_ok=True)
        shutil.copy2(path, target)
        os.chmod(target, 0o600)


def seal_patch_bundle(source: Path, dest: Path) -> Path:
    source_digest = patch_tree_digest(source)
    if dest.exists() or dest.is_symlink():
        if dest.is_symlink() or not dest.is_dir():
            raise DeployError(f"sealed D3 patch bundle is not a directory: {dest}")
        sealed_digest = patch_tree_digest(dest)
        if sealed_digest != source_digest:
            raise DeployError(
                "sealed D3 patch bundle disagrees with --patch-dir; "
                "resume the same run or start a new run id"
            )
        recorded = read_seal_tree_digest(dest)
        if recorded != sealed_digest:
            raise DeployError("sealed D3 patch digest disagrees with the sealed tree")
        return dest
    copy_patch_tree(source, dest)
    sealed_digest = patch_tree_digest(dest)
    if sealed_digest != source_digest:
        raise DeployError("sealed D3 patch bundle changed while it was being copied")
    write_seal_tree_digest(dest, sealed_digest)
    return dest


def apply_sealed_patch_bundle(
    context: D3Context,
    sealed_dir: Path,
    *,
    requested_patch_dir: str = "",
) -> dict[str, Any]:
    modules = runtime_patch_modules()
    tree_digest = require_sealed_patch_bundle(sealed_dir)
    definition = modules["load_patch_bundle_definition"](sealed_dir)
    require_patch_bundle_declares_no_assertions(definition, sealed_dir)
    sql_files = modules["resolve_patch_sql_files"](sealed_dir, definition)
    artifact_dir = context.run_root / "patch-replay"
    artifact_dir.mkdir(parents=True, mode=0o700, exist_ok=True)
    results: list[dict[str, str]] = []
    for sql_file in sql_files:
        execute_patch_sql_file(dict(context.envv), sql_file)
        results.append({"file": sql_file.name, "status": "passed"})
    summary = {
        "schema": "nexuscrm.serviceification-d3-patch-replay/v1",
        "patchDir": str(sealed_dir),
        "requestedPatchDir": requested_patch_dir,
        "manifestPath": str(definition["manifest_path"]),
        "title": str(definition["title"]),
        "createdAt": str(definition["created_at"]),
        "treeDigest": tree_digest,
        "sqlFiles": [item["file"] for item in results],
        "status": "passed",
        "results": results,
    }
    summary_path = artifact_dir / "patch-summary.json"
    pending_summary = artifact_dir / ".patch-summary.json.pending"
    if pending_summary.exists() or pending_summary.is_symlink():
        pending_summary.unlink()
    with pending_summary.open("x", encoding="utf-8") as stream:
        os.chmod(pending_summary, 0o600)
        stream.write(json.dumps(summary, indent=2, sort_keys=True) + "\n")
        stream.flush()
        os.fsync(stream.fileno())
    os.replace(pending_summary, summary_path)
    descriptor = os.open(
        artifact_dir, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0)
    )
    try:
        os.fsync(descriptor)
    finally:
        os.close(descriptor)
    return summary


def patch_seal_dir(context: D3Context) -> Path:
    env_root = Path(str(context.env_root)).expanduser().resolve()
    run_id = str(context.run_id)
    if not run_id:
        raise DeployError("D3 patch sealing requires a run id")
    return (
        env_root / ".runtime" / "serviceification-d3" / "patch-seals"
        / run_id / "patch-bundle"
    )


def require_sealed_patch_bundle(sealed: Path) -> str:
    if sealed.is_symlink() or not sealed.is_dir():
        raise DeployError(f"sealed D3 patch bundle is not a directory: {sealed}")
    recorded = read_seal_tree_digest(sealed)
    actual = patch_tree_digest(sealed)
    if actual != recorded:
        raise DeployError("sealed D3 patch bundle changed after it was sealed")
    modules = runtime_patch_modules()
    definition = modules["load_patch_bundle_definition"](sealed)
    require_patch_bundle_does_not_override_refs(definition)
    sql_files = modules["resolve_patch_sql_files"](sealed, definition)
    if not sql_files:
        raise DeployError(f"patch bundle has no sql files: {sealed}")
    for sql_file in sql_files:
        if sql_file.is_symlink() or not sql_file.is_file():
            raise DeployError(f"patch bundle refuses a symlink: {sql_file}")
        modules["validate_patch_sql_idempotency"](sql_file)
    return actual


def preflight_and_seal_patch(
    context: D3Context,
    requested_patch_dir: str,
    *,
    source_root: Path | None = None,
    expected_tree_digest: str = "",
    environment: str | None = None,
) -> Path | None:
    """Resolve and seal the operator bundle before D3 convergence mutates."""
    requested = requested_patch_dir.strip()
    expected = expected_tree_digest.strip()
    if expected and not _TREE_DIGEST_RE.fullmatch(expected):
        raise DeployError("expected D3 patch tree digest is malformed")
    sealed = patch_seal_dir(context)
    if not requested:
        if expected:
            raise DeployError("D3 patch digest was supplied without a patch bundle")
        if not (sealed.exists() or sealed.is_symlink()):
            return None
        actual = require_sealed_patch_bundle(sealed)
        definition = runtime_patch_modules()["load_patch_bundle_definition"](sealed)
        require_patch_bundle_definition_environment(
            definition,
            environment=environment,
            patch_dir=sealed,
        )
        require_patch_bundle_matches_d3_context(context, definition)
        if expected and actual != expected:
            raise DeployError("sealed D3 patch bundle disagrees with the promoted digest")
        return sealed
    source = resolve_operator_patch_dir(requested, repo_root=source_root)
    source_definition = runtime_patch_modules()["load_patch_bundle_definition"](source)
    require_patch_bundle_definition_environment(
        source_definition,
        environment=environment,
        patch_dir=source,
    )
    require_patch_bundle_matches_d3_context(context, source_definition)
    sealed.parent.mkdir(parents=True, mode=0o700, exist_ok=True)
    result = seal_patch_bundle(source, sealed)
    actual = require_sealed_patch_bundle(result)
    sealed_definition = runtime_patch_modules()["load_patch_bundle_definition"](result)
    require_patch_bundle_definition_environment(
        sealed_definition,
        environment=environment,
        patch_dir=result,
    )
    require_patch_bundle_matches_d3_context(context, sealed_definition)
    if expected and actual != expected:
        raise DeployError("sealed D3 patch bundle disagrees with the promoted digest")
    return result


def replay_patch_before_catalog(
    context: D3Context,
    *,
    requested_patch_dir: str,
    environment: str | None = None,
    apply_bundle: Callable[..., dict[str, Any]] = apply_sealed_patch_bundle,
    read_state: Callable[[Path], Mapping[str, Any]] | None = None,
) -> dict[str, Any] | None:
    """Replay the already-sealed operator patch bundle immediately before Catalog apply.

    SQL runs in the staging apply-patch window: after frozen stopped-upgrade
    quiescence and immediately before Catalog apply (preconvergence /
    apply-offline-candidate). catalog-ready is a same-process fallback only if
    this process has not already replayed. Earlier leftover phases and
    catalog-in-progress skip SQL.
    """
    phase = ""
    if read_state is not None:
        phase = active_run_phase(context, read_state=read_state)
    if phase not in PATCH_SQL_READY_PHASES:
        return None
    sealed = patch_seal_dir(context)
    requested = requested_patch_dir.strip()
    if not (sealed.exists() or sealed.is_symlink()):
        if requested:
            raise DeployError("D3 patch bundle was not sealed before Catalog")
        return None
    require_sealed_patch_bundle(sealed)
    definition = runtime_patch_modules()["load_patch_bundle_definition"](sealed)
    require_patch_bundle_definition_environment(
        definition,
        environment=environment,
        patch_dir=sealed,
    )
    require_patch_bundle_matches_d3_context(context, definition)
    return apply_bundle(context, sealed, requested_patch_dir=requested)