Spaces:

sitcod3
/

modul-x

Running

Z User

feat(jobs): server-driven bulk job runner — survives browser tab closure

b93f0d1 11 days ago

15 kB

"""Server-driven bulk job runner — the fix for the "submitting client is gone"
failure class (10 of the last 15 failed CSV runs on 2026-10-20/21).
Before this module the BROWSER orchestrated every bulk run: it chunked the
file, submitted each chunk to /validate/bulk, polled the engine, mirrored rows
into verifier_checks and finalized the job. If the tab closed, throttled or
slept mid-run, the job orphaned in 'processing' until the reaper closed it and
refunded the unprocessed remainder. The engine itself was healthy — its chunk
task even completed 500/500 while the browser row froze at 0/3000.
Contract (v2):
  POST /api/v1/jobs  {job_id, emails, deep?, final?}
    - JWT required (productFetch always attaches the Supabase session token).
    - verifier_jobs row must exist, belong to the JWT subject and still be
      open (queued/processing).
    - A background task validates the list in chunks (bulk_sync_chunk_size),
      upserts verifier_checks rows after EVERY chunk, heartbeats verifier_jobs
      (processed_count + counts + updated_at) so the dashboard and the reaper
      both see live progress, and — on the final batch — flips the job to
      'completed' and refunds risky rows server-side.
    - Any chunk-level failure finalizes the job as 'error' and refunds BOTH
      the risky rows and the unprocessed remainder (same ledger mechanics as
      the client finalize and the pg_cron reaper; idempotent via
      refunded_credits).
  The client polls verifier_jobs directly after accepting 202 — it no longer
  needs to stay alive for the run to finish.
"""
import asyncio
import json
import logging
import time
from typing import List, Optional

from fastapi import APIRouter, HTTPException, Request
from pydantic import BaseModel

from ..auth import verify_supabase_jwt
from ..config import settings
from ..db import repo
from ..models import ValidationResult
from ..validator.engine import engine

logger = logging.getLogger("emailvalidator.jobs")

router = APIRouter()

# Strong references so the event loop cannot GC an in-flight runner, and a
# semaphore so concurrent job submissions cannot spawn unbounded SMTP work.
_job_tasks: set[asyncio.Task] = set()
_job_semaphore = asyncio.Semaphore(settings.bulk_concurrency)

GROUPS = ("good", "risky", "bad")


class JobRequest(BaseModel):
    job_id: str
    emails: List[str]
    deep: bool = True
    final: bool = True
    user_id: Optional[str] = None  # informational; ownership comes from the JWT


# ---- classification (mirror of src/lib/verifier.ts classify()) ------------


def classify_engine_result(r: ValidationResult) -> dict:
    """Map an engine ValidationResult to the verifier_checks row shape the
    dashboard renders. Mirrors the frontend classify() branch order so a row
    never disagrees with a client-side re-classification:
      disposable → bad | dead domain → bad | dns inconclusive → risky |
      valid+smtp → good | engine invalid → bad | catch-all → risky | else risky.
    """
    disposable = bool(getattr(r.disposable, "is_disposable", False))
    catch_all = bool(getattr(r.catch_all, "is_catch_all", False))
    dns = r.dns
    status = (r.status.value if hasattr(r.status, "value") else str(r.status)) or "unknown"

    checks_run = r.checks_run or []
    dns_error = (dns.error or "") if dns else ""
    dns_failure = dns_error == "dns_timeout" or "timeout" in dns_error.lower()
    a_fallback_only = bool(dns and dns.domain_exists and not dns.has_mx_records)
    smtp_verified = bool(r.smtp and r.smtp.verified is True)
    reason = r.reason_code

    if disposable:
        out_status, result, grp = "disposable", "Disposable", "bad"
        reason = reason or "disposable"
    elif dns is not None and dns.domain_exists is False:
        # NXDOMAIN is definitive (the stale-build regression that showed a
        # 0% Bad column on a 3000-row list must never resurrect server-side).
        out_status, result, grp = "invalid", "Invalid", "bad"
        reason = reason or "nxdomain"
    elif dns_failure or a_fallback_only:
        out_status, result, grp = "unknown", "Unknown", "risky"
        reason = reason or ("dns_timeout" if dns_failure else "a_fallback_only")
    elif status == "valid" and smtp_verified:
        out_status, result, grp = "valid", "Ok", "good"
    elif status == "invalid":
        out_status, result, grp = "invalid", "Invalid", "bad"
    elif catch_all:
        out_status, result, grp = "catch_all", "Catch-all", "risky"
        reason = reason or "catch_all_detected"
    else:
        out_status, result, grp = "unknown", "Unknown", "risky"
        reason = reason or "smtp_unavailable"

    score = r.scoring.score if r.scoring is not None else None
    return {
        "email": r.email,
        "status": out_status,
        "result": result,
        "grp": grp,
        "score": score,
        "reason": reason,
        "pass_through": None,
        "checks_run": checks_run,
        "verified_at": r.validated_at,
        "run_id": r.run_id,
        "validator_version": r.validator_version,
    }


