from __future__ import annotations

import hashlib
import json
import uuid

import pytest
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from api.models import Base
from api.models.domain import Domain
from api.models.managed_repo import ManagedRepo
from api.models.progress_event import ProgressEvent
from api.models.task import Task
from api.models.topic import Topic
from api.models.token_event import TokenEvent
from api.models.work_record_identifier_alias import WorkRecordIdentifierAlias
from api.models.workplan import Workplan
from api.models.workplan_dependency import WorkplanDependency
from api.services.work_record_identifier_migration import (
    DERIVATION_NAMESPACE_UUID,
    IdentifierMigrationError,
    apply_repository_identifier_migration,
    repair_absent_prederivation_projection,
    reverse_repository_identifier_migration,
)


def _derived(record_id: str) -> uuid.UUID:
    return uuid.uuid5(DERIVATION_NAMESPACE_UUID, f"helixforge\n{record_id}")


def _sealed_plan(
    repo_slug: str,
    workplan_old: uuid.UUID,
    task_old: uuid.UUID,
    *,
    task_source_override: uuid.UUID | None = None,
) -> dict:
    mappings = [
        {
            "repo": repo_slug,
            "path": "workplans/TEST-WP-0001-migration.md",
            "kind": "workplan",
            "record_id": "TEST-WP-0001",
            "current_uuid": str(workplan_old),
            "derived_uuid": str(_derived("TEST-WP-0001")),
            "action": "replace",
        },
        {
            "repo": repo_slug,
            "path": "workplans/TEST-WP-0001-migration.md",
            "kind": "task",
            "record_id": "TEST-WP-0001-T01",
            "current_uuid": str(task_source_override or task_old),
            "derived_uuid": str(_derived("TEST-WP-0001-T01")),
            "action": "replace",
        },
        {
            "repo": repo_slug,
            "path": "workplans/TEST-WP-0001-migration.md",
            "kind": "task",
            "record_id": "TEST-WP-0001-T02",
            "current_uuid": None,
            "derived_uuid": str(_derived("TEST-WP-0001-T02")),
            "action": "assign",
        },
    ]
    plan = {
        "schema": "repo-manager.identifier-migration-plan.v1",
        "ok": True,
        "ready_to_apply": True,
        "root": "/test",
        "namespace": "helixforge",
        "derivation_version": "repo-manager.work-record-uuid.v1",
        "namespace_uuid": str(DERIVATION_NAMESPACE_UUID),
        "scope": "live workplans and unfinished tasks",
        "apply_policy": "all-or-nothing per repository",
        "generated_at": "2026-08-21T00:00:00Z",
        "totals": {
            "repositories": 1,
            "eligible": 1,
            "skipped": 0,
            "records": 3,
            "replace": 2,
            "assign": 1,
            "unchanged": 0,
        },
        "repositories": [
            {
                "repo": repo_slug,
                "path": "/test/repo",
                "planned_head_sha": "0" * 40,
                "source_fingerprint": "1" * 64,
                "eligible": True,
                "atomic_unit": True,
                "blockers": [],
                "mappings": mappings,
            }
        ],
    }
    canonical = json.dumps(plan, sort_keys=True, separators=(",", ":")).encode()
    plan["plan_sha256"] = hashlib.sha256(canonical).hexdigest()
    return plan


async def _seed_projection(factory, repo_slug: str):
    domain_id = uuid.uuid4()
    repo_id = uuid.uuid4()
    workplan_old = uuid.uuid4()
    task_old = uuid.uuid4()
    child_task_id = uuid.uuid4()
    progress_id = uuid.uuid4()
    token_id = uuid.uuid4()
    dependency_id = uuid.uuid4()
    async with factory() as session:
        session.add_all(
            [
                Domain(id=domain_id, slug="infotech", name="Infotech", status="active"),
                ManagedRepo(
                    id=repo_id,
                    domain_id=domain_id,
                    slug=repo_slug,
                    name="Test Repository",
                    status="active",
                ),
                Workplan(
                    id=workplan_old,
                    repo_id=repo_id,
                    slug="test-wp-0001",
                    title="Migration test",
                    status="active",
                ),
                Task(
                    id=task_old,
                    workplan_id=workplan_old,
                    title="Mapped task",
                    status="progress",
                    priority="high",
                ),
                Task(
                    id=child_task_id,
                    workplan_id=workplan_old,
                    parent_task_id=task_old,
                    title="Unmapped child",
                    status="todo",
                    priority="medium",
                ),
                ProgressEvent(
                    id=progress_id,
                    workplan_id=workplan_old,
                    task_id=task_old,
                    event_type="note",
                    summary="migration evidence",
                ),
                TokenEvent(
                    id=token_id,
                    workplan_id=workplan_old,
                    task_id=task_old,
                    tokens_in=1,
                    tokens_out=1,
                ),
                WorkplanDependency(
                    id=dependency_id,
                    from_workplan_id=workplan_old,
                    to_task_id=task_old,
                    relationship_type="blocks",
                ),
            ]
        )
        await session.commit()
    return {
        "workplan": workplan_old,
        "task": task_old,
        "child": child_task_id,
        "progress": progress_id,
        "token": token_id,
        "dependency": dependency_id,
    }


