from __future__ import annotations

import json
from typing import Any, cast

from deployer.runner import (
    DeployerError,
    FivetranApiClient,
    JobConfig,
    _render_fivetran_configuration,
    build_full_resync_config,
    build_table_resync_state,
    main,
    run_full_resync,
    run_job,
    run_table_resync,
)
from tests.pytest_compat import pytest


class FakeApiClient:
    def __init__(self, group_id: str = "group_123", connection_id: str = "conn_456"):
        self.group_id = group_id
        self.connection_id = connection_id

    def get_group_id_by_name(self, group_name: str) -> str:
        assert group_name == "tidb_snowflake"
        return self.group_id

    def get_connection_id_by_name(self, group_id: str, connection_name: str) -> str:
        assert group_id == self.group_id
        assert connection_name == "job_e2e"
        return self.connection_id


def test_job_config_reads_env(monkeypatch, tmp_path):
    pyproject = tmp_path / "pyproject.toml"
    pyproject.write_text(
        """
[project]
name = "ticdc-fivetran-connector"
version = "1.2.3"
""".strip()
    )
    monkeypatch.setenv("FIVETRAN_API_KEY_BASE64", "token")
    monkeypatch.setenv("DESTINATION_NAME", "tidb_snowflake")
    monkeypatch.setenv("CONNECTION_NAME", "job_e2e")
    monkeypatch.setenv(
        "CONNECTOR_CONFIG_JSON",
        '{"sync_task_id": "task-001", "s3_url":"s3://bucket/prefix","snapshot_prefix":"snapshot","incremental_prefix":"increment"}',
    )
    monkeypatch.setenv("PROJECT_PATH", str(tmp_path))
    monkeypatch.setenv("CONNECTOR_VERSION", "1.2.3")

    config = JobConfig.from_env()

    assert config.destination_name == "tidb_snowflake"
    assert config.connection_name == "job_e2e"
    assert config.connector_version == "1.2.3"
    assert config.connector_config == {
        "sync_task_id": "task-001",
        "s3_url": "s3://bucket/prefix",
        "snapshot_prefix": "snapshot",
        "incremental_prefix": "increment",
    }


def test_job_config_reads_connector_version_from_project(monkeypatch, tmp_path):
    (tmp_path / "pyproject.toml").write_text(
        """
[project]
name = "ticdc-fivetran-connector"
version = "1.2.3"
""".strip()
    )
    monkeypatch.setenv("FIVETRAN_API_KEY_BASE64", "token")
    monkeypatch.setenv("DESTINATION_NAME", "tidb_snowflake")
    monkeypatch.setenv("CONNECTION_NAME", "job_e2e")
    monkeypatch.setenv(
        "CONNECTOR_CONFIG_JSON",
        '{"sync_task_id": "task-001", "s3_url":"s3://bucket/prefix","snapshot_prefix":"snapshot","incremental_prefix":"increment"}',
    )
    monkeypatch.setenv("PROJECT_PATH", str(tmp_path))
    monkeypatch.delenv("CONNECTOR_VERSION", raising=False)

    config = JobConfig.from_env()

    assert config.connector_version == "1.2.3"


def test_job_config_rejects_version_mismatch(monkeypatch, tmp_path):
    (tmp_path / "pyproject.toml").write_text(
        """
[project]
name = "ticdc-fivetran-connector"
version = "9.9.9"
""".strip()
    )
    monkeypatch.setenv("FIVETRAN_API_KEY_BASE64", "token")
    monkeypatch.setenv("DESTINATION_NAME", "tidb_snowflake")
    monkeypatch.setenv("CONNECTION_NAME", "job_e2e")
    monkeypatch.setenv(
        "CONNECTOR_CONFIG_JSON",
        '{"sync_task_id": "task-001", "s3_url":"s3://bucket/prefix","snapshot_prefix":"snapshot","incremental_prefix":"increment"}',
    )
    monkeypatch.setenv("PROJECT_PATH", str(tmp_path))
    monkeypatch.setenv("CONNECTOR_VERSION", "1.2.3")

    with pytest.raises(DeployerError, match="does not match"):
        JobConfig.from_env()


