0f3a267fb6
- alembic.ini: remove hardcoded sqlite URL - alembic/env.py: inject DATABASE_URL from container env - main.py: serialize Base.metadata.create_all() + setup_pgvector() behind a Postgres advisory lock (_run_startup_ddl). Previously all 4 uvicorn workers ran the DDL in parallel and occasionally deadlocked each other on ALTER TABLE ordering, killing one worker at startup. - CLAUDE.md: add Database migrations (Alembic) section DB was stamped at 9bac7bf02e38; no schema changes in this commit.
577 lines
25 KiB
Python
577 lines
25 KiB
Python
import os
|
|
from contextlib import asynccontextmanager
|
|
|
|
from fastapi import FastAPI
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.staticfiles import StaticFiles
|
|
|
|
from app.config import settings
|
|
from app.logging_config import setup_logging
|
|
|
|
# Configure structured JSON logging before anything else
|
|
setup_logging(settings.LOG_LEVEL)
|
|
from app.database import engine, Base, SessionLocal
|
|
from app.routers import auth, documents, quizzes, attempts, admin, tts, nextcloud, categories, questions, question_categories, favorites, teach, contact, tags, flashcards, courses
|
|
from app.utils.auth import get_password_hash
|
|
from app.utils.scheduler import start_scheduler, stop_scheduler
|
|
|
|
|
|
def seed_admin():
|
|
"""Create default admin user if none exists."""
|
|
from app.models.user import User
|
|
from app.models.email_verification import EmailVerification
|
|
from app.models.password_reset import PasswordReset
|
|
db = SessionLocal()
|
|
try:
|
|
admin_exists = db.query(User).filter(User.role == "admin").first()
|
|
if not admin_exists:
|
|
admin_user = User(
|
|
email="admin@quizapp.com",
|
|
hashed_password=get_password_hash("admin123"),
|
|
name="Admin",
|
|
role="admin",
|
|
)
|
|
db.add(admin_user)
|
|
db.flush()
|
|
# Auto-verify seeded admin
|
|
from datetime import datetime
|
|
db.add(EmailVerification(
|
|
user_id=admin_user.id,
|
|
token="seeded",
|
|
expires_at=datetime.utcnow(),
|
|
verified_at=datetime.utcnow(),
|
|
))
|
|
db.commit()
|
|
else:
|
|
# Ensure existing admin has a verified email record
|
|
from app.models.email_verification import EmailVerification
|
|
from datetime import datetime
|
|
existing_v = db.query(EmailVerification).filter(EmailVerification.user_id == admin_exists.id).first()
|
|
if not existing_v:
|
|
db.add(EmailVerification(
|
|
user_id=admin_exists.id,
|
|
token=f"legacy_{admin_exists.id}",
|
|
expires_at=datetime.utcnow(),
|
|
verified_at=datetime.utcnow(),
|
|
))
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def seed_default_models():
|
|
"""Seed default AI model configs if none exist."""
|
|
from app.models.ai_model_config import AIModelConfig
|
|
db = SessionLocal()
|
|
try:
|
|
if db.query(AIModelConfig).count() == 0:
|
|
defaults = [
|
|
AIModelConfig(name="Claude Haiku 4.5", model_id="claude-haiku-4.5", task="extraction", is_active=True, is_default=True),
|
|
AIModelConfig(name="Claude Sonnet 4.6", model_id="claude-sonnet-4.6", task="extraction", is_active=True, is_default=False),
|
|
AIModelConfig(name="Gemini 2.5 Flash", model_id="gemini-2.5-flash", task="extraction", is_active=True, is_default=False),
|
|
]
|
|
db.add_all(defaults)
|
|
db.commit()
|
|
|
|
# Clean up legacy titan-embed-v2 "general" entry (was mistakenly seeded)
|
|
titan = db.query(AIModelConfig).filter(
|
|
AIModelConfig.model_id == "titan-embed-v2",
|
|
AIModelConfig.task == "general",
|
|
).first()
|
|
if titan:
|
|
db.delete(titan)
|
|
db.commit()
|
|
|
|
# Always ensure OpenAI TTS voice models exist (idempotent)
|
|
tts_voices = [
|
|
# OpenAI (work with OPENAI_API_KEY)
|
|
("OpenAI Alloy", "tts-1:alloy", True),
|
|
("OpenAI Nova", "tts-1:nova", False),
|
|
("OpenAI Echo", "tts-1:echo", False),
|
|
("OpenAI Shimmer", "tts-1:shimmer", False),
|
|
("OpenAI Onyx", "tts-1:onyx", False),
|
|
("OpenAI Fable", "tts-1:fable", False),
|
|
("OpenAI Alloy HD", "tts-1-hd:alloy", False),
|
|
("OpenAI Nova HD", "tts-1-hd:nova", False),
|
|
# ElevenLabs (work with ELEVENLABS_API_KEY)
|
|
("ElevenLabs Adam", "elevenlabs/adam", False),
|
|
# Google Cloud TTS (work with GOOGLE_TTS_API_KEY)
|
|
("Google Wavenet F (en-US)", "google/en-US-Wavenet-F", False),
|
|
("Google Wavenet D (en-US)", "google/en-US-Wavenet-D", False),
|
|
("Google Studio O (en-US)", "google/en-US-Studio-O", False),
|
|
("Google Studio Q (en-US)", "google/en-US-Studio-Q", False),
|
|
("Google Chirp 3 HD (en-US)", "google/en-US-Chirp3-HD-Aoede", False),
|
|
# AWS Polly Neural (work with AWS_ACCESS_KEY_ID + AWS_SECRET_ACCESS_KEY)
|
|
("AWS Polly Joanna (en-US)", "polly/Joanna", False),
|
|
("AWS Polly Matthew (en-US)", "polly/Matthew", False),
|
|
("AWS Polly Amy (en-GB)", "polly/Amy", False),
|
|
("AWS Polly Brian (en-GB)", "polly/Brian", False),
|
|
]
|
|
# Deactivate old generic tts-1 / tts-1-hd entries (no voice encoded)
|
|
for old_id in ("tts-1", "tts-1-hd"):
|
|
old = db.query(AIModelConfig).filter(AIModelConfig.model_id == old_id).first()
|
|
if old:
|
|
old.is_active = False
|
|
old.is_default = False
|
|
|
|
has_default_tts = db.query(AIModelConfig).filter(
|
|
AIModelConfig.task == "tts", AIModelConfig.is_default == True, AIModelConfig.is_active == True,
|
|
).first() is not None
|
|
|
|
for name, model_id, _ in tts_voices:
|
|
exists = db.query(AIModelConfig).filter(AIModelConfig.model_id == model_id).first()
|
|
if not exists:
|
|
is_def = not has_default_tts
|
|
db.add(AIModelConfig(name=name, model_id=model_id, task="tts", is_active=True, is_default=is_def))
|
|
if is_def:
|
|
has_default_tts = True
|
|
db.commit()
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def setup_pgvector():
|
|
"""Enable pgvector, add new columns/tables, run schema migrations."""
|
|
from sqlalchemy import text
|
|
# Import new models so create_all picks them up
|
|
from app.models import quiz_category, quiz_question_link, question_category, favorite # noqa
|
|
from app.models import flashcard, course # noqa
|
|
|
|
# Kill stale idle-in-transaction connections from previous killed startups.
|
|
# They hold DDL locks and cause ALTER TABLE below to hang indefinitely.
|
|
with engine.connect() as conn:
|
|
conn.execute(text("""
|
|
SELECT pg_terminate_backend(pid)
|
|
FROM pg_stat_activity
|
|
WHERE datname = current_database()
|
|
AND state = 'idle in transaction'
|
|
AND query_start < NOW() - INTERVAL '30 seconds'
|
|
AND pid != pg_backend_pid()
|
|
"""))
|
|
conn.commit()
|
|
|
|
# Retry schema migration up to 3 times — Celery tasks may hold locks briefly
|
|
for attempt in range(3):
|
|
try:
|
|
with engine.connect() as conn:
|
|
conn.execute(text("SET lock_timeout = '15s'"))
|
|
conn.execute(text("CREATE EXTENSION IF NOT EXISTS vector"))
|
|
conn.execute(text("ALTER TABLE questions ADD COLUMN IF NOT EXISTS embedding vector(1024)"))
|
|
conn.commit()
|
|
break
|
|
except Exception as e:
|
|
if attempt < 2 and "lock" in str(e).lower():
|
|
import logging; logging.getLogger(__name__).warning(f"Schema migration lock timeout (attempt {attempt+1}/3), retrying in 5s...")
|
|
import time; time.sleep(5)
|
|
continue
|
|
raise
|
|
|
|
with engine.connect() as conn:
|
|
conn.execute(text("SET lock_timeout = '15s'"))
|
|
conn.execute(text("""
|
|
CREATE INDEX IF NOT EXISTS questions_embedding_hnsw
|
|
ON questions USING hnsw (embedding vector_cosine_ops)
|
|
"""))
|
|
# Quiz categories
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS quiz_categories (
|
|
id SERIAL PRIMARY KEY,
|
|
name VARCHAR NOT NULL,
|
|
user_id INTEGER REFERENCES users(id),
|
|
created_at TIMESTAMP DEFAULT NOW()
|
|
)
|
|
"""))
|
|
conn.execute(text("""
|
|
ALTER TABLE quizzes
|
|
ADD COLUMN IF NOT EXISTS category_id INTEGER REFERENCES quiz_categories(id) ON DELETE SET NULL
|
|
"""))
|
|
# Question categories (separate from quiz categories)
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS question_categories (
|
|
id SERIAL PRIMARY KEY,
|
|
name VARCHAR NOT NULL,
|
|
description TEXT,
|
|
user_id INTEGER REFERENCES users(id),
|
|
created_at TIMESTAMP DEFAULT NOW()
|
|
)
|
|
"""))
|
|
conn.execute(text("""
|
|
ALTER TABLE questions
|
|
ADD COLUMN IF NOT EXISTS question_category_id INTEGER
|
|
REFERENCES question_categories(id) ON DELETE SET NULL
|
|
"""))
|
|
# Course quiz flag
|
|
conn.execute(text("""
|
|
ALTER TABLE quizzes
|
|
ADD COLUMN IF NOT EXISTS course_id INTEGER REFERENCES courses(id) ON DELETE CASCADE
|
|
"""))
|
|
|
|
conn.execute(text("""
|
|
ALTER TABLE quizzes
|
|
ADD COLUMN IF NOT EXISTS max_attempts INTEGER
|
|
"""))
|
|
conn.execute(text("""
|
|
ALTER TABLE quizzes
|
|
ADD COLUMN IF NOT EXISTS questions_per_attempt INTEGER
|
|
"""))
|
|
conn.execute(text("""
|
|
ALTER TABLE quiz_attempts
|
|
ADD COLUMN IF NOT EXISTS selected_question_ids JSON
|
|
"""))
|
|
|
|
# Question ownership
|
|
conn.execute(text("""
|
|
ALTER TABLE questions
|
|
ADD COLUMN IF NOT EXISTS user_id INTEGER REFERENCES users(id) ON DELETE SET NULL
|
|
"""))
|
|
conn.execute(text("""
|
|
ALTER TABLE questions
|
|
ADD COLUMN IF NOT EXISTS is_shared INTEGER DEFAULT 1
|
|
"""))
|
|
|
|
# Fix email collation to avoid en_US.utf8 B-tree index corruption
|
|
conn.execute(text('ALTER TABLE users ALTER COLUMN email TYPE varchar COLLATE "C"'))
|
|
|
|
# Fix: section deletion must not cascade-delete quizzes
|
|
conn.execute(text("ALTER TABLE quizzes ALTER COLUMN section_id DROP NOT NULL"))
|
|
try:
|
|
conn.execute(text("ALTER TABLE quizzes DROP CONSTRAINT IF EXISTS quizzes_section_id_fkey"))
|
|
conn.execute(text("ALTER TABLE quizzes ADD CONSTRAINT quizzes_section_id_fkey FOREIGN KEY (section_id) REFERENCES sections(id) ON DELETE SET NULL"))
|
|
except Exception as e:
|
|
import logging
|
|
logging.getLogger(__name__).debug(f"FK constraint migration (may already exist): {e}")
|
|
|
|
# Soft delete + publish control for quizzes
|
|
conn.execute(text("ALTER TABLE quizzes ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMP"))
|
|
conn.execute(text("ALTER TABLE quizzes ADD COLUMN IF NOT EXISTS is_published INTEGER DEFAULT 1"))
|
|
|
|
# Junction table: quiz ↔ question many-to-many
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS quiz_question_links (
|
|
quiz_id INTEGER NOT NULL REFERENCES quizzes(id) ON DELETE CASCADE,
|
|
question_id INTEGER NOT NULL REFERENCES questions(id) ON DELETE CASCADE,
|
|
position INTEGER DEFAULT 0,
|
|
PRIMARY KEY (quiz_id, question_id)
|
|
)
|
|
"""))
|
|
# Populate junction from existing questions (idempotent via ON CONFLICT DO NOTHING)
|
|
conn.execute(text("""
|
|
INSERT INTO quiz_question_links (quiz_id, question_id, position)
|
|
SELECT quiz_id, id,
|
|
ROW_NUMBER() OVER (PARTITION BY quiz_id ORDER BY id) - 1
|
|
FROM questions
|
|
WHERE quiz_id IS NOT NULL
|
|
ON CONFLICT DO NOTHING
|
|
"""))
|
|
# Make quiz_id on questions nullable (it becomes informational "source_quiz_id")
|
|
conn.execute(text("ALTER TABLE questions ALTER COLUMN quiz_id DROP NOT NULL"))
|
|
|
|
# Favorites table
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS favorites (
|
|
id SERIAL PRIMARY KEY,
|
|
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
question_id INTEGER NOT NULL REFERENCES questions(id) ON DELETE CASCADE,
|
|
created_at TIMESTAMP DEFAULT NOW(),
|
|
UNIQUE(user_id, question_id)
|
|
)
|
|
"""))
|
|
# Unthrottle flag for users (exempt from AI/TTS rate limits)
|
|
conn.execute(text("ALTER TABLE users ADD COLUMN IF NOT EXISTS is_unthrottled INTEGER DEFAULT 0"))
|
|
# Contact form submissions table
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS contact_submissions (
|
|
id SERIAL PRIMARY KEY,
|
|
name VARCHAR(120) NOT NULL,
|
|
email VARCHAR NOT NULL,
|
|
type VARCHAR NOT NULL,
|
|
message TEXT NOT NULL,
|
|
read INTEGER DEFAULT 0,
|
|
created_at TIMESTAMP DEFAULT NOW()
|
|
)
|
|
"""))
|
|
# AI question tags (subjects, diseases, keywords)
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS question_tags (
|
|
id SERIAL PRIMARY KEY,
|
|
name VARCHAR(200) NOT NULL,
|
|
type VARCHAR(50) NOT NULL,
|
|
created_at TIMESTAMP DEFAULT NOW()
|
|
)
|
|
"""))
|
|
conn.execute(text("""
|
|
CREATE UNIQUE INDEX IF NOT EXISTS uq_tag_name_type ON question_tags(LOWER(name), type)
|
|
"""))
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS question_tag_links (
|
|
question_id INTEGER REFERENCES questions(id) ON DELETE CASCADE,
|
|
tag_id INTEGER REFERENCES question_tags(id) ON DELETE CASCADE,
|
|
PRIMARY KEY (question_id, tag_id)
|
|
)
|
|
"""))
|
|
# Flashcard decks and cards
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS flashcard_decks (
|
|
id SERIAL PRIMARY KEY,
|
|
title VARCHAR NOT NULL,
|
|
section_id INTEGER REFERENCES sections(id) ON DELETE SET NULL,
|
|
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
card_count INTEGER DEFAULT 0,
|
|
created_at TIMESTAMP DEFAULT NOW()
|
|
)
|
|
"""))
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS flashcards (
|
|
id SERIAL PRIMARY KEY,
|
|
deck_id INTEGER NOT NULL REFERENCES flashcard_decks(id) ON DELETE CASCADE,
|
|
front TEXT NOT NULL,
|
|
back TEXT NOT NULL,
|
|
page_reference INTEGER,
|
|
image_path VARCHAR,
|
|
created_at TIMESTAMP DEFAULT NOW()
|
|
)
|
|
"""))
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS flashcard_tag_links (
|
|
flashcard_id INTEGER NOT NULL REFERENCES flashcards(id) ON DELETE CASCADE,
|
|
tag_id INTEGER NOT NULL REFERENCES question_tags(id) ON DELETE CASCADE,
|
|
PRIMARY KEY (flashcard_id, tag_id)
|
|
)
|
|
"""))
|
|
# Soft-delete + sharing for flashcard decks
|
|
conn.execute(text("ALTER TABLE flashcard_decks ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMP"))
|
|
conn.execute(text("ALTER TABLE flashcard_decks ADD COLUMN IF NOT EXISTS is_shared INTEGER DEFAULT 0"))
|
|
# Deck ratings
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS flashcard_deck_ratings (
|
|
id SERIAL PRIMARY KEY,
|
|
deck_id INTEGER NOT NULL REFERENCES flashcard_decks(id) ON DELETE CASCADE,
|
|
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
rating INTEGER NOT NULL CHECK (rating >= 1 AND rating <= 5),
|
|
created_at TIMESTAMP DEFAULT NOW(),
|
|
UNIQUE (user_id, deck_id)
|
|
)
|
|
"""))
|
|
# Quiz sharing
|
|
conn.execute(text("ALTER TABLE quizzes ADD COLUMN IF NOT EXISTS is_shared INTEGER DEFAULT 0"))
|
|
conn.execute(text("ALTER TABLE quizzes ADD COLUMN IF NOT EXISTS allow_review INTEGER DEFAULT 1"))
|
|
# ── Course / LMS tables ──────────────────────────────────────
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS courses (
|
|
id SERIAL PRIMARY KEY,
|
|
title VARCHAR NOT NULL,
|
|
description TEXT,
|
|
thumbnail_path VARCHAR,
|
|
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
status VARCHAR DEFAULT 'draft',
|
|
is_featured INTEGER DEFAULT 0,
|
|
requires_subscription INTEGER DEFAULT 0,
|
|
created_at TIMESTAMP DEFAULT NOW(),
|
|
updated_at TIMESTAMP DEFAULT NOW()
|
|
)
|
|
"""))
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS course_modules (
|
|
id SERIAL PRIMARY KEY,
|
|
course_id INTEGER NOT NULL REFERENCES courses(id) ON DELETE CASCADE,
|
|
title VARCHAR NOT NULL,
|
|
description TEXT,
|
|
position INTEGER DEFAULT 0
|
|
)
|
|
"""))
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS course_lessons (
|
|
id SERIAL PRIMARY KEY,
|
|
module_id INTEGER NOT NULL REFERENCES course_modules(id) ON DELETE CASCADE,
|
|
title VARCHAR NOT NULL,
|
|
description TEXT,
|
|
position INTEGER DEFAULT 0,
|
|
lesson_type VARCHAR NOT NULL,
|
|
content_text TEXT,
|
|
video_url VARCHAR,
|
|
video_provider VARCHAR,
|
|
local_file_path VARCHAR,
|
|
quiz_id INTEGER REFERENCES quizzes(id) ON DELETE SET NULL,
|
|
live_session_url VARCHAR,
|
|
live_session_start TIMESTAMP,
|
|
live_session_end TIMESTAMP,
|
|
bbb_meeting_id VARCHAR,
|
|
duration_minutes INTEGER,
|
|
is_required INTEGER DEFAULT 1
|
|
)
|
|
"""))
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS course_enrollments (
|
|
id SERIAL PRIMARY KEY,
|
|
course_id INTEGER NOT NULL REFERENCES courses(id) ON DELETE CASCADE,
|
|
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
|
|
enrolled_at TIMESTAMP DEFAULT NOW(),
|
|
completed_at TIMESTAMP,
|
|
progress_pct FLOAT DEFAULT 0.0,
|
|
status VARCHAR DEFAULT 'enrolled',
|
|
UNIQUE (course_id, user_id)
|
|
)
|
|
"""))
|
|
conn.execute(text("""
|
|
CREATE TABLE IF NOT EXISTS course_lesson_progress (
|
|
id SERIAL PRIMARY KEY,
|
|
enrollment_id INTEGER NOT NULL REFERENCES course_enrollments(id) ON DELETE CASCADE,
|
|
lesson_id INTEGER NOT NULL REFERENCES course_lessons(id) ON DELETE CASCADE,
|
|
status VARCHAR DEFAULT 'not_started',
|
|
started_at TIMESTAMP,
|
|
completed_at TIMESTAMP,
|
|
score FLOAT,
|
|
time_spent_seconds INTEGER DEFAULT 0,
|
|
UNIQUE (enrollment_id, lesson_id)
|
|
)
|
|
"""))
|
|
conn.commit()
|
|
|
|
|
|
def backfill_embeddings():
|
|
"""Generate embeddings for questions that don't have one yet (background, best-effort)."""
|
|
import threading
|
|
from app.models.question import Question
|
|
from app.services import embedding_service
|
|
|
|
def _run():
|
|
db = SessionLocal()
|
|
try:
|
|
missing = db.query(Question).filter(Question.embedding.is_(None)).all()
|
|
if not missing:
|
|
return
|
|
import logging
|
|
log = logging.getLogger(__name__)
|
|
log.info(f"Backfilling embeddings for {len(missing)} questions...")
|
|
ok = 0
|
|
for q in missing:
|
|
try:
|
|
if embedding_service.embed_question(q):
|
|
ok += 1
|
|
except Exception as e:
|
|
log.warning(f"Embedding failed for question {q.id}: {e}")
|
|
db.commit()
|
|
log.info(f"Backfill complete: {ok}/{len(missing)} embedded")
|
|
except Exception as e:
|
|
import logging
|
|
logging.getLogger(__name__).warning(f"Embedding backfill failed: {e}")
|
|
finally:
|
|
db.close()
|
|
|
|
threading.Thread(target=_run, daemon=True).start()
|
|
|
|
|
|
def _acquire_singleton_lock() -> bool:
|
|
"""Returns True if this worker wins the startup lock for singleton tasks (scheduler, backfill).
|
|
Uses Redis SETNX with a 5-minute TTL. Degrades gracefully if Redis is unavailable."""
|
|
try:
|
|
import redis as redis_lib
|
|
r = redis_lib.from_url(settings.REDIS_URL, decode_responses=True, socket_connect_timeout=2)
|
|
return bool(r.set("startup:singleton_lock", "1", nx=True, ex=300))
|
|
except Exception:
|
|
return True # Redis unavailable — assume single worker, run everything
|
|
|
|
|
|
# Stable int64 for pg_advisory_lock — arbitrary but must not collide with other uses.
|
|
_STARTUP_DDL_LOCK_KEY = 8472931
|
|
|
|
|
|
def _run_startup_ddl():
|
|
"""Serialize startup DDL across uvicorn workers using a Postgres advisory lock.
|
|
|
|
Without this, N workers run ALTER TABLE / CREATE TABLE in parallel at startup
|
|
and occasionally acquire AccessExclusiveLocks in different orders, tripping
|
|
Postgres's deadlock detector and killing one worker. The DDL itself is
|
|
idempotent (IF NOT EXISTS / create_all), so the losing worker runs it again
|
|
as a no-op after the winner releases the lock.
|
|
"""
|
|
from sqlalchemy import text
|
|
import logging
|
|
log = logging.getLogger(__name__)
|
|
|
|
# The advisory lock is session-scoped, held for the lifetime of this connection.
|
|
with engine.connect() as lock_conn:
|
|
log.info("Acquiring startup DDL advisory lock...")
|
|
lock_conn.execute(text("SELECT pg_advisory_lock(:k)"), {"k": _STARTUP_DDL_LOCK_KEY})
|
|
lock_conn.commit()
|
|
try:
|
|
Base.metadata.create_all(bind=engine)
|
|
setup_pgvector()
|
|
finally:
|
|
lock_conn.execute(text("SELECT pg_advisory_unlock(:k)"), {"k": _STARTUP_DDL_LOCK_KEY})
|
|
lock_conn.commit()
|
|
log.info("Startup DDL complete.")
|
|
|
|
|
|
@asynccontextmanager
|
|
async def lifespan(app: FastAPI):
|
|
# Startup — all workers gate DDL behind a Postgres advisory lock to avoid deadlocks.
|
|
_run_startup_ddl()
|
|
os.makedirs(settings.UPLOAD_DIR, exist_ok=True)
|
|
os.makedirs(os.path.join(settings.UPLOAD_DIR, "images"), exist_ok=True)
|
|
os.makedirs(settings.CHROMA_PERSIST_DIR, exist_ok=True)
|
|
seed_admin()
|
|
seed_default_models()
|
|
# Scheduler and backfill must only run in one worker to avoid duplicate jobs / race conditions
|
|
is_primary = _acquire_singleton_lock()
|
|
if is_primary:
|
|
backfill_embeddings()
|
|
start_scheduler()
|
|
yield
|
|
# Shutdown
|
|
if is_primary:
|
|
stop_scheduler()
|
|
|
|
|
|
app = FastAPI(
|
|
title="PedsHub",
|
|
description="Pediatric Knowledge Quiz Platform",
|
|
version="2.0.0",
|
|
lifespan=lifespan,
|
|
)
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["https://quiz.danvics.com", "https://pedshub.com", "https://www.pedshub.com"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
expose_headers=["X-New-Token"], # Allow frontend to read this header
|
|
)
|
|
|
|
# Session middleware for OIDC state (authlib needs it)
|
|
from starlette.middleware.sessions import SessionMiddleware
|
|
app.add_middleware(SessionMiddleware, secret_key=settings.SECRET_KEY)
|
|
|
|
# Add token refresh middleware AFTER CORS (middleware applies in reverse)
|
|
from app.utils.auth import TokenRefreshMiddleware
|
|
app.add_middleware(TokenRefreshMiddleware)
|
|
|
|
# Request logging middleware — logs every request with user, duration, status
|
|
from app.middleware.request_logging import RequestLoggingMiddleware
|
|
app.add_middleware(RequestLoggingMiddleware)
|
|
|
|
# Serve uploaded images as static files
|
|
app.mount("/uploads", StaticFiles(directory=settings.UPLOAD_DIR), name="uploads")
|
|
|
|
app.include_router(auth.router, prefix="/api/auth", tags=["auth"])
|
|
app.include_router(documents.router, prefix="/api/documents", tags=["documents"])
|
|
app.include_router(quizzes.router, prefix="/api/quizzes", tags=["quizzes"])
|
|
app.include_router(attempts.router, prefix="/api/attempts", tags=["attempts"])
|
|
app.include_router(admin.router, prefix="/api/admin", tags=["admin"])
|
|
app.include_router(tts.router, prefix="/api/tts", tags=["tts"])
|
|
app.include_router(nextcloud.router, prefix="/api/nextcloud", tags=["nextcloud"])
|
|
app.include_router(categories.router, prefix="/api/categories", tags=["categories"])
|
|
app.include_router(questions.router, prefix="/api/questions", tags=["questions"])
|
|
app.include_router(question_categories.router, prefix="/api/question-categories", tags=["question-categories"])
|
|
app.include_router(favorites.router, prefix="/api/favorites", tags=["favorites"])
|
|
app.include_router(teach.router, prefix="/api/teach", tags=["teach"])
|
|
app.include_router(contact.router, prefix="/api/contact", tags=["contact"])
|
|
app.include_router(tags.router, prefix="/api/tags", tags=["tags"])
|
|
app.include_router(flashcards.router, prefix="/api/flashcards", tags=["flashcards"])
|
|
app.include_router(courses.router, prefix="/api/courses", tags=["courses"])
|
|
|
|
|
|
@app.get("/api/health")
|
|
def health_check():
|
|
return {"status": "ok"}
|