@ultimat3/notify

@ultimat3/notify

Notifications as one declaration: channel fan-out, a preference gate, a digest window, a delivery ledger and an in-app inbox.

notifier() is a job factory — it returns a JobHandle, so a notification inherits retry, the dead-letter path, cancellation, x jobs show and its manifest row. It is not a ninth primitive.

The declaration

import type { MailDefinition } from '@ultimat3/mail';
import { send } from '@ultimat3/mail';
import type { Recipient } from '@ultimat3/notify';
import { inAppChannel, mailChannel, notifier, t } from '@ultimat3/notify';

interface CommentPosted {
  readonly postId: string;
  readonly orgId: string;
  readonly author: string;
}

// The app's own template, and the app's own repository — `./mail.ts` and a service its boot
// installed. Neither ships here: this package knows a mailer and an audience, not your tables.
declare const postCommented: MailDefinition<CommentPosted>;
declare const posts: {
  subscribers(postId: string, signal: AbortSignal): Promise<readonly Recipient[]>;
};

export const commentPosted = notifier({
  name: 'post.commented',
  input: t.object({ postId: t.uuid, orgId: t.uuid, author: t.string }),
  // Required, exactly as on job(): the org this run acts under.
  tenant: (params) => params.orgId,
  // Required, exactly as job().idempotencyKey is — and it is the SAME value: the queue's dedupe
  // key and the delivery ledger's event column ask one question.
  key: (params) => `comment:${params.postId}`,
  // Resolved on the worker, inside a durable step. Omit it and every enqueue names its own.
  // `ctx.signal` is the run's cancellation — hand it to anything that can block.
  recipients: ({ input, ctx }) => posts.subscribers(input.postId, ctx.signal),
  deliver: [
    // Fires immediately: no wait.
    { channel: inAppChannel() },
    // Waits ten minutes, then re-reads the condition. A subscriber who muted the thread in
    // minute three gets nothing.
    {
      // The type argument is what types `mail.batch[n].params`: `NoInfer` on `deliver` stops the
      // channel from deciding what the params are — the schema above already did.
      channel: mailChannel<CommentPosted>({
        mailer: {
          send: async (mail) => {
            // `batch` is never empty: one event for an immediate send, the whole window for a
            // digest. Oldest first, so the newest is the last.
            for (const event of mail.batch) {
              await send(postCommented, event.params, { to: mail.to, locale: mail.locale ?? 'en' });
            }
          },
        },
      }),
      wait: '10m',
      unless: ({ event }) => event.params.author === 'system',
      // Coalesce every comment on this post for one person into one email per hour.
      digest: { window: '1h', group: (event) => event.params.postId },
    },
  ],
});

Enqueued like any other job:

import type { JobHandle } from '@ultimat3/jobs';
import type { NotifyPayload } from '@ultimat3/notify';

// The declaration above: `notifier()` returns a JobHandle, so this IS the job's own enqueue.
declare const commentPosted: JobHandle<
  NotifyPayload<{ postId: string; orgId: string; author: string }>
>;
declare const postId: string;
declare const orgId: string;
declare const author: string;

await commentPosted.enqueue({ params: { postId, orgId, author } });
// …or with the audience handed in, `noticed`'s `.deliver(recipients)`:
await commentPosted.enqueue({
  params: { postId, orgId, author },
  recipients: [{ id: 'u_1', to: '[e-mail]', locale: 'en-GB', tz: 'Europe/London' }],
});

The payload nests params rather than spreading it, so an app whose notification is about recipients cannot collide with the framework's own key.

Installing the stores

One call, at boot. Whole-object replacement, never a merge.

import type { PgExecutor } from '@ultimat3/jobs';
import {
  createMemoryDigestStore,
  createPgDeliveryLedger,
  createPgInboxStore,
  setNotifyStores,
} from '@ultimat3/notify';

declare const executor: PgExecutor;   // `@ultimat3/cli`'s pgExecutorFor(client)
// The app's preference table, behind whatever taxonomy it named.
declare const prefs: {
  allows(recipient: string, notifier: string, channel: string, at: Date): boolean;
};

