feat: implement daily request rate limiting with per-user usage tracking and admin bypass

main.py

@@ -11,6 +11,8 @@ from src.summary_generator.routes import router as summary_router
 from src.rag.routes import router as tutor_router
 from src.quiz_generator.routes import router as quiz_router
 from src.auth.routes import router as auth_router
+from src.store import get_usage
+from src.dependencies import get_current_user_id
 from src.config import settings
 
 # ── Logging Setup ──────────────────────────────────────
@@ -85,3 +87,8 @@ async def root():
 @app.get("/api/health")
 async def health_check():
     return {"status": "ok", "service": "AI Tutor API"}
+
+
+@app.get("/api/usage")
+async def get_user_usage(user_id: str = Depends(get_current_user_id)):
+    return get_usage(user_id)

materials/routes.py

@@ -64,7 +64,8 @@ async def _process_pdf_background(material_id: str, file_content: bytes):
         # Skip processing if this user already has a material with this title
         mat = get_material(material_id)
         if mat and is_title_taken(mat.get("title", ""), exclude_id=material_id, user_id=mat.get("user_id")):
-            logger.info(f"Skipping processing for {material_id}: duplicate title, waiting for rename")
+            logger.info(f"Skipping processing for {material_id}: duplicate title")
+            update_material_status(material_id, "failed", "Duplicate title. Please rename to retry.")
             return
 
         loop = asyncio.get_event_loop()

quiz_generator/routes.py

@@ -4,8 +4,8 @@ from pydantic import BaseModel
 from typing import Optional
 
 from src.quiz_generator.quiz import smart_quiz_generator
-from src.store import get_material, get_chunks, get_summary, save_quiz, get_quizzes, save_quiz_result, get_quiz_results
-from src.dependencies import get_current_user_id
+from src.store import get_material, get_chunks, get_summary, save_quiz, get_quizzes, save_quiz_result, get_quiz_results, check_and_increment_daily_limit
+from src.dependencies import get_current_user_id, get_current_user
 from src.config import settings
 
 router = APIRouter(prefix="/api/quiz", tags=["Quiz"])
