feat: implement search functionality, add background task queue logic for title conflicts, and enhance auth state synchronization

auth/routes.py

@@ -1,7 +1,6 @@
 import uuid
 from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel
-from typing import Optional
 
 from src.database import get_auth_supabase
 from src.store import create_user, get_user_by_email
@@ -20,12 +19,6 @@ class SignupRequest(BaseModel):
     password: str
 
 
-class GoogleAuthRequest(BaseModel):
-    token: str
-    name: Optional[str] = None
-    email: Optional[str] = None
-
-
 @router.post("/login")
 async def login(body: LoginRequest):
     supabase = get_auth_supabase()
@@ -101,18 +94,3 @@ async def signup(body: SignupRequest):
         "token": str(uuid.uuid4()),
         "user": {"id": user["id"], "name": user["name"], "email": user["email"]},
     }
-
-
-@router.post("/google")
-async def google_auth(body: GoogleAuthRequest):
-    # Google OAuth would require supabase.auth.sign_in_with_id_token — not wired up yet
-    # Fall back to dev mode: create/fetch user by email
-    email = body.email or f"google_{uuid.uuid4().hex[:8]}@google.com"
-    name = body.name or "Google User"
-    user = get_user_by_email(email)
-    if not user:
-        user = create_user(name, email, "")
-    return {
-        "token": str(uuid.uuid4()),
-        "user": {"id": user["id"], "name": user["name"], "email": user["email"]},
-    }

main.py

@@ -48,6 +48,10 @@ async def lifespan(app: FastAPI):
         logger.info("Embedder loaded successfully.")
     except Exception as e:
         logger.warning(f"Embedder failed to load: {e}")
+
+    from src.rag.batch_workers import start_workers
+    start_workers()
+
     yield
 
 

materials/routes.py

@@ -2,13 +2,15 @@ import time
 import asyncio
 import logging
 import validators
-from fastapi import APIRouter, UploadFile, File, HTTPException, Depends, BackgroundTasks
+from fastapi import APIRouter, UploadFile, File, HTTPException, Depends, BackgroundTasks, Header
 from pydantic import BaseModel
+from postgrest.exceptions import APIError
 
 from src.materials.text_utils import text_from_pdf, chunk_text, scrap_website
-from src.rag.rag import store_embeddings
-from src.store import create_material, get_material, update_material_status, save_chunks, list_materials, delete_material, rename_material
+from src.rag.rag import store_embeddings, store_embeddings_async
+from src.store import create_material, get_material, update_material_status, save_chunks, list_materials, delete_material, rename_material, is_title_taken
 from src.dependencies import get_current_user_id, get_current_user
+from src.database import get_supabase, get_auth_supabase
 
 logger = logging.getLogger(__name__)
 
