Implement infra hardening: streaming, refresh tokens, structlog, Sentry, Redis

- SSE streaming on /generate-pseudocode with live token delivery
- HttpOnly refresh token cookies (7d) + silent refresh in AuthContext
- Access tokens in memory only (15min), no localStorage
- Dual SQLite/Postgres with refresh_tokens table
- structlog JSON/console renderer via LOG_JSON env
- Sentry init on SENTRY_DSN (FastAPI + Starlette integrations)
- Optional Redis backend for slowapi rate limiter via REDIS_URL
- Redis response cache on /v1/generate-pseudocode (1h TTL, SHA-256 key)
- Health check at GET /health
- React.lazy for auth modal pages (separate build chunks)
- CORS credentials=True only when CORS_ORIGINS is explicit (not wildcard)

app.py

@@ -1,19 +1,28 @@
+import json
 import os
 from contextlib import asynccontextmanager
 from pathlib import Path
 
+import structlog
 from dotenv import load_dotenv
 from fastapi import Depends, FastAPI, Header, HTTPException, Request, APIRouter
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import StreamingResponse
 from pydantic import BaseModel, Field
 from slowapi import Limiter, _rate_limit_exceeded_handler
 from slowapi.util import get_remote_address
 from slowapi.errors import RateLimitExceeded
 from typing import Annotated
-import logging
 
 from ai_prompts import TEMPLATES
-from utils import call_llm, call_llm_messages
+from utils import (
+    call_llm,
+    call_llm_messages,
+    call_llm_stream,
+    get_cached_response,
+    set_cached_response,
+    make_cache_key,
+)
 from database import (
     init_db,
     GUEST_DAILY_LIMIT,
@@ -27,6 +36,19 @@ from routers.auth import router as auth_router
 _BACKEND_DIR = Path(__file__).resolve().parent
 load_dotenv(dotenv_path=_BACKEND_DIR / ".env")
 
+# ── Sentry (optional) ─────────────────────────────────────────────────────────
+_SENTRY_DSN = os.getenv("SENTRY_DSN")
+if _SENTRY_DSN:
+    import sentry_sdk
+    from sentry_sdk.integrations.fastapi import FastApiIntegration
+    from sentry_sdk.integrations.starlette import StarletteIntegration
+    sentry_sdk.init(
+        dsn=_SENTRY_DSN,
+        integrations=[StarletteIntegration(), FastApiIntegration()],
+        traces_sample_rate=float(os.getenv("SENTRY_TRACES_SAMPLE_RATE", "0.2")),
+        send_default_pii=False,
+    )
+
 _REQUIRED_ENV = ["PROVIDER"]
 _PROVIDER_KEYS = {
     "openai": "OPENAI_API_KEY",
@@ -50,35 +72,41 @@ def _check_env() -> None:
 
 _check_env()
 
-logging.basicConfig(
-    level=getattr(logging, os.getenv("LOG_LEVEL", "INFO").upper(), logging.INFO),
-    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
-    datefmt="%Y-%m-%d %H:%M:%S",
-)
-logger = logging.getLogger("pseudogen")
+logger = structlog.get_logger("pseudogen.app")
 
 
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     init_db()
+    logger.info("app.started")
     yield
+    logger.info("app.stopped")
 
 
-limiter = Limiter(key_func=get_remote_address)
-app = FastAPI(title="Pseudogen API", lifespan=lifespan)
+# ── Rate limiter (Redis-backed when REDIS_URL is set) ─────────────────────────
+_redis_url = os.getenv("REDIS_URL")
+limiter = Limiter(
+    key_func=get_remote_address,
+    **{"storage_uri": _redis_url} if _redis_url else {},
+)
+
+app = FastAPI(title="Pseudogen API", lifespan=lifespan, docs_url=None, redoc_url=None)
 app.state.limiter = limiter
 app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)
 
-_cors_origins = os.getenv("CORS_ORIGINS", "*").strip()
-allow_origins = (
-    [o.strip() for o in _cors_origins.split(",") if o.strip()]
-    if _cors_origins != "*"
-    else ["*"]
-)
+# CORS: credentials require explicit origins (can't mix * with allow_credentials=True)
+_cors_origins_env = os.getenv("CORS_ORIGINS", "*").strip()
+if _cors_origins_env == "*":
+    _allow_origins = ["*"]
+    _allow_credentials = False
+else:
+    _allow_origins = [o.strip() for o in _cors_origins_env.split(",") if o.strip()]
+    _allow_credentials = True
+
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=allow_origins,
-    allow_credentials=True,
+    allow_origins=_allow_origins,
+    allow_credentials=_allow_credentials,
     allow_methods=["*"],
     allow_headers=["*"],
 )
@@ -124,9 +152,7 @@ _STYLE_SYSTEM = {
     ),
 }
 
-
 app.include_router(auth_router)
-
 v1_router = APIRouter(prefix="/v1", tags=["v1"])
 
 
@@ -135,6 +161,11 @@ async def root():
     return {"service": "Pseudogen API", "version": "1"}
 
 
