Upload folder using huggingface_hub

app/api/routes.py

@@ -25,6 +25,10 @@ async def process_bulk_sync(emails: list, webhook_url: str = None, deep: bool =
     `deep` defaults to True so bulk runs the full 7-layer pipeline (catch-all +
     SMTP) exactly like single verification — the product charges 1 credit/email
     and must not silently degrade to a syntax+DNS-only check.
+
+    Chunks feed the engine's *batched* path: DNS/MX resolves once per domain and
+    the mailbox stage takes the whole chunk in one call, so the SMTP engine reuses
+    a session per receiving host instead of opening a TCP connection per address.
     """
     results = []
     total = len(emails)
@@ -32,19 +36,38 @@ async def process_bulk_sync(emails: list, webhook_url: str = None, deep: bool =
 
     for i in range(0, total, chunk_size):
         chunk = emails[i:i + chunk_size]
-        tasks = [engine.validate(email, deep=deep) for email in chunk]
-        chunk_results = await asyncio.gather(*tasks, return_exceptions=True)
-        
-        for email, result in zip(chunk, chunk_results):
+        try:
+            chunk_results = await engine.validate_batch(chunk, deep=deep)
+        except Exception as exc:  # never lose a chunk to one bad stage
+            results.extend(
+                {
+                    "email": email,
+                    "error": str(exc) or "Unknown error",
+                    "status": "error",
+                }
+                for email in chunk
+            )
+            continue
+
+        # validate_batch guarantees one result per input, in input order.
+        if len(chunk_results) == len(chunk):
+            pairs = zip(chunk, chunk_results)
+        else:
+            # Defensive: never lose or mis-attribute a row.
+            by_email = {getattr(r, "email", "").lower(): r for r in chunk_results}
+            pairs = ((email, by_email.get(email.lower())) for email in chunk)
+
+        for email, result in pairs:
             if isinstance(result, ValidationResult):
                 results.append(result.model_dump())
             else:
-                # Handle exceptions
-                results.append({
-                    "email": email,
-                    "error": str(result) if result else "Unknown error",
-                    "status": "error"
-                })
+                results.append(
+                    {
+                        "email": email,
+                        "error": "engine returned no result for this address",
+                        "status": "error",
+                    }
+                )
     
     # Webhook notification if provided
     if webhook_url:

app/tasks/celery_tasks.py

@@ -47,38 +47,40 @@ def bulk_validate_task(self, emails: list, webhook_url: str = None, deep: bool =
     from ..db import repo
     import sys
 
+    CHUNK = 200
+
     async def run():
         eng = EmailValidationEngine()
         results = []
         total = len(emails)
 
-        for i, email in enumerate(emails):
+        # Process in chunks through the engine's batched path: DNS/MX once per
+        # domain and one mailbox call per chunk, so the SMTP engine reuses a
+        # session per receiving host instead of a TCP connection per address.
+        for start in range(0, total, CHUNK):
+            chunk = emails[start:start + CHUNK]
             try:
-                result = await eng.validate(email, deep=deep)
-                results.append(result.model_dump())
-                # Update progress
-                self.update_state(
-                    state="PROGRESS",
-                    meta={"current": i + 1, "total": total}
-                )
+                chunk_results = await eng.validate_batch(chunk, deep=deep)
+                results.extend(r.model_dump() for r in chunk_results)
             except Exception as e:
                 print(
-                    f"[CELERY TASK] Validation failed for item {i + 1}/{total}: "
-                    f"{e.__class__.__name__}",
+                    f"[CELERY TASK] Validation failed for chunk "
+                    f"{start + 1}-{start + len(chunk)}/{total}: {e.__class__.__name__}",
                     file=sys.stderr,
                 )
-                results.append({
-                    "email": email,
-                    "error": str(e),
-                    "status": "error"
-                })
+                results.extend(
+                    {"email": email, "error": str(e), "status": "error"}
+                    for email in chunk
+                )
+
+            done = min(start + len(chunk), total)
+            self.update_state(state="PROGRESS", meta={"current": done, "total": total})
 
             # Persist incremental progress to Supabase (best-effort)
-            if (i + 1) % 10 == 0 or i + 1 == total:
-                try:
-                    repo.update_bulk_job_progress(self.request.id, i + 1, total)
-                except Exception as e:
-                    print(f"[CELERY TASK] Failed to update progress: {str(e)}", file=sys.stderr)
+            try:
+                repo.update_bulk_job_progress(self.request.id, done, total)
+            except Exception as e:
+                print(f"[CELERY TASK] Failed to update progress: {str(e)}", file=sys.stderr)
 
         return results
 

app/validator/engine.py

@@ -299,6 +299,217 @@ class EmailValidationEngine:
             validator_version=settings.app_version,
         )
 
+    async def validate_batch(
+        self, emails: list[str], deep: bool = True
+    ) -> list[ValidationResult]:
+        """Validate many addresses in one pass, reusing per-domain work.
+
+        Bulk throughput comes from doing DNS/MX **once per domain** and handing
+        every (email, mx_host) pair to the mailbox provider in a **single** call,
+        so the SMTP engine can reuse one session per receiving host instead of a
+        fresh TCP handshake per address. Providers that cannot batch fall back to
+        the per-address path, so behaviour is never worse than before.
+
+        Results are returned in the same order as `emails`.
+        """
+        if not emails:
+            return []
+
+        run_id = uuid.uuid4().hex[:12]
+        out: list[ValidationResult | None] = [None] * len(emails)
+
+        def _stamp(**kw) -> ValidationResult:
+            return ValidationResult(
+                processing_time_ms=0.0,
+                validated_at=datetime.now(timezone.utc).isoformat(),
+                run_id=run_id,
+                validator_version=settings.app_version,
+                **kw,
+            )
+
+        # ---- Layer 1: syntax (per address) ---------------------------
+        parsed: list[tuple[str, SyntaxResult]] = []
+        domain_index: dict[str, list[int]] = {}
+        for idx, raw in enumerate(emails):
+            syntax = validate_syntax(raw)
+            parsed.append((raw, syntax))
+            if not syntax.passed:
+                scoring = calculate_score(syntax, None, None, None, None)
+                out[idx] = _stamp(
+                    email=raw,
+                    status=ValidationStatus.INVALID,
+                    syntax=syntax,
+                    scoring=scoring,
+                    checks_run=_build_checks_run(syntax, None, None, None, None, scoring),
+                    reason_code=_build_reason_code(None, None, None, None),
+                )
+                continue
+            domain_index.setdefault(syntax.domain, []).append(idx)
+
+        # ---- Layers 2 & 3: DNS + MX once per domain ------------------
+        domains = list(domain_index.keys())
+        dns_list = await asyncio.gather(
+            *[check_dns(d, settings.dns_timeout) for d in domains],
+            return_exceptions=True,
+        )
+        domain_dns: dict[str, DNSResult] = {}
+        for domain, dns in zip(domains, dns_list):
+            domain_dns[domain] = (
+                dns
+                if isinstance(dns, DNSResult)
+                else DNSResult(
+                    domain_exists=False,
+                    has_mx_records=False,
+                    error=str(dns),
+                    error_code="resolver_error",
+                )
+            )
+
+        pending: list[tuple[int, str, SyntaxResult, str, DNSResult]] = []
+        for domain, idxs in domain_index.items():
+            dns = domain_dns[domain]
+            inconclusive = dns.error_code in {"timeout", "resolver_error"}
+            dead = (
+                not dns.domain_exists
+                or dns.error_code == "nxdomain"
+                or (not dns.has_mx_records and not dns.a_fallback)
+            )
+            for idx in idxs:
+                raw, syntax = parsed[idx]
+                normalized = syntax.normalized_email or raw
+                if inconclusive:
+                    scoring = calculate_score(syntax, dns, None, None, None)
+                    out[idx] = _stamp(
+                        email=normalized,
+                        status=self._determine_status(scoring),
+                        syntax=syntax,
+                        dns=dns,
+                        scoring=scoring,
+                        checks_run=_build_checks_run(
+                            syntax, dns, None, None, None, scoring
+                        ),
+                        reason_code=_build_reason_code(
+                            dns, None, None, None, early="dns_timeout"
+                        ),
+                    )
+                elif dead:
+                    scoring = calculate_score(syntax, dns, None, None, None)
+                    out[idx] = _stamp(
+                        email=normalized,
+                        status=ValidationStatus.INVALID,
+                        syntax=syntax,
+                        dns=dns,
+                        scoring=scoring,
+                        checks_run=_build_checks_run(
+                            syntax, dns, None, None, None, scoring
+                        ),
+                        reason_code=_build_reason_code(
+                            dns, None, None, None, early="nxdomain"
+                        ),
+                    )
+                else:
+                    pending.append((idx, normalized, syntax, domain, dns))
+
+        # ---- Layer 4 fast path (no mailbox probe) --------------------
+        if pending and not deep:
+            for idx, normalized, syntax, domain, dns in pending:
+                disposable = check_disposable(normalized, domain)
+                scoring = calculate_score(
+                    syntax, dns, disposable, None, None,
+                    mailbox_verification_skipped=True,
+                )
+                out[idx] = _stamp(
+                    email=normalized,
+                    status=self._determine_status(scoring),
+                    syntax=syntax,
+                    dns=dns,
+                    disposable=disposable,
+                    scoring=scoring,
+                    checks_run=_build_checks_run(
+                        syntax, dns, disposable, None, None, scoring
+                    ),
+                    reason_code=_build_reason_code(dns, disposable, None, None),
+                )
+            pending = []
+
+        # ---- Layers 5 & 6: ONE batched mailbox call ------------------
+        if pending:
+            pairs = [
+                (normalized, dns.mx_records[0]["host"])
+                for _, normalized, _, _, dns in pending
+            ]
+            catch_all_map: dict[str, Optional[bool]] = {}
+            try:
+                if hasattr(self.mailbox, "verify_many"):
+                    smtp_map, caught = await self.mailbox.verify_many(pairs)
+                    catch_all_map = dict(caught or {})
+                else:
+                    # Provider cannot batch: per-address fallback.
+                    smtp_map = {}
+                    for email, mx in pairs:
+                        smtp_map[email] = await self.mailbox.verify(email, mx)
+                    for domain in {e.rsplit("@", 1)[-1].lower() for e, _ in pairs}:
+                        mx = next(
+                            m for e, m in pairs
+                            if e.rsplit("@", 1)[-1].lower() == domain
+                        )
+                        value, _conf = await self.mailbox.catch_all(domain, mx)
+                        catch_all_map[domain] = value
+            except Exception as exc:  # one bad stage must not abort the batch
+                smtp_map = {
+                    e: SMTPResult(verified=None, error=f"mailbox_stage_error: {exc}")
+                    for e, _ in pairs
+                }
+
+            # ---- Layer 7: score each row -----------------------------
+            for idx, normalized, syntax, domain, dns in pending:
+                smtp = smtp_map.get(normalized) or SMTPResult(
+                    verified=None, error="engine returned no result"
+                )
+                ca_value = catch_all_map.get(domain)
+                catch_all = CatchAllResult(
+                    is_catch_all=ca_value,
+                    confidence=0.9 if ca_value is not None else 0.0,
+                )
+                disposable = check_disposable(normalized, domain)
+                scoring = calculate_score(
+                    syntax, dns, disposable, catch_all, smtp
+                )
+                out[idx] = _stamp(
+                    email=normalized,
+                    status=self._determine_status(scoring),
+                    syntax=syntax,
+                    dns=dns,
+                    disposable=disposable,
+                    catch_all=catch_all,
+                    smtp=smtp,
+                    scoring=scoring,
+                    checks_run=_build_checks_run(
+                        syntax, dns, disposable, catch_all, smtp, scoring
+                    ),
+                    reason_code=_build_reason_code(
+                        dns, disposable, catch_all, smtp
+                    ),
+                )
+
+        # Guarantee one result per input, in order.
+        for idx, result in enumerate(out):
+            if result is None:
+                raw = emails[idx]
+                syntax = parsed[idx][1] if idx < len(parsed) else validate_syntax(raw)
+                scoring = calculate_score(syntax, None, None, None, None)
+                out[idx] = _stamp(
+                    email=raw,
+                    status=ValidationStatus.UNKNOWN,
+                    syntax=syntax,
+                    scoring=scoring,
+                    checks_run=_build_checks_run(
+                        syntax, None, None, None, None, scoring
+                    ),
+                    reason_code="engine_missing_result",
+                )
+        return [r for r in out if r is not None]
+
     def _determine_status(self, scoring: ScoringResult) -> ValidationStatus:
         if scoring.score >= 75:
             return ValidationStatus.VALID

app/validator/mailbox_provider.py

@@ -169,6 +169,22 @@ class AutonomousSmtpProvider(MailboxProvider):
             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 {}, {}
+        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 LocalSmtpProvider(MailboxProvider):
     """Direct port-25 SMTP with a circuit breaker for blocked egress."""

tests/test_bulk_depth.py

@@ -39,34 +39,30 @@ class BulkDepthTests(unittest.IsolatedAsyncioTestCase):
     Bulk must run the same deep pipeline as single verification."""
 
     async def test_process_bulk_sync_runs_deep_layers(self):
-        captured = []
+        """Bulk must run deep, via the batched path (one call for the chunk)."""
+        captured = {}
 
-        async def fake_validate(email, deep=True):
-            captured.append(deep)
-            from app.models import ValidationStatus, ScoringResult, RiskLevel
+        class FakeResult:
+            def __init__(self, email):
+                self.email = email
 
-            return type("R", (), {"model_dump": lambda self: {"email": email}})()
+            def model_dump(self):
+                return {"email": self.email}
 
-        with (
-            patch("app.api.routes.engine.validate", new=AsyncMock(side_effect=lambda e, deep=True: fake_validate(e, deep))),
-            patch("app.api.routes.settings.bulk_sync_chunk_size", 50),
-        ):
-            # patch engine.validate directly to capture the deep flag
-            async def capture_validate(email, deep=True):
-                captured.append(deep)
-
-                class FakeResult:
-                    def model_dump(self):
-                        return {"email": email}
+        async def capture_batch(emails, deep=True):
+            captured["emails"] = list(emails)
+            captured["deep"] = deep
+            return [FakeResult(e) for e in emails]
 
-                return FakeResult()
-
-            with patch("app.api.routes.engine") as eng:
-                eng.validate = AsyncMock(side_effect=capture_validate)
-                await process_bulk_sync(["a@example.com", "b@example.com"])
+        with patch(
+            "app.api.routes.engine.validate_batch",
+            new=AsyncMock(side_effect=capture_batch),
+        ):
+            out = await process_bulk_sync(["a@example.com", "b@example.com"])
 
-        self.assertTrue(len(captured) >= 1)
-        self.assertTrue(all(d is True for d in captured), f"bulk ran with deep flags: {captured}")
+        self.assertEqual(captured["emails"], ["a@example.com", "b@example.com"])
+        self.assertTrue(captured["deep"], "bulk must run deep")
+        self.assertEqual(len(out), 2)
 
     async def test_engine_deep_runs_catch_all_and_smtp(self):
         engine = EmailValidationEngine()

tests/test_engine_batch.py

@@ -0,0 +1,219 @@
+import unittest
+from unittest.mock import AsyncMock, patch
+
+from app.models import (
+    DNSResult,
+    DisposableResult,
+    SMTPResult,
+    SyntaxResult,
+    ValidationStatus,
+)
+from app.validator.engine import EmailValidationEngine
+
+
+def _dns_ok(host="mx.example.test"):
+    return DNSResult(
+        domain_exists=True,
+        has_mx_records=True,
+        mx_records=[{"host": host, "priority": 10}],
+    )
+
+
+def _disposable_clean():
+    return DisposableResult(
+        is_disposable=False, is_free_provider=False, is_role_based=False
+    )
+
+
+class FakeBatchProvider:
+    """Provider exposing the bulk capability the engine should exploit.
+
+    `verify_many` is the whole point: one call for N addresses lets the SMTP
+    engine reuse a single session per domain instead of one TCP handshake each.
+    """
+
+    def __init__(self, results=None, catch_all=None):
+        self._results = results or {}
+        self._catch_all = catch_all or {}
+        self.batch_calls: list[list[tuple[str, str]]] = []
+        self.single_calls = 0
+
+    async def verify_many(self, pairs):
+        self.batch_calls.append(list(pairs))
+        return dict(self._results), dict(self._catch_all)
+
+    async def verify(self, email, mx_host):
+        self.single_calls += 1
+        return self._results.get(email, SMTPResult(verified=None, error="n/a"))
+
+    async def catch_all(self, domain, mx_host):
+        value = self._catch_all.get(domain)
+        return (value, 0.9) if value is not None else (None, 0.0)
+
+
+class EngineBatchTests(unittest.IsolatedAsyncioTestCase):
+    """Bulk must go through ONE batched mailbox stage per run, not one
+    connection per address (the throughput requirement)."""
+
+    async def test_batch_resolves_dns_once_per_domain(self):
+        provider = FakeBatchProvider(
+            results={
+                "a@corp.test": SMTPResult(verified=True, smtp_code=250),
+                "b@corp.test": SMTPResult(verified=False, smtp_code=550,
+                                          smtp_message="550 mailbox does not exist"),
+                "c@other.test": SMTPResult(verified=True, smtp_code=250),
+            },
+            catch_all={"corp.test": False, "other.test": False},
+        )
+        engine = EmailValidationEngine(mailbox=provider)
+        with (
+            patch(
+                "app.validator.engine.check_dns",
+                new=AsyncMock(return_value=_dns_ok()),
+            ) as dns_mock,
+            patch(
+                "app.validator.engine.check_disposable",
+                return_value=_disposable_clean(),
+            ),
+        ):
+            results = await engine.validate_batch(
+                ["a@corp.test", "b@corp.test", "c@other.test"], deep=True
+            )
+
+        # Two unique domains -> two DNS resolutions, not three.
+        self.assertEqual(dns_mock.await_count, 2)
+        self.assertEqual(len(results), 3)
+
+    async def test_batch_calls_provider_verify_many_once(self):
+        provider = FakeBatchProvider(
+            results={
+                "a@corp.test": SMTPResult(verified=True, smtp_code=250),
+                "b@corp.test": SMTPResult(verified=True, smtp_code=250),
+            },
+            catch_all={"corp.test": False},
+        )
+        engine = EmailValidationEngine(mailbox=provider)
+        with (
+            patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns_ok())),
+            patch("app.validator.engine.check_disposable", return_value=_disposable_clean()),
+        ):
+            await engine.validate_batch(["a@corp.test", "b@corp.test"], deep=True)
+
+        self.assertEqual(len(provider.batch_calls), 1)
+        self.assertEqual(
+            sorted(provider.batch_calls[0]),
+            [("a@corp.test", "mx.example.test"), ("b@corp.test", "mx.example.test")],
+        )
+        # The per-address path must NOT be used when batching is available.
+        self.assertEqual(provider.single_calls, 0)
+
+    async def test_batch_results_carry_provenance(self):
+        provider = FakeBatchProvider(
+            results={"a@corp.test": SMTPResult(verified=True, smtp_code=250)},
+            catch_all={"corp.test": False},
+        )
+        engine = EmailValidationEngine(mailbox=provider)
+        with (
+            patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns_ok())),
+            patch("app.validator.engine.check_disposable", return_value=_disposable_clean()),
+        ):
+            results = await engine.validate_batch(["a@corp.test"], deep=True)
+
+        r = results[0]
+        self.assertEqual(
+            r.checks_run,
+            ["syntax", "dns", "mx", "disposable", "catch_all", "smtp", "scoring"],
+        )
+        self.assertTrue(r.run_id)
+        self.assertEqual(r.status, ValidationStatus.VALID)
+
+    async def test_batch_catch_all_rows_are_risky_not_valid(self):
+        provider = FakeBatchProvider(
+            results={
+                "x@catchall.test": SMTPResult(
+                    verified=None,
+                    error="catch_all: catchall.test accepts every recipient",
+                )
+            },
+            catch_all={"catchall.test": True},
+        )
+        engine = EmailValidationEngine(mailbox=provider)
+        with (
+            patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns_ok())),
+            patch("app.validator.engine.check_disposable", return_value=_disposable_clean()),
+        ):
+            results = await engine.validate_batch(["x@catchall.test"], deep=True)
+
+        self.assertEqual(results[0].status, ValidationStatus.RISKY)
+        self.assertEqual(results[0].reason_code, "catch_all_detected")
+
+    async def test_batch_preserves_input_order_and_handles_syntax_failures(self):
+        provider = FakeBatchProvider(
+            results={"ok@corp.test": SMTPResult(verified=True, smtp_code=250)},
+            catch_all={"corp.test": False},
+        )
+        engine = EmailValidationEngine(mailbox=provider)
+        with (
+            patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns_ok())),
+            patch("app.validator.engine.check_disposable", return_value=_disposable_clean()),
+        ):
+            results = await engine.validate_batch(
+                ["not-an-email", "ok@corp.test"], deep=True
+            )
+
+        self.assertEqual(len(results), 2)
+        self.assertEqual(results[0].email, "not-an-email")
+        self.assertEqual(results[0].status, ValidationStatus.INVALID)
+        self.assertEqual(results[1].email, "ok@corp.test")
+        self.assertEqual(results[1].status, ValidationStatus.VALID)
+
+    async def test_batch_falls_back_to_single_when_provider_cannot_batch(self):
+        class SingleOnlyProvider:
+            def __init__(self):
+                self.calls = 0
+
+            async def verify(self, email, mx_host):
+                self.calls += 1
+                return SMTPResult(verified=True, smtp_code=250)
+
+            async def catch_all(self, domain, mx_host):
+                return (False, 0.9)
+
+        provider = SingleOnlyProvider()
+        engine = EmailValidationEngine(mailbox=provider)
+        with (
+            patch("app.validator.engine.check_dns", new=AsyncMock(return_value=_dns_ok())),
+            patch("app.validator.engine.check_disposable", return_value=_disposable_clean()),
+        ):
+            results = await engine.validate_batch(["a@corp.test", "b@corp.test"], deep=True)
+
+        self.assertEqual(len(results), 2)
+        self.assertEqual(provider.calls, 2)
+
+
+class BulkRouteBatchTests(unittest.IsolatedAsyncioTestCase):
+    async def test_process_bulk_sync_uses_the_batched_path(self):
+        from app.api import routes
+
+        called = {}
+
+        async def fake_batch(emails, deep=True):
+            called["emails"] = list(emails)
+            called["deep"] = deep
+
+            class R:
+                def model_dump(self):
+                    return {"email": "x"}
+
+            return [R() for _ in emails]
+
+        with patch.object(routes.engine, "validate_batch", new=AsyncMock(side_effect=fake_batch)):
+            out = await routes.process_bulk_sync(["a@b.com", "c@d.com"], deep=True)
+
+        self.assertEqual(called["emails"], ["a@b.com", "c@d.com"])
+        self.assertTrue(called["deep"])
+        self.assertEqual(len(out), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()