"""Saved Pipeline operations and owner evidence over the production PostgreSQL boundary."""
from concurrent.futures import ThreadPoolExecutor
from dataclasses import replace
from typing import cast

import pytest

from metallo.adaptors.postgres import Database
from metallo.contexts.pipeline_coordination.application.commands import ReadPipelineQuery
from metallo.contexts.pipeline_coordination.application.ports import AuthorityUnavailable, WorkloadLookupUnavailable
from metallo.contexts.pipeline_coordination.domain.pipeline import ServiceKind
from metallo.foundation.application import ApplicationError
from metallo.foundation.identity import new_id
from metallo.tests.infrastructure.pipeline_fixtures import (
    Authority, Workloads, cancellation, command, counts, database as database, mark_completed, operations,
    save, work,
)
from metallo.tests.pipeline_coordination.application_fixtures import TEST_LIFETIME, accepted


@pytest.mark.parametrize("kind", [None, "application", "huggingface-inference"])
def test_pipeline_start_reload_cancel_uses_production_persistence(database: Database, kind: str | None) -> None:
    saved = save(database, cast(ServiceKind | None, kind))
    request = saved.request
    fresh = operations(database, Authority(), Workloads(request))
    read = ReadPipelineQuery(request.pipeline_id, new_id())
    pipeline = fresh.read(read)
    assert pipeline == accepted(request, TEST_LIFETIME)
    assert pipeline.member_id == request.member_id and pipeline.project_id == request.project_id
    assert pipeline.steps == request.steps and all(step.status == "waiting" for step in pipeline.steps)
    if kind is None:
        assert [step.environment for step in pipeline.steps] == ["k8s", "slurm", "k8s"]
    else:
        assert pipeline.steps[0].service_kind == kind
        assert pipeline.steps[0].advance_when.require_endpoint
    # The start commits one accepted-work item for each Site of its steps (#766).
    sites = len({step.environment for step in pipeline.steps})
    assert counts(database, pipeline.id) == {
        "aggregates": 1, "receipts": 1, "work": 1, "outbox": sites, "realtime": 0, "execution_records": 0,
    }
    initial = work(database, pipeline.id)
    assert initial["pipeline_version"] == 1 and initial["status"] == "pending"
    assert initial["pipeline_kind"] == "pipeline"
    cancel = cancellation(request)
    assert cancel.requested_by_member_id != pipeline.member_id
    result = fresh.cancel(cancel)
    assert result == {"aggregateId": pipeline.id, "version": 2, "status": "cancellation_requested"}
    restored = operations(database, Authority(), Workloads(request)).read(read)
    assert restored.cancellation == cancel.cancellation() and restored.steps == pipeline.steps
    assert (restored.id, restored.member_id, restored.project_id) == (pipeline.id, pipeline.member_id, pipeline.project_id)
    assert work(database, pipeline.id)["pipeline_version"] == 2
    # A requested cancellation releases nothing until the coordinator completes the Pipeline.
    assert counts(database, pipeline.id) == {
        "aggregates": 1, "receipts": 2, "work": 1, "outbox": sites, "realtime": 0, "execution_records": 0,
    }


def test_pipeline_replays_and_rejections_preserve_completed_pending_marker(database: Database) -> None:
    saved = save(database)
    request, handler = saved.request, saved.handler
    initial = {"aggregateId": request.pipeline_id, "version": 1, "status": "waiting"}
    mark_completed(request.pipeline_id)
    marker = work(database, request.pipeline_id)
    saved.workloads.available = False
    assert handler.start(request) == initial
    assert handler.start(replace(request, member_id=new_id()))["rejection"]["code"] == "idempotency_conflict"
    assert handler.start(replace(request, id=new_id()))["rejection"]["code"] == "version_conflict"
    assert work(database, request.pipeline_id) == marker
    cancel = cancellation(request)
    result = handler.cancel(cancel)
    changed = work(database, request.pipeline_id)
    assert changed["status"] == "pending" and changed["pipeline_version"] == 2
    assert changed["available_at"] != marker["available_at"]
    mark_completed(request.pipeline_id)
    marker = work(database, request.pipeline_id)
    assert handler.cancel(cancel) == result
    assert handler.cancel(replace(cancel, reason="Changed reason"))["rejection"]["code"] == "idempotency_conflict"
    assert handler.cancel(replace(cancel, id=new_id()))["rejection"]["code"] == "version_conflict"
    later = replace(cancel, id=new_id(), expected_version=2, requested_by_member_id=new_id(), reason="Another request")
    assert handler.cancel(later) == result
    assert work(database, request.pipeline_id) == marker
    assert handler.read(ReadPipelineQuery(request.pipeline_id, new_id())).cancellation == cancel.cancellation()


