diff --git a/common/database.go b/common/database.go index 1d1a04d2f190..efdfcd8c04a4 100644 --- a/common/database.go +++ b/common/database.go @@ -43,8 +43,16 @@ func UsingLogDatabase(databaseType DatabaseType) bool { var chatLogDatabaseType = DatabaseTypeSQLite +func ChatLogDatabaseType() DatabaseType { + return chatLogDatabaseType +} + func SetChatLogDatabaseType(databaseType DatabaseType) { chatLogDatabaseType = databaseType } +func UsingChatLogDatabase(databaseType DatabaseType) bool { + return chatLogDatabaseType == databaseType +} + var SQLitePath = "one-api.db?_busy_timeout=30000" diff --git a/controller/chat_log.go b/controller/chat_log.go index 14e70c654f16..d3ff96e41e39 100644 --- a/controller/chat_log.go +++ b/controller/chat_log.go @@ -8,6 +8,42 @@ import ( "github.com/gin-gonic/gin" ) +type chatSessionMeta struct { + Id int `json:"id"` + TokenId int `json:"token_id"` + UserId int `json:"user_id"` + ModelName string `json:"model_name"` + TurnCount int `json:"turn_count"` + MessageCount int `json:"message_count"` + CreatedAt int64 `json:"created_at"` + LastActiveAt int64 `json:"last_active_at"` +} + +func toChatSessionMeta(s *model.ChatSession) chatSessionMeta { + return chatSessionMeta{ + Id: s.Id, TokenId: s.TokenId, UserId: s.UserId, ModelName: s.ModelName, + TurnCount: s.TurnCount, MessageCount: s.MessageCount, + CreatedAt: s.CreatedAt, LastActiveAt: s.LastActiveAt, + } +} + +// normalizeChatLogPageLimit clamps the page-size query param once, before the +// hot-cache and database paths branch, so both see the same default (20) and +// ceiling (100). +func normalizeChatLogPageLimit(raw string) int { + limit, _ := strconv.Atoi(raw) + if limit < 1 { + limit = 20 + } + if limit > 100 { + limit = 100 + } + return limit +} + +// AdminGetChatSessions lists sessions newest-first using cursor pagination. +// The unfiltered first page is served from the hot cache when possible; every +// other query falls through to keyset queries on the chat-log database. func AdminGetChatSessions(c *gin.Context) { if !model.ChatLogDBEnabled() { c.JSON(http.StatusOK, gin.H{"success": false, "message": "对话详情库未配置"}) @@ -16,35 +52,60 @@ func AdminGetChatSessions(c *gin.Context) { tokenId, _ := strconv.Atoi(c.Query("token_id")) userId, _ := strconv.Atoi(c.Query("user_id")) modelName := c.Query("model_name") - page, _ := strconv.Atoi(c.Query("page")) - pageSize, _ := strconv.Atoi(c.Query("page_size")) + startTs, _ := strconv.ParseInt(c.Query("start_ts"), 10, 64) + endTs, _ := strconv.ParseInt(c.Query("end_ts"), 10, 64) + limit := normalizeChatLogPageLimit(c.Query("limit")) + cursorStr := c.Query("cursor") - sessions, total, err := model.SearchChatSessions(tokenId, userId, modelName, page, pageSize) - if err != nil { - c.JSON(http.StatusOK, gin.H{"success": false, "message": err.Error()}) - return + filter := model.ChatSessionFilter{ + TokenId: tokenId, UserId: userId, ModelName: modelName, + StartTs: startTs, EndTs: endTs, } - type chatSessionMeta struct { - Id int `json:"id"` - TokenId int `json:"token_id"` - UserId int `json:"user_id"` - ModelName string `json:"model_name"` - TurnCount int `json:"turn_count"` - MessageCount int `json:"message_count"` - CreatedAt int64 `json:"created_at"` - LastActiveAt int64 `json:"last_active_at"` + + var sessions []*model.ChatSession + var hasMore bool + servedFromHot := false + if filter.Empty() && cursorStr == "" { + // 近期会话默认视图：优先走内存热缓存，未命中或刷新失败再回源数据库 + if hot, hotHasMore, ok := model.ListRecentChatSessions(limit); ok { + sessions, hasMore, servedFromHot = hot, hotHasMore, true + } + } + if !servedFromHot { + cursor, err := model.DecodeChatSessionCursor(cursorStr) + if err != nil { + c.JSON(http.StatusBadRequest, gin.H{"success": false, "message": "无效的分页游标"}) + return + } + sessions, hasMore, err = model.ListChatSessions(filter, cursor, limit) + if err != nil { + c.JSON(http.StatusOK, gin.H{"success": false, "message": err.Error()}) + return + } } - out := make([]chatSessionMeta, 0, len(sessions)) + + items := make([]chatSessionMeta, 0, len(sessions)) for _, s := range sessions { - out = append(out, chatSessionMeta{ - Id: s.Id, TokenId: s.TokenId, UserId: s.UserId, ModelName: s.ModelName, - TurnCount: s.TurnCount, MessageCount: s.MessageCount, - CreatedAt: s.CreatedAt, LastActiveAt: s.LastActiveAt, - }) + items = append(items, toChatSessionMeta(s)) } - c.JSON(http.StatusOK, gin.H{"success": true, "data": out, "total": total}) + var nextCursor any + if hasMore && len(sessions) > 0 { + last := sessions[len(sessions)-1] + nextCursor = model.ChatSessionCursor{LastActiveAt: last.LastActiveAt, Id: last.Id}.Encode() + } + c.JSON(http.StatusOK, gin.H{ + "success": true, + "data": gin.H{ + "items": items, + "has_more": hasMore, + "next_cursor": nextCursor, + }, + }) } +// AdminGetChatSessionDetail returns a session's metadata plus one page of +// turns (ascending). The newest page is served from the hot cache when +// available; older pages cold-load from the database via before_id. func AdminGetChatSessionDetail(c *gin.Context) { if !model.ChatLogDBEnabled() { c.JSON(http.StatusOK, gin.H{"success": false, "message": "对话详情库未配置"}) @@ -55,15 +116,45 @@ func AdminGetChatSessionDetail(c *gin.Context) { c.JSON(http.StatusBadRequest, gin.H{"success": false, "message": "无效 ID"}) return } + limit := normalizeChatLogPageLimit(c.Query("limit")) + beforeId, _ := strconv.Atoi(c.Query("before_id")) + session, err := model.GetChatSessionById(id) if err != nil { c.JSON(http.StatusNotFound, gin.H{"success": false, "message": "会话不存在"}) return } - turns, err := model.GetChatTurnsBySessionId(id) - if err != nil { - c.JSON(http.StatusOK, gin.H{"success": false, "message": err.Error()}) - return + + var turns []*model.ChatTurn + var hasMore bool + servedFromHot := false + if beforeId == 0 { + if hot, hotHasMore, ok := model.GetHotChatTurns(id, session.TurnCount, limit); ok { + turns, hasMore, servedFromHot = hot, hotHasMore, true + } + } + if !servedFromHot { + turns, hasMore, err = model.GetChatTurnsPage(id, beforeId, limit) + if err != nil { + c.JSON(http.StatusOK, gin.H{"success": false, "message": err.Error()}) + return + } + if beforeId == 0 { + model.AdmitChatTurns(id, turns, session.TurnCount) + } + } + + var nextTurnId any + if hasMore && len(turns) > 0 { + nextTurnId = turns[0].Id } - c.JSON(http.StatusOK, gin.H{"success": true, "data": gin.H{"session": session, "turns": turns}}) + c.JSON(http.StatusOK, gin.H{ + "success": true, + "data": gin.H{ + "session": session, + "turns": turns, + "has_more": hasMore, + "next_turn_id": nextTurnId, + }, + }) } diff --git a/controller/chat_log_test.go b/controller/chat_log_test.go index 64a2f80df849..e640a56a293b 100644 --- a/controller/chat_log_test.go +++ b/controller/chat_log_test.go @@ -4,6 +4,7 @@ import ( "fmt" "net/http" "net/http/httptest" + "os" "strconv" "testing" @@ -32,6 +33,25 @@ func serveChatLog(t *testing.T, route string, handler gin.HandlerFunc, url strin return rec } +type chatSessionListResponse struct { + Success bool `json:"success"` + Data struct { + Items []map[string]any `json:"items"` + HasMore bool `json:"has_more"` + NextCursor any `json:"next_cursor"` + } `json:"data"` +} + +func listChatSessionsFor(t *testing.T, query string) chatSessionListResponse { + t.Helper() + rec := serveChatLog(t, "/chat_logs/sessions", AdminGetChatSessions, "/chat_logs/sessions"+query) + require.Equal(t, http.StatusOK, rec.Code) + var resp chatSessionListResponse + require.NoError(t, common.Unmarshal(rec.Body.Bytes(), &resp)) + require.True(t, resp.Success) + return resp +} + func TestAdminGetChatSessions_ListMetaShape(t *testing.T) { setupChatLogTestDB(t) s := &model.ChatSession{ @@ -40,20 +60,12 @@ func TestAdminGetChatSessions_ListMetaShape(t *testing.T) { } require.NoError(t, s.Insert()) - rec := serveChatLog(t, "/chat_logs/sessions", AdminGetChatSessions, "/chat_logs/sessions?page=1&page_size=10") - require.Equal(t, http.StatusOK, rec.Code) - - var resp struct { - Success bool `json:"success"` - Total int64 `json:"total"` - Data []map[string]any `json:"data"` - } - require.NoError(t, common.Unmarshal(rec.Body.Bytes(), &resp)) - require.True(t, resp.Success) - require.Equal(t, int64(1), resp.Total) - require.Len(t, resp.Data, 1) + resp := listChatSessionsFor(t, "?limit=10") + require.Len(t, resp.Data.Items, 1) + assert.False(t, resp.Data.HasMore) + assert.Nil(t, resp.Data.NextCursor) - meta := resp.Data[0] + meta := resp.Data.Items[0] for _, key := range []string{"id", "token_id", "user_id", "model_name", "turn_count", "message_count", "created_at", "last_active_at"} { assert.Contains(t, meta, key) } @@ -64,78 +76,143 @@ func TestAdminGetChatSessions_ListMetaShape(t *testing.T) { assert.Equal(t, float64(2), meta["turn_count"]) } -func TestAdminGetChatSessions_FiltersAndPaging(t *testing.T) { +func TestNormalizeChatLogPageLimit(t *testing.T) { + assert.Equal(t, 20, normalizeChatLogPageLimit("")) + assert.Equal(t, 20, normalizeChatLogPageLimit("0")) + assert.Equal(t, 20, normalizeChatLogPageLimit("-3")) + assert.Equal(t, 20, normalizeChatLogPageLimit("abc")) + assert.Equal(t, 1, normalizeChatLogPageLimit("1")) + assert.Equal(t, 50, normalizeChatLogPageLimit("50")) + assert.Equal(t, 100, normalizeChatLogPageLimit("100")) + assert.Equal(t, 100, normalizeChatLogPageLimit("5000")) +} + +// TestAdminGetChatSessions_DefaultLimitOnHotPath pins the fix for the missing +// limit param: the hot path must see the same default (20) as the DB path +// instead of clamping to a single item. +func TestAdminGetChatSessions_DefaultLimitOnHotPath(t *testing.T) { setupChatLogTestDB(t) - for i, m := range []string{"gpt-4", "gpt-4", "claude-3"} { - s := &model.ChatSession{TokenId: i + 1, UserId: 1, ModelName: m, PrefixHash: fmt.Sprintf("h%d", i)} + t.Setenv("CHAT_LOG_HOT_CACHE_ENABLED", "true") + model.InitChatLogHotCache() + t.Cleanup(func() { + os.Setenv("CHAT_LOG_HOT_CACHE_ENABLED", "false") + model.InitChatLogHotCache() + }) + + for i := 0; i < 3; i++ { + s := &model.ChatSession{TokenId: i + 1, ModelName: "gpt-4", PrefixHash: fmt.Sprintf("h%d", i)} require.NoError(t, s.Insert()) } - list := func(query string) (int64, []map[string]any) { - rec := serveChatLog(t, "/chat_logs/sessions", AdminGetChatSessions, "/chat_logs/sessions"+query) - require.Equal(t, http.StatusOK, rec.Code) - var resp struct { - Success bool `json:"success"` - Total int64 `json:"total"` - Data []map[string]any `json:"data"` - } - require.NoError(t, common.Unmarshal(rec.Body.Bytes(), &resp)) - require.True(t, resp.Success) - return resp.Total, resp.Data - } - - total, data := list("?page=1&page_size=2") - assert.Equal(t, int64(3), total) - assert.Len(t, data, 2) - - total, data = list("?page=1&page_size=10&model_name=gpt-4") - assert.Equal(t, int64(2), total) - require.Len(t, data, 2) - assert.Equal(t, "gpt-4", data[0]["model_name"]) - assert.Equal(t, "gpt-4", data[1]["model_name"]) - - total, data = list("?page=1&page_size=10&token_id=3") - assert.Equal(t, int64(1), total) - require.Len(t, data, 1) - assert.Equal(t, float64(3), data[0]["token_id"]) + // no limit param at all: default page of 20, all three sessions returned + resp := listChatSessionsFor(t, "") + require.Len(t, resp.Data.Items, 3) +} - total, data = list("?page=1&page_size=10&model_name=claude-3&token_id=3") - assert.Equal(t, int64(1), total) - assert.Len(t, data, 1) +func TestAdminGetChatSessions_FiltersAndCursorPaging(t *testing.T) { + setupChatLogTestDB(t) + now := common.GetTimestamp() + past := now - 3600 + for i, m := range []string{"gpt-4", "gpt-4", "claude-3"} { + s := &model.ChatSession{TokenId: i + 1, UserId: 1, ModelName: m, PrefixHash: fmt.Sprintf("h%d", i), LastActiveAt: past} + require.NoError(t, s.Insert()) + } - total, _ = list("?page=1&page_size=10&model_name=none") - assert.Equal(t, int64(0), total) + // first page with cursor pagination + resp := listChatSessionsFor(t, "?limit=2") + assert.Len(t, resp.Data.Items, 2) + assert.True(t, resp.Data.HasMore) + cursor, ok := resp.Data.NextCursor.(string) + require.True(t, ok) + require.NotEmpty(t, cursor) + + // walk to the older page + resp = listChatSessionsFor(t, "?limit=2&cursor="+cursor) + assert.Len(t, resp.Data.Items, 1) + assert.False(t, resp.Data.HasMore) + assert.Nil(t, resp.Data.NextCursor) + + // filters + resp = listChatSessionsFor(t, "?limit=10&model_name=gpt-4") + require.Len(t, resp.Data.Items, 2) + assert.Equal(t, "gpt-4", resp.Data.Items[0]["model_name"]) + assert.Equal(t, "gpt-4", resp.Data.Items[1]["model_name"]) + + resp = listChatSessionsFor(t, "?limit=10&token_id=3") + require.Len(t, resp.Data.Items, 1) + assert.Equal(t, float64(3), resp.Data.Items[0]["token_id"]) + + resp = listChatSessionsFor(t, "?limit=10&model_name=claude-3&token_id=3") + assert.Len(t, resp.Data.Items, 1) + + resp = listChatSessionsFor(t, "?limit=10&model_name=none") + assert.Empty(t, resp.Data.Items) + assert.False(t, resp.Data.HasMore) + + // time range filter (inclusive on last_active_at) + newer := &model.ChatSession{TokenId: 9, ModelName: "gpt-4", PrefixHash: "h9", LastActiveAt: now} + require.NoError(t, newer.Insert()) + resp = listChatSessionsFor(t, fmt.Sprintf("?limit=10&start_ts=%d", now)) + require.Len(t, resp.Data.Items, 1) + assert.Equal(t, float64(9), resp.Data.Items[0]["token_id"]) + resp = listChatSessionsFor(t, fmt.Sprintf("?limit=10&end_ts=%d", now-1)) + assert.Len(t, resp.Data.Items, 3) + + // invalid cursor is rejected ("Zm9vYmFy" decodes to "foobar", wrong shape) + rec := serveChatLog(t, "/chat_logs/sessions", AdminGetChatSessions, "/chat_logs/sessions?cursor=Zm9vYmFy") + assert.Equal(t, http.StatusBadRequest, rec.Code) } -func TestAdminGetChatSessionDetail_TurnsOrdered(t *testing.T) { +func TestAdminGetChatSessionDetail_TurnsPagedAndOrdered(t *testing.T) { setupChatLogTestDB(t) s := &model.ChatSession{TokenId: 7, UserId: 1, ModelName: "gpt-4", PrefixHash: "h7"} require.NoError(t, s.Insert()) - turn2 := &model.ChatTurn{SessionId: s.Id, TurnIndex: 2, RequestId: "r2", ModelName: "gpt-4", NewMessages: `[{"role":"assistant"}]`, ResponseBody: `{"b":2}`} - require.NoError(t, turn2.Insert()) turn1 := &model.ChatTurn{SessionId: s.Id, TurnIndex: 1, RequestId: "r1", ModelName: "gpt-4", NewMessages: `[{"role":"user"}]`, ResponseBody: `{"a":1}`} require.NoError(t, turn1.Insert()) + turn2 := &model.ChatTurn{SessionId: s.Id, TurnIndex: 2, RequestId: "r2", ModelName: "gpt-4", NewMessages: `[{"role":"assistant"}]`, ResponseBody: `{"b":2}`} + require.NoError(t, turn2.Insert()) - rec := serveChatLog(t, "/chat_logs/sessions/:id", AdminGetChatSessionDetail, "/chat_logs/sessions/"+strconv.Itoa(s.Id)) - require.Equal(t, http.StatusOK, rec.Code) - - var resp struct { - Success bool `json:"success"` - Data struct { - Session *model.ChatSession `json:"session"` - Turns []*model.ChatTurn `json:"turns"` - } `json:"data"` + fetch := func(query string) (session *model.ChatSession, turns []*model.ChatTurn, hasMore bool, nextTurnId any) { + rec := serveChatLog(t, "/chat_logs/sessions/:id", AdminGetChatSessionDetail, "/chat_logs/sessions/"+strconv.Itoa(s.Id)+query) + require.Equal(t, http.StatusOK, rec.Code) + var resp struct { + Success bool `json:"success"` + Data struct { + Session *model.ChatSession `json:"session"` + Turns []*model.ChatTurn `json:"turns"` + HasMore bool `json:"has_more"` + NextTurnId any `json:"next_turn_id"` + } `json:"data"` + } + require.NoError(t, common.Unmarshal(rec.Body.Bytes(), &resp)) + require.True(t, resp.Success) + return resp.Data.Session, resp.Data.Turns, resp.Data.HasMore, resp.Data.NextTurnId } - require.NoError(t, common.Unmarshal(rec.Body.Bytes(), &resp)) - require.True(t, resp.Success) - require.NotNil(t, resp.Data.Session) - assert.Equal(t, s.Id, resp.Data.Session.Id) - require.Len(t, resp.Data.Turns, 2) - assert.Equal(t, 1, resp.Data.Turns[0].TurnIndex) - assert.Equal(t, "r1", resp.Data.Turns[0].RequestId) - assert.Equal(t, `{"a":1}`, resp.Data.Turns[0].ResponseBody) - assert.Equal(t, 2, resp.Data.Turns[1].TurnIndex) - assert.Equal(t, "r2", resp.Data.Turns[1].RequestId) + + // default view: all turns ascending by insertion id + session, turns, hasMore, nextTurnId := fetch("") + require.NotNil(t, session) + assert.Equal(t, s.Id, session.Id) + require.Len(t, turns, 2) + assert.False(t, hasMore) + assert.Nil(t, nextTurnId) + assert.Equal(t, "r1", turns[0].RequestId) + assert.Equal(t, `{"a":1}`, turns[0].ResponseBody) + assert.Equal(t, "r2", turns[1].RequestId) + + // newest page only; next_turn_id points at the oldest turn of the page + _, turns, hasMore, nextTurnId = fetch("?limit=1") + require.Len(t, turns, 1) + assert.Equal(t, "r2", turns[0].RequestId) + assert.True(t, hasMore) + require.Equal(t, float64(turn2.Id), nextTurnId) + + // older page via before_id + _, turns, hasMore, nextTurnId = fetch("?limit=1&before_id=" + strconv.Itoa(turn2.Id)) + require.Len(t, turns, 1) + assert.Equal(t, "r1", turns[0].RequestId) + assert.False(t, hasMore) + assert.Nil(t, nextTurnId) // nonexistent id -> 404 rec404 := serveChatLog(t, "/chat_logs/sessions/:id", AdminGetChatSessionDetail, "/chat_logs/sessions/99999") diff --git a/model/chat_log.go b/model/chat_log.go index 1f308757a980..7747223e8b51 100644 --- a/model/chat_log.go +++ b/model/chat_log.go @@ -1,16 +1,22 @@ package model import ( + "encoding/base64" + "fmt" + "strconv" + "strings" + "github.com/QuantumNous/new-api/common" ) type ChatSession struct { - Id int `json:"id" gorm:"primaryKey"` - TokenId int `json:"token_id" gorm:"index;uniqueIndex:idx_chat_sessions_token_prefix"` - UserId int `json:"user_id" gorm:"index"` - ModelName string `json:"model_name" gorm:"type:varchar(128)"` - System string `json:"system" gorm:"type:text"` - TurnCount int `json:"turn_count"` + Id int `json:"id" gorm:"primaryKey"` + TokenId int `json:"token_id" gorm:"index;uniqueIndex:idx_chat_sessions_token_prefix"` + UserId int `json:"user_id" gorm:"index"` + ModelName string `json:"model_name" gorm:"type:varchar(128);index"` + System string `json:"system" gorm:"type:text"` + TurnCount int `json:"turn_count"` + // MessageCount is the number of request messages covered by PrefixHash. MessageCount int `json:"message_count"` PrefixHash string `json:"prefix_hash" gorm:"type:varchar(64);uniqueIndex:idx_chat_sessions_token_prefix"` CreatedAt int64 `json:"created_at" gorm:"bigint;index"` @@ -33,13 +39,17 @@ func (s *ChatSession) Insert() error { if s.LastActiveAt == 0 { s.LastActiveAt = now } - return CHATLOG_DB.Create(s).Error + if err := CHATLOG_DB.Create(s).Error; err != nil { + return err + } + chatLogCacheRecordSession(s) + return nil } type ChatTurn struct { Id int `json:"id" gorm:"primaryKey"` - SessionId int `json:"session_id" gorm:"index"` - TurnIndex int `json:"turn_index"` + SessionId int `json:"session_id" gorm:"index:idx_chat_turns_session_turn,priority:1"` + TurnIndex int `json:"turn_index" gorm:"index:idx_chat_turns_session_turn,priority:2"` RequestId string `json:"request_id" gorm:"type:varchar(64);index"` ModelName string `json:"model_name" gorm:"type:varchar(128)"` ChannelId int `json:"channel_id" gorm:"index"` @@ -59,13 +69,11 @@ func (t *ChatTurn) Insert() error { if t.CreatedAt == 0 { t.CreatedAt = common.GetTimestamp() } - return CHATLOG_DB.Create(t).Error -} - -func GetChatTurnsBySessionId(sessionId int) ([]*ChatTurn, error) { - var turns []*ChatTurn - err := CHATLOG_DB.Where("session_id = ?", sessionId).Order("turn_index asc").Find(&turns).Error - return turns, err + if err := CHATLOG_DB.Create(t).Error; err != nil { + return err + } + chatLogCacheRecordTurn(t) + return nil } func FindChatSessionByPrefixHashes(tokenId int, hashes []string) (*ChatSession, error) { @@ -82,41 +90,108 @@ func (s *ChatSession) Advance(modelName string, at int64) error { s.TurnCount++ s.LastActiveAt = at s.ModelName = modelName - return CHATLOG_DB.Model(s).Updates(map[string]any{ + if err := CHATLOG_DB.Model(s).Updates(map[string]any{ "turn_count": s.TurnCount, "message_count": s.MessageCount, "prefix_hash": s.PrefixHash, "last_active_at": s.LastActiveAt, "model_name": s.ModelName, - }).Error + }).Error; err != nil { + return err + } + chatLogCacheAdvanceSession(s) + return nil +} + +// ChatSessionCursor is the keyset position of a session in the newest-first +// (last_active_at desc, id desc) session list. +type ChatSessionCursor struct { + LastActiveAt int64 + Id int } -func SearchChatSessions(tokenId, userId int, modelName string, page, pageSize int) ([]*ChatSession, int64, error) { - if page < 1 { - page = 1 +func (c ChatSessionCursor) Encode() string { + raw := fmt.Sprintf("%d:%d", c.LastActiveAt, c.Id) + return base64.URLEncoding.EncodeToString([]byte(raw)) +} + +func DecodeChatSessionCursor(s string) (ChatSessionCursor, error) { + if s == "" { + return ChatSessionCursor{}, nil } - if pageSize < 1 || pageSize > 100 { - pageSize = 20 + raw, err := base64.URLEncoding.DecodeString(s) + if err != nil { + return ChatSessionCursor{}, err + } + parts := strings.Split(string(raw), ":") + if len(parts) != 2 { + return ChatSessionCursor{}, fmt.Errorf("invalid chat session cursor") + } + at, err := strconv.ParseInt(parts[0], 10, 64) + if err != nil { + return ChatSessionCursor{}, err + } + id, err := strconv.Atoi(parts[1]) + if err != nil { + return ChatSessionCursor{}, err + } + return ChatSessionCursor{LastActiveAt: at, Id: id}, nil +} + +// ChatSessionFilter narrows the session list. Zero values mean "no filter". +type ChatSessionFilter struct { + TokenId int + UserId int + ModelName string + // StartTs/EndTs bound last_active_at (unix seconds), inclusive. + StartTs int64 + EndTs int64 +} + +func (f ChatSessionFilter) Empty() bool { + return f.TokenId == 0 && f.UserId == 0 && f.ModelName == "" && f.StartTs == 0 && f.EndTs == 0 +} + +// ListChatSessions returns one keyset page of sessions, newest first. An empty +// cursor fetches the first page. hasMore reports whether older sessions may +// follow, so callers never need a full-table COUNT. +func ListChatSessions(filter ChatSessionFilter, cursor ChatSessionCursor, limit int) (sessions []*ChatSession, hasMore bool, err error) { + if limit < 1 { + limit = 20 + } + if limit > 100 { + limit = 100 } q := CHATLOG_DB.Model(&ChatSession{}) - if tokenId > 0 { - q = q.Where("token_id = ?", tokenId) + if filter.TokenId > 0 { + q = q.Where("token_id = ?", filter.TokenId) + } + if filter.UserId > 0 { + q = q.Where("user_id = ?", filter.UserId) + } + if filter.ModelName != "" { + q = q.Where("model_name = ?", filter.ModelName) } - if userId > 0 { - q = q.Where("user_id = ?", userId) + if filter.StartTs > 0 { + q = q.Where("last_active_at >= ?", filter.StartTs) } - if modelName != "" { - q = q.Where("model_name = ?", modelName) + if filter.EndTs > 0 { + q = q.Where("last_active_at <= ?", filter.EndTs) } - var total int64 - if err := q.Count(&total).Error; err != nil { - return nil, 0, err + if cursor.Id > 0 || cursor.LastActiveAt > 0 { + // expanded row-value comparison so every supported database can use the + // last_active_at index (MySQL 5.7 cannot optimize (a,b) < (?,?)) + q = q.Where("last_active_at < ? OR (last_active_at = ? AND id < ?)", + cursor.LastActiveAt, cursor.LastActiveAt, cursor.Id) } - var sessions []*ChatSession - if err := q.Order("last_active_at desc").Offset((page - 1) * pageSize).Limit(pageSize).Find(&sessions).Error; err != nil { - return nil, 0, err + if err := q.Order("last_active_at desc, id desc").Limit(limit + 1).Find(&sessions).Error; err != nil { + return nil, false, err } - return sessions, total, nil + if len(sessions) > limit { + sessions = sessions[:limit] + hasMore = true + } + return sessions, hasMore, nil } func GetChatSessionById(id int) (*ChatSession, error) { @@ -124,3 +199,31 @@ func GetChatSessionById(id int) (*ChatSession, error) { err := CHATLOG_DB.First(&s, "id = ?", id).Error return &s, err } + +// GetChatTurnsPage returns one page of a session's turns in ascending id +// order. With beforeId == 0 it returns the newest page; otherwise the page +// older than beforeId. hasMore reports whether even older turns exist. +func GetChatTurnsPage(sessionId int, beforeId int, limit int) (turns []*ChatTurn, hasMore bool, err error) { + if limit < 1 { + limit = 20 + } + if limit > 100 { + limit = 100 + } + q := CHATLOG_DB.Where("session_id = ?", sessionId) + if beforeId > 0 { + q = q.Where("id < ?", beforeId) + } + if err := q.Order("id desc").Limit(limit + 1).Find(&turns).Error; err != nil { + return nil, false, err + } + if len(turns) > limit { + turns = turns[:limit] + hasMore = true + } + // newest-first from the DB; the transcript reads oldest-first + for i, j := 0, len(turns)-1; i < j; i, j = i+1, j-1 { + turns[i], turns[j] = turns[j], turns[i] + } + return turns, hasMore, nil +} diff --git a/model/chat_log_hot_cache.go b/model/chat_log_hot_cache.go new file mode 100644 index 000000000000..38d4c84adfac --- /dev/null +++ b/model/chat_log_hot_cache.go @@ -0,0 +1,372 @@ +package model + +import ( + "container/list" + "sort" + "sync" + "time" + + "github.com/QuantumNous/new-api/common" +) + +// chatLogHotCache keeps recent chat-log data in memory so the admin views of +// active conversations do not hit the (potentially huge) chat-log database: +// +// - sessions: a newest-first window of recent session metadata, refreshed +// from the DB every refreshInterval and updated write-through on every +// session insert/advance done by this node. +// - turns: per-session turn lists (including response bodies) under a +// process-wide byte budget, LRU-evicted at session granularity. Turns are +// immutable once written, so cached turns are always safe to serve. +// +// Anything outside these structures is cold data: it is loaded from the DB on +// first access (and the newest page of a cold session is admitted to the +// cache until budget pressure evicts it). +type chatLogHotCache struct { + mu sync.Mutex + + windowSize int + maxTurnBytes int + refreshInterval time.Duration + + sessions []*ChatSession // newest-first, metadata only + sessionsById map[int]*ChatSession // id -> entry shared with sessions + windowFull bool // window holds windowSize entries: older rows may exist in the DB + lastRefresh time.Time + + turns map[int]*chatTurnCacheEntry // session id -> entry + lru *list.List // session ids, front = most recently used + lruElems map[int]*list.Element + totalTurnBytes int +} + +// chatTurnCacheEntry holds the newest turns of one session. covered is the +// DB turn count the entry is synced to: when covered equals the DB-fresh +// TurnCount, the entry's turns reach up to the session's newest turn, though +// the entry may hold only the newest page of a long session. Oversized turns +// are never cached: admission skips the whole page when one appears, and a +// skipped write-through turn leaves covered behind the DB count, so neither +// path can serve an incomplete newest page. +type chatTurnCacheEntry struct { + turns []*ChatTurn // ascending by turn id + covered int + bytes int +} + +// maxCachedChatTurnBodyBytes caps admission of a single turn's bodies; larger +// turns stay DB-only instead of pinning oversized buffers in memory. +const maxCachedChatTurnBodyBytes = 1 << 20 + +var chatLogHot *chatLogHotCache + +func InitChatLogHotCache() { + if !common.GetEnvOrDefaultBool("CHAT_LOG_HOT_CACHE_ENABLED", true) || !ChatLogDBEnabled() { + chatLogHot = nil + return + } + chatLogHot = newChatLogHotCache( + common.GetEnvOrDefault("CHAT_LOG_HOT_SESSION_WINDOW", 1000), + common.GetEnvOrDefault("CHAT_LOG_HOT_TURN_BYTES", 64<<20), + time.Second*time.Duration(common.GetEnvOrDefault("CHAT_LOG_HOT_REFRESH_SECONDS", 60)), + ) +} + +func newChatLogHotCache(windowSize, maxTurnBytes int, refreshInterval time.Duration) *chatLogHotCache { + if windowSize < 1 { + windowSize = 1 + } + if maxTurnBytes < 1 { + maxTurnBytes = 1 + } + if refreshInterval < time.Second { + refreshInterval = time.Second + } + return &chatLogHotCache{ + windowSize: windowSize, + maxTurnBytes: maxTurnBytes, + refreshInterval: refreshInterval, + sessionsById: make(map[int]*ChatSession), + turns: make(map[int]*chatTurnCacheEntry), + lru: list.New(), + lruElems: make(map[int]*list.Element), + } +} + +func (c *chatLogHotCache) recordSession(s *ChatSession) { + if s == nil || s.Id == 0 { + return + } + c.mu.Lock() + defer c.mu.Unlock() + c.upsertSessionLocked(s) +} + +func (c *chatLogHotCache) upsertSessionLocked(s *ChatSession) { + if existing, ok := c.sessionsById[s.Id]; ok { + // refresh goroutines and controllers read cached copies after the lock + // is released; mutate the shared entry only under the lock + *existing = *s + if existing != c.sessions[0] { + c.moveSessionToFrontLocked(s.Id) + } + return + } + // copy: the caller's session object is owned by the persist goroutine + entry := *s + c.sessionsById[entry.Id] = &entry + c.sessions = append([]*ChatSession{&entry}, c.sessions...) + c.trimSessionWindowLocked() +} + +func (c *chatLogHotCache) moveSessionToFrontLocked(id int) { + for i, s := range c.sessions { + if s.Id == id { + c.sessions = append(c.sessions[:i], c.sessions[i+1:]...) + break + } + } + c.sessions = append([]*ChatSession{c.sessionsById[id]}, c.sessions...) +} + +func (c *chatLogHotCache) trimSessionWindowLocked() { + if len(c.sessions) <= c.windowSize { + return + } + for _, s := range c.sessions[c.windowSize:] { + delete(c.sessionsById, s.Id) + } + c.sessions = c.sessions[:c.windowSize] + c.windowFull = true +} + +func (c *chatLogHotCache) recordTurn(t *ChatTurn) { + if t == nil || t.SessionId == 0 { + return + } + bodyBytes := len(t.NewMessages) + len(t.ResponseBody) + if bodyBytes > maxCachedChatTurnBodyBytes { + return + } + c.mu.Lock() + defer c.mu.Unlock() + entry := c.turns[t.SessionId] + if entry == nil { + entry = &chatTurnCacheEntry{} + c.turns[t.SessionId] = entry + } + turns := entry.turns + i := sort.Search(len(turns), func(i int) bool { return turns[i].Id >= t.Id }) + if i < len(turns) && turns[i].Id == t.Id { + return // already recorded + } + // the cache owns its copies: the caller's struct may be reused after insert + owned := *t + turns = append(turns, nil) + copy(turns[i+1:], turns[i:]) + turns[i] = &owned + entry.turns = turns + entry.bytes += bodyBytes + c.totalTurnBytes += bodyBytes + // this node just persisted one more turn, so whatever DB total the entry + // was synced to grew by exactly one; if the entry was already behind + // (other nodes wrote), the equality check in getTurns still fails safely + entry.covered++ + c.touchTurnsLocked(t.SessionId) + c.evictTurnsLocked() +} + +func (c *chatLogHotCache) touchTurnsLocked(sessionId int) { + if elem, ok := c.lruElems[sessionId]; ok { + c.lru.MoveToFront(elem) + return + } + c.lruElems[sessionId] = c.lru.PushFront(sessionId) +} + +func (c *chatLogHotCache) evictTurnsLocked() { + // evict down to the byte budget; the last session may go too, otherwise a + // single oversized entry could keep the cache above budget forever + for c.totalTurnBytes > c.maxTurnBytes && c.lru.Len() > 0 { + c.evictOldestTurnSessionLocked() + } +} + +func (c *chatLogHotCache) evictOldestTurnSessionLocked() { + back := c.lru.Back() + if back == nil { + return + } + sessionId := back.Value.(int) + c.lru.Remove(back) + delete(c.lruElems, sessionId) + c.totalTurnBytes -= c.turns[sessionId].bytes + delete(c.turns, sessionId) +} + +// getTurns serves the newest `limit` turns of a session from cache when the +// entry is provably current: entry.covered must equal the DB-fresh +// totalTurns (otherwise some other node appended turns this cache has not +// seen, and the "newest" page would be stale), and the cached range must +// fully cover the requested page. Returns ok=false when the caller must +// fall back to the database. +func (c *chatLogHotCache) getTurns(sessionId, totalTurns, limit int) (turns []*ChatTurn, hasMore bool, ok bool) { + if limit < 1 { + limit = 1 + } + c.mu.Lock() + defer c.mu.Unlock() + entry, exists := c.turns[sessionId] + if !exists { + return nil, false, false + } + c.touchTurnsLocked(sessionId) + if entry.covered != totalTurns { + return nil, false, false // cache is behind the database: cold start + } + n := len(entry.turns) + // a long-session entry may hold only the newest page; it cannot serve a + // page larger than what it cached + if n < limit && n < entry.covered { + return nil, false, false + } + if n >= limit { + turns = entry.turns[n-limit:] + } else { + turns = entry.turns + } + return turns, entry.covered > limit, true +} + +// admitTurns stores a cold-loaded newest page. totalTurns is the DB-fresh +// session turn count, which anchors the entry: as long as the count is +// unchanged, the entry is known to hold the session's newest turns — even +// when it holds only the newest page of a long session. Pages containing an +// oversized turn are not cached at all: the entry would claim to hold the +// newest turns while missing one inside the page. +func (c *chatLogHotCache) admitTurns(sessionId int, turns []*ChatTurn, totalTurns int) { + if sessionId == 0 || len(turns) == 0 || totalTurns < len(turns) { + return + } + bytes := 0 + kept := make([]*ChatTurn, 0, len(turns)) + for _, t := range turns { + bodyBytes := len(t.NewMessages) + len(t.ResponseBody) + if bodyBytes > maxCachedChatTurnBodyBytes { + return // keep the entry DB-served instead of incomplete + } + // the cache owns its copies: admitted turns are also handed to the + // HTTP handler for serialization + owned := *t + kept = append(kept, &owned) + bytes += bodyBytes + } + c.mu.Lock() + defer c.mu.Unlock() + c.replaceTurnsLocked(sessionId, kept, totalTurns, bytes) + c.evictTurnsLocked() +} + +func (c *chatLogHotCache) replaceTurnsLocked(sessionId int, turns []*ChatTurn, covered int, bytes int) { + c.touchTurnsLocked(sessionId) + if old, exists := c.turns[sessionId]; exists { + c.totalTurnBytes -= old.bytes + } + c.turns[sessionId] = &chatTurnCacheEntry{turns: turns, covered: covered, bytes: bytes} + c.totalTurnBytes += bytes +} + +// listRecent serves the first page of the unfiltered, newest-first session +// list from the in-memory window. The window is refreshed from the DB when +// stale (covers multi-node deployments and the cold start after boot). +// Returns ok=false when the hot cache is disabled or the refresh failed. +func (c *chatLogHotCache) listRecent(limit int) (sessions []*ChatSession, hasMore bool, ok bool) { + if limit < 1 { + limit = 1 + } + c.mu.Lock() + defer c.mu.Unlock() + if time.Since(c.lastRefresh) >= c.refreshInterval { + if err := c.refreshLocked(); err != nil { + return nil, false, false + } + } + n := len(c.sessions) + if n == 0 { + return nil, c.windowFull, true + } + if n > limit { + sessions = c.sessions[:limit] + } else { + sessions = c.sessions + } + // return copies: cached entries are mutated by later advance calls under + // the lock, readers must not share the structs + out := make([]*ChatSession, len(sessions)) + for i, s := range sessions { + copyS := *s + out[i] = ©S + } + // hasMore must compare the pre-slice window length: the window may hold + // more entries than this page even when it is not full + return out, n > limit || c.windowFull, true +} + +func (c *chatLogHotCache) refreshLocked() error { + var sessions []*ChatSession + if err := CHATLOG_DB.Order("last_active_at desc, id desc").Limit(c.windowSize).Find(&sessions).Error; err != nil { + return err + } + c.sessions = sessions + c.sessionsById = make(map[int]*ChatSession, len(sessions)) + for _, s := range sessions { + c.sessionsById[s.Id] = s + } + c.windowFull = len(sessions) == c.windowSize + c.lastRefresh = time.Now() + return nil +} + +func chatLogCacheRecordSession(s *ChatSession) { + if chatLogHot != nil { + chatLogHot.recordSession(s) + } +} + +func chatLogCacheAdvanceSession(s *ChatSession) { + if chatLogHot != nil { + chatLogHot.recordSession(s) + } +} + +func chatLogCacheRecordTurn(t *ChatTurn) { + if chatLogHot != nil { + chatLogHot.recordTurn(t) + } +} + +// ListRecentChatSessions serves the first page of the newest-first session +// list from the hot cache. ok=false means the caller should query the DB. +func ListRecentChatSessions(limit int) (sessions []*ChatSession, hasMore bool, ok bool) { + if chatLogHot == nil { + return nil, false, false + } + return chatLogHot.listRecent(limit) +} + +// GetHotChatTurns serves a session's newest turns from the hot cache. +// ok=false means the caller should query the DB (cold start). +func GetHotChatTurns(sessionId, totalTurns, limit int) (turns []*ChatTurn, hasMore bool, ok bool) { + if chatLogHot == nil { + return nil, false, false + } + return chatLogHot.getTurns(sessionId, totalTurns, limit) +} + +// AdmitChatTurns stores a cold-loaded session page in the hot cache so +// follow-up views of the same session are served from memory. totalTurns is +// the DB-fresh session turn count that anchors the entry's currency. +func AdmitChatTurns(sessionId int, turns []*ChatTurn, totalTurns int) { + if chatLogHot != nil { + chatLogHot.admitTurns(sessionId, turns, totalTurns) + } +} diff --git a/model/chat_log_test.go b/model/chat_log_test.go index 4704711cc542..b3c8b5319fb1 100644 --- a/model/chat_log_test.go +++ b/model/chat_log_test.go @@ -1,7 +1,10 @@ package model import ( + "fmt" + "strings" "testing" + "time" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -21,7 +24,7 @@ func TestChatSession_CreateAndQuery(t *testing.T) { require.NoError(t, err) assert.Equal(t, 3, got.MessageCount) - turns, err := GetChatTurnsBySessionId(s.Id) + turns, _, err := GetChatTurnsPage(s.Id, 0, 20) require.NoError(t, err) require.Len(t, turns, 2) assert.Equal(t, 0, turns[0].TurnIndex) @@ -62,36 +65,410 @@ func TestChatSession_Advance(t *testing.T) { assert.Equal(t, int64(123), got.LastActiveAt) } -func TestSearchChatSessions(t *testing.T) { +func TestListChatSessions_KeysetPagination(t *testing.T) { truncateTables(t) if !ChatLogDBEnabled() { t.Skip("CHATLOG_DB not configured") } - require.NoError(t, (&ChatSession{TokenId: 1, UserId: 1, ModelName: "gpt-4", PrefixHash: "p1", MessageCount: 1, CreatedAt: 100, LastActiveAt: 100}).Insert()) - require.NoError(t, (&ChatSession{TokenId: 1, UserId: 2, ModelName: "claude", PrefixHash: "p2", MessageCount: 2, CreatedAt: 200, LastActiveAt: 200}).Insert()) - require.NoError(t, (&ChatSession{TokenId: 2, UserId: 1, ModelName: "gpt-4", PrefixHash: "p3", MessageCount: 3, CreatedAt: 300, LastActiveAt: 300}).Insert()) + insert := func(tokenId, userId int, modelName string, at int64) *ChatSession { + s := &ChatSession{TokenId: tokenId, UserId: userId, ModelName: modelName, + PrefixHash: fmt.Sprintf("p-%d", at), MessageCount: 1, CreatedAt: at, LastActiveAt: at} + require.NoError(t, s.Insert()) + return s + } + s3 := insert(2, 1, "gpt-4", 300) + insert(1, 2, "claude", 200) + s1 := insert(1, 1, "gpt-4", 100) + + // first page, newest first + list, hasMore, err := ListChatSessions(ChatSessionFilter{}, ChatSessionCursor{}, 2) + require.NoError(t, err) + require.Len(t, list, 2) + assert.True(t, hasMore) + assert.Equal(t, int64(300), list[0].LastActiveAt) + assert.Equal(t, int64(200), list[1].LastActiveAt) + + // walk to the next page with the cursor of the last row + cursor := ChatSessionCursor{LastActiveAt: list[1].LastActiveAt, Id: list[1].Id} + list, hasMore, err = ListChatSessions(ChatSessionFilter{}, cursor, 2) + require.NoError(t, err) + require.Len(t, list, 1) + assert.False(t, hasMore) + assert.Equal(t, s1.Id, list[0].Id) + + // same last_active_at ties are broken by id descending + list, _, err = ListChatSessions(ChatSessionFilter{}, ChatSessionCursor{}, 10) + require.NoError(t, err) + require.Len(t, list, 3) - list, total, err := SearchChatSessions(1, 0, "", 1, 10) + // filters + list, hasMore, err = ListChatSessions(ChatSessionFilter{TokenId: 1}, ChatSessionCursor{}, 10) require.NoError(t, err) - assert.Equal(t, int64(2), total) - assert.Len(t, list, 2) + assert.False(t, hasMore) + require.Len(t, list, 2) - list, total, err = SearchChatSessions(0, 1, "", 1, 10) + list, _, err = ListChatSessions(ChatSessionFilter{UserId: 1}, ChatSessionCursor{}, 10) require.NoError(t, err) - assert.Equal(t, int64(2), total) + require.Len(t, list, 2) - list, total, err = SearchChatSessions(0, 0, "gpt-4", 1, 10) + list, _, err = ListChatSessions(ChatSessionFilter{ModelName: "gpt-4"}, ChatSessionCursor{}, 10) require.NoError(t, err) - assert.Equal(t, int64(2), total) + require.Len(t, list, 2) - list, total, err = SearchChatSessions(0, 0, "", 1, 1) + // time range on last_active_at, inclusive + list, _, err = ListChatSessions(ChatSessionFilter{StartTs: 200, EndTs: 300}, ChatSessionCursor{}, 10) + require.NoError(t, err) + require.Len(t, list, 2) + + // cursor combined with a filter + cursor = ChatSessionCursor{LastActiveAt: 300, Id: s3.Id} + list, _, err = ListChatSessions(ChatSessionFilter{UserId: 1}, cursor, 10) require.NoError(t, err) - assert.Equal(t, int64(3), total) require.Len(t, list, 1) - assert.Equal(t, int64(300), list[0].LastActiveAt, "ordered by last_active_at desc") + assert.Equal(t, s1.Id, list[0].Id) +} + +func TestGetChatTurnsPage(t *testing.T) { + truncateTables(t) + if !ChatLogDBEnabled() { + t.Skip("CHATLOG_DB not configured") + } + s := &ChatSession{TokenId: 1, PrefixHash: "p"} + require.NoError(t, s.Insert()) + ids := make([]int, 0, 5) + for i := 0; i < 5; i++ { + turn := &ChatTurn{SessionId: s.Id, TurnIndex: i, RequestId: "r", ResponseBody: "{}"} + require.NoError(t, turn.Insert()) + ids = append(ids, turn.Id) + } + + // newest page, ascending order inside the page + turns, hasMore, err := GetChatTurnsPage(s.Id, 0, 3) + require.NoError(t, err) + assert.True(t, hasMore) + require.Len(t, turns, 3) + assert.Equal(t, ids[2], turns[0].Id) + assert.Equal(t, ids[4], turns[2].Id) + + // page older than the newest turn id + turns, hasMore, err = GetChatTurnsPage(s.Id, ids[2], 3) + require.NoError(t, err) + assert.False(t, hasMore) + require.Len(t, turns, 2) + assert.Equal(t, ids[0], turns[0].Id) - list, total, err = SearchChatSessions(0, 0, "", 2, 10) + // other sessions stay invisible + other := &ChatSession{TokenId: 2, PrefixHash: "q"} + require.NoError(t, other.Insert()) + require.NoError(t, (&ChatTurn{SessionId: other.Id, TurnIndex: 0}).Insert()) + turns, _, err = GetChatTurnsPage(s.Id, 0, 10) require.NoError(t, err) - assert.Equal(t, int64(3), total) - assert.Empty(t, list, "page 2 of 3 items is empty") + assert.Len(t, turns, 5) +} + +func TestChatSessionCursor_Roundtrip(t *testing.T) { + c := ChatSessionCursor{LastActiveAt: 1694123456, Id: 42} + decoded, err := DecodeChatSessionCursor(c.Encode()) + require.NoError(t, err) + assert.Equal(t, c, decoded) + + decoded, err = DecodeChatSessionCursor("") + require.NoError(t, err) + assert.Equal(t, ChatSessionCursor{}, decoded) + + _, err = DecodeChatSessionCursor("not-base64!!") + assert.Error(t, err) + + _, err = DecodeChatSessionCursor("Zm9vYmFy") // "foobar", wrong shape + assert.Error(t, err) +} + +// setupChatLogHotCache installs a hot cache for the test and restores the +// previous global state afterwards. +func setupChatLogHotCache(t *testing.T, windowSize, maxTurnBytes int, refreshInterval time.Duration) *chatLogHotCache { + t.Helper() + previous := chatLogHot + cache := newChatLogHotCache(windowSize, maxTurnBytes, refreshInterval) + chatLogHot = cache + t.Cleanup(func() { + chatLogHot = previous + }) + return cache +} + +func TestChatLogHotCache_ListRecentServesWindowAndRefreshes(t *testing.T) { + truncateTables(t) + if !ChatLogDBEnabled() { + t.Skip("CHATLOG_DB not configured") + } + cache := setupChatLogHotCache(t, 3, 64<<20, time.Hour) + + s1 := &ChatSession{TokenId: 1, PrefixHash: "p1", LastActiveAt: 100} + require.NoError(t, s1.Insert()) // write-through populates the window + s2 := &ChatSession{TokenId: 2, PrefixHash: "p2", LastActiveAt: 200} + require.NoError(t, s2.Insert()) + + sessions, hasMore, ok := ListRecentChatSessions(10) + require.True(t, ok) + assert.False(t, hasMore, "window not full, all sessions fit") + require.Len(t, sessions, 2) + assert.Equal(t, s2.Id, sessions[0].Id, "newest first") + + // a session written by another node appears after the stale window is refreshed + s3 := &ChatSession{TokenId: 3, PrefixHash: "p3", LastActiveAt: 300} + require.NoError(t, CHATLOG_DB.Create(s3).Error) // bypasses write-through + _, _, ok = ListRecentChatSessions(10) + require.True(t, ok) + cache.mu.Lock() + cache.lastRefresh = time.Now().Add(-2 * time.Hour) + cache.mu.Unlock() + sessions, _, ok = ListRecentChatSessions(10) + require.True(t, ok) + require.Len(t, sessions, 3) + assert.Equal(t, s3.Id, sessions[0].Id) + + // the window (not full) holds more than one page: has_more must be true + sessions, hasMore, ok = ListRecentChatSessions(2) + require.True(t, ok) + require.Len(t, sessions, 2) + assert.True(t, hasMore, "window holds more than one page even when not full") + + // advancing a cached session moves it back to the head with fresh metadata + s1.TurnCount = 6 + s1.LastActiveAt = 400 + require.NoError(t, s1.Advance("glm-5.3", 400)) + sessions, _, ok = ListRecentChatSessions(10) + require.True(t, ok) + assert.Equal(t, s1.Id, sessions[0].Id) + assert.Equal(t, 7, sessions[0].TurnCount) +} + +func TestChatLogHotCache_ListRecentWindowFullReportsMore(t *testing.T) { + truncateTables(t) + if !ChatLogDBEnabled() { + t.Skip("CHATLOG_DB not configured") + } + setupChatLogHotCache(t, 2, 64<<20, time.Hour) + + for i := int64(1); i <= 3; i++ { + require.NoError(t, (&ChatSession{TokenId: int(i), PrefixHash: "p", LastActiveAt: i}).Insert()) + } + + sessions, hasMore, ok := ListRecentChatSessions(2) + require.True(t, ok) + assert.True(t, hasMore, "window of 2 is full, older sessions may exist") + require.Len(t, sessions, 2) +} + +func TestChatLogHotCache_TurnsServedFromCache(t *testing.T) { + truncateTables(t) + if !ChatLogDBEnabled() { + t.Skip("CHATLOG_DB not configured") + } + setupChatLogHotCache(t, 100, 64<<20, time.Hour) + + s := &ChatSession{TokenId: 1, PrefixHash: "p"} + require.NoError(t, s.Insert()) + var turns []*ChatTurn + for i := 0; i < 5; i++ { + turn := &ChatTurn{SessionId: s.Id, TurnIndex: i, NewMessages: "[]", ResponseBody: "{}"} + require.NoError(t, turn.Insert()) + turns = append(turns, turn) + } + + // newest 3 of 5 cached turns + got, hasMore, ok := GetHotChatTurns(s.Id, 5, 3) + require.True(t, ok) + assert.True(t, hasMore) + require.Len(t, got, 3) + assert.Equal(t, turns[2].Id, got[0].Id) + assert.Equal(t, turns[4].Id, got[2].Id) + + // full session in cache, complete view + got, hasMore, ok = GetHotChatTurns(s.Id, 5, 5) + require.True(t, ok) + assert.False(t, hasMore) + require.Len(t, got, 5) + + // cache holds fewer turns than requested and fewer than exist: cold start + _, _, ok = GetHotChatTurns(s.Id, 8, 10) + assert.False(t, ok) + + // cache covers the page size but is behind the DB total (another node + // appended turns): serving the "newest" page would be stale — cold start + _, _, ok = GetHotChatTurns(s.Id, 8, 3) + assert.False(t, ok) + + // a session the cache never saw: cold start + _, _, ok = GetHotChatTurns(999999, 1, 1) + assert.False(t, ok) +} + +func TestChatLogHotCache_ColdLoadAdmitsTurns(t *testing.T) { + truncateTables(t) + if !ChatLogDBEnabled() { + t.Skip("CHATLOG_DB not configured") + } + setupChatLogHotCache(t, 100, 64<<20, time.Hour) + + s := &ChatSession{TokenId: 1, PrefixHash: "p", TurnCount: 2} + require.NoError(t, s.Insert()) + t1 := &ChatTurn{SessionId: s.Id, TurnIndex: 0, NewMessages: "[]", ResponseBody: "{}"} + t2 := &ChatTurn{SessionId: s.Id, TurnIndex: 1, NewMessages: "[]", ResponseBody: "{}"} + require.NoError(t, CHATLOG_DB.Create(t1).Error) // written by another node + require.NoError(t, CHATLOG_DB.Create(t2).Error) + + // cold load path: DB page then admit, anchored to the DB turn count + page, hasMore, err := GetChatTurnsPage(s.Id, 0, 20) + require.NoError(t, err) + assert.False(t, hasMore) + AdmitChatTurns(s.Id, page, 2) + + got, hasMore, ok := GetHotChatTurns(s.Id, 2, 20) + require.True(t, ok) + assert.False(t, hasMore) + require.Len(t, got, 2) +} + +// TestChatLogHotCache_OversizedNewestTurnKeepsSessionCold pins the fix for +// the PR review finding that admitting a page which silently skips an +// oversized turn could later serve an incomplete "newest page" from cache. +func TestChatLogHotCache_OversizedNewestTurnKeepsSessionCold(t *testing.T) { + truncateTables(t) + if !ChatLogDBEnabled() { + t.Skip("CHATLOG_DB not configured") + } + setupChatLogHotCache(t, 100, 64<<20, time.Hour) + + s := &ChatSession{TokenId: 1, PrefixHash: "p", TurnCount: 3} + require.NoError(t, s.Insert()) + small1 := &ChatTurn{SessionId: s.Id, TurnIndex: 0, NewMessages: "[]", ResponseBody: "{}"} + require.NoError(t, CHATLOG_DB.Create(small1).Error) + small2 := &ChatTurn{SessionId: s.Id, TurnIndex: 1, NewMessages: "[]", ResponseBody: "{}"} + require.NoError(t, CHATLOG_DB.Create(small2).Error) + big := &ChatTurn{SessionId: s.Id, TurnIndex: 2, NewMessages: "[]", ResponseBody: strings.Repeat("x", maxCachedChatTurnBodyBytes+1)} + require.NoError(t, CHATLOG_DB.Create(big).Error) + + // cold view loads the newest page; the oversized newest turn makes the + // page uncacheable as a whole + page, hasMore, err := GetChatTurnsPage(s.Id, 0, 3) + require.NoError(t, err) + assert.False(t, hasMore) + require.Len(t, page, 3) + AdmitChatTurns(s.Id, page, 3) + + // every follow-up view must cold-start instead of serving a newest page + // that silently omits the oversized turn + _, _, ok := GetHotChatTurns(s.Id, 3, 2) + assert.False(t, ok, "entry must not claim to hold the newest turns") + _, _, ok = GetHotChatTurns(s.Id, 3, 3) + assert.False(t, ok) + + // a write-through oversized turn on an admitted entry also stays cold: + // covered only grows for cached turns, so it lags the DB count + fresh := &ChatSession{TokenId: 2, PrefixHash: "q", TurnCount: 1} + require.NoError(t, fresh.Insert()) + first := &ChatTurn{SessionId: fresh.Id, TurnIndex: 0, NewMessages: "[]", ResponseBody: "{}"} + require.NoError(t, first.Insert()) + _, _, ok = GetHotChatTurns(fresh.Id, 1, 1) + require.True(t, ok) + oversized := &ChatTurn{SessionId: fresh.Id, TurnIndex: 1, NewMessages: "[]", ResponseBody: strings.Repeat("x", maxCachedChatTurnBodyBytes+1)} + require.NoError(t, CHATLOG_DB.Create(oversized).Error) + _, _, ok = GetHotChatTurns(fresh.Id, 2, 1) + assert.False(t, ok, "covered lags the DB count after the oversized append") +} + +// the PR review finding that a strictly count-based staleness check disabled +// cache hits for long sessions: a cold-admitted newest page must serve +// follow-up views while the DB turn count is unchanged. +func TestChatLogHotCache_LongSessionNewestPageServesFromCache(t *testing.T) { + truncateTables(t) + if !ChatLogDBEnabled() { + t.Skip("CHATLOG_DB not configured") + } + setupChatLogHotCache(t, 100, 64<<20, time.Hour) + + s := &ChatSession{TokenId: 1, PrefixHash: "p", TurnCount: 10} + require.NoError(t, s.Insert()) + ids := make([]int, 0, 10) + for i := 0; i < 10; i++ { + turn := &ChatTurn{SessionId: s.Id, TurnIndex: i, NewMessages: "[]", ResponseBody: "{}"} + require.NoError(t, CHATLOG_DB.Create(turn).Error) // all written by another node + ids = append(ids, turn.Id) + } + + // first view: cold start, only the newest page is loaded and admitted + page, hasMore, err := GetChatTurnsPage(s.Id, 0, 3) + require.NoError(t, err) + assert.True(t, hasMore) + AdmitChatTurns(s.Id, page, 10) + + // second view of the same page must hit the cache, not the DB + got, hasMore, ok := GetHotChatTurns(s.Id, 10, 3) + require.True(t, ok) + assert.True(t, hasMore, "older turns exist beyond the admitted page") + require.Len(t, got, 3) + assert.Equal(t, ids[7], got[0].Id) + assert.Equal(t, ids[9], got[2].Id) + + // another node appends a turn: the anchored count no longer matches, cold start + t11 := &ChatTurn{SessionId: s.Id, TurnIndex: 10, NewMessages: "[]", ResponseBody: "{}"} + require.NoError(t, CHATLOG_DB.Create(t11).Error) + _, _, ok = GetHotChatTurns(s.Id, 11, 3) + assert.False(t, ok, "stale entry after another node appended a turn") + + // a write-through append on this node keeps an admitted entry current + AdmitChatTurns(s.Id, page, 10) + t12 := &ChatTurn{SessionId: s.Id, TurnIndex: 11, NewMessages: "[]", ResponseBody: "{}"} + require.NoError(t, t12.Insert()) + got, hasMore, ok = GetHotChatTurns(s.Id, 11, 3) + require.True(t, ok) + assert.True(t, hasMore) + require.Len(t, got, 3) + assert.Equal(t, t12.Id, got[2].Id, "local append is served as the newest turn") +} + +func TestChatLogHotCache_ByteBudgetEvictsLRU(t *testing.T) { + truncateTables(t) + if !ChatLogDBEnabled() { + t.Skip("CHATLOG_DB not configured") + } + setupChatLogHotCache(t, 100, 50, time.Hour) // tiny budget: three 20-byte turns overflow it + + mk := func(tokenId int) *ChatSession { + s := &ChatSession{TokenId: tokenId, PrefixHash: "p"} + require.NoError(t, s.Insert()) + return s + } + s1 := mk(1) + s2 := mk(2) + big := &ChatTurn{SessionId: s1.Id, TurnIndex: 0, NewMessages: "0123456789", ResponseBody: "0123456789"} // 20 bytes + require.NoError(t, big.Insert()) + first := &ChatTurn{SessionId: s2.Id, TurnIndex: 0, NewMessages: "0123456789", ResponseBody: "0123456789"} + require.NoError(t, first.Insert()) + + _, _, ok := GetHotChatTurns(s1.Id, 1, 1) + require.True(t, ok, "both small sessions fit") + + // touching s2 makes s1 the least recently used; a new turn in s2 evicts s1 + _, _, ok = GetHotChatTurns(s2.Id, 1, 1) + require.True(t, ok) + second := &ChatTurn{SessionId: s2.Id, TurnIndex: 1, NewMessages: "0123456789", ResponseBody: "0123456789"} + require.NoError(t, second.Insert()) + + _, _, ok = GetHotChatTurns(s1.Id, 1, 1) + assert.False(t, ok, "s1 evicted by byte budget") + _, _, ok = GetHotChatTurns(s2.Id, 2, 2) + require.True(t, ok, "most recently used session kept") +} + +func TestChatLogHotCache_Disabled(t *testing.T) { + previous := chatLogHot + chatLogHot = nil + t.Cleanup(func() { chatLogHot = previous }) + + _, _, ok := ListRecentChatSessions(10) + assert.False(t, ok) + _, _, ok = GetHotChatTurns(1, 1, 1) + assert.False(t, ok) } diff --git a/model/clickhouse_log_test.go b/model/clickhouse_log_test.go index d9737e6b226d..0afdcb335620 100644 --- a/model/clickhouse_log_test.go +++ b/model/clickhouse_log_test.go @@ -59,7 +59,7 @@ func TestChooseDBRejectsClickHouseForMainDatabase(t *testing.T) { }) require.NoError(t, os.Setenv("SQL_DSN", "clickhouse://default:pass@localhost:9000/logs")) - db, dbType, err := chooseDB("SQL_DSN", false) + db, dbType, err := chooseDB("SQL_DSN", false, common.SQLitePath) require.Error(t, err) assert.Nil(t, db) assert.Equal(t, common.DatabaseType(""), dbType) diff --git a/model/main.go b/model/main.go index a89be40f4264..17d385d558cc 100644 --- a/model/main.go +++ b/model/main.go @@ -130,7 +130,7 @@ func normalizeClickHouseDSN(dsn string) string { return parsed.String() } -func chooseDB(envName string, isLog bool) (*gorm.DB, common.DatabaseType, error) { +func chooseDB(envName string, isLog bool, sqlitePath string) (*gorm.DB, common.DatabaseType, error) { dsn := os.Getenv(envName) if dsn != "" { if isClickHouseDSN(dsn) { @@ -151,8 +151,8 @@ func chooseDB(envName string, isLog bool) (*gorm.DB, common.DatabaseType, error) return db, common.DatabaseTypePostgreSQL, err } if strings.HasPrefix(dsn, "local") { - common.SysLog("SQL_DSN not set, using SQLite as database") - db, err := gorm.Open(sqlite.Open(common.SQLitePath), newGormConfig(true)) + common.SysLog(envName + " is set to local, using SQLite as database") + db, err := gorm.Open(sqlite.Open(sqlitePath), newGormConfig(true)) return db, common.DatabaseTypeSQLite, err } // Use MySQL @@ -169,13 +169,13 @@ func chooseDB(envName string, isLog bool) (*gorm.DB, common.DatabaseType, error) return db, common.DatabaseTypeMySQL, err } // Use SQLite - common.SysLog("SQL_DSN not set, using SQLite as database") - db, err := gorm.Open(sqlite.Open(common.SQLitePath), newGormConfig(true)) + common.SysLog(envName + " not set, using SQLite as database") + db, err := gorm.Open(sqlite.Open(sqlitePath), newGormConfig(true)) return db, common.DatabaseTypeSQLite, err } func InitDB() (err error) { - db, dbType, err := chooseDB("SQL_DSN", false) + db, dbType, err := chooseDB("SQL_DSN", false, common.SQLitePath) if err == nil { common.SetMainDatabaseType(dbType) if os.Getenv("LOG_SQL_DSN") == "" { @@ -222,7 +222,7 @@ func InitLogDB() (err error) { initCol() return } - db, dbType, err := chooseDB("LOG_SQL_DSN", true) + db, dbType, err := chooseDB("LOG_SQL_DSN", true, common.SQLitePath) if err == nil { common.SetLogDatabaseType(dbType) initCol() @@ -411,12 +411,38 @@ func migrateLOGDB() error { return LOG_DB.AutoMigrate(&Log{}) } +// chatLogSQLiteDSN resolves the SQLite file for the chat-log database when +// CHAT_LOG_SQL_DSN is "local" or "local:". Unlike the main database, +// the default is a dedicated chatlog.db file (WAL journal) so chat details +// never bloat one-api.db. Precedence: "local:" DSN suffix, then +// CHAT_LOG_SQLITE_PATH, then the chatlog.db default. +func chatLogSQLiteDSN(dsn string) string { + path := strings.TrimSpace(strings.TrimPrefix(strings.TrimPrefix(dsn, "local"), ":")) + if path == "" { + path = common.GetEnvOrDefaultString("CHAT_LOG_SQLITE_PATH", "chatlog.db") + } + return withSQLitePragmas(path) +} + +func withSQLitePragmas(path string) string { + if strings.Contains(path, "?") { + return path + } + return path + "?_pragma=busy_timeout(10000)&_pragma=journal_mode(WAL)" +} + func InitChatLogDB() (err error) { if os.Getenv("CHAT_LOG_SQL_DSN") == "" { common.SysLog("CHAT_LOG_SQL_DSN not set, chat-log detail storage disabled") return nil } - db, dbType, err := chooseDB("CHAT_LOG_SQL_DSN", false) + dsn := os.Getenv("CHAT_LOG_SQL_DSN") + sqlitePath := common.SQLitePath + if strings.HasPrefix(dsn, "local") { + sqlitePath = chatLogSQLiteDSN(dsn) + common.SysLog("using standalone SQLite file as chat-log detail database: " + sqlitePath) + } + db, dbType, err := chooseDB("CHAT_LOG_SQL_DSN", false, sqlitePath) if err != nil { common.FatalLog(err) return err @@ -434,6 +460,7 @@ func InitChatLogDB() (err error) { sqlDB.SetMaxIdleConns(common.GetEnvOrDefault("SQL_MAX_IDLE_CONNS", 100)) sqlDB.SetMaxOpenConns(common.GetEnvOrDefault("SQL_MAX_OPEN_CONNS", 1000)) sqlDB.SetConnMaxLifetime(time.Second * time.Duration(common.GetEnvOrDefault("SQL_MAX_LIFETIME", 60))) + InitChatLogHotCache() if !common.IsMasterNode { return nil } @@ -455,7 +482,56 @@ func migrateChatLogDB(dbType common.DatabaseType) error { return err } } - return CHATLOG_DB.Exec("DROP TABLE IF EXISTS chat_logs").Error + if err := CHATLOG_DB.Exec("DROP TABLE IF EXISTS chat_logs").Error; err != nil { + return err + } + return dropLegacyChatTablesFromMainDB() +} + +// dropLegacyChatTablesFromMainDB removes chat-detail tables that previous +// versions created inside the main SQLite database (one-api.db) when +// CHAT_LOG_SQL_DSN was "local". Existing chat-detail data is explicitly not +// migrated — the new standalone database starts fresh. Only the three +// chat-log tables are ever dropped; every other main-DB table is untouched. +func dropLegacyChatTablesFromMainDB() error { + if !legacyChatTablesShareMainSQLite() { + return nil + } + for _, table := range []string{"chat_logs", "chat_turns", "chat_sessions"} { + if !DB.Migrator().HasTable(table) { + continue + } + if err := DB.Exec("DROP TABLE IF EXISTS `" + table + "`").Error; err != nil { + return err + } + common.SysLog("dropped legacy chat-log table `" + table + "` from main database") + } + return nil +} + +// legacyChatTablesShareMainSQLite reports whether the main database could be +// holding legacy chat-log tables AND the chat-log database now lives +// elsewhere. The drop is restricted to a SQLite main DB plus a proven +// different chat-log target, so an actively used chat-log database can never +// be dropped by accident. +func legacyChatTablesShareMainSQLite() bool { + if !common.UsingMainDatabase(common.DatabaseTypeSQLite) { + return false + } + if !common.UsingChatLogDatabase(common.DatabaseTypeSQLite) { + return true + } + dsn := os.Getenv("CHAT_LOG_SQL_DSN") + chatlogFile := chatLogSQLiteDSN(dsn) + return sqliteFileOf(chatlogFile) != sqliteFileOf(common.SQLitePath) +} + +// sqliteFileOf strips DSN query options so file paths compare by location only. +func sqliteFileOf(dsn string) string { + if i := strings.IndexByte(dsn, '?'); i >= 0 { + return dsn[:i] + } + return dsn } func migrateClickHouseLogDB() error { @@ -754,12 +830,18 @@ func closeDB(db *gorm.DB) error { } func CloseDB() error { - if LOG_DB != DB { + if LOG_DB != nil && LOG_DB != DB { err := closeDB(LOG_DB) if err != nil { return err } } + if CHATLOG_DB != nil && CHATLOG_DB != DB { + err := closeDB(CHATLOG_DB) + if err != nil { + return err + } + } return closeDB(DB) } diff --git a/service/chat_log_persist_test.go b/service/chat_log_persist_test.go index 3a83ee9e0d31..9fcc5d5a971e 100644 --- a/service/chat_log_persist_test.go +++ b/service/chat_log_persist_test.go @@ -73,28 +73,28 @@ func TestChatLogRecorder_PersistChainsTurns(t *testing.T) { persistTurn("req-turn-1", turn1Body) require.Eventually(t, func() bool { - sessions, total, err := model.SearchChatSessions(tokenId, 0, "", 1, 10) - return err == nil && total == 1 && len(sessions) == 1 && + sessions, hasMore, err := model.ListChatSessions(model.ChatSessionFilter{TokenId: tokenId}, model.ChatSessionCursor{}, 10) + return err == nil && !hasMore && len(sessions) == 1 && sessions[0].TurnCount == 1 && sessions[0].MessageCount == 2 }, 2*time.Second, 20*time.Millisecond) persistTurn("req-turn-2", turn2Body) require.Eventually(t, func() bool { - sessions, total, err := model.SearchChatSessions(tokenId, 0, "", 1, 10) - return err == nil && total == 1 && len(sessions) == 1 && + sessions, hasMore, err := model.ListChatSessions(model.ChatSessionFilter{TokenId: tokenId}, model.ChatSessionCursor{}, 10) + return err == nil && !hasMore && len(sessions) == 1 && sessions[0].TurnCount == 2 && sessions[0].MessageCount == 4 }, 2*time.Second, 20*time.Millisecond) - sessions, total, err := model.SearchChatSessions(tokenId, 0, "", 1, 10) + sessions, hasMore, err := model.ListChatSessions(model.ChatSessionFilter{TokenId: tokenId}, model.ChatSessionCursor{}, 10) require.NoError(t, err) - require.Equal(t, int64(1), total) + require.False(t, hasMore) require.Len(t, sessions, 1) s := sessions[0] assert.Equal(t, 2, s.TurnCount) assert.Equal(t, 4, s.MessageCount) assert.Equal(t, `"You are helpful."`, s.System) - turns, err := model.GetChatTurnsBySessionId(s.Id) + turns, _, err := model.GetChatTurnsPage(s.Id, 0, 100) require.NoError(t, err) require.Len(t, turns, 2) assert.Equal(t, 0, turns[0].TurnIndex) diff --git a/web/src/features/chat-logs/api.ts b/web/src/features/chat-logs/api.ts index 2ad85ef98be0..11f275685db1 100644 --- a/web/src/features/chat-logs/api.ts +++ b/web/src/features/chat-logs/api.ts @@ -20,6 +20,7 @@ import { api } from '@/lib/api' import type { GetChatSessionsParams, + GetSessionDetailParams, SessionDetailResponse, SessionListResponse, } from './types' @@ -32,9 +33,10 @@ export async function getChatSessions( } export async function getChatSessionDetail( - id: number + id: number, + params: GetSessionDetailParams = {} ): Promise { - const res = await api.get(`/api/chat_logs/sessions/${id}`) + const res = await api.get(`/api/chat_logs/sessions/${id}`, { params }) return res.data } diff --git a/web/src/features/chat-logs/components/chat-sessions-table.tsx b/web/src/features/chat-logs/components/chat-sessions-table.tsx index ba6c134e0cf9..44972c290a40 100644 --- a/web/src/features/chat-logs/components/chat-sessions-table.tsx +++ b/web/src/features/chat-logs/components/chat-sessions-table.tsx @@ -7,7 +7,7 @@ published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. This program is distributed in the hope that it will be useful, -but WITHOUT ANY WARRANTY; without even implied warranty of +but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. @@ -16,11 +16,11 @@ along with this program. If not, see . For commercial licensing, please contact support@quantumnous.com */ -import { useQuery } from '@tanstack/react-query' +import { useInfiniteQuery } from '@tanstack/react-query' import { getRouteApi } from '@tanstack/react-router' import type { ColumnDef } from '@tanstack/react-table' -import { Eye } from 'lucide-react' -import { useMemo, useState } from 'react' +import { Eye, Loader2 } from 'lucide-react' +import { useEffect, useMemo, useState } from 'react' import { useTranslation } from 'react-i18next' import { @@ -28,9 +28,9 @@ import { useDataTable, useDebouncedColumnFilter, } from '@/components/data-table' -import { useTableUrlState } from '@/hooks/use-table-url-state' import { Button } from '@/components/ui/button' import { Input } from '@/components/ui/input' +import { useTableUrlState } from '@/hooks/use-table-url-state' import { formatTimestampToDate } from '@/lib/format' import { chatSessionsQueryKeys, getChatSessions } from '../api' @@ -39,21 +39,17 @@ import { SessionDetailSheet } from './session-detail-sheet' const route = getRouteApi('/_authenticated/chat-logs/') +const SESSIONS_PAGE_SIZE = 20 + export function ChatSessionsTable() { const { t } = useTranslation() const [detailId, setDetailId] = useState(null) const [detailOpen, setDetailOpen] = useState(false) + const [pageIndex, setPageIndex] = useState(0) - const { - columnFilters, - onColumnFiltersChange, - pagination, - onPaginationChange, - ensurePageInRange, - } = useTableUrlState({ + const { columnFilters, onColumnFiltersChange } = useTableUrlState({ search: route.useSearch(), navigate: route.useNavigate(), - pagination: { defaultPage: 1, defaultPageSize: 10 }, globalFilter: { enabled: false }, columnFilters: [ { @@ -190,89 +186,130 @@ export function ChatSessionsTable() { [t] ) - const queryParams = { + const filterParams = { token_id: tokenIdFilter ? Number(tokenIdFilter) : undefined, user_id: userIdFilter ? Number(userIdFilter) : undefined, model_name: modelNameFilter || undefined, - page: pagination.pageIndex + 1, - page_size: pagination.pageSize, } - const { data, isLoading, isFetching } = useQuery({ - queryKey: chatSessionsQueryKeys.list(queryParams), - queryFn: () => getChatSessions(queryParams), + const query = useInfiniteQuery({ + queryKey: chatSessionsQueryKeys.list(filterParams), + queryFn: ({ pageParam }) => + getChatSessions({ + ...filterParams, + limit: SESSIONS_PAGE_SIZE, + cursor: pageParam, + }), + initialPageParam: undefined as string | undefined, + getNextPageParam: (last) => last.data?.next_cursor ?? undefined, placeholderData: (prev) => prev, }) - const sessions = data?.data ?? [] - const totalCount = data?.total ?? 0 + // filter changes restart the cursor list; drop the client-side page back to the first + useEffect(() => { + setPageIndex(0) + }, [tokenIdFilter, userIdFilter, modelNameFilter]) + + const sessions = useMemo( + () => query.data?.pages.flatMap((page) => page.data?.items ?? []) ?? [], + [query.data] + ) + const isLoading = query.isLoading + const isFetching = query.isFetching + const hasNextPage = query.hasNextPage + + const pagination = useMemo( + () => ({ + pageIndex, + pageSize: SESSIONS_PAGE_SIZE, + }), + [pageIndex] + ) const { table } = useDataTable({ data: sessions, columns, - totalCount, - columnFilters, + totalCount: sessions.length, pagination, + onPaginationChange: (updater) => { + const next = typeof updater === 'function' ? updater(pagination) : updater + setPageIndex(next.pageIndex) + }, + columnFilters, onColumnFiltersChange, - onPaginationChange, - manualPagination: true, manualFiltering: true, enableRowSelection: false, - ensurePageInRange, }) return ( <> - - - - - - ), - hasAdditionalFilters: - !!tokenIdFilter || !!userIdFilter || !!modelNameFilter, - onReset: () => { - resetTokenIdInput() - resetUserIdInput() - resetModelNameInput() - }, - hideViewOptions: true, - }} - /> +
+ + + + + + ), + hasAdditionalFilters: + !!tokenIdFilter || !!userIdFilter || !!modelNameFilter, + onReset: () => { + resetTokenIdInput() + resetUserIdInput() + resetModelNameInput() + }, + hideViewOptions: true, + }} + /> + {hasNextPage && ( +
+ +
+ )} +
. For commercial licensing, please contact support@quantumnous.com */ -import { useQuery } from '@tanstack/react-query' -import { Loader2 } from 'lucide-react' +import { useInfiniteQuery } from '@tanstack/react-query' +import { ChevronUp, Loader2 } from 'lucide-react' import { useMemo } from 'react' import { useTranslation } from 'react-i18next' import { CodeBlock } from '@/components/ai-elements/code-block' import { sideDrawerContentClassName } from '@/components/drawer-layout' +import { Button } from '@/components/ui/button' import { Sheet, SheetContent, @@ -35,6 +36,8 @@ import { formatTimestampToDate } from '@/lib/format' import { chatSessionsQueryKeys, getChatSessionDetail } from '../api' import { buildTranscript } from '../lib/transcript' +const TURNS_PAGE_SIZE = 10 + export interface SessionDetailSheetProps { open: boolean onOpenChange: (open: boolean) => void @@ -72,18 +75,33 @@ function toKeyedCodes(messages: unknown[]): KeyedCode[] { export function SessionDetailSheet(props: SessionDetailSheetProps) { const { t } = useTranslation() - const query = useQuery({ + const query = useInfiniteQuery({ queryKey: chatSessionsQueryKeys.detail(props.id ?? 0), - queryFn: () => getChatSessionDetail(props.id as number), + queryFn: ({ pageParam }) => + getChatSessionDetail(props.id as number, { + limit: TURNS_PAGE_SIZE, + before_id: pageParam, + }), + initialPageParam: undefined as number | undefined, + getNextPageParam: (last) => last.data?.next_turn_id ?? undefined, enabled: props.open && props.id != null, }) - const record = query.data?.data + const firstPage = query.data?.pages[0] + const record = firstPage?.data + // pages arrive newest-first (older pages are prepended); the transcript + // renders oldest → newest const turns = useMemo( - () => (record ? buildTranscript(record.turns) : []), - [record] + () => + query.data + ? [...query.data.pages] + .reverse() + .flatMap((page) => page.data?.turns ?? []) + : [], + [query.data] ) - const unavailable = !query.isLoading && (query.isError || !query.data?.success) + const transcript = useMemo(() => buildTranscript(turns), [turns]) + const unavailable = !query.isLoading && (query.isError || !firstPage?.success) let body: React.ReactNode = null if (query.isLoading) { @@ -96,7 +114,7 @@ export function SessionDetailSheet(props: SessionDetailSheetProps) { } else if (unavailable) { body = (

- {query.data?.message || t('No data')} + {firstPage?.message || t('No data')}

) } else if (record) { @@ -105,12 +123,19 @@ export function SessionDetailSheet(props: SessionDetailSheetProps) {
{record.session.model_name}} + value={ + + {record.session.model_name} + + } /> - +
- {turns.length === 0 ? ( + {transcript.length === 0 ? (

{t('No data')}

) : (
- {turns.map((view) => ( + {query.hasNextPage && ( + + )} + {transcript.map((view) => (
diff --git a/web/src/features/chat-logs/types.ts b/web/src/features/chat-logs/types.ts index 3bde03613141..7cd7b7662d65 100644 --- a/web/src/features/chat-logs/types.ts +++ b/web/src/features/chat-logs/types.ts @@ -45,21 +45,33 @@ export interface ChatTurn { export interface SessionDetail { session: ChatSessionMeta turns: ChatTurn[] + has_more: boolean + next_turn_id: number | null } export interface GetChatSessionsParams { - page?: number - page_size?: number + limit?: number + cursor?: string token_id?: number user_id?: number model_name?: string + start_ts?: number + end_ts?: number } export interface SessionListResponse { success: boolean message?: string - data?: ChatSessionMeta[] - total: number + data?: { + items: ChatSessionMeta[] + has_more: boolean + next_cursor: string | null + } +} + +export interface GetSessionDetailParams { + limit?: number + before_id?: number } export interface SessionDetailResponse { diff --git a/web/src/i18n/locales/en.json b/web/src/i18n/locales/en.json index acb6828bc84d..42a7cec51597 100644 --- a/web/src/i18n/locales/en.json +++ b/web/src/i18n/locales/en.json @@ -2579,6 +2579,8 @@ "LLM Leaderboard": "LLM Leaderboard", "LLM prompt helper": "LLM prompt helper", "Load Balancing": "Load Balancing", + "Load earlier turns": "Load earlier turns", + "Load more": "Load more", "Load preset": "Load preset", "Load template...": "Load template...", "Loader": "Loader", diff --git a/web/src/i18n/locales/fr.json b/web/src/i18n/locales/fr.json index 9b16835a7bc5..3eb5f65495b2 100644 --- a/web/src/i18n/locales/fr.json +++ b/web/src/i18n/locales/fr.json @@ -2579,6 +2579,8 @@ "LLM Leaderboard": "Classement des LLM", "LLM prompt helper": "Assistant prompt LLM", "Load Balancing": "Équilibrage de charge", + "Load earlier turns": "Charger les tours précédents", + "Load more": "Charger plus", "Load preset": "Charger un préréglage", "Load template...": "Charger le modèle...", "Loader": "Chargeur", diff --git a/web/src/i18n/locales/ja.json b/web/src/i18n/locales/ja.json index 7d035ed28463..a34b30716960 100644 --- a/web/src/i18n/locales/ja.json +++ b/web/src/i18n/locales/ja.json @@ -2579,6 +2579,8 @@ "LLM Leaderboard": "LLM リーダーボード", "LLM prompt helper": "LLMプロンプトヘルパー", "Load Balancing": "ロードバランシング", + "Load earlier turns": "前のターンを読み込む", + "Load more": "さらに読み込む", "Load preset": "プリセットを読み込む", "Load template...": "テンプレートをロード...", "Loader": "ローダー", diff --git a/web/src/i18n/locales/ru.json b/web/src/i18n/locales/ru.json index b99d536903f7..41d0db971210 100644 --- a/web/src/i18n/locales/ru.json +++ b/web/src/i18n/locales/ru.json @@ -2579,6 +2579,8 @@ "LLM Leaderboard": "Рейтинг LLM", "LLM prompt helper": "Помощник с промптом для LLM", "Load Balancing": "Балансировка нагрузки", + "Load earlier turns": "Загрузить более ранние ходы", + "Load more": "Загрузить ещё", "Load preset": "Загрузить пресет", "Load template...": "Загрузить шаблон...", "Loader": "Загрузчик", diff --git a/web/src/i18n/locales/vi.json b/web/src/i18n/locales/vi.json index c4c520e62ccb..2b474e4d1b21 100644 --- a/web/src/i18n/locales/vi.json +++ b/web/src/i18n/locales/vi.json @@ -2579,6 +2579,8 @@ "LLM Leaderboard": "Bảng xếp hạng LLM", "LLM prompt helper": "Trợ lý prompt LLM", "Load Balancing": "Tải cân bằng", + "Load earlier turns": "Tải các lượt cũ hơn", + "Load more": "Tải thêm", "Load preset": "Tải cấu hình mẫu", "Load template...": "Tải mẫu...", "Loader": "Trình tải", diff --git a/web/src/i18n/locales/zh-TW.json b/web/src/i18n/locales/zh-TW.json index fe4bb552a318..afe6540505e8 100644 --- a/web/src/i18n/locales/zh-TW.json +++ b/web/src/i18n/locales/zh-TW.json @@ -2579,6 +2579,8 @@ "LLM Leaderboard": "LLM 排行榜", "LLM prompt helper": "LLM 輔助設計提示詞", "Load Balancing": "負載平衡", + "Load earlier turns": "載入更早的對話", + "Load more": "載入更多", "Load preset": "載入預設", "Load template...": "載入模板...", "Loader": "載入器", diff --git a/web/src/i18n/locales/zh.json b/web/src/i18n/locales/zh.json index 7a30c4e54fe1..b36e6122eabb 100644 --- a/web/src/i18n/locales/zh.json +++ b/web/src/i18n/locales/zh.json @@ -2579,6 +2579,8 @@ "LLM Leaderboard": "LLM 排行榜", "LLM prompt helper": "LLM 辅助设计提示词", "Load Balancing": "负载均衡", + "Load earlier turns": "加载更早的对话", + "Load more": "加载更多", "Load preset": "加载预设", "Load template...": "加载模板...", "Loader": "加载器", diff --git a/web/src/routes/_authenticated/chat-logs/index.tsx b/web/src/routes/_authenticated/chat-logs/index.tsx index b22fec051688..b862ce167d66 100644 --- a/web/src/routes/_authenticated/chat-logs/index.tsx +++ b/web/src/routes/_authenticated/chat-logs/index.tsx @@ -24,7 +24,6 @@ import { ROLE } from '@/lib/roles' import { useAuthStore } from '@/stores/auth-store' const chatLogsSearchSchema = z.object({ - page: z.number().optional().catch(1), token_id: z.number().optional().catch(undefined), user_id: z.number().optional().catch(undefined), model_name: z.string().optional().catch(''),