#!/usr/bin/env python3
"""
Local end-to-end test for TiDB-Fivetran connector.

Translated from e2e-test/run_local_e2e.sh

Usage:
    python tests/e2e/run_local_e2e.py

Environment variables (all optional, with defaults):
    PYTHON_VENV=.venv          # Python virtual environment path
    SKIP_VENV=0                # Set to 1 to skip virtual environment activation
    TIDB_VERSION=v8.5.6        # TiDB version for playground
    TIUP_TAG=<random>          # TiUP playground tag (auto-generated if not set)
    CHANGEFEED_ID=tidb-fivetran-local-e2e  # TiCDC changefeed ID
    DB_NAME=test               # Database name
    TABLE_NAME=fruit1           # Table name
    DUCKDB_TABLE=tester.test_fruit_1  # DuckDB table name
    WORKSPACE=examples/local_workspace  # Workspace directory
    CONFIG_PATH=configuration.json  # Config file path
    WAREHOUSE=files/warehouse.db  # DuckDB warehouse path
    E2E_RUNTIME_DIR=e2e-test/.tmp/local-e2e  # Runtime directory
    TEST_SQL_PATH=tests/e2e/test.sql  # Test SQL file path
    TIDB_HOST=127.0.0.1        # TiDB host
    TIDB_PORT=4000             # TiDB port
    TIDB_USER=root             # TiDB user
    TIDB_PASSWORD=             # TiDB password (empty by default)
    TICDC_SERVER=127.0.0.1:8300  # TiCDC server
    DUMPLING_THREADS=8         # Dumpling threads
    DUMPLING_ROWS=200000       # Dumpling rows
    DUMPLING_FILE_SIZE=64MiB   # Dumpling file size
    READY_TIMEOUT_SECONDS=180  # Ready timeout
    FIVETRAN_DEBUG_TIMEOUT_SECONDS=180  # Fivetran debug timeout
    PLAYGROUND_LOG_TAIL_LINES=200  # Log tail lines
    KEEP_PLAYGROUND=0          # Keep playground running after test
    TIMEOUT_BIN=timeout        # Timeout binary
"""

import contextlib
import os
import re
import shutil
import signal
import subprocess
import sys
import time
import uuid
from pathlib import Path


class E2ETestError(Exception):
    """Base exception for E2E test errors."""

    pass


def is_ci() -> bool:
    """Check if running in CI environment."""
    return os.getenv("CI", "").lower() in ("true", "1", "yes") or os.getenv("GITHUB_ACTIONS", "").lower() == "true"


class Config:
    """Configuration manager for E2E test."""

    def __init__(self):
        # Python environment
        self.python_venv = os.getenv("PYTHON_VENV", ".venv")
        self.skip_venv = os.getenv("SKIP_VENV", "0") == "1"

        # TiDB/TiUP configuration
        self.tidb_version = os.getenv("TIDB_VERSION", "v8.5.6")
        # Generate random tag if not specified
        self.tiup_tag = os.getenv("TIUP_TAG", f"tidb-fivetran-e2e-{uuid.uuid4().hex[:8]}")
        self.changefeed_id = os.getenv("CHANGEFEED_ID", "tidb-fivetran-local-e2e")

        # Database configuration
        self.db_name = os.getenv("DB_NAME", "test")
        self.table_name = os.getenv("TABLE_NAME", "fruit1")
        self.duckdb_table = os.getenv("DUCKDB_TABLE", "tester.test_fruit_1")

        # Paths
        self.workspace = os.getenv("WORKSPACE", "examples/local_workspace")
        self.config_path = os.getenv("CONFIG_PATH", "configuration.json")
        self.warehouse = os.getenv("WAREHOUSE", "files/warehouse.db")
        self.runtime_dir = os.getenv("E2E_RUNTIME_DIR", "e2e-test/.tmp/local-e2e")
        self.test_sql_path = os.getenv("TEST_SQL_PATH", "tests/e2e/test.sql")

        # TiDB connection
        self.tidb_host = os.getenv("TIDB_HOST", "127.0.0.1")
        self.tidb_port = int(os.getenv("TIDB_PORT", "4000"))
        self.tidb_user = os.getenv("TIDB_USER", "root")
        self.tidb_password = os.getenv("TIDB_PASSWORD", "")

        # TiCDC configuration
        self.ticdc_server = os.getenv("TICDC_SERVER", "127.0.0.1:8300")

        # Dumpling configuration
        self.dumpling_threads = int(os.getenv("DUMPLING_THREADS", "8"))
        self.dumpling_rows = int(os.getenv("DUMPLING_ROWS", "200000"))
        self.dumpling_file_size = os.getenv("DUMPLING_FILE_SIZE", "64MiB")

        # Timeouts
        self.ready_timeout_seconds = int(os.getenv("READY_TIMEOUT_SECONDS", "180"))
        self.fivetran_debug_timeout_seconds = int(os.getenv("FIVETRAN_DEBUG_TIMEOUT_SECONDS", "180"))
        self.playground_log_tail_lines = int(os.getenv("PLAYGROUND_LOG_TAIL_LINES", "200"))

        # Other
        self.keep_playground = os.getenv("KEEP_PLAYGROUND", "0") == "1"
        self.timeout_bin = os.getenv("TIMEOUT_BIN", "timeout")

        # CI environment
        self.is_ci = is_ci()

        # Derived paths
        self.snapshot_dir = f"{self.workspace}/snapshot"
        self.increment_dir = f"{self.workspace}/increment"
        self.playground_log = f"{self.runtime_dir}/playground.log"
        self.ticdc_config = f"{self.runtime_dir}/ticdc.toml"
        self.changefeed_config = f"{self.runtime_dir}/changefeed.toml"
        self.tiup_data_dir = f"{Path.home()}/.tiup/data/{self.tiup_tag}"

        # State
        self.created_config = False
        self.playground_pid = None


