package ctxmgr import ( "context" "encoding/json" "github.com/jackc/pgx/v5" ) const chatMessageSelectColumnsLegacy = ` id, session_id, COALESCE(user_id, 0), role, content, intent, branch, metadata, COALESCE(day_key::text, ''), category, COALESCE(content_kind, ''), COALESCE(export_policy, 'none'), COALESCE(content_revision, 1), COALESCE(content_sha256, ''), created_at ` const chatMessageSelectColumnsScoped = ` id, session_id, COALESCE(user_id, 0), role, content, intent, branch, metadata, COALESCE(day_key::text, ''), category, COALESCE(content_kind, ''), COALESCE(export_policy, 'none'), COALESCE(content_revision, 1), COALESCE(content_sha256, ''), COALESCE(workflow_instance_id, 0), COALESCE(stage_instance_id, 0), COALESCE(stage_key, ''), COALESCE(scope_kind, 'legacy_unscoped'), created_at ` func (m *ContextManager) hasMessageScopeColumns(ctx context.Context) bool { if m == nil || m.pool == nil { return false } m.messageScopeOnce.Do(func() { var ok bool _ = m.pool.QueryRow(ctx, ` SELECT EXISTS ( SELECT 1 FROM information_schema.columns WHERE table_schema = 'public' AND table_name = 'chat_messages' AND column_name = 'scope_kind' ) `).Scan(&ok) m.messageScopeOK = ok }) return m.messageScopeOK } func (m *ContextManager) chatMessageSelectColumns(ctx context.Context) string { if m.hasMessageScopeColumns(ctx) { return chatMessageSelectColumnsScoped } return chatMessageSelectColumnsLegacy } func appendChatMessagesFromRows(rows pgx.Rows, scoped bool) ([]ChatMessageRecord, error) { var out []ChatMessageRecord for rows.Next() { rec, err := scanChatMessageRow(rows, scoped) if err != nil { return nil, err } out = append(out, rec) } return out, rows.Err() } func scanChatMessageRow(row pgx.Row, scoped bool) (ChatMessageRecord, error) { var r ChatMessageRecord var metaBytes []byte var err error if scoped { err = row.Scan( &r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaBytes, &r.DayKey, &r.Category, &r.ContentKind, &r.ExportPolicy, &r.ContentRevision, &r.ContentSha256, &r.WorkflowInstanceID, &r.StageInstanceID, &r.StageKey, &r.ScopeKind, &r.CreatedAt, ) } else { err = row.Scan( &r.ID, &r.SessionID, &r.UserID, &r.Role, &r.Content, &r.Intent, &r.Branch, &metaBytes, &r.DayKey, &r.Category, &r.ContentKind, &r.ExportPolicy, &r.ContentRevision, &r.ContentSha256, &r.CreatedAt, ) } if err != nil { return r, err } if len(metaBytes) > 0 { _ = json.Unmarshal(metaBytes, &r.Metadata) } if r.ScopeKind == "" { _, _, _, r.ScopeKind = scopeFieldsFromMeta(r.Metadata) } if r.StageKey == "" { if sk, ok := r.Metadata["stage_key"].(string); ok { r.StageKey = sk } } enrichMessageExportCapability(&r) return r, nil } func nullScopeInt64(v int64) any { if v <= 0 { return nil } return v } func nullScopeText(v string) any { if v == "" { return nil } return v } func scopeFieldsFromMeta(meta map[string]any) (workflowInstanceID, stageInstanceID int64, stageKey, scopeKind string) { if meta == nil { return 0, 0, "", "legacy_unscoped" } if sk, ok := meta["scope_kind"].(string); ok && sk != "" { scopeKind = sk } else { scopeKind = "legacy_unscoped" } if v, ok := meta["workflow_instance_id"].(float64); ok { workflowInstanceID = int64(v) } else if v, ok := meta["workflow_instance_id"].(int64); ok { workflowInstanceID = v } if v, ok := meta["stage_instance_id"].(float64); ok { stageInstanceID = int64(v) } else if v, ok := meta["stage_instance_id"].(int64); ok { stageInstanceID = v } if sk, ok := meta["stage_key"].(string); ok { stageKey = sk } return workflowInstanceID, stageInstanceID, stageKey, scopeKind } // ListSessionMessagesForAuth 列出会话消息轻量字段(用于授权统计同集)。 func (m *ContextManager) ListSessionMessagesForAuth(ctx context.Context, sessionID string) ([]ChatMessageRecord, error) { if sessionID == "" { return nil, nil } if m.pool == nil { m.mu.RLock() defer m.mu.RUnlock() raw := m.messages[sessionID] out := make([]ChatMessageRecord, 0, len(raw)) for _, msg := range raw { if msg.DeletedAt != nil { continue } out = append(out, msg) } return out, nil } scoped := m.hasMessageScopeColumns(ctx) rows, err := m.pool.Query(ctx, ` SELECT `+m.chatMessageSelectColumns(ctx)+` FROM chat_messages WHERE session_id = $1 AND deleted_at IS NULL ORDER BY id ASC `, sessionID) if err != nil { return nil, err } defer rows.Close() return appendChatMessagesFromRows(rows, scoped) }