from __future__ import annotations

from datetime import datetime, timedelta, timezone
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 PrincipalRef
from govoplan_core.core.automation import AutomationPrincipalResolution
from govoplan_core.core.policy import PolicyDecision
from govoplan_core.db.base import Base, utcnow
from govoplan_dataflow.backend.db.models import (
    DataflowPipeline,
    DataflowPipelineRevision,
    DataflowRun,
    DataflowTrigger,
    DataflowTriggerDelivery,
)
from govoplan_dataflow.backend.schemas import (
    DataflowEventDeliveryRequest,
    DataflowTriggerCreateRequest,
    DataflowTriggerEvent,
    DataflowTriggerSchedule,
    PipelineCreateRequest,
    PipelineDeriveRequest,
    PipelineRebaseRequest,
    PipelineUpdateRequest,
)
from govoplan_dataflow.backend.service import (
    DataflowConflictError,
    DataflowValidationError,
    create_pipeline,
    derive_pipeline,
    pipeline_response,
    rebase_pipeline,
    start_pipeline_run,
    update_pipeline,
)
from govoplan_core.core.dataflows import DataflowRunRequest
from govoplan_dataflow.backend.triggers import (
    create_trigger,
    dispatch_due_triggers,
    ingest_event_delivery,
)


POLICY_CAPABILITY = "policy.definitionGovernance"
AUTOMATION_CAPABILITY = "auth.automationPrincipalProvider"


def sample_graph(*, minimum: int = 10):
    from test_service import sample_graph as build_graph

    return build_graph(minimum=minimum)


def reusable_graph(*, minimum: int = 10):
    graph = sample_graph(minimum=minimum)
    return graph.model_copy(
        update={
            "nodes": [
                (
                    node.model_copy(
                        update={
                            "config": {
                                **node.config,
                                "input_binding": True,
                            }
                        },
                        deep=True,
                    )
                    if node.id == "source"
                    else node
                )
                for node in graph.nodes
            ]
        },
        deep=True,
    )


def referencing_graph(
    source_ref: str,
    source_revision: int,
    *,
    omit_amount: bool = False,
    input_binding: bool = False,
):
    graph = sample_graph()
    return graph.model_copy(
        update={
            "nodes": [
                (
                    node.model_copy(
                        update={
                            "config": {
                                **node.config,
                                "rows": (
                                    [{"id": 1}]
                                    if omit_amount
                                    else node.config["rows"]
                                ),
                                "input_binding": input_binding,
                            }
                        },
                        deep=True,
                    )
                    if node.id == "source"
                    else node.model_copy(
                        update={
                            "type": "subflow",
                            "label": "Governed reusable flow",
                            "config": {
                                "template_ref": source_ref,
                                "template_version": str(source_revision),
                                "parameters": {},
                                "graph": sample_graph(
                                    minimum=999
                                ).model_dump(mode="json"),
                            },
                        },
                        deep=True,
                    )
                    if node.id == "filter"
                    else node
                )
                for node in graph.nodes
            ]
        },
        deep=True,
    )


def principal() -> ApiPrincipal:
    return ApiPrincipal(
        principal=PrincipalRef(
            account_id="account-1",
            membership_id="membership-1",
            tenant_id="tenant-1",
            scopes=frozenset(
                {
                    "dataflow:pipeline:run",
                    "dataflow:trigger:read",
                    "dataflow:trigger:write",
                }
            ),
        ),
        account=object(),
        user=object(),
    )


class AllowDefinitionPolicy:
    def resolve_definition_action(self, *, request):
        allowed = not (
            request.definition_kind == "template"
            and request.action in {"run", "automate"}
        )
        if request.action == "automate":
            allowed = allowed and request.allow_automation
        if request.action == "run":
            allowed = (
                allowed
                and request.allow_run
                and request.status == "active"
            )
        if request.action in {"reuse", "derive"}:
            allowed = allowed and request.allow_reuse
        return PolicyDecision(
            allowed=allowed,
            reason=None if allowed else "Definition action denied.",
        )