# ---- persistence helpers (service-role writes, best-effort like repo) ------


async def _write_checks(rows: List[dict]) -> None:
    if not rows:
        return
    try:
        await asyncio.to_thread(repo.write_verifier_checks, rows)
    except Exception as exc:  # never lose a run to one write blip
        logger.warning("job_checks_write_failed: %s", exc)


async def _update_job(job_id: str, **fields) -> None:
    try:
        await asyncio.to_thread(repo.update_verifier_job, job_id, fields)
    except Exception as exc:
        logger.warning("job_update_failed job=%s: %s", job_id, exc)


async def _finalize_failed(job_id: str, error: str, counts: dict, processed: int) -> None:
    """Close a failed job and refund risky + unprocessed server-side.
    Mirrors the ledger mechanics of public.refund_job_risky() (migration
    20261020000000) minus its auth.uid() guard — the engine writes with the
    service-role key, under which auth.uid() is null and the SQL function
    would raise. Idempotency comes from verifier_jobs.refunded_credits.
    """
    await _update_job(
        job_id,
        status="error",
        error=error[:500],
        processed_count=processed,
        valid_count=counts.get("good", 0),
        risky_count=counts.get("risky", 0),
        bad_count=counts.get("bad", 0),
    )
    try:
        job = await asyncio.to_thread(repo.get_verifier_job, job_id)
        if not job:
            return
        total = int(job.get("total_count") or 0)
        charged = int(job.get("credits_charged") or 0)
        refunded = int(job.get("refunded_credits") or 0)
        risky = int(job.get("risky_count") or 0)
        unprocessed = max(total - int(job.get("processed_count") or 0), 0)
        due = min(risky + unprocessed, charged) - refunded
        if due > 0:
            await asyncio.to_thread(repo.refund_credits, job["user_id"], due,
                                    "Auto-refund — risky and unprocessed results are not billed")
            await _update_job(job_id, refunded_credits=refunded + due)
    except Exception as exc:
        logger.warning("job_refund_failed job=%s: %s", job_id, exc)


# ---- the runner ------------------------------------------------------------


