from __future__ import annotations

from conftest import (
    table_definition,
    write_basic_snapshot,
    write_incremental_schema,
    write_json,
    write_snapshot_schema_sql,
    write_text,
)

from app import _update, schema
from errors import ConnectorError
from fivetran.operations import RecordingOperationSink
from tests.pytest_compat import pytest


def test_update_runs_snapshot_then_incremental_without_snapshot_table_cache(workspace):
    write_basic_snapshot(workspace, pos=100)
    write_text(workspace / "snapshot/test.worker.000000001.csv", "id,name\n1,Alice\n")
    write_json(workspace / "increment/metadata", {"checkpoint-ts": 200})
    write_incremental_schema(workspace, table_definition(version=80))
    write_text(workspace / "increment/test/worker/80/2026-05-25/meta/CDC.index", "CDC000001.csv\n")
    write_text(workspace / "increment/test/worker/80/2026-05-25/CDC000001.csv", "U,worker,test,121,1,Alicia\n")
    operations = RecordingOperationSink()

    _update(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "operation_mode": "standard",
        },
        {},
        operations,
    )

    upserts = [call for call in operations.calls if call[0] == "upsert"]
    assert upserts == [
        ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Alice"}}),
        ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Alicia"}}),
    ]
    checkpoints = [call for call in operations.calls if call[0] == "checkpoint"]
    final_state = checkpoints[-1][1]["state"]
    assert final_state["snapshot"]["done"] is True
    assert final_state["snapshot"]["snapshot_tso"] == "100"
    assert final_state["incremental"]["checkpoint_ts"] == "200"
    assert "scan" not in final_state["incremental"]


def test_update_incremental_only_without_snapshot(workspace):
    write_json(workspace / "increment/metadata", {"checkpoint-ts": 200})
    write_incremental_schema(workspace, table_definition(version=120))
    write_text(workspace / "increment/test/worker/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n")
    write_text(workspace / "increment/test/worker/120/2026-05-25/CDC000001.csv", "I,worker,test,121,1,Alicia\n")
    configuration = {
        "sync_task_id": "task-001",
        "s3_url": f"file://{workspace}",
        "snapshot_prefix": "snapshot",
        "incremental_prefix": "increment",
        "snapshot_required": False,
        "operation_mode": "standard",
    }
    operations = RecordingOperationSink()

    assert schema(configuration) == [
        {
            "table": "test_worker",
            "primary_key": ["id"],
            "columns": {"id": "LONG", "name": "STRING"},
        }
    ]

    _update(configuration, {}, operations)

    upserts = [call for call in operations.calls if call[0] == "upsert"]
    assert upserts == [
        ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Alicia"}}),
    ]
    final_state = [call for call in operations.calls if call[0] == "checkpoint"][-1][1]["state"]
    assert "snapshot" not in final_state
    assert final_state["incremental"]["checkpoint_ts"] == "200"
    assert "scan" not in final_state["incremental"]


def test_update_incremental_fast_path_still_emits_checkpoint(workspace):
    write_json(workspace / "increment/metadata", {"checkpoint-ts": 200})
    write_incremental_schema(workspace, table_definition(version=120))
    configuration = {
        "sync_task_id": "task-001",
        "s3_url": f"file://{workspace}",
        "snapshot_prefix": "snapshot",
        "incremental_prefix": "increment",
        "snapshot_required": False,
        "operation_mode": "standard",
    }
    operations = RecordingOperationSink()

    _update(
        configuration,
        {
            "version": 1,
            "sync_task_id": "task-001",
            "incremental": {"checkpoint_ts": "200", "ddl_watermark": {}},
        },
        operations,
    )

    assert [call for call in operations.calls if call[0] == "upsert"] == []
    checkpoints = [call for call in operations.calls if call[0] == "checkpoint"]
    assert len(checkpoints) == 1
    assert checkpoints[0][1]["state"]["incremental"]["checkpoint_ts"] == "200"