class Logger:
    """Simple logger."""

    @staticmethod
    def log(message: str):
        """Log a message."""
        print(message)

    @staticmethod
    def error(message: str):
        """Log an error message to stderr."""
        print(f"ERROR: {message}", file=sys.stderr)

    @staticmethod
    def run_log(command: list[str]):
        """Log a command being run."""
        print(f"+ {' '.join(command)}")


class CommandRunner:
    """Command execution utilities."""

    @staticmethod
    def run(command: list[str], check: bool = True, capture_output: bool = False) -> subprocess.CompletedProcess:
        """Run a command."""
        Logger.run_log(command)
        try:
            result = subprocess.run(command, check=check, capture_output=capture_output, text=True)
            return result
        except subprocess.CalledProcessError as e:
            if check:
                raise E2ETestError(f"command failed ({e.returncode}): {' '.join(command)}") from e
            return e

    @staticmethod
    def run_background(command: list[str]) -> subprocess.Popen:
        """Run a command in background."""
        Logger.run_log(command)
        try:
            process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
            return process
        except Exception as e:
            raise E2ETestError(f"failed to start background command: {e}") from e

    @staticmethod
    def require_command(command: str):
        """Check if a command is available."""
        if shutil.which(command) is None:
            raise E2ETestError(f"missing required command: {command}")


class DatabaseManager:
    """Database operations."""

    def __init__(self, config: Config):
        self.config = config

    def mysql_exec(self, sql: str):
        """Execute SQL in TiDB."""
        command = [
            "mysql",
            "--protocol=tcp",
            "-h",
            self.config.tidb_host,
            "-P",
            str(self.config.tidb_port),
            "-u",
            self.config.tidb_user,
            f"--password={self.config.tidb_password}",
            "-e",
            sql,
        ]
        CommandRunner.run(command)

    def mysql_try_exec(self, sql: str):
        """Try to execute SQL in TiDB, ignoring failures."""
        try:
            command = [
                "mysql",
                "--protocol=tcp",
                "-h",
                self.config.tidb_host,
                "-P",
                str(self.config.tidb_port),
                "-u",
                self.config.tidb_user,
                f"--password={self.config.tidb_password}",
                "-e",
                sql,
            ]
            CommandRunner.run(command, check=False)
        except Exception:
            Logger.log(f"Optional TiDB setting failed and was ignored: {sql}")

    def mysql_exec_file(self, file_path: str):
        """Execute SQL file in TiDB."""
        if not Path(file_path).exists():
            raise E2ETestError(f"missing SQL file: {file_path}")

        command = [
            "mysql",
            "--protocol=tcp",
            "-h",
            self.config.tidb_host,
            "-P",
            str(self.config.tidb_port),
            "-u",
            self.config.tidb_user,
            f"--password={self.config.tidb_password}",
        ]

        with open(file_path) as f:
            subprocess.run(command, stdin=f, check=True, text=True)

    def duckdb_exec(self, sql: str, capture_output: bool = False) -> subprocess.CompletedProcess | None:
        """Execute SQL in DuckDB."""
        command = ["duckdb", self.config.warehouse, sql]
        return CommandRunner.run(command, capture_output=capture_output)