async def _run_verifier_job(job_id: str, user_id: str, emails: List[str], deep: bool,
                            final: bool) -> None:
    started = time.monotonic()
    counts = {g: 0 for g in GROUPS}
    processed = 0
    # Defensive dedupe (the frontend already deduped; a client bug must not
    # double-write rows — (job_id, email) upsert would silently dedupe them
    # while counts drift).
    seen: set[str] = set()
    ordered: List[str] = []
    for e in emails:
        e = (e or "").strip().lower()
        if e and "@" in e and e not in seen:
            seen.add(e)
            ordered.append(e)
    emails = ordered
    total = len(emails)
    if total == 0:
        await _finalize_failed(job_id, "No valid addresses reached the engine.", counts, 0)
        return

    chunk_size = max(int(settings.bulk_sync_chunk_size or 50), 10)
    async with _job_semaphore:
        try:
            await _update_job(job_id, status="processing")
            for i in range(0, total, chunk_size):
                chunk = emails[i:i + chunk_size]
                try:
                    results = await engine.validate_batch(chunk, deep=deep)
                except Exception as exc:  # one bad stage must not kill the job
                    logger.warning("job_chunk_failed job=%s: %s", job_id, exc)
                    await _finalize_failed(
                        job_id,
                        f"Engine chunk failed after {processed}/{total} addresses: {exc}",
                        counts, processed,
                    )
                    return
                if len(results) != len(chunk):
                    await _finalize_failed(
                        job_id,
                        f"Engine returned {len(results)} of {len(chunk)} results for one chunk",
                        counts, processed,
                    )
                    return
                rows = [classify_engine_result(r) for r in results]
                await _write_checks(rows)
                for row in rows:
                    counts[row["grp"]] = counts.get(row["grp"], 0) + 1
                processed += len(chunk)
                # Heartbeat + live progress in one write; the dashboard polls
                # this row instead of the engine, and the reaper reads it.
                await _update_job(
                    job_id,
                    processed_count=processed,
                    valid_count=counts["good"],
                    risky_count=counts["risky"],
                    bad_count=counts["bad"],
                )
            if final:
                await _update_job(
                    job_id,
                    status="completed",
                    error=None,
                    processed_count=processed,
                    valid_count=counts["good"],
                    risky_count=counts["risky"],
                    bad_count=counts["bad"],
                )
                # Risky rows are never billed; the completed-job refund covers
                # exactly the risky_count (unprocessed term is zero by design).
                try:
                    job = await asyncio.to_thread(repo.get_verifier_job, job_id)
                    if job:
                        charged = int(job.get("credits_charged") or 0)
                        refunded = int(job.get("refunded_credits") or 0)
                        due = min(counts["risky"], charged) - refunded
                        if due > 0:
                            await asyncio.to_thread(
                                repo.refund_credits, job["user_id"], due,
                                "Auto-refund — risky results are not billed")
                            await _update_job(job_id, refunded_credits=refunded + due)
                except Exception as exc:
                    logger.warning("job_risky_refund_failed job=%s: %s", job_id, exc)
            logger.info("job_batch_done job=%s processed=%s/%s final=%s secs=%.1f",
                        job_id, processed, total, final, time.monotonic() - started)
        except Exception as exc:  # absolute safety net
            logger.exception("job_runner_crashed job=%s", job_id)
            await _finalize_failed(job_id, f"Runner crashed: {exc}", counts, processed)


# ---- routes ----------------------------------------------------------------


@router.post("/jobs", tags=["Bulk Jobs"])
async def create_server_job(request: Request, payload: JobRequest):
    """Accept a bulk job for server-side processing (202 + DB polling)."""
    token = request.headers.get("authorization", "")
    token = token[7:].strip() if token.lower().startswith("bearer ") else ""
    if not token:
        raise HTTPException(401, "Missing bearer token — sign in first.")
    try:
        claims = await verify_supabase_jwt(token)
    except Exception as exc:
        raise HTTPException(401, f"Invalid or expired token: {exc}")
    user_id = claims.get("sub", "")
    if not user_id:
        raise HTTPException(401, "Token has no subject claim.")
    if not payload.emails:
        raise HTTPException(400, "emails must not be empty.")
    if not settings.supabase_configured:
        raise HTTPException(503, "Server persistence is not configured (SUPABASE_URL/SERVICE key).")

    job = await asyncio.to_thread(repo.get_verifier_job, payload.job_id)
    if not job:
        raise HTTPException(404, f"Job {payload.job_id} not found — create it first.")
    if str(job.get("user_id")) != user_id:
        # Ownership mismatch: identical response for missing/foreign jobs.
        raise HTTPException(404, f"Job {payload.job_id} not found for this account.")
    if job.get("status") not in ("queued", "processing"):
        raise HTTPException(409, f"Job {payload.job_id} is already {job.get('status')}.")
    if not payload.emails:
        raise HTTPException(400, "emails must not be empty.")

    # Cap one submission so a giant file cannot pin the runner forever: the
    # client splits into final-flagged batches (20k ≈ 600 KB JSON).
    MAX_PER_CALL = 20000
    emails = payload.emails[:MAX_PER_CALL]

    task = asyncio.create_task(
        _run_verifier_job(payload.job_id, user_id, emails, payload.deep, payload.final)
    )
    _job_tasks.add(task)
    task.add_done_callback(_job_tasks.discard)

    return {
        "accepted": len(emails),
        "job_id": payload.job_id,
        "status": "processing",
        "final": payload.final,
        "message": "Processing server-side — poll verifier_jobs for progress.",
    }


