#!/usr/bin/env python3
"""
Reset Official Mail state for a clean re-onboarding test.

What this script does:
1. Reads DB and Mailcow settings from .env.production
2. Fetches managed domains/mailboxes from the official_mail DB
3. Deletes matching mailboxes first, then domains, through the Mailcow API
4. Truncates Official Mail application tables in MySQL

Run this on the server or inside a container that can reach both:
- the MySQL server in DB_HOST/DB_PORT
- the Mailcow API base URL

Examples:
  python deploy/ubuntu/reset_official_mail.py --yes
  python deploy/ubuntu/reset_official_mail.py --yes --mailcow-base-url https://127.0.0.1:8443
  python deploy/ubuntu/reset_official_mail.py --yes --skip-mailcow
"""

from __future__ import annotations

import argparse
import json
import os
import ssl
import sys
from pathlib import Path
from typing import Any
from urllib import error, request

import pymysql


RESET_TABLES = [
    "mailbox_ai_assist_replies",
    "mailbox_ai_assist_threads",
    "mailbox_ai_assist_settings",
    "managed_team_mailboxes",
    "mailbox_delivery_logs",
    "mailbox_messages",
    "mailbox_sender_rules",
    "mailbox_folders",
    "mailboxes",
    "dns_records",
    "domains",
    "users",
]


def load_env_file(path: Path) -> dict[str, str]:
    values: dict[str, str] = {}

    if not path.exists():
        raise FileNotFoundError(f"Environment file not found: {path}")

    for raw_line in path.read_text(encoding="utf-8").splitlines():
        line = raw_line.strip()

        if not line or line.startswith("#") or "=" not in line:
            continue

        key, value = line.split("=", 1)
        values[key.strip()] = value.strip()

    return values


def env_value(values: dict[str, str], key: str, default: str = "") -> str:
    return (os.environ.get(key) or values.get(key) or default).strip()


def env_bool(values: dict[str, str], key: str, default: bool = False) -> bool:
    raw_value = env_value(values, key, "")

    if not raw_value:
        return default

    return raw_value.lower() in {"1", "true", "yes", "on"}


def require_value(values: dict[str, str], key: str) -> str:
    value = env_value(values, key)

    if not value:
        raise RuntimeError(f"Missing required env: {key}")

    return value


def connect_db(values: dict[str, str]) -> pymysql.connections.Connection:
    return pymysql.connect(
        host=require_value(values, "DB_HOST"),
        port=int(env_value(values, "DB_PORT", "3306")),
        user=require_value(values, "DB_USER"),
        password=require_value(values, "DB_PASSWORD"),
        database=require_value(values, "DB_NAME"),
        charset="utf8mb4",
        autocommit=False,
        cursorclass=pymysql.cursors.DictCursor,
    )


def fetch_managed_state(connection: pymysql.connections.Connection) -> tuple[list[str], list[str]]:
    with connection.cursor() as cursor:
        cursor.execute("SELECT email FROM mailboxes ORDER BY id ASC")
        mailboxes = [str(row["email"]).strip().lower() for row in cursor.fetchall() if row.get("email")]

        cursor.execute("SELECT domain FROM domains ORDER BY id ASC")
        domains = [str(row["domain"]).strip().lower() for row in cursor.fetchall() if row.get("domain")]

    return mailboxes, domains


def truncate_official_mail_tables(connection: pymysql.connections.Connection) -> None:
    with connection.cursor() as cursor:
        cursor.execute("SET FOREIGN_KEY_CHECKS = 0")

        for table in RESET_TABLES:
            cursor.execute(f"TRUNCATE TABLE `{table}`")

        cursor.execute("SET FOREIGN_KEY_CHECKS = 1")

    connection.commit()


def build_ssl_context(allow_self_signed: bool) -> ssl.SSLContext:
    context = ssl.create_default_context()

    if allow_self_signed:
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE

    return context


def request_mailcow(
    base_url: str,
    api_key: str,
    path: str,
    *,
    method: str = "GET",
    payload: Any | None = None,
    allow_self_signed: bool = False,
) -> Any:
    url = f"{base_url.rstrip('/')}{path}"
    data = None
    headers = {
        "Accept": "application/json",
        "X-API-Key": api_key,
    }

    if payload is not None:
        data = json.dumps(payload).encode("utf-8")
        headers["Content-Type"] = "application/json"

    req = request.Request(url, method=method, data=data, headers=headers)

    try:
        with request.urlopen(req, context=build_ssl_context(allow_self_signed), timeout=20) as response:
            raw_body = response.read().decode("utf-8", errors="replace").strip()
    except error.HTTPError as exc:
        detail = exc.read().decode("utf-8", errors="replace").strip()
        raise RuntimeError(f"Mailcow request failed: {method} {path} -> {exc.code} {detail}") from exc
    except error.URLError as exc:
        raise RuntimeError(f"Mailcow request failed: {method} {path} -> {exc}") from exc

    if not raw_body:
        return None

    try:
        return json.loads(raw_body)
    except json.JSONDecodeError:
        return raw_body


