from __future__ import annotations

from datetime import UTC, datetime, timedelta
import unittest

from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker

from govoplan_core.auth import ApiPrincipal
from govoplan_core.core.access import (
    CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER,
    PrincipalRef,
)
from govoplan_core.core.automation import AutomationPrincipalResolution
from govoplan_core.core.campaigns import (
    CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION,
    CampaignWorkHandoffInspection,
)
from govoplan_core.core.events import EventTenantRef, PlatformEvent
from govoplan_core.core.recovery import RecoveryCheckpoint, RecoveryOperation
from govoplan_core.core.runtime_coordination import (
    DistributedLease,
    RuntimeIdentity,
    bind_process_runtime_identity,
)
from govoplan_core.db.base import Base
from govoplan_workflow_engine.backend.db.models import (
    WorkflowDefinition,
    WorkflowDefinitionRevision,
    WorkflowInstance,
    WorkflowInstanceEvent,
    WorkflowInstanceStep,
    WorkflowTrigger,
    WorkflowTriggerDelivery,
    WorkflowWaitState,
)
from govoplan_workflow_engine.backend.instance_service import start_instance
from govoplan_workflow_engine.backend.schemas import (
    WorkflowDefinitionCreateRequest,
    WorkflowEdge,
    WorkflowGraph,
    WorkflowInstanceStartRequest,
    WorkflowNode,
)
from govoplan_workflow_engine.backend.service import (
    WorkflowConflictError,
    activate_definition,
    create_definition,
)
from govoplan_workflow_engine.backend.triggers import (
    SqlWorkflowTriggerDispatcher,
    reconcile_definition_triggers,
)


def principal() -> ApiPrincipal:
    return ApiPrincipal(
        principal=PrincipalRef(
            account_id="account-1",
            membership_id="membership-1",
            tenant_id="tenant-1",
            scopes=frozenset({"workflow:instance:start"}),
        ),
        account=object(),
        user=object(),
    )


def runtime_identity() -> RuntimeIdentity:
    return RuntimeIdentity(
        installation_id="workflow-trigger-tests",
        node_id="workflow-trigger-worker",
        incarnation="workflow-trigger-incarnation",
        role="worker",
        software_version="test",
        composition_hash="d" * 64,
    )


class AutomationProvider:
    def resolve_automation_principal(self, _session, *, request):
        return AutomationPrincipalResolution(
            allowed=True,
            principal=principal(),
            granted_scopes=request.grant_scopes,
            provenance={"status": "rechecked"},
        )


class CampaignHandoffProvider:
    def __init__(self) -> None:
        self.allowed = True
        self.status = "completed"
        self.revision = 2
        self.inspections: list[tuple[str, int | None]] = []

    def prepare_handoff(self, _session, _principal, *, request):
        raise AssertionError("The external wait must not create Campaign work.")

    def inspect_handoff(
        self,
        _session,
        _principal,
        *,
        tenant_id,
        assignment_id,
        expected_revision=None,
    ):
        assert tenant_id == "tenant-1"
        self.inspections.append((assignment_id, expected_revision))
        return CampaignWorkHandoffInspection(
            allowed=self.allowed,
            status=self.status,
            assignment_revision=self.revision,
            action_url="/campaigns/campaign-1/work?assignment=assignment-1",
            assignment_ref=f"campaign-work-assignment:assignment-1:r{self.revision}",
            reason=None if self.allowed else "Campaign access was revoked.",
            provenance={"access_rechecked": True},
        )


class Registry:
    def __init__(self) -> None:
        self.provider = AutomationProvider()
        self.campaign = CampaignHandoffProvider()

    def has_capability(self, name: str) -> bool:
        return (
            name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER
            or (
                name == CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION
                and self.campaign is not None
            )
        )

    def capability(self, name: str):
        if not self.has_capability(name):
            raise KeyError(name)
        if name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER:
            return self.provider
        return self.campaign


