Make inbox bulk ops report partial progress and surface walk failures (CL-7207)

packages/inbox/package.json

@@ -14,6 +14,7 @@
     "test": "bun test"
   },
   "dependencies": {
+    "@corbits/error-sink": "workspace:*",
     "@corbits/mailbox": "github:corbitsdev/corbits-mailbox#caa5214a2811a66a6f3dd3b7631b5f53591c8c14",
     "@corbits/notify": "workspace:*",
     "@intx/hub-api": "workspace:*",

packages/inbox/src/bulk.ts

@@ -22,3 +22,39 @@ export function itemsEligibleForClearDone(
 ): InboxItem[] {
   return items.filter((item) => item.status === "done");
 }
+
+export interface BulkOperationResult {
+  readonly succeeded: number;
+  readonly failed: number;
+}
+
+/**
+ * Apply `apply` to every item, one at a time, never letting one item's
+ * failure abort the rest (CL-7207). Previously `mark-all-read` and
+ * `clear-done` ran their per-item writes in a plain loop with no
+ * per-item try/catch: a throw on item N left the handler throwing out of
+ * the whole request — a 500 with items before N already mutated, item N
+ * left in a new inconsistent state, and everything after N untouched, with
+ * no way for the caller to tell how far it got. Catching per item instead
+ * means a transient failure on one row costs that one row, not the rest of
+ * the inbox, and the caller gets back exactly how many succeeded and how
+ * many didn't rather than an opaque 500.
+ */
+export async function runBulkOperation<T>(
+  items: readonly T[],
+  apply: (item: T) => Promise<void>,
+  onError?: (item: T, error: unknown) => void,
+): Promise<BulkOperationResult> {
+  let succeeded = 0;
+  let failed = 0;
+  for (const item of items) {
+    try {
+      await apply(item);
+      succeeded += 1;
+    } catch (error) {
+      failed += 1;
+      onError?.(item, error);
+    }
+  }
+  return { succeeded, failed };
+}

packages/inbox/src/routes.ts

@@ -17,20 +17,26 @@ import {
   type MailboxListCursor,
 } from "@corbits/mailbox";
 
+import { reportError } from "@corbits/error-sink";
 import type { TenantEnv } from "@intx/hub-api";
 import { getLogger } from "@intx/log";
-import { Hono } from "hono";
+import { Hono, type Context } from "hono";
 
+import {
+  itemsEligibleForClearDone,
+  itemsEligibleForMarkAllRead,
+  runBulkOperation,
+} from "./bulk";
 import { cursorScopeMismatch } from "./cursor";
 import { isInboxGroup, type InboxGroup } from "./group";
-import { itemsEligibleForClearDone, itemsEligibleForMarkAllRead } from "./bulk";
 import {
   projectInboxItem,
   projectInboxItemDetail,
   type InboxCounts,
   type InboxItem,
 } from "./project";
 import { WORKBENCH_INBOX_PRIORITIES } from "./vocabulary";
+import { walkAllOpen } from "./walk";
 
 // Page size for bulk product ops (mark-all-read, clear-done, counts). Large
 // enough that a normal inbox finishes in one round-trip; anything past it
@@ -73,9 +79,7 @@ async function listAllOpen(
   scope: { tenantId: string; principalId: string },
   filter?: MailboxFilter,
 ): Promise<InboxItem[]> {
-  const out: InboxItem[] = [];
-  let cursor: MailboxListCursor | undefined;
-  for (;;) {
+  return walkAllOpen((page) => {
     const listOpts = {
       tenantId: scope.tenantId,
       principalId: scope.principalId,
@@ -85,23 +89,16 @@ async function listAllOpen(
       priorities: WORKBENCH_INBOX_PRIORITIES,
     };
     const listOptsWithCursor =
-      cursor !== undefined ? { ...listOpts, cursor } : listOpts;
-    const page = await listUserMailbox(
+      page.cursor !== undefined
+        ? { ...listOpts, cursor: page.cursor }
+        : listOpts;
+    return listUserMailbox(
       db,
       filter !== undefined
         ? { ...listOptsWithCursor, filter }
         : listOptsWithCursor,
     );
-    for (const message of page.items) {
-      out.push(projectInboxItem(message));
-    }
-    if (page.nextCursor === undefined) break;
-    const next = decodeMailboxListCursor(page.nextCursor);
-    if (next === null) break;
-    cursor = next;
-  }
-
-  return out;
+  });
 }
 
 /**
@@ -115,14 +112,35 @@ export function createInboxRoutes(
   const { db, bus } = deps;
   const app = new Hono<TenantEnv>();
 
+  // A failed inbox walk (an undecodable cursor mid-walk, a cursor that
+  // never advances, or a pathologically large inbox — see walk.ts) must be
+  // loud, not a silently truncated count or bulk op. Reported through
+  // @corbits/error-sink and answered as a 500 with the refId a person can
+  // quote back, rather than a falsely-complete 200 (CL-7207).
+  async function listAllOpenOrReport(
+    c: Context<TenantEnv>,
+    tenantId: string,
+    operation: string,
+    scope: { tenantId: string; principalId: string },
+    filter?: MailboxFilter,
+  ): Promise<InboxItem[] | Response> {
+    try {
+      return await listAllOpen(db, scope, filter);
+    } catch (cause) {
+      const refId = reportError(cause, { operation, tenantId });
+      return c.json({ error: "could not list the inbox", refId }, 500);
+    }
+  }
+
   // Static path segments first so `/:id` never captures them.
   app.get("/counts", async (c) => {
     const tenant = c.get("tenant");
     const principal = c.get("principal");
-    const items = await listAllOpen(db, {
+    const items = await listAllOpenOrReport(c, tenant.id, "inbox_counts_walk", {
       tenantId: tenant.id,
       principalId: principal.id,
     });
+    if (items instanceof Response) return items;
     const counts: InboxCounts = {
       action: 0,
       mention: 0,
@@ -141,35 +159,75 @@ export function createInboxRoutes(
     const tenant = c.get("tenant");
     const principal = c.get("principal");
     const scope = { tenantId: tenant.id, principalId: principal.id };
-    const items = await listAllOpen(db, scope);
-    let marked = 0;
-    for (const item of itemsEligibleForMarkAllRead(items)) {
-      await enrichMailboxMessage(
-        db,
-        { ...scope, id: item.id },
-        { status: "done" },
-      );
-      await markMailboxMessageRead(db, { ...scope, id: item.id });
-      publish(bus, scope, item.id, "mark_read");
-      marked += 1;
-    }
-    return c.json({ marked });
+    const items = await listAllOpenOrReport(
+      c,
+      tenant.id,
+      "inbox_mark_all_read_walk",
+      scope,
+    );
+    if (items instanceof Response) return items;
+    const { succeeded: marked, failed } = await runBulkOperation(
+      itemsEligibleForMarkAllRead(items),
+      async (item) => {
+        // Atomic: a throw between the status flip and the read flip must
+        // never leave a row done-but-unread (CL-7207).
+        await db.transaction(async (tx) => {
+          await enrichMailboxMessage(
+            tx,
+            { ...scope, id: item.id },
+            { status: "done" },
+          );
+          await markMailboxMessageRead(tx, { ...scope, id: item.id });
+        });
+        publish(bus, scope, item.id, "mark_read");
+      },
+      (item, error) =>
+        reportError(error, {
+          operation: "inbox_mark_all_read_item",
+          tenantId: tenant.id,
+          extra: { id: item.id },
+        }),
+    );
+    // A 200 must mean "every eligible item was marked" — a partial result
+    // is reported as 207 so a caller that only checks the status code (not
+    // the body) can't mistake "half the inbox" for "success" (CL-7207).
+    return c.json(
+      { marked, failed, complete: failed === 0 },
+      failed === 0 ? 200 : 207,
+    );
   });
 
   app.post("/clear-done", async (c) => {
     const tenant = c.get("tenant");
     const principal = c.get("principal");
     const scope = { tenantId: tenant.id, principalId: principal.id };
-    const items = await listAllOpen(db, scope);
-    let cleared = 0;
-    for (const item of itemsEligibleForClearDone(items)) {
-      const ok = await trashMailboxMessage(db, { ...scope, id: item.id });
-      if (ok) {
+    const items = await listAllOpenOrReport(
+      c,
+      tenant.id,
+      "inbox_clear_done_walk",
+      scope,
+    );
+    if (items instanceof Response) return items;
+    const { succeeded: cleared, failed } = await runBulkOperation(
+      itemsEligibleForClearDone(items),
+      async (item) => {
+        const ok = await trashMailboxMessage(db, { ...scope, id: item.id });
+        if (!ok) throw new Error(`message ${item.id} not found to trash`);
         publish(bus, scope, item.id, "trash");
-        cleared += 1;
-      }
-    }
-    return c.json({ cleared });
+      },
+      (item, error) =>
+        reportError(error, {
+          operation: "inbox_clear_done_item",
+          tenantId: tenant.id,
+          extra: { id: item.id },
+        }),
+    );
+    // Same partial-vs-complete signal as mark-all-read: 207 whenever any
+    // item failed, so a status-code-only caller can't read it as success.
+    return c.json(
+      { cleared, failed, complete: failed === 0 },
+      failed === 0 ? 200 : 207,
+    );
   });
 
   app.get("/", async (c) => {

packages/inbox/src/walk.ts

@@ -0,0 +1,71 @@
+// The bulk product ops (`/counts`, `mark-all-read`, `clear-done`) all walk
+// the caller's entire open inbox before acting. `walkAllOpen` is that walk,
+// pulled out from `routes.ts` and driven through an injected `listPage` so
+// its failure modes — an undecodable cursor, a cursor that never advances,
+// an inbox large enough to need an explicit cap — are unit-testable without
+// a live Postgres (CL-7207).
+//
+// Previously this loop treated `decodeMailboxListCursor` returning `null`
+// as "done" and silently `break`-ed: a rolling deploy that lands two hub
+// versions against `@corbits/mailbox` mid-rollout could change a cursor's
+// shape under a long-running walk, and the caller would report success
+// having silently stopped partway through the inbox. Throwing instead
+// turns that into a loud, reported failure — see routes.ts's callers,
+// which report it through `@corbits/error-sink` and answer 500 rather than
+// a falsely-complete 200.
+import {
+  decodeMailboxListCursor,
+  type MailboxListCursor,
+  type MailboxPage,
+} from "@corbits/mailbox";
+
+import { projectInboxItem } from "./project";
+import type { InboxItem } from "./project";
+
+// 1000 pages * 100 rows/page = 100k rows — generous for any real inbox,
+// tight enough that a cursor bug loops a bounded number of times instead
+// of forever.
+export const MAX_WALK_PAGES = 1000;
+
+export type ListPageFn = (opts: {
+  cursor?: MailboxListCursor;
+}) => Promise<MailboxPage>;
+
+/** The walk stopped before covering the whole inbox. `message` says why. */
+export class IncompleteWalkError extends Error {}
+
+export async function walkAllOpen(
+  listPage: ListPageFn,
+  opts: { maxPages?: number } = {},
+): Promise<InboxItem[]> {
+  const maxPages = opts.maxPages ?? MAX_WALK_PAGES;
+  const out: InboxItem[] = [];
+  let cursor: MailboxListCursor | undefined;
+  let pages = 0;
+  for (;;) {
+    pages += 1;
+    if (pages > maxPages) {
+      throw new IncompleteWalkError(
+        `inbox walk exceeded ${maxPages} pages without finishing`,
+      );
+    }
+    const page = await listPage(cursor !== undefined ? { cursor } : {});
+    for (const message of page.items) out.push(projectInboxItem(message));
+    if (page.nextCursor === undefined) break;
+    const next = decodeMailboxListCursor(page.nextCursor);
+    if (next === null) {
+      throw new IncompleteWalkError(
+        "inbox walk received an undecodable cursor mid-walk",
+      );
+    }
+    if (
+      cursor !== undefined &&
+      next.createdAt === cursor.createdAt &&
+      next.id === cursor.id
+    ) {
+      throw new IncompleteWalkError("inbox walk cursor did not advance");
+    }
+    cursor = next;
+  }
+  return out;
+}