from __future__ import annotations

from typing import Any, cast

from conftest import table_definition, write_basic_snapshot, write_incremental_schema

from config import ConnectorConfig
from fivetran.schema import build_fivetran_schema
from increment.schema_index import load_storage_schema_index
from snapshot.schema import parse_create_table_sql
from snapshot.tables import SnapshotTables, load_snapshot_tables
from storage.object_store import LocalObjectStore


def _schema_config(workspace) -> ConnectorConfig:
    return ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
        }
    )


def test_builds_fivetran_schema_from_dumpling_snapshot_schema(workspace):
    write_basic_snapshot(workspace)
    store = LocalObjectStore(str(workspace))
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
            "enable_diagnostic_columns": True,
        }
    )

    snapshot_tables = load_snapshot_tables(store)
    schema_index = load_storage_schema_index(store, config.incremental_prefix)
    result = build_fivetran_schema(snapshot_tables, schema_index, config)

    assert result == [
        {
            "table": "test_worker",
            "primary_key": ["id"],
            "columns": {
                "id": "LONG",
                "name": "STRING",
                "_incremental_commit_ts": "LONG",
                "_incremental_source_schema": "STRING",
                "_incremental_source_table": "STRING",
            },
        }
    ]


def test_fivetran_schema_uses_incremental_decimal_precision_and_scale(workspace):
    raw = table_definition()
    raw["TableColumns"].append(
        {
            "ColumnName": "amount",
            "ColumnType": "DECIMAL",
            "ColumnPrecision": "20",
            "ColumnScale": "6",
        }
    )
    raw["TableColumnsTotal"] = 3
    write_incremental_schema(workspace, raw)
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "",
            "incremental_prefix": "increment",
        }
    )

    schema_index = load_storage_schema_index(LocalObjectStore(str(workspace)), config.incremental_prefix)
    result = build_fivetran_schema(SnapshotTables(tables=()), schema_index, config)

    assert result == [
        {
            "table": "test_worker",
            "primary_key": ["id"],
            "columns": {
                "id": "LONG",
                "name": "STRING",
                "amount": {"type": "DECIMAL", "precision": 20, "scale": 6},
            },
        }
    ]


def test_fivetran_schema_uses_incremental_decimal_defaults_timestamp_and_vector(workspace):
    raw = table_definition()
    raw["TableColumns"].extend(
        [
            {
                "ColumnName": "amount",
                "ColumnType": "DECIMAL",
            },
            {
                "ColumnName": "created_at",
                "ColumnType": "TIMESTAMP",
            },
            {
                "ColumnName": "embedding",
                "ColumnType": "VECTOR",
            },
        ]
    )
    raw["TableColumnsTotal"] = 5
    write_incremental_schema(workspace, raw)
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "",
            "incremental_prefix": "increment",
        }
    )

    schema_index = load_storage_schema_index(LocalObjectStore(str(workspace)), config.incremental_prefix)
    result = build_fivetran_schema(SnapshotTables(tables=()), schema_index, config)

    assert result == [
        {
            "table": "test_worker",
            "primary_key": ["id"],
            "columns": {
                "id": "LONG",
                "name": "STRING",
                "amount": {"type": "DECIMAL", "precision": 10, "scale": 0},
                "created_at": "UTC_DATETIME",
                "embedding": "STRING",
            },
        }
    ]


def test_fivetran_schema_maps_incremental_bigint_unsigned_to_decimal(workspace):
    raw = table_definition()
    raw["TableColumns"].append(
        {
            "ColumnName": "u64",
            "ColumnType": "BIGINT UNSIGNED",
            "ColumnPrecision": "20",
        }
    )
    raw["TableColumnsTotal"] = 3
    write_incremental_schema(workspace, raw)
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "",
            "incremental_prefix": "increment",
        }
    )

    schema_index = load_storage_schema_index(LocalObjectStore(str(workspace)), config.incremental_prefix)
    result = build_fivetran_schema(SnapshotTables(tables=()), schema_index, config)

    columns = cast(dict[str, Any], result[0]["columns"])
    assert columns["u64"] == {"type": "DECIMAL", "precision": 20, "scale": 0}


def test_fivetran_schema_uses_snapshot_decimal_precision_and_scale(workspace):
    table = parse_create_table_sql(
        """CREATE TABLE `payment` (
  `id` bigint NOT NULL,
  `amount` decimal(20,6) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
""",
        source_schema="test",
        source_table="payment",
        snapshot_file_stem="test.payment",
    )
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "snapshot",
            "incremental_prefix": "increment",
        }
    )
    schema_index = load_storage_schema_index(LocalObjectStore(str(workspace)), config.incremental_prefix)

    result = build_fivetran_schema(SnapshotTables(tables=(table,)), schema_index, config)

    assert result == [
        {
            "table": "test_payment",
            "primary_key": ["id"],
            "columns": {
                "id": "LONG",
                "amount": {"type": "DECIMAL", "precision": 20, "scale": 6},
            },
        }
    ]