def graph(start_type: str, *, wait: WorkflowNode | None = None) -> WorkflowGraph:
    start_config: dict[str, object] = {}
    if start_type == "workflow.start.schedule":
        start_config = {"schedule": "interval:60", "timezone": "UTC"}
    elif start_type == "workflow.start.event":
        start_config = {
            "event_type": "case.updated",
            "filter": {"payload": {"state": "ready"}},
        }
    nodes = [WorkflowNode(id="start", type=start_type, config=start_config)]
    edges: list[WorkflowEdge] = []
    previous = "start"
    if wait is not None:
        nodes.append(wait)
        edges.append(WorkflowEdge(id="start-wait", source="start", target=wait.id))
        previous = wait.id
    nodes.append(WorkflowNode(id="end", type="workflow.end.completed"))
    edges.append(
        WorkflowEdge(
            id="to-end",
            source=previous,
            source_port="timed_out" if wait is not None else "output",
            target="end",
        )
    )
    return WorkflowGraph(nodes=nodes, edges=edges)


def external_handoff_graph(*, timeout_after: str = "1h") -> WorkflowGraph:
    return WorkflowGraph(
        nodes=[
            WorkflowNode(id="start", type="workflow.start.manual"),
            WorkflowNode(
                id="campaign_work",
                type="workflow.external_handoff",
                label="Complete Campaign review",
                config={
                    "provider_capability": CAPABILITY_CAMPAIGNS_WORK_ORCHESTRATION,
                    "event_type": "campaign.work.changed",
                    "event_filter": {
                        "payload": {"assignment_id": "$input.assignment_id"}
                    },
                    "outcome_path": "payload.outcome",
                    "terminal_outcomes": {
                        "completed": "completed",
                        "rejected": "rejected",
                        "cancelled": "cancelled",
                    },
                    "observed_outcomes": ["assigned", "accepted", "reassigned"],
                    "external_id": "$input.assignment_id",
                    "expected_revision": "$input.assignment_revision",
                    "action_url": "$input.action_url",
                    "immutable_ref": "$input.assignment_ref",
                    "optional_capabilities": "$input.optional_capabilities",
                    "timeout_after": timeout_after,
                    "view_surface_ids": ["campaigns.page.work"],
                },
            ),
            WorkflowNode(id="completed", type="workflow.end.completed"),
            WorkflowNode(
                id="rejected",
                type="workflow.end.cancelled",
                config={"reason": "Campaign work rejected"},
            ),
            WorkflowNode(
                id="cancelled",
                type="workflow.end.cancelled",
                config={"reason": "Campaign work cancelled"},
            ),
            WorkflowNode(
                id="timed_out",
                type="workflow.end.cancelled",
                config={"reason": "Campaign work timed out"},
            ),
        ],
        edges=[
            WorkflowEdge(id="start-work", source="start", target="campaign_work"),
            WorkflowEdge(
                id="work-completed",
                source="campaign_work",
                source_port="completed",
                target="completed",
            ),
            WorkflowEdge(
                id="work-rejected",
                source="campaign_work",
                source_port="rejected",
                target="rejected",
            ),
            WorkflowEdge(
                id="work-cancelled",
                source="campaign_work",
                source_port="cancelled",
                target="cancelled",
            ),
            WorkflowEdge(
                id="work-timeout",
                source="campaign_work",
                source_port="timed_out",
                target="timed_out",
            ),
        ],
    )


class WorkflowTriggerTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite:///:memory:")
        self.tables = [
            DistributedLease.__table__,
            RecoveryOperation.__table__,
            RecoveryCheckpoint.__table__,
            WorkflowDefinition.__table__,
            WorkflowDefinitionRevision.__table__,
            WorkflowInstance.__table__,
            WorkflowInstanceStep.__table__,
            WorkflowInstanceEvent.__table__,
            WorkflowTrigger.__table__,
            WorkflowTriggerDelivery.__table__,
            WorkflowWaitState.__table__,
        ]
        Base.metadata.create_all(self.engine, tables=self.tables)
        self.Session = sessionmaker(bind=self.engine)
        self.session: Session = self.Session()
        bind_process_runtime_identity(runtime_identity())
        self.registry = Registry()

    def tearDown(self) -> None:
        bind_process_runtime_identity(None)
        self.session.close()
        Base.metadata.drop_all(self.engine, tables=list(reversed(self.tables)))
        self.engine.dispose()

    def _definition(
        self,
        definition_graph: WorkflowGraph,
        *,
        automation: bool,
    ) -> WorkflowDefinition:
        definition = create_definition(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=WorkflowDefinitionCreateRequest(
                name="Trigger test",
                graph=definition_graph,
                allow_automation=automation,
            ),
        )
        activate_definition(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
        )
        return definition

    def _external_input(self) -> dict[str, object]:
        return {
            "assignment_id": "assignment-1",
            "assignment_revision": 1,
            "action_url": "/campaigns/campaign-1/work?assignment=assignment-1",
            "assignment_ref": "campaign-work-assignment:assignment-1:r1",
            "optional_capabilities": {
                "tasks": False,
                "notifications": False,
            },
        }

    def test_schedule_registration_dispatch_and_replay_are_durable(self) -> None:
        definition = self._definition(
            graph("workflow.start.schedule"),
            automation=True,
        )
        summary = reconcile_definition_triggers(
            self.session,
            definition=definition,
            principal=principal(),
            registry=self.registry,
        )
        trigger = self.session.scalar(select(WorkflowTrigger))
        assert trigger is not None
        trigger.next_fire_at = datetime.now(tz=UTC) - timedelta(seconds=1)

        result = SqlWorkflowTriggerDispatcher(registry=self.registry).dispatch_due(
            self.session, now=datetime.now(tz=UTC)
        )

        self.assertEqual({"created": 1, "updated": 0, "disabled": 0}, summary)
        self.assertEqual(1, result["started"])
        delivery = self.session.scalar(select(WorkflowTriggerDelivery))
        assert delivery is not None
        self.assertEqual("succeeded", delivery.status)
        instance = self.session.get(WorkflowInstance, delivery.instance_id)
        assert instance is not None
        self.assertEqual("schedule", instance.start_origin)
        self.assertEqual("completed", instance.status)

        replay = SqlWorkflowTriggerDispatcher(registry=self.registry).dispatch_due(
            self.session, now=datetime.now(tz=UTC)
        )
        self.assertEqual(0, replay["started"])
        self.assertEqual(1, self.session.query(WorkflowInstance).count())

    def test_event_filter_queues_only_matching_event(self) -> None:
        definition = self._definition(
            graph("workflow.start.event"),
            automation=True,
        )
        reconcile_definition_triggers(
            self.session,
            definition=definition,
            principal=principal(),
            registry=self.registry,
        )
        dispatcher = SqlWorkflowTriggerDispatcher(registry=self.registry)
        ignored = dispatcher.ingest_event(
            self.session,
            event=PlatformEvent(
                type="case.updated",
                module_id="cases",
                tenant=EventTenantRef(id="tenant-1"),
                payload={"state": "draft"},
            ),
        )
        accepted = dispatcher.ingest_event(
            self.session,
            event=PlatformEvent(
                type="case.updated",
                module_id="cases",
                tenant=EventTenantRef(id="tenant-1"),
                payload={"state": "ready"},
            ),
        )
        result = dispatcher.dispatch_due(self.session)

        self.assertEqual(0, ignored["trigger_deliveries"])
        self.assertEqual(1, accepted["trigger_deliveries"])
        self.assertEqual(1, result["started"])

    def test_duration_wait_resumes_through_persisted_timer(self) -> None:
        definition = self._definition(
            graph(
                "workflow.start.manual",
                wait=WorkflowNode(
                    id="wait",
                    type="workflow.wait",
                    config={"mode": "duration", "value": "1"},
                ),
            ),
            automation=False,
        )
        instance, replayed = start_instance(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
            payload=WorkflowInstanceStartRequest(idempotency_key="wait-1"),
        )
        state = self.session.scalar(select(WorkflowWaitState))
        assert state is not None

        result = SqlWorkflowTriggerDispatcher(registry=self.registry).dispatch_due(
            self.session,
            now=datetime.now(tz=UTC) + timedelta(seconds=2),
        )

        self.assertFalse(replayed)
        self.assertEqual("timed_out", state.status)
        self.assertEqual(1, result["waits_timed_out"])
        self.assertEqual("completed", instance.status)

    def test_external_handoff_observes_duplicate_safe_events_and_resumes(self) -> None:
        definition = self._definition(
            external_handoff_graph(),
            automation=False,
        )
        instance, _replayed = start_instance(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
            payload=WorkflowInstanceStartRequest(
                idempotency_key="campaign-handoff-1",
                input=self._external_input(),
            ),
        )
        step = self.session.get(WorkflowInstanceStep, instance.current_step_id)
        state = self.session.scalar(select(WorkflowWaitState))
        assert step is not None and state is not None
        self.assertEqual("external_handoff", state.mode)
        self.assertEqual("assigned", step.handoff["state"])
        self.assertEqual(
            ["notifications", "tasks"],
            step.handoff["unavailable_optional_capabilities"],
        )
        dispatcher = SqlWorkflowTriggerDispatcher(registry=self.registry)
        accepted = PlatformEvent(
            type="campaign.work.changed",
            module_id="campaigns",
            event_id="campaign-event-accepted",
            tenant=EventTenantRef(id="tenant-1"),
            payload={
                "assignment_id": "assignment-1",
                "assignment_revision": 2,
                "outcome": "accepted",
            },
        )

        observed = dispatcher.ingest_event(self.session, event=accepted)
        duplicate = dispatcher.ingest_event(self.session, event=accepted)

        self.assertEqual(0, observed["waits_triggered"])
        self.assertEqual(0, duplicate["waits_triggered"])
        self.assertEqual("accepted", step.handoff["state"])
        self.assertEqual(
            1,
            self.session.query(WorkflowInstanceEvent)
            .filter(
                WorkflowInstanceEvent.kind
                == "workflow.external_handoff.observed"
            )
            .count(),
        )

        completed = dispatcher.ingest_event(
            self.session,
            event=PlatformEvent(
                type="campaign.work.changed",
                module_id="campaigns",
                event_id="campaign-event-completed",
                tenant=EventTenantRef(id="tenant-1"),
                payload={
                    "assignment_id": "assignment-1",
                    "assignment_revision": 2,
                    "outcome": "completed",
                },
            ),
        )
        result = dispatcher.dispatch_due(self.session)

        self.assertEqual(1, completed["waits_triggered"])
        self.assertEqual(1, result["waits_resumed"])
        self.assertEqual("completed", instance.status)
        self.assertEqual(
            [("assignment-1", 2)],
            self.registry.campaign.inspections,
        )
        self.assertEqual(
            "campaign-work-assignment:assignment-1:r2",
            step.external_ref,
        )

    def test_external_handoff_revoked_access_blocks_until_rechecked(self) -> None:
        definition = self._definition(
            external_handoff_graph(),
            automation=False,
        )
        instance, _replayed = start_instance(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
            payload=WorkflowInstanceStartRequest(
                idempotency_key="campaign-handoff-revoked",
                input=self._external_input(),
            ),
        )
        dispatcher = SqlWorkflowTriggerDispatcher(registry=self.registry)
        dispatcher.ingest_event(
            self.session,
            event=PlatformEvent(
                type="campaign.work.changed",
                module_id="campaigns",
                tenant=EventTenantRef(id="tenant-1"),
                payload={
                    "assignment_id": "assignment-1",
                    "assignment_revision": 2,
                    "outcome": "completed",
                },
            ),
        )
        self.registry.campaign.allowed = False

        blocked = dispatcher.dispatch_due(self.session)
        step = self.session.get(WorkflowInstanceStep, instance.current_step_id)
        assert step is not None

        self.assertEqual(1, blocked["waits_skipped"])
        self.assertEqual("waiting", instance.status)
        self.assertEqual("blocked", step.handoff["state"])
        self.assertIn("revoked", str(step.handoff["message"]))

        self.registry.campaign.allowed = True
        resumed = dispatcher.dispatch_due(self.session)

        self.assertEqual(1, resumed["waits_resumed"])
        self.assertEqual("completed", instance.status)

    def test_external_handoff_timeout_and_optional_provider_absence(self) -> None:
        definition = self._definition(
            external_handoff_graph(timeout_after="1s"),
            automation=False,
        )
        instance, _replayed = start_instance(
            self.session,
            tenant_id="tenant-1",
            definition_id=definition.id,
            actor_id="account-1",
            principal=principal(),
            registry=self.registry,
            payload=WorkflowInstanceStartRequest(
                idempotency_key="campaign-handoff-timeout",
                input=self._external_input(),
            ),
        )

        result = SqlWorkflowTriggerDispatcher(
            registry=self.registry
        ).dispatch_due(
            self.session,
            now=datetime.now(tz=UTC) + timedelta(seconds=2),
        )

        self.assertEqual(1, result["waits_timed_out"])
        self.assertEqual("cancelled", instance.status)

        unavailable = self._definition(
            external_handoff_graph(),
            automation=False,
        )
        self.registry.campaign = None  # type: ignore[assignment]
        with self.assertRaisesRegex(WorkflowConflictError, "is not available"):
            start_instance(
                self.session,
                tenant_id="tenant-1",
                definition_id=unavailable.id,
                actor_id="account-1",
                principal=principal(),
                registry=self.registry,
                payload=WorkflowInstanceStartRequest(
                    idempotency_key="campaign-handoff-unavailable",
                    input=self._external_input(),
                ),
            )

    def test_parent_workflow_outcome_starts_pinned_child(self) -> None:
        parent = self._definition(
            graph("workflow.start.manual"),
            automation=False,
        )
        child_graph = WorkflowGraph(
            nodes=[
                WorkflowNode(
                    id="start",
                    type="workflow.start.workflow",
                    config={
                        "parent_definition_ref": (f"workflow-definition:{parent.id}"),
                        "parent_outcome": "completed",
                        "input_mapping": {"parent_id": "$event.payload.instance_id"},
                    },
                ),
                WorkflowNode(id="end", type="workflow.end.completed"),
            ],
            edges=[WorkflowEdge(id="finish", source="start", target="end")],
        )
        child = self._definition(child_graph, automation=True)
        reconcile_definition_triggers(
            self.session,
            definition=child,
            principal=principal(),
            registry=self.registry,
        )
        dispatcher = SqlWorkflowTriggerDispatcher(registry=self.registry)
        event = PlatformEvent(
            type="workflow.instance.completed",
            module_id="workflow_engine",
            tenant=EventTenantRef(id="tenant-1"),
            payload={
                "instance_id": "parent-instance-1",
                "definition_id": parent.id,
            },
        )

        queued = dispatcher.ingest_event(self.session, event=event)
        result = dispatcher.dispatch_due(self.session)

        self.assertEqual(1, queued["trigger_deliveries"])
        self.assertEqual(1, result["started"])
        instance = self.session.scalar(
            select(WorkflowInstance).where(WorkflowInstance.definition_id == child.id)
        )
        assert instance is not None
        self.assertEqual("parent_workflow", instance.start_origin)
        self.assertEqual(
            "parent-instance-1",
            instance.input_["parent_id"],
        )


if __name__ == "__main__":
    unittest.main()