from __future__ import annotations

import io
import json
import shutil
from pathlib import Path
from typing import Any

from conftest import write_json

from config import ConnectorConfig
from fivetran.operations import RecordingOperationSink
from increment.csv import iter_events
from increment.paths import parse_schema_path
from increment.reader import sync_incremental
from increment.schema import classify_ddl_action, load_table_definition
from increment.schema_index import load_storage_schema_index
from state import normalize_state
from storage.object_store import LocalObjectStore

FIXTURE_ROOT = Path(__file__).parents[1] / "fixtures/incremental_ddl_schema"
EXPECTED_DDL_TYPES = {1, 2, 3, 4, 5, 6, 7, 8, 10, 11, 12, 14, 15, 17, 19, 22, 61}
TRUNCATE_DDL_TYPES = {4, 11}


def _fixture_index() -> list[dict[str, Any]]:
    return json.loads((FIXTURE_ROOT / "manifest.json").read_text(encoding="utf-8"))


def _normalize_query(query: str) -> str:
    return " ".join(query.split())


def test_real_incremental_ddl_schema_fixtures_cover_one_file_per_type() -> None:
    fixture_index = _fixture_index()

    assert {int(item["type"]) for item in fixture_index} == EXPECTED_DDL_TYPES
    assert len(fixture_index) == len(EXPECTED_DDL_TYPES)
    for item in fixture_index:
        assert (FIXTURE_ROOT / item["relative_path"]).is_file()


def test_real_incremental_ddl_schema_files_load_and_classify() -> None:
    for item in _fixture_index():
        schema_path = parse_schema_path(item["relative_path"], "")
        assert schema_path is not None
        raw = json.loads((FIXTURE_ROOT / item["relative_path"]).read_text(encoding="utf-8"))

        table_def = load_table_definition(raw, schema_path)
        assert table_def.ddl_type == int(item["type"])
        assert table_def.source_schema == item["schema"]
        assert table_def.source_table == item["table"]
        assert table_def.table_version == int(item["table_version"])
        assert _normalize_query(table_def.query) == item["query"]

        if item["table"]:
            assert table_def.columns
            assert table_def.primary_key
        else:
            assert table_def.columns == ()
            assert table_def.primary_key == ()

        action = classify_ddl_action(table_def)
        if int(item["type"]) in TRUNCATE_DDL_TYPES:
            assert action.operation == "truncate"
            assert action.should_warn is False
        else:
            assert action.operation == "schema_update"
            assert action.should_warn is False


def test_real_incremental_schema_columns_decode_event() -> None:
    item = next(item for item in _fixture_index() if int(item["type"]) == 3)
    schema_path = parse_schema_path(item["relative_path"], "")
    assert schema_path is not None
    raw = json.loads((FIXTURE_ROOT / item["relative_path"]).read_text(encoding="utf-8"))
    table_def = load_table_definition(raw, schema_path)
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": "file:///tmp/source-sync-fixture",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
        }
    )

    events = list(
        iter_events(
            io.StringIO(
                f"I,{table_def.source_table},{table_def.source_schema},{table_def.table_version + 1},1,payload\n"
            ),
            table_def=table_def,
            config=config,
        )
    )

    assert len(events) == 1
    assert events[0].data == {"id": 1, "payload": "payload"}


def test_real_incremental_ddl_schema_markers_sync_without_catalog(workspace: Path) -> None:
    shutil.copytree(FIXTURE_ROOT, workspace / "increment", dirs_exist_ok=True)
    high_watermark = max(int(item["table_version"]) for item in _fixture_index()) + 1
    write_json(workspace / "increment/metadata", {"checkpoint-ts": high_watermark})

    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
        }
    )
    store = LocalObjectStore(str(workspace))
    schema_index = load_storage_schema_index(store, config.incremental_prefix)
    state = normalize_state(
        {
            "version": 1,
            "sync_task_id": "task-001",
            "snapshot": {"done": True, "snapshot_tso": "1"},
            "incremental": {"checkpoint_ts": "1", "ddl_watermark": {}},
        },
        sync_task_id="task-001",
    )
    operations = RecordingOperationSink()

    sync_incremental(
        config=config,
        store=store,
        schema_index=schema_index,
        state=state,
        operations=operations,
    )

    truncates = [call for call in operations.calls if call[0] == "truncate"]
    assert ("truncate", {"table": "ft_ddl_type_test_ft_ddl_drop_recreate"}) in truncates
    assert ("truncate", {"table": "ft_ddl_type_test_ft_ddl_truncate_target"}) in truncates
    assert state.checkpoint_ts == high_watermark
    assert state.scan is None