def payload_contains_error(payload: Any) -> bool:
    if isinstance(payload, list):
        for item in payload:
            if isinstance(item, dict) and item.get("type") in {"danger", "error"}:
                return True

    if isinstance(payload, dict) and payload.get("type") in {"danger", "error"}:
        return True

    return False


def delete_mailcow_mailboxes(
    base_url: str,
    api_key: str,
    mailboxes: list[str],
    *,
    allow_self_signed: bool,
) -> None:
    for mailbox in mailboxes:
        print(f"[mailcow] deleting mailbox: {mailbox}")
        payload = request_mailcow(
            base_url,
            api_key,
            "/api/v1/delete/mailbox",
            method="POST",
            payload=[mailbox],
            allow_self_signed=allow_self_signed,
        )

        if payload_contains_error(payload):
            raise RuntimeError(f"Mailcow mailbox delete failed for {mailbox}: {payload}")


def delete_mailcow_domains(
    base_url: str,
    api_key: str,
    domains: list[str],
    *,
    allow_self_signed: bool,
) -> None:
    for domain in domains:
        print(f"[mailcow] deleting domain: {domain}")
        payload = request_mailcow(
            base_url,
            api_key,
            "/api/v1/delete/domain",
            method="POST",
            payload={"items": [domain]},
            allow_self_signed=allow_self_signed,
        )

        if payload_contains_error(payload):
            raise RuntimeError(f"Mailcow domain delete failed for {domain}: {payload}")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Reset Official Mail DB and Mailcow state.")
    parser.add_argument(
        "--env-file",
        default=".env.production",
        help="Path to the environment file to read. Default: .env.production",
    )
    parser.add_argument(
        "--mailcow-base-url",
        default="",
        help="Override Mailcow API base URL. Useful on the server when .env uses an internal hostname.",
    )
    parser.add_argument(
        "--skip-mailcow",
        action="store_true",
        help="Only reset the DB. Do not call Mailcow API.",
    )
    parser.add_argument(
        "--yes",
        action="store_true",
        help="Execute the reset. Without this flag, the script only prints what it would do.",
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    env_path = Path(args.env_file).resolve()
    values = load_env_file(env_path)

    db_name = require_value(values, "DB_NAME")
    db_host = require_value(values, "DB_HOST")
    mailcow_base_url = args.mailcow_base_url.strip() or require_value(values, "MAILCOW_BASE_URL")
    mailcow_api_key = env_value(values, "MAILCOW_API_KEY")
    allow_self_signed = env_bool(values, "MAILCOW_ALLOW_SELF_SIGNED", True)

    connection = connect_db(values)

    try:
        mailboxes, domains = fetch_managed_state(connection)

        print("Official Mail reset plan")
        print(f"- env file: {env_path}")
        print(f"- DB: {db_name} @ {db_host}")
        print(f"- mailboxes to delete: {len(mailboxes)}")
        print(f"- domains to delete: {len(domains)}")
        print(f"- tables to truncate: {', '.join(RESET_TABLES)}")
        print(f"- mailcow base url: {mailcow_base_url}")
        print(f"- skip mailcow: {args.skip_mailcow}")

        if mailboxes:
            print("  mailboxes:")
            for mailbox in mailboxes:
                print(f"    - {mailbox}")

        if domains:
            print("  domains:")
            for domain in domains:
                print(f"    - {domain}")

        if not args.yes:
            print("\nDry run only. Re-run with --yes to execute.")
            return 0

        if not args.skip_mailcow:
            if not mailcow_api_key:
                raise RuntimeError("MAILCOW_API_KEY is required unless --skip-mailcow is used.")

            delete_mailcow_mailboxes(
                mailcow_base_url,
                mailcow_api_key,
                mailboxes,
                allow_self_signed=allow_self_signed,
            )
            delete_mailcow_domains(
                mailcow_base_url,
                mailcow_api_key,
                domains,
                allow_self_signed=allow_self_signed,
            )

        print("[db] truncating official mail tables")
        truncate_official_mail_tables(connection)
        print("Reset completed.")
        return 0
    finally:
        connection.close()


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except Exception as exc:  # noqa: BLE001
        print(f"Reset failed: {exc}", file=sys.stderr)
        raise SystemExit(1)