"""Transactional State Hub projection migration for deterministic work-record UUIDs."""

from __future__ import annotations

import hashlib
import json
import re
import uuid
from dataclasses import dataclass
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import func, select, text
from sqlalchemy.ext.asyncio import AsyncSession

from api.models.managed_repo import ManagedRepo
from api.models.task import Task, TaskPriority, TaskStatus
from api.models.work_record_identifier_alias import WorkRecordIdentifierAlias
from api.models.workplan import Workplan

PLAN_SCHEMA = "repo-manager.identifier-migration-plan.v1"
FLEET_NAMESPACE = "helixforge"
DERIVATION_NAMESPACE_UUID = uuid.UUID("a4058507-5c4a-5a00-ab06-fffa4fb46009")
_RECORD_ID_RE = re.compile(r"^[A-Z][A-Z0-9-]*-WP-[0-9]{4}(?:-T[0-9]{2,})?$")


class IdentifierMigrationError(ValueError):
    """The signed plan or current projection is unsafe to migrate."""


@dataclass(frozen=True)
class IdentifierMigrationResult:
    repo_slug: str
    plan_sha256: str
    direction: str
    replacements: int
    assignments_deferred: int


@dataclass(frozen=True)
class SealedProjectionRepairReceipt:
    """Non-secret evidence for one bounded pre-derivation projection repair."""

    schema: str
    outcome: str
    repo_slug: str
    repository_id: str
    workplan_record_id: str
    workplan_old_id: str
    task_records: tuple[tuple[str, str], ...]
    plan_sha256: str
    source_revision: str
    source_fingerprint: str
    projection_identity: str
    observed_at: str


def verify_plan(plan: dict[str, Any]) -> str:
    """Validate the immutable plan envelope and deterministic UUID mappings."""
    if plan.get("schema") != PLAN_SCHEMA:
        raise IdentifierMigrationError("unsupported identifier migration plan schema")
    if plan.get("namespace") != FLEET_NAMESPACE:
        raise IdentifierMigrationError(
            f"plan namespace must be {FLEET_NAMESPACE!r}"
        )
    if plan.get("apply_policy") != "all-or-nothing per repository":
        raise IdentifierMigrationError("plan does not declare repository-atomic apply")
    if plan.get("ready_to_apply") is not True:
        raise IdentifierMigrationError("plan is not ready_to_apply")

    expected = plan.get("plan_sha256")
    if not isinstance(expected, str) or not re.fullmatch(r"[0-9a-f]{64}", expected):
        raise IdentifierMigrationError("plan has no valid SHA-256 seal")
    unsealed = {key: value for key, value in plan.items() if key != "plan_sha256"}
    canonical = json.dumps(unsealed, sort_keys=True, separators=(",", ":")).encode()
    actual = hashlib.sha256(canonical).hexdigest()
    if actual != expected:
        raise IdentifierMigrationError("plan SHA-256 mismatch")
    return expected


def _repository_mappings(
    plan: dict[str, Any], repo_slug: str
) -> tuple[list[dict[str, Any]], int]:
    matches = [item for item in plan.get("repositories", []) if item.get("repo") == repo_slug]
    if len(matches) != 1:
        raise IdentifierMigrationError(
            f"repository {repo_slug!r} must occur exactly once in plan"
        )
    repository = matches[0]
    if repository.get("eligible") is not True or repository.get("atomic_unit") is not True:
        raise IdentifierMigrationError(f"repository {repo_slug!r} is not eligible and atomic")

    replacements: list[dict[str, Any]] = []
    assignments = 0
    seen_old: set[uuid.UUID] = set()
    seen_new: set[uuid.UUID] = set()
    for raw in repository.get("mappings", []):
        action = raw.get("action")
        if action == "assign":
            assignments += 1
            continue
        if action == "unchanged":
            continue
        if action != "replace":
            raise IdentifierMigrationError(f"unsupported mapping action {action!r}")
        if raw.get("repo") != repo_slug:
            raise IdentifierMigrationError("mapping repository does not match atomic unit")
        kind = raw.get("kind")
        record_id = raw.get("record_id")
        if kind not in {"workplan", "task"}:
            raise IdentifierMigrationError(f"unsupported record kind {kind!r}")
        if not isinstance(record_id, str) or not _RECORD_ID_RE.fullmatch(record_id):
            raise IdentifierMigrationError(f"noncanonical record id {record_id!r}")
        try:
            old_id = uuid.UUID(str(raw.get("current_uuid")))
            new_id = uuid.UUID(str(raw.get("derived_uuid")))
        except (TypeError, ValueError) as exc:
            raise IdentifierMigrationError(f"invalid UUID mapping for {record_id}") from exc
        expected_new = uuid.uuid5(
            DERIVATION_NAMESPACE_UUID,
            f"{FLEET_NAMESPACE}\n{record_id}",
        )
        if new_id != expected_new:
            raise IdentifierMigrationError(f"derived UUID mismatch for {record_id}")
        if old_id == new_id or old_id in seen_old or new_id in seen_new:
            raise IdentifierMigrationError(f"non-unique replacement for {record_id}")
        seen_old.add(old_id)
        seen_new.add(new_id)
        replacements.append(
            {
                "kind": kind,
                "record_id": record_id,
                "old_id": old_id,
                "new_id": new_id,
            }
        )
    return replacements, assignments


