From fc808012c86a7b86eb2b69ffd1ea9eeaecd252de Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 13 Sep 2026 13:13:26 +0000 Subject: [PATCH 1/5] wip(service-messaging): channel availability seam + suppression record Claude-Session: https://claude.ai/code/session_01URLHobLUJB9K1ABV6ofdjj Co-authored-by: Claude --- .../src/audit/sys-notification.object.ts | 28 ++++ .../services/service-messaging/src/channel.ts | 65 ++++++++ .../service-messaging/src/email-channel.ts | 40 +++++ .../services/service-messaging/src/index.ts | 8 + .../src/messaging-service.ts | 144 ++++++++++++++++-- 5 files changed, 276 insertions(+), 9 deletions(-) diff --git a/packages/platform-objects/src/audit/sys-notification.object.ts b/packages/platform-objects/src/audit/sys-notification.object.ts index e52f65f9ba..becf15150f 100644 --- a/packages/platform-objects/src/audit/sys-notification.object.ts +++ b/packages/platform-objects/src/audit/sys-notification.object.ts @@ -101,6 +101,34 @@ export const SysNotification = ObjectSchema.create({ group: 'Event', }), + // [#17732] Channels fan-out did not even attempt, and why. + // + // A channel the tenant cannot send on (no transport configured) used to get + // one `sys_notification_delivery` row per recipient that dead-lettered on + // its first attempt. Fan-out now asks the channel first + // (`MessagingChannel.isAvailable`) and writes no delivery row at all; the + // fact is recorded HERE instead, so the suppression stays auditable at the + // event level rather than disappearing. + // + // Value: `[{ channel, reason }]`, NULL when nothing was suppressed (the + // overwhelmingly common path). JSON rather than a `select` because one + // event fans out to several channels and each carries its OWN reason — a + // scalar column would have to drop either the channel or the reason. + // + // `reason` is a CLOSED set, inlined here: `transport_not_configured`. + // `packages/platform-objects` is a lower layer than service-messaging and + // cannot import its `CHANNEL_UNAVAILABLE_REASONS`, so the two copies are + // held equal by an executable assertion in + // `packages/services/service-messaging/src/channel-availability.test.ts` + // — ⛔ a comment is not what keeps them in step. + suppressed_channels: Field.json({ + label: 'Suppressed Channels', + required: false, + description: + 'Channels fan-out skipped because they are unavailable for this tenant, as [{channel, reason}]; reason is the closed set: transport_not_configured', + group: 'Event', + }), + dedup_key: Field.text({ label: 'Dedup Key', required: false, diff --git a/packages/services/service-messaging/src/channel.ts b/packages/services/service-messaging/src/channel.ts index 7b2b4be3b8..5a26f30800 100644 --- a/packages/services/service-messaging/src/channel.ts +++ b/packages/services/service-messaging/src/channel.ts @@ -97,6 +97,42 @@ export interface SendResult { readonly error?: string; } +/** + * Why a channel is not available for a tenant. + * + * A SMALL CLOSED SET, deliberately: it is recorded on `sys_notification`, so + * every value here is a column value an operator filters and reports on. The + * same literals are inlined on the object's `suppressed_channels` field + * description (`packages/platform-objects/src/audit/sys-notification.object.ts`) + * — `packages/platform-objects` is a LOWER layer than this package and cannot + * import from it, so the two copies are held equal by an executable assertion + * (`channel-availability.test.ts`) rather than by a shared import. + * + * ⛔ Not a free-text field: a reason nobody declared is a reason nothing can + * aggregate. + */ +export const CHANNEL_UNAVAILABLE_REASONS = ['transport_not_configured'] as const; + +/** One value of {@link CHANNEL_UNAVAILABLE_REASONS}. */ +export type ChannelUnavailableReason = (typeof CHANNEL_UNAVAILABLE_REASONS)[number]; + +/** + * The tenant context an availability query is answered against. + * + * Carries the organization and nothing else: availability is a property of + * `(tenant × channel)`, not of a recipient or a topic, which is what lets + * fan-out ask ONCE PER CHANNEL PER EMIT instead of once per delivery. + */ +export interface ChannelAvailabilityQuery { + /** Tenant whose configuration decides the answer; absent on single-tenant / background emits. */ + readonly organizationId?: string; +} + +/** A channel's answer to {@link MessagingChannel.isAvailable}. */ +export type ChannelAvailability = + | { readonly available: true } + | { readonly available: false; readonly reason: ChannelUnavailableReason }; + /** Minimal context handed to a channel — just a logger in M1. */ export interface MessagingChannelContext { readonly logger: { @@ -119,4 +155,33 @@ export interface MessagingChannel { /** Optional: classify a thrown error for the (future) outbox. */ classifyError?(err: unknown): ErrorClass; + + /** + * Optional: can this tenant send on this channel AT ALL right now? + * + * Consulted by fan-out BEFORE any `sys_notification_delivery` row is + * written. A channel that answers `{ available: false }` gets no delivery + * rows for that emit; the suppression and its reason are recorded on the + * `sys_notification` event instead. Without it, every such row is work the + * pipeline is guaranteed to fail at — a delivery record that exists only to + * dead-letter. + * + * OPTIONAL, and absence means AVAILABLE — today's behaviour, unchanged, for + * every channel implementation that never heard of this member. ⛔ That + * default is not a convenience: inverting it would silently mute every + * channel a third party ships. + * + * This is a PRE-SEND fact, not a send attempt: it must not perform the + * delivery's own I/O, and it is called once per channel per emit, never per + * recipient. A channel whose answer is expensive is responsible for its own + * caching — fan-out holds none, because a cache here serves a stale answer + * across a live configuration change. + * + * A throw is treated as AVAILABLE (fail-open) and logged: an availability + * probe that breaks must not become a notification outage. + */ + isAvailable?( + ctx: MessagingChannelContext, + query: ChannelAvailabilityQuery, + ): ChannelAvailability | Promise; } diff --git a/packages/services/service-messaging/src/email-channel.ts b/packages/services/service-messaging/src/email-channel.ts index bf68a6a5c7..8d779eab5f 100644 --- a/packages/services/service-messaging/src/email-channel.ts +++ b/packages/services/service-messaging/src/email-channel.ts @@ -2,6 +2,8 @@ import type { IDataEngine } from '@objectstack/spec/contracts'; import type { + ChannelAvailability, + ChannelAvailabilityQuery, Delivery, ErrorClass, MessagingChannel, @@ -205,6 +207,44 @@ export function createEmailChannel(opts: EmailChannelOptions): MessagingChannel return { id: 'email', + /** + * Can this tenant send mail at all? (ruling on #17732 item 3.) + * + * Answered from the composition's transport configuration — the `email` + * service this channel was handed — and from nothing else. No delivery + * I/O, no recipient lookup, no settings read: a service-registry + * closure call, which is why fan-out consults it inline and holds no + * cache. + * + * ## The cost measurement the ruling asked for + * + * Mail configuration in this tree is the `mail` settings namespace, and + * that manifest declares `scope: 'global'` + * (`packages/services/service-settings/src/manifests/mail.manifest.ts`): + * one deployment-wide provider/transport, materialised ONCE into a + * single in-memory `IEmailTransport` on the `EmailService` and + * hot-swapped by the settings change bus (`EmailServicePlugin`'s + * `applySettings` → `setTransport`). So there is no per-tenant + * transport row to read, and the answer costs no round trip at all. + * + * ⇒ NO CACHE, for two independent reasons: it would save nothing, and + * it would be WRONG — a tick-scoped memo would keep answering + * "unavailable" straight through the settings save that fixed it. + * + * The query still takes the tenant context, because the seam outlives + * this measurement: the day mail configuration becomes tenant-scoped, + * the answer changes here and no published interface has to move again. + */ + isAvailable(_ctx: MessagingChannelContext, _query: ChannelAvailabilityQuery): ChannelAvailability { + // A pure probe: no logging, no I/O, no side effects. The suppression + // it causes is announced ONCE per emit by the service and recorded + // durably on `sys_notification.suppressed_channels` — a line per + // emit here would be the noisy half of an answer already written down. + return opts.getEmail() + ? { available: true } + : { available: false, reason: 'transport_not_configured' }; + }, + async send(ctx: MessagingChannelContext, delivery: Delivery): Promise { const email = opts.getEmail(); if (!email) { diff --git a/packages/services/service-messaging/src/index.ts b/packages/services/service-messaging/src/index.ts index 07ea1a05ab..6f0c647627 100644 --- a/packages/services/service-messaging/src/index.ts +++ b/packages/services/service-messaging/src/index.ts @@ -21,6 +21,7 @@ export type { MessagingServicePluginOptions } from './messaging-service-plugin.j // Service + types export { MessagingService, NOTIFICATION_EVENT_OBJECT } from './messaging-service.js'; export type { + ChannelSuppression, DeliveryOutcome, EmitResult, EmitInput, @@ -91,7 +92,14 @@ export type { Delivery, SendResult, ErrorClass, + // [#17732] The optional per-tenant availability query and its closed + // reason vocabulary — fan-out writes no delivery row for a channel that + // answers unavailable. + ChannelAvailability, + ChannelAvailabilityQuery, + ChannelUnavailableReason, } from './channel.js'; +export { CHANNEL_UNAVAILABLE_REASONS } from './channel.js'; // Reliable delivery — outbox + dispatcher (ADR-0030 P1) export type { diff --git a/packages/services/service-messaging/src/messaging-service.ts b/packages/services/service-messaging/src/messaging-service.ts index 276a8cff71..ebf3ed1a24 100644 --- a/packages/services/service-messaging/src/messaging-service.ts +++ b/packages/services/service-messaging/src/messaging-service.ts @@ -3,6 +3,8 @@ import type { IDataEngine } from '@objectstack/spec/contracts'; import { isUniqueViolationError } from '@objectstack/types'; import type { + ChannelAvailability, + ChannelUnavailableReason, MessagingChannel, MessagingChannelContext, Notification, @@ -126,6 +128,29 @@ export interface EmitResult { */ readonly enqueued: number; readonly failed: number; + /** + * Channels this emit deliberately did NOT fan out to, because the channel + * answered `{ available: false }` for the tenant (ruling on #17732). + * + * Empty on every path where nothing was suppressed — including an emit + * whose channels do not implement the availability member at all, which is + * the common case and stays byte-identical to the pre-#17732 behaviour. + * + * It is here because the alternative is a result that LIES: a two-channel + * emit that enqueued one row would otherwise report `enqueued: 1` with + * nothing saying the second channel was dropped, which is indistinguishable + * from a fan-out bug. The durable record of the same fact is + * `sys_notification.suppressed_channels`. + */ + readonly suppressed: readonly ChannelSuppression[]; +} + +/** One channel a fan-out skipped, with the channel's own reason. */ +export interface ChannelSuppression { + /** The channel id that was not fanned out to (e.g. `email`). */ + readonly channel: string; + /** Why — the closed vocabulary in `CHANNEL_UNAVAILABLE_REASONS`. */ + readonly reason: ChannelUnavailableReason; } /** Context the service needs: a logger, plus data access for the L2 event. */ @@ -857,6 +882,13 @@ export class MessagingService { * never aborts the rest of the fan-out. A `dedupKey` that matches an * existing event short-circuits: the event id is returned and no new * deliveries are produced. + * + * A channel that answers `isAvailable: { available: false }` for the tenant + * is dropped BEFORE anything is written for it (ruling on #17732): no + * `sys_notification_delivery` row, no `send()` call, and the suppression + * plus its reason recorded on the `sys_notification` event row instead. + * Every other channel — including every channel that does not implement the + * member — fans out exactly as before. */ async emit(input: EmitInput): Promise { const data = this.ctx.getData?.(); @@ -868,10 +900,23 @@ export class MessagingService { this.ctx.logger.info( `[messaging] emit: dedupKey '${input.dedupKey}' already emitted (${existing}); skipping`, ); - return { notificationId: existing, deduped: true, deliveries: [], delivered: 0, enqueued: 0, failed: 0 }; + return { notificationId: existing, deduped: true, deliveries: [], delivered: 0, enqueued: 0, failed: 0, suppressed: [] }; } } + // 1b) Channel availability (ruling on #17732 item 2). Asked ONCE PER + // CHANNEL — the answer is a property of (tenant × channel), not of a + // recipient — and asked HERE, before the event write, so the + // suppression lands in the SAME insert that creates the event row. + // That is what keeps the whole feature at zero extra writes: no + // follow-up `update` on `sys_notification`, and one fewer delivery + // row for every channel that could only have dead-lettered. + const requestedChannels = input.channels?.length ? input.channels : ['inbox']; + const { available: channels, suppressed } = await this.resolveChannelAvailability( + requestedChannels, + input.organizationId, + ); + // 2) Write the L2 event (or synthesize an id when there is no data layer). // The check at (1) is a fast-path. Where the driver materializes the // UNIQUE(dedup_key) index, it is the real guard: a concurrent emit @@ -883,7 +928,7 @@ export class MessagingService { // best-effort fast-path — the catch is then simply never taken.) let notificationId: string; try { - notificationId = await this.writeEvent(data, input); + notificationId = await this.writeEvent(data, input, suppressed); } catch (err) { if (input.dedupKey && data) { const winner = await this.findEventByDedupKey(data, input.dedupKey); @@ -891,7 +936,7 @@ export class MessagingService { this.ctx.logger.info( `[messaging] emit: dedupKey '${input.dedupKey}' raced; converged to ${winner}`, ); - return { notificationId: winner, deduped: true, deliveries: [], delivered: 0, enqueued: 0, failed: 0 }; + return { notificationId: winner, deduped: true, deliveries: [], delivered: 0, enqueued: 0, failed: 0, suppressed: [] }; } } throw err; @@ -904,13 +949,24 @@ export class MessagingService { }); if (recipients.length === 0) { this.ctx.logger.warn(`[messaging] emit: topic '${input.topic}' resolved to 0 recipients`); - return { notificationId, deduped: false, deliveries: [], delivered: 0, enqueued: 0, failed: 0 }; + return { notificationId, deduped: false, deliveries: [], delivered: 0, enqueued: 0, failed: 0, suppressed }; + } + + // 3a) Every requested channel is unavailable for this tenant. The event + // is already written and carries the reasons — there is simply + // nothing to enqueue. Said once, at `info`: the loss is FUNCTIONAL + // and fully visible on the event row, not silent data loss. + if (channels.length === 0) { + this.ctx.logger.info( + `[messaging] emit: topic '${input.topic}' has no available channel ` + + `(${suppressed.map((s) => `${s.channel}: ${s.reason}`).join(', ')}); no delivery rows written`, + ); + return { notificationId, deduped: false, deliveries: [], delivered: 0, enqueued: 0, failed: 0, suppressed }; } // 3b) Preference filter (ADR-0030 P2): drop the (recipient × channel) // pairs the user muted. Mandatory topics bypass; fail-open on error. const payload = input.payload ?? {}; - const channels = input.channels?.length ? input.channels : ['inbox']; const targets = await this.preferences.filter(recipients, channels, { topic: input.topic, organizationId: input.organizationId, @@ -918,7 +974,7 @@ export class MessagingService { }); if (targets.length === 0) { this.ctx.logger.info(`[messaging] emit: topic '${input.topic}' suppressed for all recipients by preference`); - return { notificationId, deduped: false, deliveries: [], delivered: 0, enqueued: 0, failed: 0 }; + return { notificationId, deduped: false, deliveries: [], delivered: 0, enqueued: 0, failed: 0, suppressed }; } // 4) Either enqueue durable deliveries (P1 outbox) or fan out inline (P0). @@ -938,6 +994,7 @@ export class MessagingService { return { notificationId, deduped: false, deliveries, delivered: 0, enqueued, failed: deliveries.length - enqueued, + suppressed, }; } @@ -949,7 +1006,11 @@ export class MessagingService { body: str(payload.body) ?? '', severity: input.severity ?? 'info', recipients, - channels: input.channels, + // The channels that SURVIVED the availability consult — a channel + // that answered unavailable is not advertised to the ones that did. + // Absent stays absent: a producer that named no channels still gets + // `undefined` here, exactly as before. + channels: input.channels ? channels : undefined, actionUrl: actionUrlFor(input, payload), // Who caused it, projected onto the per-recipient unit so a channel // can materialize it without reading `sys_notification` back. @@ -960,7 +1021,63 @@ export class MessagingService { // Inline (P0): every channel has already answered, so `delivered` is a // real terminal count and nothing is left in flight. const { deliveries, delivered, failed } = await this.fanOut(notification, targets); - return { notificationId, deduped: false, deliveries, delivered, enqueued: 0, failed }; + return { notificationId, deduped: false, deliveries, delivered, enqueued: 0, failed, suppressed }; + } + + /** + * Split the requested channels into the ones that can actually send for + * this tenant and the ones that answered they cannot (ruling on #17732). + * + * Three deliberate properties: + * + * 1. **A channel with no `isAvailable` is AVAILABLE.** The member is + * optional precisely so every channel implementation that predates it — + * ours and anyone else's — keeps working unchanged. Inverting this + * default would mute every channel that has not been updated, which is a + * far worse failure than the workless rows this exists to stop. + * `channel-availability.test.ts` pins it from both sides. + * 2. **An UNREGISTERED channel is left alone.** It has no implementation to + * ask, so it keeps today's path exactly: the inline fan-out reports it as + * a failed delivery, the outbox enqueues a row the dispatcher + * dead-letters. That is a real, separate defect — it is filed, ⛔ not + * widened into this ruling. + * 3. **A throw is AVAILABLE.** Fail-open, matching the preference filter one + * step down: a broken probe must degrade into today's behaviour, never + * into a silent notification outage. Logged at `warn` — the degradation + * is functional and self-announcing (the delivery still happens), not a + * durability loss. + * + * Called once per channel per emit. ⛔ No cache: the probe is answered from + * in-memory composition state, and a tick-scoped cache here would serve a + * stale answer straight through a live settings change. + */ + private async resolveChannelAvailability( + requested: readonly string[], + organizationId: string | undefined, + ): Promise<{ available: string[]; suppressed: ChannelSuppression[] }> { + const available: string[] = []; + const suppressed: ChannelSuppression[] = []; + for (const id of requested) { + const channel = this.channels.get(id); + if (!channel?.isAvailable) { + available.push(id); + continue; + } + let answer: ChannelAvailability; + try { + answer = await channel.isAvailable(this.ctx, { organizationId }); + } catch (err) { + this.ctx.logger.warn( + `[messaging] channel '${id}' availability probe threw (${(err as Error)?.message ?? String(err)}); ` + + `treating it as available and fanning out as before`, + ); + available.push(id); + continue; + } + if (answer.available) available.push(id); + else suppressed.push({ channel: id, reason: answer.reason }); + } + return { available, suppressed }; } /** @@ -1035,7 +1152,11 @@ export class MessagingService { * Persist the L2 event and return its id. With no data layer (minimal/test * stacks) we warn and synthesize an id so fan-out can still be exercised. */ - private async writeEvent(data: IDataEngine | undefined, input: EmitInput): Promise { + private async writeEvent( + data: IDataEngine | undefined, + input: EmitInput, + suppressed: readonly ChannelSuppression[], + ): Promise { if (!data) { this.ctx.logger.warn('[messaging] no data engine registered; event not persisted'); return `evt_${Math.random().toString(36).slice(2)}`; @@ -1052,6 +1173,11 @@ export class MessagingService { source_id: str(input.source?.id) ?? null, actor_id: input.actorId ?? null, organization_id: input.organizationId ?? null, + // [#17732] The channels fan-out will not even attempt, and why. + // NULL — not `[]` — when nothing was suppressed, so the column is + // empty on the overwhelmingly common path and a non-null value + // always means "something really was dropped". + suppressed_channels: suppressed.length > 0 ? suppressed.map((s) => ({ ...s })) : null, created_at: this.now(), }; const created = await data.insert(NOTIFICATION_EVENT_OBJECT, row); From da6be49f133f461930d64b5e6976d7b3980adb4b Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 13 Sep 2026 13:25:13 +0000 Subject: [PATCH 2/5] feat(service-messaging,platform-objects): fan-out consults per-tenant channel availability Claude-Session: https://claude.ai/code/session_01URLHobLUJB9K1ABV6ofdjj Co-authored-by: Claude --- .../17732-channel-availability-fanout.md | 14 + .../src/channel-availability.test.ts | 355 ++++++++++++++++++ .../src/email-channel.test.ts | 58 +++ 3 files changed, 427 insertions(+) create mode 100644 .changeset/17732-channel-availability-fanout.md create mode 100644 packages/services/service-messaging/src/channel-availability.test.ts diff --git a/.changeset/17732-channel-availability-fanout.md b/.changeset/17732-channel-availability-fanout.md new file mode 100644 index 0000000000..3ad3fb864f --- /dev/null +++ b/.changeset/17732-channel-availability-fanout.md @@ -0,0 +1,14 @@ +--- +"@objectstack/service-messaging": minor +"@objectstack/platform-objects": minor +--- + +Notification fan-out asks a channel whether the tenant can send on it before writing anything, so a channel with no transport no longer produces `sys_notification_delivery` rows that exist only to dead-letter (#17732). + +`MessagingChannel` gains one **optional** member, `isAvailable(ctx, { organizationId })`, answering `{ available: true }` or `{ available: false, reason }` from the closed vocabulary `CHANNEL_UNAVAILABLE_REASONS` (today: `transport_not_configured`). `emit()` consults it once per channel per emit — availability is a property of `(tenant × channel)`, not of a recipient — and a channel that answers unavailable gets no delivery row and no `send()` call on either the outbox (P1) or the inline (P0) path. + +- **Optional means available.** A channel that does not implement the member is treated exactly as before. Every existing implementation, in this repo and in yours, keeps working unchanged with no edit; the same is true of a channel that is registered but unknown to this version. ⛔ There is no way to configure the opposite default. +- **The suppression is recorded, not swallowed.** `sys_notification` gains one key, `suppressed_channels` — `[{ channel, reason }]`, `NULL` when nothing was suppressed — written in the *same* insert that creates the event row, so the feature costs no additional write. `EmitResult` gains the matching `suppressed` array, so a caller is never handed a delivery count that silently omits a channel it asked for. +- **The `email` channel answers from the transport it was handed** — a service-registry lookup, no I/O, nothing cached. Mail configuration in this tree is the `mail` settings namespace at `scope: 'global'`, materialised into a single in-memory transport that the settings change bus hot-swaps, so there is no per-tenant row to read and a memoized answer would survive the settings save that fixed it. The query still takes the tenant context so a future tenant-scoped transport needs no interface change. +- **A probe that throws is treated as available** and logged at `warn`: a broken availability check degrades into today's behaviour, never into a silent notification outage. +- ⚠️ **Unchanged on purpose**: a channel named in `channels` that is not *registered* at all keeps its existing path — the inline fan-out reports it as a failed delivery, the outbox enqueues a row the dispatcher dead-letters. It has no implementation to ask, and widening this ruling to cover it is filed separately. diff --git a/packages/services/service-messaging/src/channel-availability.test.ts b/packages/services/service-messaging/src/channel-availability.test.ts new file mode 100644 index 0000000000..c00450c33b --- /dev/null +++ b/packages/services/service-messaging/src/channel-availability.test.ts @@ -0,0 +1,355 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// [#17732] Fan-out consults per-tenant channel availability before it writes +// anything — ruling `5644350987` on #17732 (director seat, 2026-09-12). +// +// ## What was wrong +// +// Every `(recipient × channel)` pair got a `sys_notification_delivery` row, and +// a channel the tenant has no transport for produced rows that could only ever +// dead-letter. #17611's C capped how LONG those rows live; this is the cause. +// +// ## Why the three facts below are pinned on ONE pass +// +// Each of them alone is passed by an implementation that is badly broken in a +// different direction, so a suite that separates them proves nothing: +// +// * "the unavailable channel got no row" — also true of a fan-out that +// wrote NOTHING AT ALL. +// * "the available channel got its row" — also true of the old code. +// * "the suppression was recorded" — also true of an implementation +// that records a suppression it never actually performed. +// +// Only the conjunction — in one emit, over one outbox — says the fix +// DISCRIMINATES between the two channels rather than acting on all of them. +// +// ## And the control that matters more than the feature +// +// The ruling's item 1 is load-bearing: the member is OPTIONAL, and "a channel +// that does not implement it is treated as available (today's behaviour)". An +// implementation that inverted that default would pass every suppression test +// in this file while silently muting every channel — ours and every third +// party's — that has not been updated. That is a far worse bug than the one +// being fixed, so it gets its own pin, from both sides. + +import { describe, it, expect } from 'vitest'; +import { SysNotification } from '@objectstack/platform-objects/audit'; +import { MessagingService } from './messaging-service.js'; +import { MemoryNotificationOutbox } from './memory-outbox.js'; +import { CHANNEL_UNAVAILABLE_REASONS } from './channel.js'; +import type { + ChannelAvailability, + ChannelAvailabilityQuery, + Delivery, + MessagingChannel, +} from './channel.js'; + +function silentLogger() { + return { info: () => {}, warn: () => {}, error: () => {} }; +} + +/** + * A data engine double that captures the `sys_notification` insert. + * + * Deliberately implements only what this path reaches — `insert` for the L2 + * event, `find` for the preference lookup. `emit()` without a `dedupKey` and + * with plain user-id recipients performs no `findOne`, and no `update` or + * `delete` at all, so the double carries none: an engine double is only ever + * as wide as the calls it has to answer. + */ +function capturingEngine() { + const inserts: Array<{ object: string; row: Record }> = []; + return { + inserts, + engine: { + async insert(object: string, row: Record) { + inserts.push({ object, row }); + return { id: `evt_${inserts.length}`, ...row }; + }, + async find() { + return []; + }, + } as never, + }; +} + +/** A channel that records what it is handed and, optionally, answers availability. */ +function channelDouble( + id: string, + availability?: ChannelAvailability | (() => ChannelAvailability), +): { channel: MessagingChannel; sent: Delivery[]; probes: ChannelAvailabilityQuery[] } { + const sent: Delivery[] = []; + const probes: ChannelAvailabilityQuery[] = []; + const channel: MessagingChannel = { + id, + async send(_ctx, delivery) { + sent.push(delivery); + return { ok: true }; + }, + }; + if (availability !== undefined) { + (channel as { isAvailable?: unknown }).isAvailable = ( + _ctx: unknown, + query: ChannelAvailabilityQuery, + ): ChannelAvailability => { + probes.push(query); + return typeof availability === 'function' ? availability() : availability; + }; + } + return { channel, sent, probes }; +} + +const UNAVAILABLE: ChannelAvailability = { available: false, reason: 'transport_not_configured' }; + +describe('channel availability at fan-out (#17732)', () => { + it('suppresses the unavailable channel, still enqueues the available one, and records the reason — one pass', async () => { + const outbox = new MemoryNotificationOutbox(1); + const data = capturingEngine(); + const service = new MessagingService({ + logger: silentLogger(), + outbox, + getData: () => data.engine, + }); + + // `inbox` implements NO availability member — the ruling's default, and + // the production shape of the always-available channel. + const inbox = channelDouble('inbox'); + // `email` answers that this tenant has no transport. + const email = channelDouble('email', UNAVAILABLE); + service.registerChannel(inbox.channel); + service.registerChannel(email.channel); + + const result = await service.emit({ + topic: 'deal.won', + audience: ['user_1', 'user_2'], + channels: ['inbox', 'email'], + organizationId: 'org_1', + payload: { title: 'Deal closed' }, + }); + + const rows = await outbox.list(); + + // (1) The unavailable channel got NO delivery row … + expect(rows.filter((r) => r.channel === 'email')).toHaveLength(0); + // (2) … while the available channel in the SAME fan-out got one per + // recipient. Without this, (1) is also satisfied by writing nothing. + expect(rows.filter((r) => r.channel === 'inbox').map((r) => r.recipientId).sort()) + .toEqual(['user_1', 'user_2']); + expect(rows).toHaveLength(2); + expect(result.enqueued).toBe(2); + expect(result.failed).toBe(0); + + // (3) The suppression is recorded on the L2 event, with its reason — in + // the SAME insert that created the event, so the feature costs no + // second write. + expect(data.inserts).toHaveLength(1); + expect(data.inserts[0].object).toBe('sys_notification'); + expect(data.inserts[0].row.suppressed_channels) + .toEqual([{ channel: 'email', reason: 'transport_not_configured' }]); + + // …and the caller is told, so a two-channel emit reporting one row is + // never indistinguishable from a fan-out bug. + expect(result.suppressed).toEqual([{ channel: 'email', reason: 'transport_not_configured' }]); + }); + + it('THE CONTROL: a channel that does not implement isAvailable still gets its delivery row', async () => { + // The whole point of the optional member. An implementation that treated + // "no isAvailable" as unavailable would pass every other test in this + // file and silently mute every channel nobody has updated. + const outbox = new MemoryNotificationOutbox(1); + const data = capturingEngine(); + const service = new MessagingService({ + logger: silentLogger(), + outbox, + getData: () => data.engine, + }); + service.registerChannel(channelDouble('inbox').channel); + service.registerChannel(channelDouble('sms').channel); + + const result = await service.emit({ + topic: 'deal.won', + audience: ['user_1'], + channels: ['inbox', 'sms'], + payload: { title: 'Deal closed' }, + }); + + const rows = await outbox.list(); + expect(rows.map((r) => r.channel).sort()).toEqual(['inbox', 'sms']); + expect(result.enqueued).toBe(2); + expect(result.suppressed).toEqual([]); + // NULL, not `[]` — the column stays empty on the common path, so a + // non-null value always means something really was dropped. + expect(data.inserts[0].row.suppressed_channels).toBeNull(); + }); + + it('THE CONTROL, other side: the same channel suppressed once it DOES answer unavailable', async () => { + // Pairs with the test above on one variable — the presence of the + // member — so "it got its row" cannot be read as "suppression never + // works here". + const outbox = new MemoryNotificationOutbox(1); + const data = capturingEngine(); + const service = new MessagingService({ + logger: silentLogger(), + outbox, + getData: () => data.engine, + }); + service.registerChannel(channelDouble('inbox').channel); + service.registerChannel(channelDouble('sms', UNAVAILABLE).channel); + + await service.emit({ + topic: 'deal.won', + audience: ['user_1'], + channels: ['inbox', 'sms'], + payload: { title: 'Deal closed' }, + }); + + expect((await outbox.list()).map((r) => r.channel)).toEqual(['inbox']); + expect(data.inserts[0].row.suppressed_channels) + .toEqual([{ channel: 'sms', reason: 'transport_not_configured' }]); + }); + + it('asks each channel ONCE per emit and hands it the tenant — not once per recipient', async () => { + // The cost claim, pinned. Availability is a property of (tenant × + // channel), so a probe that ran per delivery would multiply whatever it + // costs by the audience size — the thing the ruling asked to be measured. + const outbox = new MemoryNotificationOutbox(1); + const data = capturingEngine(); + const service = new MessagingService({ + logger: silentLogger(), + outbox, + getData: () => data.engine, + }); + const email = channelDouble('email', { available: true }); + service.registerChannel(email.channel); + + await service.emit({ + topic: 'deal.won', + audience: ['user_1', 'user_2', 'user_3', 'user_4'], + channels: ['email'], + organizationId: 'org_7', + payload: { title: 'Deal closed' }, + }); + + expect((await outbox.list())).toHaveLength(4); + expect(email.probes).toEqual([{ organizationId: 'org_7' }]); + }); + + it('fails OPEN: a throwing availability probe delivers exactly as before', async () => { + const outbox = new MemoryNotificationOutbox(1); + const data = capturingEngine(); + const warnings: string[] = []; + const service = new MessagingService({ + logger: { ...silentLogger(), warn: (...a: unknown[]) => warnings.push(String(a[0])) }, + outbox, + getData: () => data.engine, + }); + const broken: MessagingChannel = { + id: 'email', + async send() { return { ok: true }; }, + isAvailable() { throw new Error('probe exploded'); }, + }; + service.registerChannel(broken); + + const result = await service.emit({ + topic: 'deal.won', + audience: ['user_1'], + channels: ['email'], + payload: { title: 'Deal closed' }, + }); + + expect((await outbox.list()).map((r) => r.channel)).toEqual(['email']); + expect(result.suppressed).toEqual([]); + expect(warnings.join('\n')).toContain('probe exploded'); + }); + + it('writes the event but no delivery rows when EVERY requested channel is unavailable', async () => { + const outbox = new MemoryNotificationOutbox(1); + const data = capturingEngine(); + const service = new MessagingService({ + logger: silentLogger(), + outbox, + getData: () => data.engine, + }); + service.registerChannel(channelDouble('email', UNAVAILABLE).channel); + + const result = await service.emit({ + topic: 'deal.won', + audience: ['user_1'], + channels: ['email'], + payload: { title: 'Deal closed' }, + }); + + // The event happened, so it is recorded — with the reason it produced + // no work. The audit trail is what replaces the dead rows. + expect(data.inserts).toHaveLength(1); + expect(data.inserts[0].row.suppressed_channels) + .toEqual([{ channel: 'email', reason: 'transport_not_configured' }]); + expect(await outbox.list()).toHaveLength(0); + expect(result.enqueued).toBe(0); + expect(result.failed).toBe(0); + expect(result.notificationId).toBeTruthy(); + }); + + it('skips send() for an unavailable channel on the inline (P0) path too', async () => { + const data = capturingEngine(); + const service = new MessagingService({ logger: silentLogger(), getData: () => data.engine }); + const inbox = channelDouble('inbox'); + const email = channelDouble('email', UNAVAILABLE); + service.registerChannel(inbox.channel); + service.registerChannel(email.channel); + + const result = await service.emit({ + topic: 'deal.won', + audience: ['user_1'], + channels: ['inbox', 'email'], + payload: { title: 'Deal closed' }, + }); + + expect(inbox.sent.map((d) => d.channel)).toEqual(['inbox']); + expect(email.sent).toHaveLength(0); + expect(result.delivered).toBe(1); + expect(result.failed).toBe(0); + expect(result.suppressed).toEqual([{ channel: 'email', reason: 'transport_not_configured' }]); + // The surviving channel list is what the materialization is told about — + // a suppressed channel is not advertised to the ones that did run. + expect(inbox.sent[0].notification.channels).toEqual(['inbox']); + }); + + it('leaves an UNREGISTERED channel on its existing path — ⛔ not folded into suppression', async () => { + // Out of the ruling's scope on purpose: an unregistered channel has no + // implementation to ask, so it keeps today's behaviour exactly. Pinned + // so the boundary is deliberate rather than accidental. + const data = capturingEngine(); + const service = new MessagingService({ logger: silentLogger(), getData: () => data.engine }); + service.registerChannel(channelDouble('inbox').channel); + + const result = await service.emit({ + topic: 'deal.won', + audience: ['user_1'], + channels: ['inbox', 'nowhere'], + payload: { title: 'Deal closed' }, + }); + + expect(result.suppressed).toEqual([]); + expect(data.inserts[0].row.suppressed_channels).toBeNull(); + expect(result.deliveries.find((d) => d.channel === 'nowhere')) + .toMatchObject({ ok: false, error: "channel 'nowhere' not registered" }); + }); + + it('the object inlines exactly the closed reason vocabulary the seam declares', async () => { + // `packages/platform-objects` is a lower layer and cannot import + // `CHANNEL_UNAVAILABLE_REASONS`, so the enum is inlined there. This is + // what keeps the two copies equal — a comment would not. + const field = (SysNotification as { fields: Record }) + .fields.suppressed_channels; + expect(field, 'sys_notification must declare the suppression key').toBeTruthy(); + const described = String(field.description); + const marker = 'closed set: '; + expect(described).toContain(marker); + const inlined = described.slice(described.indexOf(marker) + marker.length) + .split(',') + .map((s) => s.trim()) + .filter(Boolean); + expect(inlined).toEqual([...CHANNEL_UNAVAILABLE_REASONS]); + }); +}); diff --git a/packages/services/service-messaging/src/email-channel.test.ts b/packages/services/service-messaging/src/email-channel.test.ts index b2bbceffe2..876b4ccde1 100644 --- a/packages/services/service-messaging/src/email-channel.test.ts +++ b/packages/services/service-messaging/src/email-channel.test.ts @@ -571,4 +571,62 @@ describe('email channel', () => { expect(email.templated[0]).not.toHaveProperty('organizationId'); }); }); + + describe('availability at fan-out (#17732)', () => { + // The channel's half of the ruling: fan-out asks BEFORE it writes a + // delivery row, and this channel answers from the transport it was + // handed — no I/O, so the service holds no cache (see the member's + // TSDoc for the measurement). + it('answers unavailable with a declared reason when no email service is registered', () => { + const data = fakeData(); + const ch = createEmailChannel({ + getEmail: () => undefined, + getData: () => data, + store: new NotificationTemplateStore({ getData: () => data }), + }); + expect(ch.isAvailable?.(silentCtx(), { organizationId: 'org_1' })) + .toEqual({ available: false, reason: 'transport_not_configured' }); + }); + + it('answers available once an email service is registered', () => { + const data = fakeData(); + const email = fakeEmail(); + const ch = createEmailChannel({ + getEmail: () => email.service, + getData: () => data, + store: new NotificationTemplateStore({ getData: () => data }), + }); + expect(ch.isAvailable?.(silentCtx(), { organizationId: 'org_1' })).toEqual({ available: true }); + }); + + it('re-reads the transport on every call — an answer is never memoized', () => { + // The transport is hot-swapped by the mail settings change bus, so a + // cached answer would outlive the configuration that produced it. + const data = fakeData(); + const email = fakeEmail(); + let installed: ReturnType['service'] | undefined; + const ch = createEmailChannel({ + getEmail: () => installed, + getData: () => data, + store: new NotificationTemplateStore({ getData: () => data }), + }); + expect(ch.isAvailable?.(silentCtx(), {})).toEqual({ available: false, reason: 'transport_not_configured' }); + installed = email.service; + expect(ch.isAvailable?.(silentCtx(), {})).toEqual({ available: true }); + }); + + it('performs no data access to answer — the probe is not a send', async () => { + const data = fakeData(); + const email = fakeEmail(); + const ch = createEmailChannel({ + getEmail: () => email.service, + getData: () => data, + store: new NotificationTemplateStore({ getData: () => data }), + }); + const before = data.findOnes.length; + ch.isAvailable?.(silentCtx(), { organizationId: 'org_1' }); + expect(data.findOnes.length).toBe(before); + expect(email.sent).toHaveLength(0); + }); + }); }); From e57ee063cacf8fab828403f7248783724c61acd5 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 13 Sep 2026 14:31:17 +0000 Subject: [PATCH 3/5] chore(platform-objects): regenerate i18n bundles for sys_notification.suppressed_channels Claude-Session: https://claude.ai/code/session_01URLHobLUJB9K1ABV6ofdjj Co-authored-by: Claude --- .../src/apps/translations/en.objects.generated.ts | 4 ++++ .../src/apps/translations/es-ES.objects.generated.ts | 4 ++++ .../src/apps/translations/es-ES.source-hashes.generated.ts | 2 ++ .../src/apps/translations/ja-JP.objects.generated.ts | 4 ++++ .../src/apps/translations/ja-JP.source-hashes.generated.ts | 2 ++ .../src/apps/translations/zh-CN.objects.generated.ts | 4 ++++ .../src/apps/translations/zh-CN.source-hashes.generated.ts | 2 ++ 7 files changed, 22 insertions(+) diff --git a/packages/platform-objects/src/apps/translations/en.objects.generated.ts b/packages/platform-objects/src/apps/translations/en.objects.generated.ts index 16591714ad..ea8959dae4 100644 --- a/packages/platform-objects/src/apps/translations/en.objects.generated.ts +++ b/packages/platform-objects/src/apps/translations/en.objects.generated.ts @@ -2347,6 +2347,10 @@ export const enObjects: NonNullable = { critical: "critical" } }, + suppressed_channels: { + label: "Suppressed Channels", + help: "Channels fan-out skipped because they are unavailable for this tenant, as [{channel, reason}]; reason is the closed set: transport_not_configured" + }, dedup_key: { label: "Dedup Key", help: "Idempotency key within a topic window; a repeat emit is a no-op" diff --git a/packages/platform-objects/src/apps/translations/es-ES.objects.generated.ts b/packages/platform-objects/src/apps/translations/es-ES.objects.generated.ts index 432b9d7182..f2a53fb87a 100644 --- a/packages/platform-objects/src/apps/translations/es-ES.objects.generated.ts +++ b/packages/platform-objects/src/apps/translations/es-ES.objects.generated.ts @@ -2347,6 +2347,10 @@ export const esESObjects: NonNullable = { critical: "critical" } }, + suppressed_channels: { + label: "Suppressed Channels", + help: "Channels fan-out skipped because they are unavailable for this tenant, as [{channel, reason}]; reason is the closed set: transport_not_configured" + }, dedup_key: { label: "Dedup Key", help: "Idempotency key within a topic window; a repeat emit is a no-op" diff --git a/packages/platform-objects/src/apps/translations/es-ES.source-hashes.generated.ts b/packages/platform-objects/src/apps/translations/es-ES.source-hashes.generated.ts index 936f4faa00..476ae94ab8 100644 --- a/packages/platform-objects/src/apps/translations/es-ES.source-hashes.generated.ts +++ b/packages/platform-objects/src/apps/translations/es-ES.source-hashes.generated.ts @@ -289,6 +289,8 @@ export const esESGeneratedSourceHashes: Readonly> = { "objects.sys_notification.fields.severity.options.critical": "7ff0ff69c0abaf81", "objects.sys_notification.fields.severity.options.info": "3e0c8611029f253b", "objects.sys_notification.fields.severity.options.warning": "2673ee95caf83284", + "objects.sys_notification.fields.suppressed_channels.help": "1fbbc17b95ea75d2", + "objects.sys_notification.fields.suppressed_channels.label": "910cb46f687f6270", "objects.sys_notification.fields.topic.help": "80e1790edfda49df", "objects.sys_notification.fields.topic.label": "819afdb3853e9d80", "objects.sys_oauth_access_token.fields.authorization_code_id.help": "84ad4d5d8c4c1cee", diff --git a/packages/platform-objects/src/apps/translations/ja-JP.objects.generated.ts b/packages/platform-objects/src/apps/translations/ja-JP.objects.generated.ts index 537f0c79de..81f7bd0e99 100644 --- a/packages/platform-objects/src/apps/translations/ja-JP.objects.generated.ts +++ b/packages/platform-objects/src/apps/translations/ja-JP.objects.generated.ts @@ -2347,6 +2347,10 @@ export const jaJPObjects: NonNullable = { critical: "critical" } }, + suppressed_channels: { + label: "Suppressed Channels", + help: "Channels fan-out skipped because they are unavailable for this tenant, as [{channel, reason}]; reason is the closed set: transport_not_configured" + }, dedup_key: { label: "Dedup Key", help: "Idempotency key within a topic window; a repeat emit is a no-op" diff --git a/packages/platform-objects/src/apps/translations/ja-JP.source-hashes.generated.ts b/packages/platform-objects/src/apps/translations/ja-JP.source-hashes.generated.ts index 0173a66004..6145060b5a 100644 --- a/packages/platform-objects/src/apps/translations/ja-JP.source-hashes.generated.ts +++ b/packages/platform-objects/src/apps/translations/ja-JP.source-hashes.generated.ts @@ -285,6 +285,8 @@ export const jaJPGeneratedSourceHashes: Readonly> = { "objects.sys_notification.fields.severity.options.critical": "7ff0ff69c0abaf81", "objects.sys_notification.fields.severity.options.info": "3e0c8611029f253b", "objects.sys_notification.fields.severity.options.warning": "2673ee95caf83284", + "objects.sys_notification.fields.suppressed_channels.help": "1fbbc17b95ea75d2", + "objects.sys_notification.fields.suppressed_channels.label": "910cb46f687f6270", "objects.sys_notification.fields.topic.help": "80e1790edfda49df", "objects.sys_notification.fields.topic.label": "819afdb3853e9d80", "objects.sys_oauth_access_token.fields.authorization_code_id.help": "84ad4d5d8c4c1cee", diff --git a/packages/platform-objects/src/apps/translations/zh-CN.objects.generated.ts b/packages/platform-objects/src/apps/translations/zh-CN.objects.generated.ts index 7e9b0c312d..43efe4e312 100644 --- a/packages/platform-objects/src/apps/translations/zh-CN.objects.generated.ts +++ b/packages/platform-objects/src/apps/translations/zh-CN.objects.generated.ts @@ -2347,6 +2347,10 @@ export const zhCNObjects: NonNullable = { critical: "critical" } }, + suppressed_channels: { + label: "Suppressed Channels", + help: "Channels fan-out skipped because they are unavailable for this tenant, as [{channel, reason}]; reason is the closed set: transport_not_configured" + }, dedup_key: { label: "Dedup Key", help: "Idempotency key within a topic window; a repeat emit is a no-op" diff --git a/packages/platform-objects/src/apps/translations/zh-CN.source-hashes.generated.ts b/packages/platform-objects/src/apps/translations/zh-CN.source-hashes.generated.ts index 4df2457671..f9988444b8 100644 --- a/packages/platform-objects/src/apps/translations/zh-CN.source-hashes.generated.ts +++ b/packages/platform-objects/src/apps/translations/zh-CN.source-hashes.generated.ts @@ -229,6 +229,8 @@ export const zhCNGeneratedSourceHashes: Readonly> = { "objects.sys_notification.fields.severity.options.critical": "7ff0ff69c0abaf81", "objects.sys_notification.fields.severity.options.info": "3e0c8611029f253b", "objects.sys_notification.fields.severity.options.warning": "2673ee95caf83284", + "objects.sys_notification.fields.suppressed_channels.help": "1fbbc17b95ea75d2", + "objects.sys_notification.fields.suppressed_channels.label": "910cb46f687f6270", "objects.sys_notification.fields.topic.help": "80e1790edfda49df", "objects.sys_notification.fields.topic.label": "819afdb3853e9d80", "objects.sys_oauth_access_token.fields.authorization_code_id.help": "84ad4d5d8c4c1cee", From dfce7f6367e9f96588bfb61b950a1fa6df845411 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 13 Sep 2026 14:52:12 +0000 Subject: [PATCH 4/5] fix(service-messaging): name suppressed_channels on the insert only when it carries information An insert names its columns. Naming the new column on every emit made every emit depend on every sys_notification schema already carrying it, so a stack whose object predates it answered INVALID_FIELD and lost the notification -- to record that nothing was suppressed. Claude-Session: https://claude.ai/code/session_01URLHobLUJB9K1ABV6ofdjj Co-authored-by: Claude --- .../src/channel-availability.test.ts | 30 ++++++++++++++++--- .../src/messaging-service.ts | 20 +++++++++---- 2 files changed, 41 insertions(+), 9 deletions(-) diff --git a/packages/services/service-messaging/src/channel-availability.test.ts b/packages/services/service-messaging/src/channel-availability.test.ts index c00450c33b..b4aad2ad73 100644 --- a/packages/services/service-messaging/src/channel-availability.test.ts +++ b/packages/services/service-messaging/src/channel-availability.test.ts @@ -177,9 +177,31 @@ describe('channel availability at fan-out (#17732)', () => { expect(rows.map((r) => r.channel).sort()).toEqual(['inbox', 'sms']); expect(result.enqueued).toBe(2); expect(result.suppressed).toEqual([]); - // NULL, not `[]` — the column stays empty on the common path, so a - // non-null value always means something really was dropped. - expect(data.inserts[0].row.suppressed_channels).toBeNull(); + // ⭐ ABSENT, not `null` — the insert must not even NAME the new column + // when it has nothing to say. An insert names its columns, so a row + // that always carries `suppressed_channels` makes every emit depend on + // every `sys_notification` schema already having it: a stack whose + // object predates the column answers `INVALID_FIELD` and the whole + // notification is lost, to record that nothing was suppressed. That is + // not hypothetical — it is what this change did to + // `service-automation`'s zero-delivery harness before it was fixed. + expect(Object.hasOwn(data.inserts[0].row, 'suppressed_channels')).toBe(false); + }); + + it('the common path writes EXACTLY the column set it wrote before this change', async () => { + // The strongest form of the pin above: enumerate the columns, so a + // future key added "harmlessly" to every insert has to face this test + // rather than a consumer's INVALID_FIELD months later. + const data = capturingEngine(); + const service = new MessagingService({ logger: silentLogger(), getData: () => data.engine }); + service.registerChannel(channelDouble('inbox').channel); + + await service.emit({ topic: 'deal.won', audience: ['user_1'], payload: { title: 'x' } }); + + expect(Object.keys(data.inserts[0].row).sort()).toEqual([ + 'actor_id', 'created_at', 'dedup_key', 'organization_id', + 'payload', 'severity', 'source_id', 'source_object', 'topic', + ]); }); it('THE CONTROL, other side: the same channel suppressed once it DOES answer unavailable', async () => { @@ -331,7 +353,7 @@ describe('channel availability at fan-out (#17732)', () => { }); expect(result.suppressed).toEqual([]); - expect(data.inserts[0].row.suppressed_channels).toBeNull(); + expect(Object.hasOwn(data.inserts[0].row, 'suppressed_channels')).toBe(false); expect(result.deliveries.find((d) => d.channel === 'nowhere')) .toMatchObject({ ok: false, error: "channel 'nowhere' not registered" }); }); diff --git a/packages/services/service-messaging/src/messaging-service.ts b/packages/services/service-messaging/src/messaging-service.ts index ebf3ed1a24..92991ec310 100644 --- a/packages/services/service-messaging/src/messaging-service.ts +++ b/packages/services/service-messaging/src/messaging-service.ts @@ -1173,13 +1173,23 @@ export class MessagingService { source_id: str(input.source?.id) ?? null, actor_id: input.actorId ?? null, organization_id: input.organizationId ?? null, - // [#17732] The channels fan-out will not even attempt, and why. - // NULL — not `[]` — when nothing was suppressed, so the column is - // empty on the overwhelmingly common path and a non-null value - // always means "something really was dropped". - suppressed_channels: suppressed.length > 0 ? suppressed.map((s) => ({ ...s })) : null, created_at: this.now(), }; + // [#17732] The channels fan-out will not even attempt, and why — + // present ONLY when there is something to say. + // + // ⛔ Not `suppressed_channels: … ?? null` alongside the keys above. An + // insert names its columns, and naming a NEW one on EVERY emit makes + // every emit depend on every `sys_notification` schema in the world + // already carrying it: a stack whose object predates this column gets + // `INVALID_FIELD: Unknown field 'suppressed_channels'` and loses the + // whole notification — to record that nothing was suppressed. Written + // this way the common path's column set is byte-for-byte what it was + // before this change, and the column appears exactly when it carries + // information. The pin is in `channel-availability.test.ts`. + if (suppressed.length > 0) { + row.suppressed_channels = suppressed.map((s) => ({ ...s })); + } const created = await data.insert(NOTIFICATION_EVENT_OBJECT, row); const id = Array.isArray(created) ? created[0]?.id : created?.id ?? created; return id != null ? String(id) : `evt_${Math.random().toString(36).slice(2)}`; From 684df79a9e28e1685f5d085263420d044b97880c Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 13 Sep 2026 15:02:32 +0000 Subject: [PATCH 5/5] =?UTF-8?q?fix(service-messaging):=20use=20hasOwnPrope?= =?UTF-8?q?rty.call=20=E2=80=94=20Object.hasOwn=20is=20outside=20this=20pa?= =?UTF-8?q?ckage's=20lib?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01URLHobLUJB9K1ABV6ofdjj Co-authored-by: Claude --- .../service-messaging/src/channel-availability.test.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/services/service-messaging/src/channel-availability.test.ts b/packages/services/service-messaging/src/channel-availability.test.ts index b4aad2ad73..c09afd1b21 100644 --- a/packages/services/service-messaging/src/channel-availability.test.ts +++ b/packages/services/service-messaging/src/channel-availability.test.ts @@ -185,7 +185,7 @@ describe('channel availability at fan-out (#17732)', () => { // notification is lost, to record that nothing was suppressed. That is // not hypothetical — it is what this change did to // `service-automation`'s zero-delivery harness before it was fixed. - expect(Object.hasOwn(data.inserts[0].row, 'suppressed_channels')).toBe(false); + expect(Object.prototype.hasOwnProperty.call(data.inserts[0].row, 'suppressed_channels')).toBe(false); }); it('the common path writes EXACTLY the column set it wrote before this change', async () => { @@ -353,7 +353,7 @@ describe('channel availability at fan-out (#17732)', () => { }); expect(result.suppressed).toEqual([]); - expect(Object.hasOwn(data.inserts[0].row, 'suppressed_channels')).toBe(false); + expect(Object.prototype.hasOwnProperty.call(data.inserts[0].row, 'suppressed_channels')).toBe(false); expect(result.deliveries.find((d) => d.channel === 'nowhere')) .toMatchObject({ ok: false, error: "channel 'nowhere' not registered" }); });