def test_every_work_record_foreign_key_cascades_on_update():
    references = [
        foreign_key
        for table in Base.metadata.sorted_tables
        for foreign_key in table.foreign_keys
        if foreign_key.target_fullname in {"workplans.id", "tasks.id"}
    ]
    assert len(references) == 22
    assert all(foreign_key.onupdate == "CASCADE" for foreign_key in references)


@pytest.mark.asyncio
async def test_repository_migration_cascades_and_reverses(test_engine):
    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
    ids = await _seed_projection(factory, "test-repo")
    plan = _sealed_plan("test-repo", ids["workplan"], ids["task"])
    workplan_new = _derived("TEST-WP-0001")
    task_new = _derived("TEST-WP-0001-T01")

    async with factory() as session:
        result = await apply_repository_identifier_migration(session, plan, "test-repo")
    assert result.direction == "forward"
    assert result.replacements == 2
    assert result.assignments_deferred == 1

    async with factory() as session:
        assert await session.scalar(
            text("SELECT id FROM workplans WHERE id = :id"), {"id": workplan_new}
        ) == workplan_new
        assert await session.scalar(
            text("SELECT workplan_id FROM tasks WHERE id = :id"), {"id": task_new}
        ) == workplan_new
        assert await session.scalar(
            text("SELECT parent_task_id FROM tasks WHERE id = :id"), {"id": ids["child"]}
        ) == task_new
        assert tuple(
            (
                await session.execute(
                    text(
                        "SELECT workplan_id, task_id FROM progress_events WHERE id = :id"
                    ),
                    {"id": ids["progress"]},
                )
            ).one()
        ) == (workplan_new, task_new)
        assert tuple(
            (
                await session.execute(
                    text("SELECT workplan_id, task_id FROM token_events WHERE id = :id"),
                    {"id": ids["token"]},
                )
            ).one()
        ) == (workplan_new, task_new)
        assert tuple(
            (
                await session.execute(
                    text(
                        "SELECT from_workplan_id, to_task_id "
                        "FROM workplan_dependencies WHERE id = :id"
                    ),
                    {"id": ids["dependency"]},
                )
            ).one()
        ) == (workplan_new, task_new)
        aliases = list(
            (
                await session.execute(
                    select(WorkRecordIdentifierAlias).order_by(
                        WorkRecordIdentifierAlias.record_kind
                    )
                )
            ).scalars()
        )
        assert len(aliases) == 2
        assert all(alias.migration_status == "applied" for alias in aliases)

    async with factory() as session:
        result = await reverse_repository_identifier_migration(session, plan, "test-repo")
    assert result.direction == "reverse"

    async with factory() as session:
        assert await session.scalar(
            text("SELECT id FROM workplans WHERE id = :id"), {"id": ids["workplan"]}
        ) == ids["workplan"]
        assert await session.scalar(
            text("SELECT workplan_id FROM tasks WHERE id = :id"), {"id": ids["task"]}
        ) == ids["workplan"]
        assert await session.scalar(
            text("SELECT parent_task_id FROM tasks WHERE id = :id"), {"id": ids["child"]}
        ) == ids["task"]
        aliases = list((await session.execute(select(WorkRecordIdentifierAlias))).scalars())
        assert len(aliases) == 2
        assert all(alias.migration_status == "reversed" for alias in aliases)
        assert all(alias.reversed_at is not None for alias in aliases)


