Skip to content
LEARNREACTSAAS
Chapter 71Lesson 4

Wire the three call sites

So far the dispatcher has fired only from the inspector, calling dispatch() with a hand-built payload. This lesson wires it into the three product surfaces it was built for — sending an invitation, changing a member’s role, and the Stripe past-due webhook — under one rule: dispatch only after the transaction commits.

By the end, the flow runs the way it will in production. sendInvitation to an existing user writes the invitation, commits, then drops an inbox row and an email for the invitee. changeMemberRole writes both an auditLogs row and a notifications row. A customer.subscription.updated event that flips an org to past_due lands, commits, then notifies every owner. Open /inbox after any of these and you are reading real rows, not a fixture.

Your mission

Section titled “Your mission”

Move the dispatcher off the inspector demo and onto its three call sites: sendInvitation and changeMemberRole (in lib/invitations/) and the past-due branch of the Stripe webhook. The dispatcher is finished; you decide only where and when each call site invokes it, and the when is the whole lesson.

It is one discipline applied three times: do the transactional work with await withTenant(...) (or db.transaction(...) for the webhook), let it commit, then await dispatch(...), never the reverse. A notification fired inside a transaction that later rolls back is one you cannot take back: you would tell a user they were invited to an org whose invitation row never landed. sendInvitation and changeMemberRole get this almost free, since their writes already finish inside a withTenant block before any notification code.

The webhook is the hard case. The owners you notify must reflect the transition you are committing, so you read them inside the transaction, but you cannot dispatch there, because nothing has committed yet. So the handler splits the two halves: inside tx it reads the owners and pushes an org.billing.past_due descriptor onto a pendingDispatches: NotificationEvent[] array captured by the route’s closure; after db.transaction resolves, the POST drains that array through the dispatcher. The durable version is the transactional outbox: a pending_dispatches table a worker drains, surviving a crash between commit and dispatch. An in-memory array is the honest v1; you will name the upgrade, not build it.

Three constraints shape the rest. The dispatcher does no authorization: the admin check lives at the action boundary in authedAction('admin', ...), so gating is the call site’s job. The two dedup layers compose without knowing about each other: processed_events at the webhook catches a redelivered Stripe event, while notificationDedup inside the dispatcher catches a duplicate notification. And one wart is accepted on purpose: an invite to an existing user sends two emails, chapter 065’s invitation email plus the dispatcher’s org.invitation.sent. Merging them is a next step; v1 keeps both.

Out of scope: changing lib/email.ts, merging the duplicate invite emails, and short-circuiting a no-op role change.

