"""Operational checks exposed via API/panel: deliverability, RBL, e2e connectivity."""

from __future__ import annotations

import json
import logging
import os
import re
import socket
import ssl
import subprocess
import threading
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
from urllib import error as urllib_error
from urllib import request as urllib_request

from .settings import get_settings

logger = logging.getLogger(__name__)

# Public DNSBLs used only for *outbound IP reputation self-checks* (not inbound reject lists).
# Operators should run these from the mail host with a stable resolver.
DEFAULT_DNSBL_ZONES = (
    "zen.spamhaus.org",
    "bl.spamcop.net",
    "b.barracudacentral.org",
    "dnsbl.sorbs.net",
)

RBL_ENV_KEY = "LIMRISTEM_MAIL_RSPAMD_DNSBL_ENABLED"
RBL_ZONES_ENV_KEY = "LIMRISTEM_MAIL_RSPAMD_DNSBL_ZONES"


def utc_now_iso() -> str:
    return datetime.now(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")


def _run(cmd: list[str], timeout: int = 15, env: dict[str, str] | None = None) -> tuple[int, str, str]:
    try:
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout, check=False, env=env)
        return result.returncode, result.stdout or "", result.stderr or ""
    except FileNotFoundError:
        return 127, "", f"{cmd[0]} not found"
    except subprocess.TimeoutExpired:
        return 124, "", "timeout"


def dig_txt(name: str) -> list[str]:
    code, out, _ = _run(["dig", "+short", "TXT", name], timeout=10)
    if code != 0:
        return []
    return [line.strip().strip('"') for line in out.splitlines() if line.strip()]


def dig_mx(name: str) -> list[str]:
    code, out, _ = _run(["dig", "+short", "MX", name], timeout=10)
    if code != 0:
        return []
    return [line.strip() for line in out.splitlines() if line.strip()]


def dig_a(name: str) -> list[str]:
    code, out, _ = _run(["dig", "+short", "A", name], timeout=10)
    records = [line.strip() for line in out.splitlines() if line.strip()]
    code6, out6, _ = _run(["dig", "+short", "AAAA", name], timeout=10)
    records.extend([line.strip() for line in out6.splitlines() if line.strip()])
    return records


def reverse_dns(ip: str) -> str:
    code, out, _ = _run(["host", ip], timeout=10)
    if code != 0:
        return ""
    return out.strip()


def reverse_octets(ip: str) -> str | None:
    parts = ip.split(".")
    if len(parts) != 4:
        return None  # IPv6 DNSBL needs nibble form; skip for MVP simplicity
    try:
        if not all(0 <= int(p) <= 255 for p in parts):
            return None
    except ValueError:
        return None
    return ".".join(reversed(parts))


def query_dnsbl(ip: str, zone: str) -> dict[str, Any]:
    rev = reverse_octets(ip)
    if not rev:
        return {"zone": zone, "ip": ip, "listed": None, "detail": "only IPv4 DNSBL queries supported here"}
    qname = f"{rev}.{zone}"
    code, out, err = _run(["dig", "+short", "A", qname], timeout=8)
    answers = [line.strip() for line in out.splitlines() if line.strip()]
    listed = bool(answers) and code == 0
    return {
        "zone": zone,
        "ip": ip,
        "query": qname,
        "listed": listed,
        "answers": answers,
        "detail": err.strip() if code not in {0, 1} else None,
    }


