| 1 | package session |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "database/sql" |
| 6 | "encoding/base64" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "path/filepath" |
| 11 | "slices" |
| 12 | |
| 13 | "reasonix/internal/projectiondb" |
| 14 | "reasonix/internal/sessioncontent" |
| 15 | ) |
| 16 | |
| 17 | // History window reading (capability history-window-v1) pages a bounded |
| 18 | // window around an anchor: the newest page, a target message, a target turn, |
| 19 | // or an opaque continuation cursor — in either direction. It never walks |
| 20 | // pages from the newest position to reach a target: anchors resolve through |
| 21 | // the locator index. Protocol 7 HistoryPage stays unchanged for old clients. |
| 22 | const ( |
| 23 | HistoryWindowDefaultLimit = 32 |
| 24 | HistoryWindowMaxLimit = 100 |
| 25 | |
| 26 | historyWindowDirOlder = "older" |
| 27 | historyWindowDirNewer = "newer" |
| 28 | |
| 29 | // HistoryWindowUnsupported answers a peer that never negotiated |
| 30 | // history-window-v1. It is a status, not an error: the older service still |
| 31 | // serves bounded protocol-7 pages, so only the newer-direction and |
| 32 | // anchor-jump affordances are withheld until it is upgraded. |
| 33 | HistoryWindowUnsupported = "unsupported" |
| 34 | ) |
| 35 | |
| 36 | // HistoryWindowRequest locates and pages one bounded window in a single |
| 37 | // round trip. Anchor selects the entry point; Direction picks which side of |
| 38 | // the anchor the page covers ("older" ends at the anchor, "newer" starts at |
| 39 | // it). Cursor anchors ignore MessageID/Turn. |
| 40 | type HistoryWindowRequest struct { |
| 41 | Generation string `json:"generation,omitempty"` |
| 42 | SnapshotSequence *uint64 `json:"snapshotSequence,omitempty"` |
| 43 | Anchor string `json:"anchor"` // newest | message | turn | cursor |
| 44 | MessageID string `json:"messageId,omitempty"` // anchor=message |
| 45 | Turn int `json:"turn,omitempty"` // anchor=turn |
| 46 | Cursor string `json:"cursor,omitempty"` // anchor=cursor |
| 47 | Direction string `json:"direction,omitempty"` // older (default) | newer |
| 48 | Limit int `json:"limit,omitempty"` |
| 49 | } |
| 50 | |
| 51 | // HistoryWindowPage is one bounded window of a fixed durable snapshot plus |
| 52 | // the cursors to keep reading in both directions. |
| 53 | type HistoryWindowPage struct { |
| 54 | Messages []PersistentMessage `json:"messages"` |
| 55 | Status string `json:"status"` // preparing|ready|failed|stale_cursor|not_found|unsupported |
| 56 | SnapshotSequence uint64 `json:"snapshotSequence"` |
| 57 | CoverageSequence uint64 `json:"coverageSequence"` |
| 58 | Generation string `json:"generation,omitempty"` |
| 59 | TotalTurns int `json:"totalTurns"` |
| 60 | HasOlder bool `json:"hasOlder"` |
| 61 | HasNewer bool `json:"hasNewer"` |
| 62 | OlderCursor string `json:"olderCursor,omitempty"` |
| 63 | NewerCursor string `json:"newerCursor,omitempty"` |
| 64 | AnchorMessageID string `json:"anchorMessageId,omitempty"` |
| 65 | AnchorTurn int `json:"anchorTurn,omitempty"` |
| 66 | } |
| 67 | |
| 68 | // historyWindowCursor is the opaque cursor for window paging. Boundary is |
| 69 | // exclusive for older paging (positions < boundary) and inclusive for newer |
| 70 | // paging (positions >= boundary). |
| 71 | type historyWindowCursor struct { |
| 72 | SessionID string `json:"sessionId"` |
| 73 | StorageRevision int `json:"storageRevision"` |
| 74 | SnapshotSequence uint64 `json:"snapshotSequence"` |
| 75 | Boundary int64 `json:"boundary"` |
| 76 | Direction string `json:"direction"` |
| 77 | Projection int `json:"projection"` |
| 78 | Generation string `json:"generation"` |
| 79 | } |
| 80 | |
| 81 | func encodeHistoryWindowCursor(c historyWindowCursor) (string, error) { |
| 82 | data, err := json.Marshal(c) |
| 83 | if err != nil { |
| 84 | return "", err |
| 85 | } |
| 86 | return base64.RawURLEncoding.EncodeToString(data), nil |
| 87 | } |
| 88 | |
| 89 | func decodeHistoryWindowCursor(cursor string) (historyWindowCursor, error) { |
| 90 | var c historyWindowCursor |
| 91 | data, err := base64.RawURLEncoding.DecodeString(cursor) |
| 92 | if err != nil { |
| 93 | return historyWindowCursor{}, err |
| 94 | } |
| 95 | if err := json.Unmarshal(data, &c); err != nil { |
| 96 | return historyWindowCursor{}, err |
| 97 | } |
| 98 | return c, nil |
| 99 | } |
| 100 | |
| 101 | // ReadHistoryWindow resolves the anchor against the locator index and returns one |
| 102 | // page. The snapshot stays fixed across a paging session: appends do not |
| 103 | // invalidate cursors (version intervals retain the old view), while a storage |
| 104 | // replacement or projection rebuild answers stale_cursor. The client |
| 105 | // re-anchors at most once on stale_cursor and keeps the current page after a |
| 106 | // second failure. |
| 107 | func (q *Query) ReadHistoryWindow(ctx context.Context, ref SessionRef, req HistoryWindowRequest) (HistoryWindowPage, error) { |
| 108 | if q == nil { |
| 109 | return HistoryWindowPage{}, errors.New("session: nil query") |
| 110 | } |
| 111 | if err := ref.validate(q.hostID); err != nil { |
| 112 | return HistoryWindowPage{}, err |
| 113 | } |
| 114 | filesystem, ok := q.persistence.(*FilesystemPersistence) |
| 115 | if !ok { |
| 116 | return HistoryWindowPage{}, errors.New("session: history window requires filesystem persistence") |
| 117 | } |
| 118 | if req.Limit <= 0 { |
| 119 | req.Limit = HistoryWindowDefaultLimit |
| 120 | } |
| 121 | req.Limit = min(req.Limit, HistoryWindowMaxLimit) |
| 122 | if req.Direction != historyWindowDirNewer { |
| 123 | req.Direction = historyWindowDirOlder |
| 124 | } |
| 125 | anchor := req.Anchor |
| 126 | if anchor == "" { |
| 127 | if req.Cursor != "" { |
| 128 | anchor = "cursor" |
| 129 | } else { |
| 130 | anchor = "newest" |
| 131 | } |
| 132 | } |
| 133 | switch anchor { |
| 134 | case "newest", "message", "turn", "cursor": |
| 135 | default: |
| 136 | return HistoryWindowPage{}, fmt.Errorf("session: unknown history window anchor %q", req.Anchor) |
| 137 | } |
| 138 | |
| 139 | path := historyIndexPath(filesystem.Root, ref.SessionID) |
| 140 | ready, err := q.historyLocatorReady(ctx, filesystem, ref.SessionID, path) |
| 141 | if err != nil { |
| 142 | return HistoryWindowPage{}, err |
| 143 | } |
| 144 | if !ready { |
| 145 | return HistoryWindowPage{Messages: []PersistentMessage{}, Status: "preparing"}, nil |
| 146 | } |
| 147 | handle, err := projectiondb.Open(ctx, projectiondb.OpenOptions{Path: path, Migrations: historyMigrations, RequireDisk: true, MaxOpenConns: 1}) |
| 148 | if err != nil { |
| 149 | return HistoryWindowPage{}, err |
| 150 | } |
| 151 | defer handle.DB.Close() |
| 152 | metadata, err := readHistoryIndexMetadata(ctx, handle.DB) |
| 153 | if err != nil { |
| 154 | return HistoryWindowPage{}, err |
| 155 | } |
| 156 | |
| 157 | page := HistoryWindowPage{ |
| 158 | Messages: []PersistentMessage{}, |
| 159 | CoverageSequence: metadata.durableSequence, |
| 160 | Generation: metadata.generation, |
| 161 | AnchorMessageID: req.MessageID, |
| 162 | AnchorTurn: req.Turn, |
| 163 | } |
| 164 | if req.Generation != "" && req.Generation != metadata.generation { |
| 165 | page.Status = "stale_cursor" |
| 166 | return page, nil |
| 167 | } |
| 168 | page, boundary, direction, err := resolveWindowAnchor(ctx, handle.DB, ref, req, anchor, metadata, page) |
| 169 | if err != nil || page.Status != "" { |
| 170 | // A typed status is the whole answer: stale_cursor and not_found are |
| 171 | // results the client reasons about, not failures. |
| 172 | return page, err |
| 173 | } |
| 174 | result, err := q.readHistoryWindowPage(ctx, handle.DB, filesystem, ref, metadata, page.SnapshotSequence, boundary, direction, req.Limit) |
| 175 | if err != nil { |
| 176 | return HistoryWindowPage{}, err |
| 177 | } |
| 178 | if err := q.attachHistoryWindowTurnStats(ctx, handle.DB, &result); err != nil { |
| 179 | return HistoryWindowPage{}, err |
| 180 | } |
| 181 | |
| 182 | if err := q.attachToolObservations(ctx, handle.DB, ref, &result); err != nil { |
| 183 | return HistoryWindowPage{}, err |
| 184 | } |
| 185 | return attachWindowAnchor(result, page), nil |
| 186 | } |
| 187 | |
| 188 | func (q *Query) attachHistoryWindowTurnStats(ctx context.Context, db *sql.DB, page *HistoryWindowPage) error { |
| 189 | for i := range page.Messages { |
| 190 | message := &page.Messages[i] |
| 191 | var duration int64 |
| 192 | var turnID string |
| 193 | err := db.QueryRowContext(ctx, `SELECT MAX(0,ended_at-started_at),turn_id FROM turn_summaries WHERE final_message_id=? AND end_sequence>0 AND end_sequence<=? AND start_sequence<=? LIMIT 1`, message.MessageID, page.SnapshotSequence, message.EventSequence).Scan(&duration, &turnID) |
| 194 | if err == nil { |
| 195 | message.TurnFinal, message.TurnDurationMs = true, duration |
| 196 | var samples, tools int |
| 197 | if err := db.QueryRowContext(ctx, `SELECT COUNT(CASE WHEN kind='assistant/attempt' THEN 1 END),COUNT(CASE WHEN kind='tool/call' THEN 1 END) FROM turn_counts WHERE turn_id=? AND sequence<=?`, turnID, page.SnapshotSequence).Scan(&samples, &tools); err != nil { |
| 198 | return err |
| 199 | } |
| 200 | if samples > 0 || tools > 0 { |
| 201 | message.SamplingCount, message.ToolCount = &samples, &tools |
| 202 | } |
| 203 | } else if !errors.Is(err, sql.ErrNoRows) { |
| 204 | return err |
| 205 | } |
| 206 | } |
| 207 | return nil |
| 208 | } |
| 209 | |
| 210 | // resolveWindowAnchor turns a request's anchor into the boundary and direction |
| 211 | // the page read needs. A newest anchor reads the latest durable cut; the others |
| 212 | // pin a snapshot and answer through page.Status (with a nil error) when the |
| 213 | // anchor is stale or missing rather than failing the call. |
| 214 | func resolveWindowAnchor( |
| 215 | ctx context.Context, |
| 216 | db *sql.DB, |
| 217 | ref SessionRef, |
| 218 | req HistoryWindowRequest, |
| 219 | anchor string, |
| 220 | metadata historyIndexMetadata, |
| 221 | page HistoryWindowPage, |
| 222 | ) (HistoryWindowPage, int64, string, error) { |
| 223 | // Newest pages have no boundary above them; the read walks down from the end. |
| 224 | boundary := int64(^uint64(0) >> 1) |
| 225 | direction := req.Direction |
| 226 | cut := metadata.durableSequence |
| 227 | if req.SnapshotSequence != nil { |
| 228 | if *req.SnapshotSequence > cut { |
| 229 | page.Status = "stale_cursor" |
| 230 | return page, boundary, direction, nil |
| 231 | } |
| 232 | cut = *req.SnapshotSequence |
| 233 | } |
| 234 | if anchor == "newest" { |
| 235 | page.SnapshotSequence = cut |
| 236 | return page, boundary, direction, nil |
| 237 | } |
| 238 | if anchor == "cursor" { |
| 239 | parsed, err := decodeHistoryWindowCursor(req.Cursor) |
| 240 | if err != nil { |
| 241 | // An unreadable cursor means the same thing to a client as one bound |
| 242 | // to a replaced snapshot: start over. A typed status keeps that with |
| 243 | // the caller instead of surfacing an error that reads as transport. |
| 244 | page.Status = "stale_cursor" |
| 245 | return page, boundary, direction, nil |
| 246 | } |
| 247 | if parsed.SessionID != ref.SessionID || parsed.StorageRevision != StorageRevision || |
| 248 | parsed.Projection != historyIndexVersion || parsed.Generation != metadata.generation || |
| 249 | (parsed.Direction != historyWindowDirOlder && parsed.Direction != historyWindowDirNewer) || |
| 250 | parsed.Boundary <= 0 { |
| 251 | page.Status = "stale_cursor" |
| 252 | return page, boundary, direction, nil |
| 253 | } |
| 254 | if parsed.SnapshotSequence > metadata.durableSequence { |
| 255 | page.Status = "stale_cursor" |
| 256 | page.SnapshotSequence = metadata.durableSequence |
| 257 | return page, boundary, direction, nil |
| 258 | } |
| 259 | // The cursor pins the snapshot it was issued under; that pin is what |
| 260 | // keeps appends from invalidating a paging session. |
| 261 | page.SnapshotSequence = parsed.SnapshotSequence |
| 262 | return page, parsed.Boundary, parsed.Direction, nil |
| 263 | } |
| 264 | page.SnapshotSequence = cut |
| 265 | var anchorPos int64 |
| 266 | var err error |
| 267 | if anchor == "message" { |
| 268 | anchorPos, err = resolveMessagePosition(ctx, db, req.MessageID, page.SnapshotSequence) |
| 269 | } else { |
| 270 | anchorPos, err = resolveTurnPosition(ctx, db, req.Turn, page.SnapshotSequence) |
| 271 | } |
| 272 | if errors.Is(err, sql.ErrNoRows) { |
| 273 | page.Status = "not_found" |
| 274 | return page, boundary, direction, nil |
| 275 | } |
| 276 | if err != nil { |
| 277 | return HistoryWindowPage{}, boundary, direction, err |
| 278 | } |
| 279 | // Paging older from an anchor puts the anchor itself at the page's newest, |
| 280 | // so the reader sees the message or turn they asked for. |
| 281 | if direction == historyWindowDirOlder { |
| 282 | anchorPos++ |
| 283 | } |
| 284 | return page, anchorPos, direction, nil |
| 285 | } |
| 286 | |
| 287 | // attachWindowAnchor carries the resolved anchor identity onto the page the |
| 288 | // builder produced. |
| 289 | func attachWindowAnchor(result, resolved HistoryWindowPage) HistoryWindowPage { |
| 290 | // The page builder starts from a fresh value, so the anchor identity has to |
| 291 | // travel with the result: clients place the reading anchor and the visible |
| 292 | // turn range from it, and losing it makes an anchored page unanchored. |
| 293 | result.AnchorMessageID = resolved.AnchorMessageID |
| 294 | result.AnchorTurn = resolved.AnchorTurn |
| 295 | return result |
| 296 | } |
| 297 | |
| 298 | func resolveMessagePosition(ctx context.Context, db *sql.DB, messageID string, snapshot uint64) (int64, error) { |
| 299 | var position int64 |
| 300 | err := db.QueryRowContext(ctx, |
| 301 | `SELECT position FROM messages WHERE message_id=? AND event_sequence<=? AND (valid_to=0 OR valid_to>?) ORDER BY version DESC LIMIT 1`, |
| 302 | messageID, snapshot, snapshot).Scan(&position) |
| 303 | return position, err |
| 304 | } |
| 305 | |
| 306 | func resolveTurnPosition(ctx context.Context, db *sql.DB, turn int, snapshot uint64) (int64, error) { |
| 307 | var position int64 |
| 308 | err := db.QueryRowContext(ctx, |
| 309 | `SELECT MIN(position) FROM messages WHERE visible_turn=? AND event_sequence<=? AND (valid_to=0 OR valid_to>?)`, |
| 310 | turn, snapshot, snapshot).Scan(&position) |
| 311 | if err != nil { |
| 312 | return 0, err |
| 313 | } |
| 314 | if position == 0 { |
| 315 | // Zero positions never occur in the locator: MIN yielded NULL over no |
| 316 | // rows only when the turn has no current messages — confirm to |
| 317 | // distinguish not_found from a scan error. |
| 318 | var exists bool |
| 319 | if err := db.QueryRowContext(ctx, |
| 320 | `SELECT EXISTS(SELECT 1 FROM messages WHERE visible_turn=? AND event_sequence<=? AND (valid_to=0 OR valid_to>?))`, |
| 321 | turn, snapshot, snapshot).Scan(&exists); err != nil { |
| 322 | return 0, err |
| 323 | } |
| 324 | if !exists { |
| 325 | return 0, sql.ErrNoRows |
| 326 | } |
| 327 | } |
| 328 | return position, nil |
| 329 | } |
| 330 | |
| 331 | // readHistoryWindowPage pages [boundary,∞) or (−∞,boundary) of a fixed |
| 332 | // snapshot in limit-sized steps, inlining small bodies and authorizing |
| 333 | // content-range credentials for referenced ones — the same body budget as |
| 334 | // protocol 7 pages. |
| 335 | func (q *Query) readHistoryWindowPage(ctx context.Context, db *sql.DB, filesystem *FilesystemPersistence, ref SessionRef, metadata historyIndexMetadata, snapshot uint64, boundary int64, direction string, limit int) (HistoryWindowPage, error) { |
| 336 | page := HistoryWindowPage{ |
| 337 | Messages: []PersistentMessage{}, |
| 338 | Status: "ready", |
| 339 | SnapshotSequence: snapshot, |
| 340 | CoverageSequence: metadata.durableSequence, |
| 341 | Generation: metadata.generation, |
| 342 | } |
| 343 | if err := db.QueryRowContext(ctx, `SELECT COALESCE(MAX(visible_turn),0) FROM messages WHERE event_sequence<=? AND (valid_to=0 OR valid_to>?)`, snapshot, snapshot).Scan(&page.TotalTurns); err != nil { |
| 344 | return HistoryWindowPage{}, err |
| 345 | } |
| 346 | var rows *sql.Rows |
| 347 | var err error |
| 348 | if direction == historyWindowDirNewer { |
| 349 | rows, err = db.QueryContext(ctx, `SELECT message_id,position,version,role,preview,event_sequence,visible_turn,inline,content_digest,content_bytes,content_index_digest,COALESCE((SELECT submission_id FROM submissions WHERE submissions.message_id=messages.message_id AND submissions.sequence<=messages.event_sequence AND submissions.session_id=(SELECT value FROM metadata WHERE key='session_id') LIMIT 1),'') FROM messages WHERE position>=? AND event_sequence<=? AND (valid_to=0 OR valid_to>?) ORDER BY position ASC LIMIT ?`, boundary, snapshot, snapshot, limit+1) |
| 350 | } else { |
| 351 | rows, err = db.QueryContext(ctx, `SELECT message_id,position,version,role,preview,event_sequence,visible_turn,inline,content_digest,content_bytes,content_index_digest,COALESCE((SELECT submission_id FROM submissions WHERE submissions.message_id=messages.message_id AND submissions.sequence<=messages.event_sequence AND submissions.session_id=(SELECT value FROM metadata WHERE key='session_id') LIMIT 1),'') FROM messages WHERE position<? AND event_sequence<=? AND (valid_to=0 OR valid_to>?) ORDER BY position DESC LIMIT ?`, boundary, snapshot, snapshot, limit+1) |
| 352 | } |
| 353 | if err != nil { |
| 354 | return HistoryWindowPage{}, err |
| 355 | } |
| 356 | defer rows.Close() |
| 357 | encodedBytes := 0 |
| 358 | storageGeneration := q.storageGeneration(ref.SessionID) |
| 359 | scanned := 0 |
| 360 | for rows.Next() { |
| 361 | var message PersistentMessage |
| 362 | var inline []byte |
| 363 | var digest, indexDigest string |
| 364 | var contentBytes int64 |
| 365 | if err := rows.Scan(&message.MessageID, &message.Position, &message.Version, &message.Role, &message.Preview, &message.EventSequence, &message.VisibleTurn, &inline, &digest, &contentBytes, &indexDigest, &message.SubmissionID); err != nil { |
| 366 | return HistoryWindowPage{}, err |
| 367 | } |
| 368 | scanned++ |
| 369 | if len(page.Messages) == limit { |
| 370 | break |
| 371 | } |
| 372 | message.Inline = append(json.RawMessage(nil), inline...) |
| 373 | if digest != "" { |
| 374 | contentRef := sessioncontent.Ref{Digest: digest, Bytes: contentBytes, IndexDigest: indexDigest, IntegrityBlock: sessioncontent.IntegrityBlockBytes, MediaType: "application/json"} |
| 375 | if contentBytes <= recentInlineBytes && encodedBytes+int(contentBytes) <= HistoryPageMaxBytes { |
| 376 | body, readErr := contentStoreForSessionDir(filepath.Join(filesystem.Root, ref.SessionID)).ReadRange(ctx, contentRef, 0, contentBytes) |
| 377 | if readErr != nil { |
| 378 | return HistoryWindowPage{}, readErr |
| 379 | } |
| 380 | message.Inline = json.RawMessage(body) |
| 381 | } else { |
| 382 | message.ContentRef = &contentRef |
| 383 | q.authorizeContentForGeneration(ref.SessionID, storageGeneration, digest, contentBytes, indexDigest) |
| 384 | } |
| 385 | } |
| 386 | encoded, _ := json.Marshal(message) |
| 387 | if len(page.Messages) > 0 && encodedBytes+len(encoded) > HistoryPageMaxBytes { |
| 388 | break |
| 389 | } |
| 390 | encodedBytes += len(encoded) |
| 391 | page.Messages = append(page.Messages, message) |
| 392 | } |
| 393 | if err := rows.Err(); err != nil { |
| 394 | return HistoryWindowPage{}, err |
| 395 | } |
| 396 | hasMoreBeyond := scanned > len(page.Messages) |
| 397 | if direction == historyWindowDirNewer { |
| 398 | page.HasNewer = hasMoreBeyond |
| 399 | page.HasOlder = boundary > 1 |
| 400 | if page.HasNewer && len(page.Messages) > 0 { |
| 401 | last := page.Messages[len(page.Messages)-1] |
| 402 | page.NewerCursor, err = encodeHistoryWindowCursor(historyWindowCursor{SessionID: ref.SessionID, StorageRevision: StorageRevision, SnapshotSequence: snapshot, Boundary: last.Position + 1, Direction: historyWindowDirNewer, Projection: historyIndexVersion, Generation: metadata.generation}) |
| 403 | if err != nil { |
| 404 | return HistoryWindowPage{}, err |
| 405 | } |
| 406 | } |
| 407 | if page.HasOlder { |
| 408 | page.OlderCursor, err = encodeHistoryWindowCursor(historyWindowCursor{SessionID: ref.SessionID, StorageRevision: StorageRevision, SnapshotSequence: snapshot, Boundary: boundary, Direction: historyWindowDirOlder, Projection: historyIndexVersion, Generation: metadata.generation}) |
| 409 | if err != nil { |
| 410 | return HistoryWindowPage{}, err |
| 411 | } |
| 412 | } |
| 413 | // Newer pages were collected oldest-first: already display order. |
| 414 | return page, nil |
| 415 | } |
| 416 | // Older paging: the collected messages are newest-first and reversed for |
| 417 | // display; continuation cursors cover both directions. |
| 418 | page.HasOlder = hasMoreBeyond |
| 419 | if page.HasOlder && len(page.Messages) > 0 { |
| 420 | oldest := page.Messages[len(page.Messages)-1] |
| 421 | page.OlderCursor, err = encodeHistoryWindowCursor(historyWindowCursor{SessionID: ref.SessionID, StorageRevision: StorageRevision, SnapshotSequence: snapshot, Boundary: oldest.Position, Direction: historyWindowDirOlder, Projection: historyIndexVersion, Generation: metadata.generation}) |
| 422 | if err != nil { |
| 423 | return HistoryWindowPage{}, err |
| 424 | } |
| 425 | } |
| 426 | // The locator opens at most one connection: release this result set |
| 427 | // before issuing the existence probe. |
| 428 | rows.Close() |
| 429 | var newer bool |
| 430 | if err := db.QueryRowContext(ctx, `SELECT EXISTS(SELECT 1 FROM messages WHERE position>=? AND event_sequence<=? AND (valid_to=0 OR valid_to>?))`, boundary, snapshot, snapshot).Scan(&newer); err != nil { |
| 431 | return HistoryWindowPage{}, err |
| 432 | } |
| 433 | page.HasNewer = newer |
| 434 | if page.HasNewer { |
| 435 | page.NewerCursor, err = encodeHistoryWindowCursor(historyWindowCursor{SessionID: ref.SessionID, StorageRevision: StorageRevision, SnapshotSequence: snapshot, Boundary: boundary, Direction: historyWindowDirNewer, Projection: historyIndexVersion, Generation: metadata.generation}) |
| 436 | if err != nil { |
| 437 | return HistoryWindowPage{}, err |
| 438 | } |
| 439 | } |
| 440 | slices.Reverse(page.Messages) |
| 441 | return page, nil |
| 442 | } |
| 443 |