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,
    PipelineUpdateRequest,
)
from govoplan_dataflow.backend.service import (
    DataflowConflictError,
    create_pipeline,
    derive_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():
    from test_service import sample_graph as build_graph

    return build_graph()


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

    def resolve_automation_principal(self, _session, *, 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"},
        )


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

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

        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"])

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


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