package ctxmgr import ( "context" "errors" "fmt" "sort" "time" "chat-agent-system/internal/replyexport" ) const ( DefaultWindowDays = 7 MaxWindowDays = 31 MaxMsgPerDay = 300 ) // GetChatWindow 获取指定会话的按日分片窗口 (倒序取日集合 → 每日取最新 MAX_MSG_PER_DAY 条 → 正序返回) func (m *ContextManager) GetChatWindow(ctx context.Context, q ChatWindowQuery) (*ChatWindowResult, error) { if q.SessionID == "" { return nil, errors.New("session_id required") } if q.Days <= 0 { q.Days = DefaultWindowDays } if q.Days > MaxWindowDays { q.Days = MaxWindowDays } if m.pool != nil { return m.getChatWindowPG(ctx, q) } return m.getChatWindowMem(ctx, q) } func (m *ContextManager) getChatWindowPG(ctx context.Context, q ChatWindowQuery) (*ChatWindowResult, error) { // 1. 查询目标日集合 (DESC 排序取最近 N 日) var dayRowsQuery string var dayRowsArgs []any if q.BeforeDay != "" { dayRowsQuery = ` SELECT DISTINCT day_key FROM chat_messages WHERE session_id = $1 AND deleted_at IS NULL AND day_key IS NOT NULL AND day_key < $2::date ORDER BY day_key DESC LIMIT $3 ` dayRowsArgs = []any{q.SessionID, q.BeforeDay, q.Days} } else { dayRowsQuery = ` SELECT DISTINCT day_key FROM chat_messages WHERE session_id = $1 AND deleted_at IS NULL AND day_key IS NOT NULL ORDER BY day_key DESC LIMIT $2 ` dayRowsArgs = []any{q.SessionID, q.Days} } rows, err := m.pool.Query(ctx, dayRowsQuery, dayRowsArgs...) if err != nil { return nil, fmt.Errorf("query distinct days failed: %w", err) } defer rows.Close() var descDays []string for rows.Next() { var d time.Time if err := rows.Scan(&d); err == nil { descDays = append(descDays, d.Format("2006-01-02")) } } if err := rows.Err(); err != nil { return nil, fmt.Errorf("scan chat window days failed: %w", err) } if len(descDays) == 0 { return &ChatWindowResult{ SessionID: q.SessionID, Days: []ChatDayBucket{}, HasMoreBefore: false, OldestLoadedDay: "", }, nil } oldestDay := descDays[len(descDays)-1] // 2. 检查是否有更早日 var hasMoreBefore bool err = m.pool.QueryRow(ctx, ` SELECT EXISTS( SELECT 1 FROM chat_messages WHERE session_id = $1 AND deleted_at IS NULL AND day_key IS NOT NULL AND day_key < $2::date ) `, q.SessionID, oldestDay).Scan(&hasMoreBefore) if err != nil { hasMoreBefore = false } // 3. 对选中的每一个日拉取统计和最新 MAX_MSG_PER_DAY 条 buckets := make([]ChatDayBucket, 0, len(descDays)) for _, dayStr := range descDays { var totalCount int var firstID, lastID *int64 _ = m.pool.QueryRow(ctx, ` SELECT count(*), MIN(id), MAX(id) FROM chat_messages WHERE session_id = $1 AND day_key = $2::date AND deleted_at IS NULL `, q.SessionID, dayStr).Scan(&totalCount, &firstID, &lastID) fID := int64(0) lID := int64(0) if firstID != nil { fID = *firstID } if lastID != nil { lID = *lastID } // 倒序取最多 MaxMsgPerDay 条 scoped := m.hasMessageScopeColumns(ctx) msgRows, err := m.pool.Query(ctx, ` SELECT `+m.chatMessageSelectColumns(ctx)+` FROM chat_messages WHERE session_id = $1 AND day_key = $2::date AND deleted_at IS NULL ORDER BY id DESC LIMIT $3 `, q.SessionID, dayStr, MaxMsgPerDay) if err != nil { return nil, fmt.Errorf("query messages for day %s failed: %w", dayStr, err) } msgsDesc, _ := appendChatMessagesFromRows(msgRows, scoped) msgRows.Close() // 内存反转为正序 ASC msgsAsc := make([]ChatMessageRecord, len(msgsDesc)) for i, mRec := range msgsDesc { msgsAsc[len(msgsDesc)-1-i] = mRec } bucket := ChatDayBucket{ DayKey: dayStr, Count: totalCount, LoadedCount: len(msgsAsc), Truncated: totalCount > MaxMsgPerDay, FirstMessageID: fID, LastMessageID: lID, Messages: msgsAsc, } buckets = append(buckets, bucket) } // 4. 将日集合反转为正序 (最旧自然日排在最前面,供前端自顶向下按自然日渲染) sort.Slice(buckets, func(i, j int) bool { return buckets[i].DayKey < buckets[j].DayKey }) return &ChatWindowResult{ SessionID: q.SessionID, Days: buckets, HasMoreBefore: hasMoreBefore, OldestLoadedDay: oldestDay, }, nil } func (m *ContextManager) getChatWindowMem(ctx context.Context, q ChatWindowQuery) (*ChatWindowResult, error) { m.mu.RLock() defer m.mu.RUnlock() allMsgs, exists := m.messages[q.SessionID] if !exists || len(allMsgs) == 0 { return &ChatWindowResult{ SessionID: q.SessionID, Days: []ChatDayBucket{}, HasMoreBefore: false, OldestLoadedDay: "", }, nil } // 按 dayKey 分组 dayMap := make(map[string][]ChatMessageRecord) for _, msg := range allMsgs { if msg.DeletedAt != nil { continue } dKey := msg.DayKey if dKey == "" { dKey = msg.CreatedAt.In(time.FixedZone("CST", 8*3600)).Format("2006-01-02") } dayMap[dKey] = append(dayMap[dKey], msg) } var allDays []string for k := range dayMap { allDays = append(allDays, k) } sort.Slice(allDays, func(i, j int) bool { return allDays[i] > allDays[j] // DESC }) var filteredDays []string for _, d := range allDays { if q.BeforeDay != "" && d >= q.BeforeDay { continue } filteredDays = append(filteredDays, d) } if len(filteredDays) == 0 { return &ChatWindowResult{ SessionID: q.SessionID, Days: []ChatDayBucket{}, HasMoreBefore: false, OldestLoadedDay: "", }, nil } hasMoreBefore := false if len(filteredDays) > q.Days { hasMoreBefore = true filteredDays = filteredDays[:q.Days] } oldestDay := filteredDays[len(filteredDays)-1] buckets := make([]ChatDayBucket, 0, len(filteredDays)) for _, dayStr := range filteredDays { dMsgs := dayMap[dayStr] totalCount := len(dMsgs) var fID, lID int64 if totalCount > 0 { fID = dMsgs[0].ID lID = dMsgs[totalCount-1].ID } // 取最后 MAX_MSG_PER_DAY 条 start := 0 truncated := false if totalCount > MaxMsgPerDay { start = totalCount - MaxMsgPerDay truncated = true } selected := dMsgs[start:] copied := make([]ChatMessageRecord, len(selected)) copy(copied, selected) for i := range copied { enrichMessageExportCapability(&copied[i]) } buckets = append(buckets, ChatDayBucket{ DayKey: dayStr, Count: totalCount, LoadedCount: len(copied), Truncated: truncated, FirstMessageID: fID, LastMessageID: lID, Messages: copied, }) } // 排序为 ASC sort.Slice(buckets, func(i, j int) bool { return buckets[i].DayKey < buckets[j].DayKey }) return &ChatWindowResult{ SessionID: q.SessionID, Days: buckets, HasMoreBefore: hasMoreBefore, OldestLoadedDay: oldestDay, }, nil } // GetChatDayMessages 获取指定日消息,支持 anchor_id 上下文补齐或基于游标的分页拉取 func (m *ContextManager) GetChatDayMessages(ctx context.Context, sessionID, dayKey string, anchorID int64, beforeID, afterID int64, limit int) (*ChatDayMessagesResult, error) { if sessionID == "" { return nil, errors.New("session_id required") } if dayKey == "" { if anchorID <= 0 { return nil, errors.New("date or anchor_id required") } if m.pool != nil { var resolvedDay string err := m.pool.QueryRow(ctx, ` SELECT COALESCE(day_key::text, '') FROM chat_messages WHERE id = $1 AND session_id = $2 AND deleted_at IS NULL `, anchorID, sessionID).Scan(&resolvedDay) if err != nil || resolvedDay == "" { return nil, fmt.Errorf("message with anchor_id %d not found in session", anchorID) } dayKey = resolvedDay } else { m.mu.RLock() for _, msg := range m.messages[sessionID] { if msg.ID == anchorID && msg.DeletedAt == nil { dayKey = msg.DayKey if dayKey == "" { dayKey = msg.CreatedAt.In(time.FixedZone("CST", 8*3600)).Format("2006-01-02") } break } } m.mu.RUnlock() if dayKey == "" { return nil, fmt.Errorf("message with anchor_id %d not found in session", anchorID) } } } if limit <= 0 { limit = 100 } if limit > MaxMsgPerDay { limit = MaxMsgPerDay } if m.pool != nil { return m.getChatDayMessagesPG(ctx, sessionID, dayKey, anchorID, beforeID, afterID, limit) } return m.getChatDayMessagesMem(ctx, sessionID, dayKey, anchorID, beforeID, afterID, limit) } func (m *ContextManager) getChatDayMessagesPG(ctx context.Context, sessionID, dayKey string, anchorID int64, beforeID, afterID int64, limit int) (*ChatDayMessagesResult, error) { var totalCount int _ = m.pool.QueryRow(ctx, ` SELECT count(*) FROM chat_messages WHERE session_id = $1 AND day_key = $2::date AND deleted_at IS NULL `, sessionID, dayKey).Scan(&totalCount) var msgs []ChatMessageRecord var hasMoreBefore, hasMoreAfter bool scoped := m.hasMessageScopeColumns(ctx) cols := m.chatMessageSelectColumns(ctx) if anchorID > 0 { // 锚点上下文:前后各取 limit/2 条 half := limit / 2 if half < 1 { half = 1 } // before or equal to anchor rowsBefore, err := m.pool.Query(ctx, ` SELECT `+cols+` FROM chat_messages WHERE session_id = $1 AND day_key = $2::date AND id <= $3 AND deleted_at IS NULL ORDER BY id DESC LIMIT $4 `, sessionID, dayKey, anchorID, half) if err == nil { beforeDesc, _ := appendChatMessagesFromRows(rowsBefore, scoped) rowsBefore.Close() for i := len(beforeDesc) - 1; i >= 0; i-- { msgs = append(msgs, beforeDesc[i]) } } // after anchor rowsAfter, err := m.pool.Query(ctx, ` SELECT `+cols+` FROM chat_messages WHERE session_id = $1 AND day_key = $2::date AND id > $3 AND deleted_at IS NULL ORDER BY id ASC LIMIT $4 `, sessionID, dayKey, anchorID, half) if err == nil { afterMsgs, _ := appendChatMessagesFromRows(rowsAfter, scoped) rowsAfter.Close() msgs = append(msgs, afterMsgs...) } } else if beforeID > 0 { rows, err := m.pool.Query(ctx, ` SELECT `+cols+` FROM chat_messages WHERE session_id = $1 AND day_key = $2::date AND id < $3 AND deleted_at IS NULL ORDER BY id DESC LIMIT $4 `, sessionID, dayKey, beforeID, limit) if err == nil { desc, _ := appendChatMessagesFromRows(rows, scoped) rows.Close() for i := len(desc) - 1; i >= 0; i-- { msgs = append(msgs, desc[i]) } } } else if afterID > 0 { rows, err := m.pool.Query(ctx, ` SELECT `+cols+` FROM chat_messages WHERE session_id = $1 AND day_key = $2::date AND id > $3 AND deleted_at IS NULL ORDER BY id ASC LIMIT $4 `, sessionID, dayKey, afterID, limit) if err == nil { afterMsgs, _ := appendChatMessagesFromRows(rows, scoped) rows.Close() msgs = append(msgs, afterMsgs...) } } else { // 默认从当日第一条起正序 limit 条 rows, err := m.pool.Query(ctx, ` SELECT `+cols+` FROM chat_messages WHERE session_id = $1 AND day_key = $2::date AND deleted_at IS NULL ORDER BY id ASC LIMIT $3 `, sessionID, dayKey, limit) if err == nil { defaultMsgs, _ := appendChatMessagesFromRows(rows, scoped) rows.Close() msgs = append(msgs, defaultMsgs...) } } if len(msgs) > 0 { minID := msgs[0].ID maxID := msgs[len(msgs)-1].ID _ = m.pool.QueryRow(ctx, ` SELECT EXISTS(SELECT 1 FROM chat_messages WHERE session_id=$1 AND day_key=$2::date AND id < $3 AND deleted_at IS NULL) `, sessionID, dayKey, minID).Scan(&hasMoreBefore) _ = m.pool.QueryRow(ctx, ` SELECT EXISTS(SELECT 1 FROM chat_messages WHERE session_id=$1 AND day_key=$2::date AND id > $3 AND deleted_at IS NULL) `, sessionID, dayKey, maxID).Scan(&hasMoreAfter) } return &ChatDayMessagesResult{ SessionID: sessionID, DayKey: dayKey, TotalCount: totalCount, LoadedCount: len(msgs), HasMoreBefore: hasMoreBefore, HasMoreAfter: hasMoreAfter, Messages: msgs, }, nil } func (m *ContextManager) getChatDayMessagesMem(ctx context.Context, sessionID, dayKey string, anchorID int64, beforeID, afterID int64, limit int) (*ChatDayMessagesResult, error) { m.mu.RLock() defer m.mu.RUnlock() allMsgs := m.messages[sessionID] var dayMsgs []ChatMessageRecord for _, msg := range allMsgs { if msg.DeletedAt != nil { continue } dKey := msg.DayKey if dKey == "" { dKey = msg.CreatedAt.In(time.FixedZone("CST", 8*3600)).Format("2006-01-02") } if dKey == dayKey { dayMsgs = append(dayMsgs, msg) } } totalCount := len(dayMsgs) if totalCount == 0 { return &ChatDayMessagesResult{ SessionID: sessionID, DayKey: dayKey, TotalCount: 0, LoadedCount: 0, HasMoreBefore: false, HasMoreAfter: false, Messages: []ChatMessageRecord{}, }, nil } var selected []ChatMessageRecord if anchorID > 0 { anchorIdx := -1 for i, msg := range dayMsgs { if msg.ID == anchorID { anchorIdx = i break } } if anchorIdx < 0 { // 若未精确找到 anchorID,按接近位置居中截取 for i, msg := range dayMsgs { if msg.ID >= anchorID { anchorIdx = i break } } if anchorIdx < 0 { anchorIdx = len(dayMsgs) - 1 } } half := limit / 2 start := anchorIdx - half if start < 0 { start = 0 } end := start + limit if end > len(dayMsgs) { end = len(dayMsgs) } selected = dayMsgs[start:end] } else if beforeID > 0 { endIdx := -1 for i, msg := range dayMsgs { if msg.ID >= beforeID { endIdx = i break } } if endIdx < 0 { endIdx = len(dayMsgs) } start := endIdx - limit if start < 0 { start = 0 } selected = dayMsgs[start:endIdx] } else if afterID > 0 { startIdx := -1 for i, msg := range dayMsgs { if msg.ID > afterID { startIdx = i break } } if startIdx >= 0 { end := startIdx + limit if end > len(dayMsgs) { end = len(dayMsgs) } selected = dayMsgs[startIdx:end] } } else { end := limit if end > len(dayMsgs) { end = len(dayMsgs) } selected = dayMsgs[:end] } copied := make([]ChatMessageRecord, len(selected)) copy(copied, selected) hasMoreBefore := false hasMoreAfter := false if len(copied) > 0 { hasMoreBefore = copied[0].ID > dayMsgs[0].ID hasMoreAfter = copied[len(copied)-1].ID < dayMsgs[len(dayMsgs)-1].ID } return &ChatDayMessagesResult{ SessionID: sessionID, DayKey: dayKey, TotalCount: totalCount, LoadedCount: len(copied), HasMoreBefore: hasMoreBefore, HasMoreAfter: hasMoreAfter, Messages: copied, }, nil } // ListChatDates 列出会话有消息的所有日期及分类统计 (日历打点使用) func (m *ContextManager) ListChatDates(ctx context.Context, sessionID string) ([]ChatDateStat, error) { if sessionID == "" { return nil, errors.New("session_id required") } if m.pool != nil { rows, err := m.pool.Query(ctx, ` SELECT day_key::text, count(*), category FROM chat_messages WHERE session_id = $1 AND deleted_at IS NULL AND day_key IS NOT NULL GROUP BY day_key, category ORDER BY day_key ASC `, sessionID) if err != nil { return nil, fmt.Errorf("list chat dates failed: %w", err) } defer rows.Close() statMap := make(map[string]*ChatDateStat) var orderedKeys []string for rows.Next() { var dayStr, cat string var cnt int if err := rows.Scan(&dayStr, &cnt, &cat); err == nil { st, ok := statMap[dayStr] if !ok { st = &ChatDateStat{ DayKey: dayStr, Count: 0, Categories: make(map[string]int), } statMap[dayStr] = st orderedKeys = append(orderedKeys, dayStr) } st.Count += cnt st.Categories[cat] += cnt } } if err := rows.Err(); err != nil { return nil, fmt.Errorf("scan chat dates failed: %w", err) } res := make([]ChatDateStat, 0, len(orderedKeys)) for _, k := range orderedKeys { res = append(res, *statMap[k]) } return res, nil } m.mu.RLock() defer m.mu.RUnlock() allMsgs := m.messages[sessionID] statMap := make(map[string]*ChatDateStat) var orderedKeys []string for _, msg := range allMsgs { if msg.DeletedAt != nil { continue } dKey := msg.DayKey if dKey == "" { dKey = msg.CreatedAt.In(time.FixedZone("CST", 8*3600)).Format("2006-01-02") } st, ok := statMap[dKey] if !ok { st = &ChatDateStat{ DayKey: dKey, Count: 0, Categories: make(map[string]int), } statMap[dKey] = st orderedKeys = append(orderedKeys, dKey) } st.Count++ cat := msg.Category if cat == "" { cat = CategoryUncategorized } st.Categories[cat]++ } sort.Strings(orderedKeys) res := make([]ChatDateStat, 0, len(orderedKeys)) for _, k := range orderedKeys { res = append(res, *statMap[k]) } return res, nil } // ListChatCategories 统计会话各分类消息数及首尾消息ID (用于分类筛选条) func (m *ContextManager) ListChatCategories(ctx context.Context, sessionID string) ([]ChatCategoryStat, error) { if sessionID == "" { return nil, errors.New("session_id required") } if m.pool != nil { rows, err := m.pool.Query(ctx, ` SELECT category, COUNT(*), MIN(id), MAX(id) FROM chat_messages WHERE session_id = $1 AND deleted_at IS NULL GROUP BY category ORDER BY COUNT(*) DESC `, sessionID) if err != nil { return nil, fmt.Errorf("query chat categories failed: %w", err) } defer rows.Close() var res []ChatCategoryStat for rows.Next() { var cat string var count int var minID, maxID int64 if err := rows.Scan(&cat, &count, &minID, &maxID); err != nil { return nil, fmt.Errorf("scan chat category stat failed: %w", err) } if cat == "" { cat = CategoryUncategorized } res = append(res, ChatCategoryStat{ Category: cat, Label: GetCategoryLabel(cat), Count: count, FirstMessageID: minID, LastMessageID: maxID, }) } return res, rows.Err() } m.mu.RLock() defer m.mu.RUnlock() allMsgs := m.messages[sessionID] statMap := make(map[string]*ChatCategoryStat) for _, msg := range allMsgs { if msg.DeletedAt != nil { continue } cat := msg.Category if cat == "" { cat = CategoryUncategorized } st, ok := statMap[cat] if !ok { st = &ChatCategoryStat{ Category: cat, Label: GetCategoryLabel(cat), Count: 0, FirstMessageID: msg.ID, LastMessageID: msg.ID, } statMap[cat] = st } st.Count++ if msg.ID < st.FirstMessageID { st.FirstMessageID = msg.ID } if msg.ID > st.LastMessageID { st.LastMessageID = msg.ID } } res := make([]ChatCategoryStat, 0, len(statMap)) for _, st := range statMap { res = append(res, *st) } sort.Slice(res, func(i, j int) bool { if res[i].Count != res[j].Count { return res[i].Count > res[j].Count } return res[i].Category < res[j].Category }) return res, nil } func enrichMessageExportCapability(r *ChatMessageRecord) { hasCard := false if r.Metadata != nil { if _, ok := r.Metadata["interactive_card"]; ok { hasCard = true } } evalKind, evalCap := replyexport.EvaluatePolicy(replyexport.PolicyOptions{ Role: r.Role, Status: "completed", Content: r.Content, HasInteractiveCard: hasCard, Revision: r.ContentRevision, }) if r.ContentKind == "" { r.ContentKind = string(evalKind) } if r.ExportPolicy == "" { r.ExportPolicy = string(evalCap.Policy) } if r.ContentRevision <= 0 { r.ContentRevision = 1 } if r.ContentSha256 == "" { r.ContentSha256 = evalCap.ContentSha256 } evalCap.Policy = replyexport.ExportPolicy(r.ExportPolicy) evalCap.ContentRevision = r.ContentRevision evalCap.ContentSha256 = r.ContentSha256 if evalCap.Policy == replyexport.PolicyFull { evalCap.Formats = replyexport.FullFormats() evalCap.ReasonCode = "" } else if evalCap.Policy == replyexport.PolicyCopyOnly { evalCap.Formats = replyexport.CopyOnlyFormats() evalCap.ReasonCode = "" } else { evalCap.Formats = []string{} } r.ExportCapability = evalCap }