def check_public_ip_reputation(ip: str, zones: list[str] | None = None) -> dict[str, Any]:
    zone_list = zones or list(DEFAULT_DNSBL_ZONES)
    env_zones = os.getenv(RBL_ZONES_ENV_KEY, "").strip()
    if env_zones and not zones:
        zone_list = [z for z in re.split(r"[\s,]+", env_zones) if z]
    results = []
    with ThreadPoolExecutor(max_workers=min(6, max(1, len(zone_list)))) as pool:
        futures = {pool.submit(query_dnsbl, ip, zone): zone for zone in zone_list}
        for fut in as_completed(futures):
            try:
                results.append(fut.result())
            except Exception as exc:  # pragma: no cover
                logger.warning("check_public_ip_reputation: suppressed %s: %s", type(exc).__name__, exc)
                results.append({"zone": futures[fut], "ip": ip, "listed": None, "detail": str(exc)})
    listed_count = sum(1 for item in results if item.get("listed") is True)
    return {
        "ip": ip,
        "checked_at": utc_now_iso(),
        "zones_checked": len(results),
        "listed_count": listed_count,
        "status": "listed" if listed_count else "clean",
        "results": sorted(results, key=lambda item: str(item.get("zone"))),
        "ptr": reverse_dns(ip),
        "note": "Self-check only: listing status depends on your resolver and DNSBL access policy.",
    }


def run_deliverability_check(
    domain: str | None = None,
    mail_host: str | None = None,
    public_ip: str | None = None,
    dkim_selector: str | None = None,
) -> dict[str, Any]:
    settings = get_settings()
    domain = (domain or settings.primary_domain).strip().rstrip(".").lower()
    mail_host = (mail_host or settings.hostname).strip().rstrip(".").lower()
    public_ip = (public_ip or os.getenv("LIMRISTEM_MAIL_PUBLIC_IP") or "").strip()
    dkim_selector = (dkim_selector or os.getenv("LIMRISTEM_MAIL_DKIM_SELECTOR") or "default").strip().lower()

    script = settings.runtime_bin_dir / "deliverability-check.sh"
    args = [str(script), domain, mail_host]
    if public_ip:
        args.append(public_ip)
        args.append(dkim_selector)
    elif dkim_selector:
        args.extend(["", dkim_selector])

    code, out, err = _run(args, timeout=60)
    dns = {
        "mx": dig_mx(domain),
        "mail_a": dig_a(mail_host),
        "spf": dig_txt(domain),
        "dmarc": dig_txt(f"_dmarc.{domain}"),
        "dkim": dig_txt(f"{dkim_selector}._domainkey.{domain}"),
        "mta_sts": dig_txt(f"_mta-sts.{domain}"),
        "tls_rpt": dig_txt(f"_smtp._tls.{domain}"),
    }
    ip_rep = check_public_ip_reputation(public_ip) if public_ip and re.fullmatch(r"\d+\.\d+\.\d+\.\d+", public_ip) else None
    return {
        "checked_at": utc_now_iso(),
        "domain": domain,
        "mail_host": mail_host,
        "public_ip": public_ip or None,
        "dkim_selector": dkim_selector,
        "script_exit_code": code,
        "script_output": out,
        "script_error": err.strip() or None,
        "dns": dns,
        "ip_reputation": ip_rep,
        "status": "ok" if code == 0 else "degraded",
    }


def _tcp_connect(host: str, port: int, timeout: float = 5.0) -> dict[str, Any]:
    started = datetime.now(UTC)
    try:
        with socket.create_connection((host, port), timeout=timeout) as sock:
            elapsed = (datetime.now(UTC) - started).total_seconds()
            peer = sock.getpeername()
            return {"ok": True, "host": host, "port": port, "peer": f"{peer[0]}:{peer[1]}", "latency_ms": int(elapsed * 1000)}
    except OSError as exc:
        return {"ok": False, "host": host, "port": port, "error": str(exc)}


def _tls_probe(host: str, port: int, *, starttls_smtp: bool = False, timeout: float = 8.0) -> dict[str, Any]:
    try:
        raw = socket.create_connection((host, port), timeout=timeout)
        if starttls_smtp:
            raw.recv(1024)
            raw.sendall(b"EHLO limristem-probe\r\n")
            raw.recv(2048)
            raw.sendall(b"STARTTLS\r\n")
            resp = raw.recv(1024).decode("utf-8", errors="replace")
            if not resp.startswith("220"):
                raw.close()
                return {"ok": False, "host": host, "port": port, "error": f"STARTTLS refused: {resp.strip()[:120]}"}
        context = ssl.create_default_context()
        with context.wrap_socket(raw, server_hostname=host) as ssock:
            cert = ssock.getpeercert()
            cipher = ssock.cipher()
            return {
                "ok": True,
                "host": host,
                "port": port,
                "tls_version": ssock.version(),
                "cipher": cipher[0] if cipher else None,
                "subject": dict(x[0] for x in cert.get("subject", ())) if cert else None,
                "issuer": dict(x[0] for x in cert.get("issuer", ())) if cert else None,
                "notAfter": cert.get("notAfter") if cert else None,
            }
    except Exception as exc:
        logger.warning("_tls_probe: suppressed %s: %s", type(exc).__name__, exc)
        return {"ok": False, "host": host, "port": port, "error": str(exc)}


