File size: 4,698 Bytes
c94194a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
"""Result caching for the verification engine.

Two things are expensive and highly reusable in bulk verification:

  - **DNS/MX** for a domain (every address at that domain needs the same answer)
  - **the mailbox verdict** for an address (lists overlap; users re-run lists)

Both are cached here, with one hard correctness rule: **only definitive results
are cached**. Transient failures — a DNS timeout, a greylist 4xx, an unreachable
MX — are never stored, because remembering them would freeze a temporary
condition into a 24h "verdict" and suppress a later successful probe.

A cache outage degrades to "no cache"; it must never break validation.
"""
from __future__ import annotations

import json
from typing import Any, Callable, Coroutine, Optional

from ..config import settings
from ..models import DNSResult, SMTPResult
from ..utils.cache import get_cache as _default_get, set_cache as _default_set

# A DNS timeout/resolver error is inconclusive — never a cacheable fact.
_UNRESOLVED_DNS_CODES = frozenset({"timeout", "resolver_error"})

DNS_TTL = 24 * 3600        # MX/SPF/DMARC change slowly
VERDICT_TTL = 24 * 3600    # a mailbox verdict is stable within a day


class ValidationCache:
    """Thin typed layer over the shared Redis/memory cache."""

    def __init__(
        self,
        get: Optional[Callable[[str], Coroutine[Any, Any, Optional[str]]]] = None,
        set: Optional[Callable[..., Coroutine[Any, Any, None]]] = None,
        ttl_seconds: int = DNS_TTL,
    ):
        self._get = get or _default_get
        self._set = set or _default_set
        self.ttl_seconds = ttl_seconds

    # ---- keys ------------------------------------------------------
    @staticmethod
    def dns_key(domain: str) -> str:
        return f"vx:dns:{domain.strip().lower()}"

    @staticmethod
    def verdict_key(email: str, deep: bool) -> str:
        # `deep` is part of the identity: a fast-mode result carries no mailbox
        # proof and must never satisfy a deep request.
        return f"vx:v:{'d' if deep else 'f'}:{email.strip().lower()}"

    # ---- low level -------------------------------------------------
    async def _read_json(self, key: str) -> Optional[dict]:
        try:
            raw = await self._get(key)
        except Exception:
            return None  # cache outage -> treat as a miss
        if not raw:
            return None
        try:
            data = json.loads(raw)
        except Exception:
            return None  # corrupt payload -> miss, never raise
        return data if isinstance(data, dict) else None

    async def _write_json(self, key: str, payload: dict, ttl: int) -> None:
        try:
            await self._set(key, json.dumps(payload), ttl)
        except Exception:
            pass  # a cache write failure must not break validation

    # ---- DNS -------------------------------------------------------
    async def get_dns(self, domain: str) -> Optional[DNSResult]:
        data = await self._read_json(self.dns_key(domain))
        if not data:
            return None
        try:
            return DNSResult(**data)
        except Exception:
            return None

    async def put_dns(self, domain: str, result: DNSResult) -> None:
        if result.error_code in _UNRESOLVED_DNS_CODES:
            return  # transient -> never cache
        await self._write_json(
            self.dns_key(domain), result.model_dump(), self.ttl_seconds
        )

    # ---- mailbox verdict ------------------------------------------
    async def get_verdict(
        self, email: str, deep: bool
    ) -> Optional[tuple[SMTPResult, Optional[bool]]]:
        data = await self._read_json(self.verdict_key(email, deep))
        if not data:
            return None
        try:
            smtp = SMTPResult(**data["smtp"])
            catch_all = data.get("catch_all")
            return smtp, catch_all
        except Exception:
            return None

    async def put_verdict(
        self,
        email: str,
        deep: bool,
        smtp: SMTPResult,
        catch_all: Optional[bool],
    ) -> None:
        # Cacheable only when the mailbox question was actually answered, or the
        # domain is definitively catch-all (a domain fact, not a probe failure).
        definitive = smtp.verified is not None or catch_all is True
        if not definitive:
            return
        await self._write_json(
            self.verdict_key(email, deep),
            {"smtp": smtp.model_dump(), "catch_all": catch_all},
            VERDICT_TTL,
        )


# Shared instance used by the engine.
validation_cache = ValidationCache(
    ttl_seconds=getattr(settings, "dns_cache_ttl_seconds", DNS_TTL)
)