"""Narrow MySQL operations used by the integration regression."""

from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Dict, List, Sequence

import pymysql


TABLE = "active_cluster_ids"


def open_connection(defaults_file: Path) -> Any:
    if not defaults_file.is_file():
        raise ValueError(
            f"MySQL defaults file is not a regular file: {defaults_file}"
        )
    return pymysql.connect(
        read_default_file=str(defaults_file),
        read_default_group="client",
        connect_timeout=30,
        read_timeout=120,
        write_timeout=120,
        autocommit=False,
        charset="utf8mb4",
    )


def all_clusters(cursor: Any) -> List[str]:
    cursor.execute(f"SELECT cluster_id FROM {TABLE} ORDER BY cluster_id")
    clusters = [row[0] for row in cursor.fetchall()]
    if any(not isinstance(cluster, str) or not cluster for cluster in clusters):
        raise ValueError(f"{TABLE} contains an invalid cluster_id")
    if clusters != sorted(set(clusters)):
        raise ValueError(f"{TABLE} contains duplicate cluster_id values")
    return clusters


def eligible_clusters(cursor: Any, cutoff: datetime) -> List[str]:
    cursor.execute(
        f"SELECT cluster_id FROM {TABLE} "
        "WHERE updated_at > s3_data_updated_at AND updated_at <= %s "
        "ORDER BY cluster_id",
        (cutoff,),
    )
    clusters = [row[0] for row in cursor.fetchall()]
    if any(not isinstance(cluster, str) or not cluster for cluster in clusters):
        raise ValueError(f"{TABLE} returned an invalid eligible cluster_id")
    return clusters


def require_minimum_clusters(
    clusters: Sequence[str], minimum_clusters: int
) -> None:
    if len(clusters) < minimum_clusters:
        raise ValueError(
            f"expected at least {minimum_clusters} target clusters, "
            f"found {len(clusters)}"
        )


def seed_database(
    defaults_file: Path,
    sql: str,
    minimum_clusters: int,
) -> Dict[str, int]:
    statement = "\n".join(
        line for line in sql.splitlines() if not line.startswith("--")
    ).lstrip()
    if not statement.startswith("REPLACE INTO `active_cluster_ids`"):
        raise ValueError("seed SQL does not target active_cluster_ids")

    connection = open_connection(defaults_file)
    try:
        with connection.cursor() as cursor:
            cursor.execute('SET time_zone = "+00:00"')
            cursor.execute(f"DELETE FROM {TABLE}")
            cursor.execute(statement)
            cursor.execute(f"SELECT COUNT(*) FROM {TABLE}")
            cluster_count = int(cursor.fetchone()[0])
            if cluster_count < minimum_clusters:
                raise ValueError(
                    f"backup produced {cluster_count} cluster(s); at least "
                    f"{minimum_clusters} are required"
                )
        connection.commit()
    except BaseException:
        connection.rollback()
        raise
    finally:
        connection.close()
    return {"seeded_clusters": cluster_count}


def prepare_database(
    defaults_file: Path,
    cutoff: datetime,
    minimum_clusters: int,
) -> Dict[str, Any]:
    connection = open_connection(defaults_file)
    try:
        with connection.cursor() as cursor:
            cursor.execute("SET time_zone = '+00:00'")
            clusters = all_clusters(cursor)
            require_minimum_clusters(clusters, minimum_clusters)
            created_at = cutoff - timedelta(days=30)
            updated_at = cutoff - timedelta(seconds=1)
            watermark = cutoff - timedelta(days=7)
            cursor.execute(
                f"UPDATE {TABLE} SET created_at = %s, updated_at = %s, "
                "s3_data_updated_at = %s",
                (created_at, updated_at, watermark),
            )
        connection.commit()
    except BaseException:
        connection.rollback()
        raise
    finally:
        connection.close()
    return {
        "action": "prepare",
        "cluster_count": len(clusters),
        "updated_at": updated_at.isoformat(sep=" "),
        "s3_data_updated_at": watermark.isoformat(sep=" "),
    }


def advance_updated_at(
    defaults_file: Path,
    earlier_cutoff: datetime,
    later_cutoff: datetime,
    minimum_clusters: int,
) -> Dict[str, Any]:
    if earlier_cutoff >= later_cutoff:
        raise ValueError("later cutoff must be after earlier cutoff")
    updated_at = earlier_cutoff + (later_cutoff - earlier_cutoff) / 2
    if not earlier_cutoff < updated_at <= later_cutoff:
        raise ValueError("cannot choose an updated_at between backup cutoffs")

    connection = open_connection(defaults_file)
    try:
        with connection.cursor() as cursor:
            cursor.execute("SET time_zone = '+00:00'")
            clusters = all_clusters(cursor)
            require_minimum_clusters(clusters, minimum_clusters)
            cursor.execute(
                f"SELECT cluster_id, s3_data_updated_at FROM {TABLE} "
                "ORDER BY cluster_id"
            )
            watermarks_before = list(cursor.fetchall())
            cursor.execute(
                f"UPDATE {TABLE} SET updated_at = %s",
                (updated_at,),
            )
            cursor.execute(
                f"SELECT cluster_id, s3_data_updated_at FROM {TABLE} "
                "ORDER BY cluster_id"
            )
            watermarks_after = list(cursor.fetchall())
            if watermarks_after != watermarks_before:
                raise ValueError("advancing updated_at changed a watermark")
            eligible = eligible_clusters(cursor, later_cutoff)
            if eligible != clusters:
                raise ValueError(
                    "updated cluster set is not fully eligible for the "
                    f"later backup: eligible={len(eligible)} "
                    f"target={len(clusters)}"
                )
        connection.commit()
    except BaseException:
        connection.rollback()
        raise
    finally:
        connection.close()
    return {
        "action": "advance_updated_at",
        "cluster_count": len(clusters),
        "updated_at": updated_at.isoformat(sep=" "),
        "preserved_watermarks": len(watermarks_after),
    }


def list_clusters(
    defaults_file: Path,
    minimum_clusters: int,
) -> List[str]:
    connection = open_connection(defaults_file)
    try:
        with connection.cursor() as cursor:
            cursor.execute("SET time_zone = '+00:00'")
            clusters = all_clusters(cursor)
            require_minimum_clusters(clusters, minimum_clusters)
        connection.rollback()
        return clusters
    finally:
        connection.close()


def list_eligible_clusters(
    defaults_file: Path,
    cutoff: datetime,
    minimum_clusters: int,
) -> List[str]:
    connection = open_connection(defaults_file)
    try:
        with connection.cursor() as cursor:
            cursor.execute("SET time_zone = '+00:00'")
            clusters = all_clusters(cursor)
            require_minimum_clusters(clusters, minimum_clusters)
            eligible = eligible_clusters(cursor, cutoff)
        connection.rollback()
        return eligible
    finally:
        connection.close()


def assert_all_clusters_eligible(
    defaults_file: Path,
    cutoff: datetime,
    minimum_clusters: int,
) -> Dict[str, int]:
    connection = open_connection(defaults_file)
    try:
        with connection.cursor() as cursor:
            cursor.execute("SET time_zone = '+00:00'")
            clusters = all_clusters(cursor)
            require_minimum_clusters(clusters, minimum_clusters)
            eligible = eligible_clusters(cursor, cutoff)
        connection.rollback()
    finally:
        connection.close()
    if eligible != clusters:
        raise ValueError(
            "eligible cluster set differs from target set: "
            f"eligible={len(eligible)} target={len(clusters)}"
        )
    return {
        "target_clusters": len(clusters),
        "eligible_clusters": len(eligible),
    }