Merge pull request #49 from mezonai/feature/record_Agent_session

add logic interview

Author: hoangdohuy

Architect_MultiClient_Server/orchestrator_service/api/dispatch_api.py

@@ -13,6 +13,7 @@
 from pydantic import BaseModel
 from orchestrator_service.auth.verify_account import authenticate_account
 from orchestrator_service.services.livekit_client import get_livekit_service
+from orchestrator_service.services.interview_queue import get_interview_queue
 
 router = APIRouter()
 
@@ -70,7 +71,8 @@ class AccountModel(BaseModel):
 class DispatchRequestModel(BaseModel):
     account: AccountModel
     room_name: str
-
+    type: Optional[str] = "normal"
+    metadata: dict = {}
 
 async def ensure_dispatch(room_name: str) -> Dict[str, Any]:
     """
@@ -172,6 +174,20 @@ async def api_create_dispatch(body: DispatchRequestModel) -> Dict[str, Any]:
     """Create a dispatch for the specified room."""
     await verify_account(body.account.dict())
 
+    if(body.type == "interview"):
+        try:
+            # For interview dispatches, add to interview queue
+            interview_id = body.metadata.get("interview_id")
+            room_name = body.room_name
+            
+            if interview_id:
+                interview_queue = get_interview_queue()
+                interview_queue.add_by_room_name(room_name, interview_id)
+            else:
+                raise HTTPException(status_code=400, detail="interview_id is required in metadata for interview dispatch")
+    
+        except Exception as e:
+            raise HTTPException(status_code=400, detail=f"Invalid metadata for interview dispatch: {str(e)}")
     result = await ensure_dispatch(body.room_name)
     if result["status"] == DispatchStatus.ERROR:
         raise HTTPException(status_code=500, detail=result["message"])

Architect_MultiClient_Server/orchestrator_service/api/room_registry_api.py

@@ -12,6 +12,7 @@
 from orchestrator_service.services.room_registry import get_room_registry
 from orchestrator_service.services.livekit_client import get_livekit_service
 from orchestrator_service.services.transcription_service import TranscriptionService
+from orchestrator_service.services.interview_queue import get_interview_queue
 from orchestrator_service.auth.transcript_auth import verify_api_key
 
 # Import để có thể access egress_service
@@ -62,15 +63,17 @@ async def register_room(
     """
 
     registry = get_room_registry()
+    interview_queue = get_interview_queue()
     stt_room_id = None
     tracks_started = 0
 
     # 1. Start room in STT service FIRST
     try:
         stt_response = await transcription_service.start_room(request.room_name)
-        if stt_response.get("success"):
-            stt_room_id = stt_response.get("room_id")
-            logger.info(f"✅ Room '{request.room_name}' started in STT service")
+        if stt_response:
+            if stt_response.get("success"):
+                stt_room_id = stt_response.get("room_id")
+                logger.info(f"✅ Room '{request.room_name}' started in STT service")
         else:
             logger.warning(f"⚠️ Failed to start room in STT service")
     except Exception as e:
@@ -83,6 +86,12 @@ async def register_room(
             detail=f"Room '{request.room_name}' is already registered"
         )
 
+    # 2.5. Check if this room is in interview queue and update mapping
+    if stt_room_id and interview_queue.get_interview_id(request.room_name):
+        # Update from room_name to room_id
+        interview_queue.update_to_room_id(request.room_name, stt_room_id)
+        logger.info(f"✅ Interview mapping updated: room_name='{request.room_name}' → room_id='{stt_room_id}'")
+    
     # 3. Start recording for existing tracks (best effort)
     try:
         livekit_service = get_livekit_service()

Architect_MultiClient_Server/orchestrator_service/api/summary_api.py

@@ -1,9 +1,12 @@
 """
 Internal API endpoints for room summary
 """
+import asyncio
 from fastapi import APIRouter, HTTPException, Body, Depends, Header, Query
 from pydantic import BaseModel
 from orchestrator_service.services.summary_service import get_summary_service
+from orchestrator_service.services.interview_queue import get_interview_queue
+from orchestrator_service.services.interview_webhook_service import get_interview_webhook_service
 from orchestrator_service.config.application_config import get_config
 from datetime import datetime
 from typing import Optional
@@ -32,7 +35,22 @@ async def generate_room_summary(request: SummaryRequest = Body(...)):
     """
     Internal endpoint to generate a summary for a room.
     Input: {"room_id": "..."}
+    
+    If the room is associated with an interview, also sends track data to interview webhook.
     """
+    
+    # Check if this room is associated with an interview
+    interview_queue = get_interview_queue()
+    interview_id = interview_queue.get_interview_id(request.room_id)
+    
+    if interview_id:
+        # Send interview data to webhook asynchronously
+        webhook_service = get_interview_webhook_service()
+        asyncio.create_task(
+            webhook_service.send_interview_data(interview_id, request.room_id)
+        )
+        interview_queue.remove(request.room_id)  # Optionally remove from queue after processing
+
     service = get_summary_service()
     result = await service.generate_summary(request.room_id)
 

Architect_MultiClient_Server/orchestrator_service/config/application_config.py

@@ -239,6 +239,35 @@ def validate(self) -> bool:
         return True
 
 
