@classytic/arc-notifications

The blessed seam from @classytic/arc domain events to @classytic/notifications. Every arc app used to re-derive the same 40 lines — build a NotificationService, subscribe app.events, map event payloads to recipients, thread idempotency for at-least-once transports, expose a template-drafting admin surface. This package is that wiring, once.

npm i @classytic/arc-notifications @classytic/notifications
# peers: @classytic/arc >=2.20, @classytic/primitives >=0.9, @classytic/repo-core >=0.7

The pieces

1. notificationsModule — events → notifications

Composed into createApp({ modules }). Declarative notifyOn triggers subscribe arc's event bus through subscribeWithBoundary (fire-and-forget; handler failures logged, not thrown), with idempotency keys defaulted from event identity so at-least-once redelivery never double-sends.

import { createApp } from "@classytic/arc/factory";
import { getModuleExports } from "@classytic/arc/factory";
import { notificationsModule, createDbFirstResolver } from "@classytic/arc-notifications";
import { EmailChannel } from "@classytic/notifications";

const app = await createApp({
  modules: [
    notificationsModule({
      channels: [new EmailChannel({ from: "App <[email protected]>", transport: smtp })],
      templates: createDbFirstResolver({ store: templateStore, fallback: codeTemplates }),
      notifyOn: [
        {
          event: "approval.requested",                        // arc event pattern (wildcards ok)
          template: "approval-requested",
          recipient: (e) => directory.contact(e.payload.approverId),
          data: (e) => ({ title: e.payload.title, link: `/approvals/${e.payload.id}` }),
          channels: ["email"],
        },
      ],
    }),
  ],
});

// service is decorated + exported
app.notifications?.send({ event: "manual", template: "welcome", recipient: { email }, data: {} });
const svc = getModuleExports<NotificationService>(app, "notifications");

A notifyOn trigger declared without arc's eventPlugin registered fails at boot (not silently at delivery time).

2. createDbFirstResolver — runtime-draftable templates

Admins draft templates at runtime; an active DB row wins over the code-defined fallback, and a store outage falls back to code (transactional email never stops). 60s read-through cache, invalidated by the templates resource on save.

const resolver = createDbFirstResolver({
  store: { getByKey: (key) => templateRepo.getByKey(key) }, // any by-key lookup
  fallback: createSimpleResolver(codeTemplates),            // from @classytic/notifications
  cacheTtlMs: 60_000,
});

3. templatesResource — the drafting surface

A normal arc resource over any RepositoryLike (mongokit / sqlitekit / prismakit / custom) — no kit adapter needed. Fail-closed: every op + route takes one permissions gate. Mounts admin CRUD plus GET /defaults (code templates to customize from) and POST /preview (stateless draft render). Write hooks invalidate the resolver cache so a saved draft applies within one request.

templatesResource({
  repository: templateRepo,
  permissions: platformAdminOnly(),
  defaults: codeTemplates,
  resolver, // the createDbFirstResolver instance — its cache is invalidated on save
});

4. In-app feed + realtime — the full inbox lifecycle

InAppChannel persists notifications into a host-owned feed repository; inAppFeedResource is the inbox REST surface (list / unread-count / mark-read / mark-all-read, plus DELETE /:id when the repository implements delete). Grounded against Puter's NotificationService — the lifecycle pieces a bare "push to a socket" setup lacks:

import {
  InAppChannel, inAppFeedResource,
  createUnreadCountCache, createInAppBackfill,
} from "@classytic/arc-notifications";

const unreadCache = createUnreadCountCache();       // badge count — hottest query, now cached
const port = mySseHub;                              // host-owned RealtimePushPort

const channel = new InAppChannel({
  repository: feedRepo,
  port,          // combined mode: push AFTER persist, WITH the row id on the wire
  unreadCache,   // new rows move the badge immediately
});

const feed = inAppFeedResource({
  repository: feedRepo,
  resolveScope: (req) => ({ userId: ..., organizationId: ... }),
  permissions: requireAuth(),
  port,          // mark-read/-all/delete push "notification:read" → other tabs sync instantly
  unreadCache,   // GET /unread-count reads through the shared cache
});

