app/validator/mailbox_provider.py

"""Mailbox verification providers (Layer 5 catch-all + Layer 6 SMTP).

The engine is FULLY AUTONOMOUS (v3.0.0): mailbox verdicts come from its own
SMTP probing engine — no third-party verification API, no vendor key, no
per-check fee. The former MillionVerifier delegation was removed by design
decision: a stray key must never be able to re-route verification data (and
the full recipient list) through a vendor.

resolve_provider() picks the mailbox layer from settings:

    mailbox_provider = "auto"   → remote SMTP worker when SMTP_WORKER_URL is
                                  configured, else the built-in engine
    mailbox_provider = "local"  → built-in SMTP engine (this host)
    mailbox_provider = "remote" → the remote SMTP probe worker (spec §3.3)

RemoteSmtpWorkerProvider implements the four-worker architecture: the
orchestrator (this app) delegates the port-25 dialogue to a separate
deployment of the same image running on a host whose egress allows it
(see POST /api/v1/smtp/probe in routes.py). A shared secret
(SMTP_WORKER_TOKEN) guards the worker route.

Both providers embed a circuit breaker so a host with blocked port-25 egress
(or a dead worker) fails fast instead of burning the full SMTP timeout on
every address in a bulk job.
"""
from __future__ import annotations

import asyncio
import random
import time
from typing import Optional, Tuple

import httpx

from ..config import settings
from ..models import SMTPResult
from .smtp_verify import GIBBERISH_LOCAL, smtp_verifier


class MailboxProvider:
    """Common interface every mailbox provider implements."""

    name = "base"

    async def verify(self, email: str, mx_host: str) -> SMTPResult:
        raise NotImplementedError

    async def catch_all(self, domain: str, mx_host: str) -> Tuple[Optional[bool], float]:
        raise NotImplementedError


class AutonomousSmtpProvider(MailboxProvider):
    """The self-contained engine: own SMTP probing, no third-party service.

    Wraps the MX-grouped batch verifier (reused sessions, per-domain catch-all
    caching, greylist awareness). This is the default provider — the validator
    must work with no vendor account, key, or per-check fee.
    """

    name = "autonomous"

    def __init__(self, batch=None):
        if batch is None:
            from .mx_batch import CatchAllCache, MxBatchVerifier

            batch = MxBatchVerifier(
                timeout=settings.smtp_timeout,
                helo_domain=settings.smtp_helo_domain,
                mail_from=settings.smtp_mail_from,
                catch_all_cache=CatchAllCache(
                    ttl_seconds=settings.catch_all_ttl_seconds
                ),
            )
        self.batch = batch

    async def verify(self, email: str, mx_host: str) -> SMTPResult:
        results = self.batch.verify_batch([(email, mx_host)])
        return results.get(email) or SMTPResult(
            verified=None, error="engine returned no result"
        )

    async def catch_all(
        self, domain: str, mx_host: str
    ) -> Tuple[Optional[bool], float]:
        """Probe a gibberish local part to learn whether the domain accepts all."""
        gibberish = f"{GIBBERISH_LOCAL}@{domain}"
        results = self.batch.verify_batch([(gibberish, mx_host)])
        probe = results.get(gibberish)
        if probe is None:
            return None, 0.0
        if probe.verified is True:
            # The domain accepted a random address -> catch-all.
            return True, 0.9
        if probe.verified is False:
            return False, 0.9
        # Inconclusive: distinguish a detected catch-all from a dead probe.
        err = (probe.error or "").lower()
        if "catch_all" in err:
            return True, 0.95
        return None, 0.0

    async def verify_many(self, pairs):
        """Batched mailbox stage: one call for the whole list.

        Returns (email -> SMTPResult, domain -> catch-all flag). This is what
        lets the engine do one TCP session per domain instead of one per address.
        """
        pairs = [(e, mx) for e, mx in pairs if e and mx]
        if not pairs:
            return {}, {}
        # Prefer the concurrent path when the batch engine supports it.
        if hasattr(self.batch, "verify_batch_async"):
            results = await self.batch.verify_batch_async(pairs)
        else:
            results = self.batch.verify_batch(pairs)
        domains = {e.rsplit("@", 1)[-1].lower(): mx for e, mx in pairs}
        catch_all = {
            domain: self.batch.catch_all_cache.get(domain) for domain in domains
        }
        return results, catch_all


