app/services/session_store.py

"""会话持久化服务：sessions + session_messages 表（持久卷 /data/xtc.db）。

手表端可保存对话历史，重装不丢（持久卷保证）。
"""
from __future__ import annotations

import json
import time
import uuid
from typing import Optional

from ..database import get_conn


def _now_ts() -> int:
    return int(time.time())


def create_session(
    *,
    access_key: str,
    title: Optional[str] = None,
    provider: Optional[str] = None,
    model: Optional[str] = None,
) -> dict:
    sid = uuid.uuid4().hex
    now = _now_ts()
    with get_conn() as conn:
        conn.execute(
            "INSERT INTO sessions(id, title, provider, model, access_key, created_at, updated_at) "
            "VALUES(?,?,?,?,?,?,?)",
            (sid, title or "新会话", provider, model, access_key, now, now),
        )
    sess = {"id": sid, "title": title or "新会话", "provider": provider, "model": model, "created_at": now, "updated_at": now}
    # Webhook 通知
    try:
        from ..services import webhook_store
        webhook_store.notify_fire_and_forget("session.created", {"session": sess, "access_key": access_key})
    except Exception:
        pass
    return sess


def list_sessions(
    *,
    access_key: Optional[str] = None,
    limit: int = 50,
    offset: int = 0,
) -> list[dict]:
    sql = "SELECT id, title, provider, model, access_key, created_at, updated_at FROM sessions"
    args: list = []
    if access_key:
        sql += " WHERE access_key = ?"
        args.append(access_key)
    sql += " ORDER BY updated_at DESC LIMIT ? OFFSET ?"
    args += [limit, offset]
    with get_conn() as conn:
        rows = conn.execute(sql, args).fetchall()
    return [dict(r) for r in rows]


def get_session(session_id: str, *, access_key: Optional[str] = None) -> Optional[dict]:
    with get_conn() as conn:
        row = conn.execute(
            "SELECT * FROM sessions WHERE id = ?" + (" AND access_key = ?" if access_key else ""),
            (session_id,) if not access_key else (session_id, access_key),
        ).fetchone()
    return dict(row) if row else None


def append_message(
    *,
    session_id: str,
    role: str,
    content: str,
    thought: Optional[str] = None,
    provider: Optional[str] = None,
    model: Optional[str] = None,
    file_keys: Optional[list[str]] = None,
) -> dict:
    now = _now_ts()
    with get_conn() as conn:
        # 取下一个 seq
        row = conn.execute(
            "SELECT COALESCE(MAX(seq), 0) AS max_seq FROM session_messages WHERE session_id = ?",
            (session_id,),
        ).fetchone()
        next_seq = int(row["max_seq"]) + 1
        conn.execute(
            "INSERT INTO session_messages(session_id, seq, role, content, thought, provider, model, ts, file_keys) "
            "VALUES(?,?,?,?,?,?,?,?,?)",
            (
                session_id,
                next_seq,
                role,
                content,
                thought,
                provider,
                model,
                now,
                json.dumps(file_keys) if file_keys else None,
            ),
        )
        # 更新 session 时间
        new_title = None
        if role == "user" and next_seq == 1:
            # 第一条 user 消息自动作为标题
            new_title = content[:40] + ("..." if len(content) > 40 else "")
            conn.execute(
                "UPDATE sessions SET updated_at = ?, title = ? WHERE id = ?",
                (now, new_title, session_id),
            )
        else:
            conn.execute(
                "UPDATE sessions SET updated_at = ? WHERE id = ?",
                (now, session_id),
            )
    return {
        "session_id": session_id,
        "seq": next_seq,
        "role": role,
        "content": content,
        "thought": thought,
        "provider": provider,
        "model": model,
        "ts": now,
        "file_keys": file_keys,
    }


def list_messages(session_id: str, *, access_key: Optional[str] = None) -> Optional[list[dict]]:
    sess = get_session(session_id, access_key=access_key)
    if not sess:
        return None
    with get_conn() as conn:
        rows = conn.execute(
            "SELECT seq, role, content, thought, provider, model, ts, file_keys FROM session_messages "
            "WHERE session_id = ? ORDER BY seq ASC",
            (session_id,),
        ).fetchall()
    out = []
    for r in rows:
        item = dict(r)
        if r["file_keys"]:
            try:
                item["file_keys"] = json.loads(r["file_keys"])
            except Exception:
                item["file_keys"] = []
        else:
            item["file_keys"] = []
        out.append(item)
    return out


def delete_session(session_id: str, *, access_key: Optional[str] = None) -> bool:
    sess = get_session(session_id, access_key=access_key)
    if not sess:
        return False
    with get_conn() as conn:
        conn.execute("DELETE FROM session_messages WHERE session_id = ?", (session_id,))
        conn.execute("DELETE FROM sessions WHERE id = ?", (session_id,))
    # Webhook 通知
    try:
        from ..services import webhook_store
        webhook_store.notify_fire_and_forget("session.deleted", {"session_id": session_id, "access_key": access_key})
    except Exception:
        pass
    return True


def update_session(
    session_id: str,
    *,
    title: Optional[str] = None,
    access_key: Optional[str] = None,
) -> Optional[dict]:
    sess = get_session(session_id, access_key=access_key)
    if not sess:
        return None
    sets = []
    args: list = []
    if title is not None:
        sets.append("title = ?")
        args.append(title)
    if not sets:
        return sess
    sets.append("updated_at = ?")
    args.append(_now_ts())
    args.append(session_id)
    with get_conn() as conn:
        conn.execute(
            f"UPDATE sessions SET {', '.join(sets)} WHERE id = ?",
            tuple(args),
        )
    return get_session(session_id)