Events

@tulip-systems/events is the domain event atom. Use it when the application wants to say: this happened.

Events are facts. Listeners react to those facts. Queue can be used underneath to make each listener execution durable.

When to use Events

Use Events when one domain action should trigger multiple independent reactions:

create a notification when a task is created
write history when a project status changes
notify watchers when a quotation is approved
update a search index after a customer changes
enqueue integration work without coupling it to the original mutation

Do not put the side effects directly in the mutation when they are independent reactions. Publish an event and let listeners own the reactions.

Entry Points

Define events and listeners from the package entrypoint:

import { createEventListeners, defineEvent } from "@tulip-systems/events";

Create the server event bus from the server entrypoint:

import { EventBus } from "@tulip-systems/events/server";

Use queue-backed dispatch from the queue adapter entrypoint:

import {
  createEventDispatchHandler,
  defineEventDispatchTask,
  queueEventAdapter,
} from "@tulip-systems/events/queue";

Define an Event

An event has a stable name, version, and schema.

// app/admin/projects/_lib/events.ts
import { defineEvent } from "@tulip-systems/events";
import { z } from "zod";

export const projectStatusChangedEvent = defineEvent("project.status-changed", {
  version: 1,
  schema: z.object({
    projectId: z.uuid(),
    key: z.string(),
    title: z.string(),
    previousStatus: z.enum(["active", "planned", "backlog", "completed", "canceled"]),
    nextStatus: z.enum(["active", "planned", "backlog", "completed", "canceled"]),
  }),
});

The event schema is the contract listeners receive. Increment version when the contract changes incompatibly.

Define Listeners

Listeners are plain exported objects. The object keys become part of the durable listener identity.

// app/admin/notifications/_lib/listeners.server.ts
import { projectStatusChangedEvent } from "@/app/admin/projects/_lib/events";

export const notificationEventListeners = {
  projectStatusChanged: projectStatusChangedEvent.listen(async ({ payload, event }) => {
    await db.insert(notifications).values({
      title: "Project status changed",
      body: `${payload.title} moved from ${payload.previousStatus} to ${payload.nextStatus}.`,
      eventId: event.id,
      eventName: event.name,
      eventVersion: event.version,
      listenerKey: "notifications:projectStatusChanged",
      source: "project",
      sourceId: payload.projectId,
      occurredAt: new Date(event.occurredAt),
    });
  }),
};

Keep listener files server-only when they perform server work such as database writes or provider calls.

Create the Listener Collection

Use createEventListeners to compose listener objects into one dispatchable collection.

// server/events/init.ts
import { createEventListeners } from "@tulip-systems/events";
import { notificationEventListeners } from "@/app/admin/notifications/_lib/listeners.server";

export const appEventListeners = createEventListeners({
  notifications: notificationEventListeners,
});

Nested object keys are joined with :. For the example above, the listener key is:

notifications:projectStatusChanged

The durable key also includes the event name and version:

notifications:projectStatusChanged:project.status-changed@1

Object keys are durable identifiers. Renaming notifications or projectStatusChanged changes the identity used by queued jobs.

Initialize EventBus with Queue

The recommended production setup uses Queue as the event adapter. Publishing an event enqueues one queue job per matching listener.

// server/events/init.ts
import { EventBus } from "@tulip-systems/events/server";
import {
  createEventDispatchHandler,
  defineEventDispatchTask,
  queueEventAdapter,
} from "@tulip-systems/events/queue";
import { createQueueHandlers } from "@tulip-systems/queue";
import { queue } from "@/server/queue/init";

export const eventDispatchTask = defineEventDispatchTask("events.dispatch");

export const events = EventBus.init({
  listeners: appEventListeners,
  adapter: queueEventAdapter({
    queue,
    task: eventDispatchTask,
  }),
});

export const eventDispatchTaskHandler = eventDispatchTask.handle(
  "dispatch-event-listener",
  createEventDispatchHandler({ listeners: appEventListeners }),
);

export const queueHandlers = createQueueHandlers([eventDispatchTaskHandler]);

If the app already has queue handlers, include eventDispatchTaskHandler in the same collection.

export const queueHandlers = createQueueHandlers([
  queueLogTaskHandler,
  eventDispatchTaskHandler,
]);

Publish an Event

Publish from server code after the domain write succeeds.

import { projectStatusChangedEvent } from "@/app/admin/projects/_lib/events";
import { events } from "@/server/events/init";

await events.publish(projectStatusChangedEvent, {
  projectId: result.id,
  key: result.key,
  title: result.title,
  previousStatus: previousProject.status,
  nextStatus: result.status,
});

The publish flow is:

The event schema validates the payload.
EventBus finds listeners for the event name and version.
queueEventAdapter enqueues one queue job per listener.
QStash calls the queue callback route.
The event dispatch task finds the listener by listener key and event version.
The listener receives { payload, event } and runs.

Why One Queue Job Per Listener

Each listener gets its own queue job so failures are isolated.

If notifications:projectStatusChanged fails, only that listener retries. Other listeners for the same event do not run again.

This keeps side effects safer than putting all listeners in one queue job.

Listener Keys and Dedupe

Event listener keys use : because they behave like paths:

notifications:projectStatusChanged

Some queue providers restrict dedupe ids. The queue adapter sanitizes the generated default dedupe key before sending it to Queue/QStash. Internal listener keys are not changed.

You can override queue options if needed:

queueEventAdapter({
  queue,
  task: eventDispatchTask,
  options: ({ eventId, listenerKey }) => ({
    dedupeKey: `${eventId}-${listenerKey.replace(/:/g, "_")}`,
    delay: "10s",
  }),
});

Events vs Queue

Concept	Meaning	Example
Event	This happened	`task.created`
Listener	React to what happened	`notifications:taskCreated`
Queue task	Do this work durably	`events.dispatch`

Events should describe the domain. Queue should execute work.

Current Scope

Events are typed with Zod schemas.
Listener paths are built from nested object keys.
Queue integration dispatches one queue job per listener.
Apps own persistence for event side effects such as notifications or history.
Apps should publish events after successful domain writes.