from __future__ import annotations

import json
import sqlite3
import uuid
from dataclasses import asdict, dataclass
from datetime import timedelta
from typing import Any, Literal

from app.core.time import from_iso, to_iso, utc_now
from app.database.database import Database

ROTATION_LEASE_TTL_SECONDS = 300

RUN_COLUMNS = """
id, target_type, target_id, target_name, trigger, status, active_slot,
current_item_id, lease_owner, lease_until, error_code, error_message,
total_items, succeeded_items, started_at, updated_at, finished_at
"""

ITEM_COLUMNS = """
id, run_id, instance_id, position, status, stage, stage_started_at,
attempt_count, config_version, instance_display_name, aws_region,
lightsail_instance_name, aws_account_id, cloudflare_account_id,
cloudflare_zone_name, cloudflare_zone_id,
cloudflare_record_name, cloudflare_proxied, proxy_protocol, socks_port,
proxy_health_check,
health_timeout_seconds, release_grace_seconds, old_static_ip_name, old_ip,
    new_static_ip_name, new_ip, dns_ip_before, dns_ip_after, grace_until,
    aws_operation_id, error_code, error_message, rollback_from_stage,
    rollback_reason_code, rollback_reason_message, started_at, updated_at, finished_at
"""


class RotationLeaseLostError(RuntimeError):
    pass


@dataclass(slots=True)
class FleetRun:
    id: str
    target_type: str
    target_id: str
    target_name: str
    trigger: str
    status: str
    active_slot: int | None
    current_item_id: str | None
    lease_owner: str | None
    lease_until: str | None
    error_code: str | None
    error_message: str | None
    total_items: int
    succeeded_items: int
    started_at: str
    updated_at: str
    finished_at: str | None

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)


@dataclass(slots=True)
class FleetRunItem:
    id: str
    run_id: str
    instance_id: str
    position: int
    status: str
    stage: str
    stage_started_at: str
    attempt_count: int
    config_version: int
    instance_display_name: str
    aws_account_id: str | None
    cloudflare_account_id: str | None
    aws_region: str
    lightsail_instance_name: str
    cloudflare_zone_name: str
    cloudflare_zone_id: str
    cloudflare_record_name: str
    cloudflare_proxied: int
    proxy_protocol: str
    socks_port: int
    proxy_health_check: int
    health_timeout_seconds: int
    release_grace_seconds: int
    old_static_ip_name: str | None
    old_ip: str | None
    new_static_ip_name: str | None
    new_ip: str | None
    dns_ip_before: str | None
    dns_ip_after: str | None
    grace_until: str | None
    aws_operation_id: str | None
    error_code: str | None
    error_message: str | None
    rollback_from_stage: str | None
    rollback_reason_code: str | None
    rollback_reason_message: str | None
    started_at: str
    updated_at: str
    finished_at: str | None

    def to_dict(self) -> dict[str, Any]:
        data = asdict(self)
        data["proxy_health_check"] = bool(self.proxy_health_check)
        data["cloudflare_proxied"] = bool(self.cloudflare_proxied)
        return data


