"""Conversation / message persistence helpers.""" from __future__ import annotations from datetime import timedelta from typing import Any from sqlalchemy import delete, func, or_, select from sqlalchemy.orm import selectinload from app.db import SessionLocal from app.models import Conversation, Message, ToolCall, utcnow def conv_to_dict(c: Conversation) -> dict[str, Any]: return {"id": c.id, "title": c.title, "course": c.course_code, "pinned": c.pinned, "archived": c.archived, "created_at": c.created_at.isoformat(), "updated_at": c.updated_at.isoformat()} def message_to_dict(m: Message, include_tools: bool = True) -> dict[str, Any]: d: dict[str, Any] = { "id": m.id, "role": m.role, "content": m.content, "model": m.model, "created_at": m.created_at.isoformat(), "feedback": m.feedback, "attachments": m.attachments or [], "cost_usd": m.cost_usd, "tokens": {"in": m.tokens_in, "out": m.tokens_out}, "latency_ms": m.latency_ms, } if include_tools: d["tool_calls"] = [{ "id": t.id, "name": t.name, "arguments": t.arguments, "summary": t.result_summary, "payload": t.result_payload, "status": t.status, "duration_ms": t.duration_ms, } for t in m.tool_calls] return d async def list_conversations(user_id: str, q: str | None = None, archived: bool = False, limit: int = 50, offset: int = 0) -> list[dict[str, Any]]: async with SessionLocal() as session: stmt = select(Conversation).where(Conversation.user_id == user_id, Conversation.archived == archived) if q: like = f"%{q}%" sub = select(Message.conversation_id).where(Message.content.ilike(like)) stmt = stmt.where(or_(Conversation.title.ilike(like), Conversation.id.in_(sub))) stmt = stmt.order_by(Conversation.pinned.desc(), Conversation.updated_at.desc()) \ .limit(limit).offset(offset) rows = (await session.execute(stmt)).scalars() return [conv_to_dict(c) for c in rows] async def create_conversation(user_id: str, course: str, title: str | None = None) -> dict[str, Any]: async with SessionLocal() as session: c = Conversation(user_id=user_id, course_code=course.upper(), title=title or "Nouvelle conversation") session.add(c) await session.commit() await session.refresh(c) return conv_to_dict(c) async def get_conversation(conv_id: str, user_id: str | None = None) -> Conversation | None: async with SessionLocal() as session: c = await session.get(Conversation, conv_id) if c is None or (user_id is not None and c.user_id != user_id): return None return c async def get_conversation_full(conv_id: str, user_id: str) -> dict[str, Any] | None: async with SessionLocal() as session: c = await session.scalar( select(Conversation).where(Conversation.id == conv_id, Conversation.user_id == user_id) .options(selectinload(Conversation.messages).selectinload(Message.tool_calls))) if c is None: return None return {**conv_to_dict(c), "messages": [message_to_dict(m) for m in c.messages]} async def update_conversation(conv_id: str, user_id: str, **fields: Any) -> dict[str, Any] | None: async with SessionLocal() as session: c = await session.get(Conversation, conv_id) if c is None or c.user_id != user_id: return None for k, v in fields.items(): if v is not None and k in {"title", "pinned", "archived", "course_code"}: setattr(c, k, v) await session.commit() await session.refresh(c) return conv_to_dict(c) async def delete_conversation(conv_id: str, user_id: str) -> bool: async with SessionLocal() as session: c = await session.get(Conversation, conv_id) if c is None or c.user_id != user_id: return False msg_ids = [m for m in (await session.execute( select(Message.id).where(Message.conversation_id == conv_id))).scalars()] if msg_ids: await session.execute(delete(ToolCall).where(ToolCall.message_id.in_(msg_ids))) await session.execute(delete(Message).where(Message.id.in_(msg_ids))) await session.delete(c) await session.commit() return True async def history(conv_id: str, limit: int = 40) -> list[Message]: async with SessionLocal() as session: rows = (await session.execute( select(Message).where(Message.conversation_id == conv_id) .options(selectinload(Message.tool_calls)) .order_by(Message.created_at.desc()).limit(limit))).scalars() return list(reversed(list(rows))) async def add_message(conv_id: str, role: str, content: str, attachments: list[str] | None = None, **fields: Any) -> Message: async with SessionLocal() as session: m = Message(conversation_id=conv_id, role=role, content=content, attachments=attachments or [], **fields) session.add(m) c = await session.get(Conversation, conv_id) if c: c.updated_at = utcnow() await session.commit() await session.refresh(m) return m async def save_assistant_message(conv_id: str, message_id: str, content: str, model: str, usage: dict[str, Any], latency_ms: int, tool_calls: list[dict[str, Any]]) -> None: async with SessionLocal() as session: m = Message(id=message_id, conversation_id=conv_id, role="assistant", content=content, model=model, tokens_in=int(usage.get("input_tokens", 0)), tokens_out=int(usage.get("output_tokens", 0)), cost_usd=float(usage.get("cost_usd", 0.0)), latency_ms=latency_ms) session.add(m) for tc in tool_calls: session.add(ToolCall(id=tc["id"], message_id=message_id, name=tc["name"], arguments=tc.get("arguments", {}), result_summary=tc.get("summary", ""), result_payload=tc.get("payload", {}), status=tc.get("status", "ok"), duration_ms=int(tc.get("duration_ms", 0)))) c = await session.get(Conversation, conv_id) if c: c.updated_at = utcnow() await session.commit() async def set_title(conv_id: str, title: str) -> None: async with SessionLocal() as session: c = await session.get(Conversation, conv_id) if c: c.title = title[:120] await session.commit() async def set_feedback(message_id: str, user_id: str, feedback: str | None) -> bool: async with SessionLocal() as session: m = await session.get(Message, message_id) if not m: return False c = await session.get(Conversation, m.conversation_id) if not c or c.user_id != user_id: return False m.feedback = feedback await session.commit() return True async def delete_messages_after(conv_id: str, message_id: str) -> None: """Remove the assistant message (and later ones) for regeneration.""" async with SessionLocal() as session: target = await session.get(Message, message_id) if not target: return ids = [m for m in (await session.execute( select(Message.id).where(Message.conversation_id == conv_id, Message.created_at >= target.created_at))).scalars()] if ids: await session.execute(delete(ToolCall).where(ToolCall.message_id.in_(ids))) await session.execute(delete(Message).where(Message.id.in_(ids))) await session.commit() async def redact_old_messages(months: int) -> int: cutoff = utcnow() - timedelta(days=30 * months) async with SessionLocal() as session: rows = (await session.execute( select(Message).where(Message.created_at < cutoff, Message.content_redacted_at.is_(None)))).scalars() n = 0 for m in rows: m.content = "" m.content_redacted_at = utcnow() n += 1 await session.commit() return n async def count_messages_today(user_id: str) -> int: since = utcnow() - timedelta(days=1) async with SessionLocal() as session: return int(await session.scalar( select(func.count(Message.id)).join(Conversation) .where(Conversation.user_id == user_id, Message.role == "user", Message.created_at >= since)) or 0)