@@ -57,29 +59,43 @@ async def get_material_by_id(
         raise HTTPException(404, "Material not found")
     return mat
 
-def _process_pdf_background(material_id: str, file_content: bytes):
+async def _process_pdf_background(material_id: str, file_content: bytes):
     try:
+        # 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")
+            return
+
+        loop = asyncio.get_event_loop()
         from io import BytesIO
-        raw = text_from_pdf(BytesIO(file_content))
-        chunks = chunk_text(raw)
-        chunk_ids = save_chunks(material_id, chunks)
-        
+        raw = await loop.run_in_executor(None, text_from_pdf, BytesIO(file_content))
+        chunks = await loop.run_in_executor(None, chunk_text, raw)
+        chunk_ids = await loop.run_in_executor(None, save_chunks, material_id, chunks)
+
         update_material_status(material_id, "processing")
-        store_embeddings(material_id, chunk_ids, chunks)
+        await store_embeddings_async(material_id, chunk_ids, chunks)
         update_material_status(material_id, "ready")
         logger.info(f"Background processing complete for material {material_id}")
     except Exception as e:
         logger.error(f"Background processing failed for material {material_id}: {e}", exc_info=True)
         update_material_status(material_id, "failed", str(e))
 
-def _process_url_background(material_id: str, url: str):
+async def _process_url_background(material_id: str, url: str):
     try:
-        raw = scrap_website(url)
-        chunks = chunk_text(raw, chunk_size=600, chunk_overlap=100)
-        chunk_ids = save_chunks(material_id, chunks)
-        
+        # Skip 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 URL processing for {material_id}: duplicate title, waiting for rename")
+            return
+
+        loop = asyncio.get_event_loop()
+        raw = await loop.run_in_executor(None, scrap_website, url)
+        chunks = await loop.run_in_executor(None, lambda: chunk_text(raw, chunk_size=600, chunk_overlap=100))
+        chunk_ids = await loop.run_in_executor(None, save_chunks, material_id, chunks)
+
         update_material_status(material_id, "processing")
-        store_embeddings(material_id, chunk_ids, chunks)
+        await store_embeddings_async(material_id, chunk_ids, chunks)
         update_material_status(material_id, "ready")
         logger.info(f"Background processing complete for URL material {material_id}")
     except Exception as e:
@@ -134,7 +150,9 @@ async def scrape_url(
         )
         material_id = material["id"]
 
-        background_tasks.add_task(_process_url_background, material_id, input.url)
+        # Skip processing if title conflicts — user must rename first
+        if not is_title_taken(input.url, exclude_id=material_id, user_id=user_id):
+            background_tasks.add_task(_process_url_background, material_id, input.url)
 
         return {
             "status": "processing_started",
@@ -184,7 +202,20 @@ async def rename_material_endpoint(
         raise HTTPException(404, "Material not found")
     if mat.get("user_id") != user_id:
         raise HTTPException(403, "Not authorized to rename this material")
-    await loop.run_in_executor(None, lambda: rename_material(material_id, body.title))
+    new_title = body.title.strip()
+    if not new_title:
+        raise HTTPException(400, "Title cannot be empty")
+    existing = await loop.run_in_executor(None, lambda: list_materials(user_id))
+    if any(m.get("id") != material_id and m.get("title", "").strip().lower() == new_title.lower() for m in existing):
+        raise HTTPException(409, "A material with this title already exists")
+    await loop.run_in_executor(None, lambda: rename_material(material_id, new_title))
+
+    # If the material was pending due to title conflict, try processing now
+    if mat.get("source_type") == "url" and mat.get("status") == "pending":
+        url = mat.get("url")
+        if url and not is_title_taken(new_title, exclude_id=material_id, user_id=user_id):
+            asyncio.ensure_future(_process_url_background(material_id, url))
+
     return {"status": "ok"}
 
 class TopicRequest(BaseModel):
@@ -195,6 +226,8 @@ async def create_topic(
     body: TopicRequest,
     user_id: str = Depends(get_current_user_id)
 ):
+    if is_title_taken(body.topic, user_id=user_id):
+        raise HTTPException(409, "A material with this title already exists")
     mat = create_material(
         user_id=user_id,
         title=body.topic.strip(),
@@ -203,6 +236,54 @@ async def create_topic(
     update_material_status(mat["id"], "ready", "Topic ready")
     return {"material_id": mat["id"], "title": mat["title"]}
 
+
+class SearchQuery(BaseModel):
+    q: str
+
+@router.post("/search")
+async def search_materials(
+    body: SearchQuery,
+    authorization: str = Header(None),
+):
+    if not body.q.strip():
+        return {"results": []}
+
+    anon_client = get_auth_supabase()
+    if anon_client is None:
+        return {"results": []}
+
+    if not authorization:
+        raise HTTPException(401, "Missing authorization token")
+
+    token = authorization.strip()
+    if token.lower().startswith("bearer "):
+        token = token[7:].strip()
+    if not token:
+        raise HTTPException(401, "Missing authorization token")
+
+    anon_client.postgrest.auth(token)
+
+    try:
+        result = anon_client.rpc(
+            "search_materials_by_title",
+            {"p_query": body.q.strip()},
+        ).execute()
+    except APIError as e:
+        payload = e.args[0] if e.args else None
+        message = None
+        code = None
+        status = getattr(e, "status_code", None)
+        if isinstance(payload, dict):
+            message = payload.get("message")
+            code = payload.get("code")
+            status = payload.get("status") or status
+        message_text = (message or str(e) or "").lower()
+        if status == 401 or code == "PGRST303" or "jwt expired" in message_text or "unauthorized" in message_text:
+            raise HTTPException(401, "Session expired. Please sign in again.")
+        raise HTTPException(500, f"Search failed: {message or 'Unknown error'}")
+
+    return {"results": [row["material_id"] for row in (result.data or [])]}
+    
 @router.delete("/{material_id}")
 async def delete_material_endpoint(
     material_id: str,

rag/batch_workers.py

@@ -0,0 +1,202 @@
+"""
+Embedding Batch Workers — Async batching infrastructure for embedding inference.
+
+Architecture:
+  - Single asyncio.Queue for all embedding jobs.
+  - Dedicated async worker coroutines drain the queue in micro-batches.
+  - Workers offload heavy inference to a thread via run_in_executor.
+  - A shared in-memory job_store dict tracks job status + results.
+  - Warmup loop periodically does a dummy forward pass to keep OpenMP threads alive.
+"""
+
+import asyncio
+import time
+import uuid
+import logging
+from dataclasses import dataclass, field
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+
+# ═══════════════════════ Job Store ════════════════════════
+
+job_store: dict[str, dict[str, Any]] = {}
+"""
+{
+    "<job_id>": {
+        "status": "pending" | "processing" | "done" | "error",
+        "result": <list[list[float]] for EmbeddingJob> | None,
+        "error": <str> | None,
+    }
+}
+"""
+
+
+def create_job() -> str:
+    """Create a new pending job and return its ID."""
+    job_id = str(uuid.uuid4())
+    job_store[job_id] = {"status": "pending", "result": None, "error": None}
+    return job_id
+
+
+# ═══════════════════════ Request-in-Flight Gate ════════════════════════
+
+_request_in_flight_count = 0
+
+
+def set_request_in_flight(active: bool):
+    """Increment/decrement in-flight counter. Thread-safe enough for a gate."""
+    global _request_in_flight_count
+    if active:
+        _request_in_flight_count += 1
+    else:
+        _request_in_flight_count = max(0, _request_in_flight_count - 1)
+
+
+def is_request_in_flight() -> bool:
+    return _request_in_flight_count > 0
+
+
+# ═══════════════════════ Job Dataclasses ════════════════════════
+
+@dataclass
+class EmbeddingJob:
+    """Batch embedding of multiple texts (for store_embeddings)."""
+    job_id: str
+    texts: list[str]
+    done: asyncio.Event = field(default_factory=asyncio.Event)
+
+
+# ═══════════════════════ Queue ════════════════════════
+
+embedding_queue: asyncio.Queue[EmbeddingJob] = asyncio.Queue()
+
+
+# ═══════════════════════ Workers ════════════════════════
+
+_BATCH_MAX_SIZE = 8
+_BATCH_WINDOW_S = 0.05
+
+
+async def embedding_worker():
+    """
+    Drains up to {_BATCH_MAX_SIZE} embedding jobs every {_BATCH_WINDOW_S * 1000:.0f}ms.
+
+    One SentenceTransformer forward pass per batch:
+      1. Collect texts from all jobs in the batch
+      2. get_embedder().embed_documents(all_texts) → raw [B, D] embeddings
+      3. Distribute results back to individual jobs
+
+    Results are written into job_store and each job's done Event is set.
+    """
+    from src.rag.rag import get_embedder
+
+    loop = asyncio.get_event_loop()
+
+    while True:
+        # Wait for at least one job
+        first_job: EmbeddingJob = await embedding_queue.get()
+        batch: list[EmbeddingJob] = [first_job]
+
+        # Collect up to 7 more within the time window
+        deadline = loop.time() + _BATCH_WINDOW_S
+        while len(batch) < _BATCH_MAX_SIZE:
+            remaining = deadline - loop.time()
+            if remaining <= 0:
+                break
+            try:
+                job = await asyncio.wait_for(embedding_queue.get(), timeout=remaining)
+                batch.append(job)
+            except asyncio.TimeoutError:
+                break
+
+        try:
+            set_request_in_flight(True)
+
+            # Gather all texts from all jobs in the batch
+            all_texts: list[str] = []
+            text_counts: list[int] = []
+            for job in batch:
+                all_texts.extend(job.texts)
+                text_counts.append(len(job.texts))
+
+            # Single forward pass for the entire batch
+            embedder = get_embedder()
+            all_embeddings = await loop.run_in_executor(
+                None, embedder.embed_documents, all_texts
+            )
+
+            # Distribute results back to individual jobs
+            idx = 0
+            for i, job in enumerate(batch):
+                n = text_counts[i]
+                job_result = all_embeddings[idx: idx + n]
+                idx += n
+
+                job_store[job.job_id]["status"] = "done"
+                job_store[job.job_id]["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()
+        finally:
+            set_request_in_flight(False)
+
+
+# ═══════════════════════ Warmup Loop ════════════════════════
+
+_WARMUP_INTERVAL_S = 45
+
+
+async def _warmup_loop():
+    """
+    Periodically does a dummy forward pass to prevent OpenMP/MKL thread pool
+    spin-down during idle periods.
+
+    Skipped entirely if a real request is in flight.
+    """
+    from src.rag.rag import warmup_embedder
+
+    loop = asyncio.get_event_loop()
+
+    while True:
+        await asyncio.sleep(_WARMUP_INTERVAL_S)
+        if is_request_in_flight():
+            continue
+        t0 = time.monotonic()
+        try:
+            await loop.run_in_executor(None, warmup_embedder)
+        except Exception as e:
+            logger.warning(f"Warmup cycle error (non-fatal): {e}")
+            continue
+        elapsed_ms = (time.monotonic() - t0) * 1000
+        logger.info(f"Warmup cycle done ({elapsed_ms:.0f}ms)")
+
+
+# ═══════════════════════ Startup ════════════════════════
+
+_workers_started = False
+
+
+def start_workers():
+    """
+    Launch all async worker coroutines. Call once during app startup.
+
+    - 2 embedding workers (batched SentenceTransformer inference)
+    - 1 warmup loop (keeps OpenMP threads alive)
+    """
+    global _workers_started
+    if _workers_started:
+        return
+    _workers_started = True
+
+    for i in range(2):
+        asyncio.create_task(embedding_worker(), name=f"embedding_worker_{i}")
+    asyncio.create_task(_warmup_loop(), name="warmup_loop")
+
+    logger.info("Embedding batch workers started (2 workers + warmup loop)")

rag/rag.py

@@ -1,4 +1,6 @@
 import os
+import asyncio
+import uuid
 import logging
 from functools import lru_cache
 from typing import Optional
@@ -55,6 +57,45 @@ def store_embeddings(material_id: str, chunk_ids: list[str], chunks: list[str]):
     logger.info(f"Embeddings stored successfully for material {material_id}.")
 
 
+def warmup_embedder():
+    """Dummy forward pass to keep OpenMP/MKL thread pool alive during idle periods."""
+    embedder = get_embedder()
+    embedder.embed_documents(["warmup"])
+
+
+async def store_embeddings_async(material_id: str, chunk_ids: list[str], chunks: list[str]):
+    """
+    Async variant of store_embeddings that routes embedding inference through
+    the batch worker queue for batching across concurrent requests.
+    """
+    from src.rag.batch_workers import EmbeddingJob, embedding_queue, job_store
+
+    job = EmbeddingJob(job_id=str(uuid.uuid4()), texts=chunks)
+    await embedding_queue.put(job)
+    await job.done.wait()
+
+    entry = job_store[job.job_id]
+    if entry["status"] == "error":
+        raise RuntimeError(f"Embedding failed: {entry['error']}")
+
+    embeddings = entry["result"]
+
+    records = [
+        {"chunk_id": cid, "material_id": material_id, "embedding": emb}
+        for cid, emb in zip(chunk_ids, embeddings)
+    ]
+
+    db = get_supabase()
+    if db is None:
+        logger.warning("Supabase not connected — embeddings computed but NOT stored (no DB).")
+        return
+
+    logger.info(f"Storing {len(records)} embeddings in Supabase for material {material_id}...")
+    for i in range(0, len(records), 50):
+        db.table("material_embeddings").insert(records[i:i + 50]).execute()
+    logger.info(f"Embeddings stored successfully for material {material_id}.")
+
+
 def similarity_search(query: str, material_id: str, k: int = 5) -> list[dict]:
     embedder = get_embedder()
     query_embedding = embedder.embed_query(query)

store.py

@@ -133,6 +133,32 @@ def list_materials(user_id: str) -> list[dict]:
     return list(reversed([r for r in records if r.get("user_id") == user_id]))
 
 
+def is_title_taken(title: str, exclude_id: Optional[str] = None, user_id: Optional[str] = None) -> bool:
+    normalized = title.strip().lower()
+    if not normalized:
+        return False
+    try:
+        query = _table_supabase("materials").select("id,title")
+        if user_id:
+            query = query.eq("user_id", user_id)
+        result = _robust_execute(query)
+        for row in result.data:
+            if exclude_id and row.get("id") == exclude_id:
+                continue
+            if row.get("title", "").strip().lower() == normalized:
+                return True
+    except Exception:
+        pass
+    for row in _in_memory.get("materials", {}).values():
+        if exclude_id and row.get("id") == exclude_id:
+            continue
+        if user_id and row.get("user_id") != user_id:
+            continue
+        if row.get("title", "").strip().lower() == normalized:
+            return True
+    return False
+
+
 def create_material(user_id: str, source_type: str, title: str,
                     file_path: Optional[str] = None,
                     url: Optional[str] = None,