sendInvitation to an existing user writes the invitation, commits, then produces one inbox row plus one email increment for the invitee.
tested
Inviting a non-user address no-ops the dispatcher — only chapter 065’s invitation email sends, no inbox row, no email increment from the seam.
tested
changeMemberRole writes both an auditLogs row and a notifications row, with one email increment for the affected member.
tested
The Stripe past-due transition lands the webhook, commits, then produces one inbox row plus one email per org owner.
tested
A rolled-back action notifies nobody — no rows anywhere, no email increment — the fire-after-commit guarantee.
tested
A search for sendEmail( and db.insert(notifications) outside lib/notifications/ returns only the one named exception, chapter 065’s invitation email in sendInvitation — the seam holds.
untested

Coding time

Section titled “Coding time”

Add the dispatch() call after commit in src/lib/invitations/send.ts and src/lib/invitations/manage.ts, then wire the past-due path across src/lib/webhooks/stripe.ts and src/app/api/webhooks/stripe/route.ts. Each file already carries a // TODO(L4) marking where its edit goes.

Reference solution and walkthrough

The invitation: dispatch after the commit

Section titled “The invitation: dispatch after the commit”

Start with send.ts, the simplest of the three. The invitation row and its audit entry co-transact inside withTenant, which returns invitationId once that transaction commits, then chapter 065’s invitation email sends through its own sendEmail call. Your edit is the block after it: one dispatch call where the // TODO(L4) was.

const org = await db.query.organization.findFirst({
  where: eq(organization.id, ctx.orgId),
});
const orgName = org?.name ?? 'your organization';
const acceptUrl = await signedInviteUrl(invitationId, rawToken);


const sent = await sendEmail({
  to: email,
  subject: `You're invited to ${orgName}`,
  react: createElement(InviteEmail, {
    orgName,
    inviterName: ctx.user.name,
    role,
    acceptUrl,
    expiresAt: new Date(Date.now() + INVITATION_TTL_SECONDS * 1000),
  }),
  idempotencyKey: `invite:${invitationId}`,
});


await dispatch({
  type: 'org.invitation.sent',
  recipientUserIds: existingUser ? [existingUser.id] : [],
  subjectId: invitationId,
  payload: {
    invitedEmail: email,
    role,
    orgName,
    inviterName: ctx.user.name,
    acceptUrl,
  },
});


revalidatePath('/inspector');
return ok({ invitationId, emailSent: sent.ok });

The dispatch runs after withTenant has committed invitationId, so the notification can never fire for an invitation that rolled back. This is the fire-after-commit rule made structural: a notification has no chance to escape a transaction that fails.

const org = await db.query.organization.findFirst({
  where: eq(organization.id, ctx.orgId),
});
const orgName = org?.name ?? 'your organization';
const acceptUrl = await signedInviteUrl(invitationId, rawToken);


const sent = await sendEmail({
  to: email,
  subject: `You're invited to ${orgName}`,
  react: createElement(InviteEmail, {
    orgName,
    inviterName: ctx.user.name,
    role,
    acceptUrl,
    expiresAt: new Date(Date.now() + INVITATION_TTL_SECONDS * 1000),
  }),
  idempotencyKey: `invite:${invitationId}`,
});


await dispatch({
  type: 'org.invitation.sent',
  recipientUserIds: existingUser ? [existingUser.id] : [],
  subjectId: invitationId,
  payload: {
    invitedEmail: email,
    role,
    orgName,
    inviterName: ctx.user.name,
    acceptUrl,
  },
});


revalidatePath('/inspector');
return ok({ invitationId, emailSent: sent.ok });

The empty-array no-op. existingUser was resolved at the top of the action, so a non-user invitee dispatches to zero recipients and the dispatcher loops over nothing.

const org = await db.query.organization.findFirst({
  where: eq(organization.id, ctx.orgId),
});
const orgName = org?.name ?? 'your organization';
const acceptUrl = await signedInviteUrl(invitationId, rawToken);


const sent = await sendEmail({
  to: email,
  subject: `You're invited to ${orgName}`,
  react: createElement(InviteEmail, {
    orgName,
    inviterName: ctx.user.name,
    role,
    acceptUrl,
    expiresAt: new Date(Date.now() + INVITATION_TTL_SECONDS * 1000),
  }),
  idempotencyKey: `invite:${invitationId}`,
});


await dispatch({
  type: 'org.invitation.sent',
  recipientUserIds: existingUser ? [existingUser.id] : [],
  subjectId: invitationId,
  payload: {
    invitedEmail: email,
    role,
    orgName,
    inviterName: ctx.user.name,
    acceptUrl,
  },
});


revalidatePath('/inspector');
return ok({ invitationId, emailSent: sent.ok });

The dedup key for this event. The dispatcher’s notificationDedup window keys off subjectId, so this invitation can’t double-notify.

const org = await db.query.organization.findFirst({
  where: eq(organization.id, ctx.orgId),
});
const orgName = org?.name ?? 'your organization';
const acceptUrl = await signedInviteUrl(invitationId, rawToken);


const sent = await sendEmail({
  to: email,
  subject: `You're invited to ${orgName}`,
  react: createElement(InviteEmail, {
    orgName,
    inviterName: ctx.user.name,
    role,
    acceptUrl,
    expiresAt: new Date(Date.now() + INVITATION_TTL_SECONDS * 1000),
  }),
  idempotencyKey: `invite:${invitationId}`,
});


await dispatch({
  type: 'org.invitation.sent',
  recipientUserIds: existingUser ? [existingUser.id] : [],
  subjectId: invitationId,
  payload: {
    invitedEmail: email,
    role,
    orgName,
    inviterName: ctx.user.name,
    acceptUrl,
  },
});


revalidatePath('/inspector');
return ok({ invitationId, emailSent: sent.ok });

The payload: the fields the registry’s invite-sent templates consume. NotificationEvent carries no orgId, so the org name travels in the payload as copy.

const org = await db.query.organization.findFirst({
  where: eq(organization.id, ctx.orgId),
});
const orgName = org?.name ?? 'your organization';
const acceptUrl = await signedInviteUrl(invitationId, rawToken);


const sent = await sendEmail({
  to: email,
  subject: `You're invited to ${orgName}`,
  react: createElement(InviteEmail, {
    orgName,
    inviterName: ctx.user.name,
    role,
    acceptUrl,
    expiresAt: new Date(Date.now() + INVITATION_TTL_SECONDS * 1000),
  }),
  idempotencyKey: `invite:${invitationId}`,
});


await dispatch({
  type: 'org.invitation.sent',
  recipientUserIds: existingUser ? [existingUser.id] : [],
  subjectId: invitationId,
  payload: {
    invitedEmail: email,
    role,
    orgName,
    inviterName: ctx.user.name,
    acceptUrl,
  },
});


revalidatePath('/inspector');
return ok({ invitationId, emailSent: sent.ok });

Chapter 065’s invitation email, the one named exception to the seam. It predates the dispatcher and still fires on its own after commit, sending the invitee’s accept-link email directly.

1 / 1

The recipientUserIds: existingUser ? [existingUser.id] : [] ternary is worth slowing down on. An invitation can go to an email with no account yet — the normal case for inviting someone new — and that invitee has no user to notify in-app. The instinct is to guard with if (existingUser) dispatch(...); resist it. A loop over zero recipients returns { sent: 0, deduped: 0, suppressedByPrefs: 0 }, a clean no-op, so an empty array keeps the call site uniform: there is always one dispatch call, and the data decides whether anything happens. existingUser was resolved at the top of the action, by the lookup that also rejects an invite to a current member:

    const existingUser = await db.query.user.findFirst({
      where: eq(user.email, email),
    });

The only import this file gains is import { dispatch } from '@/lib/notifications'.

The role change: keep the audit write, add the notification

Section titled “The role change: keep the audit write, add the notification”

manage.ts is the same shape, with one thing to protect: the existing audit write. The role update and its logAudit(tx, ...) entry co-transact inside withTenant, deliberately, because a role that changed with no audit row is the exact gap a compliance table exists to close. Leave it alone; read the org name after the commit, then dispatch.

src/lib/invitations/manage.ts
    await withTenant(ctx.orgId, async (tx) => {
      await tx
        .update(member)
        .set({ role: newRole })
        .where(
          and(eq(member.id, memberId), eq(member.organizationId, ctx.orgId)),
        );
      await logAudit(tx, {
        action: 'member.role-changed',
        subjectType: 'member',
        subjectId: memberId,
        payload: { before: target.role, after: newRole },
      });
    });


    const org = await db.query.organization.findFirst({
      where: eq(organization.id, ctx.orgId),
    });
    const orgName = org?.name ?? 'your organization';


    await dispatch({
      type: 'org.member.role_changed',
      recipientUserIds: [target.userId],
      subjectId: memberId,
      payload: {
        newRole,
        before: target.role,
        orgName,
        actorName: ctx.user.name,
      },
    });


    revalidatePath('/inspector');
    return ok({ memberId, role: newRole });

The result is a deliberate dual write: logAudit records the change for compliance, dispatch tells the affected member their role moved. Two tables, two audiences, one action, and the test checks both: logAudit(tx, ...) still inside withTenant, dispatch(...) after it. The recipient is [target.userId], the one member whose role changed, and subjectId is the memberId so dedup keys on this membership. org is read from the global db after the transaction, not from tx: by then the commit is done and the org name is only a label for the copy. This file also gains the organization import.

The webhook: read inside the transaction, dispatch after it

Section titled “The webhook: read inside the transaction, dispatch after it”

This edit is split across two files because the read and the dispatch happen at different moments: the handler collects a descriptor inside the transaction, and the route drains it after the commit. Walk both tabs.

src/lib/webhooks/stripe.ts
export const dispatch = async (
  tx: Transaction,
  event: Stripe.Event,
  pendingDispatches: NotificationEvent[],
): Promise<void> => {
  switch (event.type) {
    case 'checkout.session.completed':
      await onCheckoutCompleted(tx, event);
      break;
    case 'customer.subscription.updated':
      await onSubscriptionUpdated(tx, event, pendingDispatches);
      break;
    case 'customer.subscription.deleted':
      await onSubscriptionDeleted(tx, event);
      break;
    default:
      log.info({ eventId: event.id, eventType: event.type }, 'unhandled');
      return;
  }
  log.info({ eventId: event.id, eventType: event.type }, 'dispatched');
};


export const onSubscriptionUpdated = async (
  tx: Transaction,
  event: Stripe.Event,
  pendingDispatches: NotificationEvent[],
): Promise<void> => {
  const sub = event.data.object as Stripe.Subscription;
  const patch = subscriptionToEntitlement(sub, loadCatalog());
  const eventAt = new Date(event.created * 1000);


  const updated = await tx
    .update(planEntitlements)
    .set({ ...patch, lastEventAt: eventAt })
    .where(
      and(
        eq(planEntitlements.subscriptionId, sub.id),
        or(
          isNull(planEntitlements.lastEventAt),
          lt(planEntitlements.lastEventAt, eventAt),
        ),
      ),
    )
    .returning({ organizationId: planEntitlements.organizationId });


  const row = updated[0];
  if (!row) {
    log.info({ eventId: event.id, subscriptionId: sub.id }, 'stale_ordering');
    return;
  }


  await logAudit(tx, {
    organizationId: row.organizationId,
    actorUserId: null,
    action: 'billing.subscription.updated',
    subjectType: 'subscription',
    subjectId: sub.id,
    payload: { plan: patch.plan, status: patch.status },
  });
  log.info(
    { eventId: event.id, orgId: row.organizationId, plan: patch.plan },
    'subscription_updated',
  );


  if (patch.status === 'past_due') {
    const org = await tx.query.organization.findFirst({
      where: eq(organization.id, row.organizationId),
    });
    const owners = await tx.query.member.findMany({
      where: and(
        eq(member.organizationId, row.organizationId),
        eq(member.role, 'owner'),
      ),
    });
    pendingDispatches.push({
      type: 'org.billing.past_due',
      recipientUserIds: owners.map((owner) => owner.userId),
      subjectId: sub.id,
      payload: {
        orgName: org?.name ?? 'your organization',
        plan: patch.plan,
      },
    });
  }
};

Collect-only: the handler never calls the notification dispatcher. Both dispatch and onSubscriptionUpdated gain a pendingDispatches: NotificationEvent[] param. When patch.status === 'past_due' (after the non-stale UPDATE returned a row), the handler reads the org and its owner members inside tx, then pushes one org.billing.past_due descriptor onto the array and returns.

The handler reads the owner ids inside tx, so they reflect the transition this transaction is committing; from the global db, a concurrent change could hand back owners that disagree with the past-due state you just wrote. It cannot fire there, so it pushes a descriptor onto the array and returns. The array lives in route.ts, declared before db.transaction so the closure can fill it, drained by a for...of loop once the transaction resolves. If the transaction throws, that loop never runs: rolled-back state has no descriptor to drain.

Two details to read closely:

  • dispatch names two functions here. route.ts already imports the Stripe router’s dispatch(tx, event, pendingDispatches), so it brings in the notification dispatcher under an alias, import { dispatch as dispatchNotification } from '@/lib/notifications', and drains with dispatchNotification(e). Different arities, different jobs; the alias keeps them apart.
  • The handler reads org and owners separately: two queries inside the transaction, both scoped to row.organizationId. The owner read filters on member.role === 'owner', and owners.map((owner) => owner.userId) becomes the recipient list. For the seeded Acme that is just Alice, so you see one row and one email; an org with three owners fans out to three.

The transactional-outbox alternative is the next reach. If the process crashes after the commit but before the drain loop runs, those notifications are lost with the in-memory array. The durable version writes the descriptors to a pending_dispatches table inside the same transaction, committing atomically with the state change, and a background worker drains that table, retrying on failure. The contract is identical — collect inside, dispatch after — so you can swap the array for a table later without touching a call site.

The seam check (requirement 6)

Section titled “The seam check (requirement 6)”

The sixth requirement is one the test suite cannot assert at runtime, because it is a property of the source, not of any single execution: outside lib/notifications/, the only sendEmail( call and the only db.insert(notifications) should be the one named exception. Run the search yourself:

Terminal window
grep -rn "db.insert(notifications)" src --include="*.ts" | grep -v "lib/notifications/"
grep -rn "sendEmail(" src --include="*.ts" | grep -v "lib/notifications/"

The first should return nothing: writeInboxChannel is the only writer of the notifications table, and it lives inside the seam. The second returns exactly one hit, chapter 065’s invitation email in send.ts — the seam holding. Anything else is a call site that learned to notify on its own instead of going through the dispatcher, the precise drift the seam exists to prevent.

For the pieces these edits lean on but do not re-teach: authedAction and withTenant come from The authedAction wrapper and the audit write from The append-only audit log; the webhook’s verify-then-claim-then-commit shape is Claim once, mutate once and its project version Claim the event inside one transaction; the dispatcher contract you are calling, NotificationEvent in and DispatchResult out, is One seam, many channels and this chapter’s Registry, dispatcher, and dedup.

Pattern: Transactional outbox
microservices.io

Chris Richardson's canonical writeup of the durable table this lesson's in-memory pendingDispatches array stands in for.

Moment of truth

Section titled “Moment of truth”

Run the lesson’s gate:

Terminal window
pnpm test:lesson 4

The suite drives all three call sites against the same local Postgres and email mock the app uses. The two action call sites run behind authedAction, which resolves a Better Auth session through next/headers, a path vitest’s node environment cannot enter. So for those, the suite drives the dispatcher with the exact event each call site builds and reads the wiring back out of the source: that dispatch sits after the commit, that the recipient list is the existing-user-or-empty ternary, that the audit write stays put. The webhook handler is a plain function, so the suite drives it directly with a real transaction and a fixture past-due subscription, then drains the descriptors the way the route does. It needs this chapter’s migration applied and pnpm db:seed run; Acme’s lone owner, Alice, is the fan-out fixture. A green run looks like this:

Terminal window
 ✓ tests/lessons/Lesson 4.test.ts (8)
   ✓ sendInvitation to an existing user notifies that user (2)
     ✓ produces one inbox row and one email increment for the invitee
     ✓ fires the dispatcher after the withTenant transaction commits
   ✓ inviting a non-user address no-ops the dispatcher (2)
     ✓ writes no inbox row and leaves the email counter flat on an empty recipient list
     ✓ builds the recipient list as the existing-user-or-empty ternary
   ✓ changeMemberRole writes both an audit row and a notification (2)
     ✓ produces one inbox row and one email increment for the affected member
     ✓ keeps the audit-log write inside the tx and dispatches after commit
   ✓ the past-due webhook fans out to every org owner after commit (1)
     ✓ collects an owner-targeted descriptor and dispatching it lands one inbox row + email per owner
   ✓ a rolled-back action notifies nobody (1)
     ✓ writes no inbox row and bumps no email counter when the transaction rolls back


 Test Files  1 passed (1)
      Tests  8 passed (8)

The suite reads rows and source, not the real surfaces. Walk them once by hand to confirm the production path does what the gate proves the wiring does:

Invite an existing user from the dashboard — an inbox row plus an email appear for the invitee, after the action commits.
untested
Invite a brand-new address — only chapter 065’s invitation email sends; the dispatcher no-ops on the empty recipient list and no inbox row lands.
untested
Change a member’s role — both an auditLogs row and a notifications row are written, with one email increment.
untested
Trigger a customer.subscription.updated with past_due status (via the stripe CLI or the inspector’s Fire billing-past-due) — processed_events, the entitlement, the audit row, and one notification per owner all land; replay the same event and the handler’s claim blocks it, with the dispatcher’s dedup as the second layer.
untested
Read the Wrap invite in rollback control’s note, and confirm through the tests that a rolled-back action writes no rows and bumps no counter.
untested
Run the seam check — sendEmail( and db.insert(notifications) outside lib/notifications/ return only the named chapter 065 invitation-email exception.
untested
Confirm in Drizzle Studio that the (userId, createdAt desc) index and the partial unread index back the inbox feed.
untested

This is where the build stops. What later chapters add on top of the same dispatcher contract:

  • Caching the inbox feed. Tag the feed read so a new notification busts exactly that cache; this project reads the table live.
  • Redis-backed dedup. Move the dedup window into Redis once a database write per check stops being free; only where isDuplicate looks changes.
  • A durable channel queue. Move channel sends behind a background worker so a slow email provider no longer blocks the action; the call site still just await dispatch(...).
  • Error and audit discipline. Formalize the channel-failure log line and the audit trail you have been writing.
  • Integration tests. Cover preferences-respected, default-on, dedup, and channel-independence against a real database.
  • Observability. Treat DispatchResult as the structured-log shape, with dashboards on dedup, suppression, and channel-failure rates.