class TestValidator:
    """Test validation utilities."""

    def __init__(self, config: Config, db_manager: DatabaseManager):
        self.config = config
        self.db_manager = db_manager

    def validate_identifier(self, value: str, label: str):
        """Validate a SQL identifier."""
        if not re.match(r"^[A-Za-z_][A-Za-z0-9_]*$", value):
            raise E2ETestError(f"invalid {label}: {value}")

    def wait_until(self, description: str, timeout_seconds: int, check_func, *args):
        """Wait until a condition is met or timeout."""
        deadline = time.time() + timeout_seconds
        while time.time() < deadline:
            if check_func(*args):
                return
            time.sleep(2)
        raise E2ETestError(f"{description} did not become ready within {timeout_seconds}s")

    def tidb_ready(self) -> bool:
        """Check if TiDB is ready."""
        try:
            command = [
                "mysql",
                "--protocol=tcp",
                "-h",
                self.config.tidb_host,
                "-P",
                str(self.config.tidb_port),
                "-u",
                self.config.tidb_user,
                f"--password={self.config.tidb_password}",
                "-e",
                "select 1",
            ]
            result = subprocess.run(command, capture_output=True, text=True)
            return result.returncode == 0
        except Exception:
            return False

    def ticdc_ready(self) -> bool:
        """Check if TiCDC is ready."""
        try:
            # First check if the port is accessible
            import socket

            host, port = self.config.ticdc_server.split(":")
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(2)
            result = sock.connect_ex((host, int(port)))
            sock.close()
            if result != 0:
                return False

            # Then try the CDC command
            command = [
                "tiup",
                f"ctl:{self.config.tidb_version}",
                "cdc",
                "changefeed",
                "list",
                f"--server={self.config.ticdc_server}",
            ]
            result = subprocess.run(command, capture_output=True, text=True, timeout=30)
            return result.returncode == 0
        except Exception:
            return False

    def has_file_matching(self, directory: str, pattern: str) -> bool:
        """Check if directory has files matching pattern."""
        try:
            result = subprocess.run(
                ["find", directory, "-type", "f", "-name", pattern, "-print", "-quit"], capture_output=True, text=True
            )
            return bool(result.stdout.strip())
        except Exception:
            return False

    def wait_for_increment_outputs(self):
        """Wait for TiCDC to write increment files."""
        deadline = time.time() + self.config.ready_timeout_seconds
        while time.time() < deadline:
            if (
                Path(self.config.increment_dir, "metadata").exists()
                and self.has_file_matching(self.config.increment_dir, "CDC*.csv")
                and self.has_file_matching(self.config.increment_dir, "CDC.index")
                and self.has_file_matching(self.config.increment_dir, "schema_*.json")
            ):
                return
            time.sleep(2)
        raise E2ETestError(f"TiCDC did not write expected increment files under {self.config.increment_dir}")

    def metadata_checkpoint_ts(self) -> int | None:
        """Get checkpoint timestamp from metadata."""
        metadata_path = Path(self.config.increment_dir, "metadata")
        if not metadata_path.exists():
            return None

        try:
            with open(metadata_path) as f:
                content = f.read()
                match = re.search(r'"checkpoint-ts"\s*:\s*"?(\d+)"?', content)
                if match:
                    return int(match.group(1))
        except Exception:
            pass
        return None

    def increment_row_commit_ts(self, needle: str) -> int | None:
        """Get commit timestamp from increment row."""
        try:
            result = subprocess.run(
                [
                    "find",
                    self.config.increment_dir,
                    "-type",
                    "f",
                    "-name",
                    "CDC*.csv",
                    "-exec",
                    "grep",
                    "-h",
                    "-F",
                    needle,
                    "{}",
                    "+",
                ],
                capture_output=True,
                text=True,
            )
            if result.stdout.strip():
                lines = result.stdout.strip().split("\n")
                if lines:
                    parts = lines[0].split(",")
                    if len(parts) >= 4:
                        return int(parts[3])
        except Exception:
            pass
        return None

    def final_ddl_row_checkpoint_ready(self) -> bool:
        """Check if final DDL row checkpoint is ready."""
        row_commit_ts = self.increment_row_commit_ts("fruit-after-truncate")
        checkpoint_ts = self.metadata_checkpoint_ts()

        if not isinstance(row_commit_ts, int) or not isinstance(checkpoint_ts, int):
            return False

        return checkpoint_ts >= row_commit_ts

    def test_sql_final_row_checkpoint_ready(self) -> bool:
        """Check if test.sql final row checkpoint is ready."""
        row_commit_ts = self.increment_row_commit_ts("cross-family value after ddl")
        checkpoint_ts = self.metadata_checkpoint_ts()

        if not isinstance(row_commit_ts, int) or not isinstance(checkpoint_ts, int):
            return False

        return checkpoint_ts >= row_commit_ts

    def wait_for_duckdb_table(self):
        """Wait for DuckDB table to be created."""
        self.wait_for_duckdb_relation(self.config.duckdb_table)

    def wait_for_duckdb_relation(self, relation: str):
        """Wait for DuckDB relation to be created."""
        deadline = time.time() + 30
        while time.time() < deadline:
            try:
                result = subprocess.run(
                    ["duckdb", self.config.warehouse, f"select 1 from {relation} limit 1;"],
                    capture_output=True,
                    text=True,
                )
                if result.returncode == 0:
                    return
            except Exception:
                pass
            time.sleep(1)

        # Show available tables for debugging
        try:
            result = subprocess.run(["duckdb", self.config.warehouse, "show tables;"], capture_output=True, text=True)
            Logger.log("DuckDB tables:")
            Logger.log(result.stdout)
        except Exception:
            pass

        raise E2ETestError(f"DuckDB relation {relation} was not created")

    def verify_duckdb_rows(self):
        """Verify DuckDB rows match expected data."""
        expected_rows_sql = f"""
        with expected_rows(id, name, color, sweetness, deleted) as (
            values
                (1, 'apple', null, null, true),
                (2, 'banana', null, null, true),
                (3, 'cherry', null, null, true),
                (4, 'date', null, null, true),
                (5, 'elderberry', null, null, true),
                (6, 'fig', null, null, true),
                (7, 'grape', null, null, true),
                (8, 'honeydew', null, null, true),
                (9, 'kiwi', null, null, true),
                (10, 'lemon', null, null, true),
                (11, 'fruit-11', null, null, true),
                (12, 'fruit-12', 'red', null, true),
                (13, 'fruit-13', 'yellow', 7, true),
                (14, 'fruit-14-long-name-after-widen', 'purple', 8, true),
                (100, 'fruit-after-truncate', 'green', 9, false)
        ),
        actual_rows as (
            select id, name, color, sweetness, coalesce(_fivetran_deleted, false) as deleted
            from {self.config.duckdb_table}
        ),
        missing_expected as (
            select id, name, color, sweetness, deleted from expected_rows
            except all
            select id, name, color, sweetness, deleted from actual_rows
        ),
        unexpected_actual as (
            select id, name, color, sweetness, deleted from actual_rows
            except all
            select id, name, color, sweetness, deleted from expected_rows
        )
        select
            (select count(*) from missing_expected) +
            (select count(*) from unexpected_actual);
        """

        result = subprocess.run(
            ["duckdb", "-noheader", "-csv", self.config.warehouse, expected_rows_sql], capture_output=True, text=True
        )
        row_diff_count = result.stdout.strip()

        if not row_diff_count or int(row_diff_count) != 0:
            Logger.log("DuckDB row data mismatch. Missing expected rows:")
            missing_sql = f"""
            with expected_rows(id, name, color, sweetness, deleted) as (
                values
                    (1, 'apple', null, null, true),
                    (2, 'banana', null, null, true),
                    (3, 'cherry', null, null, true),
                    (4, 'date', null, null, true),
                    (5, 'elderberry', null, null, true),
                    (6, 'fig', null, null, true),
                    (7, 'grape', null, null, true),
                    (8, 'honeydew', null, null, true),
                    (9, 'kiwi', null, null, true),
                    (10, 'lemon', null, null, true),
                    (11, 'fruit-11', null, null, true),
                    (12, 'fruit-12', 'red', null, true),
                    (13, 'fruit-13', 'yellow', 7, true),
                    (14, 'fruit-14-long-name-after-widen', 'purple', 8, true),
                    (100, 'fruit-after-truncate', 'green', 9, false)
            ),
            actual_rows as (
                select id, name, color, sweetness, coalesce(_fivetran_deleted, false) as deleted
                from {self.config.duckdb_table}
            )
            select id, name, color, sweetness, deleted from expected_rows
            except all
            select id, name, color, sweetness, deleted from actual_rows
            order by id;
            """

            subprocess.run(["duckdb", self.config.warehouse, missing_sql], capture_output=True, text=True)

            Logger.log("Unexpected actual rows:")
            unexpected_sql = f"""
            with expected_rows(id, name, color, sweetness, deleted) as (
                values
                    (1, 'apple', null, null, true),
                    (2, 'banana', null, null, true),
                    (3, 'cherry', null, null, true),
                    (4, 'date', null, null, true),
                    (5, 'elderberry', null, null, true),
                    (6, 'fig', null, null, true),
                    (7, 'grape', null, null, true),
                    (8, 'honeydew', null, null, true),
                    (9, 'kiwi', null, null, true),
                    (10, 'lemon', null, null, true),
                    (11, 'fruit-11', null, null, true),
                    (12, 'fruit-12', 'red', null, true),
                    (13, 'fruit-13', 'yellow', 7, true),
                    (14, 'fruit-14-long-name-after-widen', 'purple', 8, true),
                    (100, 'fruit-after-truncate', 'green', 9, false)
            ),
            actual_rows as (
                select id, name, color, sweetness, coalesce(_fivetran_deleted, false) as deleted
                from {self.config.duckdb_table}
            )
            select id, name, color, sweetness, deleted from actual_rows
            except all
            select id, name, color, sweetness, deleted from expected_rows
            order by id;
            """

            subprocess.run(["duckdb", self.config.warehouse, unexpected_sql], capture_output=True, text=True)
            raise E2ETestError("DuckDB source rows do not exactly match expected rows")

        # Check active row count
        active_sql = f"""
        select count(*)
        from {self.config.duckdb_table}
        where coalesce(_fivetran_deleted, false) = false;
        """
        result = subprocess.run(
            ["duckdb", "-noheader", "-csv", self.config.warehouse, active_sql], capture_output=True, text=True
        )
        active_row_count = result.stdout.strip()

        if not active_row_count or int(active_row_count) != 1:
            raise E2ETestError(
                f"expected exactly one active post-truncate row in {self.config.duckdb_table}, got {active_row_count}"
            )

        # Check pre-truncate deleted count
        pre_truncate_sql = f"""
        select count(*)
        from {self.config.duckdb_table}
        where id between 1 and 14 and coalesce(_fivetran_deleted, false) = true;
        """
        result = subprocess.run(
            ["duckdb", "-noheader", "-csv", self.config.warehouse, pre_truncate_sql], capture_output=True, text=True
        )
        pre_truncate_deleted_count = result.stdout.strip()

        if not pre_truncate_deleted_count or int(pre_truncate_deleted_count) != 14:
            raise E2ETestError(
                f"expected 14 soft-deleted pre-truncate rows in {self.config.duckdb_table}, got {pre_truncate_deleted_count}"
            )

        # Check no PK table count
        no_pk_sql = """
        select count(*)
        from information_schema.tables
        where table_schema = 'tester' and table_name = 'test_ddl_no_pk';
        """
        result = subprocess.run(
            ["duckdb", "-noheader", "-csv", self.config.warehouse, no_pk_sql], capture_output=True, text=True
        )
        no_pk_table_count = result.stdout.strip()

        if not no_pk_table_count or int(no_pk_table_count) != 0:
            raise E2ETestError(
                f"expected no DuckDB table for source table without primary key, got {no_pk_table_count}"
            )

        # Check metadata mismatch
        metadata_sql = f"""
        select count(*)
        from {self.config.duckdb_table}
        where
            _incremental_source_schema <> '{self.config.db_name}'
            or _incremental_source_table <> '{self.config.table_name}'
            or (id between 1 and 10 and coalesce(_incremental_commit_ts, -1) <> 0)
            or (id in (11, 12, 13, 14, 100) and coalesce(_incremental_commit_ts, 0) <= 0);
        """
        result = subprocess.run(
            ["duckdb", "-noheader", "-csv", self.config.warehouse, metadata_sql], capture_output=True, text=True
        )
        metadata_mismatch_count = result.stdout.strip()

        if not metadata_mismatch_count or int(metadata_mismatch_count) != 0:
            Logger.log("DuckDB stable metadata columns do not match expected values")
            debug_sql = f"""
            select id, name, color, sweetness, _incremental_commit_ts, _incremental_source_schema, _incremental_source_table, _fivetran_deleted
            from {self.config.duckdb_table}
            where
                _incremental_source_schema <> '{self.config.db_name}'
                or _incremental_source_table <> '{self.config.table_name}'
                or (id between 1 and 10 and coalesce(_incremental_commit_ts, -1) <> 0)
                or (id in (11, 12, 13, 14, 100) and coalesce(_incremental_commit_ts, 0) <= 0)
            order by id;
            """
            subprocess.run(["duckdb", self.config.warehouse, debug_sql], capture_output=True, text=True)
            raise E2ETestError("DuckDB stable metadata columns do not match expected values")

    def verify_test_sql_duckdb_results(self):
        """Verify test.sql DuckDB results."""
        matrix_table = "tester.ft_ddl_type_test_ft_ddl_type_matrix"
        create_table = "tester.ft_ddl_type_test_ft_ddl_create_after_snapshot"
        truncate_table = "tester.ft_ddl_type_test_ft_ddl_truncate_target"

        # Check matrix rows
        matrix_sql = f"""
        with expected_matrix_rows(
            id,
            c_varchar,
            c_added_nullable,
            c_default_toggle,
            c_changed_rename_type,
            c_type_cross_family_number
        ) as (
            values
                (1, 'after ddl update', 'after ddl update non-null', 'explicit after default drop', 9001, 'number column changed to varchar'),
                (3, null, null, 'v1', null, null),
                (4, null, null, 'v2', null, null),
                (5, null, 'insert after ddl', 'explicit_insert_v5', 9005, 'cross-family value after ddl'),
                (6, null, null, null, null, null),
                (7, null, null, null, null, null),
                (8, null, null, null, null, null)
        ),
        actual_matrix_rows as (
            select
                id,
                c_varchar,
                c_added_nullable,
                c_default_toggle,
                c_changed_rename_type,
                c_type_cross_family_number
            from {matrix_table}
            where coalesce(_fivetran_deleted, false) = false
        ),
        missing_expected as (
            select * from expected_matrix_rows
            except all
            select * from actual_matrix_rows
        ),
        unexpected_actual as (
            select * from actual_matrix_rows
            except all
            select * from expected_matrix_rows
        )
        select
            (select count(*) from missing_expected) +
            (select count(*) from unexpected_actual);
        """

        result = subprocess.run(
            ["duckdb", "-noheader", "-csv", self.config.warehouse, matrix_sql], capture_output=True, text=True
        )
        matrix_diff_count = result.stdout.strip()

        if not matrix_diff_count or int(matrix_diff_count) != 0:
            Logger.log("DuckDB ft_ddl_type_matrix active row mismatch. Active rows:")
            debug_sql = f"""
            select
                id,
                c_varchar,
                c_added_nullable,
                c_default_toggle,
                c_changed_rename_type,
                c_type_cross_family_number,
                _fivetran_deleted
            from {matrix_table}
            order by id;
            """
            subprocess.run(["duckdb", self.config.warehouse, debug_sql], capture_output=True, text=True)
            raise E2ETestError("DuckDB ft_ddl_type_matrix active rows do not match test.sql expectations")

        # Check deleted id=2
        deleted_sql = f"""
        select count(*)
        from {matrix_table}
        where id = 2 and coalesce(_fivetran_deleted, false) = true;
        """
        result = subprocess.run(
            ["duckdb", "-noheader", "-csv", self.config.warehouse, deleted_sql], capture_output=True, text=True
        )
        deleted_id_two_count = result.stdout.strip()

        if not deleted_id_two_count or int(deleted_id_two_count) != 1:
            raise E2ETestError(f"expected ft_ddl_type_matrix id=2 to be soft deleted, got {deleted_id_two_count}")

        # Check create row
        create_sql = f"""
        select count(*)
        from {create_table}
        where
            id = 1
            and payload = 'new table with stable primary key'
            and coalesce(_fivetran_deleted, false) = false;
        """
        result = subprocess.run(
            ["duckdb", "-noheader", "-csv", self.config.warehouse, create_sql], capture_output=True, text=True
        )
        create_row_count = result.stdout.strip()

        if not create_row_count or int(create_row_count) != 1:
            raise E2ETestError(f"expected ft_ddl_create_after_snapshot row to sync, got {create_row_count}")

        # Check truncate rows
        truncate_active_sql = f"""
        select count(*)
        from {truncate_table}
        where id = 3 and payload = 'after truncate' and coalesce(_fivetran_deleted, false) = false;
        """
        result = subprocess.run(
            ["duckdb", "-noheader", "-csv", self.config.warehouse, truncate_active_sql], capture_output=True, text=True
        )
        truncate_active_count = result.stdout.strip()

        truncate_deleted_sql = f"""
        select count(*)
        from {truncate_table}
        where id in (1, 2) and coalesce(_fivetran_deleted, false) = true;
        """
        result = subprocess.run(
            ["duckdb", "-noheader", "-csv", self.config.warehouse, truncate_deleted_sql], capture_output=True, text=True
        )
        truncate_deleted_count = result.stdout.strip()

        if (
            not truncate_active_count
            or int(truncate_active_count) != 1
            or not truncate_deleted_count
            or int(truncate_deleted_count) != 2
        ):
            raise E2ETestError(
                f"expected ft_ddl_truncate_target truncate semantics, active={truncate_active_count} deleted={truncate_deleted_count}"
            )

        # Check no stable PK table
        no_stable_pk_sql = """
        select count(*)
        from information_schema.tables
        where table_schema = 'tester' and table_name = 'ft_ddl_type_test_ft_ddl_no_stable_pk';
        """
        result = subprocess.run(
            ["duckdb", "-noheader", "-csv", self.config.warehouse, no_stable_pk_sql], capture_output=True, text=True
        )
        no_stable_pk_table_count = result.stdout.strip()

        if not no_stable_pk_table_count or int(no_stable_pk_table_count) != 0:
            raise E2ETestError(f"expected no DuckDB table for ft_ddl_no_stable_pk, got {no_stable_pk_table_count}")