def test_run_job_resolves_connection_id(monkeypatch, tmp_path):
    termination_log = tmp_path / "termination.log"
    config = JobConfig(
        api_key_b64="token",
        destination_name="tidb_snowflake",
        connection_name="job_e2e",
        connector_config={
            "sync_task_id": "task-001",
            "s3_url": "s3://bucket/prefix",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
        },
        request_id="req-1",
        connector_name="tidb-fivetran-connector",
        connector_version="1.2.3",
        project_path=str(tmp_path),
        python_version="3.13",
        termination_log_path=str(termination_log),
        base_url="https://api.fivetran.com/v1",
    )
    monkeypatch.setattr("deployer.runner._run_fivetran_deploy", lambda _config, _config_path: None)
    monkeypatch.setattr("deployer.runner.FivetranApiClient", lambda *_args, **_kwargs: FakeApiClient())

    result = run_job(config)

    assert result.destination_id == "group_123"
    assert result.connection_id == "conn_456"
    assert result.dashboard_url == "https://fivetran.com/dashboard/connectors/conn_456/status"


def test_render_fivetran_configuration_stringifies_values():
    rendered = _render_fivetran_configuration(
        {
            "sync_task_id": "task-001",
            "s3_url": "s3://bucket/prefix",
            "truncate_before_snapshot": False,
            "snapshot": {"csv_no_header": False},
            "incremental": {"csv_output_old_value": True},
        }
    )

    assert rendered == {
        "sync_task_id": "task-001",
        "s3_url": "s3://bucket/prefix",
        "truncate_before_snapshot": "false",
        "snapshot": '{"csv_no_header":false}',
        "incremental": '{"csv_output_old_value":true}',
    }


def test_main_writes_failure_payload(monkeypatch, tmp_path):
    log_path = tmp_path / "termination.log"
    monkeypatch.delenv("FIVETRAN_API_KEY_BASE64", raising=False)
    monkeypatch.setenv("TERMINATION_LOG_PATH", str(log_path))

    assert main() == 1

    payload = json.loads(log_path.read_text())
    assert payload["status"] == "failed"
    assert payload["error_code"] == "missing_env"


def test_fivetran_api_client_matches_group_and_schema(monkeypatch):
    client = FivetranApiClient("token", "https://api.fivetran.com/v1")

    responses = {
        "/groups": {
            "code": "Success",
            "data": {"items": [{"id": "gid", "name": "tidb_snowflake"}]},
        },
        "/connections?group_id=gid": {
            "code": "Success",
            "data": {"items": [{"id": "cid", "schema": "job_e2e"}]},
        },
    }

    monkeypatch.setattr(client, "_get_json", lambda path: responses[path])

    group_id = client.get_group_id_by_name("tidb_snowflake")
    connection_id = client.get_connection_id_by_name(group_id, "job_e2e")

    assert group_id == "gid"
    assert connection_id == "cid"


class RecordingControlPlaneApi:
    def __init__(self) -> None:
        self.calls: list[tuple[str, object]] = []
        self.connection: dict[str, object] = {"config": {}}
        self.state: dict[str, object] = {"state": {"version": 1}}

    def set_connection_paused(self, connection_id: str, paused: bool):
        self.calls.append(("pause", paused))
        return {"id": connection_id, "paused": paused}

    def get_connection(self, connection_id: str):
        self.calls.append(("get_connection", connection_id))
        return self.connection

    def get_connection_state(self, connection_id: str):
        self.calls.append(("get_state", connection_id))
        return self.state

    def patch_connection(self, connection_id: str, body):
        self.calls.append(("patch_connection", body))
        self.connection = {"config": body["config"]}
        return self.connection

    def patch_connection_state(self, connection_id: str, state):
        self.calls.append(("patch_state", state))
        self.state = {"state": state}
        return self.state

    def trigger_resync(self, connection_id: str):
        self.calls.append(("resync", connection_id))
        return {}


class MissingConfigReadBackApi(RecordingControlPlaneApi):
    def patch_connection(self, connection_id: str, body):
        self.calls.append(("patch_connection", body))
        self.connection = {}
        return self.connection