class CurrentPrincipalProvider:
    def __init__(self, actor: ApiPrincipal, *, allowed: bool = True) -> None:
        self.actor = actor
        self.allowed = allowed
        self.last_request = None

    def resolve_automation_principal(self, _session, *, request):
        self.last_request = request
        return AutomationPrincipalResolution(
            allowed=self.allowed,
            principal=self.actor if self.allowed else None,
            reason=None if self.allowed else "Owner authorization revoked.",
            granted_scopes=tuple(request.grant_scopes) if self.allowed else (),
            missing_scopes=() if self.allowed else tuple(request.grant_scopes),
            provenance={
                "status": "current" if self.allowed else "revoked",
                "event_actor": request.context.get("event_actor"),
            },
        )


class Registry:
    def __init__(
        self,
        actor: ApiPrincipal,
        *,
        automation_allowed: bool = True,
    ) -> None:
        self.automation_provider = CurrentPrincipalProvider(
            actor,
            allowed=automation_allowed,
        )
        self.capabilities = {
            POLICY_CAPABILITY: AllowDefinitionPolicy(),
            AUTOMATION_CAPABILITY: self.automation_provider,
        }

    def has_capability(self, name: str) -> bool:
        return name in self.capabilities

    def capability(self, name: str):
        return self.capabilities[name]