@@ -37,7 +37,13 @@ async def get_quiz_list(
 async def generate_quiz(
     body: QuizRequest,
     user_id: str = Depends(get_current_user_id),
+    current_user=Depends(get_current_user),
 ):
+    # Rate limit check
+    user_email = current_user.get("email") if isinstance(current_user, dict) else getattr(current_user, "email", None)
+    if not check_and_increment_daily_limit(user_id, email=user_email, limit=10):
+        raise HTTPException(429, "Daily limit of 10 requests reached. Come back tomorrow!")
+
     body.difficulty = body.difficulty.capitalize()
     if body.mcq_count < 1 or body.mcq_count > 20:
         raise HTTPException(400, "MCQ count must be between 1 and 20")

rag/batch_workers.py

@@ -134,23 +134,36 @@ async def embedding_worker():
                 job_result = all_embeddings[idx: idx + n]
                 idx += n
 
-                job_store[job.job_id]["status"] = "done"
-                job_store[job.job_id]["result"] = job_result
+                # Use .get() or setdefault to avoid KeyError if initialization was missed
+                if job.job_id not in job_store:
+                    job_store[job.job_id] = {"status": "pending", "result": None, "error": None}
+                
+                job_store[job.job_id].update({
+                    "status": "done",
+                    "result": job_result
+                })
                 job.done.set()
 
         except Exception as e:
             logger.error(f"Embedding batch failed: {e}", exc_info=True)
             for job in batch:
-                job_store[job.job_id]["status"] = "error"
-                job_store[job.job_id]["error"] = str(e)
-                job.done.set()
+                if job.job_id not in job_store:
+                    job_store[job.job_id] = {"status": "error", "result": None, "error": str(e)}
+                else:
+                    job_store[job.job_id].update({
+                        "status": "error",
+                        "error": str(e)
+                    })
+                # Critical: always set the event so the request doesn't hang
+                if not job.done.is_set():
+                    job.done.set()
         finally:
             set_request_in_flight(False)
 
 
 # ═══════════════════════ Warmup Loop ════════════════════════
 
-_WARMUP_INTERVAL_S = 45
+_WARMUP_INTERVAL_S = 300  # 5 minutes
 
 
 async def _warmup_loop():
@@ -187,7 +200,7 @@ def start_workers():
     """
     Launch all async worker coroutines. Call once during app startup.
 
-    - 2 embedding workers (batched SentenceTransformer inference)
+    - 1 embedding worker (batched SentenceTransformer inference)
     - 1 warmup loop (keeps OpenMP threads alive)
     """
     global _workers_started
@@ -195,8 +208,8 @@ def start_workers():
         return
     _workers_started = True
 
-    for i in range(2):
-        asyncio.create_task(embedding_worker(), name=f"embedding_worker_{i}")
+    # Use only 1 worker to save RAM on this environment
+    asyncio.create_task(embedding_worker(), name="embedding_worker_0")
     asyncio.create_task(_warmup_loop(), name="warmup_loop")
 
-    logger.info("Embedding batch workers started (2 workers + warmup loop)")
+    logger.info("Embedding batch workers started (1 worker + warmup loop)")

rag/rag.py

@@ -71,6 +71,7 @@ async def store_embeddings_async(material_id: str, chunk_ids: list[str], chunks:
     from src.rag.batch_workers import EmbeddingJob, embedding_queue, job_store
 
     job = EmbeddingJob(job_id=str(uuid.uuid4()), texts=chunks)
+    job_store[job.job_id] = {"status": "pending", "result": None, "error": None}
     await embedding_queue.put(job)
     await job.done.wait()
 

rag/routes.py

@@ -12,6 +12,7 @@ from src.store import (
     create_chat_session, list_chat_sessions, get_chat_session,
     rename_chat_session, delete_chat_session,
     append_session_message, get_session_messages,
+    check_and_increment_daily_limit,
     # Legacy
     save_chat_messages, get_chat_messages,
 )
@@ -43,6 +44,13 @@ async def ask_tutor(
     if not body.query.strip():
         raise HTTPException(400, "Query cannot be empty")
 
+    # Rate limit check
+    user_id = current_user.get("id") if isinstance(current_user, dict) else getattr(current_user, "id", None)
+    user_email = current_user.get("email") if isinstance(current_user, dict) else getattr(current_user, "email", None)
+    
+    if user_id and not check_and_increment_daily_limit(user_id, email=user_email, limit=10):
+        raise HTTPException(429, "Daily limit of 10 requests reached. Come back tomorrow!")
+
     # Determine memory key (prefer session_id for persistence)
     mem_key = body.session_id or body.memory_id
 

store.py

@@ -1,5 +1,6 @@
+import os
 from typing import Optional
-from datetime import datetime, timezone
+from datetime import datetime, timezone, date
 from langchain.memory import ConversationBufferMemory, ConversationBufferWindowMemory
 from src.database import get_supabase
 
@@ -12,6 +13,12 @@ _in_memory: dict = {
     "next_id": 0,
 }
 
+ADMIN_EMAILS = set(
+    email.strip() 
+    for email in os.environ.get("ADMIN_EMAILS", "").split(",")
+    if email.strip()
+)
+
 
 def _get_next_id() -> str:
     _in_memory["next_id"] += 1
@@ -266,15 +273,12 @@ def get_summary(material_id: str) -> Optional[dict]:
             _table_supabase("summaries")
             .select("*")
             .eq("material_id", material_id)
-            .maybe_single()
             .execute()
         )
-        if not result or result.data is None:
+        if not result or not result.data:
             return None
-        # _make_response returns list; unwrap the first item
-        if isinstance(result.data, list):
-            return result.data[0] if result.data else None
-        return result.data
+        # Return the most recent one if duplicates exist
+        return result.data[0]
     except Exception:
         return None
 
@@ -520,3 +524,89 @@ def get_or_create_memory(memory_id: Optional[str] = None, seed_messages: list[di
                 )
     _memories[mid] = mem
     return mem, mid
+
+
+def check_and_increment_daily_limit(user_id: str, email: Optional[str] = None, limit: int = 10) -> bool:
+    """
+    Returns True if request is allowed, False if limit exceeded.
+    Excludes Admin Emails from Limits
+    """
+    # Exclude specific email from rate limiting
+    if email in ADMIN_EMAILS:
+        return True
+
+    today = date.today().isoformat()
+
+    try:
+        # Get current profile
+        result = _robust_execute(
+            _table_supabase("profiles")
+            .select("daily_requests, last_request_date")
+            .eq("id", user_id)
+            .maybe_single()
+        )
+        
+        # If no profile or data, allow (or we could create one, but usually it exists)
+        if not result.data:
+            return True
+        
+        # Handle both single object or list response from maybe_single/execute
+        profile = result.data[0] if isinstance(result.data, list) and result.data else result.data
+        if not profile:
+            return True
+
+        last_date = profile.get("last_request_date")
+        count = profile.get("daily_requests", 0) or 0
+
+        # Reset count if it's a new day
+        if last_date != today:
+            count = 0
+
+        # Check limit
+        if count >= limit:
+            return False
+
+        # Increment
+        _robust_execute(
+            _table_supabase("profiles")
+            .update({
+                "daily_requests": count + 1,
+                "last_request_date": today,
+            })
+            .eq("id", user_id)
+        )
+        return True
+    except Exception as e:
+        # If DB fails, we default to allowing the request to not break the app
+        import logging
+        logging.getLogger(__name__).error(f"Rate limit check failed: {e}")
+        return True
+
+
+def get_usage(user_id: str) -> dict:
+    """
+    Returns current usage for a user.
+    """
+    today = date.today().isoformat()
+    try:
+        result = _robust_execute(
+            _table_supabase("profiles")
+            .select("daily_requests, last_request_date")
+            .eq("id", user_id)
+            .maybe_single()
+        )
+        if not result.data:
+            return {"used": 0, "limit": 10, "remaining": 10}
+
+        profile = result.data[0] if isinstance(result.data, list) and result.data else result.data
+        if not profile:
+            return {"used": 0, "limit": 10, "remaining": 10}
+
+        used = profile.get("daily_requests", 0) if profile.get("last_request_date") == today else 0
+        return {
+            "used": used,
+            "limit": 10,
+            "remaining": max(0, 10 - used)
+        }
+    except Exception:
+        return {"used": 0, "limit": 10, "remaining": 10}

summary_generator/routes.py

@@ -6,7 +6,7 @@ from pydantic import BaseModel
 from typing import Optional
 
 from src.summary_generator.summary import summarizer
-from src.store import get_material, get_chunks, save_summary, get_summary as get_stored_summary
+from src.store import get_material, get_chunks, save_summary, get_summary as get_stored_summary, check_and_increment_daily_limit
 from src.dependencies import get_current_user_id, get_current_user
 from src.config import settings
 
@@ -28,6 +28,11 @@ async def generate_summary(
     user_id: str = Depends(get_current_user_id),
     current_user=Depends(get_current_user),
 ):
+    # Rate limit check
+    user_email = current_user.get("email") if isinstance(current_user, dict) else getattr(current_user, "email", None)
+    if not check_and_increment_daily_limit(user_id, email=user_email, limit=10):
+        raise HTTPException(429, "Daily limit of 10 requests reached. Come back tomorrow!")
+
     mat = get_material(body.material_id)
     if not mat:
         raise HTTPException(404, "Material not found")