Spaces:

sitcod3
/

modul-x

Running

Z User

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

b93f0d1 about 14 hours ago

7.73 kB

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 <backend>/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"
    }

	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 <backend>/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"
	}