@router.get("/jobs/{job_id}", tags=["Bulk Jobs"])
async def job_status(job_id: str, request: Request):
    """Lightweight engine-side echo of the job row (debug aid; the dashboard
    polls the DB directly and does not depend on this)."""
    token = request.headers.get("authorization", "")
    token = token[7:].strip() if token.lower().startswith("bearer ") else ""
    if not token:
        raise HTTPException(401, "Missing bearer token — sign in first.")
    try:
        claims = await verify_supabase_jwt(token)
    except Exception as exc:
        raise HTTPException(401, f"Invalid or expired token: {exc}")
    job = await asyncio.to_thread(repo.get_verifier_job, job_id)
    if not job or str(job.get("user_id")) != claims.get("sub", ""):
        raise HTTPException(404, f"Job {job_id} not found for this account.")
    return {
        "job_id": job_id,
        "status": job.get("status"),
        "total": job.get("total_count"),
        "processed": job.get("processed_count"),
        "valid": job.get("valid_count"),
        "risky": job.get("risky_count"),
        "bad": job.get("bad_count"),
    }

	"""Server-driven bulk job runner — the fix for the "submitting client is gone"
	failure class (10 of the last 15 failed CSV runs on 2026-10-20/21).

	Before this module the BROWSER orchestrated every bulk run: it chunked the
	file, submitted each chunk to /validate/bulk, polled the engine, mirrored rows
	into verifier_checks and finalized the job. If the tab closed, throttled or
	slept mid-run, the job orphaned in 'processing' until the reaper closed it and
	refunded the unprocessed remainder. The engine itself was healthy — its chunk
	task even completed 500/500 while the browser row froze at 0/3000.

	Contract (v2):
	POST /api/v1/jobs {job_id, emails, deep?, final?}
	- JWT required (productFetch always attaches the Supabase session token).
	- verifier_jobs row must exist, belong to the JWT subject and still be
	open (queued/processing).
	- A background task validates the list in chunks (bulk_sync_chunk_size),
	upserts verifier_checks rows after EVERY chunk, heartbeats verifier_jobs
	(processed_count + counts + updated_at) so the dashboard and the reaper
	both see live progress, and — on the final batch — flips the job to
	'completed' and refunds risky rows server-side.
	- Any chunk-level failure finalizes the job as 'error' and refunds BOTH
	the risky rows and the unprocessed remainder (same ledger mechanics as
	the client finalize and the pg_cron reaper; idempotent via
	refunded_credits).
	The client polls verifier_jobs directly after accepting 202 — it no longer
	needs to stay alive for the run to finish.
	"""
	import asyncio
	import json
	import logging
	import time
	from typing import List, Optional

	from fastapi import APIRouter, HTTPException, Request
	from pydantic import BaseModel

	from ..auth import verify_supabase_jwt
	from ..config import settings
	from ..db import repo
	from ..models import ValidationResult
	from ..validator.engine import engine

	logger = logging.getLogger("emailvalidator.jobs")

	router = APIRouter()

	# Strong references so the event loop cannot GC an in-flight runner, and a
	# semaphore so concurrent job submissions cannot spawn unbounded SMTP work.
	_job_tasks: set[asyncio.Task] = set()
	_job_semaphore = asyncio.Semaphore(settings.bulk_concurrency)

	GROUPS = ("good", "risky", "bad")


	class JobRequest(BaseModel):
	job_id: str
	emails: List[str]
	deep: bool = True
	final: bool = True
	user_id: Optional[str] = None # informational; ownership comes from the JWT


	# ---- classification (mirror of src/lib/verifier.ts classify()) ------------


	def classify_engine_result(r: ValidationResult) -> dict:
	"""Map an engine ValidationResult to the verifier_checks row shape the
	dashboard renders. Mirrors the frontend classify() branch order so a row
	never disagrees with a client-side re-classification:
	disposable → bad \| dead domain → bad \| dns inconclusive → risky \|
	valid+smtp → good \| engine invalid → bad \| catch-all → risky \| else risky.
	"""
	disposable = bool(getattr(r.disposable, "is_disposable", False))
	catch_all = bool(getattr(r.catch_all, "is_catch_all", False))
	dns = r.dns
	status = (r.status.value if hasattr(r.status, "value") else str(r.status)) or "unknown"

	checks_run = r.checks_run or []
	dns_error = (dns.error or "") if dns else ""
	dns_failure = dns_error == "dns_timeout" or "timeout" in dns_error.lower()
	a_fallback_only = bool(dns and dns.domain_exists and not dns.has_mx_records)
	smtp_verified = bool(r.smtp and r.smtp.verified is True)
	reason = r.reason_code

	if disposable:
	out_status, result, grp = "disposable", "Disposable", "bad"
	reason = reason or "disposable"
	elif dns is not None and dns.domain_exists is False:
	# NXDOMAIN is definitive (the stale-build regression that showed a
	# 0% Bad column on a 3000-row list must never resurrect server-side).
	out_status, result, grp = "invalid", "Invalid", "bad"
	reason = reason or "nxdomain"
	elif dns_failure or a_fallback_only:
	out_status, result, grp = "unknown", "Unknown", "risky"
	reason = reason or ("dns_timeout" if dns_failure else "a_fallback_only")
	elif status == "valid" and smtp_verified:
	out_status, result, grp = "valid", "Ok", "good"
	elif status == "invalid":
	out_status, result, grp = "invalid", "Invalid", "bad"
	elif catch_all:
	out_status, result, grp = "catch_all", "Catch-all", "risky"
	reason = reason or "catch_all_detected"
	else:
	out_status, result, grp = "unknown", "Unknown", "risky"
	reason = reason or "smtp_unavailable"

	score = r.scoring.score if r.scoring is not None else None
	return {
	"email": r.email,
	"status": out_status,
	"result": result,
	"grp": grp,
	"score": score,
	"reason": reason,
	"pass_through": None,
	"checks_run": checks_run,
	"verified_at": r.validated_at,
	"run_id": r.run_id,
	"validator_version": r.validator_version,
	}


	# ---- persistence helpers (service-role writes, best-effort like repo) ------


	async def _write_checks(rows: List[dict]) -> None:
	if not rows:
	return
	try:
	await asyncio.to_thread(repo.write_verifier_checks, rows)
	except Exception as exc: # never lose a run to one write blip
	logger.warning("job_checks_write_failed: %s", exc)


	async def _update_job(job_id: str, **fields) -> None:
	try:
	await asyncio.to_thread(repo.update_verifier_job, job_id, fields)
	except Exception as exc:
	logger.warning("job_update_failed job=%s: %s", job_id, exc)


	async def _finalize_failed(job_id: str, error: str, counts: dict, processed: int) -> None:
	"""Close a failed job and refund risky + unprocessed server-side.

	Mirrors the ledger mechanics of public.refund_job_risky() (migration
	20261020000000) minus its auth.uid() guard — the engine writes with the
	service-role key, under which auth.uid() is null and the SQL function
	would raise. Idempotency comes from verifier_jobs.refunded_credits.
	"""
	await _update_job(
	job_id,
	status="error",
	error=error[:500],
	processed_count=processed,
	valid_count=counts.get("good", 0),
	risky_count=counts.get("risky", 0),
	bad_count=counts.get("bad", 0),
	)
	try:
	job = await asyncio.to_thread(repo.get_verifier_job, job_id)
	if not job:
	return
	total = int(job.get("total_count") or 0)
	charged = int(job.get("credits_charged") or 0)
	refunded = int(job.get("refunded_credits") or 0)
	risky = int(job.get("risky_count") or 0)
	unprocessed = max(total - int(job.get("processed_count") or 0), 0)
	due = min(risky + unprocessed, charged) - refunded
	if due > 0:
	await asyncio.to_thread(repo.refund_credits, job["user_id"], due,
	"Auto-refund — risky and unprocessed results are not billed")
	await _update_job(job_id, refunded_credits=refunded + due)
	except Exception as exc:
	logger.warning("job_refund_failed job=%s: %s", job_id, exc)


	# ---- the runner ------------------------------------------------------------


	async def _run_verifier_job(job_id: str, user_id: str, emails: List[str], deep: bool,
	final: bool) -> None:
	started = time.monotonic()
	counts = {g: 0 for g in GROUPS}
	processed = 0
	# Defensive dedupe (the frontend already deduped; a client bug must not
	# double-write rows — (job_id, email) upsert would silently dedupe them
	# while counts drift).
	seen: set[str] = set()
	ordered: List[str] = []
	for e in emails:
	e = (e or "").strip().lower()
	if e and "@" in e and e not in seen:
	seen.add(e)
	ordered.append(e)
	emails = ordered
	total = len(emails)
	if total == 0:
	await _finalize_failed(job_id, "No valid addresses reached the engine.", counts, 0)
	return

	chunk_size = max(int(settings.bulk_sync_chunk_size or 50), 10)
	async with _job_semaphore:
	try:
	await _update_job(job_id, status="processing")
	for i in range(0, total, chunk_size):
	chunk = emails[i:i + chunk_size]
	try:
	results = await engine.validate_batch(chunk, deep=deep)
	except Exception as exc: # one bad stage must not kill the job
	logger.warning("job_chunk_failed job=%s: %s", job_id, exc)
	await _finalize_failed(
	job_id,
	f"Engine chunk failed after {processed}/{total} addresses: {exc}",
	counts, processed,
	)
	return
	if len(results) != len(chunk):
	await _finalize_failed(
	job_id,
	f"Engine returned {len(results)} of {len(chunk)} results for one chunk",
	counts, processed,
	)
	return
	rows = [classify_engine_result(r) for r in results]
	await _write_checks(rows)
	for row in rows:
	counts[row["grp"]] = counts.get(row["grp"], 0) + 1
	processed += len(chunk)
	# Heartbeat + live progress in one write; the dashboard polls
	# this row instead of the engine, and the reaper reads it.
	await _update_job(
	job_id,
	processed_count=processed,
	valid_count=counts["good"],
	risky_count=counts["risky"],
	bad_count=counts["bad"],
	)
	if final:
	await _update_job(
	job_id,
	status="completed",
	error=None,
	processed_count=processed,
	valid_count=counts["good"],
	risky_count=counts["risky"],
	bad_count=counts["bad"],
	)
	# Risky rows are never billed; the completed-job refund covers
	# exactly the risky_count (unprocessed term is zero by design).
	try:
	job = await asyncio.to_thread(repo.get_verifier_job, job_id)
	if job:
	charged = int(job.get("credits_charged") or 0)
	refunded = int(job.get("refunded_credits") or 0)
	due = min(counts["risky"], charged) - refunded
	if due > 0:
	await asyncio.to_thread(
	repo.refund_credits, job["user_id"], due,
	"Auto-refund — risky results are not billed")
	await _update_job(job_id, refunded_credits=refunded + due)
	except Exception as exc:
	logger.warning("job_risky_refund_failed job=%s: %s", job_id, exc)
	logger.info("job_batch_done job=%s processed=%s/%s final=%s secs=%.1f",
	job_id, processed, total, final, time.monotonic() - started)
	except Exception as exc: # absolute safety net
	logger.exception("job_runner_crashed job=%s", job_id)
	await _finalize_failed(job_id, f"Runner crashed: {exc}", counts, processed)


	# ---- routes ----------------------------------------------------------------


	@router.post("/jobs", tags=["Bulk Jobs"])
	async def create_server_job(request: Request, payload: JobRequest):
	"""Accept a bulk job for server-side processing (202 + DB polling)."""
	token = request.headers.get("authorization", "")
	token = token[7:].strip() if token.lower().startswith("bearer ") else ""
	if not token:
	raise HTTPException(401, "Missing bearer token — sign in first.")
	try:
	claims = await verify_supabase_jwt(token)
	except Exception as exc:
	raise HTTPException(401, f"Invalid or expired token: {exc}")
	user_id = claims.get("sub", "")
	if not user_id:
	raise HTTPException(401, "Token has no subject claim.")
	if not payload.emails:
	raise HTTPException(400, "emails must not be empty.")
	if not settings.supabase_configured:
	raise HTTPException(503, "Server persistence is not configured (SUPABASE_URL/SERVICE key).")

	job = await asyncio.to_thread(repo.get_verifier_job, payload.job_id)
	if not job:
	raise HTTPException(404, f"Job {payload.job_id} not found — create it first.")
	if str(job.get("user_id")) != user_id:
	# Ownership mismatch: identical response for missing/foreign jobs.
	raise HTTPException(404, f"Job {payload.job_id} not found for this account.")
	if job.get("status") not in ("queued", "processing"):
	raise HTTPException(409, f"Job {payload.job_id} is already {job.get('status')}.")
	if not payload.emails:
	raise HTTPException(400, "emails must not be empty.")

	# Cap one submission so a giant file cannot pin the runner forever: the
	# client splits into final-flagged batches (20k ≈ 600 KB JSON).
	MAX_PER_CALL = 20000
	emails = payload.emails[:MAX_PER_CALL]

	task = asyncio.create_task(
	_run_verifier_job(payload.job_id, user_id, emails, payload.deep, payload.final)
	)
	_job_tasks.add(task)
	task.add_done_callback(_job_tasks.discard)

	return {
	"accepted": len(emails),
	"job_id": payload.job_id,
	"status": "processing",
	"final": payload.final,
	"message": "Processing server-side — poll verifier_jobs for progress.",
	}


	@router.get("/jobs/{job_id}", tags=["Bulk Jobs"])
	async def job_status(job_id: str, request: Request):
	"""Lightweight engine-side echo of the job row (debug aid; the dashboard
	polls the DB directly and does not depend on this)."""
	token = request.headers.get("authorization", "")
	token = token[7:].strip() if token.lower().startswith("bearer ") else ""
	if not token:
	raise HTTPException(401, "Missing bearer token — sign in first.")
	try:
	claims = await verify_supabase_jwt(token)
	except Exception as exc:
	raise HTTPException(401, f"Invalid or expired token: {exc}")
	job = await asyncio.to_thread(repo.get_verifier_job, job_id)
	if not job or str(job.get("user_id")) != claims.get("sub", ""):
	raise HTTPException(404, f"Job {job_id} not found for this account.")
	return {
	"job_id": job_id,
	"status": job.get("status"),
	"total": job.get("total_count"),
	"processed": job.get("processed_count"),
	"valid": job.get("valid_count"),
	"risky": job.get("risky_count"),
	"bad": job.get("bad_count"),
	}