class RotationRepository:
    def __init__(self, database: Database) -> None:
        self.database = database

    def create_for_instance(
        self,
        instance_id: str,
        trigger: Literal["manual", "scheduled", "recovery"] = "manual",
    ) -> FleetRun:
        return self._create_run("instance", instance_id, trigger)

    def create_for_group(
        self,
        group_id: str,
        trigger: Literal["manual", "scheduled", "recovery"] = "manual",
    ) -> FleetRun:
        return self._create_run("group", group_id, trigger)

    def target_account_bindings(
        self,
        target_type: str,
        target_id: str,
    ) -> list[tuple[str | None, str | None]]:
        with self.database.connect() as connection:
            if target_type == "instance":
                rows = connection.execute(
                    """
                    SELECT aws_account_id, cloudflare_account_id
                    FROM managed_instances
                    WHERE id = ? AND archived_at IS NULL AND enabled = 1
                    """,
                    (target_id,),
                ).fetchall()
            else:
                rows = connection.execute(
                    """
                    SELECT instance.aws_account_id, instance.cloudflare_account_id
                    FROM instance_group_members AS member
                    JOIN managed_instances AS instance ON instance.id = member.instance_id
                    WHERE member.group_id = ? AND instance.archived_at IS NULL
                        AND instance.enabled = 1
                    ORDER BY member.position, instance.created_at
                    """,
                    (target_id,),
                ).fetchall()
        return [
            (
                str(row["aws_account_id"]) if row["aws_account_id"] else None,
                str(row["cloudflare_account_id"]) if row["cloudflare_account_id"] else None,
            )
            for row in rows
        ]

    def _create_run(self, target_type: str, target_id: str, trigger: str) -> FleetRun:
        now = to_iso()
        run_id = str(uuid.uuid4())
        with self.database.connect() as connection:
            try:
                connection.execute("BEGIN IMMEDIATE")
                self._clear_expired_dns_lock(connection, now)
                if connection.execute(
                    """
                    SELECT 1 FROM fleet_operation_locks WHERE slot = 1
                    UNION ALL
                    SELECT 1 FROM fleet_runs WHERE active_slot = 1
                    LIMIT 1
                    """
                ).fetchone():
                    raise RuntimeError("ACTIVE_ROTATION")

                if target_type == "instance":
                    target = connection.execute(
                        """
                        SELECT * FROM managed_instances
                        WHERE id = ? AND archived_at IS NULL AND enabled = 1
                        """,
                        (target_id,),
                    ).fetchone()
                    if target is None:
                        raise RuntimeError("INSTANCE_NOT_FOUND")
                    target_name = str(target["display_name"])
                    instances = [target]
                else:
                    group = connection.execute(
                        """
                        SELECT * FROM instance_groups
                        WHERE id = ? AND archived_at IS NULL
                        """,
                        (target_id,),
                    ).fetchone()
                    if group is None:
                        raise RuntimeError("GROUP_NOT_FOUND")
                    if trigger == "scheduled" and not bool(group["enabled"]):
                        raise RuntimeError("GROUP_SCHEDULE_DISABLED")
                    target_name = str(group["name"])
                    instances = connection.execute(
                        """
                        SELECT instance.*
                        FROM instance_group_members AS member
                        JOIN managed_instances AS instance ON instance.id = member.instance_id
                        WHERE member.group_id = ? AND instance.archived_at IS NULL
                            AND instance.enabled = 1
                        ORDER BY member.position, instance.created_at
                        """,
                        (target_id,),
                    ).fetchall()
                    if not instances:
                        raise RuntimeError("GROUP_HAS_NO_ENABLED_INSTANCES")

                connection.execute(
                    """
                    INSERT INTO fleet_operation_locks(slot, kind, owner_id, lease_until, created_at)
                    VALUES (1, 'rotation', ?, NULL, ?)
                    """,
                    (run_id, now),
                )
                connection.execute(
                    """
                    INSERT INTO fleet_runs(
                        id, target_type, target_id, target_name, trigger, status,
                        active_slot, total_items, succeeded_items, started_at, updated_at
                    ) VALUES (?, ?, ?, ?, ?, 'queued', 1, ?, 0, ?, ?)
                    """,
                    (
                        run_id,
                        target_type,
                        target_id,
                        target_name,
                        trigger,
                        len(instances),
                        now,
                        now,
                    ),
                )
                first_item_id: str | None = None
                for position, instance in enumerate(instances):
                    item_id = str(uuid.uuid4())
                    first_item_id = first_item_id or item_id
                    connection.execute(
                        """
                        INSERT INTO fleet_run_items(
                            id, run_id, instance_id, position, status, stage,
                            stage_started_at, config_version, instance_display_name,
                            aws_region, lightsail_instance_name, aws_account_id,
                            cloudflare_account_id, cloudflare_zone_name,
                            cloudflare_zone_id, cloudflare_record_name,
                            cloudflare_proxied, proxy_protocol, socks_port,
                            proxy_health_check, health_timeout_seconds,
                            release_grace_seconds, started_at, updated_at
                        ) VALUES (
                            ?, ?, ?, ?, 'queued', 'preflight', ?, ?, ?, ?, ?, ?, ?, ?,
                            ?, ?, ?, ?, ?, ?, ?, ?, ?, ?
                        )
                        """,
                        (
                            item_id,
                            run_id,
                            instance["id"],
                            position,
                            now,
                            instance["config_version"],
                            instance["display_name"],
                            instance["aws_region"],
                            instance["lightsail_instance_name"],
                            instance["aws_account_id"],
                            instance["cloudflare_account_id"],
                            instance["cloudflare_zone_name"],
                            instance["cloudflare_zone_id"],
                            instance["cloudflare_record_name"],
                            instance["cloudflare_proxied"],
                            instance["proxy_protocol"],
                            instance["socks_port"],
                            instance["proxy_health_check"],
                            instance["health_timeout_seconds"],
                            instance["release_grace_seconds"],
                            now,
                            now,
                        ),
                    )
                connection.execute(
                    "UPDATE fleet_runs SET current_item_id = ? WHERE id = ?",
                    (first_item_id, run_id),
                )
                connection.execute(
                    """
                    INSERT INTO fleet_events(
                        run_id, item_id, occurred_at, stage, level, message, details_json
                    ) VALUES (?, ?, ?, 'queued', 'info', ?, ?)
                    """,
                    (
                        run_id,
                        first_item_id,
                        now,
                        "轮换批次已创建",
                        json.dumps({"target_type": target_type, "items": len(instances)}),
                    ),
                )
                if target_type == "group":
                    connection.execute(
                        """
                        UPDATE instance_groups SET next_run_at = ?, updated_at = ?
                        WHERE id = ?
                        """,
                        (None, now, target_id),
                    )
                connection.commit()
            except sqlite3.IntegrityError as exc:
                connection.rollback()
                raise RuntimeError("ACTIVE_ROTATION") from exc
            except Exception:
                connection.rollback()
                raise
        run = self.get(run_id)
        if run is None:  # pragma: no cover
            raise RuntimeError("RUN_NOT_FOUND")
        return run

    def get(self, run_id: str) -> FleetRun | None:
        with self.database.connect() as connection:
            row = connection.execute(
                f"SELECT {RUN_COLUMNS} FROM fleet_runs WHERE id = ?", (run_id,)
            ).fetchone()
        return FleetRun(**dict(row)) if row else None

    def get_active(self) -> FleetRun | None:
        with self.database.connect() as connection:
            row = connection.execute(
                f"SELECT {RUN_COLUMNS} FROM fleet_runs WHERE active_slot = 1"
            ).fetchone()
        return FleetRun(**dict(row)) if row else None

    def get_item(self, item_id: str) -> FleetRunItem | None:
        with self.database.connect() as connection:
            row = connection.execute(
                f"SELECT {ITEM_COLUMNS} FROM fleet_run_items WHERE id = ?", (item_id,)
            ).fetchone()
        return FleetRunItem(**dict(row)) if row else None

    def current_item(self, run_id: str) -> FleetRunItem | None:
        with self.database.connect() as connection:
            row = connection.execute(
                f"""
                SELECT {", ".join(f"item.{part.strip()}" for part in ITEM_COLUMNS.split(","))}
                FROM fleet_runs AS run
                JOIN fleet_run_items AS item ON item.id = run.current_item_id
                WHERE run.id = ?
                """,
                (run_id,),
            ).fetchone()
        return FleetRunItem(**dict(row)) if row else None

    def list_items(self, run_id: str) -> list[FleetRunItem]:
        with self.database.connect() as connection:
            rows = connection.execute(
                f"""
                SELECT {ITEM_COLUMNS} FROM fleet_run_items
                WHERE run_id = ? ORDER BY position
                """,
                (run_id,),
            ).fetchall()
        return [FleetRunItem(**dict(row)) for row in rows]

    def list_events(self, run_id: str) -> list[dict[str, Any]]:
        with self.database.connect() as connection:
            rows = connection.execute(
                """
                SELECT id, item_id, occurred_at, stage, level, message, details_json
                FROM fleet_events WHERE run_id = ? ORDER BY id
                """,
                (run_id,),
            ).fetchall()
        return [
            {
                "id": int(row["id"]),
                "item_id": row["item_id"],
                "occurred_at": row["occurred_at"],
                "stage": row["stage"],
                "level": row["level"],
                "message": row["message"],
                "details": json.loads(row["details_json"]),
            }
            for row in rows
        ]

    def list_runs(
        self,
        page: int,
        per_page: int,
        *,
        instance_id: str | None = None,
        group_id: str | None = None,
        status: str | None = None,
    ) -> tuple[list[FleetRun], int]:
        clauses: list[str] = []
        values: list[Any] = []
        if instance_id:
            clauses.append(
                "EXISTS (SELECT 1 FROM fleet_run_items item "
                "WHERE item.run_id = run.id AND item.instance_id = ?)"
            )
            values.append(instance_id)
        if group_id:
            clauses.append("run.target_type = 'group' AND run.target_id = ?")
            values.append(group_id)
        if status:
            clauses.append("run.status = ?")
            values.append(status)
        where = f"WHERE {' AND '.join(clauses)}" if clauses else ""
        offset = (page - 1) * per_page
        with self.database.connect() as connection:
            total = int(
                connection.execute(
                    f"SELECT COUNT(*) FROM fleet_runs AS run {where}", values
                ).fetchone()[0]
            )
            rows = connection.execute(
                f"""
                SELECT {", ".join(f"run.{part.strip()}" for part in RUN_COLUMNS.split(","))}
                FROM fleet_runs AS run {where}
                ORDER BY run.started_at DESC LIMIT ? OFFSET ?
                """,
                [*values, per_page, offset],
            ).fetchall()
        return [FleetRun(**dict(row)) for row in rows], total

    def history_summary(self) -> dict[str, int | str | None]:
        with self.database.connect() as connection:
            row = connection.execute(
                """
                SELECT COUNT(*) AS total,
                    SUM(CASE WHEN status = 'succeeded' THEN 1 ELSE 0 END) AS succeeded,
                    SUM(CASE WHEN status IN ('succeeded','failed','cancelled') THEN 1 ELSE 0 END)
                        AS completed,
                    MAX(CASE WHEN status = 'succeeded' THEN finished_at END) AS latest_success_at
                FROM fleet_runs
                """
            ).fetchone()
        return {
            "total": int(row["total"] or 0),
            "completed": int(row["completed"] or 0),
            "succeeded": int(row["succeeded"] or 0),
            "latest_success_at": row["latest_success_at"],
        }

    def acquire_lease(
        self,
        run_id: str,
        owner: str,
        ttl_seconds: int = ROTATION_LEASE_TTL_SECONDS,
    ) -> bool:
        now = utc_now()
        now_iso = to_iso(now)
        lease_until = to_iso(now + timedelta(seconds=ttl_seconds))
        with self.database.connect() as connection, connection:
            cursor = connection.execute(
                """
                UPDATE fleet_runs SET lease_owner = ?, lease_until = ?, updated_at = ?
                WHERE id = ? AND active_slot = 1
                    AND status IN ('queued','running')
                    AND (lease_owner IS NULL OR lease_until < ? OR lease_owner = ?)
                """,
                (owner, lease_until, now_iso, run_id, now_iso, owner),
            )
        return cursor.rowcount == 1

    def renew_lease(
        self,
        run_id: str,
        owner: str,
        ttl_seconds: int = ROTATION_LEASE_TTL_SECONDS,
    ) -> bool:
        now = utc_now()
        now_iso = to_iso(now)
        lease_until = to_iso(now + timedelta(seconds=ttl_seconds))
        with self.database.connect() as connection, connection:
            cursor = connection.execute(
                """
                UPDATE fleet_runs SET lease_until = ?, updated_at = ?
                WHERE id = ? AND active_slot = 1 AND lease_owner = ?
                    AND lease_until >= ? AND status IN ('queued','running')
                """,
                (lease_until, now_iso, run_id, owner, now_iso),
            )
        return cursor.rowcount == 1

    def transition_run(
        self,
        run_id: str,
        *,
        status: str,
        expected_owner: str,
    ) -> FleetRun:
        now = to_iso()
        with self.database.connect() as connection, connection:
            cursor = connection.execute(
                """
                UPDATE fleet_runs SET status = ?, updated_at = ?
                WHERE id = ? AND active_slot = 1 AND lease_owner = ? AND lease_until >= ?
                """,
                (status, now, run_id, expected_owner, now),
            )
            if cursor.rowcount != 1:
                raise RotationLeaseLostError("ROTATION_LEASE_LOST")
        run = self.get(run_id)
        if run is None:  # pragma: no cover
            raise RuntimeError("RUN_NOT_FOUND")
        return run

    def transition_item(
        self,
        run_id: str,
        item_id: str,
        *,
        expected_owner: str,
        stage: str | None = None,
        status: str | None = None,
        **fields: Any,
    ) -> FleetRunItem:
        allowed = {
            "old_static_ip_name",
            "old_ip",
            "new_static_ip_name",
            "new_ip",
            "dns_ip_before",
            "dns_ip_after",
            "grace_until",
            "aws_operation_id",
            "error_code",
            "error_message",
            "cloudflare_zone_id",
            "rollback_from_stage",
            "rollback_reason_code",
            "rollback_reason_message",
        }
        unknown = set(fields) - allowed
        if unknown:
            raise ValueError(f"Unsupported item fields: {sorted(unknown)}")
        now = to_iso()
        updates = dict(fields)
        if stage is not None:
            updates["stage"] = stage
            updates["stage_started_at"] = now
        if status is not None:
            updates["status"] = status
        updates["updated_at"] = now
        assignments = ", ".join(f"{name} = ?" for name in updates)
        values = [*updates.values(), item_id, run_id, expected_owner, now]
        with self.database.connect() as connection, connection:
            cursor = connection.execute(
                f"""
                UPDATE fleet_run_items SET {assignments}
                WHERE id = ? AND run_id = ? AND EXISTS (
                    SELECT 1 FROM fleet_runs run WHERE run.id = fleet_run_items.run_id
                        AND run.active_slot = 1 AND run.lease_owner = ?
                        AND run.lease_until >= ? AND run.status IN ('queued','running')
                )
                """,
                values,
            )
            if cursor.rowcount != 1:
                raise RotationLeaseLostError("ROTATION_LEASE_LOST")
        item = self.get_item(item_id)
        if item is None:  # pragma: no cover
            raise RuntimeError("ITEM_NOT_FOUND")
        return item

    def add_event(
        self,
        run_id: str,
        *,
        stage: str,
        message: str,
        item_id: str | None = None,
        level: str = "info",
        details: dict[str, Any] | None = None,
        expected_owner: str | None = None,
    ) -> None:
        now = to_iso()
        with self.database.connect() as connection:
            try:
                connection.execute("BEGIN IMMEDIATE")
                if expected_owner is not None:
                    lease = connection.execute(
                        """
                        SELECT 1 FROM fleet_runs WHERE id = ? AND active_slot = 1
                            AND lease_owner = ? AND lease_until >= ?
                            AND status IN ('queued','running')
                        """,
                        (run_id, expected_owner, now),
                    ).fetchone()
                    if lease is None:
                        raise RotationLeaseLostError("ROTATION_LEASE_LOST")
                connection.execute(
                    """
                    INSERT INTO fleet_events(
                        run_id, item_id, occurred_at, stage, level, message, details_json
                    ) VALUES (?, ?, ?, ?, ?, ?, ?)
                    """,
                    (
                        run_id,
                        item_id,
                        now,
                        stage,
                        level,
                        message,
                        json.dumps(details or {}, ensure_ascii=False),
                    ),
                )
                connection.commit()
            except Exception:
                connection.rollback()
                raise

    def complete_item(self, run_id: str, item_id: str, expected_owner: str) -> bool:
        now = to_iso()
        with self.database.connect() as connection:
            try:
                connection.execute("BEGIN IMMEDIATE")
                lease = connection.execute(
                    """
                    SELECT target_type, target_id FROM fleet_runs
                    WHERE id = ? AND active_slot = 1 AND lease_owner = ?
                        AND lease_until >= ? AND status IN ('queued','running')
                    """,
                    (run_id, expected_owner, now),
                ).fetchone()
                if lease is None:
                    raise RotationLeaseLostError("ROTATION_LEASE_LOST")
                connection.execute(
                    """
                    UPDATE fleet_run_items SET status = 'succeeded', stage = 'succeeded',
                        error_code = NULL, error_message = NULL, updated_at = ?, finished_at = ?
                    WHERE id = ? AND run_id = ?
                    """,
                    (now, now, item_id, run_id),
                )
                completed_item = connection.execute(
                    "SELECT instance_id, new_ip FROM fleet_run_items WHERE id = ?",
                    (item_id,),
                ).fetchone()
                connection.execute(
                    """
                    UPDATE managed_instances SET last_known_ip = ?, last_checked_at = ?,
                        updated_at = ? WHERE id = ?
                    """,
                    (
                        completed_item["new_ip"],
                        now,
                        now,
                        completed_item["instance_id"],
                    ),
                )
                next_item = connection.execute(
                    """
                    SELECT id FROM fleet_run_items
                    WHERE run_id = ? AND status = 'queued'
                    ORDER BY position LIMIT 1
                    """,
                    (run_id,),
                ).fetchone()
                succeeded = int(
                    connection.execute(
                        """
                        SELECT COUNT(*) FROM fleet_run_items
                        WHERE run_id = ? AND status = 'succeeded'
                        """,
                        (run_id,),
                    ).fetchone()[0]
                )
                if next_item is not None:
                    connection.execute(
                        """
                        UPDATE fleet_runs SET current_item_id = ?, succeeded_items = ?,
                            status = 'running', updated_at = ? WHERE id = ?
                        """,
                        (next_item["id"], succeeded, now, run_id),
                    )
                    connection.execute(
                        """
                        INSERT INTO fleet_events(
                            run_id,item_id,occurred_at,stage,level,message,details_json
                        ) VALUES (?, ?, ?, 'preflight', 'info', '开始下一个实例', '{}')
                        """,
                        (run_id, next_item["id"], now),
                    )
                    connection.commit()
                    return False

                connection.execute(
                    """
                    UPDATE fleet_runs SET status = 'succeeded', active_slot = NULL,
                        current_item_id = NULL, lease_owner = NULL, lease_until = NULL,
                        succeeded_items = ?, error_code = NULL, error_message = NULL,
                        updated_at = ?, finished_at = ? WHERE id = ?
                    """,
                    (succeeded, now, now, run_id),
                )
                connection.execute(
                    "DELETE FROM fleet_operation_locks WHERE owner_id = ?", (run_id,)
                )
                if lease["target_type"] == "group":
                    self._advance_group_schedule(connection, str(lease["target_id"]), now)
                connection.commit()
                return True
            except Exception:
                connection.rollback()
                raise

    def fail_execution(
        self,
        run_id: str,
        item_id: str,
        *,
        error_code: str,
        error_message: str,
        outcome: Literal["failed", "needs_attention", "cleanup_pending"],
        expected_owner: str,
    ) -> FleetRun:
        now = to_iso()
        with self.database.connect() as connection:
            try:
                connection.execute("BEGIN IMMEDIATE")
                row = connection.execute(
                    """
                    SELECT item.stage, run.target_type, run.target_id FROM fleet_runs run
                    JOIN fleet_run_items item ON item.id = run.current_item_id
                    WHERE run.id = ? AND item.id = ? AND run.active_slot = 1
                        AND run.lease_owner = ? AND run.lease_until >= ?
                    """,
                    (run_id, item_id, expected_owner, now),
                ).fetchone()
                if row is None:
                    raise RotationLeaseLostError("ROTATION_LEASE_LOST")
                connection.execute(
                    """
                    INSERT INTO fleet_events(
                        run_id,item_id,occurred_at,stage,level,message,details_json
                    ) VALUES (?, ?, ?, ?, 'error', ?, ?)
                    """,
                    (
                        run_id,
                        item_id,
                        now,
                        row["stage"],
                        error_message,
                        json.dumps({"code": error_code}, ensure_ascii=False),
                    ),
                )
                finished_at = now if outcome == "failed" else None
                connection.execute(
                    """
                    UPDATE fleet_run_items SET status = ?, error_code = ?, error_message = ?,
                        updated_at = ?, finished_at = ? WHERE id = ?
                    """,
                    (outcome, error_code, error_message, now, finished_at, item_id),
                )
                if outcome == "failed":
                    connection.execute(
                        """
                        UPDATE fleet_run_items SET status = 'cancelled', stage = 'cancelled',
                            error_code = 'BATCH_ABORTED',
                            error_message = '同一批次中的前序实例失败，未执行此实例',
                            updated_at = ?, finished_at = ?
                        WHERE run_id = ? AND status = 'queued'
                        """,
                        (now, now, run_id),
                    )
                    connection.execute(
                        """
                        UPDATE fleet_runs SET status = 'failed', active_slot = NULL,
                            lease_owner = NULL, lease_until = NULL, error_code = ?,
                            error_message = ?, updated_at = ?, finished_at = ? WHERE id = ?
                        """,
                        (error_code, error_message, now, now, run_id),
                    )
                    connection.execute(
                        "DELETE FROM fleet_operation_locks WHERE owner_id = ?", (run_id,)
                    )
                    if row["target_type"] == "group":
                        self._advance_group_schedule(
                            connection,
                            str(row["target_id"]),
                            now,
                        )
                else:
                    connection.execute(
                        """
                        UPDATE fleet_runs SET status = ?, lease_owner = NULL,
                            lease_until = NULL, error_code = ?, error_message = ?, updated_at = ?
                        WHERE id = ?
                        """,
                        (outcome, error_code, error_message, now, run_id),
                    )
                connection.commit()
            except Exception:
                connection.rollback()
                raise
        run = self.get(run_id)
        if run is None:  # pragma: no cover
            raise RuntimeError("RUN_NOT_FOUND")
        return run

    def prepare_resume(self, run_id: str) -> FleetRun:
        now = to_iso()
        with self.database.connect() as connection:
            try:
                connection.execute("BEGIN IMMEDIATE")
                run = connection.execute(
                    """
                    SELECT current_item_id FROM fleet_runs
                    WHERE id = ? AND active_slot = 1
                        AND status IN ('needs_attention','cleanup_pending')
                    """,
                    (run_id,),
                ).fetchone()
                if run is None:
                    raise RuntimeError("RUN_NOT_RESUMABLE")
                connection.execute(
                    """
                    UPDATE fleet_runs SET status = 'running', trigger = 'recovery',
                        lease_owner = NULL, lease_until = NULL, error_code = NULL,
                        error_message = NULL, updated_at = ? WHERE id = ?
                    """,
                    (now, run_id),
                )
                connection.execute(
                    """
                    UPDATE fleet_run_items SET status = 'running',
                        attempt_count = attempt_count + 1, error_code = NULL,
                        error_message = NULL, updated_at = ? WHERE id = ?
                    """,
                    (now, run["current_item_id"]),
                )
                connection.execute(
                    """
                    INSERT INTO fleet_events(
                        run_id,item_id,occurred_at,stage,level,message,details_json
                    ) SELECT ?, id, ?, stage, 'info', '管理员已继续当前任务', '{}'
                      FROM fleet_run_items WHERE id = ?
                    """,
                    (run_id, now, run["current_item_id"]),
                )
                connection.commit()
            except Exception:
                connection.rollback()
                raise
        resumed = self.get(run_id)
        if resumed is None:  # pragma: no cover
            raise RuntimeError("RUN_NOT_FOUND")
        return resumed

    def cancel(self, run_id: str) -> FleetRun:
        now = to_iso()
        message = "任务已由管理员放弃，请人工核对 AWS 静态 IP 与 Cloudflare DNS"
        with self.database.connect() as connection:
            try:
                connection.execute("BEGIN IMMEDIATE")
                run = connection.execute(
                    """
                    SELECT target_type, target_id, current_item_id FROM fleet_runs
                    WHERE id = ? AND active_slot = 1
                        AND status IN ('needs_attention','cleanup_pending')
                    """,
                    (run_id,),
                ).fetchone()
                if run is None:
                    raise RuntimeError("RUN_NOT_CANCELABLE")
                connection.execute(
                    """
                    UPDATE fleet_run_items SET status = 'cancelled', error_code =
                        CASE WHEN id = ? THEN 'ROTATION_CANCELLED' ELSE error_code END,
                        error_message = CASE WHEN id = ? THEN ? ELSE error_message END,
                        updated_at = ?, finished_at = COALESCE(finished_at, ?)
                    WHERE run_id = ? AND status != 'succeeded'
                    """,
                    (run["current_item_id"], run["current_item_id"], message, now, now, run_id),
                )
                connection.execute(
                    """
                    UPDATE fleet_runs SET status = 'cancelled', active_slot = NULL,
                        lease_owner = NULL, lease_until = NULL, error_code = 'ROTATION_CANCELLED',
                        error_message = ?, updated_at = ?, finished_at = ? WHERE id = ?
                    """,
                    (message, now, now, run_id),
                )
                connection.execute(
                    "DELETE FROM fleet_operation_locks WHERE owner_id = ?", (run_id,)
                )
                if run["target_type"] == "group":
                    connection.execute(
                        """
                        UPDATE instance_groups SET enabled = 0, next_run_at = NULL, updated_at = ?
                        WHERE id = ?
                        """,
                        (now, run["target_id"]),
                    )
                else:
                    connection.execute(
                        """
                        UPDATE instance_groups SET enabled = 0, next_run_at = NULL, updated_at = ?
                        WHERE id IN (
                            SELECT group_id FROM instance_group_members WHERE instance_id = ?
                        )
                        """,
                        (now, run["target_id"]),
                    )
                connection.execute(
                    """
                    INSERT INTO fleet_events(
                        run_id,item_id,occurred_at,stage,level,message,details_json
                    ) VALUES (?, ?, ?, 'cancelled', 'warning', ?, ?)
                    """,
                    (
                        run_id,
                        run["current_item_id"],
                        now,
                        message,
                        json.dumps({"code": "ROTATION_CANCELLED"}),
                    ),
                )
                connection.commit()
            except Exception:
                connection.rollback()
                raise
        cancelled = self.get(run_id)
        if cancelled is None:  # pragma: no cover
            raise RuntimeError("RUN_NOT_FOUND")
        return cancelled

    def due_group_ids(self) -> list[str]:
        now = to_iso()
        with self.database.connect() as connection:
            rows = connection.execute(
                """
                SELECT group_record.id FROM instance_groups AS group_record
                WHERE group_record.archived_at IS NULL AND group_record.enabled = 1
                    AND group_record.next_run_at IS NOT NULL
                    AND group_record.next_run_at <= ?
                    AND EXISTS (
                        SELECT 1 FROM instance_group_members member
                        JOIN managed_instances instance ON instance.id = member.instance_id
                        WHERE member.group_id = group_record.id
                            AND instance.archived_at IS NULL AND instance.enabled = 1
                    )
                ORDER BY group_record.next_run_at
                """,
                (now,),
            ).fetchall()
        return [str(row["id"]) for row in rows]

    @staticmethod
    def _advance_group_schedule(
        connection: sqlite3.Connection,
        group_id: str,
        finished_at: str,
    ) -> None:
        group = connection.execute(
            """
            SELECT enabled, interval_minutes FROM instance_groups
            WHERE id = ? AND archived_at IS NULL
            """,
            (group_id,),
        ).fetchone()
        if group is None:
            return
        next_run_at = (
            to_iso(from_iso(finished_at) + timedelta(minutes=int(group["interval_minutes"])))
            if bool(group["enabled"])
            else None
        )
        connection.execute(
            """
            UPDATE instance_groups
            SET last_run_at = ?, next_run_at = ?, updated_at = ?
            WHERE id = ?
            """,
            (finished_at, next_run_at, finished_at, group_id),
        )

    def defer_group(self, group_id: str) -> None:
        now = utc_now()
        with self.database.connect() as connection, connection:
            row = connection.execute(
                """
                SELECT interval_minutes FROM instance_groups
                WHERE id = ? AND archived_at IS NULL AND enabled = 1
                """,
                (group_id,),
            ).fetchone()
            if row is None:
                return
            connection.execute(
                """
                UPDATE instance_groups SET next_run_at = ?, updated_at = ? WHERE id = ?
                """,
                (
                    to_iso(now + timedelta(minutes=int(row["interval_minutes"]))),
                    to_iso(now),
                    group_id,
                ),
            )

    def acquire_dns_sync_lock(self, owner_id: str, ttl_seconds: int = 300) -> bool:
        now = utc_now()
        now_iso = to_iso(now)
        lease_until = to_iso(now + timedelta(seconds=ttl_seconds))
        with self.database.connect() as connection:
            try:
                connection.execute("BEGIN IMMEDIATE")
                self._clear_expired_dns_lock(connection, now_iso)
                if connection.execute("SELECT 1 FROM fleet_runs WHERE active_slot = 1").fetchone():
                    connection.rollback()
                    return False
                connection.execute(
                    """
                    INSERT INTO fleet_operation_locks(slot,kind,owner_id,lease_until,created_at)
                    VALUES (1,'dns_sync',?,?,?)
                    """,
                    (owner_id, lease_until, now_iso),
                )
                connection.commit()
                return True
            except sqlite3.IntegrityError:
                connection.rollback()
                return False

    def renew_operation_lock(self, owner_id: str, ttl_seconds: int = 300) -> bool:
        now = utc_now()
        with self.database.connect() as connection, connection:
            cursor = connection.execute(
                """
                UPDATE fleet_operation_locks SET lease_until = ?
                WHERE owner_id = ? AND kind = 'dns_sync' AND lease_until >= ?
                """,
                (to_iso(now + timedelta(seconds=ttl_seconds)), owner_id, to_iso(now)),
            )
        return cursor.rowcount == 1

    def release_operation_lock(self, owner_id: str) -> None:
        with self.database.connect() as connection, connection:
            connection.execute("DELETE FROM fleet_operation_locks WHERE owner_id = ?", (owner_id,))

    @staticmethod
    def static_ip_name(instance_name: str, item_id: str) -> str:
        normalized = "".join(
            char.lower() if char.isascii() and char.isalnum() else "-" for char in instance_name
        )
        normalized = "-".join(filter(None, normalized.split("-")))[:48] or "instance"
        return f"fluxip-{normalized}-{item_id.replace('-', '')[:12]}"

    @staticmethod
    def stage_elapsed_seconds(item: FleetRunItem) -> float:
        return (utc_now() - from_iso(item.stage_started_at)).total_seconds()

    @staticmethod
    def _clear_expired_dns_lock(connection: sqlite3.Connection, now: str) -> None:
        connection.execute(
            """
            DELETE FROM fleet_operation_locks
            WHERE kind = 'dns_sync' AND lease_until IS NOT NULL AND lease_until < ?
            """,
            (now,),
        )