Skip to content

Commit ffbb7a1

Browse files
claude[bot]os-samclaude
authored
fix(messaging): stamp organization_id on flow-produced notifications and markRead receipts (#11698)
`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. Claude-Session: https://claude.ai/code/session_01APWX2AwT3a4xDcjPCe8bk4 Co-authored-by: os-sam <sam@objectstack.ai> Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
1 parent 6da3601 commit ffbb7a1

5 files changed

Lines changed: 486 additions & 0 deletions

File tree

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
---
2+
'@objectstack/service-automation': patch
3+
'@objectstack/service-messaging': patch
4+
---
5+
6+
Stamp `organization_id` on flow-produced notifications and on `markRead`
7+
receipts, so the notification family stops writing org-less rows
8+
9+
An application project's read-only inventory found `sys_inbox_message`,
10+
`sys_notification`, `sys_notification_receipt` and `sys_notification_delivery`
11+
carrying `organization_id = NULL` on **100%** of their rows — existing rows and
12+
same-day new ones alike, while `sys_approval_request` in the same database
13+
carried an organization on every row. Ruled a gap, not a design choice.
14+
15+
Everything below the messaging ingress was already threaded: `emit()` stamps the
16+
`sys_notification` event, the inbox channel stamps `sys_inbox_message` and its
17+
`delivered` receipt, and the outbox carries the value onto
18+
`sys_notification_delivery`. Each of them reads `EmitInput.organizationId`
19+
and the `notify` flow node, the dominant producer, never supplied it. Its local
20+
structural mirror of `emit()` did not even declare the field, so the value could
21+
not have been passed. One missing argument, four tables at 100% null.
22+
23+
The node now threads the organization from the run's own acting context
24+
(`AutomationContext.tenantId`), the same source the `collab.mention` producer in
25+
`@objectstack/plugin-audit` already uses, so the two notification producers agree
26+
about whose organization a notification carries.
27+
28+
A second producer of the same table is fixed alongside it: the `read` receipt
29+
`markRead` inserts — written when a user reads a notification whose delivered
30+
receipt never landed — named no organization at all. It now carries the
31+
organization of the `sys_notification` row it is about.
32+
33+
There is deliberately **no fallback limb** in either producer: not "the current
34+
organization", not the install's first organization, not the recipient's first
35+
membership. A run with no organization in scope still emits and still writes its
36+
rows, and the `notify` node warns audibly naming the topic and the consequence.
37+
A wrong `organization_id` is worse than a null — a null is visibly missing,
38+
while a wrong value is silently authoritative to every report, export and
39+
cleanup script that filters by organization.
40+
41+
Forward-stamping only. Existing rows are not backfilled and no migration ships.

packages/services/service-automation/src/builtin/notify-node.ts

Lines changed: 56 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,17 @@ export interface MessagingServiceSurface {
2525
dedupKey?: string;
2626
source?: { object: string; id: string };
2727
actorId?: string;
28+
/**
29+
* [#11303] The organization the notification belongs to — the field the
30+
* whole downstream chain stamps from. `MessagingService.writeEvent`
31+
* puts it on `sys_notification`, the inbox channel puts it on
32+
* `sys_inbox_message` and on the `delivered` receipt, and the outbox
33+
* carries it onto the `sys_notification_delivery` row. It was missing
34+
* from this structural mirror, so the node could not have passed it
35+
* even if it had tried: four tables landed 100% org-less on every
36+
* flow-produced notification.
37+
*/
38+
organizationId?: string;
2839
channels?: string[];
2940
}): Promise<{
3041
notificationId: string;
@@ -312,6 +323,46 @@ export function registerNotifyNode(engine: AutomationEngine, ctx: PluginContext)
312323
};
313324
}
314325

326+
// [#11303] The organization this notification belongs to, THREADED
327+
// from the run's own acting context — never fabricated.
328+
//
329+
// Maintainer ruling, 2026-08-24, verbatim: 「11303
330+
// sys_inbox_message/sys_notification/sys_email 应该写
331+
// organization_id。」 — a gap, not a design choice, and the
332+
// PRODUCERS are the fix site. Everything below `emit()` was already
333+
// threaded; this node was the origin that never supplied a value.
334+
//
335+
// `AutomationContext.tenantId` is the acting run's organization —
336+
// the same source `audit-writers.ts` hands its own `collab.mention`
337+
// emit, so the two notification producers agree about whose
338+
// organization a notification carries.
339+
//
340+
// ⛔ There is deliberately NO fallback limb here — not "the current
341+
// organization", not the first organization on the install, not the
342+
// recipient's first membership. A wrong `organization_id` is worse
343+
// than a null: a null is visibly missing, while a wrong value is
344+
// silently authoritative to every report, export and cleanup script
345+
// that filters by organization. When the run carries no
346+
// organization, the notification carries none and says so (below).
347+
const organizationId = toStr(context.tenantId);
348+
if (!organizationId) {
349+
// Fail-LOUD, not fail-guess — and deliberately not fail-CLOSED.
350+
// Refusing here would break the two deployments that legitimately
351+
// have no organization to thread: a `single`-posture install, and
352+
// every stack before its first organization exists. So the
353+
// org-less write stays permitted and becomes a VISIBLE event
354+
// instead of a silent one.
355+
ctx.logger.warn(
356+
`[notify] no organization in scope for topic '${topic ?? 'notify'}' — the ` +
357+
`sys_notification / sys_inbox_message / sys_notification_receipt / ` +
358+
`sys_notification_delivery rows for this emit will carry organization_id = NULL ` +
359+
`and will be invisible to any report or cleanup that filters by organization. ` +
360+
`On a multi-organization install this means the triggering context lost its ` +
361+
`tenant: give the flow's trigger an acting organization (AutomationContext.tenantId). ` +
362+
`On a single-organization install this is expected and can be ignored.`,
363+
);
364+
}
365+
315366
try {
316367
// ADR-0030 single ingress: hand the messaging service a topic +
317368
// audience + payload; it writes the L2 event and materializes
@@ -338,6 +389,11 @@ export function registerNotifyNode(engine: AutomationEngine, ctx: PluginContext)
338389
severity,
339390
source,
340391
actorId,
392+
// [#11303] Absent (not null) when the run has no
393+
// organization: `EmitInput.organizationId` is optional, and
394+
// the chain below normalizes a missing value to NULL exactly
395+
// once, in `writeEvent`.
396+
...(organizationId ? { organizationId } : {}),
341397
channels: channels.length ? channels : undefined,
342398
});
343399
const delivered = Number(result.delivered) || 0;
Lines changed: 247 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,247 @@
1+
// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license.
2+
3+
import { describe, it, expect } from 'vitest';
4+
import {
5+
MessagingService,
6+
MemoryNotificationOutbox,
7+
createInboxChannel,
8+
INBOX_OBJECT,
9+
RECEIPT_OBJECT,
10+
NOTIFICATION_EVENT_OBJECT,
11+
} from '@objectstack/service-messaging';
12+
import { AutomationEngine } from '../engine.js';
13+
import { registerNotifyNode } from './notify-node.js';
14+
import type { MessagingServiceSurface } from './notify-node.js';
15+
16+
/**
17+
* [#11303] The `notify` node is the producer that decides whether the whole
18+
* notification family carries an organization.
19+
*
20+
* Maintainer ruling, 2026-08-24, verbatim: 「11303
21+
* sys_inbox_message/sys_notification/sys_email 应该写 organization_id。」 — a
22+
* GAP, not a design choice.
23+
*
24+
* The measurement that shapes these pins: the messaging chain BELOW `emit()`
25+
* already threads an organization end to end — `writeEvent` stamps
26+
* `organization_id` on `sys_notification`, the inbox channel stamps it on
27+
* `sys_inbox_message` AND on the `delivered` receipt, and the outbox carries it
28+
* onto the `sys_notification_delivery` row. Every one of those reads
29+
* `notification.organizationId`, which is `EmitInput.organizationId`. The break
30+
* is at the ORIGIN: the `notify` node never passes it, so a flow-produced
31+
* notification lands org-less in four tables at once — which is exactly the
32+
* 100%-null reading the card reports for all four.
33+
*
34+
* ⭐ The organization is THREADED from the run's own acting context
35+
* (`AutomationContext.tenantId`), never fabricated. There is deliberately no
36+
* "first organization" / "the current organization" fallback: a wrong
37+
* `organization_id` is worse than a null, because a null is visibly missing
38+
* while a wrong value is silently authoritative to every report, export and
39+
* cleanup script that filters by organization.
40+
*/
41+
42+
function silentLogger(): any {
43+
const l: any = { info: () => {}, warn: () => {}, error: () => {}, debug: () => {} };
44+
l.child = () => l;
45+
return l;
46+
}
47+
48+
/** A logger that records every warning line, for the fail-loud pin. */
49+
function recordingLogger(): { logger: any; warnings: string[] } {
50+
const warnings: string[] = [];
51+
const l: any = {
52+
info: () => {},
53+
warn: (...args: unknown[]) => { warnings.push(args.map(String).join(' ')); },
54+
error: () => {},
55+
debug: () => {},
56+
};
57+
l.child = () => l;
58+
return { logger: l, warnings };
59+
}
60+
61+
/** Every row this stack wrote, in insertion order, with the object it landed in. */
62+
interface WrittenRow { object: string; row: Record<string, unknown> }
63+
64+
/**
65+
* A capturing data engine. Reads answer empty (no preference rows, no dedup
66+
* hit) so the default always-on `inbox` channel is the one that runs; writes
67+
* are recorded verbatim, which is the only thing these pins assert on.
68+
*/
69+
function capturingEngine(): { engine: any; written: WrittenRow[] } {
70+
const written: WrittenRow[] = [];
71+
let seq = 0;
72+
const engine = {
73+
async insert(object: string, row: Record<string, unknown>) {
74+
written.push({ object, row: { ...row } });
75+
const id = row.id != null ? String(row.id) : `row_${++seq}`;
76+
return { ...row, id };
77+
},
78+
async find() { return []; },
79+
async findOne() { return undefined; },
80+
};
81+
return { engine, written };
82+
}
83+
84+
/** The four tables the ruling names for the notification family. */
85+
const NOTIFICATION_FAMILY = new Set<string>([
86+
NOTIFICATION_EVENT_OBJECT,
87+
INBOX_OBJECT,
88+
RECEIPT_OBJECT,
89+
]);
90+
91+
/**
92+
* The identity list this suite asserts on — `object:organization_id` per row,
93+
* in write order. ⭐ Identities, not a count: an offsetting error (one row
94+
* gaining an organization while another loses it) holds a count constant while
95+
* the identity list inverts.
96+
*/
97+
function orgIdentities(written: WrittenRow[]): string[] {
98+
return written
99+
.filter((w) => NOTIFICATION_FAMILY.has(w.object))
100+
.map((w) => `${w.object}:${w.row.organization_id ?? 'NULL'}`);
101+
}
102+
103+
function notifyFlow(): any {
104+
return {
105+
name: 'nudge',
106+
label: 'Nudge',
107+
type: 'autolaunched' as const,
108+
nodes: [
109+
{ id: 'start', type: 'start' as const, label: 'Start' },
110+
{
111+
id: 'notify',
112+
type: 'notify' as const,
113+
label: 'Notify',
114+
config: {
115+
topic: 'deal.won',
116+
recipients: ['user_1'],
117+
title: 'Renewal due',
118+
message: 'Ping',
119+
channels: ['inbox'],
120+
},
121+
},
122+
{ id: 'end', type: 'end' as const, label: 'End' },
123+
],
124+
edges: [
125+
{ id: 'e1', source: 'start', target: 'notify' },
126+
{ id: 'e2', source: 'notify', target: 'end' },
127+
],
128+
};
129+
}
130+
131+
/**
132+
* The REAL messaging service with the REAL inbox channel behind the notify
133+
* node — the seam under test is precisely the handoff between them, so a fake
134+
* that answers `emit()` in one shot could not express it.
135+
*/
136+
function bootInlineStack(logger: any = silentLogger()) {
137+
const { engine: data, written } = capturingEngine();
138+
const messaging = new MessagingService({ logger, getData: () => data });
139+
messaging.registerChannel(createInboxChannel({ getData: () => data }));
140+
141+
const engine = new AutomationEngine(logger);
142+
registerNotifyNode(engine, {
143+
logger,
144+
getService: (name: string) => (name === 'messaging' ? messaging : undefined),
145+
} as any);
146+
engine.registerFlow('nudge', notifyFlow());
147+
return { engine, messaging, written };
148+
}
149+
150+
describe('#11303 — the notify producer stamps organization_id on the notification family', () => {
151+
it('PIN A: threads the run\'s own organization onto the emit input', async () => {
152+
const emitted: any[] = [];
153+
const service: MessagingServiceSurface = {
154+
async emit(n: any) {
155+
emitted.push(n);
156+
return { notificationId: 'evt_1', delivered: n.audience.length, failed: 0 };
157+
},
158+
};
159+
const engine = new AutomationEngine(silentLogger());
160+
registerNotifyNode(engine, {
161+
logger: silentLogger(),
162+
getService: (name: string) => (name === 'messaging' ? service : undefined),
163+
} as any);
164+
engine.registerFlow('nudge', notifyFlow());
165+
166+
const run = await engine.execute('nudge', { tenantId: 'org_pin_alpha' } as any);
167+
168+
expect(run.success).toBe(true);
169+
expect(emitted).toHaveLength(1);
170+
// The named producer pin: the organization reaching `emit()` is the
171+
// run's acting tenant, verbatim — not a derived or defaulted value.
172+
expect(emitted[0].organizationId).toBe('org_pin_alpha');
173+
});
174+
175+
it('PIN B: a run under an organization writes ZERO org-less rows into the notification family', async () => {
176+
const { engine, written } = bootInlineStack();
177+
178+
const run = await engine.execute('nudge', { tenantId: 'org_pin_alpha' } as any);
179+
expect(run.success).toBe(true);
180+
181+
// The end-to-end pin the ruling names, asserted as an IDENTITY list so a
182+
// producer nobody enumerated cannot hide behind a stable count.
183+
expect(orgIdentities(written)).toEqual([
184+
`${NOTIFICATION_EVENT_OBJECT}:org_pin_alpha`,
185+
`${INBOX_OBJECT}:org_pin_alpha`,
186+
`${RECEIPT_OBJECT}:org_pin_alpha`,
187+
]);
188+
// Said the second way, so the pin still bites if the write ORDER changes:
189+
// no row of the family may carry NULL.
190+
expect(orgIdentities(written).filter((i) => i.endsWith(':NULL'))).toEqual([]);
191+
});
192+
193+
it('PIN B2: the durable delivery row carries the same organization', async () => {
194+
const { engine: data } = capturingEngine();
195+
const outbox = new MemoryNotificationOutbox(1);
196+
const messaging = new MessagingService({ logger: silentLogger(), getData: () => data, outbox });
197+
messaging.registerChannel(createInboxChannel({ getData: () => data }));
198+
const engine = new AutomationEngine(silentLogger());
199+
registerNotifyNode(engine, {
200+
logger: silentLogger(),
201+
getService: (name: string) => (name === 'messaging' ? messaging : undefined),
202+
} as any);
203+
engine.registerFlow('nudge', notifyFlow());
204+
205+
const run = await engine.execute('nudge', { tenantId: 'org_pin_alpha' } as any);
206+
expect(run.success).toBe(true);
207+
208+
const rows = await outbox.list();
209+
expect(rows).toHaveLength(1);
210+
expect(rows[0].organizationId).toBe('org_pin_alpha');
211+
});
212+
213+
it('PIN C (over-denial control): a stack with no organization in scope still delivers', async () => {
214+
// The control that stops the fix from degenerating into "refuse unless
215+
// an organization is present". A `single`-posture deployment — and every
216+
// fresh boot before the first organization exists — has no organization
217+
// to thread, and a notify there must still emit and still write its rows.
218+
// ⭐ A suite that only pinned "organization_id is present" would score
219+
// green on an implementation that breaks exactly this deployment.
220+
const { engine, written } = bootInlineStack();
221+
222+
const run = await engine.execute('nudge');
223+
224+
expect(run.success).toBe(true);
225+
expect(orgIdentities(written)).toEqual([
226+
`${NOTIFICATION_EVENT_OBJECT}:NULL`,
227+
`${INBOX_OBJECT}:NULL`,
228+
`${RECEIPT_OBJECT}:NULL`,
229+
]);
230+
});
231+
232+
it('PIN D (fail-loud, not fail-guess): an unresolvable organization warns audibly', async () => {
233+
// Fail-LOUD by warning rather than refusing — see PIN C for why a
234+
// refusal is not available here. The warning is what makes the org-less
235+
// row a visible event instead of a silent one, and it must name the
236+
// topic so the operator can find the producer.
237+
const { logger, warnings } = recordingLogger();
238+
const { engine } = bootInlineStack(logger);
239+
240+
const run = await engine.execute('nudge');
241+
242+
expect(run.success).toBe(true);
243+
const line = warnings.find((w) => w.includes('organization'));
244+
expect(line, `no organization warning in: ${JSON.stringify(warnings)}`).toBeDefined();
245+
expect(line).toContain('deal.won');
246+
});
247+
});

0 commit comments

Comments
 (0)