from datetime import datetime, timezone from typing import Dict, Any import json from fastapi import WebSocket # In-memory store of active visitor sessions: # { sessionId: { "sessionId": str, "siteId": str, "connectedAt": str, "status": str, "connectionStatus": str, "disconnectedAt": str|None, "user": dict|None, "ipAddress": str|None, "country": str|None, "websocket": WebSocket, "assignedAgentId": str|None, "assignedAt": str|None } } active_sessions: Dict[str, Dict[str, Any]] = {} # Grace-period background timers for IN_CHAT disconnected visitors: { sessionId: asyncio.Task } import asyncio grace_period_timers: Dict[str, asyncio.Task] = {} def cancel_grace_period_timer(session_id: str) -> None: timer_task = grace_period_timers.pop(session_id, None) if timer_task and not timer_task.done(): timer_task.cancel() async def start_grace_period_timer(session_id: str, site_id: str, grace_seconds: float = 30.0) -> None: cancel_grace_period_timer(session_id) async def _timer_coro(): try: await asyncio.sleep(grace_seconds) if session_id in active_sessions: active_sessions[session_id]["connectionStatus"] = "offline" active_sessions[session_id]["status"] = "ABANDONED" active_sessions[session_id]["assignedAgentId"] = None active_sessions[session_id]["assignedAt"] = None from app.database import get_database db = get_database() if db is not None: from app.repositories.session_repository import update_session_connection_status, update_session_status now = datetime.now(timezone.utc) await update_session_status(db, session_id, "ABANDONED") await update_session_connection_status(db, session_id, "offline") await db["sessions"].update_one( {"sessionId": session_id}, {"$set": {"assignedAgentId": None, "assignedAt": None}} ) await db["conversations"].update_one( {"sessionId": session_id}, {"$set": {"status": "ABANDONED", "assignedAgentId": None, "updatedAt": now}} ) await broadcast_to_agents({ "type": "visitor_timeout", "sessionId": session_id, "siteId": site_id, "connectionStatus": "offline", "status": "ABANDONED" }) remove_active_session(session_id) except asyncio.CancelledError: pass finally: grace_period_timers.pop(session_id, None) task = asyncio.create_task(_timer_coro()) grace_period_timers[session_id] = task # Structured pool of connected agents: # { agentId: { "agentId": str, "websocket": WebSocket, "connectedAt": str, "status": "AVAILABLE" | "BUSY" | "OFFLINE" } } connected_agents: Dict[str, Dict[str, Any]] = {} # --- Visitor Session Management --- def add_active_session( session_id: str, site_id: str, websocket: WebSocket, user_data: dict | None = None, ip_address: str | None = None, country: str | None = None, status_val: str = "RINGING", assigned_agent_id: str | None = None, assigned_at: str | None = None, connection_status: str = "online", disconnected_at: str | None = None ) -> Dict[str, Any]: now_iso = datetime.now(timezone.utc).isoformat() session_data = { "sessionId": session_id, "siteId": site_id, "status": status_val, "connectionStatus": connection_status, "disconnectedAt": disconnected_at, "user": user_data, "ipAddress": ip_address, "country": country or "Unknown", "connectedAt": now_iso, "websocket": websocket, "assignedAgentId": assigned_agent_id, "assignedAt": assigned_at } active_sessions[session_id] = session_data return session_data def remove_active_session(session_id: str) -> None: cancel_grace_period_timer(session_id) if session_id in active_sessions: del active_sessions[session_id] def get_active_sessions(site_id: str | None = None) -> Dict[str, Dict[str, Any]]: active = {sid: data for sid, data in active_sessions.items() if data.get("status") not in ("ENDED", "ABANDONED")} if site_id: return {sid: data for sid, data in active.items() if data.get("siteId") == site_id} return active def update_active_session_status(session_id: str, new_status: str, ended_by: str | None = None) -> None: if session_id in active_sessions: active_sessions[session_id]["status"] = new_status if ended_by: active_sessions[session_id]["endedBy"] = ended_by # --- Multi-Agent Connection Pool Management --- def add_agent_connection( agent_id: str, websocket: WebSocket, status_val: str = "AVAILABLE" ) -> Dict[str, Any]: """ Registers an agent WebSocket connection in the multi-agent presence pool. Safely replaces stale connections if the same agent reconnects. """ if not agent_id: agent_id = "agent_1" now_iso = datetime.now(timezone.utc).isoformat() agent_info = { "agentId": agent_id, "websocket": websocket, "status": status_val if status_val in ("AVAILABLE", "BUSY", "OFFLINE") else "AVAILABLE", "connectedAt": now_iso } connected_agents[agent_id] = agent_info return agent_info def remove_agent_connection(agent_id: str, websocket: WebSocket | None = None) -> None: """ Removes an agent connection. If a websocket is provided, only removes if it matches the currently registered active socket for that agent identity. """ if agent_id in connected_agents: if websocket is not None: active_ws = connected_agents[agent_id].get("websocket") if active_ws != websocket: return # Connection was replaced by a newer reconnect; do not delete active replacement del connected_agents[agent_id] def get_agent_connection(agent_id: str) -> Dict[str, Any] | None: """Returns presence info for a specific connected agent ID.""" return connected_agents.get(agent_id) def get_connected_agents() -> Dict[str, Dict[str, Any]]: """Returns a dict of all currently connected agents.""" return connected_agents def set_agent_status(agent_id: str, new_status: str) -> bool: """Updates presence status (AVAILABLE, BUSY, OFFLINE) for a connected agent.""" if new_status not in ("AVAILABLE", "BUSY", "OFFLINE"): return False if agent_id in connected_agents: connected_agents[agent_id]["status"] = new_status return True return False def get_available_agents() -> Dict[str, Dict[str, Any]]: """Returns all connected agents whose status is currently AVAILABLE.""" return {aid: data for aid, data in connected_agents.items() if data.get("status") == "AVAILABLE"} # Legacy Single-Agent Aliases for Backward Compatibility def add_agent_websocket(websocket: WebSocket) -> None: add_agent_connection("agent_1", websocket) def remove_agent_websocket(websocket: WebSocket) -> None: remove_agent_connection("agent_1", websocket) # --- Messaging & Notification Dispatch --- async def broadcast_to_agents(event_data: dict, target_agent_ids: list | None = None) -> None: """ Broadcasts a JSON event payload to connected agent dashboards in the pool. Optionally filters by target_agent_ids. """ to_remove = [] message_str = json.dumps(event_data) agents_to_notify = list(connected_agents.items()) if target_agent_ids is not None: target_set = set(target_agent_ids) agents_to_notify = [(aid, data) for aid, data in agents_to_notify if aid in target_set] for aid, data in agents_to_notify: ws = data.get("websocket") if ws: try: await ws.send_text(message_str) except Exception: to_remove.append((aid, ws)) for aid, ws in to_remove: remove_agent_connection(aid, ws) async def notify_visitor_socket(session_id: str, event_data: dict) -> None: """Sends a JSON event payload to a specific visitor's WebSocket connection.""" print(f"[VIDEO][TRACE][BACKEND] resolving visitor socket") print(f"[VIDEO][TRACE][BACKEND] sessionId = {session_id}") if session_id in active_sessions: ws = active_sessions[session_id].get("websocket") if ws: print(f"[VIDEO][TRACE][BACKEND] visitor socket found = true") print(f"[VIDEO][TRACE][BACKEND] socket state = open") print(f"[VIDEO][TRACE][BACKEND] forwarding event") print(f"[VIDEO][TRACE][BACKEND] sending to visitor:\n{json.dumps(event_data, indent=2)}") try: await ws.send_text(json.dumps(event_data)) return except Exception as e: print(f"[VIDEO][TRACE][BACKEND] Error sending to visitor socket: {e}") else: print(f"[VIDEO][TRACE][BACKEND] visitor socket found = false (websocket is None in active_sessions)") else: print(f"[VIDEO][TRACE][BACKEND] visitor socket found = false ({session_id} not in active_sessions: {list(active_sessions.keys())})") async def handle_agent_disconnect(db, agent_id: str) -> None: """ Handles agent WebSocket disconnection. 1. Finds active IN_CHAT sessions assigned to that agent in DB and in memory. 2. Requeues each IN_CHAT session back to RINGING state atomically via requeue_session(db, sid, agent_id). 3. Notifies visitor socket and remaining connected agents. """ if not agent_id: return from app.repositories.session_repository import requeue_session # Find IN_CHAT sessions assigned to this agent in MongoDB sessions_to_requeue = {} if db is not None: try: cursor = db["sessions"].find({"assignedAgentId": agent_id, "status": "IN_CHAT"}) async for doc in cursor: sessions_to_requeue[doc["sessionId"]] = doc.get("siteId", "") except Exception as e: print(f"Error querying sessions for disconnected agent {agent_id}: {e}") # Also check in-memory active sessions for sid, sdata in list(active_sessions.items()): if sdata.get("status") == "IN_CHAT" and sdata.get("assignedAgentId") == agent_id: sessions_to_requeue[sid] = sdata.get("siteId", "") for sid, site_id in sessions_to_requeue.items(): # Atomically requeue in DB passing agent_id to prevent race conditions requeued_doc = await requeue_session(db, sid, agent_id=agent_id) if db is not None else True if requeued_doc: # Update memory update_active_session_status(sid, "RINGING") if sid in active_sessions: active_sessions[sid]["assignedAgentId"] = None active_sessions[sid]["assignedAt"] = None # Notify visitor socket await notify_visitor_socket(sid, { "type": "status_changed", "status": "RINGING", "message": "Your agent disconnected. We're reconnecting you to another agent." }) # Broadcast visitor_requeued to remaining agents await broadcast_to_agents({ "type": "visitor_requeued", "sessionId": sid, "siteId": site_id, "status": "RINGING", "reason": "agent_disconnected" })