| 1 | package sessioncatalog |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "strings" |
| 7 | ) |
| 8 | |
| 9 | func (c *Catalog) ListTopics(ctx context.Context, req TopicPageRequest) (TopicPage, error) { |
| 10 | out := TopicPage{Items: []TopicRecord{}, Revision: c.revision.Load()} |
| 11 | req.Scope, req.WorkspaceRoot = normalizeScope(req.Scope, req.WorkspaceRoot) |
| 12 | if req.Limit <= 0 { |
| 13 | req.Limit = DefaultLimit |
| 14 | } |
| 15 | if req.Limit > MaxLimit { |
| 16 | req.Limit = MaxLimit |
| 17 | } |
| 18 | cursor, err := decodeCursor(req.Cursor) |
| 19 | if err != nil { |
| 20 | return out, err |
| 21 | } |
| 22 | if cursor != nil && cursor.ManualOrder != req.ManualOrder { |
| 23 | return out, errCursorSortModeChanged |
| 24 | } |
| 25 | if cursor != nil && cursor.Binding != req.CursorBinding { |
| 26 | return out, errCursorSortModeChanged |
| 27 | } |
| 28 | rootKey := c.workspaceRootKey(req.Scope, req.WorkspaceRoot) |
| 29 | args := []any{req.Scope, rootKey} |
| 30 | where := `scope=? AND workspace_root_key=?` |
| 31 | if query := strings.TrimSpace(req.Query); query != "" { |
| 32 | where += ` AND lower(title) LIKE ?` |
| 33 | args = append(args, "%"+strings.ToLower(query)+"%") |
| 34 | } |
| 35 | if cutoff := timeFilterCutoff(req.TimeFilter, c.opts.Now()); cutoff > 0 { |
| 36 | where += ` AND last_activity_at>=?` |
| 37 | args = append(args, cutoff) |
| 38 | } |
| 39 | where += ` AND (?='' OR topic_id IN (SELECT value FROM json_each(COALESCE(NULLIF(?,''),'[]'))))` + |
| 40 | ` AND (?='' OR topic_id NOT IN (SELECT value FROM json_each(COALESCE(NULLIF(?,''),'[]'))))` + |
| 41 | ` AND (?=0 OR pinned=0)` |
| 42 | args = append(args, req.IncludeTopicIDsJSON, req.IncludeTopicIDsJSON, |
| 43 | req.ExcludeTopicIDsJSON, req.ExcludeTopicIDsJSON, req.ExcludePinned) |
| 44 | scanCursor := cursor |
| 45 | scanLimit := max(req.Limit+1, 64) |
| 46 | for len(out.Items) <= req.Limit { |
| 47 | query, pageArgs := topicPageQuery(req, where, args, scanCursor, scanLimit) |
| 48 | rows, queryErr := c.db.QueryContext(ctx, query, pageArgs...) |
| 49 | if queryErr != nil { |
| 50 | return out, queryErr |
| 51 | } |
| 52 | scanned, scanErr := scanTopicRows(rows, scanLimit) |
| 53 | if scanErr != nil { |
| 54 | return out, scanErr |
| 55 | } |
| 56 | rawCount := len(scanned) |
| 57 | overflow := false |
| 58 | for _, item := range scanned { |
| 59 | sessions, listErr := c.listTopicSessionsByRootKey(ctx, TopicKey{ |
| 60 | Scope: item.Scope, WorkspaceRoot: item.WorkspaceRoot, TopicID: item.TopicID, |
| 61 | }, rootKey) |
| 62 | if listErr != nil { |
| 63 | return TopicPage{Items: []TopicRecord{}, Revision: out.Revision}, listErr |
| 64 | } |
| 65 | if len(sessions) == 0 { |
| 66 | continue |
| 67 | } |
| 68 | // Skip recovery shells that lost their ordinary representative while |
| 69 | // lineage is re-anchored, unless the topic is explicitly pinned. |
| 70 | hasOrdinary := false |
| 71 | for _, session := range sessions { |
| 72 | if session.OrdinaryVisible || (!session.Recovered && !session.RecoveryCopy) { |
| 73 | hasOrdinary = true |
| 74 | break |
| 75 | } |
| 76 | } |
| 77 | if !hasOrdinary && !item.Pinned { |
| 78 | continue |
| 79 | } |
| 80 | item.Sessions = sessions |
| 81 | hydrateTopicDisplay(&item) |
| 82 | out.Items = append(out.Items, item) |
| 83 | if len(out.Items) > req.Limit { |
| 84 | overflow = true |
| 85 | break |
| 86 | } |
| 87 | } |
| 88 | if overflow || rawCount < scanLimit || rawCount == 0 { |
| 89 | break |
| 90 | } |
| 91 | scanCursor = cursorForTopic(scanned[rawCount-1], req) |
| 92 | } |
| 93 | more := len(out.Items) > req.Limit |
| 94 | if more { |
| 95 | out.Items = out.Items[:req.Limit] |
| 96 | } |
| 97 | if more && len(out.Items) > 0 { |
| 98 | out.NextCursor = encodeCursor(*cursorForTopic(out.Items[len(out.Items)-1], req)) |
| 99 | } |
| 100 | return out, nil |
| 101 | } |
| 102 | |
| 103 | func topicPageQuery(req TopicPageRequest, where string, args []any, cursor *pageCursor, limit int) (string, []any) { |
| 104 | sortExpression := topicPageSortExpression(req.SortMode) |
| 105 | manualSortExpression := topicPageManualSortExpression() |
| 106 | pageArgs := append([]any(nil), args...) |
| 107 | if cursor != nil && req.ManualOrder { |
| 108 | where += ` AND (pinned<? OR (pinned=? AND ` + manualSortExpression + `>?) OR ` + |
| 109 | `(pinned=? AND ` + manualSortExpression + `=? AND ` + sortExpression + `<?) OR ` + |
| 110 | `(pinned=? AND ` + manualSortExpression + `=? AND ` + sortExpression + `=? AND topic_id>?))` |
| 111 | pageArgs = append(pageArgs, |
| 112 | cursor.Pinned, |
| 113 | cursor.Pinned, cursor.SortOrder, |
| 114 | cursor.Pinned, cursor.SortOrder, cursor.Activity, |
| 115 | cursor.Pinned, cursor.SortOrder, cursor.Activity, cursor.TopicID, |
| 116 | ) |
| 117 | } else if cursor != nil { |
| 118 | where += ` AND (pinned<? OR (pinned=? AND ` + sortExpression + `<?) OR (pinned=? AND ` + sortExpression + `=? AND topic_id>?))` |
| 119 | pageArgs = append(pageArgs, cursor.Pinned, cursor.Pinned, cursor.Activity, |
| 120 | cursor.Pinned, cursor.Activity, cursor.TopicID) |
| 121 | } |
| 122 | orderBy := `pinned DESC,` + sortExpression + ` DESC,topic_id ASC` |
| 123 | if req.ManualOrder { |
| 124 | orderBy = `pinned DESC,` + manualSortExpression + ` ASC,` + sortExpression + ` DESC,topic_id ASC` |
| 125 | } |
| 126 | pageArgs = append(pageArgs, limit) |
| 127 | return `SELECT scope,workspace_root,topic_id,title,title_source,pinned, |
| 128 | CASE WHEN metadata_present=1 THEN sort_order ELSE -1 END, |
| 129 | turns,turns_state,created_at,last_activity_at,recovery_state,recovery_branch_count, |
| 130 | recovery_unresolved_count,recovery_cleanup_eligible_count,health |
| 131 | FROM catalog_topics WHERE ` + where + ` ORDER BY ` + orderBy + ` LIMIT ?`, pageArgs |
| 132 | } |
| 133 | |
| 134 | func scanTopicRows(rows *sql.Rows, capacity int) ([]TopicRecord, error) { |
| 135 | defer rows.Close() |
| 136 | // Drain before hydrating sessions: the nested read needs another connection |
| 137 | // and an open cursor deadlocks when the in-memory pool is saturated. |
| 138 | scanned := make([]TopicRecord, 0, capacity) |
| 139 | for rows.Next() { |
| 140 | var item TopicRecord |
| 141 | if err := rows.Scan(&item.Scope, &item.WorkspaceRoot, &item.TopicID, &item.Title, |
| 142 | &item.TitleSource, &item.Pinned, &item.SortOrder, &item.Turns, &item.TurnsState, |
| 143 | &item.CreatedAt, &item.LastActivityAt, &item.RecoveryState, &item.RecoveryBranchCount, |
| 144 | &item.RecoveryUnresolvedCount, &item.RecoveryCleanupEligibleCount, &item.Health); err != nil { |
| 145 | return nil, err |
| 146 | } |
| 147 | scanned = append(scanned, item) |
| 148 | } |
| 149 | if err := rows.Err(); err != nil { |
| 150 | return nil, err |
| 151 | } |
| 152 | return scanned, nil |
| 153 | } |
| 154 | |
| 155 | func cursorForTopic(topic TopicRecord, req TopicPageRequest) *pageCursor { |
| 156 | pinned := 0 |
| 157 | if topic.Pinned { |
| 158 | pinned = 1 |
| 159 | } |
| 160 | return &pageCursor{ |
| 161 | Pinned: pinned, ManualOrder: req.ManualOrder, |
| 162 | SortOrder: topicPageManualSortValue(topic), |
| 163 | Activity: topicPageSortValue(topic, req.SortMode), TopicID: topic.TopicID, |
| 164 | Binding: req.CursorBinding, |
| 165 | } |
| 166 | } |
| 167 |