def test_update_rejects_resync_snapshot_prefix_path_escape_before_catalog_discovery(workspace):
    write_basic_snapshot(workspace, pos=100)

    with pytest.raises(ConnectorError, match=r"state\.resync\.snapshot_prefix"):
        _update(
            {
                "sync_task_id": "task-001",
                "s3_url": f"file://{workspace}",
                "snapshot_prefix": "snapshot",
                "incremental_prefix": "increment",
                "operation_mode": "standard",
            },
            {
                "version": 1,
                "sync_task_id": "task-001",
                "snapshot": {"done": True, "snapshot_tso": "100"},
                "resync": {
                    "request_id": "table-resync-001",
                    "phase": "snapshot",
                    "tables": ["`test`.`worker`"],
                    "snapshot_prefix": "../outside",
                    "snapshot_tso": "150",
                },
                "incremental": {"checkpoint_ts": "100", "ddl_watermark": {}},
            },
            RecordingOperationSink(),
        )


def test_update_keeps_existing_tables_incrementing_while_new_table_has_no_rows(workspace):
    write_basic_snapshot(workspace, pos=100)
    write_snapshot_schema_sql(workspace, table="new_worker")
    write_json(workspace / "increment/metadata", {"checkpoint-ts": 200})
    write_incremental_schema(workspace, table_definition(version=80))
    write_text(workspace / "increment/test/worker/80/2026-05-25/meta/CDC.index", "CDC000001.csv\n")
    write_text(workspace / "increment/test/worker/80/2026-05-25/CDC000001.csv", "U,worker,test,121,1,Alicia\n")
    operations = RecordingOperationSink()

    _update(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "operation_mode": "standard",
        },
        {
            "version": 1,
            "sync_task_id": "task-001",
            "snapshot": {"done": True, "snapshot_tso": "100"},
            "incremental": {
                "checkpoint_ts": "0",
                "ddl_watermark": {},
            },
        },
        operations,
    )

    upserts = [call for call in operations.calls if call[0] == "upsert"]
    assert upserts == [
        ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Alicia"}}),
    ]
    checkpoints = [call for call in operations.calls if call[0] == "checkpoint"]
    final_state = checkpoints[-1][1]["state"]
    assert final_state["snapshot"]["done"] is True
    assert "tables" not in final_state["snapshot"]


def test_update_resyncs_top_level_resync_table_from_state_without_s3_marker(workspace):
    write_basic_snapshot(workspace, pos=150)
    write_text(workspace / "snapshot/test.worker.000000001.csv", "id,name\n1,Snapshot\n")
    write_snapshot_schema_sql(workspace, table="team")
    write_json(workspace / "increment/metadata", {"checkpoint-ts": 200})
    write_incremental_schema(workspace, table_definition(table="worker", version=120))
    write_incremental_schema(workspace, table_definition(table="team", version=120))
    write_text(workspace / "increment/test/worker/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n")
    write_text(
        workspace / "increment/test/worker/120/2026-05-25/CDC000001.csv",
        "U,worker,test,121,1,TooOld\nU,worker,test,151,1,Ready\n",
    )
    write_text(workspace / "increment/test/team/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n")
    write_text(workspace / "increment/test/team/120/2026-05-25/CDC000001.csv", "U,team,test,121,2,Team\n")
    operations = RecordingOperationSink()

    _update(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "operation_mode": "standard",
        },
        {
            "version": 1,
            "sync_task_id": "task-001",
            "snapshot": {"done": True, "snapshot_tso": "100"},
            "resync": {
                "request_id": "table-resync-001",
                "phase": "snapshot",
                "tables": ["`test`.`worker`"],
                "snapshot_prefix": "snapshot",
                "snapshot_tso": "130",
            },
            "incremental": {
                "checkpoint_ts": "0",
                "ddl_watermark": {"`test`.`worker`": 120, "`test`.`team`": 120},
            },
        },
        operations,
    )

    assert ("truncate", {"table": "test_worker"}) in operations.calls
    assert ("truncate", {"table": "test_team"}) not in operations.calls
    upserts = [call for call in operations.calls if call[0] == "upsert"]
    assert ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Snapshot"}}) in upserts
    assert ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Ready"}}) in upserts
    assert ("upsert", {"table": "test_team", "data": {"id": 2, "name": "Team"}}) in upserts
    assert all(call[1]["data"].get("name") != "TooOld" for call in upserts)

    final_state = [call for call in operations.calls if call[0] == "checkpoint"][-1][1]["state"]
    assert final_state["snapshot"]["done"] is True
    assert "tables" not in final_state["snapshot"]
    assert "resync" not in final_state


