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.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 (
    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 Registry:
    def __init__(self) -> None:
        self.provider = AutomationProvider()

    def has_capability(self, name: str) -> bool:
        return name == CAPABILITY_AUTH_AUTOMATION_PRINCIPAL_PROVIDER

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


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)


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 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_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()