from fastapi import FastAPI, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import JSONResponse from slowapi import Limiter, _rate_limit_exceeded_handler from slowapi.util import get_remote_address from slowapi.errors import RateLimitExceeded from contextlib import asynccontextmanager import asyncio import time from .config import settings from .api.routes import router from .api.jobs import router as jobs_router from .auth import license_middleware from .utils.logger import setup_logging from .db import repo # Setup structured logging setup_logging() # Rate limiter limiter = Limiter(key_func=get_remote_address) @asynccontextmanager async def lifespan(app: FastAPI): """Startup and shutdown events.""" print(f"BRIDGE Modul - X v{settings.app_version} starting up...") # Verify Supabase connectivity at boot (best-effort, never fatal) if settings.supabase_configured: try: status = await repo.db_status() if status.get("reachable"): print(f"[DB] Supabase reachable — tables_ready={status.get('tables_ready')}") if not status.get("tables_ready"): print("[DB] WARNING: run supabase/schema.sql in the Supabase SQL editor " "to create the tables.") else: print(f"[DB] WARNING: Supabase not reachable — {status.get('detail')}") except Exception as exc: # pragma: no cover - defensive print(f"[DB] WARNING: DB check failed — {exc}") else: print("[DB] Supabase not configured — persistence disabled (see .env.example)") # Port-25 egress preflight: mailbox verification cannot work without it, so # report it at boot instead of letting operators discover it from a wall of # "smtp_unreachable" results. Best-effort and never fatal. try: from .validator.egress import preflight_or_warn await asyncio.to_thread(preflight_or_warn) except Exception as exc: # pragma: no cover - defensive print(f"[EGRESS] preflight skipped — {exc}") # Greylist retry-queue drain loop (in-process — the Space runs no Celery # beat, so without this the queue would fill forever and greylisted # mailboxes would stay permanently risky). Fail-open, never fatal. drain_task = None try: from .tasks.drain_loop import DrainLoop from .validator.engine import engine as _engine async def _prober(email: str, mx_host: str): return await _engine.mailbox.verify(email, mx_host) loop = DrainLoop(prober=_prober, interval_seconds=300) drain_task = asyncio.create_task(loop.start()) print("[RETRY] drain loop started (every 300s)") except Exception as exc: # pragma: no cover - defensive print(f"[RETRY] drain loop skipped — {exc}") # Warm the advisory model at startup: the first request otherwise pays the # torch + checkpoint load (measured 42.6s cold start) even though inference # now runs off the event loop. Fail-open and never fatal. try: from .ml.runtime import get_model_async model = await get_model_async() if model is not None and getattr(model, "loaded", False): print("[ML] advisory model preloaded") else: print("[ML] advisory model unavailable - deterministic mode") except Exception as exc: # pragma: no cover - defensive print(f"[ML] preload skipped — {exc}") # Warm the LOCAL risk scorer at startup (Part B architecture): weights are # baked into the image (HF_HOME=/app/models), so this is a fast disk-cache # load (~seconds) with ZERO network. Without it, the first paying request # would pay the model load. Fail-open: a failed load trips the circuit # breaker inside the scorer and the heuristic takes over (spec §2.3). try: from .ml.risk_scorer import ai_scorer as _scorer if _scorer._inference_configured(): await asyncio.to_thread(_scorer._ensure_loaded) st = _scorer.status() print( f"[AI] local risk scorer preloaded: model={st['model']} " f"loaded={st['model_loaded']} " f"circuit_open={st['circuit']['open']}" ) except Exception as exc: # pragma: no cover - defensive print(f"[AI] risk scorer preload skipped — {exc}") yield if drain_task is not None: drain_task.cancel() print("BRIDGE Modul - X shutting down...") # Create FastAPI app app = FastAPI( title=settings.app_name, description=""" BRIDGE Modul - X — 7-Layer Email Validation Engine. Features: - Layer 1 — RFC 5321/5322 Syntax Validation - Layer 2 — DNS Domain Existence Check - Layer 3 — MX Record Verification - Layer 4 — Disposable/Role-Based/Free Provider Detection - Layer 5 — Catch-All Domain Detection - Layer 6 — SMTP Mailbox Verification (with proxy support) - Layer 7 — Deliverability Scoring & Risk Assessment Provider support: Gmail, Outlook, Yahoo, iCloud, ProtonMail, Zoho, 100K+ domains """, version=settings.app_version, docs_url="/docs", redoc_url="/redoc", lifespan=lifespan ) # ---- Middleware ----------------------------------------------------- app.state.limiter = limiter app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler) app.add_middleware( CORSMiddleware, allow_origins=settings.cors_origins, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], expose_headers=["X-Process-Time"] ) @app.middleware("http") async def add_process_time_header(request: Request, call_next): start_time = time.monotonic() response = await call_next(request) process_time = (time.monotonic() - start_time) * 1000 response.headers["X-Process-Time"] = f"{process_time:.2f}ms" return response # Central Supabase JWT + license gate for paid API routes. # /api/v1/health and /api/health stay open (HF + website productApi probe). # No-op while LICENSE_ENFORCEMENT != "true" (demo mode). See app/auth.py. @app.middleware("http") async def license_gate(request: Request, call_next): return await license_middleware(request, call_next) # Global error handler @app.exception_handler(Exception) async def global_exception_handler(request: Request, exc: Exception): return JSONResponse( status_code=500, content={ "error": "Internal server error", "detail": str(exc) if settings.debug else "Contact support" } ) # ---- Include Routes ------------------------------------------------- app.include_router(router, prefix="/api/v1") # Server-driven bulk jobs (POST /api/v1/jobs): the engine finalizes the run # with DB heartbeats so a closed browser tab can no longer orphan it. app.include_router(jobs_router, prefix="/api/v1") # ---- Health alias for the website productApi probe ------------------- # productApi tries GET /api/health then /health. The versioned # route lives at /api/v1/health; this alias keeps the standard probe path # working (stays open via the license_gate allow-list in app/auth.py). @app.get("/api/health", tags=["System"]) async def api_health_alias(): import time as _time return { "status": "healthy", "app": settings.app_name, "version": settings.app_version, "timestamp": _time.time(), } # ---- Root ----------------------------------------------------------- @app.get("/") async def root(): return { "app": settings.app_name, "version": settings.app_version, "docs": "/docs", "health": "/api/v1/health" }