def run_e2e_connectivity_tests(mail_host: str | None = None) -> dict[str, Any]:
    """Local/public connectivity probes (not a full mail transaction with credentials)."""
    settings = get_settings()
    host = (mail_host or settings.hostname).strip().rstrip(".").lower()
    ports = {
        "smtp_25": _tcp_connect(host, 25),
        "submission_587": _tcp_connect(host, 587),
        "smtps_465": _tcp_connect(host, 465),
        "imap_143": _tcp_connect(host, 143),
        "imaps_993": _tcp_connect(host, 993),
        "https_443": _tcp_connect(host, 443),
    }
    tls = {
        "smtp_starttls_25": _tls_probe(host, 25, starttls_smtp=True),
        "submission_starttls_587": _tls_probe(host, 587, starttls_smtp=True),
        "smtps_465": _tls_probe(host, 465, starttls_smtp=False),
        "imaps_993": _tls_probe(host, 993, starttls_smtp=False),
        "https_443": _tls_probe(host, 443, starttls_smtp=False),
    }
    health = {"ok": False, "detail": None}
    try:
        url = f"https://{host}/health" if settings.ssl_mode != "plain" else f"http://127.0.0.1:{settings.api_port}/health"
        with urllib_request.urlopen(url, timeout=8) as resp:  # noqa: S310
            body = resp.read(4096).decode("utf-8", errors="replace")
            health = {"ok": resp.status == 200, "status": resp.status, "body": body[:500]}
    except Exception as exc:
        logger.warning("run_e2e_connectivity_tests: suppressed %s: %s", type(exc).__name__, exc)
        health = {"ok": False, "error": str(exc)}

    ok_ports = sum(1 for item in ports.values() if item.get("ok"))
    ok_tls = sum(1 for item in tls.values() if item.get("ok"))
    return {
        "checked_at": utc_now_iso(),
        "mail_host": host,
        "ports": ports,
        "tls": tls,
        "health": health,
        "summary": {
            "ports_ok": ok_ports,
            "ports_total": len(ports),
            "tls_ok": ok_tls,
            "tls_total": len(tls),
            "status": "ok" if ok_ports >= 4 and health.get("ok") else "degraded",
        },
        "note": "Does not authenticate as a mailbox user; use a client for full login tests.",
    }


def rbl_inbound_config() -> dict[str, Any]:
    enabled = (os.getenv(RBL_ENV_KEY) or "no").strip().lower() in {"1", "true", "yes", "on"}
    zones_raw = os.getenv(RBL_ZONES_ENV_KEY, "zen.spamhaus.org bl.spamcop.net").strip()
    zones = [z for z in re.split(r"[\s,]+", zones_raw) if z]
    return {
        "enabled": enabled,
        "zones": zones,
        "env_enabled_key": RBL_ENV_KEY,
        "env_zones_key": RBL_ZONES_ENV_KEY,
        "warning": (
            "Inbound DNSBL can false-positive on shared/public resolvers. "
            "Prefer Rspamd modules with a dedicated resolver. Enable only if you understand the risk."
        ),
    }