@pytest.mark.asyncio
async def test_repository_migration_is_atomic_when_a_source_is_missing(test_engine):
    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
    ids = await _seed_projection(factory, "atomic-repo")
    plan = _sealed_plan(
        "atomic-repo",
        ids["workplan"],
        ids["task"],
        task_source_override=uuid.uuid4(),
    )

    async with factory() as session:
        with pytest.raises(IdentifierMigrationError, match="source task"):
            await apply_repository_identifier_migration(session, plan, "atomic-repo")

    async with factory() as session:
        assert await session.scalar(
            text("SELECT id FROM workplans WHERE id = :id"), {"id": ids["workplan"]}
        ) == ids["workplan"]
        assert await session.scalar(
            text("SELECT count(*) FROM work_record_identifier_aliases")
        ) == 0


async def _seed_empty_repository(factory, repo_slug: str):
    domain_id = uuid.uuid4()
    topic_id = uuid.uuid4()
    repo_id = uuid.uuid4()
    async with factory() as session:
        session.add(Domain(id=domain_id, slug="infotech", name="Infotech", status="active"))
        await session.flush()
        session.add(
            Topic(
                id=topic_id,
                domain_id=domain_id,
                slug="infotech",
                title="Infotech",
                status="active",
            )
        )
        await session.flush()
        session.add(
            ManagedRepo(
                id=repo_id,
                domain_id=domain_id,
                topic_id=topic_id,
                slug=repo_slug,
                name="Repair Repository",
                status="active",
            )
        )
        await session.commit()
    return repo_id, topic_id


def _repair_unit(plan: dict, topic_id: uuid.UUID) -> dict:
    repository = plan["repositories"][0]
    replacements = [m for m in repository["mappings"] if m["action"] == "replace"]
    workplan = next(m for m in replacements if m["kind"] == "workplan")
    task = next(m for m in replacements if m["kind"] == "task")
    return {
        "workplan": {
            "record_id": workplan["record_id"],
            "id": workplan["current_uuid"],
            "topic_id": str(topic_id),
            "slug": "test-wp-0001",
            "title": "Migration test",
            "description": "Exact authoritative projection",
            "status": "active",
            "owner": "codex",
            "planning_priority": "high",
            "planning_order": 1,
        },
        "tasks": [
            {
                "record_id": task["record_id"],
                "id": task["current_uuid"],
                "title": "Mapped task",
                "description": "Exact task projection",
                "status": "progress",
                "priority": "high",
                "assignee": "codex",
            }
        ],
    }


def _repair_kwargs(plan: dict) -> dict:
    repository = plan["repositories"][0]
    return {
        "expected_plan_sha256": plan["plan_sha256"],
        "source_revision": repository["planned_head_sha"],
        "source_fingerprint": repository["source_fingerprint"],
        "source_clean": True,
        "source_synchronized": True,
        "primary_confirmed": True,
        "projection_identity": "test-projection",
    }


@pytest.mark.asyncio
async def test_sealed_repair_restores_absent_projection_and_retries_as_noop(test_engine):
    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
    repo_id, topic_id = await _seed_empty_repository(factory, "repair-repo")
    workplan_old = uuid.uuid4()
    task_old = uuid.uuid4()
    plan = _sealed_plan("repair-repo", workplan_old, task_old)
    unit = _repair_unit(plan, topic_id)

    async with factory() as session:
        receipt = await repair_absent_prederivation_projection(
            session, plan, "repair-repo", unit, **_repair_kwargs(plan)
        )
    assert receipt.outcome == "repaired"
    assert receipt.repository_id == str(repo_id)
    assert receipt.workplan_old_id == str(workplan_old)
    assert receipt.task_records == (("TEST-WP-0001-T01", str(task_old)),)

    async with factory() as session:
        assert await session.get(Workplan, workplan_old) is not None
        assert await session.get(Task, task_old) is not None
        assert await session.get(Workplan, _derived("TEST-WP-0001")) is None
        assert await session.get(Task, _derived("TEST-WP-0001-T01")) is None

    async with factory() as session:
        retry = await repair_absent_prederivation_projection(
            session, plan, "repair-repo", unit, **_repair_kwargs(plan)
        )
    assert retry.outcome == "verified_noop"


