""" Art-DAG L1 Server Application Factory. Creates and configures the FastAPI application with all routers and middleware. """ import secrets import time from pathlib import Path from urllib.parse import quote from fastapi import FastAPI, Request from fastapi.responses import JSONResponse, RedirectResponse from fastapi.staticfiles import StaticFiles from artdag_common import create_jinja_env from artdag_common.middleware.auth import get_user_from_cookie from .config import settings # Paths that should never trigger a silent auth check _SKIP_PREFIXES = ("/auth/", "/static/", "/api/", "/ipfs/", "/download/", "/inbox", "/health", "/internal/", "/oembed") _SILENT_CHECK_COOLDOWN = 300 # 5 minutes _DEVICE_COOKIE = "artdag_did" _DEVICE_COOKIE_MAX_AGE = 30 * 24 * 3600 # 30 days # Derive external base URL from oauth_redirect_uri (e.g. https://celery-artdag.rose-ash.com) _EXTERNAL_BASE = settings.oauth_redirect_uri.rsplit("/auth/callback", 1)[0] def _external_url(request: Request) -> str: """Build external URL from request path + query, using configured base domain.""" url = f"{_EXTERNAL_BASE}{request.url.path}" if request.url.query: url += f"?{request.url.query}" return url def create_app() -> FastAPI: """ Create and configure the L1 FastAPI application. Returns: Configured FastAPI instance """ app = FastAPI( title="Art-DAG L1 Server", description="Content-addressed media processing with distributed execution", version="1.0.0", ) # Database lifecycle events from database import init_db, close_db @app.on_event("startup") async def startup(): await init_db() @app.on_event("shutdown") async def shutdown(): await close_db() # Silent auth check — auto-login via prompt=none OAuth # NOTE: registered BEFORE device_id so device_id is outermost (runs first) @app.middleware("http") async def silent_auth_check(request: Request, call_next): path = request.url.path if ( request.method != "GET" or any(path.startswith(p) for p in _SKIP_PREFIXES) or request.headers.get("hx-request") # skip HTMX ): return await call_next(request) # Already logged in — but verify account hasn't logged out if get_user_from_cookie(request): device_id = getattr(request.state, "device_id", None) if device_id: try: from .dependencies import get_redis_client r = get_redis_client() if not r.get(f"did_auth:{device_id}"): # Account logged out — clear our cookie response = await call_next(request) response.delete_cookie("artdag_session") response.delete_cookie("pnone_at") return response except Exception: pass return await call_next(request) # Check cooldown — don't re-check within 5 minutes pnone_at = request.cookies.get("pnone_at") if pnone_at: try: pnone_ts = float(pnone_at) if (time.time() - pnone_ts) < _SILENT_CHECK_COOLDOWN: # But first check if account signalled a login via inbox delivery device_id = getattr(request.state, "device_id", None) if device_id: try: from .dependencies import get_redis_client r = get_redis_client() auth_ts = r.get(f"did_auth:{device_id}") if auth_ts and float(auth_ts) > pnone_ts: # Login happened since our last check — retry current_url = _external_url(request) return RedirectResponse( url=f"/auth/login?prompt=none&next={quote(current_url, safe='')}", status_code=302, ) except Exception: pass return await call_next(request) except (ValueError, TypeError): pass # Redirect to silent OAuth check current_url = _external_url(request) return RedirectResponse( url=f"/auth/login?prompt=none&next={quote(current_url, safe='')}", status_code=302, ) # Device ID middleware — track browser identity across domains # Registered AFTER silent_auth_check so it's outermost (always runs) @app.middleware("http") async def device_id_middleware(request: Request, call_next): did = request.cookies.get(_DEVICE_COOKIE) if did: request.state.device_id = did request.state._new_device_id = False else: request.state.device_id = secrets.token_urlsafe(32) request.state._new_device_id = True response = await call_next(request) if getattr(request.state, "_new_device_id", False): response.set_cookie( key=_DEVICE_COOKIE, value=request.state.device_id, max_age=_DEVICE_COOKIE_MAX_AGE, httponly=True, samesite="lax", secure=True, ) return response # Initialize Jinja2 templates template_dir = Path(__file__).parent / "templates" app.state.templates = create_jinja_env(template_dir) # Custom 404 handler @app.exception_handler(404) async def not_found_handler(request: Request, exc): from artdag_common.middleware import wants_html if wants_html(request): from artdag_common import render return render(app.state.templates, "404.html", request, user=None, status_code=404, ) return JSONResponse({"detail": "Not found"}, status_code=404) # Include routers from .routers import auth, storage, api, recipes, cache, runs, home, effects, inbox, fragments, oembed # Home and auth routers (root level) app.include_router(home.router, tags=["home"]) app.include_router(auth.router, prefix="/auth", tags=["auth"]) app.include_router(inbox.router, tags=["inbox"]) app.include_router(fragments.router, tags=["fragments"]) app.include_router(oembed.router, tags=["oembed"]) # Feature routers app.include_router(storage.router, prefix="/storage", tags=["storage"]) app.include_router(api.router, prefix="/api", tags=["api"]) # Runs and recipes routers app.include_router(runs.router, prefix="/runs", tags=["runs"]) app.include_router(recipes.router, prefix="/recipes", tags=["recipes"]) # Cache router - handles /cache and /media app.include_router(cache.router, prefix="/cache", tags=["cache"]) # Also mount cache router at /media for convenience app.include_router(cache.router, prefix="/media", tags=["media"]) # Effects router app.include_router(effects.router, prefix="/effects", tags=["effects"]) return app # Create the default app instance app = create_app()