// Reconnect catch-up: push everything persisted while the user was offline.
// Requires feedRepo.claimUndelivered (ATOMIC fetch-and-mark-delivered).
const backfill = createInAppBackfill({ repository: feedRepo, port });
sseHub.onConnect((userId, orgId) => backfill.onConnect(userId, orgId)); // debounced 2s (multi-tab)

• Combined mode over a separate SseChannel when you persist: the kernel dispatches channels in parallel, so only persist-then-push can carry the feed-row id — which is what lets a live toast be marked read without a refetch. SseChannel remains for realtime-only setups.
• Delivered ≠ read. claimUndelivered tracks "has any client received this over realtime" (backfill's axis); the read flag tracks the user's action (the badge's axis). Both stay host-owned in the repository.
• All realtime is best-effort on top of the durable feed — a dead socket never fails a request or a send; clients converge via GET /notifications/.

5. Durable dispatch — crash-safe bulk sends (10k+)

The service's default queue is in-memory: a 10k birthday/newsletter send that crashes at #4000 loses the rest. There is one durable path — arc's jobs plugin. arc-notifications does not own a queue; background work in arc goes through @classytic/arc/integrations/jobs (an fp() plugin that decorates fastify.jobs, owns the BullMQ queues/workers, and adds an arc-level semaphore, cron reconciliation, event bridging, dead-letter queues, GET /jobs/:id/status, and onClose teardown). The bridge routes sends through it — so a durable notification send reuses the same job subsystem the rest of your app already runs. No second BullMQ, no second connection pool.

import { jobsPlugin } from "@classytic/arc/integrations/jobs";
import { notificationsJob, notificationsModule, createRepositoryDeliveryLog } from "@classytic/arc-notifications";

const notify = notificationsJob({ concurrency: 5, rateLimit: { max: 10, duration: 1000 } });

await createApp({
  plugins: async (f) => f.register(jobsPlugin, { connection: redis, jobs: [notify.job] }),
  modules: [
    notificationsModule({
      channels: [new EmailChannel({ from, transport })],
      queue: (f) => notify.toQueueAdapter(f.jobs),   // resolved at bootstrap — fastify.jobs exists by then
      deliveryLog: createRepositoryDeliveryLog(deliveryRepo),
      notifyOn: [ /* ... */ ],
    }),
  ],
});

Works without Redis — just don't wire the queue: omit queue (and don't register jobsPlugin) and sends run on the service's in-memory queue. The module warns loudly at boot if you declare notifyOn triggers without a durable queue, so "in-memory in production" is never silent.

Why a queue, not a workflow engine: a bulk send is a fan-out, not a multi-step workflow. One durable job per recipient (never one job for 10k) keyed by idempotencyKey (→ BullMQ jobId via fastify.jobs, so crash-recovery redelivery is a no-op, not a duplicate). Streamline (a workflow engine) is the right tool only when the campaign itself has steps/waits/approval gates — it would then orchestrate and delegate the sending to the jobs plugin exactly like this.

Durability levels: in-memory → lost on crash (dev only). fastify.jobs (BullMQ) → survives app crashes (add Redis AOF for Redis-crash safety). Plus a persistent DeliveryLog → observable + resumable (query the failed set, re-drive only those).

Composing with arc's outbox (strongest). The outbox and the queue are different layers and compose — they are not merged. Arc's EventOutbox guarantees the domain event survives (transactional same-DB-txn write + relay); the jobs queue guarantees the send survives. Wire the trigger to an outbox-relayed event and the whole chain is at-least-once end to end: business write + outbox event (one txn) → relay → event bus → notifyOn fires → durable send. idempotencyKey (defaulted from event identity) dedupes the at-least-once overlap at both hops.

BullMQ is a peer of arc's jobs plugin, not of arc-notifications — this package has no direct queue dependency.

Design rules

• Channels and stores come from @classytic/notifications (peer) — this package adds the arc-native durable QueueAdapter and repo-backed DeliveryLog, plus the event seam and templates; it never re-exports notifications' channels.
• One-way dependency: @classytic/arc never depends on this package.

Generalized from be-prod's production shared/notifications (db-first resolver) + resources/notifications (email-template model, admin resource, event-trigger registry).

Trademark

MIT-licensed code. "Classytic"/"arc" names + logos are trademarks of Classytic LLC — see TRADEMARK.md.