@pytest.mark.asyncio
@pytest.mark.parametrize(
    ("override", "message"),
    [
        ({"source_clean": False}, "clean synchronized"),
        ({"source_synchronized": False}, "clean synchronized"),
        ({"source_revision": "f" * 40}, "source revision drifted"),
        ({"source_fingerprint": "e" * 64}, "source fingerprint drifted"),
        ({"expected_plan_sha256": "0" * 64}, "explicit plan SHA-256"),
        ({"primary_confirmed": False}, "primary confirmation"),
    ],
)
async def test_sealed_repair_rejects_untrusted_source(test_engine, override, message):
    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
    _repo_id, topic_id = await _seed_empty_repository(factory, "untrusted-repo")
    workplan_old = uuid.uuid4()
    task_old = uuid.uuid4()
    plan = _sealed_plan("untrusted-repo", workplan_old, task_old)
    kwargs = _repair_kwargs(plan) | override

    async with factory() as session:
        with pytest.raises(IdentifierMigrationError, match=message):
            await repair_absent_prederivation_projection(
                session, plan, "untrusted-repo", _repair_unit(plan, topic_id), **kwargs
            )
    async with factory() as session:
        assert await session.get(Workplan, workplan_old) is None
        assert await session.get(Task, task_old) is None


@pytest.mark.asyncio
async def test_sealed_repair_rejects_partial_or_derived_projection_presence(test_engine):
    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
    repo_id, topic_id = await _seed_empty_repository(factory, "partial-repo")
    workplan_old = uuid.uuid4()
    task_old = uuid.uuid4()
    plan = _sealed_plan("partial-repo", workplan_old, task_old)
    unit = _repair_unit(plan, topic_id)

    async with factory() as session:
        session.add(
            Workplan(
                id=workplan_old,
                repo_id=repo_id,
                topic_id=topic_id,
                slug="test-wp-0001",
                title="Migration test",
                status="active",
            )
        )
        await session.commit()
    async with factory() as session:
        with pytest.raises(IdentifierMigrationError, match="partial old projection"):
            await repair_absent_prederivation_projection(
                session, plan, "partial-repo", unit, **_repair_kwargs(plan)
            )

    async with factory() as session:
        await session.execute(text("DELETE FROM workplans WHERE id = :id"), {"id": workplan_old})
        session.add(
            Workplan(
                id=_derived("TEST-WP-0001"),
                repo_id=repo_id,
                topic_id=topic_id,
                slug="derived-conflict",
                title="Derived conflict",
                status="active",
            )
        )
        await session.commit()
    async with factory() as session:
        with pytest.raises(IdentifierMigrationError, match="derived target presence"):
            await repair_absent_prederivation_projection(
                session, plan, "partial-repo", unit, **_repair_kwargs(plan)
            )


@pytest.mark.asyncio
async def test_sealed_repair_http_surface_returns_receipt(client, test_engine):
    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
    repo_id, topic_id = await _seed_empty_repository(factory, "http-repair-repo")
    workplan_old = uuid.uuid4()
    task_old = uuid.uuid4()
    plan = _sealed_plan("http-repair-repo", workplan_old, task_old)
    response = await client.post(
        "/identifier-migrations/sealed-projection-repairs",
        json={
            "plan": plan,
            "repo_slug": "http-repair-repo",
            "unit": _repair_unit(plan, topic_id),
            **_repair_kwargs(plan),
        },
    )
    assert response.status_code == 200, response.text
    receipt = response.json()
    assert receipt["outcome"] == "repaired"
    assert receipt["repository_id"] == str(repo_id)
    assert receipt["task_records"] == [["TEST-WP-0001-T01", str(task_old)]]
    assert (await client.get(f"/workplans/{workplan_old}")).status_code == 200
    assert (await client.get(f"/tasks/{task_old}")).status_code == 200
    assert (await client.get(f"/workplans/{_derived('TEST-WP-0001')}")).status_code == 404
    assert (await client.get(f"/tasks/{_derived('TEST-WP-0001-T01')}")).status_code == 404


@pytest.mark.asyncio
async def test_sealed_repair_http_surface_rejects_missing_confirmation(client, test_engine):
    factory = async_sessionmaker(test_engine, class_=AsyncSession, expire_on_commit=False)
    _repo_id, topic_id = await _seed_empty_repository(factory, "http-reject-repo")
    plan = _sealed_plan("http-reject-repo", uuid.uuid4(), uuid.uuid4())
    payload = {
        "plan": plan,
        "repo_slug": "http-reject-repo",
        "unit": _repair_unit(plan, topic_id),
        **_repair_kwargs(plan),
    }
    payload["primary_confirmed"] = False
    response = await client.post(
        "/identifier-migrations/sealed-projection-repairs", json=payload
    )
    assert response.status_code == 409
    assert "primary confirmation" in response.json()["detail"]
