From 8321b6a82930f4eaa0631c941a7143f1fa4141c2 Mon Sep 17 00:00:00 2001 From: os-sam Date: Mon, 24 Aug 2026 12:24:59 +0000 Subject: [PATCH] fix(messaging): stamp organization_id on flow-produced notifications and markRead receipts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `sys_inbox_message`, `sys_notification`, `sys_notification_receipt` and `sys_notification_delivery` were measured at 100% `organization_id = NULL` on a live install, while `sys_approval_request` in the same database carried an organization on every row. Ruled a gap, not a design choice. The chain below the messaging ingress was already threaded end to end — every writer reads `EmitInput.organizationId`. The break was at the origin: the `notify` flow node never passed it, and its local structural mirror of `emit()` did not declare the field, so it could not have. The node now threads the organization from the run's acting context (`AutomationContext.tenantId`), the same source `plugin-audit`'s `collab.mention` producer already uses. A second producer of the same table is fixed alongside: the `read` receipt `markRead` inserts named no organization at all, and now carries the organization of the `sys_notification` row it is about. No fallback limb in either producer, by design: a run with no organization in scope still emits and still writes, and the node warns audibly instead of guessing. A wrong organization_id is worse than a null — a null is visibly missing, a wrong value is silently authoritative. Forward-stamping only; no backfill and no migration. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01APWX2AwT3a4xDcjPCe8bk4 --- ...amp-organization-on-notification-writes.md | 41 +++ .../src/builtin/notify-node.ts | 56 ++++ ...ify-organization-stamp.integration.test.ts | 247 ++++++++++++++++++ .../src/messaging-service.ts | 55 ++++ .../src/read-receipt-organization.test.ts | 87 ++++++ 5 files changed, 486 insertions(+) create mode 100644 .changeset/stamp-organization-on-notification-writes.md create mode 100644 packages/services/service-automation/src/builtin/notify-organization-stamp.integration.test.ts create mode 100644 packages/services/service-messaging/src/read-receipt-organization.test.ts diff --git a/.changeset/stamp-organization-on-notification-writes.md b/.changeset/stamp-organization-on-notification-writes.md new file mode 100644 index 0000000000..158a2d3263 --- /dev/null +++ b/.changeset/stamp-organization-on-notification-writes.md @@ -0,0 +1,41 @@ +--- +'@objectstack/service-automation': patch +'@objectstack/service-messaging': patch +--- + +Stamp `organization_id` on flow-produced notifications and on `markRead` +receipts, so the notification family stops writing org-less rows + +An application project's read-only inventory found `sys_inbox_message`, +`sys_notification`, `sys_notification_receipt` and `sys_notification_delivery` +carrying `organization_id = NULL` on **100%** of their rows — existing rows and +same-day new ones alike, while `sys_approval_request` in the same database +carried an organization on every row. Ruled a gap, not a design choice. + +Everything below the messaging ingress was already threaded: `emit()` stamps the +`sys_notification` event, the inbox channel stamps `sys_inbox_message` and its +`delivered` receipt, and the outbox carries the value onto +`sys_notification_delivery`. Each of them reads `EmitInput.organizationId` — +and the `notify` flow node, the dominant producer, never supplied it. Its local +structural mirror of `emit()` did not even declare the field, so the value could +not have been passed. One missing argument, four tables at 100% null. + +The node now threads the organization from the run's own acting context +(`AutomationContext.tenantId`), the same source the `collab.mention` producer in +`@objectstack/plugin-audit` already uses, so the two notification producers agree +about whose organization a notification carries. + +A second producer of the same table is fixed alongside it: the `read` receipt +`markRead` inserts — written when a user reads a notification whose delivered +receipt never landed — named no organization at all. It now carries the +organization of the `sys_notification` row it is about. + +There is deliberately **no fallback limb** in either producer: not "the current +organization", not the install's first organization, not the recipient's first +membership. A run with no organization in scope still emits and still writes its +rows, and the `notify` node warns audibly naming the topic and the consequence. +A wrong `organization_id` is worse than a null — a null is visibly missing, +while a wrong value is silently authoritative to every report, export and +cleanup script that filters by organization. + +Forward-stamping only. Existing rows are not backfilled and no migration ships. diff --git a/packages/services/service-automation/src/builtin/notify-node.ts b/packages/services/service-automation/src/builtin/notify-node.ts index 391d968bd3..41f68dd486 100644 --- a/packages/services/service-automation/src/builtin/notify-node.ts +++ b/packages/services/service-automation/src/builtin/notify-node.ts @@ -25,6 +25,17 @@ export interface MessagingServiceSurface { dedupKey?: string; source?: { object: string; id: string }; actorId?: string; + /** + * [#11303] The organization the notification belongs to — the field the + * whole downstream chain stamps from. `MessagingService.writeEvent` + * puts it on `sys_notification`, the inbox channel puts it on + * `sys_inbox_message` and on the `delivered` receipt, and the outbox + * carries it onto the `sys_notification_delivery` row. It was missing + * from this structural mirror, so the node could not have passed it + * even if it had tried: four tables landed 100% org-less on every + * flow-produced notification. + */ + organizationId?: string; channels?: string[]; }): Promise<{ notificationId: string; @@ -312,6 +323,46 @@ export function registerNotifyNode(engine: AutomationEngine, ctx: PluginContext) }; } + // [#11303] The organization this notification belongs to, THREADED + // from the run's own acting context — never fabricated. + // + // Maintainer ruling, 2026-08-24, verbatim: 「11303 + // sys_inbox_message/sys_notification/sys_email 应该写 + // organization_id。」 — a gap, not a design choice, and the + // PRODUCERS are the fix site. Everything below `emit()` was already + // threaded; this node was the origin that never supplied a value. + // + // `AutomationContext.tenantId` is the acting run's organization — + // the same source `audit-writers.ts` hands its own `collab.mention` + // emit, so the two notification producers agree about whose + // organization a notification carries. + // + // ⛔ There is deliberately NO fallback limb here — not "the current + // organization", not the first organization on the install, not the + // recipient's first membership. A wrong `organization_id` is worse + // than a null: a null is visibly missing, while a wrong value is + // silently authoritative to every report, export and cleanup script + // that filters by organization. When the run carries no + // organization, the notification carries none and says so (below). + const organizationId = toStr(context.tenantId); + if (!organizationId) { + // Fail-LOUD, not fail-guess — and deliberately not fail-CLOSED. + // Refusing here would break the two deployments that legitimately + // have no organization to thread: a `single`-posture install, and + // every stack before its first organization exists. So the + // org-less write stays permitted and becomes a VISIBLE event + // instead of a silent one. + ctx.logger.warn( + `[notify] no organization in scope for topic '${topic ?? 'notify'}' — the ` + + `sys_notification / sys_inbox_message / sys_notification_receipt / ` + + `sys_notification_delivery rows for this emit will carry organization_id = NULL ` + + `and will be invisible to any report or cleanup that filters by organization. ` + + `On a multi-organization install this means the triggering context lost its ` + + `tenant: give the flow's trigger an acting organization (AutomationContext.tenantId). ` + + `On a single-organization install this is expected and can be ignored.`, + ); + } + try { // ADR-0030 single ingress: hand the messaging service a topic + // audience + payload; it writes the L2 event and materializes @@ -338,6 +389,11 @@ export function registerNotifyNode(engine: AutomationEngine, ctx: PluginContext) severity, source, actorId, + // [#11303] Absent (not null) when the run has no + // organization: `EmitInput.organizationId` is optional, and + // the chain below normalizes a missing value to NULL exactly + // once, in `writeEvent`. + ...(organizationId ? { organizationId } : {}), channels: channels.length ? channels : undefined, }); const delivered = Number(result.delivered) || 0; diff --git a/packages/services/service-automation/src/builtin/notify-organization-stamp.integration.test.ts b/packages/services/service-automation/src/builtin/notify-organization-stamp.integration.test.ts new file mode 100644 index 0000000000..f404afa4b1 --- /dev/null +++ b/packages/services/service-automation/src/builtin/notify-organization-stamp.integration.test.ts @@ -0,0 +1,247 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +import { describe, it, expect } from 'vitest'; +import { + MessagingService, + MemoryNotificationOutbox, + createInboxChannel, + INBOX_OBJECT, + RECEIPT_OBJECT, + NOTIFICATION_EVENT_OBJECT, +} from '@objectstack/service-messaging'; +import { AutomationEngine } from '../engine.js'; +import { registerNotifyNode } from './notify-node.js'; +import type { MessagingServiceSurface } from './notify-node.js'; + +/** + * [#11303] The `notify` node is the producer that decides whether the whole + * notification family carries an organization. + * + * Maintainer ruling, 2026-08-24, verbatim: 「11303 + * sys_inbox_message/sys_notification/sys_email 应该写 organization_id。」 — a + * GAP, not a design choice. + * + * The measurement that shapes these pins: the messaging chain BELOW `emit()` + * already threads an organization end to end — `writeEvent` stamps + * `organization_id` on `sys_notification`, the inbox channel stamps it on + * `sys_inbox_message` AND on the `delivered` receipt, and the outbox carries it + * onto the `sys_notification_delivery` row. Every one of those reads + * `notification.organizationId`, which is `EmitInput.organizationId`. The break + * is at the ORIGIN: the `notify` node never passes it, so a flow-produced + * notification lands org-less in four tables at once — which is exactly the + * 100%-null reading the card reports for all four. + * + * ⭐ The organization is THREADED from the run's own acting context + * (`AutomationContext.tenantId`), never fabricated. There is deliberately no + * "first organization" / "the current organization" fallback: a wrong + * `organization_id` is worse than a null, because a null is visibly missing + * while a wrong value is silently authoritative to every report, export and + * cleanup script that filters by organization. + */ + +function silentLogger(): any { + const l: any = { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} }; + l.child = () => l; + return l; +} + +/** A logger that records every warning line, for the fail-loud pin. */ +function recordingLogger(): { logger: any; warnings: string[] } { + const warnings: string[] = []; + const l: any = { + info: () => {}, + warn: (...args: unknown[]) => { warnings.push(args.map(String).join(' ')); }, + error: () => {}, + debug: () => {}, + }; + l.child = () => l; + return { logger: l, warnings }; +} + +/** Every row this stack wrote, in insertion order, with the object it landed in. */ +interface WrittenRow { object: string; row: Record } + +/** + * A capturing data engine. Reads answer empty (no preference rows, no dedup + * hit) so the default always-on `inbox` channel is the one that runs; writes + * are recorded verbatim, which is the only thing these pins assert on. + */ +function capturingEngine(): { engine: any; written: WrittenRow[] } { + const written: WrittenRow[] = []; + let seq = 0; + const engine = { + async insert(object: string, row: Record) { + written.push({ object, row: { ...row } }); + const id = row.id != null ? String(row.id) : `row_${++seq}`; + return { ...row, id }; + }, + async find() { return []; }, + async findOne() { return undefined; }, + }; + return { engine, written }; +} + +/** The four tables the ruling names for the notification family. */ +const NOTIFICATION_FAMILY = new Set([ + NOTIFICATION_EVENT_OBJECT, + INBOX_OBJECT, + RECEIPT_OBJECT, +]); + +/** + * The identity list this suite asserts on — `object:organization_id` per row, + * in write order. ⭐ Identities, not a count: an offsetting error (one row + * gaining an organization while another loses it) holds a count constant while + * the identity list inverts. + */ +function orgIdentities(written: WrittenRow[]): string[] { + return written + .filter((w) => NOTIFICATION_FAMILY.has(w.object)) + .map((w) => `${w.object}:${w.row.organization_id ?? 'NULL'}`); +} + +function notifyFlow(): any { + return { + name: 'nudge', + label: 'Nudge', + type: 'autolaunched' as const, + nodes: [ + { id: 'start', type: 'start' as const, label: 'Start' }, + { + id: 'notify', + type: 'notify' as const, + label: 'Notify', + config: { + topic: 'deal.won', + recipients: ['user_1'], + title: 'Renewal due', + message: 'Ping', + channels: ['inbox'], + }, + }, + { id: 'end', type: 'end' as const, label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'notify' }, + { id: 'e2', source: 'notify', target: 'end' }, + ], + }; +} + +/** + * The REAL messaging service with the REAL inbox channel behind the notify + * node — the seam under test is precisely the handoff between them, so a fake + * that answers `emit()` in one shot could not express it. + */ +function bootInlineStack(logger: any = silentLogger()) { + const { engine: data, written } = capturingEngine(); + const messaging = new MessagingService({ logger, getData: () => data }); + messaging.registerChannel(createInboxChannel({ getData: () => data })); + + const engine = new AutomationEngine(logger); + registerNotifyNode(engine, { + logger, + getService: (name: string) => (name === 'messaging' ? messaging : undefined), + } as any); + engine.registerFlow('nudge', notifyFlow()); + return { engine, messaging, written }; +} + +describe('#11303 — the notify producer stamps organization_id on the notification family', () => { + it('PIN A: threads the run\'s own organization onto the emit input', async () => { + const emitted: any[] = []; + const service: MessagingServiceSurface = { + async emit(n: any) { + emitted.push(n); + return { notificationId: 'evt_1', delivered: n.audience.length, failed: 0 }; + }, + }; + const engine = new AutomationEngine(silentLogger()); + registerNotifyNode(engine, { + logger: silentLogger(), + getService: (name: string) => (name === 'messaging' ? service : undefined), + } as any); + engine.registerFlow('nudge', notifyFlow()); + + const run = await engine.execute('nudge', { tenantId: 'org_pin_alpha' } as any); + + expect(run.success).toBe(true); + expect(emitted).toHaveLength(1); + // The named producer pin: the organization reaching `emit()` is the + // run's acting tenant, verbatim — not a derived or defaulted value. + expect(emitted[0].organizationId).toBe('org_pin_alpha'); + }); + + it('PIN B: a run under an organization writes ZERO org-less rows into the notification family', async () => { + const { engine, written } = bootInlineStack(); + + const run = await engine.execute('nudge', { tenantId: 'org_pin_alpha' } as any); + expect(run.success).toBe(true); + + // The end-to-end pin the ruling names, asserted as an IDENTITY list so a + // producer nobody enumerated cannot hide behind a stable count. + expect(orgIdentities(written)).toEqual([ + `${NOTIFICATION_EVENT_OBJECT}:org_pin_alpha`, + `${INBOX_OBJECT}:org_pin_alpha`, + `${RECEIPT_OBJECT}:org_pin_alpha`, + ]); + // Said the second way, so the pin still bites if the write ORDER changes: + // no row of the family may carry NULL. + expect(orgIdentities(written).filter((i) => i.endsWith(':NULL'))).toEqual([]); + }); + + it('PIN B2: the durable delivery row carries the same organization', async () => { + const { engine: data } = capturingEngine(); + const outbox = new MemoryNotificationOutbox(1); + const messaging = new MessagingService({ logger: silentLogger(), getData: () => data, outbox }); + messaging.registerChannel(createInboxChannel({ getData: () => data })); + const engine = new AutomationEngine(silentLogger()); + registerNotifyNode(engine, { + logger: silentLogger(), + getService: (name: string) => (name === 'messaging' ? messaging : undefined), + } as any); + engine.registerFlow('nudge', notifyFlow()); + + const run = await engine.execute('nudge', { tenantId: 'org_pin_alpha' } as any); + expect(run.success).toBe(true); + + const rows = await outbox.list(); + expect(rows).toHaveLength(1); + expect(rows[0].organizationId).toBe('org_pin_alpha'); + }); + + it('PIN C (over-denial control): a stack with no organization in scope still delivers', async () => { + // The control that stops the fix from degenerating into "refuse unless + // an organization is present". A `single`-posture deployment — and every + // fresh boot before the first organization exists — has no organization + // to thread, and a notify there must still emit and still write its rows. + // ⭐ A suite that only pinned "organization_id is present" would score + // green on an implementation that breaks exactly this deployment. + const { engine, written } = bootInlineStack(); + + const run = await engine.execute('nudge'); + + expect(run.success).toBe(true); + expect(orgIdentities(written)).toEqual([ + `${NOTIFICATION_EVENT_OBJECT}:NULL`, + `${INBOX_OBJECT}:NULL`, + `${RECEIPT_OBJECT}:NULL`, + ]); + }); + + it('PIN D (fail-loud, not fail-guess): an unresolvable organization warns audibly', async () => { + // Fail-LOUD by warning rather than refusing — see PIN C for why a + // refusal is not available here. The warning is what makes the org-less + // row a visible event instead of a silent one, and it must name the + // topic so the operator can find the producer. + const { logger, warnings } = recordingLogger(); + const { engine } = bootInlineStack(logger); + + const run = await engine.execute('nudge'); + + expect(run.success).toBe(true); + const line = warnings.find((w) => w.includes('organization')); + expect(line, `no organization warning in: ${JSON.stringify(warnings)}`).toBeDefined(); + expect(line).toContain('deal.won'); + }); +}); diff --git a/packages/services/service-messaging/src/messaging-service.ts b/packages/services/service-messaging/src/messaging-service.ts index ace6fc2b6a..6ff39483a4 100644 --- a/packages/services/service-messaging/src/messaging-service.ts +++ b/packages/services/service-messaging/src/messaging-service.ts @@ -749,6 +749,28 @@ export class MessagingService { // flight) can win the (notification_id, user_id, channel) unique index // between our read and write. Treat that collision as "someone else // created it" and flip the now-present row to `read` instead of failing. + // [#11303] The organization this receipt belongs to, THREADED from the + // notification the receipt is ABOUT. + // + // Maintainer ruling, 2026-08-24, verbatim: 「11303 + // sys_inbox_message/sys_notification/sys_email 应该写 organization_id。」 + // + // The inbox channel's `delivered` receipt already stamps the + // organization it was handed at fan-out. This is the OTHER + // `sys_notification_receipt` producer — the `read` receipt, inserted + // when a user reads a notification whose delivered-receipt never landed + // — and it named no organization at all, so it wrote a NULL however + // well the emit path upstream was threaded. One table, two producers. + // + // The subject record's own organization is the platform's standing + // answer for a platform row (#8287), and here it is also the only + // honest one: `markRead(userId, ids)` carries no tenant of its own, and + // ⛔ inventing one — the reader's active organization, their first + // membership — would let a receipt claim an organization its own + // notification does not have. A null is visibly missing; a wrong value + // is silently authoritative. + const organizationId = await this.notificationOrganization(data, notificationId); + try { await data.insert(RECEIPT_OBJECT, { notification_id: notificationId, @@ -757,6 +779,7 @@ export class MessagingService { channel: 'inbox', state: 'read', at, + organization_id: organizationId, created_at: at, }); return 1; @@ -766,6 +789,38 @@ export class MessagingService { } } + /** + * [#11303] The organization of the `sys_notification` a receipt is about, + * or `null`. + * + * Best-effort in one direction only: a read that fails or finds nothing + * yields `null`, which is the same value the row carried before this + * existed — so a degraded read can never turn into a WRONG organization, + * only into the missing one it already was. That asymmetry is the point: + * the failure mode this closes is silence, and the failure mode it must not + * open is confident invention. + */ + private async notificationOrganization( + data: IDataEngine, + notificationId: string, + ): Promise { + try { + const row = await data.findOne(NOTIFICATION_EVENT_OBJECT, { + where: { id: notificationId }, + fields: ['id', 'organization_id'], + }); + const org = (row as Record | undefined)?.organization_id; + return org != null && String(org) !== '' ? String(org) : null; + } catch (err) { + this.ctx.logger.warn( + `[messaging] could not read the organization of notification '${notificationId}' ` + + `(${(err as Error).message}); its read receipt is written with organization_id = NULL ` + + `rather than a guessed organization`, + ); + return null; + } + } + /** * The single notification ingress. Writes the L2 event, resolves the * audience, and fans the result out to its channels. An unregistered diff --git a/packages/services/service-messaging/src/read-receipt-organization.test.ts b/packages/services/service-messaging/src/read-receipt-organization.test.ts new file mode 100644 index 0000000000..62e0926b15 --- /dev/null +++ b/packages/services/service-messaging/src/read-receipt-organization.test.ts @@ -0,0 +1,87 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +import { describe, it, expect } from 'vitest'; +import { MessagingService, NOTIFICATION_EVENT_OBJECT } from './messaging-service.js'; +import { RECEIPT_OBJECT } from './inbox-channel.js'; + +/** + * [#11303] The SECOND `sys_notification_receipt` producer. + * + * Maintainer ruling, 2026-08-24, verbatim: 「11303 + * sys_inbox_message/sys_notification/sys_email 应该写 organization_id。」 + * + * The inbox channel's `delivered` receipt already stamps the organization it + * was handed. `markRead` writes the OTHER receipt — the `read` one, inserted + * when a user reads a notification whose delivered-receipt never landed — and + * that insert names no organization at all, so it lands org-less however well + * the emit path was threaded. One table, two producers, and a per-producer + * suite that only enumerated the first would have scored green with this one + * still emitting nulls. + * + * ⭐ Threaded, never fabricated: the organization is read off the + * `sys_notification` row the receipt is ABOUT — the subject record's own + * organization, which is the platform's standing answer for a platform row + * (#8287). When the notification itself carries none, the receipt carries none: + * a null is visibly missing, a guess is silently authoritative. + */ + +function silentLogger(): any { + const l: any = { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} }; + l.child = () => l; + return l; +} + +/** + * A data engine holding one `sys_notification` row and recording every insert. + * `findOne` answers the notification row for the event object and `undefined` + * for the receipt (so `markRead` takes the insert limb, not the flip limb). + */ +function engineWithNotification(notificationOrg: string | null) { + const inserted: Array<{ object: string; row: Record }> = []; + const engine: any = { + async insert(object: string, row: Record) { + inserted.push({ object, row: { ...row } }); + return { ...row, id: 'rec_1' }; + }, + async findOne(object: string, opts: any) { + if (object === NOTIFICATION_EVENT_OBJECT) { + return { id: 'evt_pin', organization_id: notificationOrg }; + } + void opts; + return undefined; + }, + async find() { return []; }, + }; + return { engine, inserted }; +} + +describe('#11303 — the markRead receipt producer stamps organization_id', () => { + it('PIN E: the `read` receipt carries the notification\'s own organization', async () => { + const { engine, inserted } = engineWithNotification('org_pin_beta'); + const messaging = new MessagingService({ logger: silentLogger(), getData: () => engine }); + + const result = await messaging.markRead('user_1', ['evt_pin']); + expect(result).toMatchObject({ success: true, readCount: 1 }); + + const receipts = inserted.filter((i) => i.object === RECEIPT_OBJECT); + expect(receipts).toHaveLength(1); + // Identity, not presence: the receipt must carry the SAME organization + // the notification does, not merely some organization. + expect(receipts[0].row.organization_id).toBe('org_pin_beta'); + }); + + it('PIN E2 (over-denial control): an org-less notification still yields a receipt', async () => { + // The single-tenant / no-organization deployment. Nothing is refused and + // nothing is invented — the receipt simply carries the same null its + // notification does. + const { engine, inserted } = engineWithNotification(null); + const messaging = new MessagingService({ logger: silentLogger(), getData: () => engine }); + + const result = await messaging.markRead('user_1', ['evt_pin']); + expect(result).toMatchObject({ success: true, readCount: 1 }); + + const receipts = inserted.filter((i) => i.object === RECEIPT_OBJECT); + expect(receipts).toHaveLength(1); + expect(receipts[0].row.organization_id ?? null).toBeNull(); + }); +});