class MissingStateReadBackApi(RecordingControlPlaneApi):
    def patch_connection_state(self, connection_id: str, state):
        self.calls.append(("patch_state", state))
        self.state = {"state": {"version": 1, "sync_task_id": "task-001"}}
        return self.state


def test_build_full_resync_config_sets_snapshot_prefix_and_truncate():
    config = build_full_resync_config(
        {
            "sync_task_id": "task-001",
            "s3_url": "s3://bucket/workspace",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
        },
        snapshot_prefix="/snapshot/full-resync-001/",
    )

    assert config["snapshot_prefix"] == "snapshot/full-resync-001"
    assert config["truncate_before_snapshot"] is True


def test_run_full_resync_pauses_patches_readbacks_resyncs_and_unpauses():
    api = RecordingControlPlaneApi()

    run_full_resync(
        api=api,
        connection_id="conn_123",
        connector_config={
            "sync_task_id": "task-001",
            "s3_url": "s3://bucket/workspace",
            "incremental_prefix": "increment",
        },
        snapshot_prefix="snapshot/full-resync-001",
    )

    assert api.calls == [
        ("pause", True),
        ("get_connection", "conn_123"),
        ("get_state", "conn_123"),
        (
            "patch_connection",
            {
                "config": {
                    "sync_task_id": "task-001",
                    "s3_url": "s3://bucket/workspace",
                    "snapshot_prefix": "snapshot/full-resync-001",
                    "incremental_prefix": "increment",
                    "truncate_before_snapshot": True,
                }
            },
        ),
        ("get_connection", "conn_123"),
        ("resync", "conn_123"),
        ("pause", False),
    ]


def test_run_full_resync_rejects_missing_config_readback_and_unpauses():
    api = MissingConfigReadBackApi()

    with pytest.raises(DeployerError, match="patched config was not visible"):
        run_full_resync(
            api=api,
            connection_id="conn_123",
            connector_config={
                "sync_task_id": "task-001",
                "s3_url": "s3://bucket/workspace",
                "incremental_prefix": "increment",
            },
            snapshot_prefix="snapshot/full-resync-001",
        )

    assert ("resync", "conn_123") not in api.calls
    assert ("pause", False) not in api.calls


def test_build_full_resync_config_rejects_missing_incremental_prefix():
    with pytest.raises(DeployerError, match="incremental_prefix"):
        build_full_resync_config(
            {"sync_task_id": "task-001", "s3_url": "s3://bucket/workspace", "snapshot_prefix": "snapshot"},
            snapshot_prefix="/snapshot/full-resync-001/",
        )


def test_build_table_resync_state_adds_top_level_marker():
    state = build_table_resync_state(
        {
            "version": 1,
            "sync_task_id": "task-001",
            "snapshot": {"done": True, "snapshot_tso": "100"},
            "incremental": {"checkpoint_ts": "100", "ddl_watermark": {}},
        },
        request_id="table-resync-001",
        tables=["test.worker", "`a.b`.`c`", "`te.st`.`t.t`"],
        snapshot_tso=150,
        snapshot_prefix="/snapshot/table-resync-001/",
    )

    assert state["resync"] == {
        "request_id": "table-resync-001",
        "phase": "snapshot",
        "tables": ["`test`.`worker`", "`a.b`.`c`", "`te.st`.`t.t`"],
        "snapshot_tso": "150",
        "snapshot_prefix": "snapshot/table-resync-001",
        "truncate_done": False,
        "current_file": None,
        "next_row": 0,
    }


