"""Outbound port-25 preflight.

Mailbox verification *requires* outbound TCP to the recipient's MX on port 25.
Most PaaS platforms block it — Hugging Face Spaces and AWS (EC2/Lambda) both
block 25 by default — so an engine that looks correct can still be unable to
produce a single real verdict.

This probe answers the only question that matters before trusting a host:
"can this machine reach an MX on port 25?" It is deliberately tiny, side-effect
free (banner + EHLO, then QUIT), and safe to run at boot or from CI.
"""
from __future__ import annotations

import socket
from dataclasses import asdict, dataclass
from typing import Callable, Optional

# A highly-available MX; if this is unreachable, port 25 egress is blocked.
DEFAULT_TARGET = ("gmail-smtp-in.l.google.com", 25)
DEFAULT_HELO = "validator.local"

_BLOCKED_ADVICE = (
    "Outbound port 25 is blocked. Hugging Face Spaces, AWS (EC2/Lambda) and most "
    "PaaS providers block it by default, so direct SMTP verification cannot run "
    "there. Fix: run this service on a VPS/host with port-25 egress (and set a "
    "PTR record + SMTP_HELO_DOMAIN so strict receivers accept the probes). "
    "Until then every mailbox row is honestly reported as smtp_unreachable."
)


@dataclass
class EgressReport:
    target: str
    reachable: bool
    banner: str = ""
    helo_accepted: bool = False
    diagnosis: str = ""
    advice: str = ""

    @property
    def ok(self) -> bool:
        return self.reachable

    def as_dict(self) -> dict:
        data = asdict(self)
        data["ok"] = self.ok
        return data


def _default_connect(host: str, port: int, timeout: float):
    sock = socket.create_connection((host, port), timeout=timeout)
    sock.settimeout(timeout)
    return sock


def check_port25_egress(
    connect: Optional[Callable[[str, int, float], object]] = None,
    target: tuple[str, int] = DEFAULT_TARGET,
    helo_domain: str = DEFAULT_HELO,
    timeout: float = 8.0,
) -> EgressReport:
    """Probe whether outbound port 25 works from this host."""
    host, port = target
    label = f"{host}:{port}"
    sock = None
    try:
        sock = (connect or _default_connect)(host, port, timeout)
    except Exception as exc:
        # A timeout / refusal / unreachable network on port 25 is the signature
        # of egress filtering. Say so explicitly — "timed out" alone sends the
        # operator hunting for a DNS or app bug instead of a firewall.
        kind = type(exc).__name__
        detail = str(exc)
        if isinstance(exc, socket.timeout) or "timed out" in detail.lower():
            diagnosis = f"connect timed out (port 25 is blocked/filtered): {detail}"
        elif isinstance(exc, ConnectionRefusedError) or "refused" in detail.lower():
            diagnosis = f"connection refused on port 25 (blocked): {detail}"
        elif isinstance(exc, OSError) and getattr(exc, "errno", None) == 101:
            diagnosis = f"network unreachable (port 25 egress blocked): {detail}"
        else:
            diagnosis = f"connect failed, port 25 likely blocked: {kind}: {detail}"
        return EgressReport(
            target=label,
            reachable=False,
            diagnosis=diagnosis,
            advice=_BLOCKED_ADVICE,
        )

    banner = ""
    helo_accepted = False
    try:
        try:
            raw = sock.recv(1024)  # type: ignore[attr-defined]
            banner = raw.decode("utf-8", "replace").strip() if raw else ""
        except Exception as exc:
            return EgressReport(
                target=label,
                reachable=False,
                diagnosis=f"no SMTP banner: {type(exc).__name__}: {exc}",
                advice=_BLOCKED_ADVICE,
            )

        try:
            sock.sendall(f"EHLO {helo_domain}\r\n".encode())  # type: ignore[attr-defined]
            reply = sock.recv(1024)  # type: ignore[attr-defined]
            helo_accepted = b"250" in reply
        except Exception:
            helo_accepted = False

        try:
            sock.sendall(b"QUIT\r\n")  # type: ignore[attr-defined]
        except Exception:
            pass
    finally:
        if sock is not None:
            try:
                sock.close()
            except Exception:
                pass

    if not banner.startswith("220"):
        return EgressReport(
            target=label,
            reachable=False,
            banner=banner,
            helo_accepted=helo_accepted,
            diagnosis=f"unexpected banner: {banner[:80]!r}",
            advice=_BLOCKED_ADVICE,
        )

    return EgressReport(
        target=label,
        reachable=True,
        banner=banner,
        helo_accepted=helo_accepted,
        diagnosis="port 25 reachable — real mailbox verification is possible here",
        advice=(
            "Set SMTP_HELO_DOMAIN to this host's real hostname and give that IP a "
            "matching PTR record; strict receivers reject generic HELO names."
        ),
    )


def preflight_or_warn() -> EgressReport:
    """Run the preflight and print a one-line operational summary."""
    report = check_port25_egress()
    status = "OK" if report.ok else "BLOCKED"
    print(f"[EGRESS] port-25 preflight {status} — {report.target} — {report.diagnosis}")
    if not report.ok:
        print(f"[EGRESS] {report.advice}")
    return report