from __future__ import annotations

from collections.abc import Mapping, Sequence
from dataclasses import dataclass, field
from datetime import datetime
from typing import Protocol, runtime_checkable

from govoplan_core.core.automation import AutomationInvocation
from govoplan_core.core.events import PlatformEvent


CAPABILITY_DATAFLOW_RUN_LIFECYCLE = "dataflow.runLifecycle"
CAPABILITY_DATAFLOW_RUN_WORKER = "dataflow.runWorker"
CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER = "dataflow.triggerDispatcher"
CAPABILITY_DATAFLOW_DATASET_OUTPUT = "dataflow.dataset_output"


class DataflowRunError(ValueError):
    """Stable base error for module-neutral Dataflow run operations."""


class DataflowRunNotFoundError(DataflowRunError):
    pass


class DataflowRunConflictError(DataflowRunError):
    pass


class DataflowRunUnavailableError(DataflowRunError):
    pass


@dataclass(frozen=True, slots=True)
class DataflowDatasetDescriptor:
    pipeline_ref: str
    name: str
    revision: int
    definition_hash: str
    status: str
    description: str | None = None
    updated_at: datetime | None = None
    parameters: Mapping[str, object] = field(default_factory=dict)
    provenance: Mapping[str, object] = field(default_factory=dict)


@dataclass(frozen=True, slots=True)
class DataflowDatasetRequest:
    pipeline_ref: str
    revision: int
    parameters: Mapping[str, object] = field(default_factory=dict)
    row_limit: int = 500
    expected_definition_hash: str | None = None
    expected_source_fingerprints: tuple[Mapping[str, object], ...] = ()


@dataclass(frozen=True, slots=True)
class DataflowDatasetResult:
    pipeline_ref: str
    revision: int
    definition_hash: str
    rows: tuple[Mapping[str, object], ...]
    total_rows: int
    truncated: bool
    output_hash: str
    executor_version: str
    run_ref: str | None = None
    source_fingerprints: tuple[Mapping[str, object], ...] = ()
    diagnostics: tuple[Mapping[str, object], ...] = ()
    generated_at: datetime | None = None
    provenance: Mapping[str, object] = field(default_factory=dict)


@runtime_checkable
class DataflowDatasetOutputProvider(Protocol):
    def list_outputs(
        self,
        session: object,
        principal: object,
        *,
        query: str = "",
        limit: int = 100,
    ) -> Sequence[DataflowDatasetDescriptor]: ...

    def read_output(
        self,
        session: object,
        principal: object,
        *,
        request: DataflowDatasetRequest,
    ) -> DataflowDatasetResult: ...


@dataclass(frozen=True, slots=True)
class DataflowPublicationTarget:
    target_datasource_ref: str | None = None
    name: str | None = None
    source_name: str | None = None
    description: str | None = None
    freeze: bool = False
    frozen_label: str | None = None
    set_current: bool = True
    metadata: Mapping[str, object] = field(default_factory=dict)


@dataclass(frozen=True, slots=True)
class DataflowRunRequest:
    pipeline_ref: str
    revision: int
    idempotency_key: str
    row_limit: int = 500
    execution_backend: str = "auto"
    environment: str = "development"
    max_attempts: int = 3
    retention_days: int = 30
    publication: DataflowPublicationTarget | None = None
    invocation: AutomationInvocation = field(
        default_factory=AutomationInvocation
    )


@dataclass(frozen=True, slots=True)
class DataflowRunDescriptor:
    ref: str
    pipeline_ref: str
    revision: int
    status: str
    definition_hash: str
    executor_version: str
    input_row_count: int = 0
    output_row_count: int = 0
    output_publication_ref: str | None = None
    output_datasource_ref: str | None = None
    output_materialization_ref: str | None = None
    invocation_kind: str = "manual"
    trigger_ref: str | None = None
    delivery_ref: str | None = None
    error: str | None = None
    started_at: datetime | None = None
    finished_at: datetime | None = None
    replayed: bool = False
    metadata: Mapping[str, object] = field(default_factory=dict)