def test_build_table_resync_state_rejects_invalid_request():
    base_state = {
        "version": 1,
        "sync_task_id": "task-001",
        "snapshot": {"done": True, "snapshot_tso": "100"},
        "incremental": {"checkpoint_ts": "100", "ddl_watermark": {}},
    }

    with pytest.raises(DeployerError, match="duplicate table"):
        build_table_resync_state(
            base_state,
            request_id="table-resync-001",
            tables=["`test`.`worker`", "test.worker"],
            snapshot_tso=150,
            snapshot_prefix="snapshot/table-resync-001",
        )

    with pytest.raises(DeployerError, match="schema/table separator") as exc_info:
        build_table_resync_state(
            base_state,
            request_id="table-resync-001",
            tables=["te.st.t.t"],
            snapshot_tso=150,
            snapshot_prefix="snapshot/table-resync-001",
        )
    assert exc_info.value.code == "invalid_resync_request"
    assert "invalid_resync_request" not in exc_info.value.message

    with pytest.raises(DeployerError, match="snapshot_tso"):
        build_table_resync_state(
            base_state,
            request_id="table-resync-001",
            tables=["`test`.`worker`"],
            snapshot_tso=0,
            snapshot_prefix="snapshot/table-resync-001",
        )

    with pytest.raises(DeployerError, match="snapshot_prefix"):
        build_table_resync_state(
            base_state,
            request_id="table-resync-001",
            tables=["`test`.`worker`"],
            snapshot_tso=150,
            snapshot_prefix="../snapshot/table-resync-001",
        )


def test_build_table_resync_state_rejects_uninitialized_state():
    with pytest.raises(DeployerError, match="initialized connector state"):
        build_table_resync_state(
            {},
            request_id="table-resync-001",
            tables=["`test`.`worker`"],
            snapshot_tso=150,
            snapshot_prefix="snapshot/table-resync-001",
        )

    with pytest.raises(DeployerError, match="missing sync_task_id"):
        build_table_resync_state(
            {
                "version": 1,
                "snapshot": {"done": True, "snapshot_tso": "100"},
                "incremental": {"checkpoint_ts": "100", "ddl_watermark": {}},
            },
            request_id="table-resync-001",
            tables=["`test`.`worker`"],
            snapshot_tso=150,
            snapshot_prefix="snapshot/table-resync-001",
        )


def test_run_table_resync_pauses_patches_state_readbacks_and_unpauses():
    api = RecordingControlPlaneApi()
    api.state = {
        "state": {
            "version": 1,
            "sync_task_id": "task-001",
            "snapshot": {"done": True, "snapshot_tso": "100"},
            "incremental": {"checkpoint_ts": "200", "ddl_watermark": {}},
        }
    }

    run_table_resync(
        api=api,
        connection_id="conn_123",
        request_id="table-resync-001",
        tables=["`test`.`worker`"],
        snapshot_tso=150,
        snapshot_prefix="snapshot/table-resync-001",
    )

    assert [call[0] for call in api.calls] == [
        "pause",
        "get_state",
        "patch_state",
        "get_state",
        "pause",
    ]
    assert api.calls[0] == ("pause", True)
    assert api.calls[-1] == ("pause", False)
    patch_state = cast(dict[str, Any], next(call[1] for call in api.calls if call[0] == "patch_state"))
    assert patch_state["incremental"]["checkpoint_ts"] == "200"
    assert patch_state["resync"]["tables"] == ["`test`.`worker`"]


def test_run_table_resync_rejects_empty_state_without_patch_and_unpauses():
    api = RecordingControlPlaneApi()
    api.state = {"state": {}}

    with pytest.raises(DeployerError, match="initialized connector state"):
        run_table_resync(
            api=api,
            connection_id="conn_123",
            request_id="table-resync-001",
            tables=["`test`.`worker`"],
            snapshot_tso=150,
            snapshot_prefix="snapshot/table-resync-001",
        )

    assert [call[0] for call in api.calls] == ["pause", "get_state", "pause"]
    assert api.calls[-1] == ("pause", False)


def test_run_table_resync_keeps_paused_after_state_readback_mismatch():
    api = MissingStateReadBackApi()
    api.state = {
        "state": {
            "version": 1,
            "sync_task_id": "task-001",
            "snapshot": {"done": True, "snapshot_tso": "100"},
            "incremental": {"checkpoint_ts": "200", "ddl_watermark": {}},
        }
    }

    with pytest.raises(DeployerError, match="patched state was not visible"):
        run_table_resync(
            api=api,
            connection_id="conn_123",
            request_id="table-resync-001",
            tables=["`test`.`worker`"],
            snapshot_tso=150,
            snapshot_prefix="snapshot/table-resync-001",
        )

    assert ("pause", False) not in api.calls