fix gestione update e reload sessioni
This commit is contained in:
Samuele E. Locatelli
+61
-66
@@ -1,49 +1,73 @@
|
||||
# api/v1/sessions.py
|
||||
|
||||
from typing import List
|
||||
from fastapi import Body, Query, Path
|
||||
from fastapi import Body, Query, Path, WebSocket, WebSocketDisconnect
|
||||
from services import redis_service
|
||||
from fastapi import APIRouter
|
||||
from models.session import SessionMeta
|
||||
import json
|
||||
import asyncio
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@router.post("/sessions", response_model=dict)
|
||||
async def create_session_endpoint(
|
||||
user_id: str = Query(..., description="User ID"),
|
||||
first_message: str = Body("", embed=True)
|
||||
):
|
||||
"""
|
||||
Create a new chat session for a user.
|
||||
Session name is taken from `first_message` or defaults to "New Chat".
|
||||
"""
|
||||
meta = redis_service.create_session(user_id, first_message)
|
||||
return meta
|
||||
# -------------------------
|
||||
# WebSocket per aggiornamenti sessioni
|
||||
# -------------------------
|
||||
@router.websocket("/ws/sessions")
|
||||
async def sessions_ws(websocket: WebSocket, user_id: str = Query(...)):
|
||||
await websocket.accept()
|
||||
try:
|
||||
# Invia subito la lista completa
|
||||
sessions = redis_service.get_sessions(user_id)
|
||||
await websocket.send_json({"type": "full_list", "sessions": sessions})
|
||||
|
||||
# Sottoscrizione al canale Redis
|
||||
pubsub = redis_service.r.pubsub()
|
||||
channel = f"sessions:{user_id}"
|
||||
await pubsub.subscribe(channel)
|
||||
|
||||
async for message in pubsub.listen():
|
||||
if message["type"] != "message":
|
||||
continue
|
||||
try:
|
||||
data = json.loads(message["data"])
|
||||
await websocket.send_json(data)
|
||||
except Exception as e:
|
||||
print(f"[WS] Errore parsing messaggio: {e}")
|
||||
|
||||
except WebSocketDisconnect:
|
||||
print(f"[WS] Disconnesso: {user_id}")
|
||||
finally:
|
||||
await pubsub.unsubscribe(channel)
|
||||
await pubsub.close()
|
||||
|
||||
# -------------------------
|
||||
# Endpoint REST
|
||||
# -------------------------
|
||||
|
||||
@router.get("/sessions", response_model=List[dict])
|
||||
async def list_sessions_endpoint(
|
||||
user_id: str = Query(..., description="User ID")
|
||||
):
|
||||
"""
|
||||
Get a list of all saved sessions for a user (most recent first).
|
||||
"""
|
||||
#print(f"[DEBUG] user_id={user_id!r}")
|
||||
sessions = redis_service.get_sessions(user_id)
|
||||
print(f"[DEBUG] sessions={sessions}")
|
||||
return sessions
|
||||
|
||||
async def list_sessions_endpoint(user_id: str = Query(...)):
|
||||
return redis_service.get_sessions(user_id)
|
||||
|
||||
@router.get("/sessions/{session_id}", response_model=dict)
|
||||
async def get_session_meta_endpoint(
|
||||
user_id: str = Query(..., description="User ID"),
|
||||
session_id: str = Path(..., description="Session ID")
|
||||
):
|
||||
"""
|
||||
Get metadata for a single session.
|
||||
"""
|
||||
return redis_service.get_session_meta(user_id, session_id) or {}
|
||||
|
||||
@router.post("/sessions", response_model=dict)
|
||||
async def create_session_endpoint(
|
||||
user_id: str = Query(..., description="User ID"),
|
||||
first_message: str = Body("", embed=True)
|
||||
):
|
||||
meta = redis_service.create_session(user_id, first_message)
|
||||
# Notifica WS
|
||||
redis_service.r.publish(
|
||||
f"sessions:{user_id}",
|
||||
json.dumps({"type": "created", "session": meta})
|
||||
)
|
||||
return meta
|
||||
|
||||
@router.patch("/sessions/{session_id}", response_model=dict)
|
||||
async def update_session_endpoint(
|
||||
@@ -51,53 +75,24 @@ async def update_session_endpoint(
|
||||
session_id: str = Path(..., description="Session ID"),
|
||||
session_name: str = Body(..., embed=True)
|
||||
):
|
||||
"""
|
||||
Rename a session or update metadata fields.
|
||||
"""
|
||||
return redis_service.update_session_meta(user_id, session_id, session_name=session_name) or {}
|
||||
|
||||
updated = redis_service.update_session_meta(user_id, session_id, session_name=session_name) or {}
|
||||
if updated:
|
||||
redis_service.r.publish(
|
||||
f"sessions:{user_id}",
|
||||
json.dumps({"type": "updated", "session": updated})
|
||||
)
|
||||
return updated
|
||||
|
||||
@router.delete("/sessions/{session_id}", response_model=dict)
|
||||
async def delete_session_endpoint(
|
||||
user_id: str = Query(..., description="User ID"),
|
||||
session_id: str = Path(..., description="Session ID")
|
||||
):
|
||||
"""
|
||||
Delete a session and its chat history.
|
||||
"""
|
||||
redis_service.delete_session(user_id, session_id)
|
||||
redis_service.r.publish(
|
||||
f"sessions:{user_id}",
|
||||
json.dumps({"type": "deleted", "session_id": session_id})
|
||||
)
|
||||
return {"status": "deleted"}
|
||||
|
||||
|
||||
# -------------------------
|
||||
# Gestione TTL sessioni
|
||||
# -------------------------
|
||||
|
||||
@router.post("/sessions/{session_id}/extend_ttl", response_model=dict)
|
||||
async def extend_ttl_endpoint(
|
||||
user_id: str = Query(..., description="User ID"),
|
||||
session_id: str = Path(..., description="Session ID"),
|
||||
extra_seconds: int = Body(..., embed=True, description="Seconds to extend the TTL")
|
||||
):
|
||||
"""
|
||||
Extend the TTL (time-to-live) for a given session and its related data.
|
||||
"""
|
||||
redis_service.extend_session_ttl(user_id, session_id, extra_seconds)
|
||||
return {"status": "ttl_extended", "extra_seconds": extra_seconds}
|
||||
|
||||
|
||||
@router.post("/sessions/{session_id}/refresh_ttl", response_model=dict)
|
||||
async def refresh_ttl_endpoint(
|
||||
user_id: str = Query(..., description="User ID"),
|
||||
session_id: str = Path(..., description="Session ID")
|
||||
):
|
||||
"""
|
||||
Reset the TTL for a given session and its related data back to the default value.
|
||||
"""
|
||||
redis_service.refresh_session_ttl(user_id, session_id)
|
||||
return {
|
||||
"status": "ttl_refreshed",
|
||||
"default_ttl_seconds": redis_service.DEFAULT_TTL_SECONDS
|
||||
}
|
||||
|
||||
|
||||
|
||||
+2
-2
@@ -6,8 +6,8 @@ class Settings(BaseSettings):
|
||||
REDIS_PORT: int = int(os.getenv("REDIS_PORT", 6379))
|
||||
REDIS_DB: int = int(os.getenv("REDIS_DB", 0))
|
||||
LM_STUDIO_URL: str = os.getenv("LM_STUDIO_URL", "http://10.74.83.100:1234/v1/chat/completions")
|
||||
MODEL_NAME: str = os.getenv("MODEL_NAME", "qwen/qwen3-4b-thinking-2507")
|
||||
#MODEL_NAME: str = os.getenv("MODEL_NAME", "qwen/qwen3-4b-2507")
|
||||
#MODEL_NAME: str = os.getenv("MODEL_NAME", "qwen/qwen3-4b-thinking-2507")
|
||||
MODEL_NAME: str = os.getenv("MODEL_NAME", "qwen/qwen3-4b-2507")
|
||||
REQUEST_TIMEOUT: float = float(os.getenv("REQUEST_TIMEOUT", 30.0))
|
||||
|
||||
settings = Settings()
|
||||
|
||||
@@ -10,6 +10,7 @@ from typing import List, Optional
|
||||
# TTL di default: 30 giorni
|
||||
DEFAULT_TTL_SECONDS = 30 * 24 * 60 * 60
|
||||
|
||||
# Connessione principale
|
||||
r = redis.Redis(
|
||||
host=settings.REDIS_HOST,
|
||||
port=settings.REDIS_PORT,
|
||||
@@ -17,24 +18,37 @@ r = redis.Redis(
|
||||
decode_responses=True
|
||||
)
|
||||
|
||||
# -------------------------
|
||||
# Nuovo: helper per WS
|
||||
# -------------------------
|
||||
|
||||
def publish_session_event(user_id: str, event_type: str, **kwargs):
|
||||
"""
|
||||
Pubblica un evento sul canale WS dell'utente.
|
||||
event_type: "full_list", "created", "updated", "deleted"
|
||||
kwargs: dati extra (es. session, session_id)
|
||||
"""
|
||||
channel = f"sessions:{user_id}"
|
||||
payload = {"type": event_type}
|
||||
payload.update(kwargs)
|
||||
try:
|
||||
r.publish(channel, json.dumps(payload))
|
||||
except Exception as e:
|
||||
print(f"[Redis] Errore publish su {channel}: {e}")
|
||||
|
||||
# -------------------------
|
||||
# Chat message operations
|
||||
# -------------------------
|
||||
|
||||
def save_chat(user_id: str, session_id: str, message: dict):
|
||||
"""Save a message for a user/session, adding a UTC ISO timestamp.
|
||||
Also updates session metadata and TTL.
|
||||
"""
|
||||
key = f"chatHistory:{user_id}:{session_id}"
|
||||
enriched = {**message, "timestamp": datetime.utcnow().isoformat() + "Z"}
|
||||
r.rpush(key, json.dumps(enriched))
|
||||
r.expire(key, DEFAULT_TTL_SECONDS) # rinnova TTL history
|
||||
r.expire(key, DEFAULT_TTL_SECONDS)
|
||||
_update_session_stats(user_id, session_id)
|
||||
refresh_session_ttl(user_id, session_id) # rinnova TTL meta e indice
|
||||
|
||||
refresh_session_ttl(user_id, session_id)
|
||||
|
||||
def get_chat(user_id: str, session_id: str, limit: Optional[int] = None):
|
||||
"""Retrieve messages for a user/session, sorted by timestamp."""
|
||||
key = f"chatHistory:{user_id}:{session_id}"
|
||||
data = r.lrange(key, 0, -1)
|
||||
messages = [json.loads(item) for item in data]
|
||||
@@ -43,21 +57,17 @@ def get_chat(user_id: str, session_id: str, limit: Optional[int] = None):
|
||||
messages = messages[-limit:]
|
||||
return messages
|
||||
|
||||
|
||||
def clear_chat(user_id: str, session_id: str):
|
||||
"""Delete all messages for a user/session and reset stats."""
|
||||
key = f"chatHistory:{user_id}:{session_id}"
|
||||
r.delete(key)
|
||||
_update_session_stats(user_id, session_id, reset=True)
|
||||
refresh_session_ttl(user_id, session_id)
|
||||
|
||||
|
||||
# -------------------------
|
||||
# Session metadata ops
|
||||
# -------------------------
|
||||
|
||||
def create_session(user_id: str, first_message: str) -> dict:
|
||||
"""Create a new session with initial metadata and TTL."""
|
||||
session_id = str(uuid.uuid4())
|
||||
created_at = datetime.utcnow().isoformat() + "Z"
|
||||
session_name = (first_message.strip()[:50] or "New Chat")
|
||||
@@ -72,64 +82,56 @@ def create_session(user_id: str, first_message: str) -> dict:
|
||||
index_key = f"chatSessionsIndex:{user_id}"
|
||||
|
||||
r.set(meta_key, json.dumps(meta), ex=DEFAULT_TTL_SECONDS)
|
||||
|
||||
# Score come timestamp float, ma non sarà usato per filtrare
|
||||
score = float(datetime.utcnow().timestamp())
|
||||
r.zadd(index_key, {session_id: score})
|
||||
r.expire(index_key, DEFAULT_TTL_SECONDS)
|
||||
|
||||
# Notifica WS
|
||||
publish_session_event(user_id, "created", session=meta)
|
||||
|
||||
return meta
|
||||
|
||||
|
||||
def get_sessions(user_id: str) -> List[dict]:
|
||||
"""Return all sessions for a user, newest first, ignoring score filtering."""
|
||||
key = f"chatSessionsIndex:{user_id}"
|
||||
|
||||
# Recupera tutti i membri in ordine di score decrescente
|
||||
session_ids = r.zrevrange(key, 0, -1)
|
||||
|
||||
print(f"[DEBUG] key={key}")
|
||||
print(f"[DEBUG] session_ids={session_ids}")
|
||||
|
||||
sessions = []
|
||||
for sid in session_ids:
|
||||
session_key = f"chatSession:{user_id}:{sid}"
|
||||
raw = r.get(session_key)
|
||||
print(f"[DEBUG] GET {session_key} -> {raw}")
|
||||
if raw:
|
||||
sessions.append(json.loads(raw))
|
||||
return sessions
|
||||
|
||||
|
||||
def get_session_meta(user_id: str, session_id: str) -> Optional[dict]:
|
||||
raw = r.get(f"chatSession:{user_id}:{session_id}")
|
||||
return json.loads(raw) if raw else None
|
||||
|
||||
|
||||
def update_session_meta(user_id: str, session_id: str, **updates) -> Optional[dict]:
|
||||
"""Update metadata fields (e.g., session_name) for a session."""
|
||||
meta = get_session_meta(user_id, session_id)
|
||||
if not meta:
|
||||
return None
|
||||
meta.update(updates)
|
||||
r.set(f"chatSession:{user_id}:{session_id}", json.dumps(meta), ex=DEFAULT_TTL_SECONDS)
|
||||
refresh_session_ttl(user_id, session_id)
|
||||
|
||||
# Notifica WS
|
||||
publish_session_event(user_id, "updated", session=meta)
|
||||
|
||||
return meta
|
||||
|
||||
|
||||
def delete_session(user_id: str, session_id: str):
|
||||
"""Delete metadata, chat history, and remove from index."""
|
||||
r.delete(f"chatSession:{user_id}:{session_id}")
|
||||
r.delete(f"chatHistory:{user_id}:{session_id}")
|
||||
r.zrem(f"chatSessionsIndex:{user_id}", session_id)
|
||||
|
||||
# Notifica WS
|
||||
publish_session_event(user_id, "deleted", session_id=session_id)
|
||||
|
||||
# -------------------------
|
||||
# TTL management
|
||||
# -------------------------
|
||||
|
||||
def extend_session_ttl(user_id: str, session_id: str, extra_seconds: int):
|
||||
"""Extend TTL for session metadata, history, and index."""
|
||||
keys = [
|
||||
f"chatSession:{user_id}:{session_id}",
|
||||
f"chatHistory:{user_id}:{session_id}",
|
||||
@@ -143,9 +145,7 @@ def extend_session_ttl(user_id: str, session_id: str, extra_seconds: int):
|
||||
else:
|
||||
r.expire(key, extra_seconds)
|
||||
|
||||
|
||||
def refresh_session_ttl(user_id: str, session_id: str):
|
||||
"""Reset TTL for session metadata, history, and index to default."""
|
||||
keys = [
|
||||
f"chatSession:{user_id}:{session_id}",
|
||||
f"chatHistory:{user_id}:{session_id}",
|
||||
@@ -155,13 +155,11 @@ def refresh_session_ttl(user_id: str, session_id: str):
|
||||
if r.exists(key):
|
||||
r.expire(key, DEFAULT_TTL_SECONDS)
|
||||
|
||||
|
||||
# -------------------------
|
||||
# Internal helpers
|
||||
# -------------------------
|
||||
|
||||
def _update_session_stats(user_id: str, session_id: str, reset: bool = False):
|
||||
"""Refresh message_count and history_size_bytes in metadata."""
|
||||
meta = get_session_meta(user_id, session_id)
|
||||
if not meta:
|
||||
return
|
||||
@@ -175,4 +173,3 @@ def _update_session_stats(user_id: str, session_id: str, reset: bool = False):
|
||||
meta["history_size_bytes"] = sum(len(m.encode("utf-8")) for m in messages)
|
||||
r.set(f"chatSession:{user_id}:{session_id}", json.dumps(meta), ex=DEFAULT_TTL_SECONDS)
|
||||
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
// src/SessionTable.jsx
|
||||
import React, { useEffect, useState } from "react";
|
||||
import React, { useEffect, useState, useRef } from "react";
|
||||
import { getSessionId } from "./useSessionId";
|
||||
|
||||
export default function SessionTable({ userId, onSelectSession, onClosePanel }) {
|
||||
@@ -8,8 +8,10 @@ export default function SessionTable({ userId, onSelectSession, onClosePanel })
|
||||
const [editingSessionId, setEditingSessionId] = useState(null);
|
||||
const [editName, setEditName] = useState("");
|
||||
const activeSessionId = getSessionId();
|
||||
const wsRef = useRef(null);
|
||||
|
||||
const fetchSessions = async () => {
|
||||
if (!userId) return;
|
||||
setLoading(true);
|
||||
try {
|
||||
const res = await fetch(`/v1/sessions?user_id=${userId}`);
|
||||
@@ -23,7 +25,51 @@ export default function SessionTable({ userId, onSelectSession, onClosePanel })
|
||||
};
|
||||
|
||||
useEffect(() => {
|
||||
if (userId) fetchSessions();
|
||||
if (!userId) return;
|
||||
|
||||
// Caricamento iniziale
|
||||
fetchSessions();
|
||||
|
||||
// Apri WebSocket
|
||||
const wsUrl = `${location.protocol === "https:" ? "wss" : "ws"}://${location.host}/v1/ws/sessions?user_id=${userId}`;
|
||||
const ws = new WebSocket(wsUrl);
|
||||
wsRef.current = ws;
|
||||
|
||||
ws.onopen = () => {
|
||||
console.log("[WS] Connected to sessions stream");
|
||||
};
|
||||
|
||||
ws.onmessage = (event) => {
|
||||
try {
|
||||
const msg = JSON.parse(event.data);
|
||||
// Possibili tipi di evento: full_list, created, updated, deleted
|
||||
if (msg.type === "full_list") {
|
||||
setSessions(msg.sessions);
|
||||
} else if (msg.type === "created") {
|
||||
setSessions((prev) => [msg.session, ...prev]);
|
||||
} else if (msg.type === "updated" && msg.session?.session_id) {
|
||||
setSessions((prev) =>
|
||||
prev.map((s) =>
|
||||
String(s.session_id) === String(msg.session.session_id)
|
||||
? { ...s, ...msg.session }
|
||||
: s
|
||||
)
|
||||
);
|
||||
} else if (msg.type === "deleted") {
|
||||
setSessions((prev) => prev.filter((s) => s.session_id !== msg.session_id));
|
||||
}
|
||||
} catch (err) {
|
||||
console.error("[WS] Error parsing message", err);
|
||||
}
|
||||
};
|
||||
|
||||
ws.onclose = () => {
|
||||
console.log("[WS] Disconnected from sessions stream");
|
||||
};
|
||||
|
||||
return () => {
|
||||
ws.close();
|
||||
};
|
||||
}, [userId]);
|
||||
|
||||
const startEditing = (session) => {
|
||||
@@ -38,8 +84,13 @@ export default function SessionTable({ userId, onSelectSession, onClosePanel })
|
||||
headers: { "Content-Type": "application/json" },
|
||||
body: JSON.stringify({ session_name: editName }),
|
||||
});
|
||||
// Aggiornamento ottimistico
|
||||
setSessions((prev) =>
|
||||
prev.map((s) =>
|
||||
s.session_id === sessionId ? { ...s, session_name: editName } : s
|
||||
)
|
||||
);
|
||||
setEditingSessionId(null);
|
||||
fetchSessions();
|
||||
} catch (err) {
|
||||
console.error("Failed to rename session", err);
|
||||
}
|
||||
@@ -49,7 +100,7 @@ export default function SessionTable({ userId, onSelectSession, onClosePanel })
|
||||
if (!window.confirm("Delete this session and its history?")) return;
|
||||
try {
|
||||
await fetch(`/v1/sessions/${sessionId}?user_id=${userId}`, { method: "DELETE" });
|
||||
fetchSessions();
|
||||
// Non serve fetchSessions: il WS notificherà la cancellazione
|
||||
} catch (err) {
|
||||
console.error("Failed to delete session", err);
|
||||
}
|
||||
@@ -115,7 +166,7 @@ export default function SessionTable({ userId, onSelectSession, onClosePanel })
|
||||
<td>{new Date(s.created_at).toLocaleString()}</td>
|
||||
<td title={s.history_size_bytes}>
|
||||
{s.message_count}
|
||||
</td>
|
||||
</td>
|
||||
<td className="text-end text-nowrap">
|
||||
<button
|
||||
className="btn btn-sm px-1 btn-outline-danger"
|
||||
@@ -131,3 +182,4 @@ export default function SessionTable({ userId, onSelectSession, onClosePanel })
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user