feat(gateway): persist chat sessions and messages in PostgreSQL

- store: new chat_sessions/chat_messages tables with CRUD, ILIKE message
  search, and auto-titling from the first user message
- context: optional Persister on the sliding-window store — messages are
  saved asynchronously on Add, sessions not in memory are restored from
  the database on Get
- orchestrator: ProcessInput/FinishRound now carry userID so persisted
  messages have an owner
- api: real session handlers (list/create/delete), message history and
  clear endpoints, working message search and three-format export
  (markdown/txt/json), fixing the short-ID panic in export
- main: migration, store wiring, ctxStore persister adapter

Tests: chat store handler lifecycle (fake store), context persister
save/restore, stub auth login flow updated.
This commit is contained in:
Aska committed 2026-08-20 20:34:08 +08:00
1 parent f44bf26050
commit f8196c111c
11 files changed
+787 -27

No files matched your search

+47 -1
View File
@@ -10,12 +10,23 @@ import (
"git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model"
)
// Persister saves conversation messages durably and can restore them
// (e.g. from PostgreSQL) when a session is not in memory.
type Persister interface {
// SaveMessage persists one message. Called synchronously by Add —
// implementations should be fast or dispatch async work themselves.
SaveMessage(sessionID, userID string, msg model.Message)
// LoadRecent returns the most recent messages of a session, ascending.
LoadRecent(sessionID string, limit int) []model.Message
}
// Store holds conversation history with a sliding window.
type Store struct {
mu sync.RWMutex
sessions map[string]*SessionHistory
maxTokens int // Maximum tokens to include in context
maxTurns int // Maximum message turns to keep
persister Persister
}
// SessionHistory holds messages for a single session.
@@ -39,14 +50,31 @@ func NewStore(maxTokens, maxTurns int) *Store {
}
}
// SetPersister injects durable persistence for conversation messages.
func (s *Store) SetPersister(p Persister) {
s.mu.Lock()
s.persister = p
s.mu.Unlock()
}
// Add appends a message to the session history.
func (s *Store) Add(sessionID string, msg model.Message) {
s.add(sessionID, "", msg)
}
// AddWithUser appends a message and records the owning user for persistence.
func (s *Store) AddWithUser(sessionID, userID string, msg model.Message) {
s.add(sessionID, userID, msg)
}
func (s *Store) add(sessionID, userID string, msg model.Message) {
s.mu.Lock()
sh, ok := s.sessions[sessionID]
if !ok {
sh = &SessionHistory{}
s.sessions[sessionID] = sh
}
persister := s.persister
s.mu.Unlock()
sh.mu.Lock()
@@ -56,15 +84,33 @@ func (s *Store) Add(sessionID string, msg model.Message) {
sh.Messages = sh.Messages[len(sh.Messages)-s.maxTurns:]
}
sh.mu.Unlock()
// Persist asynchronously — never block the conversation on the database.
if persister != nil && (msg.Role == "user" || msg.Role == "assistant") {
go persister.SaveMessage(sessionID, userID, msg)
}
}
// Get retrieves recent messages for a session, trimmed to fit token budget.
// Sessions missing from memory are restored from the persister if available.
func (s *Store) Get(sessionID string) []model.Message {
s.mu.RLock()
sh, ok := s.sessions[sessionID]
persister := s.persister
s.mu.RUnlock()
if !ok {
return nil
if persister == nil {
return nil
}
loaded := persister.LoadRecent(sessionID, s.maxTurns)
if len(loaded) == 0 {
return nil
}
sh = &SessionHistory{Messages: loaded}
s.mu.Lock()
s.sessions[sessionID] = sh
s.mu.Unlock()
}
sh.mu.RLock()
@@ -0,0 +1,98 @@
package context
import (
"sync"
"testing"
"time"
"git.yeij.top/AskaEth/Cyrene_v2/services/ai-core/model"
)
// fakePersister records saved messages and can restore them.
type fakePersister struct {
mu sync.Mutex
saved map[string][]model.Message
}
func newFakePersister() *fakePersister {
return &fakePersister{saved: map[string][]model.Message{}}
}
func (f *fakePersister) SaveMessage(sessionID, userID string, msg model.Message) {
f.mu.Lock()
defer f.mu.Unlock()
f.saved[sessionID] = append(f.saved[sessionID], msg)
}
func (f *fakePersister) LoadRecent(sessionID string, limit int) []model.Message {
f.mu.Lock()
defer f.mu.Unlock()
return f.saved[sessionID]
}
func (f *fakePersister) count(sessionID string) int {
f.mu.Lock()
defer f.mu.Unlock()
return len(f.saved[sessionID])
}
func TestPersisterSaveOnAdd(t *testing.T) {
s := NewStore(4000, 50)
p := newFakePersister()
s.SetPersister(p)
s.AddWithUser("s1", "u1", model.Message{Role: "user", Content: "你好"})
// SaveMessage is dispatched asynchronously — wait briefly
for i := 0; i < 50 && p.count("s1") == 0; i++ {
time.Sleep(10 * time.Millisecond)
}
if p.count("s1") != 1 {
t.Fatalf("expected 1 persisted message, got %d", p.count("s1"))
}
}
func TestPersisterSkipsNonChatRoles(t *testing.T) {
s := NewStore(4000, 50)
p := newFakePersister()
s.SetPersister(p)
s.Add("s1", model.Message{Role: "system", Content: "上下文块"})
time.Sleep(50 * time.Millisecond)
if p.count("s1") != 0 {
t.Fatalf("system messages should not be persisted, got %d", p.count("s1"))
}
}
func TestRestoreFromPersisterOnGet(t *testing.T) {
s := NewStore(4000, 50)
p := newFakePersister()
p.saved["s2"] = []model.Message{
{Role: "user", Content: "以前说过的话"},
{Role: "assistant", Content: "以前的回复"},
}
s.SetPersister(p)
// Session not in memory — should restore from persister
msgs := s.Get("s2")
if len(msgs) != 2 {
t.Fatalf("expected 2 restored messages, got %d", len(msgs))
}
if msgs[0].Content != "以前说过的话" {
t.Errorf("unexpected restored content: %q", msgs[0].Content)
}
// Second Get should hit memory, not reload
p.saved["s2"] = nil
msgs = s.Get("s2")
if len(msgs) != 2 {
t.Errorf("expected cached messages after restore, got %d", len(msgs))
}
}
func TestGetWithoutPersister(t *testing.T) {
s := NewStore(4000, 50)
if msgs := s.Get("missing"); msgs != nil {
t.Errorf("expected nil for unknown session, got %v", msgs)
}
}
@@ -79,7 +79,7 @@ func (p *Pipeline) SetToolExecutor(t ToolExecutor) {
// ProcessInput processes a user message through the full pipeline.
// Returns a channel of StreamToken for the LLM response.
func (p *Pipeline) ProcessInput(ctx context.Context, sessionID, userMessage string) (<-chan model.StreamToken, *model.Intent, error) {
func (p *Pipeline) ProcessInput(ctx context.Context, sessionID, userID, userMessage string) (<-chan model.StreamToken, *model.Intent, error) {
// 1. Cache check
if cached := p.cache.Get(userMessage); cached != nil {
ch := make(chan model.StreamToken, 1)
@@ -122,7 +122,7 @@ func (p *Pipeline) ProcessInput(ctx context.Context, sessionID, userMessage stri
}
// 7. Store user message
p.ctxStore.Add(sessionID, model.Message{Role: "user", Content: userMessage})
p.ctxStore.AddWithUser(sessionID, userID, model.Message{Role: "user", Content: userMessage})
return tokenCh, intent, nil
}
@@ -193,7 +193,7 @@ func (p *Pipeline) streamWithTools(ctx context.Context, adapter *llm.Adapter, me
// after the response stream completes.
func (p *Pipeline) FinishRound(sessionID, userID, assistantReply string) {
if assistantReply != "" {
p.ctxStore.Add(sessionID, model.Message{Role: "assistant", Content: assistantReply})
p.ctxStore.AddWithUser(sessionID, userID, model.Message{Role: "assistant", Content: assistantReply})
}
p.PostProcess(sessionID, userID)
}
@@ -87,7 +87,7 @@ func TestToolCallLoop(t *testing.T) {
exec := &fakeToolExec{}
p.SetToolExecutor(exec)
tokenCh, _, err := p.ProcessInput(context.Background(), "s1", "打开客厅灯")
tokenCh, _, err := p.ProcessInput(context.Background(), "s1", "u1", "打开客厅灯")
if err != nil {
t.Fatalf("ProcessInput: %v", err)
}