+@app.get("/health")
+async def health():
+    return {"status": "ok"}
+
+
 @app.get("/usage")
 async def usage(
     user: dict | None = Depends(get_optional_user),
@@ -152,12 +183,7 @@ async def usage(
         return {"used": 0, "limit": GUEST_DAILY_LIMIT, "remaining": GUEST_DAILY_LIMIT, "is_guest": True}
 
     used = get_usage_today(identifier)
-    return {
-        "used": used,
-        "limit": limit,
-        "remaining": max(0, limit - used),
-        "is_guest": is_guest,
-    }
+    return {"used": used, "limit": limit, "remaining": max(0, limit - used), "is_guest": is_guest}
 
 
 @app.post("/summarize")
@@ -173,76 +199,124 @@ async def summarize_title(request: Request, req: SummarizeRequest):
         title = title.strip().split("\n")[0][:60]
         return {"title": title}
     except Exception:
-        logger.exception("Summarize failed")
+        logger.exception("summarize.failed")
         raise HTTPException(status_code=502, detail="Summarization failed")
 
 
-@v1_router.post("/generate-pseudocode")
+def _resolve_identity(user: dict | None, x_session_id: str | None):
+    if user:
+        return f"user:{user['id']}", USER_DAILY_LIMIT, False
+    if x_session_id:
+        return f"session:{x_session_id}", GUEST_DAILY_LIMIT, True
+    raise HTTPException(status_code=400, detail="A session ID or account is required.")
+
+
+def _build_messages(req: GenerateRequest) -> list | None:
+    if not req.context:
+        return None
+    system_msg = (
+        f"{_STYLE_SYSTEM.get(req.style, 'You generate pseudocode.')} "
+        f"Detail level: {req.detail}. "
+        "When asked to modify or improve, update the pseudocode accordingly."
+    )
+    context = req.context[-10:]
+    return [
+        {"role": "system", "content": system_msg},
+        *[{"role": m.role, "content": m.content} for m in context],
+        {"role": "user", "content": req.problem_description},
+    ]
+
+
+# ── Main endpoint — SSE streaming ─────────────────────────────────────────────
+
+@app.post("/generate-pseudocode")
 @limiter.limit("30/minute")
-async def generate_v1(
+async def generate(
     request: Request,
     req: GenerateRequest,
     user: dict | None = Depends(get_optional_user),
     x_session_id: str | None = Header(default=None),
 ):
-    return await _generate(req, user, x_session_id)
+    identifier, limit, is_guest = _resolve_identity(user, x_session_id)
+    used = get_usage_today(identifier)
+    if used >= limit:
+        raise HTTPException(
+            status_code=429,
+            detail=(
+                f"You've used all {limit} free prompts for today. Create a free account to get {USER_DAILY_LIMIT} per day."
+                if is_guest
+                else f"Daily limit of {limit} prompts reached. Resets at midnight UTC."
+            ),
+        )
 
+    # Increment before streaming to prevent quota abuse via cancel
+    new_count = increment_usage_today(identifier)
+    remaining = max(0, limit - new_count)
+    messages = _build_messages(req)
 
-app.include_router(v1_router)
+    def _sse():
+        try:
+            if messages:
+                token_stream = call_llm_stream(messages)
+            else:
+                template = TEMPLATES.get(req.style)
+                if template is None:
+                    yield f"data: {json.dumps({'error': 'Unknown style'})}\n\n"
+                    return
+                prompt = template.format(user_input=req.problem_description, detail=req.detail)
+                token_stream = call_llm_stream([{"role": "user", "content": prompt}])
 
+            for token in token_stream:
+                yield f"data: {json.dumps({'token': token})}\n\n"
 
-@app.post("/generate-pseudocode")
+            yield f"data: {json.dumps({'usage': {'used': new_count, 'limit': limit, 'remaining': remaining, 'is_guest': is_guest}})}\n\n"
+            yield "data: [DONE]\n\n"
+        except Exception as exc:
+            logger.error("generate.stream.error", error=str(exc))
+            yield f"data: {json.dumps({'error': 'Generation failed. Please try again.'})}\n\n"
+
+    return StreamingResponse(_sse(), media_type="text/event-stream")
+
+
+# ── v1 endpoint — non-streaming with Redis cache ──────────────────────────────
+
+@v1_router.post("/generate-pseudocode")
 @limiter.limit("30/minute")
-async def generate(
+async def generate_v1(
     request: Request,
     req: GenerateRequest,
     user: dict | None = Depends(get_optional_user),
     x_session_id: str | None = Header(default=None),
 ):
-    return await _generate(req, user, x_session_id)
-
-
-async def _generate(req: GenerateRequest, user: dict | None, x_session_id: str | None):
-    if user:
-        identifier = f"user:{user['id']}"
-        limit = USER_DAILY_LIMIT
-        is_guest = False
-    elif x_session_id:
-        identifier = f"session:{x_session_id}"
-        limit = GUEST_DAILY_LIMIT
-        is_guest = True
-    else:
-        raise HTTPException(
-            status_code=400,
-            detail="A session ID or account is required.",
-        )
+    identifier, limit, is_guest = _resolve_identity(user, x_session_id)
+
+    cache_key = make_cache_key(req.problem_description, req.style, req.detail)
+    cached = get_cached_response(cache_key)
+    if cached:
+        used = get_usage_today(identifier)
+        return {
+            "markdown": cached,
+            "used": used,
+            "limit": limit,
+            "remaining": max(0, limit - used),
+            "is_guest": is_guest,
+            "cached": True,
+        }
 
     used = get_usage_today(identifier)
     if used >= limit:
-        if is_guest:
-            raise HTTPException(
-                status_code=429,
-                detail=f"You've used all {limit} free prompts for today. Create a free account to get {USER_DAILY_LIMIT} per day.",
-            )
-        else:
-            raise HTTPException(
-                status_code=429,
-                detail=f"Daily limit of {limit} prompts reached. Resets at midnight UTC.",
-            )
+        raise HTTPException(
+            status_code=429,
+            detail=(
+                f"You've used all {limit} free prompts for today. Create a free account to get {USER_DAILY_LIMIT} per day."
+                if is_guest
+                else f"Daily limit of {limit} prompts reached. Resets at midnight UTC."
+            ),
+        )
 
     try:
-        if req.context:
-            system_msg = (
-                f"{_STYLE_SYSTEM.get(req.style, 'You generate pseudocode.')} "
-                f"Detail level: {req.detail}. "
-                "When asked to modify or improve, update the pseudocode accordingly."
-            )
-            context = req.context[-10:]
-            messages = [
-                {"role": "system", "content": system_msg},
-                *[{"role": m.role, "content": m.content} for m in context],
-                {"role": "user", "content": req.problem_description},
-            ]
+        messages = _build_messages(req)
+        if messages:
             response_text = call_llm_messages(messages)
         else:
             template = TEMPLATES.get(req.style)
@@ -253,9 +327,10 @@ async def _generate(req: GenerateRequest, user: dict | None, x_session_id: str |
     except HTTPException:
         raise
     except Exception:
-        logger.exception("LLM call failed")
+        logger.exception("generate_v1.failed")
         raise HTTPException(status_code=502, detail="Failed to generate pseudocode. Please try again.")
 
+    set_cached_response(cache_key, response_text)
     new_count = increment_usage_today(identifier)
     remaining = max(0, limit - new_count)
 
@@ -265,4 +340,8 @@ async def _generate(req: GenerateRequest, user: dict | None, x_session_id: str |
         "limit": limit,
         "remaining": remaining,
         "is_guest": is_guest,
+        "cached": False,
     }
+
+
+app.include_router(v1_router)

auth.py

@@ -1,7 +1,3 @@
-# backend/auth.py
-"""
-JWT and password hashing for Pseudogen. get_current_user dependency for protected routes.
-"""
 import os
 from datetime import datetime, timedelta, timezone
 
@@ -18,7 +14,7 @@ load_dotenv(dotenv_path=Path(__file__).resolve().parent / ".env")
 
 SECRET_KEY = os.getenv("SECRET_KEY", "change-me-in-production-use-env")
 ALGORITHM = "HS256"
-ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7  # 7 days
+ACCESS_TOKEN_EXPIRE_MINUTES = 15
 
 pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
 security = HTTPBearer(auto_error=False)

database.py

@@ -1,41 +1,113 @@
-import sqlite3
-from datetime import datetime, timezone
+import os
+import hashlib
+import secrets
+from datetime import datetime, timezone, timedelta
 from pathlib import Path
 
 _BACKEND_DIR = Path(__file__).resolve().parent
 DB_PATH = _BACKEND_DIR / "pseudogen.db"
 
+_USE_PG = bool(os.getenv("DATABASE_URL"))
+P = "%s" if _USE_PG else "?"
+
 GUEST_DAILY_LIMIT = 5
 USER_DAILY_LIMIT = 10
 
 
 def get_connection():
-    conn = sqlite3.connect(DB_PATH)
-    conn.row_factory = sqlite3.Row
-    return conn
+    if _USE_PG:
+        import psycopg2
+        url = os.getenv("DATABASE_URL", "")
+        if url.startswith("postgres://"):
+            url = url.replace("postgres://", "postgresql://", 1)
+        return psycopg2.connect(url)
+    else:
+        import sqlite3
+        conn = sqlite3.connect(DB_PATH)
+        conn.row_factory = sqlite3.Row
+        return conn
+
+
+def _one(conn, sql, params=()):
+    if _USE_PG:
+        import psycopg2.extras
+        cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
+        cur.execute(sql, params)
+        row = cur.fetchone()
+        return dict(row) if row else None
+    else:
+        row = conn.execute(sql, params).fetchone()
+        return dict(row) if row else None
+
+
+def _exec(conn, sql, params=()):
+    if _USE_PG:
+        cur = conn.cursor()
+        cur.execute(sql, params)
+        return cur
+    else:
+        return conn.execute(sql, params)
 
 
 def init_db():
     conn = get_connection()
     try:
-        conn.execute("""
-            CREATE TABLE IF NOT EXISTS users (
-                id INTEGER PRIMARY KEY AUTOINCREMENT,
-                email TEXT UNIQUE NOT NULL,
-                hashed_password TEXT NOT NULL,
-                plan TEXT NOT NULL DEFAULT 'free',
-                created_at TEXT NOT NULL DEFAULT (datetime('now'))
-            )
-        """)
-        conn.execute("""
-            CREATE TABLE IF NOT EXISTS daily_usage (
-                id INTEGER PRIMARY KEY AUTOINCREMENT,
-                identifier TEXT NOT NULL,
-                date TEXT NOT NULL,
-                count INTEGER NOT NULL DEFAULT 0,
-                UNIQUE(identifier, date)
-            )
-        """)
+        if _USE_PG:
+            _exec(conn, """
+                CREATE TABLE IF NOT EXISTS users (
+                    id BIGSERIAL PRIMARY KEY,
+                    email TEXT UNIQUE NOT NULL,
+                    hashed_password TEXT NOT NULL,
+                    plan TEXT NOT NULL DEFAULT 'free',
+                    created_at TEXT NOT NULL DEFAULT NOW()::TEXT
+                )
+            """)
+            _exec(conn, """
+                CREATE TABLE IF NOT EXISTS daily_usage (
+                    id BIGSERIAL PRIMARY KEY,
+                    identifier TEXT NOT NULL,
+                    date TEXT NOT NULL,
+                    count INTEGER NOT NULL DEFAULT 0,
+                    UNIQUE(identifier, date)
+                )
+            """)
+            _exec(conn, """
+                CREATE TABLE IF NOT EXISTS refresh_tokens (
+                    id BIGSERIAL PRIMARY KEY,
+                    token_hash TEXT UNIQUE NOT NULL,
+                    user_id BIGINT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+                    expires_at TEXT NOT NULL,
+                    created_at TEXT NOT NULL DEFAULT NOW()::TEXT
+                )
+            """)
+        else:
+            _exec(conn, """
+                CREATE TABLE IF NOT EXISTS users (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    email TEXT UNIQUE NOT NULL,
+                    hashed_password TEXT NOT NULL,
+                    plan TEXT NOT NULL DEFAULT 'free',
+                    created_at TEXT NOT NULL DEFAULT (datetime('now'))
+                )
+            """)
+            _exec(conn, """
+                CREATE TABLE IF NOT EXISTS daily_usage (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    identifier TEXT NOT NULL,
+                    date TEXT NOT NULL,
+                    count INTEGER NOT NULL DEFAULT 0,
+                    UNIQUE(identifier, date)
+                )
+            """)
+            _exec(conn, """
+                CREATE TABLE IF NOT EXISTS refresh_tokens (
+                    id INTEGER PRIMARY KEY AUTOINCREMENT,
+                    token_hash TEXT UNIQUE NOT NULL,
+                    user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+                    expires_at TEXT NOT NULL,
+                    created_at TEXT NOT NULL DEFAULT (datetime('now'))
+                )
+            """)
         conn.commit()
     finally:
         conn.close()
@@ -44,11 +116,11 @@ def init_db():
 def get_user_by_email(email: str) -> dict | None:
     conn = get_connection()
     try:
-        row = conn.execute(
-            "SELECT id, email, hashed_password, plan, created_at FROM users WHERE email = ?",
+        return _one(
+            conn,
+            f"SELECT id, email, hashed_password, plan, created_at FROM users WHERE email = {P}",
             (email.strip().lower(),),
-        ).fetchone()
-        return dict(row) if row else None
+        )
     finally:
         conn.close()
 
@@ -56,11 +128,11 @@ def get_user_by_email(email: str) -> dict | None:
 def get_user_by_id(user_id: int) -> dict | None:
     conn = get_connection()
     try:
-        row = conn.execute(
-            "SELECT id, email, plan, created_at FROM users WHERE id = ?",
+        return _one(
+            conn,
+            f"SELECT id, email, plan, created_at FROM users WHERE id = {P}",
             (user_id,),
-        ).fetchone()
-        return dict(row) if row else None
+        )
     finally:
         conn.close()
 
@@ -68,12 +140,22 @@ def get_user_by_id(user_id: int) -> dict | None:
 def create_user(email: str, hashed_password: str, plan: str = "free") -> dict:
     conn = get_connection()
     try:
-        cursor = conn.execute(
-            "INSERT INTO users (email, hashed_password, plan) VALUES (?, ?, ?)",
-            (email.strip().lower(), hashed_password, plan),
-        )
-        conn.commit()
-        return {"id": cursor.lastrowid, "email": email.strip().lower(), "plan": plan}
+        if _USE_PG:
+            row = _one(
+                conn,
+                f"INSERT INTO users (email, hashed_password, plan) VALUES ({P}, {P}, {P}) RETURNING id, email, plan",
+                (email.strip().lower(), hashed_password, plan),
+            )
+            conn.commit()
+            return row
+        else:
+            cur = _exec(
+                conn,
+                f"INSERT INTO users (email, hashed_password, plan) VALUES ({P}, {P}, {P})",
+                (email.strip().lower(), hashed_password, plan),
+            )
+            conn.commit()
+            return {"id": cur.lastrowid, "email": email.strip().lower(), "plan": plan}
     finally:
         conn.close()
 
@@ -85,10 +167,11 @@ def _today_utc() -> str:
 def get_usage_today(identifier: str) -> int:
     conn = get_connection()
     try:
-        row = conn.execute(
-            "SELECT count FROM daily_usage WHERE identifier = ? AND date = ?",
+        row = _one(
+            conn,
+            f"SELECT count FROM daily_usage WHERE identifier = {P} AND date = {P}",
             (identifier, _today_utc()),
-        ).fetchone()
+        )
         return row["count"] if row else 0
     finally:
         conn.close()
@@ -98,19 +181,93 @@ def increment_usage_today(identifier: str) -> int:
     today = _today_utc()
     conn = get_connection()
     try:
-        conn.execute(
-            """
+        _exec(
+            conn,
+            f"""
             INSERT INTO daily_usage (identifier, date, count)
-            VALUES (?, ?, 1)
-            ON CONFLICT(identifier, date) DO UPDATE SET count = count + 1
+            VALUES ({P}, {P}, 1)
+            ON CONFLICT(identifier, date) DO UPDATE SET count = daily_usage.count + 1
             """,
             (identifier, today),
         )
         conn.commit()
-        row = conn.execute(
-            "SELECT count FROM daily_usage WHERE identifier = ? AND date = ?",
+        row = _one(
+            conn,
+            f"SELECT count FROM daily_usage WHERE identifier = {P} AND date = {P}",
             (identifier, today),
-        ).fetchone()
-        return row["count"]
+        )
+        return row["count"] if row else 1
+    finally:
+        conn.close()
+
+
+# ── Refresh tokens ────────────────────────────────────────────────────────────
+
+REFRESH_TOKEN_DAYS = 7
+
+
+def _hash_token(value: str) -> str:
+    return hashlib.sha256(value.encode()).hexdigest()
+
+
+def create_refresh_token(user_id: int) -> str:
+    """Create a refresh token, persist its hash, return the raw value."""
+    value = secrets.token_urlsafe(32)
+    token_hash = _hash_token(value)
+    expires_at = (
+        datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_DAYS)
+    ).isoformat()
+    conn = get_connection()
+    try:
+        _exec(
+            conn,
+            f"INSERT INTO refresh_tokens (token_hash, user_id, expires_at) VALUES ({P}, {P}, {P})",
+            (token_hash, user_id, expires_at),
+        )
+        conn.commit()
+    finally:
+        conn.close()
+    return value
+
+
+def validate_refresh_token(value: str) -> int | None:
+    """Return user_id if valid and not expired, else None."""
+    token_hash = _hash_token(value)
+    conn = get_connection()
+    try:
+        row = _one(
+            conn,
+            f"SELECT user_id, expires_at FROM refresh_tokens WHERE token_hash = {P}",
+            (token_hash,),
+        )
+        if not row:
+            return None
+        expires = datetime.fromisoformat(row["expires_at"])
+        if expires.tzinfo is None:
+            expires = expires.replace(tzinfo=timezone.utc)
+        if datetime.now(timezone.utc) > expires:
+            _exec(conn, f"DELETE FROM refresh_tokens WHERE token_hash = {P}", (token_hash,))
+            conn.commit()
+            return None
+        return row["user_id"]
+    finally:
+        conn.close()
+
+
+def revoke_refresh_token(value: str) -> None:
+    token_hash = _hash_token(value)
+    conn = get_connection()
+    try:
+        _exec(conn, f"DELETE FROM refresh_tokens WHERE token_hash = {P}", (token_hash,))
+        conn.commit()
+    finally:
+        conn.close()
+
+
+def revoke_all_refresh_tokens(user_id: int) -> None:
+    conn = get_connection()
+    try:
+        _exec(conn, f"DELETE FROM refresh_tokens WHERE user_id = {P}", (user_id,))
+        conn.commit()
     finally:
         conn.close()

requirements.txt

@@ -19,6 +19,16 @@ bcrypt==4.0.1
 PyJWT==2.10.1
 email-validator==2.2.0
 
+# Database
+psycopg2-binary==2.9.10
+
+# Observability
+structlog==24.4.0
+sentry-sdk[fastapi]==2.19.2
+
+# Caching / Redis rate-limit backend
+redis==5.2.1
+
 # Testing
 pytest==8.3.4
 pytest-asyncio==0.24.0

routers/auth.py

@@ -1,11 +1,41 @@
-from fastapi import APIRouter, Depends, HTTPException, status
+import os
+from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
 from pydantic import BaseModel, EmailStr
 
-from database import get_user_by_email, create_user
+from database import (
+    get_user_by_email,
+    create_user,
+    get_user_by_id,
+    create_refresh_token,
+    validate_refresh_token,
+    revoke_refresh_token,
+    revoke_all_refresh_tokens,
+)
 from auth import hash_password, verify_password, create_access_token, get_current_user
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 
+_COOKIE_NAME = "pseudogen_rt"
+_COOKIE_SECURE = os.getenv("COOKIE_SECURE", "false").lower() == "true"
+_COOKIE_SAMESITE = os.getenv("COOKIE_SAMESITE", "lax")
+_COOKIE_MAX_AGE = 7 * 24 * 3600
+
+
+def _set_refresh_cookie(response: Response, token_value: str) -> None:
+    response.set_cookie(
+        key=_COOKIE_NAME,
+        value=token_value,
+        httponly=True,
+        secure=_COOKIE_SECURE,
+        samesite=_COOKIE_SAMESITE,
+        max_age=_COOKIE_MAX_AGE,
+        path="/auth",
+    )
+
+
+def _clear_refresh_cookie(response: Response) -> None:
+    response.delete_cookie(key=_COOKIE_NAME, path="/auth")
+
 
 class RegisterRequest(BaseModel):
     email: EmailStr
@@ -29,7 +59,7 @@ class UserResponse(BaseModel):
 
 
 @router.post("/register", response_model=TokenResponse)
-def register(req: RegisterRequest):
+def register(req: RegisterRequest, response: Response):
     if len(req.password) < 8:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
@@ -42,20 +72,60 @@ def register(req: RegisterRequest):
         )
     hashed = hash_password(req.password)
     user = create_user(req.email, hashed)
-    token = create_access_token({"sub": str(user["id"])})
-    return TokenResponse(access_token=token)
+    access_token = create_access_token({"sub": str(user["id"])})
+    refresh_value = create_refresh_token(user["id"])
+    _set_refresh_cookie(response, refresh_value)
+    return TokenResponse(access_token=access_token)
 
 
 @router.post("/login", response_model=TokenResponse)
-def login(req: LoginRequest):
+def login(req: LoginRequest, response: Response):
     user = get_user_by_email(req.email)
     if not user or not verify_password(req.password, user["hashed_password"]):
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Invalid email or password",
         )
-    token = create_access_token({"sub": str(user["id"])})
-    return TokenResponse(access_token=token)
+    access_token = create_access_token({"sub": str(user["id"])})
+    refresh_value = create_refresh_token(user["id"])
+    _set_refresh_cookie(response, refresh_value)
+    return TokenResponse(access_token=access_token)
+
+
+@router.post("/refresh", response_model=TokenResponse)
+def refresh(request: Request, response: Response):
+    token_value = request.cookies.get(_COOKIE_NAME)
+    if not token_value:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="No refresh token")
+    user_id = validate_refresh_token(token_value)
+    if user_id is None:
+        _clear_refresh_cookie(response)
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Refresh token expired or invalid")
+    user = get_user_by_id(user_id)
+    if user is None:
+        _clear_refresh_cookie(response)
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
+    revoke_refresh_token(token_value)
+    new_access = create_access_token({"sub": str(user["id"])})
+    new_refresh = create_refresh_token(user["id"])
+    _set_refresh_cookie(response, new_refresh)
+    return TokenResponse(access_token=new_access)
+
+
+@router.post("/logout")
+def logout(request: Request, response: Response):
+    token_value = request.cookies.get(_COOKIE_NAME)
+    if token_value:
+        revoke_refresh_token(token_value)
+    _clear_refresh_cookie(response)
+    return {"detail": "Logged out"}
+
+
+@router.post("/logout-all")
+def logout_all(response: Response, user: dict = Depends(get_current_user)):
+    revoke_all_refresh_tokens(user["id"])
+    _clear_refresh_cookie(response)
+    return {"detail": "All sessions revoked"}
 
 
 @router.get("/me", response_model=UserResponse)

utils.py

@@ -1,8 +1,11 @@
 import os
 import time
+import json
+import hashlib
+import logging
 import openai
 import requests
-import logging
+import structlog
 from anthropic import Anthropic
 from requests.exceptions import RequestException
 from dotenv import load_dotenv
@@ -13,6 +16,74 @@ load_dotenv(dotenv_path=Path(__file__).resolve().parent / ".env")
 openai.api_key = os.getenv("OPENAI_API_KEY")
 _claude = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
 
+# ── Structlog setup ───────────────────────────────────────────────────────────
+
+def configure_logging() -> None:
+    log_level = getattr(logging, os.getenv("LOG_LEVEL", "INFO").upper(), logging.INFO)
+    use_json = os.getenv("LOG_JSON", "false").lower() == "true"
+
+    processors = [
+        structlog.stdlib.filter_by_level,
+        structlog.stdlib.add_logger_name,
+        structlog.stdlib.add_log_level,
+        structlog.processors.TimeStamper(fmt="iso"),
+        structlog.stdlib.PositionalArgumentsFormatter(),
+        structlog.processors.StackInfoRenderer(),
+        structlog.processors.format_exc_info,
+        structlog.processors.JSONRenderer() if use_json else structlog.dev.ConsoleRenderer(),
+    ]
+    structlog.configure(
+        processors=processors,
+        wrapper_class=structlog.stdlib.BoundLogger,
+        context_class=dict,
+        logger_factory=structlog.stdlib.LoggerFactory(),
+        cache_logger_on_first_use=True,
+    )
+    logging.basicConfig(level=log_level, format="%(message)s")
+
+
+configure_logging()
+logger = structlog.get_logger("pseudogen")
+
+# ── Redis (optional) ──────────────────────────────────────────────────────────
+
+_redis = None
+_redis_url = os.getenv("REDIS_URL")
+if _redis_url:
+    try:
+        import redis as _redis_lib
+        _redis = _redis_lib.from_url(_redis_url, decode_responses=True, socket_connect_timeout=2)
+        _redis.ping()
+        logger.info("redis.connected", url=_redis_url.split("@")[-1])
+    except Exception as e:
+        logger.warning("redis.unavailable", error=str(e))
+        _redis = None
+
+
+def get_cached_response(key: str) -> str | None:
+    if not _redis:
+        return None
+    try:
+        return _redis.get(f"pgcache:{key}")
+    except Exception:
+        return None
+
+
+def set_cached_response(key: str, value: str, ttl: int = 3600) -> None:
+    if not _redis:
+        return
+    try:
+        _redis.setex(f"pgcache:{key}", ttl, value)
+    except Exception:
+        pass
+
+
+def make_cache_key(problem: str, style: str, detail: str) -> str:
+    payload = f"{style}:{detail}:{problem.strip().lower()}"
+    return hashlib.sha256(payload.encode()).hexdigest()
+
+
+# ── Single-turn LLM (non-streaming) ──────────────────────────────────────────
 
 def call_openai_with_retries(prompt: str, model: str = None, max_retries: int = 3, backoff: float = 1.0) -> str:
     model = model or os.getenv("OPENAI_MODEL", "gpt-4o-mini")
@@ -30,7 +101,7 @@ def call_openai_with_retries(prompt: str, model: str = None, max_retries: int =
             raise RuntimeError("Empty response from OpenAI")
         except Exception as e:
             last_err = e
-            logging.warning(f"OpenAI attempt {attempt} failed: {e}")
+            logger.warning("openai.retry", attempt=attempt, error=str(e))
             if attempt < max_retries:
                 time.sleep(backoff * attempt)
     raise RuntimeError(f"OpenAI failed after {max_retries} attempts: {last_err}")
@@ -54,7 +125,7 @@ def call_claude_with_retries(prompt: str, model: str = None, max_retries: int =
             raise RuntimeError("Empty response from Claude")
         except Exception as e:
             last_err = e
-            logging.warning(f"Claude attempt {attempt} failed: {e}")
+            logger.warning("claude.retry", attempt=attempt, error=str(e))
             if attempt < max_retries:
                 time.sleep(backoff * attempt)
     raise RuntimeError(f"Claude failed after {max_retries} attempts: {last_err}")
@@ -66,37 +137,25 @@ def call_groq_with_retries(prompt: str, model: str = None, max_retries: int = 3,
         raise RuntimeError("Missing GROQ_API_KEY")
     model = model or os.getenv("GROQ_MODEL", "llama3-8b-8192")
     ssl_verify = os.getenv("GROQ_SSL_VERIFY", "true").lower() != "false"
-    headers = {
-        "Authorization": f"Bearer {api_key}",
-        "Content-Type": "application/json",
-    }
-    payload = {
-        "model": model,
-        "messages": [{"role": "user", "content": prompt}],
-        "temperature": 0.2,
-        "max_tokens": 1000,
-    }
+    headers = {"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"}
+    payload = {"model": model, "messages": [{"role": "user", "content": prompt}], "temperature": 0.2, "max_tokens": 1000}
     last_err = None
     for attempt in range(1, max_retries + 1):
         try:
             resp = requests.post(
                 "https://api.groq.com/openai/v1/chat/completions",
-                headers=headers,
-                json=payload,
-                timeout=30,
-                verify=ssl_verify,
+                headers=headers, json=payload, timeout=30, verify=ssl_verify,
             )
             if resp.status_code == 200:
-                data = resp.json()
-                content = data.get("choices", [{}])[0].get("message", {}).get("content")
+                content = resp.json().get("choices", [{}])[0].get("message", {}).get("content")
                 if content:
                     return content.strip()
                 raise RuntimeError("Empty response from Groq")
-            logging.error(f"Groq API error ({resp.status_code}): {resp.text}")
+            logger.error("groq.error", status=resp.status_code)
             resp.raise_for_status()
         except (RequestException, Exception) as e:
             last_err = e
-            logging.warning(f"Groq attempt {attempt} failed: {e}")
+            logger.warning("groq.retry", attempt=attempt, error=str(e))
             if attempt < max_retries:
                 time.sleep(backoff * attempt)
     raise RuntimeError(f"Groq failed after {max_retries} attempts: {last_err}")
@@ -104,7 +163,7 @@ def call_groq_with_retries(prompt: str, model: str = None, max_retries: int = 3,
 
 def call_llm(prompt: str) -> str:
     provider = os.getenv("PROVIDER", "openai").lower()
-    if provider == "claude" or provider == "anthropic":
+    if provider in ("claude", "anthropic"):
         return call_claude_with_retries(prompt)
     elif provider == "openai":
         return call_openai_with_retries(prompt)
@@ -113,6 +172,8 @@ def call_llm(prompt: str) -> str:
     raise RuntimeError(f"Unsupported PROVIDER: {provider}")
 
 
+# ── Multi-turn LLM (non-streaming) ───────────────────────────────────────────
+
 def call_groq_with_messages(messages: list, model: str = None, max_retries: int = 3, backoff: float = 1.0) -> str:
     api_key = os.getenv("GROQ_API_KEY")
     if not api_key:
@@ -133,11 +194,10 @@ def call_groq_with_messages(messages: list, model: str = None, max_retries: int
                 if content:
                     return content.strip()
                 raise RuntimeError("Empty response from Groq")
-            logging.error(f"Groq API error ({resp.status_code}): {resp.text}")
             resp.raise_for_status()
         except (RequestException, Exception) as e:
             last_err = e
-            logging.warning(f"Groq attempt {attempt} failed: {e}")
+            logger.warning("groq.retry", attempt=attempt, error=str(e))
             if attempt < max_retries:
                 time.sleep(backoff * attempt)
     raise RuntimeError(f"Groq failed after {max_retries} attempts: {last_err}")
@@ -156,7 +216,7 @@ def call_openai_with_messages(messages: list, model: str = None, max_retries: in
             raise RuntimeError("Empty response from OpenAI")
         except Exception as e:
             last_err = e
-            logging.warning(f"OpenAI attempt {attempt} failed: {e}")
+            logger.warning("openai.retry", attempt=attempt, error=str(e))
             if attempt < max_retries:
                 time.sleep(backoff * attempt)
     raise RuntimeError(f"OpenAI failed after {max_retries} attempts: {last_err}")
@@ -180,7 +240,7 @@ def call_claude_with_messages(messages: list, model: str = None, max_retries: in
             raise RuntimeError("Empty response from Claude")
         except Exception as e:
             last_err = e
-            logging.warning(f"Claude attempt {attempt} failed: {e}")
+            logger.warning("claude.retry", attempt=attempt, error=str(e))
             if attempt < max_retries:
                 time.sleep(backoff * attempt)
     raise RuntimeError(f"Claude failed after {max_retries} attempts: {last_err}")
@@ -195,3 +255,72 @@ def call_llm_messages(messages: list) -> str:
     elif provider == "groq":
         return call_groq_with_messages(messages)
     raise RuntimeError(f"Unsupported PROVIDER: {provider}")
+
+
+# ── Streaming LLM ─────────────────────────────────────────────────────────────
+
+def call_groq_stream(messages: list, model: str = None):
+    """Sync generator — yields text tokens from Groq SSE stream."""
+    api_key = os.getenv("GROQ_API_KEY")
+    if not api_key:
+        raise RuntimeError("Missing GROQ_API_KEY")
+    model = model or os.getenv("GROQ_MODEL", "llama3-8b-8192")
+    ssl_verify = os.getenv("GROQ_SSL_VERIFY", "true").lower() != "false"
+    resp = requests.post(
+        "https://api.groq.com/openai/v1/chat/completions",
+        headers={"Authorization": f"Bearer {api_key}", "Content-Type": "application/json"},
+        json={"model": model, "messages": messages, "temperature": 0.2, "max_tokens": 1000, "stream": True},
+        timeout=60,
+        verify=ssl_verify,
+        stream=True,
+    )
+    resp.raise_for_status()
+    for line in resp.iter_lines():
+        if not line or line == b"data: [DONE]":
+            continue
+        if line.startswith(b"data: "):
+            try:
+                data = json.loads(line[6:])
+                delta = data["choices"][0]["delta"].get("content", "")
+                if delta:
+                    yield delta
+            except (json.JSONDecodeError, KeyError, IndexError):
+                pass
+
+
+def call_openai_stream(messages: list, model: str = None):
+    """Sync generator — yields text tokens from OpenAI streaming."""
+    model = model or os.getenv("OPENAI_MODEL", "gpt-4o-mini")
+    resp = openai.ChatCompletion.create(
+        model=model, messages=messages, temperature=0.2, max_tokens=1200, stream=True,
+    )
+    for chunk in resp:
+        delta = chunk["choices"][0]["delta"].get("content", "")
+        if delta:
+            yield delta
+
+
+def call_claude_stream(messages: list, model: str = None):
+    """Sync generator — yields text tokens from Claude streaming."""
+    model = model or os.getenv("CLAUDE_MODEL", "claude-3-5-haiku-20241022")
+    system_parts = [m["content"] for m in messages if m.get("role") == "system"]
+    non_system = [m for m in messages if m.get("role") != "system"]
+    kwargs = {"model": model, "max_tokens": 1000, "messages": non_system}
+    if system_parts:
+        kwargs["system"] = " ".join(system_parts)
+    with _claude.messages.stream(**kwargs) as stream:
+        for text in stream.text_stream:
+            yield text
+
+
+def call_llm_stream(messages: list):
+    """Sync generator — dispatches to the configured provider's streaming function."""
+    provider = os.getenv("PROVIDER", "openai").lower()
+    if provider in ("claude", "anthropic"):
+        yield from call_claude_stream(messages)
+    elif provider == "openai":
+        yield from call_openai_stream(messages)
+    elif provider == "groq":
+        yield from call_groq_stream(messages)
+    else:
+        raise RuntimeError(f"Unsupported PROVIDER: {provider}")