diff --git a/internal/server/handlers_items.go b/internal/server/handlers_items.go index e43d917b..6d0b1105 100644 --- a/internal/server/handlers_items.go +++ b/internal/server/handlers_items.go @@ -76,9 +76,14 @@ func (s *Server) handleListItems(w http.ResponseWriter, r *http.Request) { // itemsIndexResponse wraps the skinny-projection items list with bookkeeping // the local-first read model (PLAN-1343) needs: // - total: the row count so the client can size its in-memory array. -// - cursor: a placeholder for the Phase 2 monotonic seq cursor — for now -// we return the maximum updated_at (RFC3339), or "0" when the workspace -// is empty. Clients should treat its value as opaque. +// - cursor: the workspace-scoped monotonic `seq` cursor (TASK-1353). +// Holds MAX(seq) across the requested scope as a decimal-encoded +// string so it forward-compatibly tolerates future encoding changes +// (opaque on the wire — clients MUST NOT parse it as an integer). +// When the requested scope returns zero rows, the cursor falls back +// to the workspace's current MAX(seq) so /items-changes?since=cursor +// starts from the right floor on the next poll instead of replaying +// every prior mutation. Empty workspaces return "0". type itemsIndexResponse struct { Items []models.Item `json:"items"` Total int `json:"total"` @@ -127,6 +132,32 @@ func (s *Server) handleListItemsIndex(w http.ResponseWriter, r *http.Request) { params.ItemIDs = grantedItemIDs } + // Snapshot the workspace's MAX(seq) BEFORE the list query so the + // fallback cursor cannot leapfrog a concurrent insert that + // committed between the list query and a post-list MaxItemSeq + // read. Per Codex review of TASK-1353 round 1 [P1]: + // + // List → empty. + // Concurrent INSERT lands with seq = M+1 (visible to a future + // /items-changes call). + // MaxItemSeq → M+1. + // Cursor = M+1, response = []. + // Client polls /items-changes?since=M+1 → seq > M+1 returns + // nothing. The new row is lost. + // + // Capturing M BEFORE the list eliminates that window: any insert + // after the snapshot has seq > M (workspace counter is strictly + // monotonic per TASK-1352), so /items-changes?since=M will return + // it. Rows the list DOES see may have seq > M (a concurrent insert + // the list query happened to observe); MAX(rows.seq) bumps the + // cursor for that case so the client never re-fetches what was + // already in the response. + wsMaxBefore, err := s.store.MaxItemSeq(workspaceID) + if err != nil { + writeInternalError(w, err) + return + } + result, err := s.store.ListItemsIndex(workspaceID, params) if err != nil { writeInternalError(w, err) @@ -137,13 +168,16 @@ func (s *Server) handleListItemsIndex(w http.ResponseWriter, r *http.Request) { } s.enrichItemsWithParent(workspaceID, result, visibleIDs) - // Cursor placeholder: max(updated_at) across the returned set. ListItemsIndex - // already sorts by updated_at DESC, so the first row holds it. Phase 2 will - // replace this with a monotonic `seq` cursor (see PLAN-1343). - cursor := "0" - if len(result) > 0 { - cursor = result[0].UpdatedAt.UTC().Format(time.RFC3339Nano) + // Cursor: max(pre-list workspace floor, MAX(returned-rows.seq)). + // See the long comment above MaxItemSeq for why the floor is read + // BEFORE ListItemsIndex. Empty workspace + empty result → "0". + cursorSeq := wsMaxBefore + for _, it := range result { + if it.Seq > cursorSeq { + cursorSeq = it.Seq + } } + cursor := strconv.FormatInt(cursorSeq, 10) writeJSON(w, http.StatusOK, itemsIndexResponse{ Items: result, diff --git a/internal/server/handlers_items_test.go b/internal/server/handlers_items_test.go index c773faf9..200697b1 100644 --- a/internal/server/handlers_items_test.go +++ b/internal/server/handlers_items_test.go @@ -6,6 +6,7 @@ import ( "io" "net/http" "net/http/httptest" + "strconv" "testing" "time" @@ -1255,7 +1256,8 @@ type itemsIndexBody struct { // behavior of the local-first read model bootstrap endpoint (TASK-1344): // - response shape: {items, total, cursor} // - content body excluded (skinny projection) -// - cursor placeholder reflects the newest updated_at +// - cursor reflects MAX(seq) across the result set (TASK-1353) +// - per-row seq is populated and non-zero // - core projected fields populate (ref, collection_slug, fields, …) // // Sort order is exercised separately by TestListItemsIndex_SortByUpdatedAt, @@ -1295,11 +1297,23 @@ func TestListItemsIndex_SkinnyProjectionAndShape(t *testing.T) { t.Fatalf("expected cursor to be populated for non-empty workspace, got %q", resp.Cursor) } - // Cursor must match the newest updated_at — and because the sort is - // updated_at DESC, that's the row at index 0. - want := resp.Items[0].UpdatedAt.UTC().Format(time.RFC3339Nano) - if resp.Cursor != want { - t.Fatalf("cursor mismatch: got %q, want %q", resp.Cursor, want) + // Cursor must be the decimal-encoded MAX(seq) across the returned + // rows (TASK-1353). Per-row seq is populated by the skinny scan. + cursorSeq, err := strconv.ParseInt(resp.Cursor, 10, 64) + if err != nil { + t.Fatalf("cursor not decimal-encoded int: %q (err=%v)", resp.Cursor, err) + } + var maxSeq int64 + for _, it := range resp.Items { + if it.Seq == 0 { + t.Errorf("items[%s]: seq must be non-zero", it.Ref) + } + if it.Seq > maxSeq { + maxSeq = it.Seq + } + } + if cursorSeq != maxSeq { + t.Fatalf("cursor mismatch: got %d, want MAX(seq)=%d", cursorSeq, maxSeq) } // Skinny projection: content body excluded for every row. @@ -1360,7 +1374,7 @@ func TestListItemsIndex_SortByUpdatedAt(t *testing.T) { } // TestListItemsIndex_EmptyWorkspace covers the edge case the cursor -// placeholder is documented to handle: no items → cursor "0". +// is documented to handle: no items → cursor "0". func TestListItemsIndex_EmptyWorkspace(t *testing.T) { srv := testServer(t) slug := createWSWithCollections(t, srv) @@ -1389,6 +1403,93 @@ func TestListItemsIndex_EmptyWorkspace(t *testing.T) { } } +// TestListItemsIndex_EmptyResultFallsBackToWorkspaceMax verifies the +// cursor-fallback path (TASK-1353 acceptance): when a filtered query +// returns zero rows but the workspace itself is non-empty, the cursor +// must hold the workspace's MAX(seq) so the client's next +// /items-changes?since=cursor poll starts at the right floor rather +// than replaying every prior mutation from 0. +func TestListItemsIndex_EmptyResultFallsBackToWorkspaceMax(t *testing.T) { + srv := testServer(t) + slug := createWSWithCollections(t, srv) + + // Populate the workspace so MAX(seq) is non-zero. + createItem(t, srv, slug, "tasks", map[string]interface{}{ + "title": "T1", + "fields": `{"status":"open"}`, + }) + createItem(t, srv, slug, "ideas", map[string]interface{}{ + "title": "I1", + "fields": `{"status":"new"}`, + }) + + // Establish the baseline MAX(seq) via the unfiltered call. + rrAll := doRequest(srv, "GET", "/api/v1/workspaces/"+slug+"/items-index", nil) + if rrAll.Code != http.StatusOK { + t.Fatalf("items-index all: expected 200, got %d: %s", rrAll.Code, rrAll.Body.String()) + } + var all itemsIndexBody + parseJSON(t, rrAll, &all) + if all.Cursor == "" || all.Cursor == "0" { + t.Fatalf("baseline cursor should be non-zero, got %q", all.Cursor) + } + + // Now request a collection that exists but has no items — the response + // slice is empty, but the workspace MAX(seq) fallback should still + // produce the same cursor. + rrFiltered := doRequest(srv, "GET", "/api/v1/workspaces/"+slug+"/items-index?collection=docs", nil) + if rrFiltered.Code != http.StatusOK { + t.Fatalf("items-index docs: expected 200, got %d: %s", rrFiltered.Code, rrFiltered.Body.String()) + } + var filtered itemsIndexBody + parseJSON(t, rrFiltered, &filtered) + + if filtered.Total != 0 { + t.Fatalf("expected total=0 for empty collection filter, got %d", filtered.Total) + } + if filtered.Cursor != all.Cursor { + t.Fatalf("empty-result cursor should equal workspace MAX(seq) baseline %q, got %q", all.Cursor, filtered.Cursor) + } + if _, err := strconv.ParseInt(filtered.Cursor, 10, 64); err != nil { + t.Fatalf("fallback cursor not decimal-encoded int: %q", filtered.Cursor) + } +} + +// TestListItemsIndex_CursorMonotonicAcrossMutations confirms that the +// cursor advances after every items mutation and can be re-used as the +// `since` parameter on a follow-up /items-changes call (cursor contract +// for PLAN-1343 Phase 2). +func TestListItemsIndex_CursorMonotonicAcrossMutations(t *testing.T) { + srv := testServer(t) + slug := createWSWithCollections(t, srv) + + createItem(t, srv, slug, "tasks", map[string]interface{}{ + "title": "C1", + "fields": `{"status":"open"}`, + }) + + rr1 := doRequest(srv, "GET", "/api/v1/workspaces/"+slug+"/items-index", nil) + var resp1 itemsIndexBody + parseJSON(t, rr1, &resp1) + c1, _ := strconv.ParseInt(resp1.Cursor, 10, 64) + if c1 == 0 { + t.Fatalf("expected non-zero cursor after first create, got %q", resp1.Cursor) + } + + createItem(t, srv, slug, "ideas", map[string]interface{}{ + "title": "C2", + "fields": `{"status":"new"}`, + }) + + rr2 := doRequest(srv, "GET", "/api/v1/workspaces/"+slug+"/items-index", nil) + var resp2 itemsIndexBody + parseJSON(t, rr2, &resp2) + c2, _ := strconv.ParseInt(resp2.Cursor, 10, 64) + if c2 <= c1 { + t.Fatalf("cursor should advance after second create: c1=%d, c2=%d", c1, c2) + } +} + // TestListItemsIndex_CollectionFilter covers ?collection=. func TestListItemsIndex_CollectionFilter(t *testing.T) { srv := testServer(t) diff --git a/internal/store/items.go b/internal/store/items.go index 8c529fe2..4bafe40e 100644 --- a/internal/store/items.go +++ b/internal/store/items.go @@ -816,6 +816,27 @@ func (s *Store) ListItemsIndex(workspaceID string, params ItemIndexParams) ([]mo return scanItemsIndex(rows) } +// MaxItemSeq returns the largest items.seq across the workspace, or 0 +// if the workspace has no items. This is the cursor floor for the +// local-first read model (PLAN-1343 / TASK-1353): /items-index hands +// it back when its filtered result set is empty so the client can +// poll /items-changes?since= against the workspace's true +// current position instead of restarting from 0. +// +// Soft-deleted items DO contribute to MAX(seq) — the seq column +// bumps on tombstone writes (DeleteItem) so a client's cursor must +// move past those events for the next /items-changes scan to skip +// them. Filtering by `deleted_at IS NULL` here would silently regress +// the cursor whenever the most recent mutation was a delete. +func (s *Store) MaxItemSeq(workspaceID string) (int64, error) { + var seq int64 + err := s.db.QueryRow(s.q(`SELECT COALESCE(MAX(seq), 0) FROM items WHERE workspace_id = ?`), workspaceID).Scan(&seq) + if err != nil { + return 0, fmt.Errorf("max item seq: %w", err) + } + return seq, nil +} + // ItemCheckboxProgress is the per-item count of markdown checkboxes // (`- [ ]` / `- [x]`) extracted from item content. Used by the // collection page to render checklist progress badges without diff --git a/web/src/lib/api/client.ts b/web/src/lib/api/client.ts index dbe4877d..ea4466dd 100644 --- a/web/src/lib/api/client.ts +++ b/web/src/lib/api/client.ts @@ -328,7 +328,14 @@ export const api = { * Skinny-projection cross-collection listing for the local-first * read model (PLAN-1343 / TASK-1344). Returns every item in a * workspace MINUS the rich-text `content` body, plus a `total` - * count and a forward-looking `cursor` placeholder. + * count and a real workspace-scoped `seq` cursor (TASK-1353). + * + * The response `cursor` is the decimal-encoded `MAX(seq)` across + * the requested scope (or the workspace's true `MAX(seq)` when + * the filtered set is empty, so /items-changes?since=cursor + * starts at the right floor). Each row carries its own `seq` + * field so the client can reason about ordering without parsing + * the cursor. * * Optional filters mirror the server: `collection` narrows to one * collection slug, `include_archived` flips the soft-delete gate. diff --git a/web/src/lib/types/index.ts b/web/src/lib/types/index.ts index 72290512..6b3acf76 100644 --- a/web/src/lib/types/index.ts +++ b/web/src/lib/types/index.ts @@ -429,6 +429,12 @@ export interface Item { collection_icon?: string; collection_prefix?: string; item_number?: number; + // `seq` is the workspace-scoped monotonic mutation cursor (PLAN-1343 / + // DOC-1342 decision #1). Stamped server-side on every create / update / + // soft-delete / restore. Clients track the max `seq` they have seen + // and request `?since=` deltas to resume. Optional because old + // snapshots may predate the column; new responses always populate it. + seq?: number; parent_link_id?: string; parent_ref?: string; parent_title?: string; @@ -455,10 +461,14 @@ export type ItemIndexRow = Omit; export interface ItemIndexResponse { items: ItemIndexRow[]; total: number; - // `cursor` is a placeholder until Phase 2 lands the monotonic `seq` - // column — today the server returns the maximum `updated_at` across - // the result set (RFC3339Nano), or `"0"` when the workspace is empty. - // Clients should treat the value as opaque. + // `cursor` is the workspace-scoped monotonic `seq` cursor (TASK-1353). + // Holds MAX(seq) across the requested scope as a decimal-encoded + // string. When the result set is empty but the workspace has items, + // it falls back to the workspace's current MAX(seq) so the next + // /items-changes?since= poll starts at the right floor. + // Empty workspaces return `"0"`. Treat the value as opaque — the + // encoding may change in future, and clients should not parse it + // as an integer beyond passing it back as `since`. cursor: string; }