# Rspamd only picks up a local.d file when its name matches a module: local.d/rbl.conf
# merges into the rbl module, local.d/anything_else.conf is never read. Earlier revisions
# wrote local.d/limristem_rbl.conf, so enabling inbound DNSBL from the panel reported
# success and changed nothing at all in the filtering path.
RSPAMD_LOCAL_D = Path("/etc/rspamd/local.d")
RSPAMD_RBL_CONF = RSPAMD_LOCAL_D / "rbl.conf"
RSPAMD_LEGACY_RBL_CONF = RSPAMD_LOCAL_D / "limristem_rbl.conf"
RBL_SYMBOL_PREFIX = "LIMRISTEM_RBL_"


# Strict validation at the API edge. render_rspamd_rbl_config() still whitelists on the
# way out — it is the last line before the config file — but rejecting bad input here
# gives the operator an error instead of a silently mangled zone name.
DNSBL_ZONE_RE = re.compile(
    r"^(?=.{1,253}$)[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?"
    r"(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)+$"
)


def validate_dnsbl_zone(value: str) -> str:
    zone = (value or "").strip().lower().rstrip(".")
    if not DNSBL_ZONE_RE.fullmatch(zone):
        raise ValueError(f"Invalid DNSBL zone: {value!r}")
    return zone


def normalize_dnsbl_zones(values: list[str], *, limit: int = 20) -> list[str]:
    cleaned: list[str] = []
    for item in values:
        if not str(item).strip():
            continue
        zone = validate_dnsbl_zone(str(item))
        if zone not in cleaned:
            cleaned.append(zone)
    return cleaned[:limit]


def rbl_symbol_for(zone: str) -> str:
    """Stable, syntactically valid Rspamd symbol name for a DNSBL zone."""
    safe = re.sub(r"[^A-Za-z0-9]", "_", zone).strip("_").upper()
    return (RBL_SYMBOL_PREFIX + safe)[:60]


def render_rspamd_rbl_config(*, enabled: bool, zones: list[str]) -> str:
    """Render the rbl module snippet. Zone names are whitelisted, never interpolated raw."""
    lines = [
        "# Generated by Limristem eMail — do not edit by hand.",
        "# Managed via: limristem-mail rbl set <yes|no> [zones...]",
        "# Reference: https://rspamd.com/doc/modules/rbl.html",
        f"# enabled={str(bool(enabled)).lower()}",
        "",
    ]
    entries: list[str] = []
    if enabled:
        seen: set[str] = set()
        for zone in zones:
            # Whitelist rather than escape: everything reaching this file is a DNS zone
            # name, and a stray quote or brace here would inject Rspamd configuration.
            safe = re.sub(r"[^A-Za-z0-9._-]", "", zone).strip(".")
            if not safe or safe in seen:
                continue
            seen.add(safe)
            symbol = rbl_symbol_for(safe)
            entries.append(
                "\n".join(
                    [
                        f"  {symbol} {{",
                        f'    rbl = "{safe}";',
                        f'    symbol = "{symbol}";',
                        "    ipv4 = true;",
                        "    ipv6 = false;",
                        "    from = true;",
                        "    received = false;",
                        "  }",
                    ]
                )
            )
    if entries:
        lines.append("rbls {")
        lines.append("\n\n".join(entries))
        lines.append("}")
    else:
        lines.append("# No Limristem-managed DNSBL zones are active.")
    return "\n".join(lines) + "\n"


def write_rspamd_rbl_config(*, enabled: bool, zones: list[str]) -> Path:
    """Write the Rspamd rbl snippet. Requires manage-rbl.sh or root for the live path."""
    # Staged copy for when /etc is not writable. It goes to the service-writable state
    # directory, not config/: that one must stay root-owned so the account running this
    # code cannot replace the env files and secrets the root helpers read from it.
    from .limristem_mail_backend import _state_dir

    staged = _state_dir() / "rspamd-rbl-enabled.inc"
    content = render_rspamd_rbl_config(enabled=enabled, zones=zones)
    staged.parent.mkdir(parents=True, exist_ok=True)
    staged.write_text(content, encoding="utf-8")
    os.chmod(staged, 0o640)
    try:
        if RSPAMD_LOCAL_D.is_dir() and os.access(RSPAMD_LOCAL_D, os.W_OK):
            RSPAMD_RBL_CONF.write_text(content, encoding="utf-8")
            os.chmod(RSPAMD_RBL_CONF, 0o640)
            # Drop the file earlier versions wrote; Rspamd never read it, but leaving it
            # behind makes the next person think the feature lives there.
            RSPAMD_LEGACY_RBL_CONF.unlink(missing_ok=True)
    except OSError as exc:
        logger.warning("Unable to write live Rspamd RBL config: %s", exc)
    return staged


