diff --git a/.changeset/notification-mark-all-read-full-sweep.md b/.changeset/notification-mark-all-read-full-sweep.md new file mode 100644 index 0000000000..21e141c26f --- /dev/null +++ b/.changeset/notification-mark-all-read-full-sweep.md @@ -0,0 +1,52 @@ +--- +"@objectstack/service-messaging": patch +--- + +fix(services): `markAllRead` clears the WHOLE inbox, not one 200-row window (#6436) + +`POST /api/v1/notifications/read/all` is published as "mark **every** +currently-unread inbox message as read". It swept +`listInbox(userId, { read: false, limit: 200 })` — one page of the LIST, and +`200` is that list's hard cap — so it cleared at most 200 receipts per call. + +Measured over the real stack (sqlite-wasm + ObjectQL + service-messaging + hono ++ dispatcher), one user, 260 unread: + +| request | before | after | +|:---|---:|---:| +| `POST /notifications/read/all` | `readCount: 200` | `readCount: 260` | +| `GET /notifications` (same user, next) | `unreadCount: 60` | `unreadCount: 0` | + +**#6363 did not cause this — it removed the cover.** While `unreadCount` was +itself window-scoped the shortfall was self-consistent and invisible: clear 200, +poll, see a window with nothing unread in it, badge 0. Now that the badge is a +true total, the same request pair states the contradiction out loud, which also +raises the severity — a user presses "mark all read" and the badge stays lit. + +**A second, sharper face of the same defect, fixed with it.** That window was +`created_at desc` over ALL rows, with the `read` filter applied in memory AFTER +the truncation. An inbox whose newest 200 were already read therefore handed the +sweep an EMPTY id list and marked **nothing at all**, however much older unread +sat behind it. That is also why "loop the pages until one comes back empty" is +not the fix: it exits on exactly that empty first page. + +**What it does now.** It reads the unread SET rather than a page of the list, in +a FIXED two reads whatever the inbox size — the same one-column, unwindowed +projection of `sys_inbox_message` that #6363's `countUnreadTotal` already issues +to answer the badge, joined against the receipt spine `listInbox` already reads +unbounded. No loop and no page count to bound; nothing is asked of the data +layer that the bell's poll does not already ask on every saturated page. The +write stays one receipt per unread notification — that is the receipt model +itself (ADR-0030) — and `markRead`'s check-then-act upsert, its unique-conflict +convergence and its "no receipt row yet" insert are untouched. + +`readCount` now reports the number of **distinct notifications this call flipped +to `read`** (it reported "the unread ones inside the newest 200 rows"). Two +consequences: a notification materialized by several inbox rows counts once, and +an inbox row carrying no `notification_id` is skipped rather than counted — +read-state is keyed by the event id, so the receipt the old code wrote for those +(keyed by the inbox ROW id) was one the join could never read back. + +Unchanged: the list window (default 50, cap 200, newest first), `unreadCount`, +`markRead`, and an inbox smaller than the old window — which does the same +writes it always did, and no extra ones. diff --git a/packages/runtime/src/notifications.hono.integration.test.ts b/packages/runtime/src/notifications.hono.integration.test.ts index 80c6ad7048..f43e068f2c 100644 --- a/packages/runtime/src/notifications.hono.integration.test.ts +++ b/packages/runtime/src/notifications.hono.integration.test.ts @@ -104,12 +104,14 @@ describe('in-app notifications over a real hono server (integration, #3362)', () } }, 30_000); - const authed = (path: string, init?: RequestInit) => + const as = (user: string, path: string, init?: RequestInit) => fetch(`${baseUrl}${path}`, { ...init, - headers: { 'x-test-user': TEST_USER, 'content-type': 'application/json', ...(init?.headers ?? {}) }, + headers: { 'x-test-user': user, 'content-type': 'application/json', ...(init?.headers ?? {}) }, }); + const authed = (path: string, init?: RequestInit) => as(TEST_USER, path, init); + it('resolves the notification service in discovery (declared === enforced)', async () => { const res = await fetch(`${baseUrl}/api/v1/discovery`); expect(res.status).toBe(200); @@ -173,4 +175,54 @@ describe('in-app notifications over a real hono server (integration, #3362)', () expect(receipts.length).toBe(2); expect(receipts.every((r: any) => r.state === 'read')).toBe(true); }); + + it('[#6436] mark-all-read clears an inbox LARGER than the list window — no readCount/unreadCount contradiction', async () => { + // The issue, over the wire, on the real stack. `markAllRead` swept one page + // of the list (`limit: 200`, that list's hard cap), so a user with more + // unread than the cap pressed "mark all read" and the badge did not clear: + // + // POST /api/v1/notifications/read/all → { readCount: 200 } + // GET /api/v1/notifications → { unreadCount: 150 } + // + // #6363 did not cause that — it removed the cover. While `unreadCount` was + // window-scoped the shortfall was self-consistent and invisible; once the + // badge became a true total the same request pair states it out loud, which + // is why this pin lives at the wire and not only under the service. + // + // Rows are seeded through the data engine rather than 260 `emit()` calls: + // this asserts the READ/SWEEP path over HTTP, and the delivery path is + // already covered by the test above. + const BULK_USER = 'usr_notif_bulk'; + const TOTAL = 260; // > the list's hard cap of 200 + const data = kernel.getService('data'); + for (let i = 0; i < TOTAL; i++) { + await data.insert('sys_inbox_message', { + user_id: BULK_USER, + notification_id: `bulk_n${String(i).padStart(3, '0')}`, + topic: 'deal.won', + title: `Bulk ${i}`, + body_md: 'body', + severity: 'info', + created_at: `2026-02-01T00:00:00.${String(i).padStart(3, '0')}Z`, + }); + } + + const before = await (await as(BULK_USER, '/api/v1/notifications')).json(); + expect(before.data.unreadCount).toBe(TOTAL); // the true total (#6363) + expect(before.data.notifications).toHaveLength(50); // the list is still one window + + const readAll = await (await as(BULK_USER, '/api/v1/notifications/read/all', { method: 'POST' })).json(); + expect(readAll.data).toMatchObject({ success: true, readCount: TOTAL }); // was 200 + + const after = await (await as(BULK_USER, '/api/v1/notifications')).json(); + expect(after.data.unreadCount).toBe(0); // was 60 + expect(after.data.notifications.every((n: any) => n.read === true)).toBe(true); + + // Persisted, not computed: one `read` receipt per notification. + const receipts = await data.find('sys_notification_receipt', { + where: { user_id: BULK_USER, channel: 'inbox' }, + }); + expect(receipts.length).toBe(TOTAL); + expect(receipts.every((r: any) => r.state === 'read')).toBe(true); + }, 120_000); }); diff --git a/packages/services/service-messaging/src/messaging-service.test.ts b/packages/services/service-messaging/src/messaging-service.test.ts index 7233f978fa..ff7972aa8c 100644 --- a/packages/services/service-messaging/src/messaging-service.test.ts +++ b/packages/services/service-messaging/src/messaging-service.test.ts @@ -793,3 +793,192 @@ describe('[#6363] listInbox — unreadCount is the TOTAL unread, not the fetched expect((await svc.listInbox('u1', { limit: 120 })).notifications).toHaveLength(120); }); }); + +/** + * [#6436] `markAllRead` swept `listInbox(userId, { read: false, limit: 200 })` + * — one page of the LIST — and `200` is that list's hard cap, so the route + * documented as "mark **every** currently-unread inbox message as read" + * cleared at most 200 receipts per call. + * + * #6363 did not introduce this; it removed the cover. While `unreadCount` was + * counted over the window the truncation was self-consistent and invisible + * (clear 200, poll, see a window with nothing unread in it, badge 0). Now that + * the badge is the true total, one response pair states the contradiction on + * its own: `POST /read/all → { readCount: 200 }` then + * `GET /notifications → { unreadCount: 150 }`. + * + * Route C — redefine "all" as "the current window" — was excluded by the + * maintainer's #6363 Option A ruling (make the declaration true). The sweep now + * reads the unread SET directly instead of a page of the list. + */ +describe('[#6436] markAllRead — sweeps the whole inbox, not one 200-row window', () => { + const logger = silentLogger(); + + it("clears an inbox holding more unread than the list's hard cap (the issue's 350)", async () => { + const engine = inboxEngine({ inbox: seedInbox('u1', 350) }); + const svc = new MessagingService({ logger, getData: () => engine }); + + const res = await svc.markAllRead('u1'); + // Before: `readCount: 200`, and 150 messages still unread behind a + // badge that — since #6363 — reported them correctly. + expect(res).toEqual({ success: true, readCount: 350 }); + expect((await svc.listInbox('u1')).unreadCount).toBe(0); + + // Persisted read-state, not a view-layer computation: one receipt per + // notification, every one of them `read`. + const receipts = engine.store.sys_notification_receipt; + expect(receipts).toHaveLength(350); + expect(receipts.every((r: any) => r.state === 'read')).toBe(true); + }); + + it('marks the older unread even when the newest 200 are already read', async () => { + // The sharper face of the same defect, and the reason "loop `listInbox` + // until it comes back empty" is not merely costly but WRONG: the window + // is `created_at desc` over ALL rows and the `read` filter is applied + // in memory AFTER the truncation. An inbox whose newest 200 are read + // therefore handed the sweep an EMPTY id list — it marked nothing at + // all, however much older unread sat behind it, and a paging loop would + // have exited on that same empty first page. Reading the unread SET + // makes a message's position in the inbox stop mattering. + const engine = inboxEngine({ + inbox: seedInbox('u1', 350), + // m151…m350 are the NEWEST 200 (created_at .150 … .349). + receipts: Array.from({ length: 200 }, (_, i) => readReceipt('u1', i + 151)), + }); + const svc = new MessagingService({ logger, getData: () => engine }); + expect((await svc.listInbox('u1')).unreadCount).toBe(150); + + expect((await svc.markAllRead('u1')).readCount).toBe(150); // before: 0 + expect((await svc.listInbox('u1')).unreadCount).toBe(0); + }); + + it('a small inbox behaves exactly as before — only the unread flip', async () => { + const engine = inboxEngine({ + inbox: seedInbox('u1', 10), + receipts: [readReceipt('u1', 1), readReceipt('u1', 2)], + }); + const before = engine.store.sys_notification_receipt.map((r: any) => ({ ...r })); + const svc = new MessagingService({ logger, getData: () => engine }); + + expect(await svc.markAllRead('u1')).toEqual({ success: true, readCount: 8 }); + expect((await svc.listInbox('u1')).unreadCount).toBe(0); + + // The two already-read receipts are not re-stamped: an inbox smaller + // than the old window is the case that was never broken, and it must + // not start doing extra writes to prove it. + const after = engine.store.sys_notification_receipt; + expect(after).toHaveLength(10); + expect(after.slice(0, 2)).toEqual(before); + }); + + it('costs a fixed two reads however large the inbox is — no paging loop', async () => { + for (const n of [10, 350]) { + const engine = inboxEngine({ inbox: seedInbox('u1', n) }); + const calls = recordFinds(engine); + const svc = new MessagingService({ logger, getData: () => engine }); + + await svc.markAllRead('u1'); + + expect(calls, `inbox of ${n}`).toHaveLength(2); + const inboxRead = calls.find((c) => c.object === 'sys_inbox_message')!; + // Unwindowed, unordered and one column wide — the same projection + // #6363's `countUnreadTotal` already reads to answer the badge, so + // the sweep asks the data layer for nothing the bell poll does not + // ask it on every saturated page. + expect(inboxRead.query.where).toEqual({ user_id: 'u1' }); + expect(inboxRead.query.fields).toEqual(['notification_id']); + expect(inboxRead.query.limit).toBeUndefined(); + expect(inboxRead.query.orderBy).toBeUndefined(); + expect(calls.filter((c) => c.object === 'sys_notification_receipt')).toHaveLength(1); + } + }); + + it('touches only the addressed user', async () => { + const engine = inboxEngine({ + inbox: [ + ...seedInbox('u1', 350), + { id: 'x1', user_id: 'u2', notification_id: 'xn1', title: 'X', body_md: 'x', created_at: '2026-01-01T00:00:00.900Z' }, + { id: 'x2', user_id: 'u2', notification_id: 'xn2', title: 'Y', body_md: 'y', created_at: '2026-01-01T00:00:00.901Z' }, + ], + }); + const svc = new MessagingService({ logger, getData: () => engine }); + + expect((await svc.markAllRead('u1')).readCount).toBe(350); + expect((await svc.listInbox('u2')).unreadCount).toBe(2); + expect(engine.store.sys_notification_receipt.every((r: any) => r.user_id === 'u1')).toBe(true); + }); + + it('is idempotent — a second sweep writes nothing and reports 0', async () => { + const engine = inboxEngine({ inbox: seedInbox('u1', 350) }); + const svc = new MessagingService({ logger, getData: () => engine }); + + expect((await svc.markAllRead('u1')).readCount).toBe(350); + expect((await svc.markAllRead('u1')).readCount).toBe(0); + expect(engine.store.sys_notification_receipt).toHaveLength(350); + }); + + it('counts a notification once when several inbox rows materialize it', async () => { + // `readCount` reports NOTIFICATIONS flipped, and the receipt is keyed + // `(notification_id, user_id, channel)` — one row per notification + // however many inbox rows point at it. Feeding the id twice would have + // counted a second upsert that wrote nothing new. + const inbox = seedInbox('u1', 3); + inbox[2].notification_id = 'n1'; // m3 re-materializes n1 + const engine = inboxEngine({ inbox }); + const svc = new MessagingService({ logger, getData: () => engine }); + + expect((await svc.markAllRead('u1')).readCount).toBe(2); + expect(engine.store.sys_notification_receipt).toHaveLength(2); + expect((await svc.listInbox('u1')).unreadCount).toBe(0); + }); + + it('reads the receipt spine best-effort, exactly as listInbox does', async () => { + // Read-state lives on a DIFFERENT object which a minimal stack may not + // have registered (`listInbox` degrades to "everything unread" for the + // same reason). Degrading to "sweep them all" is the safe direction: + // re-marking a read message is idempotent, skipping an unread one is + // the defect this issue is about. + const engine = inboxEngine({ + inbox: seedInbox('u1', 3), + receipts: [readReceipt('u1', 1)], + }); + const real = engine.find.bind(engine); + engine.find = async (object: string, query: any = {}) => { + if (object === 'sys_notification_receipt') throw new Error('receipts unavailable'); + return real(object, query); + }; + const svc = new MessagingService({ logger, getData: () => engine }); + + expect((await svc.markAllRead('u1')).readCount).toBe(3); + }); + + it('skips rows carrying no event id — they key no receipt at all', async () => { + // Recorded, NOT endorsed. Read-state is keyed by the EVENT id + // (ADR-0030) and the inbox channel writes no receipt for a row without + // one, so there is nothing this sweep can write for it. The old code + // fed `markRead` the inbox ROW id (`listInbox` views it as `nid ?? + // String(m.id)`), which inserted a receipt the join never reads back — + // it could not make the row read and still counted itself into + // `readCount`. Skipping it keeps `readCount` honest; #6363's count goes + // on reporting the row as unread, which is the true state. Whether such + // a row should be readable at all is #6448 — a gap in the receipt KEY, + // not in this sweep, and dormant: the single `emit()` ingress always + // carries an event id, so only data written around it can be null. + const inbox = seedInbox('u1', 3); + inbox[0].notification_id = null; + const engine = inboxEngine({ inbox }); + const svc = new MessagingService({ logger, getData: () => engine }); + + expect((await svc.markAllRead('u1')).readCount).toBe(2); + expect(engine.store.sys_notification_receipt).toHaveLength(2); + expect((await svc.listInbox('u1')).unreadCount).toBe(1); + }); + + it('still degrades to a no-op without a data engine or user id', async () => { + const noData = new MessagingService({ logger }); + expect(await noData.markAllRead('u1')).toEqual({ success: true, readCount: 0 }); + + const svc = new MessagingService({ logger, getData: () => inboxEngine({ inbox: seedInbox('u1', 3) }) }); + expect(await svc.markAllRead('')).toEqual({ success: true, readCount: 0 }); + }); +}); diff --git a/packages/services/service-messaging/src/messaging-service.ts b/packages/services/service-messaging/src/messaging-service.ts index 1525abcfa9..a070bdfbbc 100644 --- a/packages/services/service-messaging/src/messaging-service.ts +++ b/packages/services/service-messaging/src/messaging-service.ts @@ -309,27 +309,11 @@ export class MessagingService { const where: Record = { user_id: userId }; if (opts.type) where.topic = opts.type; - const [rows, receipts] = await Promise.all([ + const [rows, stateByNotif] = await Promise.all([ data.find(INBOX_OBJECT, { where, orderBy: [{ field: 'created_at', order: 'desc' }], limit }) as Promise>>, - // Read-state spine. Best-effort: if receipts are unavailable the - // inbox still lists (everything reads as unread) rather than erroring. - (data.find(RECEIPT_OBJECT, { where: { user_id: userId, channel: 'inbox' } }) as Promise>>) - .catch(() => [] as Array>), + this.readReceiptStates(data, userId), ]); - // notification_id → most-advanced receipt state (read/clicked/dismissed - // wins over a plain delivered one). - const stateByNotif = new Map(); - for (const r of receipts) { - const nid = r?.notification_id != null ? String(r.notification_id) : ''; - if (!nid) continue; - const state = String(r.state ?? 'delivered'); - const prev = stateByNotif.get(nid); - if (!prev || (!READ_RECEIPT_STATES.has(prev) && READ_RECEIPT_STATES.has(state))) { - stateByNotif.set(nid, state); - } - } - let windowUnread = 0; const all: InboxNotificationView[] = rows.map((m) => { const nid = m?.notification_id != null ? String(m.notification_id) : null; @@ -403,6 +387,37 @@ export class MessagingService { return unread; } + /** + * The user's inbox read-state spine: `notification_id` → most-advanced + * receipt state, where `read`/`clicked`/`dismissed` wins over a plain + * `delivered` one (ADR-0030 keeps read-state on `sys_notification_receipt`, + * not on the inbox row). + * + * Best-effort by design: receipts are a DIFFERENT object which a minimal + * stack may not have registered at all, so an unavailable spine degrades to + * "nothing is known to be read" rather than failing the caller. Both + * callers survive that direction — `listInbox` lists everything as unread, + * and `markAllRead` sweeps everything (re-marking a read message is + * idempotent; skipping an unread one is #6436). + */ + private async readReceiptStates(data: IDataEngine, userId: string): Promise> { + const receipts = await (data.find(RECEIPT_OBJECT, { + where: { user_id: userId, channel: 'inbox' }, + }) as Promise>>).catch(() => [] as Array>); + + const stateByNotif = new Map(); + for (const r of receipts) { + const nid = r?.notification_id != null ? String(r.notification_id) : ''; + if (!nid) continue; + const state = String(r.state ?? 'delivered'); + const prev = stateByNotif.get(nid); + if (!prev || (!READ_RECEIPT_STATES.has(prev) && READ_RECEIPT_STATES.has(state))) { + stateByNotif.set(nid, state); + } + } + return stateByNotif; + } + /** * Mark specific notifications read by upserting their inbox receipts to * `read`. Updates the existing `delivered` receipt in place (keyed @@ -428,14 +443,96 @@ export class MessagingService { } /** - * Mark every currently-unread inbox message for the user as read. Returns - * `{ success, readCount }` (`MarkAllNotificationsReadResponseSchema`). + * Mark every currently-unread inbox message for the user as read — the + * whole inbox, not a page of it (#6436). Returns `{ success, readCount }` + * (`MarkAllNotificationsReadResponseSchema`), where `readCount` is the + * number of DISTINCT notifications this call flipped to `read`: the + * unread set it found, minus any whose receipt write failed (those are + * logged by `markRead` and left for the next sweep, which is idempotent). + * + * It used to sweep `listInbox(userId, { read: false, limit: 200 })` — one + * page of the LIST, and 200 is that list's hard cap — so the route + * documented as "mark **every** currently-unread inbox message as read" + * cleared at most 200 receipts per call. Two ways that showed: + * + * * 350 unread → `{ readCount: 200 }`, 150 still unread. Invisible while + * `unreadCount` was itself window-scoped; since #6363 made the badge a + * true total, one response pair states the contradiction on its own. + * * Worse, and the reason a paging loop is not the fix: that window is + * `created_at desc` over ALL rows, with the `read` filter applied in + * memory AFTER the truncation. An inbox whose newest 200 are already + * read handed the sweep an EMPTY list and marked NOTHING, however much + * older unread sat behind it — and "loop until the page comes back + * empty" exits on exactly that empty first page. + * + * So the sweep reads the unread SET instead of a page of the list, in a + * FIXED two reads whatever the inbox size: the same one-column, unwindowed + * projection of `sys_inbox_message` that #6363's `countUnreadTotal` already + * issues to answer the badge, joined against the receipt spine that + * `listInbox` already reads unbounded. No loop, no page count to bound, and + * nothing asked of the data layer that the bell's poll does not ask on + * every saturated page. What remains linear is the WRITE — one receipt per + * unread notification, which is the receipt model itself (ADR-0030), and + * the only thing that could collapse it is a predicate-shaped bulk write + * (route B), which would have to redesign `markRead`'s check-then-act + * upsert and its "no receipt row yet" insert face. Deliberately not done + * here. + * + * No cap: a numeric safety valve is route C wearing a larger number — above + * it, "all" would be a lie again, which is the reading the maintainer ruled + * against on #6363 (make the declaration true rather than document the + * shortfall). What bounds a pathological inbox instead is that the work is + * idempotent and resumable — a failed receipt write is logged, skipped, and + * picked up by the next sweep. */ async markAllRead(userId: string): Promise<{ success: boolean; readCount: number }> { const data = this.ctx.getData?.(); if (!data || !userId) return { success: true, readCount: 0 }; - const { notifications } = await this.listInbox(userId, { read: false, limit: 200 }); - return this.markRead(userId, notifications.map((n) => n.id)); + return this.markRead(userId, await this.unreadNotificationIds(data, userId)); + } + + /** + * Every notification id in the user's inbox that has no `read`-class + * receipt yet — the set `markAllRead` must flip, deduplicated so a + * notification materialized by several inbox rows is counted (and upserted) + * once, since its receipt is keyed `(notification_id, user_id, channel)`. + * + * The inbox read is NOT best-effort, matching `countUnreadTotal`: it is the + * primary read, and a failure means we do not know what to mark — far + * better to surface it than to report a confident `readCount` over a set we + * could not see. The receipt read degrades (see `readReceiptStates`). + * + * Rows carrying no `notification_id` are skipped: read-state is keyed by + * the event id, and the inbox channel writes no receipt for a row without + * one, so there is nothing addressable to write. The previous sweep fed + * `markRead` the inbox ROW id for those (`listInbox` views them as + * `nid ?? String(m.id)`), inserting a receipt the join never reads back — + * it could not make the row read and still counted itself into + * `readCount`. They keep reporting as unread, which is the true state. + * Whether such a row should be readable at all is a gap in the receipt KEY, + * not in this sweep: filed as #6448 (dormant — the single `emit()` ingress + * always carries an event id, so only data written around it can be null). + */ + private async unreadNotificationIds(data: IDataEngine, userId: string): Promise { + const [rows, stateByNotif] = await Promise.all([ + data.find(INBOX_OBJECT, { + where: { user_id: userId }, + fields: ['notification_id'], + }) as Promise>>, + this.readReceiptStates(data, userId), + ]); + + const ids: string[] = []; + const seen = new Set(); + for (const row of rows) { + const nid = row?.notification_id != null ? String(row.notification_id) : ''; + if (!nid || seen.has(nid)) continue; + const state = stateByNotif.get(nid); + if (state && READ_RECEIPT_STATES.has(state)) continue; + seen.add(nid); + ids.push(nid); + } + return ids; } /** Upsert a `read` receipt for one notification; returns 1 when it persisted. */