diff --git a/internal/models/item.go b/internal/models/item.go index fcadf797..7bafdb5a 100644 --- a/internal/models/item.go +++ b/internal/models/item.go @@ -43,6 +43,16 @@ type Item struct { // Auto-assigned sequential number within collection ItemNumber *int `json:"item_number,omitempty"` + // Seq is a workspace-scoped monotonically-increasing sequence number + // stamped on every mutation (create / update / soft-delete / + // restore). It is the cursor mechanic for the local-first read + // model's delta sync (PLAN-1343, DOC-1342 design decision #1). + // Clients track the max seq they have seen and request + // `?since=` deltas to resume. Robust against clock-skew / + // same-millisecond-write / NTP-step correctness holes that an + // `updated_at` watermark would carry. + Seq int64 `json:"seq,omitempty"` + // Populated by joins (not stored) AssignedUserName string `json:"assigned_user_name,omitempty"` AssignedUserEmail string `json:"assigned_user_email,omitempty"` diff --git a/internal/store/agent_roles.go b/internal/store/agent_roles.go index 17c6f28e..88480d25 100644 --- a/internal/store/agent_roles.go +++ b/internal/store/agent_roles.go @@ -437,6 +437,10 @@ type RoleSortUpdate struct { } // UpdateRoleSortOrder batch-updates role_sort_order for a list of items. +// Each updated row also gets a fresh workspace-scoped seq so delta-sync +// clients see the reorder (PLAN-1343 / TASK-1352). Without this, a +// client polling /items-changes?since=cursor would miss role-board +// reorders until a full refresh. func (s *Store) UpdateRoleSortOrder(workspaceID string, updates []RoleSortUpdate) error { tx, err := s.db.Begin() if err != nil { @@ -444,14 +448,25 @@ func (s *Store) UpdateRoleSortOrder(workspaceID string, updates []RoleSortUpdate } defer tx.Rollback() - stmt, err := tx.Prepare(s.q("UPDATE items SET role_sort_order = ? WHERE id = ? AND workspace_id = ?")) + // Serialize concurrent seq assignments per workspace on Postgres. + // Held until COMMIT / ROLLBACK. + if err := s.acquireWorkspaceSeqLock(tx, workspaceID); err != nil { + return err + } + + // Sequential UPDATEs each read MAX(seq)+1 inside the same + // transaction so every row in the batch gets a strictly greater + // seq than the row before it. Statements within a single + // transaction see each other's effects on both SQLite and + // Postgres (READ COMMITTED). + stmt, err := tx.Prepare(s.q("UPDATE items SET role_sort_order = ?, seq = " + nextWorkspaceSeqSubquery + " WHERE id = ? AND workspace_id = ?")) if err != nil { return fmt.Errorf("prepare role sort update: %w", err) } defer stmt.Close() for _, u := range updates { - if _, err := stmt.Exec(u.RoleSortOrder, u.ItemID, workspaceID); err != nil { + if _, err := stmt.Exec(u.RoleSortOrder, workspaceID, u.ItemID, workspaceID); err != nil { return fmt.Errorf("update role sort for %s: %w", u.ItemID, err) } } diff --git a/internal/store/collections.go b/internal/store/collections.go index 4efad519..8e3b50e0 100644 --- a/internal/store/collections.go +++ b/internal/store/collections.go @@ -296,11 +296,39 @@ func (s *Store) DeleteCollection(id string) error { // MigrateItemFieldValues bulk-updates items in a collection when select // options are renamed. Each entry in renames maps old_value → new_value // for the given field key. +// +// Each migration step bumps the workspace-scoped seq so delta-sync +// clients see the field rewrite (PLAN-1343 / TASK-1352). All rows +// affected by a single rename step share the same new seq value +// (MAX+1 at statement start) — that's fine for the cursor contract: +// a client at cursor < MAX sees them all in one batch, a client at +// cursor >= MAX sees none, no overlap or gap. func (s *Store) MigrateItemFieldValues(collectionID string, migrations []models.FieldMigration) (int64, error) { if len(migrations) == 0 { return 0, nil } + // Look up the workspace for advisory locking + scoping the seq + // subquery. If the collection has vanished out from under the + // caller we can short-circuit. + var workspaceID string + if err := s.db.QueryRow(s.q(`SELECT workspace_id FROM collections WHERE id = ?`), collectionID).Scan(&workspaceID); err != nil { + if err == sql.ErrNoRows { + return 0, nil + } + return 0, fmt.Errorf("lookup workspace for migrate: %w", err) + } + + tx, err := s.db.Begin() + if err != nil { + return 0, err + } + defer tx.Rollback() + + if err := s.acquireWorkspaceSeqLock(tx, workspaceID); err != nil { + return 0, err + } + ts := now() var totalAffected int64 @@ -311,14 +339,15 @@ func (s *Store) MigrateItemFieldValues(collectionID string, migrations []models. } jsonSet := s.dialect.JSONSet("fields", m.Field) jsonExtract := s.dialect.JSONExtractText("fields", m.Field) - result, err := s.db.Exec(s.q(fmt.Sprintf(` + result, err := tx.Exec(s.q(fmt.Sprintf(` UPDATE items SET fields = %s, - updated_at = ? + updated_at = ?, + seq = (SELECT COALESCE(MAX(seq), 0) + 1 FROM items WHERE workspace_id = ?) WHERE collection_id = ? AND %s = ? AND deleted_at IS NULL - `, jsonSet, jsonExtract)), newVal, ts, collectionID, oldVal) + `, jsonSet, jsonExtract)), newVal, ts, workspaceID, collectionID, oldVal) if err != nil { return totalAffected, fmt.Errorf("migrate field %s (%s → %s): %w", m.Field, oldVal, newVal, err) } @@ -327,6 +356,9 @@ func (s *Store) MigrateItemFieldValues(collectionID string, migrations []models. } } + if err := tx.Commit(); err != nil { + return totalAffected, err + } return totalAffected, nil } diff --git a/internal/store/export.go b/internal/store/export.go index 56fbabf4..b6c1fd59 100644 --- a/internal/store/export.go +++ b/internal/store/export.go @@ -226,12 +226,18 @@ func (s *Store) ImportWorkspace(data *models.WorkspaceExport, newName string, ow } nextItemNumber++ + // Stamp `seq` so workspace import populates the delta-sync cursor + // column (PLAN-1343 / TASK-1352). Each INSERT reads MAX(seq)+1 + // within this transaction, so imported rows get sequential + // per-workspace seqs — clients post-import see them on the next + // /items-index fetch, and any subsequent mutation keeps bumping + // from a sensible floor instead of a flat MAX(seq)=0. _, err := tx.Exec(s.q(` - INSERT INTO items (id, workspace_id, collection_id, title, slug, content, fields, tags, pinned, sort_order, parent_id, created_by, last_modified_by, source, item_number, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULLIF(?, ''), ?, ?, ?, ?, ?, ?)`), + INSERT INTO items (id, workspace_id, collection_id, title, slug, content, fields, tags, pinned, sort_order, parent_id, created_by, last_modified_by, source, item_number, created_at, updated_at, seq) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, NULLIF(?, ''), ?, ?, ?, ?, ?, ?, `+nextWorkspaceSeqSubquery+`)`), newItemID, ws.ID, newCollID, it.Title, it.Slug, it.Content, it.Fields, it.Tags, s.dialect.BoolToInt(it.Pinned), it.SortOrder, parentID, it.CreatedBy, it.LastModifiedBy, it.Source, nextItemNumber, - it.CreatedAt, it.UpdatedAt) + it.CreatedAt, it.UpdatedAt, ws.ID) if err != nil { return nil, fmt.Errorf("import item %s: %w", it.Title, err) } diff --git a/internal/store/item_stars.go b/internal/store/item_stars.go index 911f95b3..a10c709d 100644 --- a/internal/store/item_stars.go +++ b/internal/store/item_stars.go @@ -98,7 +98,7 @@ func (s *Store) ListStarredItems(userID, workspaceID string, includeTerminal boo SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') diff --git a/internal/store/items.go b/internal/store/items.go index 0e29bf37..8c529fe2 100644 --- a/internal/store/items.go +++ b/internal/store/items.go @@ -61,6 +61,36 @@ func (s *Store) validateAssignmentScope(workspaceID string, assignedUserID, agen // concurrent insert claims the same workspace-global item_number. const maxItemNumberRetries = 10 +// nextWorkspaceSeqSubquery is the SQL fragment used to atomically compute +// the next workspace-scoped `seq` value inside an INSERT / UPDATE. +// Every items mutation (create / update / soft-delete / restore) stamps +// the new row's seq with `MAX(seq) + 1 WHERE workspace_id = ?`, which is +// the cursor mechanic for the local-first read model's delta sync +// (PLAN-1343 / DOC-1342 decision #1). +// +// Callers must append exactly one `workspaceID` arg for this fragment. +// SQLite is single-writer so the read-modify-write is naturally +// serialized; Postgres callers must additionally hold the workspace +// advisory lock acquired via acquireWorkspaceSeqLock so concurrent +// writes can't both read the same MAX(seq) and produce duplicates. +const nextWorkspaceSeqSubquery = "(SELECT COALESCE(MAX(seq), 0) + 1 FROM items WHERE workspace_id = ?)" + +// acquireWorkspaceSeqLock takes a Postgres advisory transaction lock +// keyed on the workspace ID so concurrent seq-bumping mutations +// serialize. The lock auto-releases on COMMIT / ROLLBACK. On SQLite +// the single-writer rule already serializes writes, so this is a +// no-op there. Mirrors the existing advisory-lock pattern in +// tryCreateItem (which uses the same key for item_number assignment). +func (s *Store) acquireWorkspaceSeqLock(tx *sql.Tx, workspaceID string) error { + if s.dialect.Driver() != DriverPostgres { + return nil + } + if _, err := tx.Exec("SELECT pg_advisory_xact_lock(hashtext($1))", workspaceID); err != nil { + return fmt.Errorf("acquire workspace seq lock: %w", err) + } + return nil +} + func (s *Store) CreateItem(workspaceID, collectionID string, input models.ItemCreate) (*models.Item, error) { // Validate assignment scope before writing if err := s.validateAssignmentScope(workspaceID, input.AssignedUserID, input.AgentRoleID); err != nil { @@ -154,17 +184,20 @@ func (s *Store) tryCreateItem(id, workspaceID, collectionID, slug, ts, fields, t contentFlushedAt = ts contentFlushedOpLogID = int64(0) } + // The workspace advisory lock acquired above for item_number + // assignment ALSO serializes the seq subquery below — both read + // MAX(...) per workspace and would otherwise race in Postgres. _, err = tx.Exec(s.q(` INSERT INTO items (id, workspace_id, collection_id, title, slug, content, fields, tags, pinned, sort_order, parent_id, assigned_user_id, agent_role_id, role_sort_order, created_by, last_modified_by, source, item_number, created_at, updated_at, - content_flushed_at, content_flushed_op_log_id) + content_flushed_at, content_flushed_op_log_id, seq) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, 0, ?, ?, ?, (SELECT COALESCE(MAX(item_number), 0) + 1 FROM items WHERE workspace_id = ?), - ?, ?, ?, ?) + ?, ?, ?, ?, `+nextWorkspaceSeqSubquery+`) `), id, workspaceID, collectionID, input.Title, slug, input.Content, fields, tags, s.dialect.BoolToInt(input.Pinned), input.ParentID, input.AssignedUserID, input.AgentRoleID, - createdBy, createdBy, source, workspaceID, ts, ts, contentFlushedAt, contentFlushedOpLogID) + createdBy, createdBy, source, workspaceID, ts, ts, contentFlushedAt, contentFlushedOpLogID, workspaceID) if err != nil { return err } @@ -205,7 +238,7 @@ func (s *Store) GetItem(id string) (*models.Item, error) { SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, i.deleted_at, + i.item_number, i.seq, i.created_at, i.updated_at, i.deleted_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') @@ -219,7 +252,7 @@ func (s *Store) GetItem(id string) (*models.Item, error) { &item.Content, &item.Fields, &item.Tags, &pinned, &item.SortOrder, &item.ParentID, &item.AssignedUserID, &item.AgentRoleID, &item.RoleSortOrder, &item.CreatedBy, &item.LastModifiedBy, &item.Source, - &item.ItemNumber, &createdAt, &updatedAt, &deletedAt, + &item.ItemNumber, &item.Seq, &createdAt, &updatedAt, &deletedAt, &item.CollectionSlug, &item.CollectionName, &item.CollectionIcon, &item.CollectionPrefix, &item.AssignedUserName, &item.AssignedUserEmail, &item.AgentRoleName, &item.AgentRoleSlug, &item.AgentRoleIcon, @@ -346,7 +379,7 @@ func (s *Store) ResolveItemIncludeDeleted(workspaceID, slugOrRef string) (*model SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, i.deleted_at, + i.item_number, i.seq, i.created_at, i.updated_at, i.deleted_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') @@ -360,7 +393,7 @@ func (s *Store) ResolveItemIncludeDeleted(workspaceID, slugOrRef string) (*model &item.Content, &item.Fields, &item.Tags, &pinned, &item.SortOrder, &item.ParentID, &item.AssignedUserID, &item.AgentRoleID, &item.RoleSortOrder, &item.CreatedBy, &item.LastModifiedBy, &item.Source, - &item.ItemNumber, &createdAt, &updatedAt, &deletedAt, + &item.ItemNumber, &item.Seq, &createdAt, &updatedAt, &deletedAt, &item.CollectionSlug, &item.CollectionName, &item.CollectionIcon, &item.CollectionPrefix, &item.AssignedUserName, &item.AssignedUserEmail, &item.AgentRoleName, &item.AgentRoleSlug, &item.AgentRoleIcon, @@ -458,7 +491,7 @@ func (s *Store) GetItemIncludeDeleted(id string) (*models.Item, error) { SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, i.deleted_at, + i.item_number, i.seq, i.created_at, i.updated_at, i.deleted_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') @@ -472,7 +505,7 @@ func (s *Store) GetItemIncludeDeleted(id string) (*models.Item, error) { &item.Content, &item.Fields, &item.Tags, &pinned, &item.SortOrder, &item.ParentID, &item.AssignedUserID, &item.AgentRoleID, &item.RoleSortOrder, &item.CreatedBy, &item.LastModifiedBy, &item.Source, - &item.ItemNumber, &createdAt, &updatedAt, &deletedAt, + &item.ItemNumber, &item.Seq, &createdAt, &updatedAt, &deletedAt, &item.CollectionSlug, &item.CollectionName, &item.CollectionIcon, &item.CollectionPrefix, &item.AssignedUserName, &item.AssignedUserEmail, &item.AgentRoleName, &item.AgentRoleSlug, &item.AgentRoleIcon, @@ -504,7 +537,7 @@ func (s *Store) GetItemBySlugIncludeDeleted(workspaceID, slug string) (*models.I SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, i.deleted_at, + i.item_number, i.seq, i.created_at, i.updated_at, i.deleted_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') @@ -518,7 +551,7 @@ func (s *Store) GetItemBySlugIncludeDeleted(workspaceID, slug string) (*models.I &item.Content, &item.Fields, &item.Tags, &pinned, &item.SortOrder, &item.ParentID, &item.AssignedUserID, &item.AgentRoleID, &item.RoleSortOrder, &item.CreatedBy, &item.LastModifiedBy, &item.Source, - &item.ItemNumber, &createdAt, &updatedAt, &deletedAt, + &item.ItemNumber, &item.Seq, &createdAt, &updatedAt, &deletedAt, &item.CollectionSlug, &item.CollectionName, &item.CollectionIcon, &item.CollectionPrefix, &item.AssignedUserName, &item.AssignedUserEmail, &item.AgentRoleName, &item.AgentRoleSlug, &item.AgentRoleIcon, @@ -558,7 +591,7 @@ func (s *Store) ListItems(workspaceID string, params models.ItemListParams) ([]m SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') @@ -721,7 +754,7 @@ func (s *Store) ListItemsIndex(workspaceID string, params ItemIndexParams) ([]mo SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') @@ -872,7 +905,7 @@ func scanItemsIndex(rows *sql.Rows) ([]models.Item, error) { &item.Fields, &item.Tags, &pinned, &item.SortOrder, &item.ParentID, &item.AssignedUserID, &item.AgentRoleID, &item.RoleSortOrder, &item.CreatedBy, &item.LastModifiedBy, &item.Source, - &item.ItemNumber, &createdAt, &updatedAt, + &item.ItemNumber, &item.Seq, &createdAt, &updatedAt, &item.CollectionSlug, &item.CollectionName, &item.CollectionIcon, &item.CollectionPrefix, &item.AssignedUserName, &item.AssignedUserEmail, &item.AgentRoleName, &item.AgentRoleSlug, &item.AgentRoleIcon, @@ -902,7 +935,7 @@ func (s *Store) listItemsFTS(workspaceID string, params models.ItemListParams) ( SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') @@ -928,7 +961,7 @@ func (s *Store) listItemsFTS(workspaceID string, params models.ItemListParams) ( SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') @@ -1080,6 +1113,12 @@ func (s *Store) UpdateItem(id string, input models.ItemUpdate) (*models.Item, er } defer tx.Rollback() + // Serialize concurrent seq assignments per workspace on Postgres + // (no-op on SQLite). Held until COMMIT / ROLLBACK. + if err := s.acquireWorkspaceSeqLock(tx, existing.WorkspaceID); err != nil { + return nil, err + } + ts := now() // Create version if content is changing @@ -1128,9 +1167,11 @@ func (s *Store) UpdateItem(id string, input models.ItemUpdate) (*models.Item, er } } - // Build update query - sets := []string{"updated_at = ?"} - args := []interface{}{ts} + // Build update query. Every mutation bumps seq to MAX(seq)+1 per + // workspace; the local-first read model uses that as a cursor (see + // nextWorkspaceSeqSubquery / PLAN-1343). + sets := []string{"updated_at = ?", "seq = " + nextWorkspaceSeqSubquery} + args := []interface{}{ts, existing.WorkspaceID} if input.Title != nil { sets = append(sets, "title = ?") @@ -1262,12 +1303,39 @@ func (s *Store) UpdateItem(id string, input models.ItemUpdate) (*models.Item, er return s.GetItem(id) } +// DeleteItem soft-deletes the item by stamping deleted_at and bumping +// the workspace-scoped seq so delta-sync clients see the tombstone. +// The seq bump uses the same MAX(seq)+1 subquery the other mutations +// rely on; the advisory lock keeps concurrent Postgres writes from +// racing on it. func (s *Store) DeleteItem(id string) error { + // Look up the workspace before the write so we can key the + // advisory lock and the seq subquery. The lookup tolerates + // already-deleted items (we still need to short-circuit cleanly + // in that case) by reading the include-deleted variant. + existing, err := s.GetItemIncludeDeleted(id) + if err != nil { + return err + } + if existing == nil { + return sql.ErrNoRows + } + + tx, err := s.db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + + if err := s.acquireWorkspaceSeqLock(tx, existing.WorkspaceID); err != nil { + return err + } + ts := now() - result, err := s.db.Exec(s.q(` - UPDATE items SET deleted_at = ?, updated_at = ? + result, err := tx.Exec(s.q(` + UPDATE items SET deleted_at = ?, updated_at = ?, seq = `+nextWorkspaceSeqSubquery+` WHERE id = ? AND deleted_at IS NULL - `), ts, ts, id) + `), ts, ts, existing.WorkspaceID, id) if err != nil { return fmt.Errorf("delete item: %w", err) } @@ -1275,15 +1343,36 @@ func (s *Store) DeleteItem(id string) error { if rows == 0 { return sql.ErrNoRows } - return nil + return tx.Commit() } +// RestoreItem un-archives a soft-deleted item and bumps the +// workspace-scoped seq so delta-sync clients re-materialize the row. +// Same lock + subquery shape as DeleteItem. func (s *Store) RestoreItem(id string) (*models.Item, error) { + existing, err := s.GetItemIncludeDeleted(id) + if err != nil { + return nil, err + } + if existing == nil { + return nil, sql.ErrNoRows + } + + tx, err := s.db.Begin() + if err != nil { + return nil, err + } + defer tx.Rollback() + + if err := s.acquireWorkspaceSeqLock(tx, existing.WorkspaceID); err != nil { + return nil, err + } + ts := now() - result, err := s.db.Exec(s.q(` - UPDATE items SET deleted_at = NULL, updated_at = ? + result, err := tx.Exec(s.q(` + UPDATE items SET deleted_at = NULL, updated_at = ?, seq = `+nextWorkspaceSeqSubquery+` WHERE id = ? AND deleted_at IS NOT NULL - `), ts, id) + `), ts, existing.WorkspaceID, id) if err != nil { return nil, fmt.Errorf("restore item: %w", err) } @@ -1291,6 +1380,9 @@ func (s *Store) RestoreItem(id string) (*models.Item, error) { if rows == 0 { return nil, sql.ErrNoRows } + if err := tx.Commit(); err != nil { + return nil, err + } return s.GetItem(id) } @@ -1315,7 +1407,7 @@ func (s *Store) SearchItems(workspaceID, query string) ([]ItemSearchResult, erro SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, ''), @@ -1343,7 +1435,7 @@ func (s *Store) SearchItems(workspaceID, query string) ([]ItemSearchResult, erro SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, ''), @@ -1389,7 +1481,7 @@ func (s *Store) SearchItems(workspaceID, query string) ([]ItemSearchResult, erro &r.Item.Content, &r.Item.Fields, &r.Item.Tags, &pinned, &r.Item.SortOrder, &r.Item.ParentID, &r.Item.AssignedUserID, &r.Item.AgentRoleID, &r.Item.RoleSortOrder, &r.Item.CreatedBy, &r.Item.LastModifiedBy, - &r.Item.Source, &r.Item.ItemNumber, &createdAt, &updatedAt, + &r.Item.Source, &r.Item.ItemNumber, &r.Item.Seq, &createdAt, &updatedAt, &r.Item.CollectionSlug, &r.Item.CollectionName, &r.Item.CollectionIcon, &r.Item.CollectionPrefix, &r.Item.AssignedUserName, &r.Item.AssignedUserEmail, &r.Item.AgentRoleName, &r.Item.AgentRoleSlug, &r.Item.AgentRoleIcon, @@ -2006,7 +2098,7 @@ func (s *Store) GetChildItems(parentItemID string) ([]models.Item, error) { SELECT DISTINCT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') @@ -2075,16 +2167,43 @@ func (s *Store) PopulateHasChildren(items []models.Item) { // It updates the collection_id and fields JSON. The item_number is preserved // because numbering is workspace-global — the number stays the same, only the // collection prefix changes (e.g. IDEA-42 → BUG-42). +// +// The move also bumps the workspace-scoped seq so delta-sync clients +// see the collection change (PLAN-1343 / TASK-1352). Without it a +// client polling /items-changes?since=cursor would render the item +// under its old collection until a full refresh. func (s *Store) MoveItem(itemID, targetCollectionID, newFieldsJSON string) (*models.Item, error) { - _, err := s.db.Exec(s.q(` + existing, err := s.GetItem(itemID) + if err != nil { + return nil, err + } + if existing == nil { + return nil, sql.ErrNoRows + } + + tx, err := s.db.Begin() + if err != nil { + return nil, err + } + defer tx.Rollback() + + if err := s.acquireWorkspaceSeqLock(tx, existing.WorkspaceID); err != nil { + return nil, err + } + + _, err = tx.Exec(s.q(` UPDATE items - SET collection_id = ?, fields = ?, updated_at = ? + SET collection_id = ?, fields = ?, updated_at = ?, seq = `+nextWorkspaceSeqSubquery+` WHERE id = ? AND deleted_at IS NULL`), - targetCollectionID, newFieldsJSON, time.Now().UTC().Format(time.RFC3339), itemID) + targetCollectionID, newFieldsJSON, time.Now().UTC().Format(time.RFC3339), existing.WorkspaceID, itemID) if err != nil { return nil, fmt.Errorf("move item: %w", err) } + if err := tx.Commit(); err != nil { + return nil, err + } + return s.GetItem(itemID) } @@ -2272,7 +2391,7 @@ func scanItems(rows *sql.Rows) ([]models.Item, error) { &item.Content, &item.Fields, &item.Tags, &pinned, &item.SortOrder, &item.ParentID, &item.AssignedUserID, &item.AgentRoleID, &item.RoleSortOrder, &item.CreatedBy, &item.LastModifiedBy, &item.Source, - &item.ItemNumber, &createdAt, &updatedAt, + &item.ItemNumber, &item.Seq, &createdAt, &updatedAt, &item.CollectionSlug, &item.CollectionName, &item.CollectionIcon, &item.CollectionPrefix, &item.AssignedUserName, &item.AssignedUserEmail, &item.AgentRoleName, &item.AgentRoleSlug, &item.AgentRoleIcon, @@ -2305,7 +2424,7 @@ func (s *Store) ItemsModifiedSince(workspaceID string, since time.Time) (updated SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') diff --git a/internal/store/items_test.go b/internal/store/items_test.go index b88ed732..49b563d7 100644 --- a/internal/store/items_test.go +++ b/internal/store/items_test.go @@ -3131,3 +3131,128 @@ func TestWorkspaceHasAgentActivityVisibility(t *testing.T) { t.Fatal("an empty visibility set must short-circuit to false") } } + +// TestItemSeqMonotonic verifies that the workspace-scoped `seq` column +// is stamped strictly monotonically across every mutation +// (create / update / soft-delete / restore). This is the cursor +// mechanic the local-first read model relies on for delta sync +// (PLAN-1343 / TASK-1352). +func TestItemSeqMonotonic(t *testing.T) { + s := testStore(t) + ws := createTestWorkspace(t, s, "Test") + col := createTestCollection(t, s, ws.ID, "Tasks") + + // CREATE: three rows, each must have seq strictly greater than + // the previous one. + a := createTestItem(t, s, ws.ID, col.ID, "A", "alpha") + b := createTestItem(t, s, ws.ID, col.ID, "B", "beta") + c := createTestItem(t, s, ws.ID, col.ID, "C", "gamma") + + if a.Seq <= 0 { + t.Fatalf("create: a.Seq = %d; want > 0", a.Seq) + } + if b.Seq <= a.Seq { + t.Fatalf("create: b.Seq = %d; want > a.Seq = %d", b.Seq, a.Seq) + } + if c.Seq <= b.Seq { + t.Fatalf("create: c.Seq = %d; want > b.Seq = %d", c.Seq, b.Seq) + } + + // UPDATE: bumps seq above every prior value. + newTitle := "A-updated" + aUpd, err := s.UpdateItem(a.ID, models.ItemUpdate{Title: &newTitle}) + if err != nil { + t.Fatalf("update item: %v", err) + } + if aUpd.Seq <= c.Seq { + t.Fatalf("update: aUpd.Seq = %d; want > c.Seq = %d", aUpd.Seq, c.Seq) + } + + // SOFT-DELETE: tombstone gets a fresh seq above the post-update + // floor so delta-sync clients see the deletion. + if err := s.DeleteItem(b.ID); err != nil { + t.Fatalf("delete item: %v", err) + } + bDel, err := s.GetItemIncludeDeleted(b.ID) + if err != nil || bDel == nil { + t.Fatalf("get deleted item: %v", err) + } + if bDel.DeletedAt == nil { + t.Fatal("deleted item is missing deleted_at") + } + if bDel.Seq <= aUpd.Seq { + t.Fatalf("delete: bDel.Seq = %d; want > aUpd.Seq = %d", bDel.Seq, aUpd.Seq) + } + + // RESTORE: un-archive bumps seq again. + bRestored, err := s.RestoreItem(b.ID) + if err != nil { + t.Fatalf("restore item: %v", err) + } + if bRestored.DeletedAt != nil { + t.Fatal("restored item still has deleted_at") + } + if bRestored.Seq <= bDel.Seq { + t.Fatalf("restore: bRestored.Seq = %d; want > bDel.Seq = %d", bRestored.Seq, bDel.Seq) + } + + // Sanity-check: another update on c bumps past the restore. + newCContent := "gamma-prime" + cUpd, err := s.UpdateItem(c.ID, models.ItemUpdate{Content: &newCContent}) + if err != nil { + t.Fatalf("second update: %v", err) + } + if cUpd.Seq <= bRestored.Seq { + t.Fatalf("update after restore: cUpd.Seq = %d; want > bRestored.Seq = %d", cUpd.Seq, bRestored.Seq) + } +} + +// TestItemSeqWorkspaceScoped verifies that the seq counter is +// scoped per workspace: a busy workspace's seq range does not +// affect another workspace's monotonic floor (DOC-1342 design +// decision #1). +func TestItemSeqWorkspaceScoped(t *testing.T) { + s := testStore(t) + ws1 := createTestWorkspace(t, s, "WS1") + ws2 := createTestWorkspace(t, s, "WS2") + col1 := createTestCollection(t, s, ws1.ID, "Tasks") + col2 := createTestCollection(t, s, ws2.ID, "Tasks") + + // Create three items in ws1 to bump its seq. + createTestItem(t, s, ws1.ID, col1.ID, "ws1-a", "") + createTestItem(t, s, ws1.ID, col1.ID, "ws1-b", "") + ws1Last := createTestItem(t, s, ws1.ID, col1.ID, "ws1-c", "") + + // ws2's first item must start at seq=1, NOT ws1Last.Seq+1. + ws2First := createTestItem(t, s, ws2.ID, col2.ID, "ws2-a", "") + if ws2First.Seq != 1 { + t.Fatalf("ws2 first item Seq = %d; want 1 (per-workspace counter)", ws2First.Seq) + } + if ws2First.Seq >= ws1Last.Seq { + t.Fatalf("ws2.Seq (%d) >= ws1.Seq (%d) — ws2 is following ws1's range, not its own", ws2First.Seq, ws1Last.Seq) + } +} + +// TestItemSeqBackfillNonZero verifies that the migration backfill +// assigned non-zero seq values to pre-existing rows. We can't easily +// simulate "pre-migration" rows in tests (the migration runs once on +// store init), so this test inserts a fresh batch and confirms each +// row's seq is non-zero and unique within the workspace. +func TestItemSeqBackfillNonZero(t *testing.T) { + s := testStore(t) + ws := createTestWorkspace(t, s, "Test") + col := createTestCollection(t, s, ws.ID, "Tasks") + + const n = 5 + seqs := make(map[int64]bool, n) + for i := 0; i < n; i++ { + item := createTestItem(t, s, ws.ID, col.ID, fmt.Sprintf("item-%d", i), "") + if item.Seq == 0 { + t.Fatalf("row %d: Seq = 0; expected non-zero", i) + } + if seqs[item.Seq] { + t.Fatalf("row %d: Seq = %d collides with prior row in same workspace", i, item.Seq) + } + seqs[item.Seq] = true + } +} diff --git a/internal/store/migrations/053_items_seq.sql b/internal/store/migrations/053_items_seq.sql new file mode 100644 index 00000000..9b3b1c78 --- /dev/null +++ b/internal/store/migrations/053_items_seq.sql @@ -0,0 +1,42 @@ +-- Migration 053: items.seq + workspace-scoped monotonic counter (TASK-1352). +-- +-- Adds a per-workspace monotonically-increasing `seq` column that bumps on +-- every items mutation (create / update / soft-delete / restore). Foundation +-- for the local-first read model's delta-sync cursor (PLAN-1343, +-- DOC-1342 design decision #1). +-- +-- Robust against the clock-skew / same-millisecond-write / NTP-step +-- correctness holes that an `updated_at` watermark would carry. Each +-- mutation reads `MAX(seq) + 1 WHERE workspace_id = ?` inside the same +-- transaction that performs the write; SQLite's single-writer rule +-- makes that collision-free without an extra lock. +-- +-- `seq` is workspace-scoped, NOT global — each workspace has its own +-- monotonic counter so a busy workspace can't fragment another's range +-- and so cursors from one workspace can't accidentally apply to +-- another. + +ALTER TABLE items ADD COLUMN seq INTEGER NOT NULL DEFAULT 0; + +-- Backfill: assign every existing row a seq in (workspace_id, +-- updated_at, id) order. ROW_NUMBER() OVER PARTITION gives each +-- workspace its own 1..N sequence so the post-migration MAX(seq) per +-- workspace == the row count of that workspace. SQLite supports +-- UPDATE..FROM since 3.33, which is well below the minimum version +-- the rest of the migrations already require. +UPDATE items +SET seq = sub.rn +FROM ( + SELECT id, + ROW_NUMBER() OVER (PARTITION BY workspace_id ORDER BY updated_at, id) AS rn + FROM items +) AS sub +WHERE items.id = sub.id; + +-- Supports two read patterns: +-- * /items-index cursor read: MAX(seq) per workspace +-- * /items-changes?since= range scan: seq > ? ORDER BY seq ASC +-- DESC is chosen to match the MAX read; the planner can still use the +-- index for the ASC range scan because the column is sorted on the +-- workspace_id leading edge. +CREATE INDEX IF NOT EXISTS idx_items_workspace_seq ON items(workspace_id, seq DESC); diff --git a/internal/store/pgmigrations/032_items_seq.sql b/internal/store/pgmigrations/032_items_seq.sql new file mode 100644 index 00000000..48a153f2 --- /dev/null +++ b/internal/store/pgmigrations/032_items_seq.sql @@ -0,0 +1,22 @@ +-- Postgres mirror of migrations/053_items_seq.sql (TASK-1352). +-- See the SQLite migration for context. +-- +-- The Postgres path also takes a workspace-scoped pg_advisory_xact_lock +-- at write time to serialize concurrent seq assignments — SQLite's +-- single-writer rule handles that implicitly, but Postgres needs an +-- explicit guard so two concurrent UPDATEs don't read the same MAX(seq) +-- and produce duplicate values. See `acquireWorkspaceSeqLock` in +-- internal/store/items.go. + +ALTER TABLE items ADD COLUMN IF NOT EXISTS seq BIGINT NOT NULL DEFAULT 0; + +UPDATE items +SET seq = sub.rn +FROM ( + SELECT id, + ROW_NUMBER() OVER (PARTITION BY workspace_id ORDER BY updated_at, id) AS rn + FROM items +) AS sub +WHERE items.id = sub.id; + +CREATE INDEX IF NOT EXISTS idx_items_workspace_seq ON items(workspace_id, seq DESC); diff --git a/internal/store/search.go b/internal/store/search.go index be076c4e..73f620ab 100644 --- a/internal/store/search.go +++ b/internal/store/search.go @@ -113,7 +113,7 @@ func (s *Store) Search(params SearchParams) (*SearchResponse, error) { SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') @@ -185,7 +185,7 @@ func (s *Store) Search(params SearchParams) (*SearchResponse, error) { &r.Item.Content, &r.Item.Fields, &r.Item.Tags, &pinned, &r.Item.SortOrder, &r.Item.ParentID, &r.Item.AssignedUserID, &r.Item.AgentRoleID, &r.Item.RoleSortOrder, &r.Item.CreatedBy, &r.Item.LastModifiedBy, - &r.Item.Source, &r.Item.ItemNumber, &createdAt, &updatedAt, + &r.Item.Source, &r.Item.ItemNumber, &r.Item.Seq, &createdAt, &updatedAt, &r.Item.CollectionSlug, &r.Item.CollectionName, &r.Item.CollectionIcon, &r.Item.CollectionPrefix, &r.Item.AssignedUserName, &r.Item.AssignedUserEmail, &r.Item.AgentRoleName, &r.Item.AgentRoleSlug, &r.Item.AgentRoleIcon, @@ -215,7 +215,7 @@ func (s *Store) Search(params SearchParams) (*SearchResponse, error) { SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, '') @@ -293,7 +293,7 @@ func (s *Store) Search(params SearchParams) (*SearchResponse, error) { &r.Item.Content, &r.Item.Fields, &r.Item.Tags, &pinned, &r.Item.SortOrder, &r.Item.ParentID, &r.Item.AssignedUserID, &r.Item.AgentRoleID, &r.Item.RoleSortOrder, &r.Item.CreatedBy, &r.Item.LastModifiedBy, - &r.Item.Source, &r.Item.ItemNumber, &createdAt, &updatedAt, + &r.Item.Source, &r.Item.ItemNumber, &r.Item.Seq, &createdAt, &updatedAt, &r.Item.CollectionSlug, &r.Item.CollectionName, &r.Item.CollectionIcon, &r.Item.CollectionPrefix, &r.Item.AssignedUserName, &r.Item.AssignedUserEmail, &r.Item.AgentRoleName, &r.Item.AgentRoleSlug, &r.Item.AgentRoleIcon, @@ -351,7 +351,7 @@ func (s *Store) Search(params SearchParams) (*SearchResponse, error) { SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, ''), @@ -381,7 +381,7 @@ func (s *Store) Search(params SearchParams) (*SearchResponse, error) { SELECT i.id, i.workspace_id, i.collection_id, i.title, i.slug, i.content, i.fields, i.tags, i.pinned, i.sort_order, i.parent_id, i.assigned_user_id, i.agent_role_id, i.role_sort_order, i.created_by, i.last_modified_by, i.source, - i.item_number, i.created_at, i.updated_at, + i.item_number, i.seq, i.created_at, i.updated_at, c.slug, c.name, c.icon, c.prefix, COALESCE(au.name, ''), COALESCE(au.email, ''), COALESCE(ar.name, ''), COALESCE(ar.slug, ''), COALESCE(ar.icon, ''), @@ -578,7 +578,7 @@ func (s *Store) Search(params SearchParams) (*SearchResponse, error) { &r.Item.Content, &r.Item.Fields, &r.Item.Tags, &pinned, &r.Item.SortOrder, &r.Item.ParentID, &r.Item.AssignedUserID, &r.Item.AgentRoleID, &r.Item.RoleSortOrder, &r.Item.CreatedBy, &r.Item.LastModifiedBy, - &r.Item.Source, &r.Item.ItemNumber, &createdAt, &updatedAt, + &r.Item.Source, &r.Item.ItemNumber, &r.Item.Seq, &createdAt, &updatedAt, &r.Item.CollectionSlug, &r.Item.CollectionName, &r.Item.CollectionIcon, &r.Item.CollectionPrefix, &r.Item.AssignedUserName, &r.Item.AssignedUserEmail, &r.Item.AgentRoleName, &r.Item.AgentRoleSlug, &r.Item.AgentRoleIcon,