from __future__ import annotations

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


CAPABILITY_DATASOURCE_CATALOGUE = "datasources.catalogue"
CAPABILITY_DATASOURCE_LIFECYCLE = "datasources.lifecycle"
CAPABILITY_DATASOURCE_PUBLICATION = "datasources.publication"
CAPABILITY_DATASOURCE_ORIGINS = "connectors.datasourceOrigins"

DatasourceMode = Literal["live", "cached", "static"]
DatasourceKind = Literal[
    "upload",
    "database",
    "http",
    "rest",
    "directory",
    "file",
    "feed",
    "custom",
]
DatasourceShape = Literal["tabular", "document", "binary", "directory", "stream"]
DatasourceConsistency = Literal["current", "live", "frozen"]


class DatasourceError(ValueError):
    """Stable base error for provider-neutral datasource operations."""


class DatasourceNotFoundError(DatasourceError):
    pass


class DatasourceAccessError(DatasourceError):
    pass


class DatasourceValidationError(DatasourceError):
    pass


class DatasourceUnavailableError(DatasourceError):
    pass


@dataclass(frozen=True, slots=True)
class DatasourceField:
    name: str
    data_type: str
    nullable: bool = True


@dataclass(frozen=True, slots=True)
class DatasourceDescriptor:
    ref: str
    source_name: str
    name: str
    kind: DatasourceKind
    mode: DatasourceMode
    shape: DatasourceShape
    status: str = "active"
    description: str | None = None
    provider: str | None = None
    provider_ref: str | None = None
    schema: tuple[DatasourceField, ...] = ()
    schema_version: str = "1"
    fingerprint: str = ""
    current_materialization_ref: str | None = None
    row_count: int | None = None
    byte_count: int | None = None
    updated_at: datetime | None = None
    capabilities: tuple[str, ...] = ("read",)
    provenance: Mapping[str, object] = field(default_factory=dict)
    metadata: Mapping[str, object] = field(default_factory=dict)


@dataclass(frozen=True, slots=True)
class DatasourceMaterialization:
    ref: str
    datasource_ref: str
    revision: int
    state: str
    fingerprint: str
    schema: tuple[DatasourceField, ...] = ()
    row_count: int | None = None
    byte_count: int | None = None
    frozen_at: datetime | None = None
    frozen_label: str | None = None
    source_timestamp: datetime | None = None
    created_at: datetime | None = None
    provenance: Mapping[str, object] = field(default_factory=dict)
    metadata: Mapping[str, object] = field(default_factory=dict)


@dataclass(frozen=True, slots=True)
class DatasourceStage:
    ref: str
    name: str
    source_name: str
    kind: DatasourceKind
    mode: DatasourceMode
    shape: DatasourceShape
    state: str
    target_datasource_ref: str | None = None
    fingerprint: str = ""
    schema: tuple[DatasourceField, ...] = ()
    row_count: int | None = None
    byte_count: int | None = None
    validation: Mapping[str, object] = field(default_factory=dict)
    created_at: datetime | None = None
    promoted_at: datetime | None = None
    promoted_materialization_ref: str | None = None
    provenance: Mapping[str, object] = field(default_factory=dict)
    metadata: Mapping[str, object] = field(default_factory=dict)


@dataclass(frozen=True, slots=True)
class DatasourceReadRequest:
    datasource_ref: str
    materialization_ref: str | None = None
    consistency: DatasourceConsistency = "current"
    limit: int = 250
    offset: int = 0
    columns: tuple[str, ...] = ()
    expected_fingerprint: str | None = None


@dataclass(frozen=True, slots=True)
class DatasourceReadResult:
    datasource: DatasourceDescriptor
    rows: tuple[Mapping[str, object], ...]
    total_rows: int
    truncated: bool
    materialization: DatasourceMaterialization | None = None


@dataclass(frozen=True, slots=True)
class DatasourceStageInput:
    name: str
    source_name: str
    kind: DatasourceKind
    mode: DatasourceMode
    shape: DatasourceShape
    rows: tuple[Mapping[str, object], ...]
    description: str | None = None
    target_datasource_ref: str | None = None
    provider: str | None = None
    provider_ref: str | None = None
    provenance: Mapping[str, object] = field(default_factory=dict)
    metadata: Mapping[str, object] = field(default_factory=dict)


@dataclass(frozen=True, slots=True)
class DatasourcePublicationRequest:
    producer_module: str
    producer_run_ref: str
    idempotency_key: str
    rows: tuple[Mapping[str, object], ...]
    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
    source_timestamp: datetime | None = None
    provenance: Mapping[str, object] = field(default_factory=dict)
    metadata: Mapping[str, object] = field(default_factory=dict)


@dataclass(frozen=True, slots=True)
class DatasourcePublicationResult:
    ref: str
    status: str
    datasource: DatasourceDescriptor
    materialization: DatasourceMaterialization
    replayed: bool = False


@dataclass(frozen=True, slots=True)
class DatasourceOrigin:
    ref: str
    source_name: str
    name: str
    kind: DatasourceKind
    shape: DatasourceShape
    supported_modes: tuple[DatasourceMode, ...]
    provider: str
    description: str | None = None
    schema: tuple[DatasourceField, ...] = ()
    schema_version: str = "1"
    fingerprint: str = ""
    row_count: int | None = None
    byte_count: int | None = None
    updated_at: datetime | None = None
    capabilities: tuple[str, ...] = ("read",)
    metadata: Mapping[str, object] = field(default_factory=dict)