@pytest.mark.parametrize("changed", ["member", "project", "revision", "environment", "order"])
def test_pipeline_receipt_covers_identity_order_and_step_material(database: Database, changed: str) -> None:
    saved = save(database)
    request = saved.request
    if changed == "member":
        conflict = replace(request, member_id=new_id())
    elif changed == "project":
        conflict = replace(request, project_id=new_id())
    else:
        first, second, third = request.steps
        if changed == "revision":
            first = replace(first, workload_revision=3)
        elif changed == "environment":
            first = replace(first, environment="slurm")
        else:
            conflict = replace(request, steps=(second, first, third))
        if changed != "order":
            conflict = replace(request, steps=(first, second, third))
    before = work(database, request.pipeline_id)
    assert saved.handler.start(conflict)["rejection"]["code"] == "idempotency_conflict"
    assert work(database, request.pipeline_id) == before
    assert counts(database, request.pipeline_id)["receipts"] == 1


@pytest.mark.parametrize("provider_state", ["restored", "unavailable"])
def test_pipeline_recorded_rejection_replays_through_changed_owner_evidence(
        database: Database, provider_state: str) -> None:
    request = command()
    workloads = Workloads(request)
    valid = dict(workloads.revisions)
    workloads.revisions.clear()
    handler = operations(database, Authority(), workloads)
    result = handler.start(request)
    assert result["rejection"]["code"] == "workload_revision_not_found"
    workloads.revisions = valid
    workloads.available = provider_state != "unavailable"
    assert handler.start(request) == result
    assert handler.start(replace(request, member_id=new_id()))["rejection"]["code"] == "idempotency_conflict"
    assert counts(database, request.pipeline_id) == {
        "aggregates": 0, "receipts": 1, "work": 0, "outbox": 0, "realtime": 0, "execution_records": 0,
    }


@pytest.mark.parametrize("owner", ["authority", "workload"])
def test_pipeline_provider_outage_has_no_recorded_rejection_or_work(database: Database, owner: str) -> None:
    request = command()
    authority, workloads = Authority(), Workloads(request)
    authority.available = owner != "authority"
    workloads.available = owner != "workload"
    handler = operations(database, authority, workloads)
    with pytest.raises(AuthorityUnavailable if owner == "authority" else WorkloadLookupUnavailable):
        handler.start(request)
    assert all(value == 0 for value in counts(database, request.pipeline_id).values())
    authority.available = workloads.available = True
    assert handler.start(request)["status"] == "waiting"


@pytest.mark.parametrize("operation", ["start", "read", "cancel"])
@pytest.mark.parametrize("failure", ["denied", "unavailable"])
def test_pipeline_current_access_is_required_after_saved_outcomes(
        database: Database, operation: str, failure: str) -> None:
    saved = save(database)
    cancel = cancellation(saved.request)
    if operation == "cancel":
        saved.handler.cancel(cancel)
    before, marker = counts(database, saved.request.pipeline_id), work(database, saved.request.pipeline_id)
    saved.authority.allowed = failure != "denied"
    saved.authority.available = failure != "unavailable"
    with pytest.raises(ApplicationError if failure == "denied" else AuthorityUnavailable):
        if operation == "start":
            saved.handler.start(saved.request)
        elif operation == "read":
            saved.handler.read(ReadPipelineQuery(saved.request.pipeline_id, new_id()))
        else:
            saved.handler.cancel(cancel)
    assert counts(database, saved.request.pipeline_id) == before and work(database, saved.request.pipeline_id) == marker


def test_concurrent_pipeline_starts_replay_one_record_and_competing_cancels_fence_versions(database: Database) -> None:
    request = command()
    handler = operations(database, Authority(), Workloads(request))
    with ThreadPoolExecutor(max_workers=4) as workers:
        outcomes = list(workers.map(handler.start, [request] * 4))
    assert all(outcome == outcomes[0] for outcome in outcomes)
    assert counts(database, request.pipeline_id)["receipts"] == 1
    requests = [cancellation(request), cancellation(request)]
    with ThreadPoolExecutor(max_workers=2) as workers:
        cancelled = list(workers.map(handler.cancel, requests))
    assert sum("rejection" not in outcome for outcome in cancelled) == 1
    assert [outcome["rejection"]["code"] for outcome in cancelled if "rejection" in outcome] == ["version_conflict"]
    restored = handler.read(ReadPipelineQuery(request.pipeline_id, new_id()))
    assert restored.version == 2 and restored.cancellation is not None
    assert restored.cancellation.request_id in {value.id for value in requests}
    assert work(database, request.pipeline_id)["pipeline_version"] == 2
    assert counts(database, request.pipeline_id)["receipts"] == 3