def test_fivetran_schema_merges_snapshot_and_incremental_columns(workspace):
    table = parse_create_table_sql(
        """CREATE TABLE `worker` (
  `id` bigint NOT NULL,
  `name` varchar(64) DEFAULT NULL,
  `old_name` varchar(64) DEFAULT NULL,
  `age` int DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
""",
        source_schema="test",
        source_table="worker",
        snapshot_file_stem="test.worker",
    )
    raw = table_definition()
    raw["TableColumns"].append({"ColumnName": "new_name", "ColumnType": "VARCHAR"})
    raw["TableColumnsTotal"] = 3
    write_incremental_schema(workspace, raw)
    config = _schema_config(workspace)
    schema_index = load_storage_schema_index(LocalObjectStore(str(workspace)), config.incremental_prefix)

    result = build_fivetran_schema(SnapshotTables(tables=(table,)), schema_index, config)

    assert result == [
        {
            "table": "test_worker",
            "primary_key": ["id"],
            "columns": {
                "id": "LONG",
                "name": "STRING",
                "old_name": "STRING",
                "age": "INT",
                "new_name": "STRING",
            },
        }
    ]


def test_fivetran_schema_keeps_old_and_new_tables_for_table_rename(workspace):
    old_table = parse_create_table_sql(
        """CREATE TABLE `old_worker` (
  `id` bigint NOT NULL,
  `name` varchar(64) DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
""",
        source_schema="test",
        source_table="old_worker",
        snapshot_file_stem="test.old_worker",
    )
    write_incremental_schema(
        workspace,
        table_definition(
            table="new_worker",
            version=120,
            ddl_type=14,
            query="rename table `test`.`old_worker` to `test`.`new_worker`",
        ),
    )
    config = _schema_config(workspace)
    schema_index = load_storage_schema_index(LocalObjectStore(str(workspace)), config.incremental_prefix)

    result = build_fivetran_schema(SnapshotTables(tables=(old_table,)), schema_index, config)

    assert [table["table"] for table in result] == ["test_new_worker", "test_old_worker"]


def test_fivetran_schema_keeps_dropped_table_from_snapshot(workspace):
    table = parse_create_table_sql(
        """CREATE TABLE `worker` (
  `id` bigint NOT NULL,
  `name` varchar(64) DEFAULT NULL,
  `age` int DEFAULT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin;
""",
        source_schema="test",
        source_table="worker",
        snapshot_file_stem="test.worker",
    )
    write_incremental_schema(
        workspace,
        table_definition(
            version=120,
            ddl_type=4,
            query="drop table `test`.`worker`",
        ),
    )
    config = _schema_config(workspace)
    schema_index = load_storage_schema_index(LocalObjectStore(str(workspace)), config.incremental_prefix)

    result = build_fivetran_schema(SnapshotTables(tables=(table,)), schema_index, config)

    assert result == [
        {
            "table": "test_worker",
            "primary_key": ["id"],
            "columns": {
                "id": "LONG",
                "name": "STRING",
                "age": "INT",
            },
        }
    ]


def test_fivetran_schema_maps_bit_one_and_skips_multi_bit_columns(workspace):
    raw = table_definition()
    raw["TableColumns"].extend(
        [
            {
                "ColumnName": "flag",
                "ColumnType": "BIT UNSIGNED",
                "ColumnPrecision": "1",
            },
            {
                "ColumnName": "bitmap",
                "ColumnType": "BIT UNSIGNED",
                "ColumnPrecision": "8",
            },
        ]
    )
    raw["TableColumnsTotal"] = 4
    write_incremental_schema(workspace, raw)
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "",
            "incremental_prefix": "increment",
        }
    )

    schema_index = load_storage_schema_index(LocalObjectStore(str(workspace)), config.incremental_prefix)
    result = build_fivetran_schema(SnapshotTables(tables=()), schema_index, config)

    assert result == [
        {
            "table": "test_worker",
            "primary_key": ["id"],
            "columns": {
                "id": "LONG",
                "name": "STRING",
                "flag": "BOOLEAN",
            },
        }
    ]


def test_fivetran_schema_skips_table_when_multi_bit_column_is_part_of_primary_key(workspace):
    raw = table_definition()
    raw["TableColumns"].append(
        {
            "ColumnName": "tenant_bits",
            "ColumnType": "BIT UNSIGNED",
            "ColumnPrecision": "8",
            "ColumnIsPk": "true",
        }
    )
    raw["TableColumnsTotal"] = 3
    write_incremental_schema(workspace, raw)
    config = ConnectorConfig.from_dict(
        {
            "sync_task_id": "task-001",
            "s3_url": f"file://{workspace}",
            "snapshot_prefix": "",
            "incremental_prefix": "increment",
        }
    )

    schema_index = load_storage_schema_index(LocalObjectStore(str(workspace)), config.incremental_prefix)
    result = build_fivetran_schema(SnapshotTables(tables=()), schema_index, config)

    assert result == []