mirror of
https://github.com/idrainformatica/PecFlow.git
synced 2026-06-16 12:45:42 +02:00
115 lines
3.2 KiB
Python
115 lines
3.2 KiB
Python
"""
|
||
WebSocket endpoint – connessione real-time per aggiornamenti inbox.
|
||
|
||
URL: ws://<host>/api/v1/ws?token=<jwt_access_token>
|
||
|
||
Protocollo messaggi (dal server al client):
|
||
{
|
||
"type": "mailbox:new_message",
|
||
"mailbox_id": "<uuid>",
|
||
"message_id": "<uuid>",
|
||
"subject": "...",
|
||
"from_address": "...",
|
||
"received_at": "2026-03-18T14:00:00Z"
|
||
}
|
||
|
||
{
|
||
"type": "mailbox:sync_error",
|
||
"mailbox_id": "<uuid>",
|
||
"error": "...",
|
||
"status": "error"
|
||
}
|
||
|
||
{ "type": "ping" } ← heartbeat ogni 30s per mantenere connessione viva
|
||
"""
|
||
|
||
import asyncio
|
||
import uuid
|
||
|
||
from fastapi import APIRouter, Query, WebSocket, WebSocketDisconnect
|
||
from jose import JWTError
|
||
|
||
from app.core.security import decode_token
|
||
from app.core.logging import get_logger
|
||
from app.websocket.manager import manager
|
||
|
||
router = APIRouter(tags=["WebSocket"])
|
||
logger = get_logger(__name__)
|
||
|
||
|
||
@router.websocket("/ws")
|
||
async def websocket_endpoint(
|
||
websocket: WebSocket,
|
||
token: str = Query(..., description="JWT access token per autenticazione"),
|
||
) -> None:
|
||
"""
|
||
WebSocket endpoint autenticato via JWT query param.
|
||
Invia eventi real-time per il tenant dell'utente connesso.
|
||
"""
|
||
# Autenticazione: valida JWT
|
||
try:
|
||
payload = decode_token(token)
|
||
if payload.get("type") != "access":
|
||
await websocket.close(code=4001, reason="Token non valido")
|
||
return
|
||
|
||
tenant_id_str = payload.get("tid")
|
||
user_id_str = payload.get("sub")
|
||
if not tenant_id_str or not user_id_str:
|
||
await websocket.close(code=4001, reason="Token malformato")
|
||
return
|
||
|
||
tenant_id = uuid.UUID(tenant_id_str)
|
||
user_id = uuid.UUID(user_id_str)
|
||
|
||
except (JWTError, ValueError):
|
||
await websocket.close(code=4001, reason="Token non valido")
|
||
return
|
||
|
||
# Connessione accettata
|
||
await manager.connect(websocket, tenant_id)
|
||
logger.info(
|
||
"WS autenticato",
|
||
extra={"user_id": str(user_id), "tenant_id": str(tenant_id)},
|
||
)
|
||
|
||
# Invia ack di connessione
|
||
try:
|
||
await websocket.send_json({
|
||
"type": "connected",
|
||
"tenant_id": str(tenant_id),
|
||
"user_id": str(user_id),
|
||
})
|
||
except Exception:
|
||
await manager.disconnect(websocket, tenant_id)
|
||
return
|
||
|
||
# Heartbeat task
|
||
async def send_pings() -> None:
|
||
while True:
|
||
try:
|
||
await asyncio.sleep(30)
|
||
await websocket.send_json({"type": "ping"})
|
||
except Exception:
|
||
break
|
||
|
||
ping_task = asyncio.create_task(send_pings())
|
||
|
||
try:
|
||
# Mantieni la connessione aperta, gestisci messaggi client (pong, ecc.)
|
||
while True:
|
||
try:
|
||
data = await asyncio.wait_for(websocket.receive_text(), timeout=35.0)
|
||
# Gestisci pong dal client (opzionale)
|
||
if data == "pong":
|
||
continue
|
||
except asyncio.TimeoutError:
|
||
# Nessun messaggio dal client in 35s: connessione morta
|
||
break
|
||
except WebSocketDisconnect:
|
||
break
|
||
|
||
finally:
|
||
ping_task.cancel()
|
||
await manager.disconnect(websocket, tenant_id)
|