From dfd3811eee5083995823dde03b40cd618a540baa Mon Sep 17 00:00:00 2001 From: xarmian Date: Sat, 30 May 2026 18:41:55 -0400 Subject: [PATCH] feat(api): bulk-mutation endpoint + single SSE batch event (TASK-1668) (#669) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(api): bulk-mutation endpoint + single SSE batch event (TASK-1668) Add POST /workspaces/{ws}/items/bulk accepting item IDs + a verb (archive, move, tag, untag, set-priority, assign). The lane-header bulk actions operate on a whole filtered lane, so the endpoint emits ONE items_bulk_updated SSE event and ONE item.bulk_updated webhook for the batch instead of per-item fan-out. Reuses the existing store paths (UpdateItemWithPreCheck / MoveItem / DeleteItem) rather than re-implementing writes; the open-children guard runs per status-bearing move exactly as the single PATCH path does (force-overridable). Per-row failures are collected into the response envelope (updated/failed/total) rather than aborting the batch. Editor/owner gated. Frontend client + TS types follow in TASK-1669; UI wiring in TASK-1672. Parent: PLAN-1667. * fix(api): per-item visibility + collection-move guard on bulk endpoint per Codex review (round 1) - Enforce per-item collection visibility (checkItemVisible) in the bulk loop so a member with collection_access="specific" can't bulk-mutate items in hidden collections by guessing refs; report invisible rows as not-found. Also gate the move target collection on visibility. - Route bulk collection moves through MoveItemWithPreCheck with the open-children guard (destination schema), closing the bypass where a collection move + terminal status could mark a parent terminal with open children. Status-only moves already ran the guard. - Tests: status-move + collection-move guard coverage (reject + force override + mutation-safety). * fix(web): consume items_bulk_updated SSE event per Codex review (round 2) The bulk endpoint emits one items_bulk_updated event, but the SSE service only listened for the fixed ITEM_EVENTS list — so a bulk mutation left other tabs/sessions stale until an unrelated sync fired. Route the batch event through the existing sync_required path: it carries item_ids + a max seq but no per-item field payload, so an incremental /items-changes delta reconciles every affected row by seq. Broadcast so peer tabs reconcile too. * fix(api): scope bulk SSE event per-collection, drop item_ids per Codex review (round 3) The batch event published with an empty Collection, which the SSE filter treats as workspace-level: restricted members received bulk events for hidden collections (leaking item_ids/op/count) while guests with grants were dropped entirely and stayed stale. Emit one items_bulk_updated event per affected collection with Collection set, so the existing visibility filter routes it like any collection-scoped event. Drop per-item IDs from the SSE payload — a batch can't be item-grant-filtered for guests on a broadcast bus, so IDs would leak; recipients reconcile via the /items-changes delta, which is visibility-filtered server-side (Seq carries the cursor). The webhook (a trusted workspace integration) keeps the full id list. Test asserts the event is collection-scoped and carries no item_ids. * fix(api): bulk collection move notifies both source and target scopes per Codex review (round 4) A cross-collection move only emitted a batch event for the target collection, so a restricted member watching the source lane wouldn't reconcile the item leaving it. Notify both the source and target collection scopes for moves (still no per-item IDs). Test asserts both events fire. * fix(api): suppress itemless batch SSE events for item-grant-only subscribers per Codex review (round 5) A guest/restricted member with only item-level grants in a collection could still receive the collection-scoped items_bulk_updated event (itemless), learning op/count/timing for items they can't see. Extract the SSE visibility filter into sseEventVisibleFor and add a rule: itemless collection-scoped events go only to subscribers with FULL collection access; item-grant-only subscribers reconcile their granted items via the next resume/reconnect /items-changes sync instead. Adds a unit test covering the visibility matrix. * fix(api): validate status override against target schema on bulk collection move per Codex review (round 6) A status override on a collection move was applied after MigrateFields but never validated against the target schema, so an out-of-options value (e.g. status=bogus) could be written. Run ValidateFields on the final field map before the move. Test asserts the invalid value is rejected per-row and the item stays put. --- internal/events/bus.go | 21 + internal/server/handlers_events.go | 70 +- .../server/handlers_events_filter_test.go | 58 ++ internal/server/handlers_items_bulk.go | 617 ++++++++++++++++++ internal/server/handlers_items_bulk_test.go | 464 +++++++++++++ internal/server/server.go | 4 + web/src/lib/services/sse.svelte.ts | 12 + 7 files changed, 1220 insertions(+), 26 deletions(-) create mode 100644 internal/server/handlers_events_filter_test.go create mode 100644 internal/server/handlers_items_bulk.go create mode 100644 internal/server/handlers_items_bulk_test.go diff --git a/internal/events/bus.go b/internal/events/bus.go index 9c3a58d4..339ec8b9 100644 --- a/internal/events/bus.go +++ b/internal/events/bus.go @@ -36,6 +36,13 @@ const ( // Composite events ItemUpdatedWithComment = "item_updated_with_comment" + + // Batch events. Emitted once for a whole bulk mutation (TASK-1668) + // instead of one ItemUpdated/ItemArchived per row — the lane-header + // bulk actions (archive/move/tag/untag/set-priority/assign all) can + // touch a whole filtered lane, so per-item fan-out would flood both + // SSE subscribers and webhooks. + ItemsBulkUpdated = "items_bulk_updated" ) // Default replay buffer settings. @@ -69,6 +76,20 @@ type Event struct { // comment_*, reaction_*) and for legacy publishers that // haven't been upgraded. Seq int64 `json:"seq,omitempty"` + // Op / Count describe a batch event (ItemsBulkUpdated, TASK-1668). + // Op is the verb applied (archive/move/tag/untag/set-priority/ + // assign); Count is the number of items affected in this event's + // Collection. Zero/empty for single-item events. + // + // A batch event is scoped to ONE Collection (the bulk endpoint emits + // one per affected collection) so the SSE visibility filter routes + // it like any collection-scoped event. It deliberately carries NO + // per-item IDs: a batch can't be item-grant-filtered for guests on a + // broadcast bus, so IDs would leak. Recipients react by running a + // /items-changes delta, which IS visibility-filtered server-side; + // Seq holds the max seq across the batch as the reconcile cursor. + Op string `json:"op,omitempty"` + Count int `json:"count,omitempty"` } // EventBus is the interface for pub/sub event distribution. diff --git a/internal/server/handlers_events.go b/internal/server/handlers_events.go index e2fe3276..0431d559 100644 --- a/internal/server/handlers_events.go +++ b/internal/server/handlers_events.go @@ -148,32 +148,7 @@ func (s *Server) handleSSE(w http.ResponseWriter, r *http.Request) { // Reads from the current `vis` snapshot so recomputes on each // revalidation tick take effect immediately for the next event. sseEventVisible := func(event events.Event) bool { - // User-scoped events (e.g. star/unstar) are only sent to the user who triggered them - if event.UserID != "" && event.UserID != sseUserID { - return false - } - collection := event.Collection - itemID := event.ItemID - if vis.visibleSlugSet == nil { - return true // all access - } - if collection == "" { - // Events without a collection (workspace-level, legacy docs) are - // only sent to actual members, not guests — they may contain - // operational metadata like member invites, role changes, etc. - if vis.isGuest { - return false - } - return true - } - if !vis.visibleSlugSet[collection] { - return false - } - // For guests with item-level grants, additionally check the item ID - if vis.grantedItemSet != nil && !vis.fullCollSet[collection] && itemID != "" { - return vis.grantedItemSet[itemID] - } - return true + return sseEventVisibleFor(vis, sseUserID, event) } // Send initial connected event. If even this first write fails the @@ -344,6 +319,49 @@ var sseMembershipRevalInterval = 60 * time.Second // fields are rebuilt atomically on each revalidation tick so that // permission-tightening changes take effect for the very next event // without tearing down and rebuilding the stream. +// sseEventVisibleFor decides whether a single event should be delivered +// to a subscriber with the given visibility snapshot. Extracted from the +// per-connection closure so the rule matrix is unit-testable in +// isolation (the live handler just binds `vis` + `sseUserID`). +func sseEventVisibleFor(vis sseVisibility, sseUserID string, event events.Event) bool { + // User-scoped events (e.g. star/unstar) only go to the triggering user. + if event.UserID != "" && event.UserID != sseUserID { + return false + } + collection := event.Collection + itemID := event.ItemID + if vis.visibleSlugSet == nil { + return true // all access + } + if collection == "" { + // Events without a collection (workspace-level, legacy docs) are + // only sent to actual members, not guests — they may contain + // operational metadata like member invites, role changes, etc. + if vis.isGuest { + return false + } + return true + } + if !vis.visibleSlugSet[collection] { + return false + } + // For subscribers filtered to item-level grants in this collection + // (no full-collection access): + if vis.grantedItemSet != nil && !vis.fullCollSet[collection] { + // Item-scoped events: gate on the specific granted item. + if itemID != "" { + return vis.grantedItemSet[itemID] + } + // Itemless collection-scoped events (e.g. the items_bulk_updated + // batch event, TASK-1668) can't be item-grant-filtered — they'd + // otherwise leak op/count/timing for items the subscriber can't + // see. Suppress; these subscribers reconcile their granted items + // via the next resume/reconnect /items-changes sync instead. + return false + } + return true +} + type sseVisibility struct { // visibleSlugSet == nil → user has unrestricted access (admin / owner / // editor with no collection scope). A non-nil empty map means deny all diff --git a/internal/server/handlers_events_filter_test.go b/internal/server/handlers_events_filter_test.go new file mode 100644 index 00000000..69366b7d --- /dev/null +++ b/internal/server/handlers_events_filter_test.go @@ -0,0 +1,58 @@ +package server + +import ( + "testing" + + "github.com/PerpetualSoftware/pad/internal/events" +) + +// TestSSEEventVisibleFor covers the visibility matrix, with focus on the +// itemless batch event (items_bulk_updated, TASK-1668): an item-grant- +// only subscriber must NOT receive a collection-scoped event that carries +// no item ID, since it can't be item-filtered and would leak op/count/ +// timing for items they can't see. +func TestSSEEventVisibleFor(t *testing.T) { + bulk := func(coll string) events.Event { + return events.Event{Type: events.ItemsBulkUpdated, Collection: coll, Op: "archive", Count: 3} + } + perItem := func(coll, itemID string) events.Event { + return events.Event{Type: events.ItemUpdated, Collection: coll, ItemID: itemID} + } + + allAccess := sseVisibility{visibleSlugSet: nil} + + fullColl := sseVisibility{ + visibleSlugSet: map[string]bool{"tasks": true}, + grantedItemSet: map[string]bool{"item-1": true}, + fullCollSet: map[string]bool{"tasks": true}, + } + + itemGrantOnly := sseVisibility{ + visibleSlugSet: map[string]bool{"tasks": true}, + grantedItemSet: map[string]bool{"item-1": true}, + fullCollSet: map[string]bool{}, // no full access to tasks + isGuest: true, + } + + cases := []struct { + name string + vis sseVisibility + ev events.Event + want bool + }{ + {"all-access sees bulk", allAccess, bulk("tasks"), true}, + {"full-collection access sees bulk", fullColl, bulk("tasks"), true}, + {"item-grant-only suppressed for itemless bulk", itemGrantOnly, bulk("tasks"), false}, + {"item-grant-only suppressed for bulk in invisible collection", itemGrantOnly, bulk("secrets"), false}, + {"item-grant-only still gets granted per-item event", itemGrantOnly, perItem("tasks", "item-1"), true}, + {"item-grant-only denied non-granted per-item event", itemGrantOnly, perItem("tasks", "item-2"), false}, + {"any subscriber denied bulk in unseen collection", fullColl, bulk("other"), false}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + if got := sseEventVisibleFor(tc.vis, "user-x", tc.ev); got != tc.want { + t.Errorf("got %v, want %v", got, tc.want) + } + }) + } +} diff --git a/internal/server/handlers_items_bulk.go b/internal/server/handlers_items_bulk.go new file mode 100644 index 00000000..d5fe86ed --- /dev/null +++ b/internal/server/handlers_items_bulk.go @@ -0,0 +1,617 @@ +package server + +import ( + "database/sql" + "encoding/json" + "fmt" + "net/http" + "strings" + + "github.com/PerpetualSoftware/pad/internal/events" + "github.com/PerpetualSoftware/pad/internal/items" + "github.com/PerpetualSoftware/pad/internal/models" +) + +// maxBulkItems caps how many items a single bulk request may touch. +// The lane-header bulk actions (TASK-1668) operate on a whole filtered +// lane, which is realistically tens of items; the cap is a guardrail +// against a pathological request, not an expected ceiling. +const maxBulkItems = 1000 + +// bulkItemsRequest is the body of POST /workspaces/{ws}/items/bulk. +// `ids` accepts issue refs (TASK-5) or UUIDs; `op` selects the verb. +// The remaining fields are op-specific params — see handleBulkItems for +// which op consumes which. +type bulkItemsRequest struct { + IDs []string `json:"ids"` + Op string `json:"op"` + + // move + Status string `json:"status,omitempty"` // move-to-status (within or across collection) + Collection string `json:"collection,omitempty"` // move-to-collection (target slug) + + // set-priority + Priority string `json:"priority,omitempty"` + + // tag / untag + Tags []string `json:"tags,omitempty"` + + // assign + AssignedUserID *string `json:"assigned_user_id,omitempty"` + AgentRoleID *string `json:"agent_role_id,omitempty"` + ClearAssignedUser bool `json:"clear_assigned_user,omitempty"` + ClearAgentRole bool `json:"clear_agent_role,omitempty"` + + // Force overrides the open-children guard on status-bearing moves, + // mirroring `pad item update --force` and the move handler's + // ?force=true. No effect on ops that don't flip a terminal status. + Force bool `json:"force,omitempty"` +} + +// bulkItemOutcome is one successfully-mutated row. +type bulkItemOutcome struct { + Ref string `json:"ref"` + ID string `json:"id"` +} + +// bulkItemFailure is one row that failed, carrying the structured +// server error (code + details) when present — e.g. an open_children +// rejection — so MCP/web callers see the same shape the single PATCH +// surfaces, not just a flattened string. +type bulkItemFailure struct { + Ref string `json:"ref"` + Error string `json:"error"` + Code string `json:"code,omitempty"` + Details json.RawMessage `json:"details,omitempty"` +} + +// bulkItemsResponse is the structured envelope returned to the caller. +type bulkItemsResponse struct { + Op string `json:"op"` + Updated []bulkItemOutcome `json:"updated"` + Failed []bulkItemFailure `json:"failed"` + Total int `json:"total"` +} + +// bulkOpError carries a per-row failure with an optional structured +// code/details (currently only open_children). +type bulkOpError struct { + message string + code string + details json.RawMessage +} + +func (e *bulkOpError) Error() string { return e.message } + +// handleBulkItems applies one mutation verb to many items in a single +// request, emitting ONE SSE batch event and ONE webhook for the whole +// batch instead of per-item fan-out (TASK-1668). Editor/owner gated: +// the lane-header bulk actions are `canEdit`-only in the UI, so the +// endpoint requires workspace editor role (owner satisfies it too). +// +// Reuses the store mutation paths (UpdateItemWithPreCheck / MoveItem / +// DeleteItem) rather than re-implementing writes; the open-children +// guard runs per status-bearing move exactly as the single PATCH path +// does. Per-row failures are collected, not fatal — the response +// envelope reports updated vs failed so the caller can react. +func (s *Server) handleBulkItems(w http.ResponseWriter, r *http.Request) { + workspaceID, ok := s.getWorkspaceID(w, r) + if !ok { + return + } + + // Owner/editor gated. Bulk lane actions are canEdit-only; viewers + // and guests (grant-based access) cannot bulk-mutate. + if !requireRole(r, "editor") { + writeError(w, http.StatusForbidden, "forbidden", "Bulk mutations require editor or owner role") + return + } + + var req bulkItemsRequest + if err := decodeJSON(r, &req); err != nil { + writeError(w, http.StatusBadRequest, "bad_request", err.Error()) + return + } + if len(req.IDs) == 0 { + writeError(w, http.StatusBadRequest, "bad_request", "ids is required") + return + } + if len(req.IDs) > maxBulkItems { + writeError(w, http.StatusBadRequest, "bad_request", + fmt.Sprintf("too many items: %d (max %d per request)", len(req.IDs), maxBulkItems)) + return + } + + // Validate the verb + its required params up front so a malformed + // request fails fast before touching any rows. + switch req.Op { + case "archive": + case "move": + if req.Status == "" && req.Collection == "" { + writeError(w, http.StatusBadRequest, "bad_request", "move requires status or collection") + return + } + case "set-priority": + if req.Priority == "" { + writeError(w, http.StatusBadRequest, "bad_request", "set-priority requires priority") + return + } + case "tag", "untag": + if len(req.Tags) == 0 { + writeError(w, http.StatusBadRequest, "bad_request", req.Op+" requires tags") + return + } + case "assign": + if req.AssignedUserID == nil && req.AgentRoleID == nil && !req.ClearAssignedUser && !req.ClearAgentRole { + writeError(w, http.StatusBadRequest, "bad_request", "assign requires assigned_user_id, agent_role_id, or a clear flag") + return + } + default: + writeError(w, http.StatusBadRequest, "bad_request", fmt.Sprintf("unknown op %q", req.Op)) + return + } + + actor, source := actorFromRequest(r) + actorName := actorNameFromRequest(r) + user := currentUser(r) + role := workspaceRole(r) + + // Pre-compute collection visibility once. A member with + // collection_access="specific" (even an editor) must not be able to + // bulk-mutate items in collections they can't see — the single-item + // handlers enforce this per row via requireItemVisible, so the bulk + // path must too. nil = all-access (admin / fresh install). + visibleIDs, err := s.visibleCollectionIDs(r, workspaceID) + if err != nil { + writeInternalError(w, err) + return + } + + resp := bulkItemsResponse{ + Op: req.Op, + Updated: []bulkItemOutcome{}, + Failed: []bulkItemFailure{}, + } + affectedIDs := make([]string, 0, len(req.IDs)) + // Group affected items by their resulting collection slug so each + // batch SSE event carries a Collection and routes through the SSE + // visibility filter correctly (restricted members only get events + // for collections they can see; guests with grants still get the + // reconcile trigger). For the dominant lane-header case (one lane = + // one collection) this is exactly one event. + type collBatch struct { + count int + maxSeq int64 + } + batches := map[string]*collBatch{} + + for _, ref := range req.IDs { + item, err := s.store.ResolveItem(workspaceID, ref) + if err != nil { + resp.Failed = append(resp.Failed, bulkItemFailure{Ref: ref, Error: err.Error()}) + continue + } + if item == nil { + resp.Failed = append(resp.Failed, bulkItemFailure{Ref: ref, Error: "item not found"}) + continue + } + + // Per-item visibility gate. Report invisible items as + // not-found so a restricted member can't probe existence by ref. + visible, verr := s.checkItemVisible(workspaceID, item, user, role) + if verr != nil { + resp.Failed = append(resp.Failed, bulkItemFailure{Ref: ref, Error: verr.Error()}) + continue + } + if !visible { + resp.Failed = append(resp.Failed, bulkItemFailure{Ref: ref, Error: "item not found"}) + continue + } + + updated, opErr := s.applyBulkOp(r, workspaceID, item, &req, actor, source, visibleIDs) + if opErr != nil { + resp.Failed = append(resp.Failed, bulkItemFailure{ + Ref: itemRefOrSlug(*item), + Error: opErr.message, + Code: opErr.code, + Details: opErr.details, + }) + continue + } + + // Per-row activity log keeps the audit trail intact — it's a + // DB write, not the SSE/webhook fan-out the task is avoiding. + action := "updated" + if req.Op == "archive" { + action = "archived" + } + s.logActivityWithMeta(workspaceID, item.ID, action, + r, auditMeta(map[string]string{"bulk_op": req.Op})) + + resp.Updated = append(resp.Updated, bulkItemOutcome{Ref: itemRefOrSlug(*item), ID: item.ID}) + affectedIDs = append(affectedIDs, item.ID) + + // Determine which collection scopes need a reconcile event. + // Every op notifies the collection the item lives in. A + // cross-collection move ALSO notifies the target, so a member + // watching the source lane (the item leaving) AND one watching + // the target lane (the item arriving) both reconcile. (Move + // rejects same-collection, so source != target here.) + scopes := []string{item.CollectionSlug} + if req.Op == "move" && req.Collection != "" && req.Collection != item.CollectionSlug { + scopes = append(scopes, req.Collection) + } + var seq int64 + if updated != nil { + seq = updated.Seq + } + for _, sc := range scopes { + b := batches[sc] + if b == nil { + b = &collBatch{} + batches[sc] = b + } + b.count++ + if seq > b.maxSeq { + b.maxSeq = seq + } + } + } + + resp.Total = len(resp.Updated) + len(resp.Failed) + + // One SSE batch event per affected collection + ONE webhook for the + // whole batch (only when something actually changed). The core of + // the task: a whole-lane bulk action must not emit N per-item events. + // Per-collection (not fully per-item) keeps SSE visibility routing + // correct while still collapsing a lane action to a single event. + if len(affectedIDs) > 0 { + for collSlug, b := range batches { + s.publishBulkItemsEvent(workspaceID, req.Op, collSlug, b.count, actor, actorName, source, b.maxSeq) + } + // The webhook is a trusted workspace integration (not + // visibility-scoped per subscriber), so it keeps the full id + // list for the whole batch. + s.dispatchWebhook(workspaceID, "item.bulk_updated", map[string]any{ + "op": req.Op, + "count": len(affectedIDs), + "item_ids": affectedIDs, + }) + } + + writeJSON(w, http.StatusOK, resp) +} + +// applyBulkOp dispatches one verb against one item, reusing the same +// store paths as the single-item handlers. Returns the post-mutation +// item (for seq) on success, or a structured per-row error. +func (s *Server) applyBulkOp(r *http.Request, workspaceID string, item *models.Item, req *bulkItemsRequest, actor, source string, visibleIDs []string) (*models.Item, *bulkOpError) { + switch req.Op { + case "archive": + if err := s.store.DeleteItem(item.ID); err != nil { + return nil, &bulkOpError{message: err.Error()} + } + // DeleteItem bumps seq; re-read so the batch event carries the + // post-archive cursor. Falls back to the pre-delete row on a + // lookup miss (downstream backfills on a stale/zero seq). + if d, derr := s.store.GetItemIncludeDeleted(item.ID); derr == nil && d != nil { + return d, nil + } + return item, nil + + case "move": + if req.Collection != "" { + return s.bulkMoveCollection(r, workspaceID, item, req, visibleIDs) + } + // Status-only move = a field update on the same collection. + return s.bulkFieldUpdate(r, workspaceID, item, map[string]any{"status": req.Status}, req.Force, visibleIDs, actor, source) + + case "set-priority": + return s.bulkFieldUpdate(r, workspaceID, item, map[string]any{"priority": req.Priority}, req.Force, visibleIDs, actor, source) + + case "tag": + return s.bulkTagUpdate(item, req.Tags, true, actor, source) + + case "untag": + return s.bulkTagUpdate(item, req.Tags, false, actor, source) + + case "assign": + input := models.ItemUpdate{ + AssignedUserID: req.AssignedUserID, + AgentRoleID: req.AgentRoleID, + ClearAssignedUser: req.ClearAssignedUser, + ClearAgentRole: req.ClearAgentRole, + LastModifiedBy: actor, + Source: source, + } + updated, err := s.store.UpdateItem(item.ID, input) + if err != nil { + return nil, &bulkOpError{message: err.Error()} + } + return updated, nil + } + // Unreachable: op was validated in handleBulkItems. + return nil, &bulkOpError{message: fmt.Sprintf("unsupported op %q", req.Op)} +} + +// bulkFieldUpdate merges field changes into the item's existing fields, +// validates against the collection schema, runs the open-children guard +// (unless force), and writes via UpdateItemWithPreCheck — the same path +// the single PATCH handler uses. Used by status moves and set-priority. +func (s *Server) bulkFieldUpdate(r *http.Request, workspaceID string, item *models.Item, changes map[string]any, force bool, visibleIDs []string, actor, source string) (*models.Item, *bulkOpError) { + coll, err := s.store.GetCollection(item.CollectionID) + if err != nil || coll == nil { + return nil, &bulkOpError{message: "failed to load collection"} + } + var schema models.CollectionSchema + if err := json.Unmarshal([]byte(coll.Schema), &schema); err != nil { + return nil, &bulkOpError{message: "failed to parse collection schema"} + } + + fieldMap := make(map[string]any) + if item.Fields != "" && item.Fields != "{}" { + _ = json.Unmarshal([]byte(item.Fields), &fieldMap) + } + for k, v := range changes { + fieldMap[k] = v + } + + if err := items.ValidateFields(fieldMap, schema); err != nil { + return nil, &bulkOpError{message: err.Error(), code: "validation_error"} + } + if err := s.checkUniqueFields(workspaceID, item.CollectionID, item.ID, schema, fieldMap); err != nil { + return nil, &bulkOpError{message: err.Error(), code: "conflict"} + } + autoPopulateDates(fieldMap, item.Fields, schema) + + var precheck func(tx *sql.Tx, existing *models.Item) error + if !force { + var settings models.CollectionSettings + if coll.Settings != "" { + _ = json.Unmarshal([]byte(coll.Settings), &settings) + } + guestFull, guestGranted, gerr := s.guestResourceFilter(r, workspaceID) + if gerr != nil { + return nil, &bulkOpError{message: gerr.Error()} + } + gctx := openChildrenGuardContext{ + r: r, + workspaceID: workspaceID, + itemID: item.ID, + parentSchema: schema, + parentSettings: settings, + newFieldMap: fieldMap, + visibleCollectionIDs: visibleIDs, + guestFullCollIDs: guestFull, + guestGrantedItemIDs: guestGranted, + } + precheck = func(tx *sql.Tx, existing *models.Item) error { + txCtx := gctx + txCtx.currentFieldsJS = existing.Fields + details, derr := s.runOpenChildrenGuard(tx, txCtx) + if derr != nil { + return derr + } + if details != nil { + return &openChildrenGuardError{details: details} + } + return nil + } + } + + fieldsJSON, err := json.Marshal(fieldMap) + if err != nil { + return nil, &bulkOpError{message: "failed to marshal fields"} + } + fieldsStr := string(fieldsJSON) + input := models.ItemUpdate{ + Fields: &fieldsStr, + LastModifiedBy: actor, + Source: source, + } + + updated, err := s.store.UpdateItemWithPreCheck(item.ID, input, precheck) + if err != nil { + if details, ok := asOpenChildrenGuardError(err); ok { + raw, _ := json.Marshal(details) + return nil, &bulkOpError{ + message: "cannot mark item terminal while it has open children", + code: "open_children", + details: raw, + } + } + return nil, &bulkOpError{message: err.Error()} + } + if updated == nil { + return nil, &bulkOpError{message: "item not found"} + } + return updated, nil +} + +// bulkTagUpdate adds (add=true) or removes (add=false) the given tags +// from the item's tag set, preserving existing order and de-duplicating. +func (s *Server) bulkTagUpdate(item *models.Item, tags []string, add bool, actor, source string) (*models.Item, *bulkOpError) { + existing := []string{} + if item.Tags != "" && item.Tags != "[]" { + _ = json.Unmarshal([]byte(item.Tags), &existing) + } + + remove := make(map[string]bool) + if !add { + for _, t := range tags { + remove[t] = true + } + } + seen := make(map[string]bool) + result := make([]string, 0, len(existing)+len(tags)) + for _, t := range existing { + if remove[t] || seen[t] { + continue + } + seen[t] = true + result = append(result, t) + } + if add { + for _, t := range tags { + t = strings.TrimSpace(t) + if t == "" || seen[t] { + continue + } + seen[t] = true + result = append(result, t) + } + } + + tagsJSON, err := json.Marshal(result) + if err != nil { + return nil, &bulkOpError{message: "failed to marshal tags"} + } + tagsStr := string(tagsJSON) + updated, err := s.store.UpdateItem(item.ID, models.ItemUpdate{ + Tags: &tagsStr, + LastModifiedBy: actor, + Source: source, + }) + if err != nil { + return nil, &bulkOpError{message: err.Error()} + } + return updated, nil +} + +// bulkMoveCollection moves one item into req.Collection, migrating its +// fields between schemas — the same core as handleMoveItem, applied +// per row. A status override (req.Status) lands as a field override on +// the migrated set. +func (s *Server) bulkMoveCollection(r *http.Request, workspaceID string, item *models.Item, req *bulkItemsRequest, visibleIDs []string) (*models.Item, *bulkOpError) { + targetColl, err := s.store.GetCollectionBySlug(workspaceID, req.Collection) + if err != nil || targetColl == nil { + return nil, &bulkOpError{message: "target collection not found", code: "invalid_collection"} + } + // Target-collection visibility gate — same as handleMoveItem. A + // restricted member must not be able to move items into a + // collection they can't see. + if !isCollectionVisible(targetColl.ID, visibleIDs) { + return nil, &bulkOpError{message: "target collection not found", code: "invalid_collection"} + } + if targetColl.ID == item.CollectionID { + return nil, &bulkOpError{message: "item is already in this collection", code: "same_collection"} + } + sourceColl, err := s.store.GetCollection(item.CollectionID) + if err != nil || sourceColl == nil { + return nil, &bulkOpError{message: "failed to load source collection"} + } + + var sourceSchema, targetSchema models.CollectionSchema + if err := json.Unmarshal([]byte(sourceColl.Schema), &sourceSchema); err != nil { + return nil, &bulkOpError{message: "failed to parse source schema"} + } + if err := json.Unmarshal([]byte(targetColl.Schema), &targetSchema); err != nil { + return nil, &bulkOpError{message: "failed to parse target schema"} + } + + currentFields := make(map[string]any) + if err := json.Unmarshal([]byte(item.Fields), ¤tFields); err != nil { + currentFields = make(map[string]any) + } + + result := items.MigrateFields(currentFields, sourceSchema.Fields, targetSchema.Fields) + if req.Status != "" { + result.Fields["status"] = req.Status + } + if len(result.Errors) > 0 { + return nil, &bulkOpError{ + message: "required fields missing: " + strings.Join(result.Errors, ", "), + code: "missing_required_fields", + } + } + // Validate the final field map (including any status override) + // against the TARGET schema — MigrateFields validates migrated + // values but an override can smuggle in a value the target schema + // doesn't allow (e.g. a status not in the target's options). + if err := items.ValidateFields(result.Fields, targetSchema); err != nil { + return nil, &bulkOpError{message: err.Error(), code: "validation_error"} + } + + fieldsJSON, err := json.Marshal(result.Fields) + if err != nil { + return nil, &bulkOpError{message: "failed to serialize fields"} + } + + // Open-children guard (unless force), classified against the + // DESTINATION schema — same as handleMoveItem. A collection move + // that also sets a terminal status would otherwise mark a parent + // terminal while it still has open children. Routing every + // collection move through MoveItemWithPreCheck closes the bypass + // Codex flagged (a status-only move already ran the guard). + var precheck func(tx *sql.Tx, existing *models.Item) error + if !req.Force { + var destSettings models.CollectionSettings + if targetColl.Settings != "" { + _ = json.Unmarshal([]byte(targetColl.Settings), &destSettings) + } + guestFull, guestGranted, gerr := s.guestResourceFilter(r, workspaceID) + if gerr != nil { + return nil, &bulkOpError{message: gerr.Error()} + } + mgctx := openChildrenGuardContext{ + r: r, + workspaceID: workspaceID, + itemID: item.ID, + parentSchema: targetSchema, + parentSettings: destSettings, + newFieldMap: result.Fields, + visibleCollectionIDs: visibleIDs, + guestFullCollIDs: guestFull, + guestGrantedItemIDs: guestGranted, + } + precheck = func(tx *sql.Tx, existing *models.Item) error { + txCtx := mgctx + txCtx.currentFieldsJS = existing.Fields + details, derr := s.runOpenChildrenGuard(tx, txCtx) + if derr != nil { + return derr + } + if details != nil { + return &openChildrenGuardError{details: details} + } + return nil + } + } + + moved, err := s.store.MoveItemWithPreCheck(item.ID, targetColl.ID, string(fieldsJSON), precheck) + if err != nil { + if details, ok := asOpenChildrenGuardError(err); ok { + raw, _ := json.Marshal(details) + return nil, &bulkOpError{ + message: "cannot mark item terminal while it has open children", + code: "open_children", + details: raw, + } + } + return nil, &bulkOpError{message: err.Error()} + } + return moved, nil +} + +// publishBulkItemsEvent emits one ItemsBulkUpdated SSE event for the +// slice of a batch mutation that landed in `collection` (TASK-1668). +// Collection is set so the SSE visibility filter routes it like any +// collection-scoped event; the payload carries the verb, the +// per-collection count, and the max seq as the reconcile cursor — but +// no per-item IDs (see events.Event doc for why). +func (s *Server) publishBulkItemsEvent(workspaceID, op, collection string, count int, actor, actorName, source string, maxSeq int64) { + if s.events == nil { + return + } + s.events.Publish(events.Event{ + Type: events.ItemsBulkUpdated, + WorkspaceID: workspaceID, + Collection: collection, + Op: op, + Count: count, + Actor: actor, + ActorName: actorName, + Source: source, + Seq: maxSeq, + }) +} diff --git a/internal/server/handlers_items_bulk_test.go b/internal/server/handlers_items_bulk_test.go new file mode 100644 index 00000000..fdd88167 --- /dev/null +++ b/internal/server/handlers_items_bulk_test.go @@ -0,0 +1,464 @@ +package server + +import ( + "encoding/json" + "net/http" + "strings" + "testing" + "time" + + "github.com/PerpetualSoftware/pad/internal/events" + "github.com/PerpetualSoftware/pad/internal/models" +) + +// createBulkTestItem creates a task and returns it. +func createBulkTestItem(t *testing.T, srv *Server, ws, title, fields string) models.Item { + t.Helper() + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/collections/tasks/items", map[string]interface{}{ + "title": title, + "fields": fields, + }) + if rr.Code != http.StatusCreated { + t.Fatalf("create item %q: expected 201, got %d: %s", title, rr.Code, rr.Body.String()) + } + var item models.Item + parseJSON(t, rr, &item) + return item +} + +func itemFields(t *testing.T, srv *Server, ws, slug string) map[string]any { + t.Helper() + rr := doRequest(srv, "GET", "/api/v1/workspaces/"+ws+"/items/"+slug, nil) + if rr.Code != http.StatusOK { + t.Fatalf("get item %s: %d: %s", slug, rr.Code, rr.Body.String()) + } + var it models.Item + parseJSON(t, rr, &it) + f := map[string]any{} + _ = json.Unmarshal([]byte(it.Fields), &f) + return f +} + +func TestBulkItems_SetPriority(t *testing.T) { + srv := testServer(t) + ws := createWSWithCollections(t, srv) + + a := createBulkTestItem(t, srv, ws, "A", `{"status":"open","priority":"low"}`) + b := createBulkTestItem(t, srv, ws, "B", `{"status":"open","priority":"low"}`) + + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{a.Ref, b.Ref}, + "op": "set-priority", + "priority": "high", + }) + if rr.Code != http.StatusOK { + t.Fatalf("bulk set-priority: expected 200, got %d: %s", rr.Code, rr.Body.String()) + } + var resp bulkItemsResponse + parseJSON(t, rr, &resp) + if len(resp.Updated) != 2 || len(resp.Failed) != 0 || resp.Total != 2 { + t.Fatalf("expected 2 updated / 0 failed, got %+v", resp) + } + for _, it := range []models.Item{a, b} { + if got := itemFields(t, srv, ws, it.Slug)["priority"]; got != "high" { + t.Errorf("%s priority: expected high, got %v", it.Ref, got) + } + } +} + +func TestBulkItems_MoveStatus(t *testing.T) { + srv := testServer(t) + ws := createWSWithCollections(t, srv) + + a := createBulkTestItem(t, srv, ws, "A", `{"status":"open"}`) + b := createBulkTestItem(t, srv, ws, "B", `{"status":"open"}`) + + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{a.Ref, b.Ref}, + "op": "move", + "status": "in-progress", + }) + if rr.Code != http.StatusOK { + t.Fatalf("bulk move: expected 200, got %d: %s", rr.Code, rr.Body.String()) + } + var resp bulkItemsResponse + parseJSON(t, rr, &resp) + if len(resp.Updated) != 2 { + t.Fatalf("expected 2 updated, got %+v", resp) + } + if got := itemFields(t, srv, ws, a.Slug)["status"]; got != "in-progress" { + t.Errorf("status: expected in-progress, got %v", got) + } +} + +func TestBulkItems_TagAndUntag(t *testing.T) { + srv := testServer(t) + ws := createWSWithCollections(t, srv) + + a := createBulkTestItem(t, srv, ws, "A", `{"status":"open"}`) + + // Tag + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{a.Ref}, + "op": "tag", + "tags": []string{"urgent", "frontend"}, + }) + if rr.Code != http.StatusOK { + t.Fatalf("bulk tag: expected 200, got %d: %s", rr.Code, rr.Body.String()) + } + rr = doRequest(srv, "GET", "/api/v1/workspaces/"+ws+"/items/"+a.Slug, nil) + var it models.Item + parseJSON(t, rr, &it) + var tags []string + _ = json.Unmarshal([]byte(it.Tags), &tags) + if len(tags) != 2 { + t.Fatalf("expected 2 tags after tag, got %v", tags) + } + + // Re-tagging the same tag is idempotent (no duplicates). + doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{a.Ref}, "op": "tag", "tags": []string{"urgent"}, + }) + rr = doRequest(srv, "GET", "/api/v1/workspaces/"+ws+"/items/"+a.Slug, nil) + parseJSON(t, rr, &it) + _ = json.Unmarshal([]byte(it.Tags), &tags) + if len(tags) != 2 { + t.Fatalf("expected tags to stay 2 after duplicate tag, got %v", tags) + } + + // Untag one + rr = doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{a.Ref}, "op": "untag", "tags": []string{"urgent"}, + }) + if rr.Code != http.StatusOK { + t.Fatalf("bulk untag: expected 200, got %d: %s", rr.Code, rr.Body.String()) + } + rr = doRequest(srv, "GET", "/api/v1/workspaces/"+ws+"/items/"+a.Slug, nil) + parseJSON(t, rr, &it) + _ = json.Unmarshal([]byte(it.Tags), &tags) + if len(tags) != 1 || tags[0] != "frontend" { + t.Fatalf("expected [frontend] after untag, got %v", tags) + } +} + +func TestBulkItems_Archive(t *testing.T) { + srv := testServer(t) + ws := createWSWithCollections(t, srv) + + a := createBulkTestItem(t, srv, ws, "A", `{"status":"open"}`) + b := createBulkTestItem(t, srv, ws, "B", `{"status":"open"}`) + + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{a.Ref, b.Ref}, "op": "archive", + }) + if rr.Code != http.StatusOK { + t.Fatalf("bulk archive: expected 200, got %d: %s", rr.Code, rr.Body.String()) + } + var resp bulkItemsResponse + parseJSON(t, rr, &resp) + if len(resp.Updated) != 2 { + t.Fatalf("expected 2 archived, got %+v", resp) + } + + rr = doRequest(srv, "GET", "/api/v1/workspaces/"+ws+"/items", nil) + var items []models.Item + parseJSON(t, rr, &items) + if len(items) != 0 { + t.Errorf("expected 0 live items after bulk archive, got %d", len(items)) + } +} + +func TestBulkItems_PartialFailure(t *testing.T) { + srv := testServer(t) + ws := createWSWithCollections(t, srv) + + a := createBulkTestItem(t, srv, ws, "A", `{"status":"open"}`) + + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{a.Ref, "TASK-9999"}, + "op": "set-priority", + "priority": "high", + }) + if rr.Code != http.StatusOK { + t.Fatalf("expected 200 with partial failures, got %d: %s", rr.Code, rr.Body.String()) + } + var resp bulkItemsResponse + parseJSON(t, rr, &resp) + if len(resp.Updated) != 1 { + t.Errorf("expected 1 updated, got %d", len(resp.Updated)) + } + if len(resp.Failed) != 1 { + t.Errorf("expected 1 failed, got %d", len(resp.Failed)) + } +} + +func TestBulkItems_Validation(t *testing.T) { + srv := testServer(t) + ws := createWSWithCollections(t, srv) + + cases := []struct { + name string + body map[string]any + }{ + {"empty ids", map[string]any{"ids": []string{}, "op": "archive"}}, + {"unknown op", map[string]any{"ids": []string{"TASK-1"}, "op": "frobnicate"}}, + {"move without params", map[string]any{"ids": []string{"TASK-1"}, "op": "move"}}, + {"set-priority without priority", map[string]any{"ids": []string{"TASK-1"}, "op": "set-priority"}}, + {"tag without tags", map[string]any{"ids": []string{"TASK-1"}, "op": "tag"}}, + {"assign without target", map[string]any{"ids": []string{"TASK-1"}, "op": "assign"}}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", tc.body) + if rr.Code != http.StatusBadRequest { + t.Errorf("expected 400, got %d: %s", rr.Code, rr.Body.String()) + } + }) + } +} + +// TestBulkItems_StatusMoveRunsOpenChildrenGuard confirms a bulk +// status move to a terminal value is rejected (per-row) while the +// item still has open children — same guard the single PATCH path runs. +func TestBulkItems_StatusMoveRunsOpenChildrenGuard(t *testing.T) { + srv := testServer(t) + ws := createWSWithCollections(t, srv) + + plan, _ := seedParentAndChildren(t, srv, ws, []string{"open"}) + + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{plan.Ref}, "op": "move", "status": "completed", + }) + if rr.Code != http.StatusOK { + t.Fatalf("expected 200 envelope, got %d: %s", rr.Code, rr.Body.String()) + } + var resp bulkItemsResponse + parseJSON(t, rr, &resp) + if len(resp.Updated) != 0 || len(resp.Failed) != 1 { + t.Fatalf("expected 0 updated / 1 failed, got %+v", resp) + } + if resp.Failed[0].Code != "open_children" { + t.Errorf("expected open_children code, got %q (%s)", resp.Failed[0].Code, resp.Failed[0].Error) + } + + // force=true escapes the guard. + rr = doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{plan.Ref}, "op": "move", "status": "completed", "force": true, + }) + parseJSON(t, rr, &resp) + if len(resp.Updated) != 1 { + t.Fatalf("force should bypass guard: %+v", resp) + } +} + +// TestBulkItems_CollectionMoveRunsOpenChildrenGuard confirms a bulk +// collection move that also sets a terminal status runs the guard +// against the destination schema (Codex round-1 finding). +func TestBulkItems_CollectionMoveRunsOpenChildrenGuard(t *testing.T) { + srv := testServer(t) + ws := createWSWithCollections(t, srv) + + collResp := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/collections", map[string]interface{}{ + "name": "Programs", + "icon": "package", + "schema": `{"fields":[{"key":"status","label":"Status","type":"select","options":["active","completed"],"terminal_options":["completed"]}]}`, + }) + if collResp.Code != http.StatusCreated { + t.Fatalf("create programs: %d %s", collResp.Code, collResp.Body.String()) + } + + plan, _ := seedParentAndChildren(t, srv, ws, []string{"open"}) + + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{plan.Ref}, "op": "move", "collection": "programs", "status": "completed", + }) + if rr.Code != http.StatusOK { + t.Fatalf("expected 200 envelope, got %d: %s", rr.Code, rr.Body.String()) + } + var resp bulkItemsResponse + parseJSON(t, rr, &resp) + if len(resp.Failed) != 1 || resp.Failed[0].Code != "open_children" { + t.Fatalf("expected open_children failure, got %+v", resp) + } + + // The plan must NOT have moved. + rr = doRequest(srv, "GET", "/api/v1/workspaces/"+ws+"/items/"+plan.Ref, nil) + var fresh models.Item + parseJSON(t, rr, &fresh) + if fresh.CollectionSlug != "plans" { + t.Errorf("plan moved despite guard rejection — now in %q", fresh.CollectionSlug) + } +} + +// TestBulkItems_EmitsCollectionScopedBatchEvent asserts the batch SSE +// event is scoped to its collection (so the SSE visibility filter routes +// it correctly) and carries NO per-item IDs (no leak on a broadcast bus). +func TestBulkItems_EmitsCollectionScopedBatchEvent(t *testing.T) { + srv := testServerWithEvents(t) + ws := createWSWithCollections(t, srv) + wsRow, err := srv.store.GetWorkspaceBySlug(ws) + if err != nil || wsRow == nil { + t.Fatalf("resolve workspace: %v", err) + } + ch := srv.events.Subscribe(wsRow.ID) + defer srv.events.Unsubscribe(ch) + + a := createBulkTestItem(t, srv, ws, "A", `{"status":"open","priority":"low"}`) + b := createBulkTestItem(t, srv, ws, "B", `{"status":"open","priority":"low"}`) + + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{a.Ref, b.Ref}, "op": "set-priority", "priority": "high", + }) + if rr.Code != http.StatusOK { + t.Fatalf("bulk: %d: %s", rr.Code, rr.Body.String()) + } + + var bulk *events.Event + deadline := time.After(2 * time.Second) +loop: + for { + select { + case ev := <-ch: + if ev.Type == events.ItemsBulkUpdated { + e := ev + bulk = &e + break loop + } + case <-deadline: + break loop + } + } + if bulk == nil { + t.Fatal("no items_bulk_updated event published") + } + if bulk.Collection != "tasks" { + t.Errorf("expected Collection=tasks, got %q", bulk.Collection) + } + if bulk.Count != 2 { + t.Errorf("expected Count=2, got %d", bulk.Count) + } + if bulk.Op != "set-priority" { + t.Errorf("expected Op=set-priority, got %q", bulk.Op) + } + // The wire payload must not leak per-item IDs. + raw, _ := json.Marshal(bulk) + if strings.Contains(string(raw), "item_ids") { + t.Errorf("batch SSE event must not carry item_ids: %s", raw) + } +} + +// TestBulkItems_CollectionMoveNotifiesBothScopes asserts a bulk +// collection move emits a batch event for BOTH the source and target +// collections, so a member watching either lane reconciles. +func TestBulkItems_CollectionMoveNotifiesBothScopes(t *testing.T) { + srv := testServerWithEvents(t) + ws := createWSWithCollections(t, srv) + wsRow, err := srv.store.GetWorkspaceBySlug(ws) + if err != nil || wsRow == nil { + t.Fatalf("resolve workspace: %v", err) + } + + collResp := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/collections", map[string]interface{}{ + "name": "Programs", + "icon": "package", + "schema": `{"fields":[{"key":"status","label":"Status","type":"select","options":["active","completed"]}]}`, + }) + if collResp.Code != http.StatusCreated { + t.Fatalf("create programs: %d %s", collResp.Code, collResp.Body.String()) + } + + a := createBulkTestItem(t, srv, ws, "A", `{"status":"open"}`) + + ch := srv.events.Subscribe(wsRow.ID) + defer srv.events.Unsubscribe(ch) + + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{a.Ref}, "op": "move", "collection": "programs", + }) + if rr.Code != http.StatusOK { + t.Fatalf("bulk move: %d: %s", rr.Code, rr.Body.String()) + } + + gotScopes := map[string]bool{} + deadline := time.After(2 * time.Second) +collect: + for { + select { + case ev := <-ch: + if ev.Type == events.ItemsBulkUpdated { + gotScopes[ev.Collection] = true + if gotScopes["tasks"] && gotScopes["programs"] { + break collect + } + } + case <-deadline: + break collect + } + } + if !gotScopes["tasks"] { + t.Error("expected a batch event for source collection 'tasks'") + } + if !gotScopes["programs"] { + t.Error("expected a batch event for target collection 'programs'") + } +} + +// TestBulkItems_CollectionMoveValidatesStatusOverride confirms a status +// override on a collection move is validated against the target schema — +// an out-of-options value is rejected (per-row), not written. +func TestBulkItems_CollectionMoveValidatesStatusOverride(t *testing.T) { + srv := testServer(t) + ws := createWSWithCollections(t, srv) + + collResp := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/collections", map[string]interface{}{ + "name": "Programs", + "icon": "package", + "schema": `{"fields":[{"key":"status","label":"Status","type":"select","options":["active","completed"]}]}`, + }) + if collResp.Code != http.StatusCreated { + t.Fatalf("create programs: %d %s", collResp.Code, collResp.Body.String()) + } + + a := createBulkTestItem(t, srv, ws, "A", `{"status":"open"}`) + + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{a.Ref}, "op": "move", "collection": "programs", "status": "bogus", + }) + if rr.Code != http.StatusOK { + t.Fatalf("expected 200 envelope, got %d: %s", rr.Code, rr.Body.String()) + } + var resp bulkItemsResponse + parseJSON(t, rr, &resp) + if len(resp.Updated) != 0 || len(resp.Failed) != 1 { + t.Fatalf("expected 0 updated / 1 failed for invalid status, got %+v", resp) + } + if resp.Failed[0].Code != "validation_error" { + t.Errorf("expected validation_error, got %q (%s)", resp.Failed[0].Code, resp.Failed[0].Error) + } + + // The item must NOT have moved. + rr = doRequest(srv, "GET", "/api/v1/workspaces/"+ws+"/items/"+a.Slug, nil) + var fresh models.Item + parseJSON(t, rr, &fresh) + if fresh.CollectionSlug != "tasks" { + t.Errorf("item moved despite invalid status — now in %q", fresh.CollectionSlug) + } +} + +// TestBulkItems_RouteDoesNotShadowItemSlug guards the route-ordering +// fix: /items/bulk is a static segment registered before the +// /items/{itemSlug} param route, so it must not be treated as an item +// slug, and a GET to it (no handler) must not resolve as an item. +func TestBulkItems_RouteRegistered(t *testing.T) { + srv := testServer(t) + ws := createWSWithCollections(t, srv) + a := createBulkTestItem(t, srv, ws, "A", `{"status":"open"}`) + + // POST hits the bulk handler (200), not a 404 / item-slug path. + rr := doRequest(srv, "POST", "/api/v1/workspaces/"+ws+"/items/bulk", map[string]any{ + "ids": []string{a.Ref}, "op": "set-priority", "priority": "high", + }) + if rr.Code != http.StatusOK { + t.Fatalf("bulk route: expected 200, got %d: %s", rr.Code, rr.Body.String()) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 4ed3bb59..982fa072 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -1113,6 +1113,10 @@ func (s *Server) setupRouter() { // Items (cross-collection, v2) r.Get("/items", s.handleListItems) + // Bulk mutation (TASK-1668). Static segment must be + // registered before the /items/{itemSlug} param route + // so "bulk" isn't captured as an item slug. + r.Post("/items/bulk", s.handleBulkItems) r.Route("/items/{itemSlug}", func(r chi.Router) { r.Get("/", s.handleGetItem) r.Patch("/", s.handleUpdateItem) diff --git a/web/src/lib/services/sse.svelte.ts b/web/src/lib/services/sse.svelte.ts index 831cc4ec..87fabeac 100644 --- a/web/src/lib/services/sse.svelte.ts +++ b/web/src/lib/services/sse.svelte.ts @@ -182,6 +182,18 @@ function createSSEService() { broadcast({ type: 'sync_required' }); }); + // Bulk mutations (TASK-1668) emit ONE `items_bulk_updated` event + // for the whole batch instead of per-item item_updated/archived + // events. It carries item_ids + a max seq but no per-item field + // payload, so rather than apply N in-place deltas we route it + // through the same incremental backfill the gap path uses — a + // /items-changes delta reconciles every affected row by seq. + // Broadcast so peer tabs reconcile too. + eventSource.addEventListener('items_bulk_updated', () => { + dispatchSyncRequired(); + broadcast({ type: 'sync_required' }); + }); + // Handle unauthorized: the server's periodic membership revalidation // detected that this session has lost access to the workspace. We // must close the EventSource ourselves — otherwise the browser