class RemoteSmtpWorkerProvider(MailboxProvider):
    """Delegates the port-25 dialogue to a remote SMTP probe worker (§3.3).

    The worker is any deployment of this same image whose SMTP_WORKER_URL this
    orchestrator can reach. Egress requirement moves to the worker host; the
    orchestrator stays PaaS-deployable.
    """

    name = "remote"

    def __init__(
        self,
        url: Optional[str] = None,
        token: Optional[str] = None,
        timeout: Optional[int] = None,
        fail_threshold: Optional[int] = None,
        cooldown_seconds: Optional[int] = None,
    ):
        self.url = (url if url is not None else settings.smtp_worker_url).rstrip("/")
        self.token = token if token is not None else settings.smtp_worker_token
        self.timeout = timeout if timeout is not None else settings.smtp_worker_timeout
        self.fail_threshold = (
            fail_threshold if fail_threshold is not None else settings.smtp_fail_threshold
        )
        self.cooldown = (
            cooldown_seconds
            if cooldown_seconds is not None
            else settings.smtp_cooldown_seconds
        )
        self._consecutive_failures = 0
        self._open_until = 0.0

    # ---- circuit breaker (same contract as LocalSmtpProvider) --------------
    @property
    def circuit_open(self) -> bool:
        return time.monotonic() < self._open_until

    def record_failure(self) -> None:
        self._consecutive_failures += 1
        if self._consecutive_failures >= self.fail_threshold:
            self._open_until = time.monotonic() + self.cooldown

    def record_success(self) -> None:
        self._consecutive_failures = 0
        self._open_until = 0.0

    def _headers(self) -> dict:
        h = {"Content-Type": "application/json"}
        if self.token:
            h["X-Worker-Token"] = self.token
        return h

    async def _post(self, path: str, payload: dict):
        async with httpx.AsyncClient(timeout=self.timeout + 5) as client:
            resp = await client.post(
                f"{self.url}/smtp{path}", json=payload, headers=self._headers()
            )
            resp.raise_for_status()
            return resp.json()

    async def verify(self, email: str, mx_host: str) -> SMTPResult:
        if self.circuit_open:
            return SMTPResult(
                verified=None, error="SMTP worker unavailable (circuit open)"
            )
        try:
            data = await self._post(
                "/probe", {"email": email, "mx_host": mx_host}
            )
            result = SMTPResult(**data)
        except Exception as exc:
            self.record_failure()
            return SMTPResult(
                verified=None, error=f"smtp_worker_unreachable: {exc}"
            )
        self.record_success()
        return result

    async def catch_all(
        self, domain: str, mx_host: str
    ) -> Tuple[Optional[bool], float]:
        if self.circuit_open:
            return None, 0.0
        try:
            data = await self._post(
                "/probe", {"domain": domain, "mx_host": mx_host, "probe": "catch_all"}
            )
        except Exception:
            self.record_failure()
            return None, 0.0
        self.record_success()
        value = data.get("is_catch_all")
        confidence = float(data.get("confidence", 0.0) or 0.0)
        if value is None or confidence == 0.0:
            return None, 0.0
        return bool(value), confidence

    async def verify_many(self, pairs):
        pairs = [(e, mx) for e, mx in pairs if e and mx]
        if not pairs:
            return {}, {}
        if self.circuit_open:
            return (
                {
                    e: SMTPResult(
                        verified=None, error="SMTP worker unavailable (circuit open)"
                    )
                    for e, _ in pairs
                },
                {},
            )
        try:
            data = await self._post(
                "/probe/batch", {"pairs": [[e, mx] for e, mx in pairs]}
            )
            results = {
                e: SMTPResult(**row) for e, row in (data.get("results") or {}).items()
            }
            catch_all = {d: v for d, v in (data.get("catch_all") or {}).items()}
            self.record_success()
            return results, catch_all
        except Exception as exc:
            self.record_failure()
            return (
                {
                    e: SMTPResult(
                        verified=None, error=f"smtp_worker_unreachable: {exc}"
                    )
                    for e, _ in pairs
                },
                {},
            )