@runtime_checkable
class DataflowRunLifecycleProvider(Protocol):
    def start_run(
        self,
        session: object,
        principal: object,
        *,
        request: DataflowRunRequest,
    ) -> DataflowRunDescriptor:
        ...

    def get_run(
        self,
        session: object,
        principal: object,
        *,
        run_ref: str,
    ) -> DataflowRunDescriptor | None:
        ...

    def cancel_run(
        self,
        session: object,
        principal: object,
        *,
        run_ref: str,
    ) -> DataflowRunDescriptor:
        ...


@runtime_checkable
class DataflowTriggerDispatcher(Protocol):
    def dispatch_due(
        self,
        session: object,
        *,
        now: datetime | None = None,
        limit: int = 50,
    ) -> Mapping[str, object]:
        ...

    def ingest_event(
        self,
        session: object,
        *,
        event: PlatformEvent,
    ) -> Mapping[str, object]:
        ...


@runtime_checkable
class DataflowRunWorker(Protocol):
    """Durable worker boundary for queued Dataflow execution.

    Implementations own claim transaction boundaries so a lease is committed
    before potentially long-running execution starts.
    """

    def dispatch_pending(
        self,
        session: object,
        *,
        now: datetime | None = None,
        limit: int = 10,
        worker_id: str | None = None,
    ) -> Mapping[str, object]:
        ...

    def purge_expired(
        self,
        session: object,
        *,
        now: datetime | None = None,
        limit: int = 500,
    ) -> Mapping[str, object]:
        ...


def dataflow_run_lifecycle(
    registry: object | None,
) -> DataflowRunLifecycleProvider | None:
    capability = _capability(registry, CAPABILITY_DATAFLOW_RUN_LIFECYCLE)
    return capability if isinstance(capability, DataflowRunLifecycleProvider) else None


def dataflow_run_worker(
    registry: object | None,
) -> DataflowRunWorker | None:
    capability = _capability(registry, CAPABILITY_DATAFLOW_RUN_WORKER)
    return capability if isinstance(capability, DataflowRunWorker) else None


def dataflow_trigger_dispatcher(
    registry: object | None,
) -> DataflowTriggerDispatcher | None:
    capability = _capability(registry, CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER)
    return (
        capability
        if isinstance(capability, DataflowTriggerDispatcher)
        else None
    )


def dataflow_dataset_output(
    registry: object | None,
) -> DataflowDatasetOutputProvider | None:
    capability = _capability(registry, CAPABILITY_DATAFLOW_DATASET_OUTPUT)
    return capability if isinstance(capability, DataflowDatasetOutputProvider) else None


def _capability(registry: object | None, name: str) -> object | None:
    if (
        registry is None
        or not hasattr(registry, "has_capability")
        or not hasattr(registry, "capability")
        or not registry.has_capability(name)
    ):
        return None
    return registry.capability(name)


__all__ = [
    "CAPABILITY_DATAFLOW_RUN_LIFECYCLE",
    "CAPABILITY_DATAFLOW_RUN_WORKER",
    "CAPABILITY_DATAFLOW_TRIGGER_DISPATCHER",
    "CAPABILITY_DATAFLOW_DATASET_OUTPUT",
    "DataflowDatasetDescriptor",
    "DataflowDatasetOutputProvider",
    "DataflowDatasetRequest",
    "DataflowDatasetResult",
    "DataflowPublicationTarget",
    "DataflowRunConflictError",
    "DataflowRunDescriptor",
    "DataflowRunError",
    "DataflowRunLifecycleProvider",
    "DataflowRunNotFoundError",
    "DataflowRunRequest",
    "DataflowRunUnavailableError",
    "DataflowRunWorker",
    "DataflowTriggerDispatcher",
    "dataflow_run_lifecycle",
    "dataflow_run_worker",
    "dataflow_trigger_dispatcher",
    "dataflow_dataset_output",
]