setNotifyStores({
  ledger: createPgDeliveryLedger({ executor }),
  inbox: createPgInboxStore({ executor }),
  digest: createMemoryDigestStore(),
  preferences: {
    // The GATE ships; what it reads never does. Your taxonomy, your quiet hours.
    allows: ({ recipient, notifier: name, channel, ctx }) =>
      prefs.allows(recipient.id, name, channel, ctx.now()),
  },
});

Store	Default	Why
ledger	createMemoryDeliveryLedger()	one process is genuinely deduped by a heap map, and no ledger at all means a replay sends twice
preferences	allowAllPreferences()	denying by default is a notifier that silently delivers nothing
inbox	none — X_NOTIFY_STORE_MISSING	a message written to nowhere is worse than a refusal
digest	none — X_NOTIFY_STORE_MISSING	same

executor is a structural { query(sql, params) } — @ultimat3/cli's pgExecutorFor(client) over a DbClient is the framework's own. Bun.sql does not satisfy it.

Channels

import { bulkChannel, channel } from '@ultimat3/notify';

// The app's push vendor and its Slack webhook — a channel is where an SDK belongs.
declare const pushService: {
  send(token: string, payload: unknown, init: { signal: AbortSignal }): Promise<void>;
};
declare const SLACK_URL: string;

// One call per recipient. The retry unit is one address.
export const push = channel('push', async ({ recipient, event, signal }) => {
  await pushService.send(recipient.to ?? '', event.params, { signal });
});

// ONE call for the whole audience — a Slack post, a webhook, a digest to an ops channel.
export const slack = bulkChannel('slack', async ({ recipients, event, signal }) => {
  await fetch(SLACK_URL, {
    method: 'POST',
    body: JSON.stringify({ text: `${String(recipients.length)} watchers`, event: event.key }),
    signal,
  });
});

inAppChannel() and mailChannel({ mailer }) ship. mailChannel takes a structural Mailer — one method, no dependency on @ultimat3/mail, which is the same tier.

The inbox

import type { Ctx } from '@ultimat3/core';
import { requireInbox } from '@ultimat3/notify';

declare const ctx: Ctx;   // the request context; `ctx.actor` is who is reading

const store = requireInbox('post.commented');
const page = await store.list({ recipient: ctx.actor.id, limit: 50 });
const badge = await store.unreadCount(ctx.actor.id);
await store.markSeen({ recipient: ctx.actor.id, at: ctx.now() });
await store.markRead({ recipient: ctx.actor.id, ids: [page[0]?.id ?? ''], at: ctx.now() });

seenAt and readAt are two facts — showing the badge does not dismiss the message — and the unread count is derived from readAt is null, never stored, so the badge and the list cannot disagree. Every write is scoped by recipient: an id somebody else named is simply absent.

Delivery order

deliver entries are sorted by wait ascending and the fan-out sleeps the delta, so a channel with no wait fires immediately even when a later one waits an hour. if and unless are evaluated after the wait, on the attempt that delivers — a condition that goes false during a ten-minute delay sends nothing.

At-least-once

A job body runs before its checkpoint lands. Two layers stop a double send: the step checkpoint (deliver:<channel>:<recipient>), and the delivery ledger's atomic claim on (notifier, key, channel, coalesce(recipient, '')) — taken before the send, settled after. A claim that already reads sent answers false.

A bulk channel claims one row for the whole audience, with a null recipient: half a bulk POST is not a state this package can represent. That null is why the key coalesces rather than naming the column: NULLs are distinct in a plain unique index, so a bulk claim would otherwise be claimable without bound and every replay would re-send the whole audience.

Errors

Code	When
X_NOTIFY_CHANNELS_EMPTY	deliver: [] — refused at declaration
X_NOTIFY_CHANNEL_DUPLICATE	two deliveries name one channel; the ledger keys on it
X_NOTIFY_DIGEST_UNSUPPORTED	a digest window on a bulk channel
X_NOTIFY_FANOUT_TOO_WIDE	more recipients than maxRecipients (default 500)
X_NOTIFY_STORE_MISSING	an inbox or digest channel with no store installed
X_NOTIFY_DELIVERY_FAILED	a channel's deliver threw; the run retries on its policy

Boundary

Tier 4. May import tiers 0-3 — enforced by bun run boundaries. Its real imports are core, schema, time and jobs, so tier 4 is its floor. See CLAUDE.md.