feat(backend): WebSocket endpoint for real-time indexing progress (#149)

- Add /api/v1/ws/playground/{job_id} WebSocket endpoint
- Implement Redis Pub/Sub for real-time event streaming
- Publish events: connected, cloning, progress, completed, error
- Progress events include current_file for streaming file list
- Skip duplicate PROCESSING events (handled by update_progress)
- Add comprehensive test suite (7 tests)

Closes #149

Author: DevanshuNEU

backend/main.py

@@ -27,6 +27,7 @@
 from routes.api_keys import router as api_keys_router
 from routes.users import router as users_router
 from routes.search_v2 import router as search_v2_router
+from routes.ws_playground import websocket_playground_index
 
 
 # Lifespan context manager for startup/shutdown
@@ -91,8 +92,9 @@ async def dispatch(self, request: Request, call_next):
 app.include_router(users_router, prefix=API_PREFIX)
 app.include_router(search_v2_router, prefix=API_PREFIX)
 
-# WebSocket endpoint (versioned)
+# WebSocket endpoints (versioned)
 app.add_api_websocket_route(f"{API_PREFIX}/ws/index/{{repo_id}}", websocket_index)
+app.add_api_websocket_route(f"{API_PREFIX}/ws/playground/{{job_id}}", websocket_playground_index)
 
 
 # ===== ERROR HANDLERS =====

backend/routes/ws_playground.py

@@ -0,0 +1,140 @@
+"""
+WebSocket endpoint for real-time playground indexing progress.
+
+This provides instant updates as files are indexed, giving users
+a smooth streaming experience instead of polling every 2 seconds.
+
+Channel format: job:{job_id}:events
+Message types: started, progress, completed, error
+"""
+import json
+import asyncio
+from typing import Optional
+
+from fastapi import WebSocket, WebSocketDisconnect
+
+from dependencies import redis_client
+from services.observability import logger
+
+
+# How long to wait for first message before giving up
+INITIAL_TIMEOUT_SECONDS = 30
+
+# How long between messages before assuming job is dead
+MESSAGE_TIMEOUT_SECONDS = 60
+
+
+async def websocket_playground_index(websocket: WebSocket, job_id: str):
+    """
+    Stream indexing progress to client via WebSocket.
+    
+    Subscribes to Redis pub/sub channel for this job and forwards
+    all events to the connected client. Closes when job completes
+    or fails, or if client disconnects.
+    
+    No auth required - job_id is an unguessable UUID that acts as
+    a bearer token. Only the session that created the job knows it.
+    """
+    # Validate we have Redis (required for pub/sub)
+    if not redis_client:
+        logger.error("WebSocket failed - no Redis connection")
+        await websocket.close(code=4500, reason="Service unavailable")
+        return
+    
+    # Validate job_id format (basic sanity check)
+    if not job_id or len(job_id) < 10:
+        await websocket.close(code=4400, reason="Invalid job ID")
+        return
+    
+    channel = f"job:{job_id}:events"
+    
+    # Accept the WebSocket connection
+    await websocket.accept()
+    logger.info("WebSocket connected", job_id=job_id[:12], channel=channel)
+    
+    # Set up Redis pub/sub
+    pubsub = redis_client.pubsub()
+    
+    try:
+        # Subscribe to job's event channel
+        await asyncio.to_thread(pubsub.subscribe, channel)
+        logger.debug("Subscribed to channel", channel=channel)
+        
+        # Send initial ack so client knows we're connected
+        await websocket.send_json({
+            "type": "connected",
+            "job_id": job_id,
+            "message": "Listening for indexing events"
+        })
+        
+        # Listen for messages
+        while True:
+            # Check for new message (non-blocking with timeout)
+            message = await asyncio.to_thread(
+                pubsub.get_message,
+                ignore_subscribe_messages=True,
+                timeout=MESSAGE_TIMEOUT_SECONDS
+            )
+            
+            if message is None:
+                # Timeout - check if client is still connected
+                try:
+                    await websocket.send_json({"type": "ping"})
+                except Exception:
+                    logger.debug("Client disconnected during timeout", job_id=job_id[:12])
+                    break
+                continue
+            
+            if message["type"] != "message":
+                continue
+            
+            # Parse and forward the event
+            try:
+                event_data = json.loads(message["data"])
+                await websocket.send_json(event_data)
+                
+                # Close connection after terminal events
+                event_type = event_data.get("type")
+                if event_type in ("completed", "error"):
+                    logger.info(
+                        "Job finished, closing WebSocket",
+                        job_id=job_id[:12],
+                        event_type=event_type
+                    )
+                    break
+                    
+            except json.JSONDecodeError:
+                logger.warning("Invalid JSON in pub/sub message", job_id=job_id[:12])
+                continue
+            except Exception as e:
+                logger.error("Error forwarding message", error=str(e), job_id=job_id[:12])
+                continue
+                
+    except WebSocketDisconnect:
+        logger.debug("WebSocket disconnected by client", job_id=job_id[:12])
+        
+    except Exception as e:
+        logger.error("WebSocket error", error=str(e), job_id=job_id[:12])
+        try:
+            await websocket.send_json({
+                "type": "error",
+                "message": "Internal server error"
+            })
+        except Exception:
+            pass
+            
+    finally:
+        # Clean up pub/sub subscription
+        try:
+            await asyncio.to_thread(pubsub.unsubscribe, channel)
+            await asyncio.to_thread(pubsub.close)
+        except Exception:
+            pass
+        
+        # Close WebSocket if still open
+        try:
+            await websocket.close()
+        except Exception:
+            pass
+        
+        logger.debug("WebSocket cleanup complete", job_id=job_id[:12])

backend/services/anonymous_indexer.py

@@ -60,6 +60,7 @@ class AnonymousIndexingJob:
     """
 
     REDIS_PREFIX = "anon_job:"
+    PUBSUB_PREFIX = "job:"  # Channel: job:{job_id}:events
     JOB_TTL_SECONDS = 3600  # 1 hour for job metadata
     REPO_TTL_HOURS = 24  # 24 hours for indexed data
     TEMP_DIR = "/tmp/anon_repos"
@@ -71,6 +72,27 @@ def __init__(self, redis_client):
         # Ensure temp directory exists
         Path(self.TEMP_DIR).mkdir(parents=True, exist_ok=True)
 
+    def _get_channel(self, job_id: str) -> str:
+        """Get Redis pub/sub channel for job events."""
+        return f"{self.PUBSUB_PREFIX}{job_id}:events"
+
+    def _publish_event(self, job_id: str, event: dict) -> None:
+        """
+        Publish event to Redis pub/sub for real-time WebSocket updates.
+        
+        Events are fire-and-forget - if no one is listening, that's fine.
+        The job state in Redis is the source of truth for polling fallback.
+        """
+        if not self.redis:
+            return
+        
+        try:
+            channel = self._get_channel(job_id)
+            self.redis.publish(channel, json.dumps(event))
+        except Exception as e:
+            # Don't fail the job if pub/sub fails - it's nice-to-have
+            logger.warning("Failed to publish event", job_id=job_id, error=str(e))
+
     @staticmethod
     def generate_job_id() -> str:
         """Generate unique job ID."""
@@ -160,7 +182,7 @@ def update_status(
         error: Optional[str] = None,
         error_message: Optional[str] = None
     ) -> bool:
-        """Update job status in Redis."""
+        """Update job status in Redis and publish event to WebSocket clients."""
         if not self.redis:
             return False
 
@@ -185,6 +207,30 @@ def update_status(
         key = self._get_key(job_id)
         self.redis.setex(key, self.JOB_TTL_SECONDS, json.dumps(job))
 
+        # Publish status change event for WebSocket clients
+        # Note: PROCESSING events are handled by update_progress() to avoid duplicates
+        event = {"type": status.value, "job_id": job_id}
+        
+        if status == JobStatus.QUEUED:
+            event["message"] = "Job queued for processing"
+        elif status == JobStatus.CLONING:
+            event["message"] = "Cloning repository..."
+            event["repo_name"] = job.get("repo_name")
+        elif status == JobStatus.PROCESSING:
+            # Skip publishing here - update_progress() sends granular progress events
+            return True
+        elif status == JobStatus.COMPLETED:
+            event["message"] = "Indexing complete!"
+            event["repo_id"] = repo_id
+            if stats:
+                event["stats"] = stats.to_dict()
+        elif status == JobStatus.FAILED:
+            event["message"] = error_message or "Indexing failed"
+            event["error"] = error
+            event["recoverable"] = error not in ("timeout", "clone_failed")
+        
+        self._publish_event(job_id, event)
+
         return True
 
     def update_progress(
@@ -195,7 +241,26 @@ def update_progress(
         files_total: int,
         current_file: Optional[str] = None
     ) -> bool:
-        """Update job progress (called during indexing)."""
+        """
+        Update job progress during indexing.
+        
+        Publishes real-time progress event for WebSocket clients,
+        then updates the job state in Redis.
+        """
+        # Publish progress event for real-time streaming
+        # This is separate from status events - more granular
+        percent = int((files_processed / files_total) * 100) if files_total > 0 else 0
+        
+        self._publish_event(job_id, {
+            "type": "progress",
+            "files_processed": files_processed,
+            "files_total": files_total,
+            "functions_found": functions_found,
+            "current_file": current_file,
+            "percent": percent
+        })
+        
+        # Update job state in Redis (for polling fallback)
         progress = JobProgress(
             files_total=files_total,
             files_processed=files_processed,