Registry, dispatcher, and dedup
Your first job is the spine of the project: name the three events the app can notify about, and ship a dispatch() that collapses a burst of the same event into a single notification.
By the end, the inspector tells the story. One fire sends to both stubbed channels and writes a single notification_dedup row; a second fire inside the sixty-second window is deduped and writes nothing; five rapid fires deliver once and swallow four duplicates. Wait past the window and the count starts fresh.
The channels stay stubbed: each only bumps the sent count, writing no inbox rows and sending no email. Dedup is the one behavior you verify here, and it is observable only through a dispatch() that consults the registry.
Your mission
Section titled “Your mission”Write three pieces that ship as a unit because they only become confirmable together: the registry that names what is notifiable, the dedup helper that collapses a burst, and the dispatch() that ties them together. Dedup lives inside the dispatcher, keyed by (eventType, dedupKey, recipientUserId), with a per-event window read from the registry. A registry miss is a programmer error, not a channel failure, so the dispatcher throws NotificationError('REGISTRY_MISS') rather than swallowing it.
Out of scope this lesson: the real channels, the real preference resolution, and the call-site wiring all stay as the starter’s no-op stubs. Those are the next two lessons.
notifications, user_notification_preferences, notification_dedup) exist after migration, with the dedup composite index on (eventType, dedupKey, recipientUserId, firedAt desc) and the partial unread index on notifications, confirmable in Drizzle Studio.Fire invite-sent from the inspector returns { sent: 2, deduped: 0, suppressedByPrefs: 0 } and writes exactly one notification_dedup row.deduped: 1 and writes no second dedup row.Rapid-fire invite-sent (5x in 2s) aggregates to one delivery with deduped: 4.deduped: 0 — the window has released.NotificationError('REGISTRY_MISS') rather than a silent no-op.Coding time
Section titled “Coding time”Implement the three notification tables in db/schema.ts, the registry in registry.ts, the dedup helpers in dedup.ts, and the dispatch() body in dispatcher.ts, against the reference signatures and the lesson’s tests. The inspector’s fire buttons already call dispatch() through provided server actions, and index.ts already re-exports it, so there is no wiring to do.
Reference solution and walkthrough
The three tables
Section titled “The three tables”Open db/schema.ts and find the // TODO(L2) block near the bottom: the three notification tables ship commented out. Uncomment them, add user to the auth import (the FKs reference it), and add index to the drizzle-orm/pg-core import. The columns are already written for you.
import { sql } from 'drizzle-orm';import { bigint, boolean, index, integer, jsonb, pgEnum, pgTable, text, timestamp, unique, uuid,} from 'drizzle-orm/pg-core';import { uuidv7 } from 'uuidv7';
import { timestamps } from '@/db/columns';import { organization, user } from '@/db/schema/auth';
// ...email_suppressions, processed_events, plan_entitlements unchanged...
// The in-app inbox feed. One row per delivered in-app notification, written ONLY by// writeInboxChannel (the dispatcher's inbox channel); any direct write to this table// outside lib/notifications/ is a regression. title/body are rendered once at dispatch and// frozen here (render-at-dispatch), so the inbox UI is a pure read with no joins, immune to// later actor-name drift. userId/orgId are `text` (Better Auth ids); ids default to uuidv7// in app code, never a database-side default. readAt null = unread.export const notifications = pgTable( 'notifications', {14 collapsed lines
id: uuid() .primaryKey() .$defaultFn(() => uuidv7()), userId: text() .notNull() .references(() => user.id, { onDelete: 'cascade' }), orgId: text().references(() => organization.id, { onDelete: 'cascade' }), eventType: text().notNull(), subjectId: text().notNull(), title: text().notNull(), body: text().notNull(), payload: jsonb().$type<Record<string, unknown>>().notNull().default({}), readAt: timestamp({ withTimezone: true }), createdAt: timestamp({ withTimezone: true }).defaultNow().notNull(), }, (t) => [ index('idx_notifications_user_created').on(t.userId, t.createdAt.desc()), index('idx_notifications_user_unread') .on(t.userId) .where(sql`read_at is null`), ],);
export type Notification = typeof notifications.$inferSelect;export type NewNotification = typeof notifications.$inferInsert;
// Per-category, per-channel opt-out. Read once per dispatch (batched) and applied// default-on (`?? true` — a missing row receives everything). NO orgId: prefs are// user-scoped, they follow the user across orgs, so the tenant-leading-column rule does// not apply. The named (userId, category) unique is what the inspector's UPSERT conflicts// on. push is reserved at the column with no channel consumer in this project.export const userNotificationPreferences = pgTable( 'user_notification_preferences', {14 collapsed lines
id: uuid() .primaryKey() .$defaultFn(() => uuidv7()), userId: text() .notNull() .references(() => user.id, { onDelete: 'cascade' }), category: text().notNull(), email: boolean().notNull().default(true), inbox: boolean().notNull().default(true), push: boolean().notNull().default(true), updatedAt: timestamp({ withTimezone: true }) .notNull() .defaultNow() .$onUpdate(() => new Date()), }, (t) => [ unique('user_notification_preferences_user_id_category_unique').on( t.userId, t.category, ), ],);
export type UserNotificationPreference = typeof userNotificationPreferences.$inferSelect;export type NewUserNotificationPreference = typeof userNotificationPreferences.$inferInsert;
// The time-windowed dedup ledger. One row per (eventType, dedupKey, recipientUserId)// fan-out; the dispatcher's isDuplicate selects the most-recent row inside the registry's// window before fanning out, collapsing a burst to one. recipientUserId is load-bearing in// the key (two recipients getting the same event is not a duplicate). NO orgId// (user-scoped bookkeeping). The composite index leads with the lookup columns and sorts// firedAt desc for the most-recent probe.export const notificationDedup = pgTable( 'notification_dedup', {9 collapsed lines
id: uuid() .primaryKey() .$defaultFn(() => uuidv7()), eventType: text().notNull(), dedupKey: text().notNull(), recipientUserId: text() .notNull() .references(() => user.id, { onDelete: 'cascade' }), firedAt: timestamp({ withTimezone: true }).defaultNow().notNull(), }, (t) => [ index('idx_notification_dedup_lookup').on( t.eventType, t.dedupKey, t.recipientUserId, t.firedAt.desc(), ), ],);
export type NotificationDedup = typeof notificationDedup.$inferSelect;export type NewNotificationDedup = typeof notificationDedup.$inferInsert;One column is easy to get wrong: userId must be text, not uuid, to match Better Auth’s text user.id; a mismatched FK type fails the migration. The title and body columns on notifications are plain text because the dispatcher renders them once and freezes them onto the row, so the inbox reads them with no join and never drifts when an actor is later renamed.
Generate and apply the migration:
pnpm db:generate --name add_notificationsOpen the file Drizzle Kit writes (drizzle/0011_add_notifications.sql) before running it. You should see three CREATE TABLE statements, the foreign keys to user/organization, the dedup composite index, and one partial index ending in WHERE read_at is null.
CREATE INDEX "idx_notification_dedup_lookup" ON "notification_dedup" USING btree ("event_type","dedup_key","recipient_user_id","fired_at" DESC NULLS LAST);--> statement-breakpointCREATE INDEX "idx_notifications_user_created" ON "notifications" USING btree ("user_id","created_at" DESC NULLS LAST);--> statement-breakpointCREATE INDEX "idx_notifications_user_unread" ON "notifications" USING btree ("user_id") WHERE read_at is null;pnpm db:migrateThe tests assert behavior, not table shape, so confirm the schema and indexes by hand in Drizzle Studio.
The registry
Section titled “The registry”The registry is the source of truth for what is notifiable: each entry names the channels the event fans out to, its templates, its preference category, and its dedup configuration. The as const satisfies Record<string, NotifiableEvent> does two jobs: satisfies checks every entry against the NotifiableEvent shape, and as const freezes the keys so EventType = keyof typeof notifiableEvents infers the exact union of the three string literals, making an unknown event type a compile error at every call site. It is the same as const satisfies move from Derive types from values, applied to a real registry.
import BillingPastDueEmail from '@/emails/BillingPastDueEmail';import InviteSentEmail from '@/emails/InviteSentEmail';import RoleChangedEmail from '@/emails/RoleChangedEmail';
import type { NotifiableEvent } from './types';
8 collapsed lines
// The registry is the source of truth: it enumerates what is notifiable and how. Adding// an event is one entry; adding a channel later is one function of the same signature.// `as const satisfies Record<string, NotifiableEvent>` makes an unknown key a compile// error and infers EventType. Each entry's email template declares its own typed payload —// the permissive `(props: any) => ReactElement` field accepts every one. The inbox// formatter renders title/body from the payload (frozen onto the row at dispatch). The// dedup window is per-event; org.billing.past_due names email its critical channel so it// flows even when the user toggled billing email off.export const notifiableEvents = { 'org.invitation.sent': { channels: ['email', 'inbox'],7 collapsed lines
templates: { email: InviteSentEmail, inbox: (payload) => ({ title: `Invitation to ${String(payload.orgName)}`, body: `${String(payload.inviterName)} invited you to join ${String(payload.orgName)} as a ${String(payload.role)}.`, }), }, preferenceCategory: 'team', dedup: { windowSeconds: 60, keyBy: ['subjectId'] }, description: 'A member was invited to the organization.', }, 'org.member.role_changed': { channels: ['email', 'inbox'],7 collapsed lines
templates: { email: RoleChangedEmail, inbox: (payload) => ({ title: `Your role in ${String(payload.orgName)} changed`, body: `${String(payload.actorName)} changed your role from ${String(payload.before)} to ${String(payload.newRole)}.`, }), }, preferenceCategory: 'team', dedup: { windowSeconds: 60, keyBy: ['subjectId', 'newRole'] }, description: "A member's role in the organization changed.", }, 'org.billing.past_due': { channels: ['email', 'inbox'],7 collapsed lines
templates: { email: BillingPastDueEmail, inbox: (payload) => ({ title: `Payment past due for ${String(payload.orgName)}`, body: `The latest payment for ${String(payload.orgName)}'s ${String(payload.plan)} plan did not go through.`, }), }, preferenceCategory: 'billing', criticalChannel: 'email', dedup: { windowSeconds: 60, keyBy: ['subjectId'] }, description: 'An organization subscription went past due.', },} as const satisfies Record<string, NotifiableEvent>;
export type EventType = keyof typeof notifiableEvents;Notice the per-event dedup config. Two events key by ['subjectId'] alone, but org.member.role_changed keys by ['subjectId', 'newRole']: promoting someone and then demoting them are two distinct notifications even within the window, so the new role is part of the identity.
One thing in types.ts looks wrong but is required: the email-template field is typed (props: any) => ReactElement, not the tighter (props: Record<string, unknown>) => ReactElement. Each template declares its own typed props, and under TS 6 strict a component with typed props does not assign to a field whose parameter is Record<string, unknown>: that is parameter contravariance (a TS2322), since a function demanding a specific shape cannot stand in for one that must accept any object. Only the permissive any form accepts every typed template while staying callable, and it carries a biome-ignore with that rationale.
The dedup helpers
Section titled “The dedup helpers”Three small functions back the window. computeDedupKey joins the registry entry’s keyBy values with :. isDuplicate selects the most-recent matching row inside the window and returns true on a hit. recordDedup inserts one row marking the triple as fired now.
import 'server-only';
import { and, desc, eq, gt, sql } from 'drizzle-orm';
import { db } from '@/db';import { notificationDedup } from '@/db/schema';
import { notifiableEvents } from './registry';import type { NotificationEvent } from './types';
type DedupArgs = { event: NotificationEvent; userId: string; payload: Record<string, unknown>;};
// Build the dedup key by joining the registry entry's keyBy field values with ':'.// subjectId is read off the event; every other key is read off the payload.export const computeDedupKey = ( event: NotificationEvent, payload: Record<string, unknown>,): string => { const eventDef = notifiableEvents[event.type]; return eventDef.dedup.keyBy .map((key) => key === 'subjectId' ? event.subjectId : String(payload[key]), ) .join(':');};
export const isDuplicate = async ({ event, userId, payload,}: DedupArgs): Promise<boolean> => { const eventDef = notifiableEvents[event.type]; const dedupKey = computeDedupKey(event, payload); const since = sql`now() - make_interval(secs => ${eventDef.dedup.windowSeconds})`; const row = await db .select({ id: notificationDedup.id }) .from(notificationDedup) .where( and( eq(notificationDedup.eventType, event.type), eq(notificationDedup.dedupKey, dedupKey), eq(notificationDedup.recipientUserId, userId), gt(notificationDedup.firedAt, since), ), ) .orderBy(desc(notificationDedup.firedAt)) .limit(1); return row.length > 0;};
export const recordDedup = async ({ event, userId, payload,}: DedupArgs): Promise<void> => { await db.insert(notificationDedup).values({ eventType: event.type, dedupKey: computeDedupKey(event, payload), recipientUserId: userId, });};computeDedupKey builds the key string by joining the registry entry’s keyBy field values with :. subjectId is read off the event itself; every other key is read off the payload, so role_changed can fold its newRole into the identity.
import 'server-only';
import { and, desc, eq, gt, sql } from 'drizzle-orm';
import { db } from '@/db';import { notificationDedup } from '@/db/schema';
import { notifiableEvents } from './registry';import type { NotificationEvent } from './types';
type DedupArgs = { event: NotificationEvent; userId: string; payload: Record<string, unknown>;};
// Build the dedup key by joining the registry entry's keyBy field values with ':'.// subjectId is read off the event; every other key is read off the payload.export const computeDedupKey = ( event: NotificationEvent, payload: Record<string, unknown>,): string => { const eventDef = notifiableEvents[event.type]; return eventDef.dedup.keyBy .map((key) => key === 'subjectId' ? event.subjectId : String(payload[key]), ) .join(':');};
export const isDuplicate = async ({ event, userId, payload,}: DedupArgs): Promise<boolean> => { const eventDef = notifiableEvents[event.type]; const dedupKey = computeDedupKey(event, payload); const since = sql`now() - make_interval(secs => ${eventDef.dedup.windowSeconds})`; const row = await db .select({ id: notificationDedup.id }) .from(notificationDedup) .where( and( eq(notificationDedup.eventType, event.type), eq(notificationDedup.dedupKey, dedupKey), eq(notificationDedup.recipientUserId, userId), gt(notificationDedup.firedAt, since), ), ) .orderBy(desc(notificationDedup.firedAt)) .limit(1); return row.length > 0;};
export const recordDedup = async ({ event, userId, payload,}: DedupArgs): Promise<void> => { await db.insert(notificationDedup).values({ eventType: event.type, dedupKey: computeDedupKey(event, payload), recipientUserId: userId, });};isDuplicate returns true when a matching row was fired inside the window — firedAt > now() - make_interval(secs => windowSeconds). The orderBy firedAt desc + limit 1 walks the composite index’s lookup columns to the most-recent probe.
import 'server-only';
import { and, desc, eq, gt, sql } from 'drizzle-orm';
import { db } from '@/db';import { notificationDedup } from '@/db/schema';
import { notifiableEvents } from './registry';import type { NotificationEvent } from './types';
type DedupArgs = { event: NotificationEvent; userId: string; payload: Record<string, unknown>;};
// Build the dedup key by joining the registry entry's keyBy field values with ':'.// subjectId is read off the event; every other key is read off the payload.export const computeDedupKey = ( event: NotificationEvent, payload: Record<string, unknown>,): string => { const eventDef = notifiableEvents[event.type]; return eventDef.dedup.keyBy .map((key) => key === 'subjectId' ? event.subjectId : String(payload[key]), ) .join(':');};
export const isDuplicate = async ({ event, userId, payload,}: DedupArgs): Promise<boolean> => { const eventDef = notifiableEvents[event.type]; const dedupKey = computeDedupKey(event, payload); const since = sql`now() - make_interval(secs => ${eventDef.dedup.windowSeconds})`; const row = await db .select({ id: notificationDedup.id }) .from(notificationDedup) .where( and( eq(notificationDedup.eventType, event.type), eq(notificationDedup.dedupKey, dedupKey), eq(notificationDedup.recipientUserId, userId), gt(notificationDedup.firedAt, since), ), ) .orderBy(desc(notificationDedup.firedAt)) .limit(1); return row.length > 0;};
export const recordDedup = async ({ event, userId, payload,}: DedupArgs): Promise<void> => { await db.insert(notificationDedup).values({ eventType: event.type, dedupKey: computeDedupKey(event, payload), recipientUserId: userId, });};recordDedup inserts one row marking this (eventType, dedupKey, recipientUserId) triple as fired now — landing only after a successful fan-out, so a fully-suppressed recipient never opens a window.
Because make_interval(secs => windowSeconds) reads the registry’s number at query time, the window releases on its own: a row stamped sixty-one seconds ago falls outside a sixty-second window. Widening it for a chatty event is a one-number change in the registry.
Between the isDuplicate read and the recordDedup write, two truly concurrent fires can both read “not a duplicate” and both send. You accept that: one duplicate per rare concurrent burst, not a correctness hole. The hardening, when throughput justifies it, is a partial unique constraint on the key plus onConflictDoNothing to move the decision into the database, where it is atomic. That is named, not built.
The dispatcher
Section titled “The dispatcher”The dispatcher is the one seam: every call site builds a NotificationEvent and await dispatch(...), and nothing else imports a channel or writes the notifications table. The body runs in phases: look up the registry entry, read preferences once for all recipients, render the content once, then loop per recipient to resolve channels, check dedup, fan out behind a per-channel try/catch, and record the dedup row last.
import 'server-only';
import { logger } from '@/lib/logger';
import { sendEmailChannel } from './channels/email';import { writeInboxChannel } from './channels/inbox';import { isDuplicate, recordDedup } from './dedup';import { NotificationError } from './errors';import { readPrefsForCategory, resolveChannels } from './prefs';import { notifiableEvents } from './registry';import type { ChannelFn, ChannelName, DispatchResult, NotificationEvent, RenderedContent,} from './types';
// The uniform channel table: the dispatcher loops `await channelFns[channel](args)` with no// branch on channel name. Adding a channel later is one entry of the same signature.const channelFns = { email: sendEmailChannel, inbox: writeInboxChannel,} satisfies Record<ChannelName, ChannelFn>;
export const dispatch = async ( event: NotificationEvent,): Promise<DispatchResult> => { const eventDef = notifiableEvents[event.type]; if (!eventDef) { throw new NotificationError('REGISTRY_MISS', event.type); }
const result: DispatchResult = { sent: 0, deduped: 0, suppressedByPrefs: 0 };
// One batched read across all recipients (never per-recipient). // TODO(L3) const prefsByUser = await readPrefsForCategory( event.recipientUserIds, eventDef.preferenceCategory, );
// Rendered once per dispatch, frozen onto every inbox row / passed to the email template. const rendered: RenderedContent = { emailProps: event.payload, inbox: eventDef.templates.inbox(event.payload), orgId: null, };
for (const userId of event.recipientUserIds) { const channels = resolveChannels(eventDef, prefsByUser.get(userId)); result.suppressedByPrefs += eventDef.channels.length - channels.length; if (channels.length === 0) { continue; }
const duplicate = await isDuplicate({ event, userId, payload: event.payload, }); if (duplicate) { result.deduped++; continue; }
for (const channel of channels) { try { await channelFns[channel]({ recipient: { userId }, event, payload: event.payload, rendered, }); result.sent++; } catch (e) { logger.error( { seam: 'notifications.channel', channel, err: e }, 'channel failed', ); } }
await recordDedup({ event, userId, payload: event.payload }); }
logger.info({ seam: 'notifications.dispatch', ...result }, 'dispatch settled'); return result;};channelFns is a plain object typed satisfies Record<ChannelName, ChannelFn>, so the inner loop is await channelFns[channel](args) with no if (channel === 'email') branch anywhere. Adding a third channel later is one new key.
import 'server-only';
import { logger } from '@/lib/logger';
import { sendEmailChannel } from './channels/email';import { writeInboxChannel } from './channels/inbox';import { isDuplicate, recordDedup } from './dedup';import { NotificationError } from './errors';import { readPrefsForCategory, resolveChannels } from './prefs';import { notifiableEvents } from './registry';import type { ChannelFn, ChannelName, DispatchResult, NotificationEvent, RenderedContent,} from './types';
// The uniform channel table: the dispatcher loops `await channelFns[channel](args)` with no// branch on channel name. Adding a channel later is one entry of the same signature.const channelFns = { email: sendEmailChannel, inbox: writeInboxChannel,} satisfies Record<ChannelName, ChannelFn>;
export const dispatch = async ( event: NotificationEvent,): Promise<DispatchResult> => { const eventDef = notifiableEvents[event.type]; if (!eventDef) { throw new NotificationError('REGISTRY_MISS', event.type); }
const result: DispatchResult = { sent: 0, deduped: 0, suppressedByPrefs: 0 };
// One batched read across all recipients (never per-recipient). // TODO(L3) const prefsByUser = await readPrefsForCategory( event.recipientUserIds, eventDef.preferenceCategory, );
// Rendered once per dispatch, frozen onto every inbox row / passed to the email template. const rendered: RenderedContent = { emailProps: event.payload, inbox: eventDef.templates.inbox(event.payload), orgId: null, };
for (const userId of event.recipientUserIds) { const channels = resolveChannels(eventDef, prefsByUser.get(userId)); result.suppressedByPrefs += eventDef.channels.length - channels.length; if (channels.length === 0) { continue; }
const duplicate = await isDuplicate({ event, userId, payload: event.payload, }); if (duplicate) { result.deduped++; continue; }
for (const channel of channels) { try { await channelFns[channel]({ recipient: { userId }, event, payload: event.payload, rendered, }); result.sent++; } catch (e) { logger.error( { seam: 'notifications.channel', channel, err: e }, 'channel failed', ); } }
await recordDedup({ event, userId, payload: event.payload }); }
logger.info({ seam: 'notifications.dispatch', ...result }, 'dispatch settled'); return result;};The registry lookup runs first, and a miss throws REGISTRY_MISS before the loop — an unknown event type is a programmer error in the calling code, the one failure the dispatcher never swallows.
import 'server-only';
import { logger } from '@/lib/logger';
import { sendEmailChannel } from './channels/email';import { writeInboxChannel } from './channels/inbox';import { isDuplicate, recordDedup } from './dedup';import { NotificationError } from './errors';import { readPrefsForCategory, resolveChannels } from './prefs';import { notifiableEvents } from './registry';import type { ChannelFn, ChannelName, DispatchResult, NotificationEvent, RenderedContent,} from './types';
// The uniform channel table: the dispatcher loops `await channelFns[channel](args)` with no// branch on channel name. Adding a channel later is one entry of the same signature.const channelFns = { email: sendEmailChannel, inbox: writeInboxChannel,} satisfies Record<ChannelName, ChannelFn>;
export const dispatch = async ( event: NotificationEvent,): Promise<DispatchResult> => { const eventDef = notifiableEvents[event.type]; if (!eventDef) { throw new NotificationError('REGISTRY_MISS', event.type); }
const result: DispatchResult = { sent: 0, deduped: 0, suppressedByPrefs: 0 };
// One batched read across all recipients (never per-recipient). // TODO(L3) const prefsByUser = await readPrefsForCategory( event.recipientUserIds, eventDef.preferenceCategory, );
// Rendered once per dispatch, frozen onto every inbox row / passed to the email template. const rendered: RenderedContent = { emailProps: event.payload, inbox: eventDef.templates.inbox(event.payload), orgId: null, };
for (const userId of event.recipientUserIds) { const channels = resolveChannels(eventDef, prefsByUser.get(userId)); result.suppressedByPrefs += eventDef.channels.length - channels.length; if (channels.length === 0) { continue; }
const duplicate = await isDuplicate({ event, userId, payload: event.payload, }); if (duplicate) { result.deduped++; continue; }
for (const channel of channels) { try { await channelFns[channel]({ recipient: { userId }, event, payload: event.payload, rendered, }); result.sent++; } catch (e) { logger.error( { seam: 'notifications.channel', channel, err: e }, 'channel failed', ); } }
await recordDedup({ event, userId, payload: event.payload }); }
logger.info({ seam: 'notifications.dispatch', ...result }, 'dispatch settled'); return result;};The batched readPrefsForCategory and the render-once both run against the starter’s stubs this lesson — readPrefsForCategory returns an empty map and the no-op channels ignore rendered. The // TODO(L3) marks where they go live next lesson.
import 'server-only';
import { logger } from '@/lib/logger';
import { sendEmailChannel } from './channels/email';import { writeInboxChannel } from './channels/inbox';import { isDuplicate, recordDedup } from './dedup';import { NotificationError } from './errors';import { readPrefsForCategory, resolveChannels } from './prefs';import { notifiableEvents } from './registry';import type { ChannelFn, ChannelName, DispatchResult, NotificationEvent, RenderedContent,} from './types';
// The uniform channel table: the dispatcher loops `await channelFns[channel](args)` with no// branch on channel name. Adding a channel later is one entry of the same signature.const channelFns = { email: sendEmailChannel, inbox: writeInboxChannel,} satisfies Record<ChannelName, ChannelFn>;
export const dispatch = async ( event: NotificationEvent,): Promise<DispatchResult> => { const eventDef = notifiableEvents[event.type]; if (!eventDef) { throw new NotificationError('REGISTRY_MISS', event.type); }
const result: DispatchResult = { sent: 0, deduped: 0, suppressedByPrefs: 0 };
// One batched read across all recipients (never per-recipient). // TODO(L3) const prefsByUser = await readPrefsForCategory( event.recipientUserIds, eventDef.preferenceCategory, );
// Rendered once per dispatch, frozen onto every inbox row / passed to the email template. const rendered: RenderedContent = { emailProps: event.payload, inbox: eventDef.templates.inbox(event.payload), orgId: null, };
for (const userId of event.recipientUserIds) { const channels = resolveChannels(eventDef, prefsByUser.get(userId)); result.suppressedByPrefs += eventDef.channels.length - channels.length; if (channels.length === 0) { continue; }
const duplicate = await isDuplicate({ event, userId, payload: event.payload, }); if (duplicate) { result.deduped++; continue; }
for (const channel of channels) { try { await channelFns[channel]({ recipient: { userId }, event, payload: event.payload, rendered, }); result.sent++; } catch (e) { logger.error( { seam: 'notifications.channel', channel, err: e }, 'channel failed', ); } }
await recordDedup({ event, userId, payload: event.payload }); }
logger.info({ seam: 'notifications.dispatch', ...result }, 'dispatch settled'); return result;};The per-recipient loop. The stub resolveChannels returns every channel, so nothing is suppressed yet; isDuplicate skips a recipient already inside the window, incrementing deduped and continue-ing past the fan-out.
import 'server-only';
import { logger } from '@/lib/logger';
import { sendEmailChannel } from './channels/email';import { writeInboxChannel } from './channels/inbox';import { isDuplicate, recordDedup } from './dedup';import { NotificationError } from './errors';import { readPrefsForCategory, resolveChannels } from './prefs';import { notifiableEvents } from './registry';import type { ChannelFn, ChannelName, DispatchResult, NotificationEvent, RenderedContent,} from './types';
// The uniform channel table: the dispatcher loops `await channelFns[channel](args)` with no// branch on channel name. Adding a channel later is one entry of the same signature.const channelFns = { email: sendEmailChannel, inbox: writeInboxChannel,} satisfies Record<ChannelName, ChannelFn>;
export const dispatch = async ( event: NotificationEvent,): Promise<DispatchResult> => { const eventDef = notifiableEvents[event.type]; if (!eventDef) { throw new NotificationError('REGISTRY_MISS', event.type); }
const result: DispatchResult = { sent: 0, deduped: 0, suppressedByPrefs: 0 };
// One batched read across all recipients (never per-recipient). // TODO(L3) const prefsByUser = await readPrefsForCategory( event.recipientUserIds, eventDef.preferenceCategory, );
// Rendered once per dispatch, frozen onto every inbox row / passed to the email template. const rendered: RenderedContent = { emailProps: event.payload, inbox: eventDef.templates.inbox(event.payload), orgId: null, };
for (const userId of event.recipientUserIds) { const channels = resolveChannels(eventDef, prefsByUser.get(userId)); result.suppressedByPrefs += eventDef.channels.length - channels.length; if (channels.length === 0) { continue; }
const duplicate = await isDuplicate({ event, userId, payload: event.payload, }); if (duplicate) { result.deduped++; continue; }
for (const channel of channels) { try { await channelFns[channel]({ recipient: { userId }, event, payload: event.payload, rendered, }); result.sent++; } catch (e) { logger.error( { seam: 'notifications.channel', channel, err: e }, 'channel failed', ); } }
await recordDedup({ event, userId, payload: event.payload }); }
logger.info({ seam: 'notifications.dispatch', ...result }, 'dispatch settled'); return result;};The fan-out is wrapped in a per-channel try/catch, not a per-dispatch one: a failing channel (email down, address won’t resolve) is logged and walked past so the other channel still fires, and each success bumps sent once.
import 'server-only';
import { logger } from '@/lib/logger';
import { sendEmailChannel } from './channels/email';import { writeInboxChannel } from './channels/inbox';import { isDuplicate, recordDedup } from './dedup';import { NotificationError } from './errors';import { readPrefsForCategory, resolveChannels } from './prefs';import { notifiableEvents } from './registry';import type { ChannelFn, ChannelName, DispatchResult, NotificationEvent, RenderedContent,} from './types';
// The uniform channel table: the dispatcher loops `await channelFns[channel](args)` with no// branch on channel name. Adding a channel later is one entry of the same signature.const channelFns = { email: sendEmailChannel, inbox: writeInboxChannel,} satisfies Record<ChannelName, ChannelFn>;
export const dispatch = async ( event: NotificationEvent,): Promise<DispatchResult> => { const eventDef = notifiableEvents[event.type]; if (!eventDef) { throw new NotificationError('REGISTRY_MISS', event.type); }
const result: DispatchResult = { sent: 0, deduped: 0, suppressedByPrefs: 0 };
// One batched read across all recipients (never per-recipient). // TODO(L3) const prefsByUser = await readPrefsForCategory( event.recipientUserIds, eventDef.preferenceCategory, );
// Rendered once per dispatch, frozen onto every inbox row / passed to the email template. const rendered: RenderedContent = { emailProps: event.payload, inbox: eventDef.templates.inbox(event.payload), orgId: null, };
for (const userId of event.recipientUserIds) { const channels = resolveChannels(eventDef, prefsByUser.get(userId)); result.suppressedByPrefs += eventDef.channels.length - channels.length; if (channels.length === 0) { continue; }
const duplicate = await isDuplicate({ event, userId, payload: event.payload, }); if (duplicate) { result.deduped++; continue; }
for (const channel of channels) { try { await channelFns[channel]({ recipient: { userId }, event, payload: event.payload, rendered, }); result.sent++; } catch (e) { logger.error( { seam: 'notifications.channel', channel, err: e }, 'channel failed', ); } }
await recordDedup({ event, userId, payload: event.payload }); }
logger.info({ seam: 'notifications.dispatch', ...result }, 'dispatch settled'); return result;};recordDedup lands last, after a successful fan-out, opening the window that the next fire inside sixty seconds will dedup against.
Because preferences and content resolve against stubs this lesson — an empty prefs map and pass-through channels — Fire invite-sent reads sent: 2 with suppressedByPrefs: 0. Dedup is the one path fully wired, and the part the tests check.
The .on(...) composite index, the .where(sql`...`) partial index, and the unique() constraint you write into the three tables.
make_interval(secs => …) is the function the dedup window predicate builds its now()-relative cutoff from.
Moment of truth
Section titled “Moment of truth”Run the test suite:
pnpm test:lesson 2It runs your dispatch() against the app’s local Postgres and reads the dedup ledger directly, so the Lesson 2 migration must already be applied. A green run looks like this:
✓ tests/lessons/Lesson 2.test.ts (5) ✓ a first fire delivers and opens the dedup window ✓ a refire inside the window is deduped ✓ a rapid burst of five collapses to one delivery ✓ a fire after the window has elapsed delivers fresh ✓ an unknown event type surfaces a REGISTRY_MISS
Test Files 1 passed (1) Tests 5 passed (5)Then confirm by hand what the tests do not reach:
Fire invite-sent in the inspector shows { sent: 2, deduped: 0, suppressedByPrefs: 0 } and one new notification_dedup row.deduped: 1 with no second dedup row.Rapid-fire invite-sent ×5 shows deduped: 4.deduped: 0 — the window has released.The channels are still no-op stubs, so no inbox rows appear and the email-sent counter stays put. The next lesson wires them up.