| 1 | package sessioncatalog |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/base64" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "strings" |
| 11 | "time" |
| 12 | ) |
| 13 | |
| 14 | type sessionPageCursor struct { |
| 15 | Revision uint64 `json:"r"` |
| 16 | Activity int64 `json:"a"` |
| 17 | Path string `json:"p"` |
| 18 | } |
| 19 | |
| 20 | const sessionSelectColumns = `path,path_key,directory,scope,workspace_root,topic_id,topic_title, |
| 21 | custom_title,created_at,last_activity_at,preview,turns,turns_state,recovered, |
| 22 | recovery_reason,recovery_digest,parent_id,recovery_copy,recovery_group_id, |
| 23 | recovery_role,recovery_canonical,logical_topic_id,ordinary_visible,content_fingerprint, |
| 24 | meta_fingerprint,health,missing_since,log_format,head_count,selected_head_id` |
| 25 | |
| 26 | func scanSession(scanner interface{ Scan(...any) error }) (SessionRecord, error) { |
| 27 | var record SessionRecord |
| 28 | var recoveryCopy, recoveryCanonical, ordinaryVisible int |
| 29 | err := scanner.Scan(&record.Path, &record.pathKey, &record.Directory, &record.Scope, &record.WorkspaceRoot, |
| 30 | &record.TopicID, &record.TopicTitle, &record.CustomTitle, &record.CreatedAt, |
| 31 | &record.LastActivityAt, &record.Preview, &record.Turns, &record.TurnsState, |
| 32 | &record.Recovered, &record.RecoveryReason, &record.RecoveryDigest, |
| 33 | &record.ParentID, &recoveryCopy, &record.RecoveryGroupID, &record.RecoveryRole, |
| 34 | &recoveryCanonical, &record.LogicalTopicID, &ordinaryVisible, &record.ContentFingerprint, &record.MetaFingerprint, |
| 35 | &record.Health, &record.MissingSince, &record.LogFormat, &record.HeadCount, &record.SelectedHeadID) |
| 36 | record.RecoveryCopy = recoveryCopy != 0 |
| 37 | record.RecoveryCanonical = recoveryCanonical != 0 |
| 38 | record.OrdinaryVisible = ordinaryVisible != 0 |
| 39 | if record.LogicalTopicID == "" { |
| 40 | record.LogicalTopicID = record.TopicID |
| 41 | } |
| 42 | if record.RecoveryRole == "" { |
| 43 | if record.RecoveryCopy { |
| 44 | record.RecoveryRole = RecoveryRoleCoveredCopy |
| 45 | } else if record.Recovered { |
| 46 | record.RecoveryRole = RecoveryRoleDiverged |
| 47 | } else { |
| 48 | record.RecoveryRole = RecoveryRoleNormal |
| 49 | } |
| 50 | } |
| 51 | return record, err |
| 52 | } |
| 53 | |
| 54 | // ListSessions returns only catalog metadata. It never opens a transcript or |
| 55 | // sidecar and therefore remains safe on startup and UI pagination paths. |
| 56 | func (c *Catalog) ListSessions(ctx context.Context, req SessionPageRequest) (SessionPage, error) { |
| 57 | out := SessionPage{Items: []SessionRecord{}, Revision: c.revision.Load()} |
| 58 | if req.Limit <= 0 { |
| 59 | req.Limit = DefaultLimit |
| 60 | } |
| 61 | if req.Limit > MaxLimit { |
| 62 | req.Limit = MaxLimit |
| 63 | } |
| 64 | cursor, err := decodeSessionCursor(req.Cursor) |
| 65 | if err != nil { |
| 66 | return out, err |
| 67 | } |
| 68 | if cursor != nil && cursor.Revision != out.Revision { |
| 69 | out.StaleCursor = true |
| 70 | return out, nil |
| 71 | } |
| 72 | where := []string{`missing_since=0`, `health<>'missing'`} |
| 73 | args := []any{} |
| 74 | switch strings.ToLower(strings.TrimSpace(req.Scope)) { |
| 75 | case "", "all": |
| 76 | case "project": |
| 77 | where = append(where, `scope='project'`, `workspace_root_key=?`) |
| 78 | args = append(args, c.workspaceRootKey("project", req.WorkspaceRoot)) |
| 79 | case "global": |
| 80 | where = append(where, `scope='global'`) |
| 81 | default: |
| 82 | return out, fmt.Errorf("invalid session catalog scope %q", req.Scope) |
| 83 | } |
| 84 | if directory := strings.TrimSpace(req.Directory); directory != "" { |
| 85 | where = append(where, `directory_key=?`) |
| 86 | args = append(args, c.pathKey(directory)) |
| 87 | } |
| 88 | if query := strings.ToLower(strings.TrimSpace(req.Query)); query != "" { |
| 89 | where = append(where, `(lower(custom_title) LIKE ? OR lower(preview) LIKE ? OR lower(topic_title) LIKE ? OR lower(topic_id) LIKE ?)`) |
| 90 | like := "%" + query + "%" |
| 91 | args = append(args, like, like, like, like) |
| 92 | } |
| 93 | appendSessionTimeFilter(&where, &args, req.TimeFilter, c.opts.Now()) |
| 94 | scanCursor := cursor |
| 95 | scanLimit := max(req.Limit+1, 64) |
| 96 | for len(out.Items) <= req.Limit { |
| 97 | pageWhere := append([]string(nil), where...) |
| 98 | pageArgs := append([]any(nil), args...) |
| 99 | if scanCursor != nil { |
| 100 | pageWhere = append(pageWhere, `(last_activity_at<? OR (last_activity_at=? AND path>?))`) |
| 101 | pageArgs = append(pageArgs, scanCursor.Activity, scanCursor.Activity, scanCursor.Path) |
| 102 | } |
| 103 | pageArgs = append(pageArgs, scanLimit) |
| 104 | rows, err := c.db.QueryContext(ctx, `SELECT `+sessionSelectColumns+` FROM catalog_sessions WHERE `+ |
| 105 | strings.Join(pageWhere, ` AND `)+` ORDER BY last_activity_at DESC,path ASC LIMIT ?`, pageArgs...) |
| 106 | if err != nil { |
| 107 | return out, err |
| 108 | } |
| 109 | rawCount := 0 |
| 110 | var lastScanned SessionRecord |
| 111 | for rows.Next() { |
| 112 | record, err := scanSession(rows) |
| 113 | if err != nil { |
| 114 | _ = rows.Close() |
| 115 | return out, err |
| 116 | } |
| 117 | rawCount++ |
| 118 | lastScanned = record |
| 119 | if c.pathRemovedKey(record.pathKey, record.Path) { |
| 120 | continue |
| 121 | } |
| 122 | out.Items = append(out.Items, record) |
| 123 | if len(out.Items) > req.Limit { |
| 124 | break |
| 125 | } |
| 126 | } |
| 127 | rowsErr := rows.Err() |
| 128 | _ = rows.Close() |
| 129 | if rowsErr != nil { |
| 130 | return out, rowsErr |
| 131 | } |
| 132 | if len(out.Items) > req.Limit || rawCount < scanLimit || rawCount == 0 { |
| 133 | break |
| 134 | } |
| 135 | scanCursor = &sessionPageCursor{Activity: lastScanned.LastActivityAt, Path: lastScanned.Path} |
| 136 | } |
| 137 | if len(out.Items) > req.Limit { |
| 138 | out.Items = out.Items[:req.Limit] |
| 139 | last := out.Items[len(out.Items)-1] |
| 140 | out.NextCursor = encodeSessionCursor(sessionPageCursor{Revision: out.Revision, Activity: last.LastActivityAt, Path: last.Path}) |
| 141 | } |
| 142 | return out, nil |
| 143 | } |
| 144 | |
| 145 | func appendSessionTimeFilter(where *[]string, args *[]any, filter string, now time.Time) { |
| 146 | value := strings.ToLower(strings.TrimSpace(filter)) |
| 147 | startToday := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()) |
| 148 | switch value { |
| 149 | case "", "all": |
| 150 | case "today": |
| 151 | *where = append(*where, `last_activity_at>=?`) |
| 152 | *args = append(*args, startToday.UnixMilli()) |
| 153 | case "yesterday": |
| 154 | *where = append(*where, `last_activity_at>=?`, `last_activity_at<?`) |
| 155 | *args = append(*args, startToday.AddDate(0, 0, -1).UnixMilli(), startToday.UnixMilli()) |
| 156 | case "older": |
| 157 | *where = append(*where, `last_activity_at<?`) |
| 158 | *args = append(*args, startToday.AddDate(0, 0, -1).UnixMilli()) |
| 159 | default: |
| 160 | if cutoff := timeFilterCutoff(value, now); cutoff > 0 { |
| 161 | *where = append(*where, `last_activity_at>=?`) |
| 162 | *args = append(*args, cutoff) |
| 163 | } |
| 164 | } |
| 165 | } |
| 166 | |
| 167 | func (c *Catalog) GetSession(ctx context.Context, path string) (SessionRecord, bool, error) { |
| 168 | path = cleanCatalogAccessPath(path) |
| 169 | if path == "" { |
| 170 | return SessionRecord{}, false, nil |
| 171 | } |
| 172 | if c.pathRemoved(path) { |
| 173 | return SessionRecord{}, false, nil |
| 174 | } |
| 175 | record, err := scanSession(c.db.QueryRowContext(ctx, `SELECT `+sessionSelectColumns+` FROM catalog_sessions WHERE path_key=?`, c.pathKey(path))) |
| 176 | if errors.Is(err, sql.ErrNoRows) { |
| 177 | return SessionRecord{}, false, nil |
| 178 | } |
| 179 | return record, err == nil, err |
| 180 | } |
| 181 | |
| 182 | func encodeSessionCursor(cursor sessionPageCursor) string { |
| 183 | b, _ := json.Marshal(cursor) |
| 184 | return base64.RawURLEncoding.EncodeToString(b) |
| 185 | } |
| 186 | |
| 187 | // CursorAfter returns an exclusive pagination cursor after the given session. |
| 188 | func CursorAfter(revision uint64, lastActivityAt int64, path string) string { |
| 189 | return encodeSessionCursor(sessionPageCursor{Revision: revision, Activity: lastActivityAt, Path: path}) |
| 190 | } |
| 191 | |
| 192 | func decodeSessionCursor(encoded string) (*sessionPageCursor, error) { |
| 193 | if strings.TrimSpace(encoded) == "" { |
| 194 | return nil, nil |
| 195 | } |
| 196 | b, err := base64.RawURLEncoding.DecodeString(encoded) |
| 197 | if err != nil { |
| 198 | return nil, fmt.Errorf("invalid session catalog cursor: %w", err) |
| 199 | } |
| 200 | var cursor sessionPageCursor |
| 201 | if err := json.Unmarshal(b, &cursor); err != nil || cursor.Path == "" { |
| 202 | return nil, errors.New("invalid session catalog cursor") |
| 203 | } |
| 204 | return &cursor, nil |
| 205 | } |
| 206 |