From 08fe60e3b5e532a0b74ba76c6be451b7cbe083ec Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 05:58:53 -0700 Subject: [PATCH 1/2] Add inbox walk-guard and bulk-operation partial-progress primitives New, DB-free pure modules for CL-7207, unit-tested without a live Postgres (bulk.ts already documents this as the package's convention for testing product rules against the mailbox FKs): - walk.ts: walkAllOpen() extracts the paging walk `/counts`, mark-all-read, and clear-done all share, driven through an injected listPage so its failure modes are directly testable: an undecodable nextCursor, a cursor that never advances, and a page cap (MAX_WALK_PAGES) now throw IncompleteWalkError instead of looping forever or silently breaking. - bulk.ts: runBulkOperation() applies a per-item action without letting one item's failure abort the rest, returning {succeeded, failed} instead of throwing out of the whole loop. Committed together with their tests, rather than as a red-then-green pair: both are net-new extractions with no prior standalone behavior to reproduce as a failing test, and typechecking one half without the other isn't meaningful. The next commit wires both into routes.ts, which is where CL-7207's actual user-visible behavior changes. --- packages/inbox/src/bulk.ts | 36 ++++++++++++ packages/inbox/src/walk.ts | 71 +++++++++++++++++++++++ packages/inbox/test/bulk.test.ts | 38 ++++++++++++ packages/inbox/test/walk.test.ts | 99 ++++++++++++++++++++++++++++++++ 4 files changed, 244 insertions(+) create mode 100644 packages/inbox/src/walk.ts create mode 100644 packages/inbox/test/walk.test.ts diff --git a/packages/inbox/src/bulk.ts b/packages/inbox/src/bulk.ts index c4991dd5a..d840afff1 100644 --- a/packages/inbox/src/bulk.ts +++ b/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( + items: readonly T[], + apply: (item: T) => Promise, + onError?: (item: T, error: unknown) => void, +): Promise { + 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 }; +} diff --git a/packages/inbox/src/walk.ts b/packages/inbox/src/walk.ts new file mode 100644 index 000000000..60dba9896 --- /dev/null +++ b/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; + +/** 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 { + 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; +} diff --git a/packages/inbox/test/bulk.test.ts b/packages/inbox/test/bulk.test.ts index 67214ff8f..aa9ecf5fe 100644 --- a/packages/inbox/test/bulk.test.ts +++ b/packages/inbox/test/bulk.test.ts @@ -3,6 +3,7 @@ import { describe, expect, test } from "bun:test"; import { itemsEligibleForClearDone, itemsEligibleForMarkAllRead, + runBulkOperation, } from "../src/bulk"; import type { InboxItem } from "../src/project"; @@ -41,3 +42,40 @@ describe("itemsEligibleForClearDone", () => { expect(itemsEligibleForClearDone(items).map((i) => i.id)).toEqual(["m"]); }); }); + +describe("runBulkOperation", () => { + test("a thrown error on one item does not stop the rest", async () => { + const applied: string[] = []; + const failures: { id: string; error: unknown }[] = []; + const result = await runBulkOperation( + ["a", "b", "c"], + async (id) => { + if (id === "b") throw new Error("transient write failure"); + applied.push(id); + }, + (id, error) => failures.push({ id, error }), + ); + + // Both non-failing items still ran, despite "b" throwing between them — + // the caller can tell exactly how far the operation got. + expect(applied).toEqual(["a", "c"]); + expect(result).toEqual({ succeeded: 2, failed: 1 }); + expect(failures).toHaveLength(1); + expect(failures[0]?.id).toBe("b"); + expect((failures[0]?.error as Error).message).toBe( + "transient write failure", + ); + }); + + test("every item succeeding reports zero failures", async () => { + const result = await runBulkOperation(["a", "b"], async () => {}); + expect(result).toEqual({ succeeded: 2, failed: 0 }); + }); + + test("onError is optional", async () => { + const result = await runBulkOperation(["a"], async () => { + throw new Error("boom"); + }); + expect(result).toEqual({ succeeded: 0, failed: 1 }); + }); +}); diff --git a/packages/inbox/test/walk.test.ts b/packages/inbox/test/walk.test.ts new file mode 100644 index 000000000..7587397e9 --- /dev/null +++ b/packages/inbox/test/walk.test.ts @@ -0,0 +1,99 @@ +import { describe, expect, test } from "bun:test"; +import type { MailboxPage } from "@corbits/mailbox"; +import { encodeMailboxListCursor } from "@corbits/mailbox"; +import { IncompleteWalkError, walkAllOpen } from "../src/walk"; + +function message(id: string, createdAt: string) { + return { + id, + messageId: `${id}@test`, + from: "routine:test", + to: ["prn_1@inbox.test"], + date: createdAt, + read: false, + status: "open" as const, + createdAt, + }; +} + +const SHAPE = { view: "all" as const, sort: "date" as const, filter: "" }; + +describe("walkAllOpen", () => { + test("walks every page and stops once nextCursor is undefined", async () => { + let calls = 0; + const page: MailboxPage = { + items: [message("msg_1", "2026-01-02T00:00:00.000000Z")], + }; + const items = await walkAllOpen(async () => { + calls += 1; + return page; + }); + expect(items.map((item) => item.id)).toEqual(["msg_1"]); + expect(calls).toBe(1); + }); + + test("follows nextCursor across multiple pages", async () => { + const cursorFor = (id: string, createdAt: string) => + encodeMailboxListCursor({ createdAt, id }, SHAPE); + const pages: MailboxPage[] = [ + { + items: [message("msg_1", "2026-01-03T00:00:00.000000Z")], + nextCursor: cursorFor("msg_1", "2026-01-03T00:00:00.000000Z"), + }, + { items: [message("msg_2", "2026-01-02T00:00:00.000000Z")] }, + ]; + let call = 0; + const items = await walkAllOpen(async () => { + const page = pages[call]; + call += 1; + if (page === undefined) throw new Error("listPage called too many times"); + return page; + }); + expect(items.map((item) => item.id)).toEqual(["msg_1", "msg_2"]); + expect(call).toBe(2); + }); + + test("throws IncompleteWalkError when the next cursor is undecodable", async () => { + const page: MailboxPage = { + items: [message("msg_1", "2026-01-01T00:00:00.000000Z")], + nextCursor: "not-a-valid-cursor", + }; + await expect(walkAllOpen(async () => page)).rejects.toBeInstanceOf( + IncompleteWalkError, + ); + }); + + test("throws IncompleteWalkError when the cursor does not advance", async () => { + const stuckCursor = encodeMailboxListCursor( + { createdAt: "2026-01-01T00:00:00.000000Z", id: "msg_1" }, + SHAPE, + ); + const page: MailboxPage = { + items: [message("msg_1", "2026-01-01T00:00:00.000000Z")], + nextCursor: stuckCursor, + }; + // Every call returns the exact same page/cursor — an infinite loop + // without the advance guard. + await expect(walkAllOpen(async () => page)).rejects.toBeInstanceOf( + IncompleteWalkError, + ); + }); + + test("throws IncompleteWalkError once maxPages is exceeded", async () => { + let call = 0; + const listPage = async (): Promise => { + call += 1; + return { + items: [message(`msg_${call}`, `2026-01-01T00:00:0${call}.000000Z`)], + nextCursor: encodeMailboxListCursor( + { createdAt: `2026-01-01T00:00:0${call}.000000Z`, id: `msg_${call}` }, + SHAPE, + ), + }; + }; + await expect(walkAllOpen(listPage, { maxPages: 2 })).rejects.toBeInstanceOf( + IncompleteWalkError, + ); + expect(call).toBe(2); + }); +}); From 2c854d23069543ca65b70cb61122d5c681149307 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 30 Aug 2026 06:04:40 -0700 Subject: [PATCH 2/2] Make inbox bulk ops report partial progress and surface walk failures (CL-7207) mark-all-read, clear-done, and /counts shared a walk (listAllOpen) that silently truncated on an undecodable nextCursor -- rolling deploy skew between two hub versions against @corbits/mailbox could change a cursor's shape mid-walk, and the caller would see success having silently stopped partway through the inbox. mark-all-read and clear-done also mutated one item at a time with no transaction tying each item's writes together and no per-item error isolation: a throw on item N threw the whole handler as a 500, left items before N mutated, item N in a state the UI had never rendered (done but unread), and everything after N untouched, with no way for a retry to resume instead of re-walking and re-mutating everything from scratch. - listAllOpen now delegates to walk.ts's walkAllOpen, and its callers (/counts, mark-all-read, clear-done) report a failed walk through @corbits/error-sink and answer 500 with a refId instead of silently returning wrong data. - mark-all-read's status flip + read flip, and clear-done's trash, run inside db.transaction so a failure between them never leaves a row half-mutated. - Both now use bulk.ts's runBulkOperation so one item's failure doesn't abort the rest: the response is {marked/cleared, failed, complete} rather than a 500 with no signal of how far the operation got. A 200 with `failed: 0` is the only "fully done" shape; any failed item answers 207 instead, so a caller that only checks the status code (not the body) can't mistake a partial run for a complete one. Adds @corbits/error-sink as a dependency of @corbits/inbox. --- bun.lock | 13 ++- packages/inbox/package.json | 1 + packages/inbox/src/routes.ts | 138 ++++++++++++++++++++--------- packages/inbox/test/routes.test.ts | 45 +++++++++- 4 files changed, 148 insertions(+), 49 deletions(-) diff --git a/bun.lock b/bun.lock index 6db572c5c..cc2eb879b 100644 --- a/bun.lock +++ b/bun.lock @@ -236,7 +236,7 @@ }, "packages/agent-directory-tools": { "name": "@corbits/agent-directory-tools", - "version": "0.0.4", + "version": "0.0.5", "dependencies": { "@intx/agent": "workspace:*", "@intx/types": "workspace:*", @@ -589,7 +589,7 @@ }, "packages/connections-tools": { "name": "@corbits/connections-tools", - "version": "0.0.5", + "version": "0.0.6", "dependencies": { "@intx/agent": "workspace:*", "@intx/types": "workspace:*", @@ -763,7 +763,7 @@ }, "packages/github-tools": { "name": "@corbits/github-tools", - "version": "0.0.6", + "version": "0.0.8", "dependencies": { "@intx/agent": "workspace:*", "@intx/types": "workspace:*", @@ -843,6 +843,7 @@ "name": "@corbits/inbox", "version": "0.0.1", "dependencies": { + "@corbits/error-sink": "workspace:*", "@corbits/mailbox": "github:corbitsdev/corbits-mailbox#caa5214a2811a66a6f3dd3b7631b5f53591c8c14", "@corbits/notify": "workspace:*", "@intx/hub-api": "workspace:*", @@ -969,7 +970,7 @@ }, "packages/manus-tools": { "name": "@corbits/manus-tools", - "version": "0.0.1", + "version": "0.0.11", "dependencies": { "@intx/agent": "0.3.0", "@intx/types": "0.3.0", @@ -3566,8 +3567,6 @@ "@babel/helper-compilation-targets/semver": ["semver@6.3.1", "", { "bin": { "semver": "bin/semver.js" } }, "sha512-BR7VvDCVHO+q2xBEWskxS6DJE1qRnb7DxzUrogb71CWoSficBxYsiAGd+Kl0mmq/MprG9yArRkyrQxTO6XjMzA=="], - "@corbits/artifacts-hub/@corbits/artifacts": ["@corbits/artifacts@github:corbitsdev/corbits-artifacts#81049ed", { "dependencies": { "@hono/standard-validator": "^0.2.3" }, "peerDependencies": { "@intx/types": "^0.2.2", "arktype": "^2.1.29", "drizzle-orm": "^0.45.2", "hono": "^4.12.32", "hono-openapi": "^1.2.0", "postgres": "^3.4.9" } }, "corbitsdev-corbits-artifacts-81049ed", "sha512-oTE0iFDyQdz0ifG1epo39pwaCaYaw19YcKXwfaZqAEQ56a1g9YIozXwH9CG4NaUTwcJKUeYGuNls6oJsMPisCw=="], - "@esbuild-kit/core-utils/esbuild": ["esbuild@0.18.20", "", { "optionalDependencies": { "@esbuild/android-arm": "0.18.20", "@esbuild/android-arm64": "0.18.20", "@esbuild/android-x64": "0.18.20", "@esbuild/darwin-arm64": "0.18.20", "@esbuild/darwin-x64": "0.18.20", "@esbuild/freebsd-arm64": "0.18.20", "@esbuild/freebsd-x64": "0.18.20", "@esbuild/linux-arm": "0.18.20", "@esbuild/linux-arm64": "0.18.20", "@esbuild/linux-ia32": "0.18.20", "@esbuild/linux-loong64": "0.18.20", "@esbuild/linux-mips64el": "0.18.20", "@esbuild/linux-ppc64": "0.18.20", "@esbuild/linux-riscv64": "0.18.20", "@esbuild/linux-s390x": "0.18.20", "@esbuild/linux-x64": "0.18.20", "@esbuild/netbsd-x64": "0.18.20", "@esbuild/openbsd-x64": "0.18.20", "@esbuild/sunos-x64": "0.18.20", "@esbuild/win32-arm64": "0.18.20", "@esbuild/win32-ia32": "0.18.20", "@esbuild/win32-x64": "0.18.20" }, "bin": { "esbuild": "bin/esbuild" } }, "sha512-ceqxoedUrcayh7Y7ZX6NdbbDzGROiyVBgC4PriJThBKSVPWnnFHZAkfI1lJT8QFkOwH4qOS2SJkS4wvpGl8BpA=="], "@eslint-community/eslint-utils/eslint-visitor-keys": ["eslint-visitor-keys@3.4.3", "", {}, "sha512-wpc+LXeiyiisxPlEkUzU6svyS1frIO3Mgxj1fdy7Pm8Ygzguax2N3Fa/D/ag1WqbOprdI+uY6wMUl8/a2G+iag=="], @@ -3590,7 +3589,7 @@ "@typescript-eslint/eslint-plugin/ignore": ["ignore@7.0.6", "", {}, "sha512-BAg6QkE8W+TuQLrrw0Ugr7HegXduRuuj8/ti2kSOc+jz1dmx8/WNcjr6XGnq5YpDWxFwwaavqD0+jIUOKelTsw=="], - "@workbench/hub/@corbits/artifacts": ["@corbits/artifacts@github:corbitsdev/corbits-artifacts#81049ed", { "dependencies": { "@hono/standard-validator": "^0.2.3" }, "peerDependencies": { "@intx/types": "^0.2.2", "arktype": "^2.1.29", "drizzle-orm": "^0.45.2", "hono": "^4.12.32", "hono-openapi": "^1.2.0", "postgres": "^3.4.9" } }, "corbitsdev-corbits-artifacts-81049ed", "sha512-oTE0iFDyQdz0ifG1epo39pwaCaYaw19YcKXwfaZqAEQ56a1g9YIozXwH9CG4NaUTwcJKUeYGuNls6oJsMPisCw=="], + "@workbench/hub/@corbits/mailbox": ["@corbits/mailbox@github:corbitsdev/corbits-mailbox#caa5214", { "dependencies": { "@hono/standard-validator": "0.2.3", "@standard-community/standard-json": "0.3.5", "@standard-community/standard-openapi": "0.2.9", "arktype": "2.1.29", "hono-openapi": "1.3.1" }, "peerDependencies": { "@intx/log": "^0.2.2", "@intx/mime": "^0.2.2", "@intx/types": "^0.2.2", "drizzle-orm": "^0.45.2", "hono": "^4.12.0", "postgres": "^3.4.0" } }, "corbitsdev-corbits-mailbox-caa5214", "sha512-z8DRBFgA4ukM8p29COeaMjfKZYe5jAUF4OBMiaIQFuW592+DGD/y6Ws6SjGlXmR9azkHNWh8oTzjlWlRP24vsQ=="], "ajv-formats/ajv": ["ajv@8.20.0", "", { "dependencies": { "fast-deep-equal": "^3.1.3", "fast-uri": "^3.0.1", "json-schema-traverse": "^1.0.0", "require-from-string": "^2.0.2" } }, "sha512-Thbli+OlOj+iMPYFBVBfJ3OmCAnaSyNn4M1vz9T6Gka5Jt9ba/HIR56joy65tY6kx/FCF5VXNB819Y7/GUrBGA=="], diff --git a/packages/inbox/package.json b/packages/inbox/package.json index 7d2aeabcb..eaeb0caea 100644 --- a/packages/inbox/package.json +++ b/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:*", diff --git a/packages/inbox/src/routes.ts b/packages/inbox/src/routes.ts index 21123b3bb..10d0bdd2c 100644 --- a/packages/inbox/src/routes.ts +++ b/packages/inbox/src/routes.ts @@ -17,13 +17,18 @@ 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, @@ -31,6 +36,7 @@ import { 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 { - 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(); + // 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, + tenantId: string, + operation: string, + scope: { tenantId: string; principalId: string }, + filter?: MailboxFilter, + ): Promise { + 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) => { diff --git a/packages/inbox/test/routes.test.ts b/packages/inbox/test/routes.test.ts index e96e82789..ac1bab718 100644 --- a/packages/inbox/test/routes.test.ts +++ b/packages/inbox/test/routes.test.ts @@ -25,9 +25,20 @@ function neverCalledDb(): MailboxDb { ) as MailboxDb; } -function mount(): Hono { +function throwingDb(message: string): MailboxDb { + return new Proxy( + {}, + { + get() { + throw new Error(message); + }, + }, + ) as MailboxDb; +} + +function mount(db: MailboxDb = neverCalledDb()): Hono { const routes = createInboxRoutes({ - db: neverCalledDb(), + db, bus: createInMemoryMailboxEventBus(), }); const app = new Hono(); @@ -71,3 +82,33 @@ describe("GET / cursor/filter cross-check", () => { expect(body.error).toBe("malformed cursor"); }); }); + +describe("bulk ops surface a failed inbox walk instead of silently truncating it (CL-7207)", () => { + test("GET /counts reports and 500s rather than swallowing a walk failure", async () => { + const app = mount(throwingDb("connection reset mid-walk")); + const response = await app.request("/counts"); + expect(response.status).toBe(500); + const body = (await response.json()) as { error: string; refId: string }; + expect(body.error).toBe("could not list the inbox"); + expect(typeof body.refId).toBe("string"); + expect(body.refId.length).toBeGreaterThan(0); + }); + + test("POST /mark-all-read reports and 500s rather than silently marking nothing", async () => { + const app = mount(throwingDb("connection reset mid-walk")); + const response = await app.request("/mark-all-read", { method: "POST" }); + expect(response.status).toBe(500); + const body = (await response.json()) as { error: string; refId: string }; + expect(body.error).toBe("could not list the inbox"); + expect(typeof body.refId).toBe("string"); + }); + + test("POST /clear-done reports and 500s rather than silently clearing nothing", async () => { + const app = mount(throwingDb("connection reset mid-walk")); + const response = await app.request("/clear-done", { method: "POST" }); + expect(response.status).toBe(500); + const body = (await response.json()) as { error: string; refId: string }; + expect(body.error).toBe("could not list the inbox"); + expect(typeof body.refId).toBe("string"); + }); +});