class LocalSmtpProvider(MailboxProvider):
    """Direct port-25 SMTP (one dialogue per call) with a circuit breaker.

    Kept as the building block for single-probe paths and tests; bulk traffic
    goes through AutonomousSmtpProvider's session-reusing batch engine.
    """

    name = "local"

    def __init__(
        self,
        fail_threshold: Optional[int] = None,
        cooldown_seconds: Optional[int] = None,
    ):
        self.fail_threshold = (
            fail_threshold if fail_threshold is not None else settings.smtp_fail_threshold
        )
        self.cooldown = (
            cooldown_seconds if cooldown_seconds is not None else settings.smtp_cooldown_seconds
        )
        self._consecutive_failures = 0
        self._open_until = 0.0

    @property
    def circuit_open(self) -> bool:
        return time.monotonic() < self._open_until

    def circuit_open_after_cooldown(self) -> bool:
        """True only while still within the cooldown window (0s ⇒ closed)."""
        return time.monotonic() < self._open_until

    def record_failure(self) -> None:
        self._consecutive_failures += 1
        if self._consecutive_failures >= self.fail_threshold:
            self._open_until = time.monotonic() + self.cooldown

    def record_success(self) -> None:
        self._consecutive_failures = 0
        self._open_until = 0.0

    async def verify(self, email: str, mx_host: str) -> SMTPResult:
        if self.circuit_open:
            return SMTPResult(
                verified=None, error="SMTP unavailable (egress circuit open)"
            )
        result = await smtp_verifier.verify(email, mx_host, settings.use_proxies)
        # A connection-class failure (timeout/refused/unreachable) counts toward
        # opening the circuit; a real SMTP dialogue does not.
        if result.verified is None and result.error and any(
            k in result.error.lower()
            for k in ("timeout", "timed out", "refused", "unreachable", "unavailable")
        ):
            self.record_failure()
        elif result.smtp_code is not None or result.verified is not None:
            self.record_success()
        return result

    async def catch_all(self, domain: str, mx_host: str) -> Tuple[Optional[bool], float]:
        if self.circuit_open:
            return None, 0.0
        try:
            is_catch_all, confidence = await smtp_verifier.check_catch_all(domain, mx_host)
        except Exception:
            return None, 0.0
        # check_catch_all returns (False, 0.0) when the probe is inconclusive —
        # that means UNKNOWN, not "not catch-all". Confidence 0.0 is the signal.
        if confidence == 0.0:
            self.record_failure()
            return None, 0.0
        self.record_success()
        return is_catch_all, confidence


def resolve_provider() -> MailboxProvider:
    """Pick the mailbox layer.

    Fully autonomous by default: the built-in SMTP engine answers with no
    external dependency. A configured SMTP_WORKER_URL shifts the port-25
    dialogue to the dedicated worker host — nothing else changes.
    """
    mode = (settings.mailbox_provider or "auto").strip().lower()
    if mode == "remote":
        if settings.smtp_worker_url:
            return RemoteSmtpWorkerProvider()
        # Misconfiguration must not crash the API: degrade to the built-in
        # engine, which reports honest transport failures.
        return AutonomousSmtpProvider()
    if mode == "local":
        return AutonomousSmtpProvider()
    # auto → remote worker when configured, else built-in.
    if settings.smtp_worker_url:
        return RemoteSmtpWorkerProvider()
    return AutonomousSmtpProvider()