class E2ETest:
    """Main E2E test orchestrator."""

    def __init__(self):
        self.config = Config()
        self.db_manager = DatabaseManager(self.config)
        self.validator = TestValidator(self.config, self.db_manager)
        # Use tiup ctl:{version} cdc as the CDC CLI command
        self.cdc_cli = ["tiup", f"ctl:{self.config.tidb_version}", "cdc"]

    def cleanup(self):
        """Cleanup resources."""
        Logger.log(f"Cleaning up with TIUP_TAG={self.config.tiup_tag}")

        if self.config.playground_pid:
            if self.config.keep_playground:
                Logger.log(
                    f"Leaving playground running with pid {self.config.playground_pid} because KEEP_PLAYGROUND=1"
                )
            else:
                with contextlib.suppress(ProcessLookupError, PermissionError):
                    os.killpg(os.getpgid(self.config.playground_pid), signal.SIGTERM)
                with contextlib.suppress(ProcessLookupError, PermissionError):
                    os.kill(self.config.playground_pid, signal.SIGTERM)

                time.sleep(2)

                with contextlib.suppress(ProcessLookupError, PermissionError):
                    os.killpg(os.getpgid(self.config.playground_pid), signal.SIGKILL)
                with contextlib.suppress(ProcessLookupError, PermissionError):
                    os.kill(self.config.playground_pid, signal.SIGKILL)

                with contextlib.suppress(ChildProcessError, ProcessLookupError):
                    os.waitpid(self.config.playground_pid, 0)

        if self.config.created_config:
            with contextlib.suppress(Exception):
                Path(self.config.config_path).unlink(missing_ok=True)

        # Clean up runtime configs
        with contextlib.suppress(Exception):
            Path(self.config.ticdc_config).unlink(missing_ok=True)
            Path(self.config.changefeed_config).unlink(missing_ok=True)

        # Always clean up TiUP tag
        try:
            Logger.log(f"Running tiup clean {self.config.tiup_tag}")
            subprocess.run(["tiup", "clean", self.config.tiup_tag], capture_output=True, check=False)
        except Exception as e:
            Logger.log(f"Warning: tiup clean failed: {e}")

        # Kill tagged tiup processes
        try:
            if shutil.which("pgrep"):
                result = subprocess.run(["pgrep", "-f", self.config.tiup_data_dir], capture_output=True, text=True)
                if result.stdout.strip():
                    pids = result.stdout.strip().split("\n")
                    for pid in pids:
                        with contextlib.suppress(ProcessLookupError, ValueError):
                            os.kill(int(pid), signal.SIGTERM)
                    time.sleep(2)

                    # Force kill if still alive
                    result = subprocess.run(["pgrep", "-f", self.config.tiup_data_dir], capture_output=True, text=True)
                    if result.stdout.strip():
                        pids = result.stdout.strip().split("\n")
                        for pid in pids:
                            with contextlib.suppress(ProcessLookupError, ValueError):
                                os.kill(int(pid), signal.SIGKILL)
        except Exception:
            pass

        # Clean up runtime directory (skip in CI to allow artifact upload)
        if not self.config.is_ci:
            try:
                if Path(self.config.runtime_dir).exists():
                    shutil.rmtree(self.config.runtime_dir)
                    Logger.log(f"Removed runtime directory: {self.config.runtime_dir}")
            except Exception as e:
                Logger.log(f"Warning: failed to remove runtime directory: {e}")

    def activate_python(self):
        """Activate Python virtual environment if needed."""
        if self.config.skip_venv or os.getenv("VIRTUAL_ENV"):
            return

        activate_path = Path(self.config.python_venv) / "bin" / "activate"
        if activate_path.exists():
            # Source the virtual environment
            activate_cmd = f"source {activate_path}"
            subprocess.run(activate_cmd, shell=True, check=True)
            return

        if shutil.which("fivetran"):
            Logger.log(f"No {activate_path} found; using current Python environment.")
            return

        raise E2ETestError(f"missing {activate_path} and fivetran is not on PATH")

    def write_runtime_configs(self):
        """Write runtime configuration files."""
        Path(self.config.runtime_dir).mkdir(parents=True, exist_ok=True)

        with open(self.config.ticdc_config, "w") as f:
            f.write("newarch = true\n")

        with open(self.config.changefeed_config, "w") as f:
            f.write("[sink]\n[sink.csv]\ninclude-commit-ts = true\n")

    def prepare_test_sql_environment(self):
        """Prepare test SQL environment."""
        self.db_manager.mysql_try_exec("set global tidb_enable_check_constraint = on;")
        self.db_manager.mysql_try_exec("set global tidb_enable_foreign_key = on;")

    def apply_ddl_test_cases(self):
        """Apply DDL test cases."""
        ddl_sql = f"""
        use {self.config.db_name};
        insert into {self.config.table_name}(name) values ('fruit-11');
        alter table {self.config.table_name} add column color varchar(20) default null;
        insert into {self.config.table_name}(name, color) values ('fruit-12', 'red');
        alter table {self.config.table_name} add column sweetness int not null default 0;
        insert into {self.config.table_name}(name, color, sweetness) values ('fruit-13', 'yellow', 7);
        alter table {self.config.table_name} modify column name varchar(64) not null default '';
        insert into {self.config.table_name}(name, color, sweetness) values ('fruit-14-long-name-after-widen', 'purple', 8);
        alter table {self.config.table_name} comment = 'local e2e ddl coverage';
        create table ddl_no_pk(note varchar(20));
        insert into ddl_no_pk(note) values ('ignored');
        truncate table {self.config.table_name};
        insert into {self.config.table_name}(id, name, color, sweetness) values (100, 'fruit-after-truncate', 'green', 9);
        """
        self.db_manager.mysql_exec(ddl_sql)

    def run_fivetran_debug_cycle(self, phase: str):
        """Run fivetran debug cycle."""
        Logger.log(f"+ fivetran debug --configuration {self.config.config_path} ({phase})")

        try:
            result = subprocess.run(
                [
                    self.config.timeout_bin,
                    f"{self.config.fivetran_debug_timeout_seconds}s",
                    "fivetran",
                    "debug",
                    "--configuration",
                    self.config.config_path,
                ],
                capture_output=True,
                text=True,
                timeout=self.config.fivetran_debug_timeout_seconds + 10,
            )

            if result.returncode == 124:
                Logger.log(
                    f"fivetran debug timed out after {self.config.fivetran_debug_timeout_seconds}s during {phase}; continuing to verification"
                )
            elif result.returncode != 0:
                Logger.log(f"fivetran debug stdout: {result.stdout}")
                Logger.log(f"fivetran debug stderr: {result.stderr}")
                raise E2ETestError(f"fivetran debug failed during {phase} with exit code {result.returncode}")
        except subprocess.TimeoutExpired:
            Logger.log(
                f"fivetran debug timed out after {self.config.fivetran_debug_timeout_seconds}s during {phase}; continuing to verification"
            )

    def run(self):
        """Run the E2E test."""
        status = 0
        Logger.log(f"Starting E2E test with TIUP_TAG={self.config.tiup_tag}")
        Logger.log(f"CI environment: {self.config.is_ci}")

        try:
            # Activate Python environment
            self.activate_python()

            # Check required commands
            for tool in [
                "tiup",
                "mysql",
                "fivetran",
                "duckdb",
                self.config.timeout_bin,
                "find",
                "realpath",
                "grep",
                "tr",
                "sed",
                "awk",
            ]:
                CommandRunner.require_command(tool)

            # Check required files
            if not Path("examples/local_configuration.json").exists():
                raise E2ETestError("missing examples/local_configuration.json")
            if not Path(self.config.test_sql_path).exists():
                raise E2ETestError(f"missing {self.config.test_sql_path}")

            # Validate identifiers
            self.validator.validate_identifier(self.config.db_name, "DB_NAME")
            self.validator.validate_identifier(self.config.table_name, "TABLE_NAME")

            # Create config if needed
            if not Path(self.config.config_path).exists():
                shutil.copy("examples/local_configuration.json", self.config.config_path)
                self.config.created_config = True

            # Clean up old files
            for path in [
                self.config.snapshot_dir,
                self.config.increment_dir,
                self.config.warehouse,
                "files/state.json",
            ]:
                if Path(path).exists():
                    if Path(path).is_dir():
                        shutil.rmtree(path)
                    else:
                        Path(path).unlink()

            # Create directories
            Path(self.config.workspace).mkdir(parents=True, exist_ok=True)
            Path("files").mkdir(parents=True, exist_ok=True)

            # Write runtime configs
            self.write_runtime_configs()

            # Clean TiUP
            subprocess.run(["tiup", "clean", self.config.tiup_tag], capture_output=True, check=False)

            # Start playground
            Logger.log(
                f"+ tiup playground {self.config.tidb_version} --tag {self.config.tiup_tag} --db 1 --pd 1 --kv 1 --tiflash 0 --ticdc 1 --ticdc.config {self.config.ticdc_config}"
            )

            with open(self.config.playground_log, "w") as log_file:
                process = subprocess.Popen(
                    [
                        "tiup",
                        "playground",
                        self.config.tidb_version,
                        "--tag",
                        self.config.tiup_tag,
                        "--db",
                        "1",
                        "--pd",
                        "1",
                        "--kv",
                        "1",
                        "--tiflash",
                        "0",
                        "--ticdc",
                        "1",
                        "--ticdc.config",
                        self.config.ticdc_config,
                    ],
                    stdout=log_file,
                    stderr=subprocess.STDOUT,
                    preexec_fn=os.setsid,
                )
                self.config.playground_pid = process.pid

            # Wait for services to be ready
            self.validator.wait_until("TiDB", self.config.ready_timeout_seconds, self.validator.tidb_ready)
            self.validator.wait_until("TiCDC", self.config.ready_timeout_seconds, self.validator.ticdc_ready)

            # Seed data
            seed_sql = f"""
            drop database if exists {self.config.db_name};
            create database {self.config.db_name};
            use {self.config.db_name};
            create table {self.config.table_name}(id bigint not null auto_increment primary key, name varchar(20) not null default '');
            insert into {self.config.table_name}(name) values ('apple'), ('banana'), ('cherry'), ('date'), ('elderberry'), ('fig'), ('grape'), ('honeydew'), ('kiwi'), ('lemon');
            """
            self.db_manager.mysql_exec(seed_sql)

            # Run dumpling
            dumpling_cmd = [
                "tiup",
                "dumpling",
                "-u",
                self.config.tidb_user,
                "-P",
                str(self.config.tidb_port),
                "-h",
                self.config.tidb_host,
                "--filetype",
                "csv",
                "-t",
                str(self.config.dumpling_threads),
                "-r",
                str(self.config.dumpling_rows),
                "-F",
                self.config.dumpling_file_size,
                "-B",
                self.config.db_name,
                "-o",
                f"./{self.config.snapshot_dir}",
            ]
            if self.config.tidb_password:
                dumpling_cmd.extend(["-p", self.config.tidb_password])

            CommandRunner.run(dumpling_cmd)

            # Verify dumpling output
            if not Path(self.config.snapshot_dir, "metadata").exists() or not self.validator.has_file_matching(
                self.config.snapshot_dir, f"{self.config.db_name}.{self.config.table_name}.*.csv"
            ):
                raise E2ETestError(
                    f"Dumpling did not create expected CSV snapshot files under {self.config.snapshot_dir}"
                )

            # Run fivetran debug for snapshot
            self.run_fivetran_debug_cycle("snapshot")

            # Create TiCDC changefeed
            increment_abs = str(Path(self.config.increment_dir).resolve())
            sink_uri = f"file://{increment_abs}?protocol=csv&output-old-value=true"

            CommandRunner.run(
                [
                    *self.cdc_cli,
                    "changefeed",
                    "create",
                    f"--server={self.config.ticdc_server}",
                    f"--sink-uri={sink_uri}",
                    f"--changefeed-id={self.config.changefeed_id}",
                    f"--config={self.config.changefeed_config}",
                    "--no-confirm",
                ]
            )

            # Apply DDL test cases
            self.apply_ddl_test_cases()

            # Prepare test SQL environment
            self.prepare_test_sql_environment()

            # Execute test SQL
            self.db_manager.mysql_exec_file(self.config.test_sql_path)

            # Wait for increment outputs
            self.validator.wait_for_increment_outputs()

            # Wait for checkpoints
            self.validator.wait_until(
                "TiCDC final DDL row checkpoint",
                self.config.ready_timeout_seconds,
                self.validator.final_ddl_row_checkpoint_ready,
            )
            self.validator.wait_until(
                "TiCDC test.sql final row checkpoint",
                self.config.ready_timeout_seconds,
                self.validator.test_sql_final_row_checkpoint_ready,
            )

            # Run fivetran debug for incremental
            self.run_fivetran_debug_cycle("incremental")

            # Verify warehouse exists
            if not Path(self.config.warehouse).exists():
                raise E2ETestError(f"missing DuckDB warehouse: {self.config.warehouse}")

            # Wait for DuckDB tables
            self.validator.wait_for_duckdb_table()
            self.validator.wait_for_duckdb_relation("tester.ft_ddl_type_test_ft_ddl_type_matrix")
            self.validator.wait_for_duckdb_relation("tester.ft_ddl_type_test_ft_ddl_create_after_snapshot")
            self.validator.wait_for_duckdb_relation("tester.ft_ddl_type_test_ft_ddl_truncate_target")

            # Verify row counts
            result = subprocess.run(
                [
                    "duckdb",
                    "-noheader",
                    "-csv",
                    self.config.warehouse,
                    f"select count(*) from {self.config.duckdb_table};",
                ],
                capture_output=True,
                text=True,
            )
            row_count = result.stdout.strip()

            result = subprocess.run(
                [
                    "duckdb",
                    "-noheader",
                    "-csv",
                    self.config.warehouse,
                    f"select count(*) from {self.config.duckdb_table} where name = 'fruit-after-truncate' and coalesce(_fivetran_deleted, false) = false;",
                ],
                capture_output=True,
                text=True,
            )
            incremental_count = result.stdout.strip()

            if not row_count or int(row_count) != 15:
                raise E2ETestError(
                    f"expected exactly 15 rows in {self.config.duckdb_table} after DDL coverage, got {row_count}"
                )

            if not incremental_count or int(incremental_count) < 1:
                raise E2ETestError(
                    f"expected active post-truncate row fruit-after-truncate in {self.config.duckdb_table}"
                )

            # Verify data
            self.validator.verify_duckdb_rows()
            self.validator.verify_test_sql_duckdb_results()

            Logger.log(
                f"OK: DuckDB {self.config.warehouse} contains exactly matching DDL lifecycle rows and test.sql matrix results"
            )

        except Exception as e:
            status = 1
            Logger.error(str(e))

            # Show playground log tail on failure
            if Path(self.config.playground_log).exists():
                Logger.log(f"\n===== playground log tail ({self.config.playground_log_tail_lines} lines) =====")
                try:
                    with open(self.config.playground_log) as f:
                        lines = f.readlines()
                        for line in lines[-self.config.playground_log_tail_lines :]:
                            Logger.log(line.rstrip())
                except Exception:
                    pass
                Logger.log("===== end playground log tail =====")
        finally:
            self.cleanup()

        return status


def main():
    """Main entry point."""
    test = E2ETest()
    return test.run()


if __name__ == "__main__":
    sys.exit(main())