def _repository_entry(plan: dict[str, Any], repo_slug: str) -> dict[str, Any]:
    matches = [item for item in plan.get("repositories", []) if item.get("repo") == repo_slug]
    if len(matches) != 1:
        raise IdentifierMigrationError(
            f"repository {repo_slug!r} must occur exactly once in plan"
        )
    return matches[0]


def _repair_unit_mappings(
    plan: dict[str, Any], repo_slug: str, unit: dict[str, Any]
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    replacements, _assignments = _repository_mappings(plan, repo_slug)
    workplan = unit.get("workplan")
    tasks = unit.get("tasks")
    if not isinstance(workplan, dict) or not isinstance(tasks, list):
        raise IdentifierMigrationError("repair unit requires one workplan and a task list")

    workplan_record_id = workplan.get("record_id")
    workplan_matches = [
        mapping
        for mapping in replacements
        if mapping["kind"] == "workplan" and mapping["record_id"] == workplan_record_id
    ]
    if len(workplan_matches) != 1:
        raise IdentifierMigrationError(
            "repair workplan must occur exactly once as a replacement in the sealed plan"
        )
    workplan_mapping = workplan_matches[0]
    raw_workplan_mapping = next(
        raw
        for raw in _repository_entry(plan, repo_slug).get("mappings", [])
        if raw.get("kind") == "workplan" and raw.get("record_id") == workplan_record_id
    )
    authoritative_path = raw_workplan_mapping.get("path")
    if not isinstance(authoritative_path, str) or not authoritative_path:
        raise IdentifierMigrationError("sealed workplan mapping has no authoritative path")

    task_record_ids = [task.get("record_id") for task in tasks if isinstance(task, dict)]
    if len(task_record_ids) != len(tasks) or len(set(task_record_ids)) != len(tasks):
        raise IdentifierMigrationError("repair task record ids must be explicit and unique")
    task_mappings = [
        mapping
        for mapping in replacements
        if mapping["kind"] == "task" and mapping["record_id"] in task_record_ids
    ]
    mapped_task_ids = {mapping["record_id"] for mapping in task_mappings}
    if mapped_task_ids != set(task_record_ids):
        raise IdentifierMigrationError(
            "repair tasks do not exactly match replacement mappings in the sealed plan"
        )

    raw_unit_task_ids = {
        raw.get("record_id")
        for raw in _repository_entry(plan, repo_slug).get("mappings", [])
        if raw.get("kind") == "task"
        and raw.get("action") == "replace"
        and raw.get("path") == authoritative_path
    }
    if raw_unit_task_ids != set(task_record_ids):
        raise IdentifierMigrationError(
            "repair must include every replaced task in the sealed workplan unit"
        )
    return workplan_mapping, task_mappings


def _uuid_field(value: Any, label: str) -> uuid.UUID:
    try:
        return uuid.UUID(str(value))
    except (TypeError, ValueError) as exc:
        raise IdentifierMigrationError(f"invalid UUID for {label}") from exc


def _repair_payloads(
    unit: dict[str, Any],
    workplan_mapping: dict[str, Any],
    task_mappings: list[dict[str, Any]],
) -> tuple[dict[str, Any], list[dict[str, Any]]]:
    raw_workplan = unit["workplan"]
    if _uuid_field(raw_workplan.get("id"), "workplan") != workplan_mapping["old_id"]:
        raise IdentifierMigrationError("repair workplan UUID does not match sealed old UUID")
    slug = raw_workplan.get("slug")
    title = raw_workplan.get("title")
    status = raw_workplan.get("status")
    if not isinstance(slug, str) or not slug or not isinstance(title, str) or not title:
        raise IdentifierMigrationError("repair workplan requires explicit slug and title")
    if status not in {
        "proposed", "ready", "active", "blocked", "backlog", "finished", "archived"
    }:
        raise IdentifierMigrationError(f"unsupported repair workplan status {status!r}")
    workplan_payload = {
        "id": workplan_mapping["old_id"],
        "topic_id": _uuid_field(raw_workplan.get("topic_id"), "workplan topic"),
        "slug": slug,
        "title": title,
        "description": raw_workplan.get("description"),
        "status": status,
        "owner": raw_workplan.get("owner"),
        "planning_priority": raw_workplan.get("planning_priority"),
        "planning_order": raw_workplan.get("planning_order"),
    }

    mappings_by_record = {mapping["record_id"]: mapping for mapping in task_mappings}
    task_payloads: list[dict[str, Any]] = []
    for raw_task in unit["tasks"]:
        mapping = mappings_by_record[raw_task["record_id"]]
        if _uuid_field(raw_task.get("id"), raw_task["record_id"]) != mapping["old_id"]:
            raise IdentifierMigrationError(
                f"repair task UUID does not match sealed old UUID for {raw_task['record_id']}"
            )
        title = raw_task.get("title")
        if not isinstance(title, str) or not title:
            raise IdentifierMigrationError(
                f"repair task {raw_task['record_id']} requires an explicit title"
            )
        try:
            status_value = TaskStatus(raw_task.get("status", "todo"))
            priority_value = TaskPriority(raw_task.get("priority", "medium"))
        except ValueError as exc:
            raise IdentifierMigrationError(
                f"invalid task state for {raw_task['record_id']}"
            ) from exc
        parent_record_id = raw_task.get("parent_record_id")
        if parent_record_id is not None and parent_record_id not in mappings_by_record:
            raise IdentifierMigrationError(
                f"parent task for {raw_task['record_id']} is outside the sealed repair unit"
            )
        task_payloads.append(
            {
                "record_id": raw_task["record_id"],
                "id": mapping["old_id"],
                "title": title,
                "description": raw_task.get("description"),
                "status": status_value,
                "priority": priority_value,
                "assignee": raw_task.get("assignee"),
                "blocking_reason": raw_task.get("blocking_reason"),
                "needs_human": bool(raw_task.get("needs_human", False)),
                "intervention_note": raw_task.get("intervention_note"),
                "parent_task_id": (
                    mappings_by_record[parent_record_id]["old_id"]
                    if parent_record_id is not None
                    else None
                ),
            }
        )
        if task_payloads[-1]["needs_human"] and not task_payloads[-1]["intervention_note"]:
            raise IdentifierMigrationError(
                f"repair task {raw_task['record_id']} needs an intervention note"
            )
        if (
            task_payloads[-1]["status"] == TaskStatus.wait
            and task_payloads[-1]["needs_human"]
            and not task_payloads[-1]["blocking_reason"]
        ):
            raise IdentifierMigrationError(
                f"repair task {raw_task['record_id']} needs a blocking reason"
            )

    pending = list(task_payloads)
    ordered: list[dict[str, Any]] = []
    emitted: set[uuid.UUID] = set()
    while pending:
        ready = [
            task
            for task in pending
            if task["parent_task_id"] is None or task["parent_task_id"] in emitted
        ]
        if not ready:
            raise IdentifierMigrationError("repair task parentage contains a cycle")
        for task in ready:
            pending.remove(task)
            ordered.append(task)
            emitted.add(task["id"])
    return workplan_payload, ordered


def _same_workplan(row: Workplan, expected: dict[str, Any], repo_id: uuid.UUID) -> bool:
    fields = (
        "id", "topic_id", "slug", "title", "description", "status", "owner",
        "planning_priority", "planning_order",
    )
    return row.repo_id == repo_id and all(
        getattr(row, field) == expected[field] for field in fields
    )


def _same_task(row: Task, expected: dict[str, Any], workplan_id: uuid.UUID) -> bool:
    fields = (
        "id", "title", "description", "status", "priority", "assignee",
        "blocking_reason", "needs_human", "intervention_note", "parent_task_id",
    )
    return row.workplan_id == workplan_id and all(
        getattr(row, field) == expected[field] for field in fields
    )


async def repair_absent_prederivation_projection(
    session: AsyncSession,
    plan: dict[str, Any],
    repo_slug: str,
    unit: dict[str, Any],
    *,
    expected_plan_sha256: str,
    source_revision: str,
    source_fingerprint: str,
    source_clean: bool,
    source_synchronized: bool,
    primary_confirmed: bool,
    projection_identity: str,
    observed_at: datetime | None = None,
) -> SealedProjectionRepairReceipt:
    """Restore one exact random-ID workplan unit into an absent projection.

    This is intentionally separate from ordinary consistency repair. The caller
    must independently prove Git cleanliness/synchronization and provide the
    authoritative unit parsed from the pinned source revision.
    """
    plan_sha256 = verify_plan(plan)
    if plan_sha256 != expected_plan_sha256:
        raise IdentifierMigrationError("explicit plan SHA-256 does not match sealed plan")
    if not primary_confirmed:
        raise IdentifierMigrationError("projection repair requires explicit primary confirmation")
    if not source_clean or not source_synchronized:
        raise IdentifierMigrationError("projection repair requires a clean synchronized source")
    if not re.fullmatch(r"[0-9a-f]{40}", source_revision):
        raise IdentifierMigrationError("source revision must be a full Git SHA")
    if not re.fullmatch(r"[0-9a-f]{64}", source_fingerprint):
        raise IdentifierMigrationError("source fingerprint must be SHA-256")
    if not projection_identity.strip():
        raise IdentifierMigrationError("projection identity is required")

    repository_entry = _repository_entry(plan, repo_slug)
    if source_revision != repository_entry.get("planned_head_sha"):
        raise IdentifierMigrationError("source revision drifted from sealed plan")
    if source_fingerprint != repository_entry.get("source_fingerprint"):
        raise IdentifierMigrationError("source fingerprint drifted from sealed plan")
    workplan_mapping, task_mappings = _repair_unit_mappings(plan, repo_slug, unit)
    workplan_payload, task_payloads = _repair_payloads(
        unit, workplan_mapping, task_mappings
    )
    if session.in_transaction():
        raise IdentifierMigrationError("projection repair requires a fresh database session")

    outcome = "repaired"
    async with session.begin():
        repo = await session.scalar(
            select(ManagedRepo).where(ManagedRepo.slug == repo_slug).with_for_update()
        )
        if repo is None:
            raise IdentifierMigrationError(f"repository projection is absent: {repo_slug}")
        if repo.topic_id != workplan_payload["topic_id"]:
            raise IdentifierMigrationError("repair topic does not match repository projection")

        old_ids = [workplan_mapping["old_id"], *(m["old_id"] for m in task_mappings)]
        new_workplan = await session.get(Workplan, workplan_mapping["new_id"])
        new_tasks = [await session.get(Task, mapping["new_id"]) for mapping in task_mappings]
        if new_workplan is not None or any(task is not None for task in new_tasks):
            raise IdentifierMigrationError("derived target presence blocks projection repair")
        alias_count = await session.scalar(
            select(func.count()).select_from(WorkRecordIdentifierAlias).where(
                WorkRecordIdentifierAlias.old_id.in_(old_ids)
            )
        )
        if alias_count:
            raise IdentifierMigrationError("durable alias presence blocks projection repair")

        old_workplan = await session.get(Workplan, workplan_mapping["old_id"])
        old_tasks = [await session.get(Task, mapping["old_id"]) for mapping in task_mappings]
        present_count = int(old_workplan is not None) + sum(task is not None for task in old_tasks)
        expected_count = 1 + len(task_mappings)
        if present_count not in {0, expected_count}:
            raise IdentifierMigrationError("partial old projection presence blocks repair")

        if present_count == expected_count:
            if not _same_workplan(old_workplan, workplan_payload, repo.id):
                raise IdentifierMigrationError(
                    "existing old workplan differs from sealed repair unit"
                )
            expected_tasks = {task["id"]: task for task in task_payloads}
            if any(
                not _same_task(task, expected_tasks[task.id], workplan_mapping["old_id"])
                for task in old_tasks
            ):
                raise IdentifierMigrationError("existing old tasks differ from sealed repair unit")
            outcome = "verified_noop"
        else:
            slug_conflict = await session.scalar(
                select(Workplan.id).where(Workplan.slug == workplan_payload["slug"])
            )
            if slug_conflict is not None:
                raise IdentifierMigrationError("workplan slug is already projected")
            session.add(
                Workplan(repo_id=repo.id, **workplan_payload)
            )
            await session.flush()
            for task_payload in task_payloads:
                values = {key: value for key, value in task_payload.items() if key != "record_id"}
                session.add(Task(workplan_id=workplan_mapping["old_id"], **values))
            await session.flush()

    timestamp = observed_at or datetime.now(UTC)
    if timestamp.tzinfo is None:
        raise IdentifierMigrationError("receipt observation time must be timezone-aware")
    return SealedProjectionRepairReceipt(
        schema="state-hub.sealed-prederivation-projection-repair-receipt.v1",
        outcome=outcome,
        repo_slug=repo_slug,
        repository_id=str(repo.id),
        workplan_record_id=workplan_mapping["record_id"],
        workplan_old_id=str(workplan_mapping["old_id"]),
        task_records=tuple(
            (mapping["record_id"], str(mapping["old_id"])) for mapping in task_mappings
        ),
        plan_sha256=plan_sha256,
        source_revision=source_revision,
        source_fingerprint=source_fingerprint,
        projection_identity=projection_identity,
        observed_at=timestamp.astimezone(UTC).isoformat().replace("+00:00", "Z"),
    )


async def _assert_projection_preconditions(
    session: AsyncSession,
    repo_slug: str,
    replacements: list[dict[str, Any]],
    *,
    reverse: bool,
) -> None:
    for mapping in replacements:
        source_id = mapping["new_id"] if reverse else mapping["old_id"]
        target_id = mapping["old_id"] if reverse else mapping["new_id"]
        if mapping["kind"] == "workplan":
            source_query = text(
                "SELECT workplans.id FROM workplans "
                "JOIN managed_repos ON managed_repos.id = workplans.repo_id "
                "WHERE managed_repos.slug = :repo_slug AND workplans.id = :record_id "
                "FOR UPDATE"
            )
            target_query = text("SELECT id FROM workplans WHERE id = :record_id")
        else:
            source_query = text(
                "SELECT tasks.id FROM tasks "
                "JOIN workplans ON workplans.id = tasks.workplan_id "
                "JOIN managed_repos ON managed_repos.id = workplans.repo_id "
                "WHERE managed_repos.slug = :repo_slug AND tasks.id = :record_id "
                "FOR UPDATE"
            )
            target_query = text("SELECT id FROM tasks WHERE id = :record_id")
        source = await session.execute(
            source_query,
            {"repo_slug": repo_slug, "record_id": source_id},
        )
        if source.scalar_one_or_none() is None:
            raise IdentifierMigrationError(
                f"source {mapping['kind']} {source_id} is absent from repository {repo_slug}"
            )
        target = await session.execute(target_query, {"record_id": target_id})
        if target.scalar_one_or_none() is not None:
            raise IdentifierMigrationError(
                f"target {mapping['kind']} {target_id} already exists"
            )


async def apply_repository_identifier_migration(
    session: AsyncSession,
    plan: dict[str, Any],
    repo_slug: str,
) -> IdentifierMigrationResult:
    """Replace one repository's projected UUIDs in one database transaction."""
    plan_sha256 = verify_plan(plan)
    replacements, assignments = _repository_mappings(plan, repo_slug)
    if session.in_transaction():
        raise IdentifierMigrationError("migration requires a fresh database session")

    async with session.begin():
        await _assert_projection_preconditions(
            session, repo_slug, replacements, reverse=False
        )
        aliases = {
            alias.old_id: alias
            for alias in (
                await session.execute(
                    select(WorkRecordIdentifierAlias).where(
                        WorkRecordIdentifierAlias.old_id.in_(
                            [mapping["old_id"] for mapping in replacements]
                        )
                    )
                )
            ).scalars()
        }
        for mapping in replacements:
            alias = aliases.get(mapping["old_id"])
            if alias is None:
                alias = WorkRecordIdentifierAlias(
                    old_id=mapping["old_id"],
                    new_id=mapping["new_id"],
                    record_kind=mapping["kind"],
                    record_id=mapping["record_id"],
                    repo_slug=repo_slug,
                    namespace=FLEET_NAMESPACE,
                    plan_sha256=plan_sha256,
                )
                session.add(alias)
                aliases[mapping["old_id"]] = alias
            elif (
                alias.new_id != mapping["new_id"]
                or alias.record_kind != mapping["kind"]
                or alias.record_id != mapping["record_id"]
                or alias.repo_slug != repo_slug
                or alias.namespace != FLEET_NAMESPACE
                or alias.plan_sha256 != plan_sha256
                or alias.migration_status != "reversed"
            ):
                raise IdentifierMigrationError(
                    f"conflicting durable alias for {mapping['old_id']}"
                )
        await session.flush()

        for kind in ("workplan", "task"):
            table = "workplans" if kind == "workplan" else "tasks"
            for mapping in (item for item in replacements if item["kind"] == kind):
                result = await session.execute(
                    text(f"UPDATE {table} SET id = :new_id WHERE id = :old_id"),
                    {"old_id": mapping["old_id"], "new_id": mapping["new_id"]},
                )
                if result.rowcount != 1:
                    raise IdentifierMigrationError(
                        f"failed to replace {mapping['record_id']}"
                    )

        now = datetime.now(UTC)
        for alias in aliases.values():
            alias.migration_status = "applied"
            alias.applied_at = now
            alias.reversed_at = None

    return IdentifierMigrationResult(
        repo_slug=repo_slug,
        plan_sha256=plan_sha256,
        direction="forward",
        replacements=len(replacements),
        assignments_deferred=assignments,
    )


async def reverse_repository_identifier_migration(
    session: AsyncSession,
    plan: dict[str, Any],
    repo_slug: str,
) -> IdentifierMigrationResult:
    """Restore one repository's prior projected UUIDs and retain aliases."""
    plan_sha256 = verify_plan(plan)
    replacements, assignments = _repository_mappings(plan, repo_slug)
    if session.in_transaction():
        raise IdentifierMigrationError("migration requires a fresh database session")

    async with session.begin():
        aliases = list(
            (
                await session.execute(
                    select(WorkRecordIdentifierAlias).where(
                        WorkRecordIdentifierAlias.plan_sha256 == plan_sha256,
                        WorkRecordIdentifierAlias.repo_slug == repo_slug,
                    )
                )
            ).scalars()
        )
        aliases_by_old = {alias.old_id: alias for alias in aliases}
        if len(aliases_by_old) != len(replacements):
            raise IdentifierMigrationError("durable alias set is incomplete")
        for mapping in replacements:
            alias = aliases_by_old.get(mapping["old_id"])
            if (
                alias is None
                or alias.new_id != mapping["new_id"]
                or alias.record_kind != mapping["kind"]
                or alias.record_id != mapping["record_id"]
                or alias.migration_status != "applied"
            ):
                raise IdentifierMigrationError(
                    f"durable alias is not applied for {mapping['record_id']}"
                )
        await _assert_projection_preconditions(
            session, repo_slug, replacements, reverse=True
        )

        for kind in ("task", "workplan"):
            table = "tasks" if kind == "task" else "workplans"
            for mapping in (item for item in replacements if item["kind"] == kind):
                result = await session.execute(
                    text(f"UPDATE {table} SET id = :old_id WHERE id = :new_id"),
                    {"old_id": mapping["old_id"], "new_id": mapping["new_id"]},
                )
                if result.rowcount != 1:
                    raise IdentifierMigrationError(
                        f"failed to reverse {mapping['record_id']}"
                    )

        now = datetime.now(UTC)
        for alias in aliases:
            alias.migration_status = "reversed"
            alias.reversed_at = now

    return IdentifierMigrationResult(
        repo_slug=repo_slug,
        plan_sha256=plan_sha256,
        direction="reverse",
        replacements=len(replacements),
        assignments_deferred=assignments,
    )