class DataflowTriggerTests(unittest.TestCase):
    def setUp(self) -> None:
        self.engine = create_engine("sqlite:///:memory:")
        Base.metadata.create_all(
            self.engine,
            tables=[
                DataflowPipeline.__table__,
                DataflowPipelineRevision.__table__,
                DataflowRun.__table__,
                DataflowTrigger.__table__,
                DataflowTriggerDelivery.__table__,
            ],
        )
        self.Session = sessionmaker(bind=self.engine)
        self.session: Session = self.Session()
        self.actor = principal()
        self.registry = Registry(self.actor)
        self.pipeline = create_pipeline(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=PipelineCreateRequest(
                name="Automated comparison",
                status="active",
                allow_automation=True,
                graph=sample_graph(),
            ),
        )
        self.session.commit()

    def tearDown(self) -> None:
        self.session.close()
        Base.metadata.drop_all(
            self.engine,
            tables=[
                DataflowTriggerDelivery.__table__,
                DataflowTrigger.__table__,
                DataflowRun.__table__,
                DataflowPipelineRevision.__table__,
                DataflowPipeline.__table__,
            ],
        )
        self.engine.dispose()

    def _event_trigger(self) -> DataflowTrigger:
        trigger = create_trigger(
            self.session,
            tenant_id="tenant-1",
            pipeline_id=self.pipeline.id,
            actor_id="account-1",
            principal=self.actor,
            registry=self.registry,
            payload=DataflowTriggerCreateRequest(
                name="Monthly file arrived",
                kind="event",
                enabled=True,
                event=DataflowTriggerEvent(
                    event_type="files.upload.completed",
                    module_id="files",
                    filters={"folder_id": "incoming"},
                ),
            ),
        )
        self.session.commit()
        return trigger

    def test_interval_delivery_runs_pinned_revision_with_provenance(self) -> None:
        trigger = create_trigger(
            self.session,
            tenant_id="tenant-1",
            pipeline_id=self.pipeline.id,
            actor_id="account-1",
            principal=self.actor,
            registry=self.registry,
            payload=DataflowTriggerCreateRequest(
                name="Hourly",
                kind="interval",
                enabled=True,
                schedule=DataflowTriggerSchedule(
                    interval_seconds=3600,
                    timezone="Europe/Berlin",
                ),
            ),
        )
        now = utcnow()
        trigger.next_fire_at = now - timedelta(minutes=1)
        self.session.commit()

        result = dispatch_due_triggers(
            self.session,
            registry=self.registry,
            now=now,
        )
        self.session.commit()

        delivery = self.session.scalar(
            select(DataflowTriggerDelivery).where(
                DataflowTriggerDelivery.trigger_id == trigger.id
            )
        )
        run = self.session.get(DataflowRun, delivery.run_id)
        self.assertEqual(1, result.queued)
        self.assertEqual(1, result.succeeded)
        self.assertEqual("succeeded", delivery.status)
        self.assertEqual("schedule", run.invocation_kind)
        self.assertEqual(trigger.id, run.trigger_id)
        self.assertEqual(delivery.id, run.trigger_delivery_id)
        self.assertEqual("current", delivery.authorization_provenance["status"])

    def test_skip_catch_up_records_missed_interval_without_running(self) -> None:
        trigger = create_trigger(
            self.session,
            tenant_id="tenant-1",
            pipeline_id=self.pipeline.id,
            actor_id="account-1",
            principal=self.actor,
            registry=self.registry,
            payload=DataflowTriggerCreateRequest(
                name="Skip stale intervals",
                kind="interval",
                enabled=True,
                catch_up_policy="skip",
                schedule=DataflowTriggerSchedule(
                    anchor_at=datetime(
                        2000,
                        1,
                        1,
                        tzinfo=timezone.utc,
                    ),
                    interval_seconds=60,
                ),
            ),
        )
        now = utcnow()
        trigger.next_fire_at = now - timedelta(minutes=3)
        self.session.commit()

        result = dispatch_due_triggers(
            self.session,
            registry=self.registry,
            now=now,
        )
        self.session.commit()

        delivery = self.session.scalar(
            select(DataflowTriggerDelivery).where(
                DataflowTriggerDelivery.trigger_id == trigger.id
            )
        )
        self.assertEqual(0, result.queued)
        self.assertEqual(1, result.skipped)
        self.assertEqual("skipped", delivery.status)
        self.assertIsNone(delivery.run_id)
        next_fire_at = trigger.next_fire_at
        self.assertIsNotNone(next_fire_at)
        if next_fire_at.tzinfo is None:
            next_fire_at = next_fire_at.replace(tzinfo=timezone.utc)
        self.assertGreater(next_fire_at, now)

    def test_disabled_trigger_can_be_prepared_for_a_draft(self) -> None:
        self.pipeline.status = "draft"
        self.session.flush()

        trigger = create_trigger(
            self.session,
            tenant_id="tenant-1",
            pipeline_id=self.pipeline.id,
            actor_id="account-1",
            principal=self.actor,
            registry=self.registry,
            payload=DataflowTriggerCreateRequest(
                name="Prepared event",
                kind="event",
                enabled=False,
                event=DataflowTriggerEvent(
                    event_type="files.upload.completed",
                ),
            ),
        )

        self.assertEqual("disabled", trigger.status)
        self.assertIsNone(trigger.next_fire_at)

    def test_event_ingress_is_filtered_and_idempotent(self) -> None:
        trigger = self._event_trigger()
        event = DataflowEventDeliveryRequest(
            event_id="event-1",
            type="files.upload.completed",
            module_id="files",
            payload={"folder_id": "incoming", "file_id": "file-1"},
            occurred_at=utcnow(),
            actor={
                "type": "user",
                "id": "account-2",
                "label": "Uploader",
            },
        )

        self.assertEqual(
            1,
            ingest_event_delivery(
                self.session,
                tenant_id="tenant-1",
                event=event,
            ),
        )
        self.assertEqual(
            0,
            ingest_event_delivery(
                self.session,
                tenant_id="tenant-1",
                event=event,
            ),
        )
        result = dispatch_due_triggers(
            self.session,
            registry=self.registry,
        )
        self.session.commit()

        delivery = self.session.scalar(
            select(DataflowTriggerDelivery).where(
                DataflowTriggerDelivery.trigger_id == trigger.id
            )
        )
        run = self.session.get(DataflowRun, delivery.run_id)
        self.assertEqual(1, result.succeeded)
        self.assertEqual("event", run.invocation_kind)
        self.assertEqual("event-1", run.request_["invocation"]["event_id"])
        self.assertEqual(
            {
                "type": "user",
                "id": "account-2",
                "label": "Uploader",
            },
            self.registry.automation_provider.last_request.context[
                "event_actor"
            ],
        )
        self.assertEqual(
            "account-2",
            run.request_["invocation"]["metadata"][
                "authorization_provenance"
            ]["event_actor"]["id"],
        )

    def test_revoked_owner_blocks_delivery_before_execution(self) -> None:
        trigger = self._event_trigger()
        ingest_event_delivery(
            self.session,
            tenant_id="tenant-1",
            event=DataflowEventDeliveryRequest(
                event_id="event-revoked",
                type="files.upload.completed",
                module_id="files",
                payload={"folder_id": "incoming"},
                occurred_at=utcnow(),
            ),
        )

        result = dispatch_due_triggers(
            self.session,
            registry=Registry(self.actor, automation_allowed=False),
        )
        self.session.commit()

        delivery = self.session.scalar(
            select(DataflowTriggerDelivery).where(
                DataflowTriggerDelivery.trigger_id == trigger.id
            )
        )
        self.assertEqual(1, result.blocked)
        self.assertEqual("blocked", delivery.status)
        self.assertIn("revoked", delivery.error)
        self.assertIsNone(delivery.run_id)

    def test_restricted_event_requires_durable_bridge(self) -> None:
        self._event_trigger()
        with self.assertRaises(DataflowConflictError):
            ingest_event_delivery(
                self.session,
                tenant_id="tenant-1",
                event=DataflowEventDeliveryRequest(
                    event_id="event-secret",
                    type="files.upload.completed",
                    module_id="files",
                    payload={"folder_id": "incoming"},
                    occurred_at=utcnow(),
                    classification="restricted",
                ),
            )

    def test_template_derivation_pins_source_and_template_cannot_run(self) -> None:
        template = create_pipeline(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=PipelineCreateRequest(
                name="System import template",
                graph=sample_graph(),
                scope_type="system",
                definition_kind="template",
                inherit_to_lower_scopes=True,
                allow_reuse=True,
            ),
        )
        derived = derive_pipeline(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            principal=self.actor,
            registry=self.registry,
            source_pipeline_id=template.id,
            payload=PipelineDeriveRequest(
                name="Tenant monthly import",
                definition_kind="flow",
                allow_run=True,
            ),
        )
        self.session.commit()

        self.assertEqual(template.id, derived.derived_from_pipeline_id)
        self.assertEqual(1, derived.derived_from_revision)
        self.assertEqual(
            template.revisions[0].content_hash,
            derived.derived_from_hash,
        )
        self.assertEqual(
            "system",
            derived.derivation_provenance["source_scope"]["scope_type"],
        )
        with self.assertRaises(DataflowConflictError):
            start_pipeline_run(
                self.session,
                tenant_id="tenant-1",
                actor_id="account-1",
                principal=self.actor,
                registry=self.registry,
                request=DataflowRunRequest(
                    pipeline_ref=f"pipeline:{template.id}",
                    revision=1,
                    idempotency_key="template-must-not-run",
                ),
            )

    def test_reusable_reference_is_policy_resolved_with_typed_contracts(
        self,
    ) -> None:
        template = create_pipeline(
            self.session,
            tenant_id="tenant-1",
            actor_id="template-author",
            payload=PipelineCreateRequest(
                name="Typed reusable filter",
                graph=reusable_graph(minimum=10),
                definition_kind="template",
                allow_reuse=True,
            ),
        )
        consumer = create_pipeline(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            principal=self.actor,
            registry=self.registry,
            payload=PipelineCreateRequest(
                name="Resolved consumer",
                graph=referencing_graph(f"pipeline:{template.id}", 1),
            ),
        )
        self.session.commit()

        stored = consumer.revisions[0].graph
        subflow = next(
            node for node in stored["nodes"] if node["id"] == "filter"
        )
        self.assertEqual(template.revisions[0].content_hash, subflow["config"]["template_hash"])
        self.assertEqual(
            10,
            next(
                node
                for node in subflow["config"]["graph"]["nodes"]
                if node["id"] == "filter"
            )["config"]["value"],
        )
        self.assertEqual(
            {"id", "amount"},
            {
                field["name"]
                for field in subflow["config"]["input_schema"]
            },
        )
        self.assertEqual(
            {"id", "amount"},
            {
                field["name"]
                for field in subflow["config"]["output_schema"]
            },
        )
        self.assertTrue(
            subflow["config"]["reference_provenance"][
                "policy_decision"
            ]["allowed"]
        )

        with self.assertRaises(DataflowValidationError):
            create_pipeline(
                self.session,
                tenant_id="tenant-1",
                actor_id="account-1",
                principal=self.actor,
                registry=self.registry,
                payload=PipelineCreateRequest(
                    name="Incompatible consumer",
                    graph=referencing_graph(
                        f"pipeline:{template.id}",
                        1,
                        omit_amount=True,
                    ),
                ),
            )

    def test_reusable_reference_cycles_are_rejected_across_revisions(
        self,
    ) -> None:
        left = create_pipeline(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=PipelineCreateRequest(
                name="Left template",
                graph=reusable_graph(),
                definition_kind="template",
                allow_reuse=True,
            ),
        )
        right = create_pipeline(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=PipelineCreateRequest(
                name="Right template",
                graph=reusable_graph(),
                definition_kind="template",
                allow_reuse=True,
            ),
        )
        self.session.flush()
        update_pipeline(
            self.session,
            tenant_id="tenant-1",
            pipeline_id=left.id,
            actor_id="account-1",
            principal=self.actor,
            registry=self.registry,
            payload=PipelineUpdateRequest(
                name=left.name,
                graph=referencing_graph(
                    f"pipeline:{right.id}",
                    1,
                    input_binding=True,
                ),
                status="draft",
                expected_revision=1,
                definition_kind="template",
                allow_reuse=True,
            ),
        )

        with self.assertRaisesRegex(DataflowConflictError, "cannot reference itself"):
            update_pipeline(
                self.session,
                tenant_id="tenant-1",
                pipeline_id=right.id,
                actor_id="account-1",
                principal=self.actor,
                registry=self.registry,
                payload=PipelineUpdateRequest(
                    name=right.name,
                    graph=referencing_graph(
                        f"pipeline:{left.id}",
                        2,
                        input_binding=True,
                    ),
                    status="draft",
                    expected_revision=1,
                    definition_kind="template",
                    allow_reuse=True,
                ),
            )

    def test_derived_limits_cannot_be_broadened_transitively(self) -> None:
        template = create_pipeline(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=PipelineCreateRequest(
                name="Restricted template",
                graph=sample_graph(),
                definition_kind="template",
                allow_reuse=True,
                allow_automation=False,
                inherit_to_lower_scopes=False,
            ),
        )
        derived = derive_pipeline(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            principal=self.actor,
            registry=self.registry,
            source_pipeline_id=template.id,
            payload=PipelineDeriveRequest(
                name="Tenant flow",
                allow_reuse=True,
                allow_automation=True,
                inherit_to_lower_scopes=True,
            ),
        )
        self.assertFalse(derived.allow_automation)
        self.assertFalse(derived.inherit_to_lower_scopes)

        update_pipeline(
            self.session,
            tenant_id="tenant-1",
            pipeline_id=derived.id,
            actor_id="account-1",
            payload=PipelineUpdateRequest(
                name=derived.name,
                graph=sample_graph(),
                status="draft",
                expected_revision=1,
                allow_reuse=True,
                allow_automation=True,
                inherit_to_lower_scopes=True,
            ),
        )
        grandchild = derive_pipeline(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            principal=self.actor,
            registry=self.registry,
            source_pipeline_id=derived.id,
            payload=PipelineDeriveRequest(
                name="User flow",
                scope_type="user",
                scope_id="membership-1",
                allow_automation=True,
                inherit_to_lower_scopes=True,
            ),
        )

        self.assertFalse(derived.allow_automation)
        self.assertFalse(derived.inherit_to_lower_scopes)
        self.assertFalse(grandchild.allow_automation)
        self.assertFalse(grandchild.inherit_to_lower_scopes)

    def test_source_update_is_detected_and_rebased_as_reviewed_revision(
        self,
    ) -> None:
        template = create_pipeline(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            payload=PipelineCreateRequest(
                name="Reusable import",
                graph=sample_graph(),
                definition_kind="template",
                allow_reuse=True,
                allow_automation=True,
            ),
        )
        derived = derive_pipeline(
            self.session,
            tenant_id="tenant-1",
            actor_id="account-1",
            principal=self.actor,
            registry=self.registry,
            source_pipeline_id=template.id,
            payload=PipelineDeriveRequest(
                name="Tenant import",
                allow_run=True,
                allow_automation=True,
            ),
        )
        self.session.commit()

        before = pipeline_response(
            self.session,
            derived,
            principal=self.actor,
            registry=self.registry,
        )
        self.assertTrue(before.governance.source_available)
        self.assertFalse(before.governance.update_available)
        original_child_hash = derived.revisions[0].content_hash

        update_pipeline(
            self.session,
            tenant_id="tenant-1",
            pipeline_id=template.id,
            actor_id="template-author",
            payload=PipelineUpdateRequest(
                name=template.name,
                graph=sample_graph(minimum=20),
                status="draft",
                expected_revision=1,
                definition_kind="template",
                allow_reuse=True,
                allow_automation=True,
            ),
        )
        self.session.commit()
        source_hash = template.revisions[-1].content_hash

        available = pipeline_response(
            self.session,
            derived,
            principal=self.actor,
            registry=self.registry,
        )
        self.assertTrue(available.governance.update_available)
        self.assertEqual(2, available.governance.source_current_revision)
        self.assertEqual(source_hash, available.governance.source_current_hash)
        self.assertEqual(original_child_hash, derived.revisions[0].content_hash)

        with self.assertRaisesRegex(DataflowConflictError, "source hash"):
            rebase_pipeline(
                self.session,
                tenant_id="tenant-1",
                pipeline_id=derived.id,
                actor_id="reviewer-1",
                principal=self.actor,
                registry=self.registry,
                payload=PipelineRebaseRequest(
                    expected_revision=1,
                    source_revision=2,
                    source_hash="0" * 64,
                    reason="Reviewed the changed filter threshold.",
                ),
            )

        rebased = rebase_pipeline(
            self.session,
            tenant_id="tenant-1",
            pipeline_id=derived.id,
            actor_id="reviewer-1",
            principal=self.actor,
            registry=self.registry,
            payload=PipelineRebaseRequest(
                expected_revision=1,
                source_revision=2,
                source_hash=source_hash,
                reason="Reviewed the changed filter threshold.",
            ),
        )
        self.session.commit()

        self.assertEqual(2, rebased.current_revision)
        self.assertEqual("draft", rebased.status)
        self.assertEqual(2, rebased.derived_from_revision)
        self.assertEqual(source_hash, rebased.derived_from_hash)
        self.assertEqual(source_hash, rebased.revisions[-1].content_hash)
        self.assertEqual(original_child_hash, rebased.revisions[0].content_hash)
        history = rebased.derivation_provenance["rebase_history"]
        self.assertEqual(1, len(history))
        self.assertEqual("reviewer-1", history[0]["rebased_by"])
        self.assertEqual(
            "Reviewed the changed filter threshold.",
            history[0]["reason"],
        )
        current = pipeline_response(
            self.session,
            rebased,
            principal=self.actor,
            registry=self.registry,
        )
        self.assertFalse(current.governance.update_available)


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