+# ============================================================================
+# Interview Configuration
+# ============================================================================
+
+@dataclass
+class InterviewConfig:
+    """Configuration for interview webhook service"""
+    webhook_url: str = ""
+    webhook_api_key: str = ""
+    timeout: float = 30.0
+    enabled: bool = False
+    
+    @classmethod
+    def from_env(cls) -> 'InterviewConfig':
+        """Create Interview config from environment variables"""
+        return cls(
+            webhook_url=os.getenv('INTERVIEW_WEBHOOK_URL', ''),
+            webhook_api_key=os.getenv('INTERVIEW_WEBHOOK_API_KEY', ''),
+            timeout=float(os.getenv('INTERVIEW_WEBHOOK_TIMEOUT', '30.0')),
+            enabled=os.getenv('INTERVIEW_ENABLED', 'false').lower() == 'true',
+        )
+    
+    def validate(self) -> bool:
+        """Validate interview configuration"""
+        if self.enabled and not self.webhook_url:
+            return False
+        return True
+
+
 # ============================================================================
 # Main Application Configuration (Singleton)
 # ============================================================================
@@ -268,6 +297,7 @@ def __init__(self):
         self.logger = LoggerConfig.from_env()
         self.minio = MinIOConfig.from_env()
         self.llm = LLMConfig.from_env()
+        self.interview = InterviewConfig.from_env()
 
         self._initialized = True
         self._validate_all()
@@ -278,6 +308,8 @@ def _validate_all(self):
             raise ValueError("Invalid LiveKit configuration")
         if not self.minio.validate():
             raise ValueError("Invalid MinIO configuration")
+        if not self.interview.validate():
+            raise ValueError("Invalid Interview configuration")
 
     @classmethod
     def get_instance(cls) -> 'Config':

Architect_MultiClient_Server/orchestrator_service/services/interview_queue.py

@@ -0,0 +1,128 @@
+"""
+Interview Queue Service - Singleton for managing interview room mappings
+"""
+from typing import Dict, Optional
+from orchestrator_service.utils.logger import get_logger
+
+logger = get_logger(__name__)
+
+
+class InterviewQueue:
+    """
+    Singleton service to manage interview room mappings.
+    
+    Lifecycle:
+    1. When interview dispatch is created: {room_name: interview_id}
+    2. When room is registered: update to {room_id: interview_id}
+    3. When summary is generated: retrieve interview_id by room_id and send to webhook
+    """
+    
+    _instance = None
+    
+    def __new__(cls):
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._instance._initialized = False
+        return cls._instance
+    
+    def __init__(self):
+        if self._initialized:
+            return
+        
+        # Store mappings: key can be room_name or room_id, value is interview_id
+        self._queue: Dict[str, str] = {}
+        self._initialized = True
+        logger.info("InterviewQueue initialized")
+    
+    def add_by_room_name(self, room_name: str, interview_id: str) -> bool:
+        """
+        Add interview mapping by room_name.
+        Called when dispatch is created with interview type.
+        
+        Args:
+            room_name: LiveKit room name
+            interview_id: Interview identifier
+            
+        Returns:
+            True if added successfully
+        """
+        if not room_name or not interview_id:
+            logger.warning(f"Invalid parameters: room_name={room_name}, interview_id={interview_id}")
+            return False
+        
+        self._queue[room_name] = interview_id
+        logger.info(f"✅ Interview added: room_name='{room_name}' → interview_id='{interview_id}'")
+        return True
+    
+    def update_to_room_id(self, room_name: str, room_id: str) -> bool:
+        """
+        Update mapping from room_name to room_id when room is registered.
+        
+        Args:
+            room_name: LiveKit room name
+            room_id: MongoDB room _id
+            
+        Returns:
+            True if updated successfully, False if room_name not found
+        """
+        if room_name not in self._queue:
+            logger.debug(f"Room '{room_name}' not in interview queue, skipping update")
+            return False
+        
+        interview_id = self._queue[room_name]
+        
+        # Remove old room_name mapping
+        del self._queue[room_name]
+        
+        # Add new room_id mapping
+        self._queue[room_id] = interview_id
+        
+        logger.info(f"✅ Interview mapping updated: room_name='{room_name}' → room_id='{room_id}', interview_id='{interview_id}'")
+        return True
+    
+    def get_interview_id(self, key: str) -> Optional[str]:
+        """
+        Get interview_id by room_name or room_id.
+        
+        Args:
+            key: Either room_name or room_id
+            
+        Returns:
+            interview_id if found, None otherwise
+        """
+        return self._queue.get(key)
+    
+    def remove(self, key: str) -> bool:
+        """
+        Remove interview mapping.
+        
+        Args:
+            key: Either room_name or room_id
+            
+        Returns:
+            True if removed, False if not found
+        """
+        if key in self._queue:
+            interview_id = self._queue.pop(key)
+            logger.info(f"🗑️ Interview mapping removed: key='{key}', interview_id='{interview_id}'")
+            return True
+        return False
+    
+    def clear(self):
+        """Clear all interview mappings."""
+        count = len(self._queue)
+        self._queue.clear()
+        logger.info(f"🗑️ Interview queue cleared: {count} mappings removed")
+    
+    def list_all(self) -> Dict[str, str]:
+        """Get all interview mappings."""
+        return self._queue.copy()
+    
+    def count(self) -> int:
+        """Get count of interview mappings."""
+        return len(self._queue)
+
+
+def get_interview_queue() -> InterviewQueue:
+    """Get the singleton interview queue instance."""
+    return InterviewQueue()