def test_update_table_resync_catchup_consumes_target_rows_before_clearing_resync(workspace):
    write_basic_snapshot(workspace, pos=150)
    write_text(workspace / "snapshot/test.worker.000000001.csv", "id,name\n1,Snapshot\n")
    write_snapshot_schema_sql(workspace, table="team")
    write_json(workspace / "increment/metadata", {"checkpoint-ts": 200})
    write_incremental_schema(workspace, table_definition(table="worker", version=120))
    write_incremental_schema(workspace, table_definition(table="team", version=120))
    write_text(workspace / "increment/test/worker/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n")
    write_text(
        workspace / "increment/test/worker/120/2026-05-25/CDC000001.csv",
        "U,worker,test,121,1,TooOld\nU,worker,test,151,1,Ready\n",
    )
    write_text(workspace / "increment/test/team/120/2026-05-25/meta/CDC.index", "CDC000001.csv\n")
    write_text(workspace / "increment/test/team/120/2026-05-25/CDC000001.csv", "U,team,test,151,2,Team\n")
    operations = RecordingOperationSink()

    _update(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "operation_mode": "standard",
        },
        {
            "version": 1,
            "sync_task_id": "task-001",
            "snapshot": {"done": True, "snapshot_tso": "100"},
            "resync": {
                "request_id": "table-resync-001",
                "phase": "snapshot",
                "tables": ["`test`.`worker`"],
                "snapshot_prefix": "snapshot",
                "snapshot_tso": "150",
            },
            "incremental": {
                "checkpoint_ts": "200",
                "ddl_watermark": {"`test`.`worker`": 120, "`test`.`team`": 120},
            },
        },
        operations,
    )

    upserts = [call for call in operations.calls if call[0] == "upsert"]
    assert ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Snapshot"}}) in upserts
    assert ("upsert", {"table": "test_worker", "data": {"id": 1, "name": "Ready"}}) in upserts
    assert ("upsert", {"table": "test_team", "data": {"id": 2, "name": "Team"}}) not in upserts
    assert all(call[1]["data"].get("name") != "TooOld" for call in upserts)

    final_state = [call for call in operations.calls if call[0] == "checkpoint"][-1][1]["state"]
    assert final_state["incremental"]["checkpoint_ts"] == "200"
    assert "resync" not in final_state


def test_update_table_resync_discovers_new_empty_table_from_resync_snapshot_prefix(workspace):
    write_basic_snapshot(workspace, pos=100)
    write_text(
        workspace / "snapshot/table-resync-001/metadata",
        "\n".join(
            [
                "Started dump at: 2026-05-26 06:19:14",
                "SHOW MASTER STATUS:",
                "    Log: tidb-binlog",
                "    Pos: 150",
                "    GTID:",
                "",
                "Finished dump at: 2026-05-26 06:19:36",
                "",
            ]
        ),
    )
    write_text(
        workspace / "snapshot/table-resync-001/test.new_worker-schema.sql",
        """CREATE TABLE `new_worker` (
  `id` bigint NOT NULL,
  `name` varchar(64) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
""",
    )
    operations = RecordingOperationSink()

    _update(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "operation_mode": "standard",
        },
        {
            "version": 1,
            "sync_task_id": "task-001",
            "snapshot": {"done": True, "snapshot_tso": "100"},
            "resync": {
                "request_id": "table-resync-001",
                "phase": "snapshot",
                "tables": ["`test`.`new_worker`"],
                "snapshot_prefix": "snapshot/table-resync-001",
                "snapshot_tso": "150",
            },
            "incremental": {"checkpoint_ts": "100", "ddl_watermark": {}},
        },
        operations,
    )

    assert [call for call in operations.calls if call[0] == "truncate"] == [("truncate", {"table": "test_new_worker"})]
    assert [call for call in operations.calls if call[0] == "upsert"] == []
    final_state = [call for call in operations.calls if call[0] == "checkpoint"][-1][1]["state"]
    assert final_state["resync"]["phase"] == "align"
    assert final_state["resync"]["tables"] == ["`test`.`new_worker`"]