from __future__ import annotations

from collections.abc import Iterable
from dataclasses import dataclass
from urllib.parse import unquote

from errors import ManifestError
from metadata import TableName
from observability import warning
from snapshot.schema import (
    SNAPSHOT_PREFIX,
    SNAPSHOT_SCHEMA_CREATE_SUFFIX,
    SNAPSHOT_SCHEMA_SUFFIX,
    SnapshotTable,
    fivetran_table_name,
    parse_create_table_sql,
    quote_object_path_segment,
)
from storage.object_store import ObjectStore


@dataclass(frozen=True)
class SnapshotSchemaFile:
    source_schema: str
    source_table: str
    fivetran_table: str
    snapshot_file_stem: str

    @property
    def table_name(self) -> TableName:
        return TableName(self.source_schema, self.source_table)


@dataclass(frozen=True)
class SnapshotTables:
    tables: tuple[SnapshotTable, ...]

    @property
    def table_by_name(self) -> dict[TableName, SnapshotTable]:
        return {table.name: table for table in self.tables}

    @property
    def table_names(self) -> set[TableName]:
        return {table.name for table in self.tables}


def load_snapshot_tables(
    store: ObjectStore,
    snapshot_prefix: str = SNAPSHOT_PREFIX,
) -> SnapshotTables:
    tables = tuple(iter_snapshot_tables(store, snapshot_prefix))
    if not tables:
        raise ManifestError("no supported tables discovered from snapshot schema files")
    return SnapshotTables(tables=tables)


def iter_snapshot_tables(
    store: ObjectStore,
    snapshot_prefix: str,
) -> Iterable[SnapshotTable]:
    seen: set[TableName] = set()
    for schema_file in iter_snapshot_schema_files(store, snapshot_prefix):
        table = parse_create_table_sql(
            store.read_text(_snapshot_schema_key(snapshot_prefix, schema_file.snapshot_file_stem)),
            source_schema=schema_file.source_schema,
            source_table=schema_file.source_table,
            snapshot_file_stem=schema_file.snapshot_file_stem,
        )
        if not table.primary_key:
            warning(
                "skipping table without primary key",
                phase="schema",
                source_schema=schema_file.source_schema,
                source_table=schema_file.source_table,
                file_path=_snapshot_schema_key(snapshot_prefix, schema_file.snapshot_file_stem),
            )
            continue
        if table.name in seen:
            raise ManifestError(f"duplicate snapshot schema for table: {table.quoted}")
        seen.add(table.name)
        yield table


def iter_snapshot_schema_files(
    store: ObjectStore,
    snapshot_prefix: str,
) -> Iterable[SnapshotSchemaFile]:
    seen: set[TableName] = set()
    for key in store.iter_keys(snapshot_prefix):
        stem = _snapshot_schema_stem(key, snapshot_prefix)
        if stem is None:
            continue
        source_schema, source_table = _parse_snapshot_schema_stem(stem)
        table_name = TableName(source_schema, source_table)
        if table_name in seen:
            raise ManifestError(f"duplicate snapshot schema for table: {table_name.quoted}")
        seen.add(table_name)
        yield SnapshotSchemaFile(
            source_schema=source_schema,
            source_table=source_table,
            fivetran_table=fivetran_table_name(source_schema, source_table),
            snapshot_file_stem=stem,
        )


def _snapshot_schema_key(snapshot_prefix: str, stem: str) -> str:
    prefix = snapshot_prefix.strip("/")
    filename = f"{stem}{SNAPSHOT_SCHEMA_SUFFIX}"
    return f"{prefix}/{filename}" if prefix else filename


def _snapshot_schema_stem(key: str, snapshot_prefix: str) -> str | None:
    relative = key.strip("/")
    prefix = snapshot_prefix.strip("/")
    if prefix and relative.startswith(prefix + "/"):
        relative = relative[len(prefix) + 1 :]
    if "/" in relative:
        return None
    if not relative.endswith(SNAPSHOT_SCHEMA_SUFFIX):
        return None
    if relative.endswith(SNAPSHOT_SCHEMA_CREATE_SUFFIX):
        return None
    return relative[: -len(SNAPSHOT_SCHEMA_SUFFIX)]


def _parse_snapshot_schema_stem(stem: str) -> tuple[str, str]:
    if "." not in stem:
        raise ManifestError(f"snapshot schema file name must be <schema>.<table>-schema.sql: {stem}")
    schema_segment, table_segment = stem.split(".", 1)
    source_schema = _decode_object_path_segment(schema_segment, stem)
    source_table = _decode_object_path_segment(table_segment, stem)
    return source_schema, source_table


def _decode_object_path_segment(segment: str, stem: str) -> str:
    decoded = unquote(segment)
    if not decoded or decoded in {".", ".."} or "/" in decoded or "\\" in decoded:
        raise ManifestError(f"invalid snapshot schema path segment in {stem!r}")
    if quote_object_path_segment(decoded) != segment:
        raise ManifestError(f"snapshot schema path segment must be URL-encoded in {stem!r}")
    return decoded


__all__ = [
    "SnapshotSchemaFile",
    "SnapshotTables",
    "iter_snapshot_schema_files",
    "iter_snapshot_tables",
    "load_snapshot_tables",
]