@dataclass(frozen=True, slots=True)
class DatasourceOriginReadRequest:
    origin_ref: str
    limit: int = 250
    offset: int = 0
    columns: tuple[str, ...] = ()
    expected_fingerprint: str | None = None


@dataclass(frozen=True, slots=True)
class DatasourceOriginReadResult:
    origin: DatasourceOrigin
    rows: tuple[Mapping[str, object], ...]
    total_rows: int
    truncated: bool


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

    def get_datasource(
        self,
        session: object,
        principal: object,
        *,
        datasource_ref: str,
    ) -> DatasourceDescriptor | None:
        ...

    def read_datasource(
        self,
        session: object,
        principal: object,
        *,
        request: DatasourceReadRequest,
    ) -> DatasourceReadResult:
        ...

    def list_materializations(
        self,
        session: object,
        principal: object,
        *,
        datasource_ref: str,
    ) -> Sequence[DatasourceMaterialization]:
        ...


@runtime_checkable
class DatasourceLifecycleProvider(Protocol):
    def list_stages(
        self,
        session: object,
        principal: object,
        *,
        limit: int = 100,
    ) -> Sequence[DatasourceStage]:
        ...

    def create_stage(
        self,
        session: object,
        principal: object,
        *,
        stage: DatasourceStageInput,
    ) -> DatasourceStage:
        ...

    def promote_stage(
        self,
        session: object,
        principal: object,
        *,
        stage_ref: str,
        freeze: bool = False,
        frozen_label: str | None = None,
    ) -> tuple[DatasourceDescriptor, DatasourceMaterialization]:
        ...

    def register_origin(
        self,
        session: object,
        principal: object,
        *,
        origin_ref: str,
        name: str,
        source_name: str,
        mode: DatasourceMode,
        description: str | None = None,
    ) -> DatasourceDescriptor:
        ...

    def refresh_datasource(
        self,
        session: object,
        principal: object,
        *,
        datasource_ref: str,
    ) -> tuple[DatasourceDescriptor, DatasourceMaterialization]:
        ...

    def freeze_datasource(
        self,
        session: object,
        principal: object,
        *,
        datasource_ref: str,
        label: str | None = None,
    ) -> DatasourceMaterialization:
        ...

    def retire_datasource(
        self,
        session: object,
        principal: object,
        *,
        datasource_ref: str,
    ) -> DatasourceDescriptor:
        ...


@runtime_checkable
class DatasourcePublicationProvider(Protocol):
    def publish_rows(
        self,
        session: object,
        principal: object,
        *,
        request: DatasourcePublicationRequest,
    ) -> DatasourcePublicationResult:
        ...


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

    def get_origin(
        self,
        session: object,
        principal: object,
        *,
        origin_ref: str,
    ) -> DatasourceOrigin | None:
        ...

    def read_origin(
        self,
        session: object,
        principal: object,
        *,
        request: DatasourceOriginReadRequest,
    ) -> DatasourceOriginReadResult:
        ...


def datasource_catalogue(registry: object | None) -> DatasourceCatalogueProvider | None:
    capability = _capability(registry, CAPABILITY_DATASOURCE_CATALOGUE)
    return capability if isinstance(capability, DatasourceCatalogueProvider) else None


def datasource_lifecycle(registry: object | None) -> DatasourceLifecycleProvider | None:
    capability = _capability(registry, CAPABILITY_DATASOURCE_LIFECYCLE)
    return capability if isinstance(capability, DatasourceLifecycleProvider) else None


def datasource_publication(
    registry: object | None,
) -> DatasourcePublicationProvider | None:
    capability = _capability(registry, CAPABILITY_DATASOURCE_PUBLICATION)
    return capability if isinstance(capability, DatasourcePublicationProvider) else None


def datasource_origins(registry: object | None) -> DatasourceOriginProvider | None:
    capability = _capability(registry, CAPABILITY_DATASOURCE_ORIGINS)
    return capability if isinstance(capability, DatasourceOriginProvider) 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_DATASOURCE_CATALOGUE",
    "CAPABILITY_DATASOURCE_LIFECYCLE",
    "CAPABILITY_DATASOURCE_ORIGINS",
    "DatasourceAccessError",
    "DatasourceCatalogueProvider",
    "DatasourceConsistency",
    "DatasourceDescriptor",
    "DatasourceError",
    "DatasourceField",
    "DatasourceKind",
    "DatasourceLifecycleProvider",
    "DatasourceMaterialization",
    "DatasourceMode",
    "DatasourceNotFoundError",
    "DatasourceOrigin",
    "DatasourceOriginProvider",
    "DatasourceOriginReadRequest",
    "DatasourceOriginReadResult",
    "DatasourceReadRequest",
    "DatasourceReadResult",
    "DatasourceShape",
    "DatasourceStage",
    "DatasourceStageInput",
    "DatasourceUnavailableError",
    "DatasourceValidationError",
    "datasource_catalogue",
    "datasource_lifecycle",
    "datasource_origins",
]
