fix(messaging): stamp organization_id on flow-produced notifications and markRead receipts

.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.

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;

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<string, unknown> }
+
+/**
+ * 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<string, unknown>) {
+ 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<string>([
+ 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');
+ });
+});