_APT_JOB_LOCK = threading.Lock()
_APT_JOB_STATE: dict[str, Any] = {
    "running": False,
    "started_at": None,
    "completed_at": None,
    "status": "idle",
    "message": "Nessun aggiornamento in corso.",
    "output": "",
    "code": None,
}


def get_apt_job_state() -> dict[str, Any]:
    with _APT_JOB_LOCK:
        return dict(_APT_JOB_STATE)


def _package_helper_command(action: str) -> list[str]:
    """sudo command for the fixed-operation package helper.

    apt itself is not sudo-reachable on purpose: its configuration options can run
    arbitrary commands as root. manage-packages.sh exposes only update/full-upgrade
    and forwards no argument from the caller.
    """
    helper = get_settings().runtime_bin_dir / "manage-packages.sh"
    return ["/usr/bin/sudo", "-n", str(helper), action]


def _async_apt_upgrade_worker() -> None:
    global _APT_JOB_STATE
    logger.info("Starting background APT update & upgrade worker...")

    code1, out1, err1 = _run(_package_helper_command("update"), timeout=180)
    logger.info("APT Update returned %s. Out: %s, Err: %s", code1, out1, err1)

    code2, out2, err2 = _run(_package_helper_command("full-upgrade"), timeout=600)
    logger.info("APT Upgrade returned %s. Out: %s, Err: %s", code2, out2, err2)

    success = (code1 == 0 and code2 == 0)
    output = f"=== APT UPDATE ===\n{out1}\n{err1}\n\n=== APT UPGRADE ===\n{out2}\n{err2}"

    with _APT_JOB_LOCK:
        _APT_JOB_STATE["running"] = False
        _APT_JOB_STATE["completed_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
        _APT_JOB_STATE["code"] = 0 if success else (code1 or code2 or 1)
        _APT_JOB_STATE["status"] = "success" if success else "failed"
        _APT_JOB_STATE["message"] = (
            "Aggiornamento pacchetti APT completato con successo."
            if success
            else "Aggiornamento pacchetti APT terminato con errori."
        )
        _APT_JOB_STATE["output"] = output.strip()

    try:
        from .alerts import collect_system_alerts

        collect_system_alerts(force_refresh=True)
    except Exception as exc:
        logger.warning("_async_apt_upgrade_worker: suppressed %s: %s", type(exc).__name__, exc)


def start_async_apt_upgrade() -> tuple[bool, str]:
    global _APT_JOB_STATE
    with _APT_JOB_LOCK:
        if _APT_JOB_STATE["running"]:
            return False, "Un aggiornamento pacchetti APT è già in corso nel sistema. Attendere il completamento."

        _APT_JOB_STATE["running"] = True
        _APT_JOB_STATE["started_at"] = time.strftime("%Y-%m-%d %H:%M:%S")
        _APT_JOB_STATE["status"] = "running"
        _APT_JOB_STATE["message"] = "Aggiornamento pacchetti APT in corso in background (in queue)..."
        _APT_JOB_STATE["output"] = ""
        _APT_JOB_STATE["code"] = None

    t = threading.Thread(target=_async_apt_upgrade_worker, daemon=True)
    t.start()
    return True, "Aggiornamento pacchetti APT avviato in background. Il progresso è visibile nella box delle operazioni."


def run_apt_upgrade() -> dict[str, Any]:
    """Synchronous fallback alias for start_async_apt_upgrade."""
    started, msg = start_async_apt_upgrade()
    return {"ok": started, "message": msg, "state": get_apt_job_state()}