npm package: @l-etabli/events
Description: The purpose of this repository is to make it easy to setup event driven architecture using outbox pattern
Latest version: 0.8.1

# @l-etabli/events

TypeScript library for event-driven architecture with the **outbox pattern**.

Events are persisted in the same transaction as your domain changes, then published asynchronously with retry support.

## Installation

```bash
bun add @l-etabli/events
```

For Kysely/PostgreSQL:

```bash
bun add @l-etabli/events kysely pg
```

## Recommended API

The recommended way to model events is now a single canonical definitions object.

```ts
import {
  defineEvent,
  defineEvents,
  type InferEventsFromDefinitions,
} from "@l-etabli/events";

type Project = { id: string; name: string };
type ProjectContext = { projectId: string };
type ProjectMemberContext = { projectId: string; memberId: string };
type ProjectRole = "admin" | "editor";

const eventDefinitions = defineEvents({
  ProjectCreated: defineEvent<{ project: Project }, ProjectContext>({
    priority: 1,
  }),
  ProjectUpdated: defineEvent<{ project: Project }, ProjectContext>(),
  UserAddedToProject: defineEvent<
    { projectId: string; userId: string; role: ProjectRole },
    ProjectMemberContext
  >({
    priority: 20,
  }),
  PingSent: defineEvent<{ at: Date }>(),
});

type AppEvents = InferEventsFromDefinitions<typeof eventDefinitions>;
```

Benefits:

- one object to maintain
- topics derived from object keys
- `priority` lives in the same canonical definition
- no handwritten `GenericEvent<...> | ...` union in the app

## Quick Start

### 1. Setup infrastructure

```ts
import {
  createUserActor,
  createWorkerActor,
  createEventCrawler,
  createInMemoryEventBus,
  createInMemoryEventRepositoryAndQueries,
} from "@l-etabli/events";

const { eventQueries, withUow } =
  createInMemoryEventRepositoryAndQueries<AppEvents>();

const { eventBus, createNewEvent } = createInMemoryEventBus(withUow, {
  eventDefinitions,
});

const crawler = createEventCrawler({
  withUow,
  eventQueries,
  eventBus,
});
```

### 2. Subscribe to events

```ts
eventBus.subscribe({
  topic: "ProjectCreated",
  subscriptionId: "send-project-created-email",
  callBack: async (event) => {
    await emailService.send(event.payload.project.id);
  },
});
```

### 3. Emit events

```ts
await withUow(async (uow) => {
  await projectRepository.save(project);

  await uow.eventRepository.saveNewEventsBatch([
    createNewEvent({
      topic: "ProjectCreated",
      payload: { project },
      context: { projectId: project.id },
      triggeredByActor: createUserActor(currentUserId),
      flowId: requestId,
    }),
  ]);
});
```

### 4. Process events

Traditional server:

```ts
crawler.start();
```

Serverless:

```ts
await withUow(
  async (uow) => {
    await uow.eventRepository.saveNewEventsBatch([event]);
  },
  {
    afterCommit: async () => {
      await crawler.triggerProcessing();
    },
  },
);
```

## Event shape

`GenericEvent` remains the base type.

```ts
type GenericEvent<Topic, Payload, Context = undefined> = {
  id: EventId;
  occurredAt: Date;
  topic: Topic;
  payload: Payload;
  status: EventStatus;
  publications: EventPublication[];
  triggeredByActor: Actor;
  flowId?: string;
  causedByEventId?: EventId;
  priority?: number;
  context?: Context;
};

type Actor =
  | UserActor
  | SystemActor
  | WorkerActor
  | ApiKeyActor
  | AnonymousActor;
type UserActor<Id extends string = string> = { kind: "user"; id: Id };
type SystemActor = { kind: "system" };
type WorkerActor<Id extends string = string> = { kind: "worker"; id?: Id };
type ApiKeyActor<Id extends string = string> = { kind: "api-key"; id: Id };
type AnonymousActor = { kind: "anonymous" };
```

Rules:

- `payload` is always required
- `context` is only required at the call site for topics that declare one
- `priority` is injected automatically from `eventDefinitions` when provided

## Creating events

### From event definitions

```ts
import { createUserActor, makeCreateNewEvent } from "@l-etabli/events";

const createNewEvent = makeCreateNewEvent({
  eventDefinitions,
});

createNewEvent({
  topic: "ProjectCreated",
  payload: { project },
  context: { projectId: project.id },
  triggeredByActor: createUserActor(currentUserId),
});

createNewEvent({
  topic: "PingSent",
  payload: { at: new Date() },
  triggeredByActor: createWorkerActor("nightly-sync"),
});
```

### From an existing union

The previous union-based approach still works.

```ts
import { type GenericEvent, makeCreateNewEvent } from "@l-etabli/events";

type LegacyEvents =
  | GenericEvent<"UserCreated", { userId: string; email: string }>
  | GenericEvent<"OrderPlaced", { orderId: string }, { tenantId: string }>;

const createNewEvent = makeCreateNewEvent<LegacyEvents>();
```

## Event lifecycle

```text
never-published -> in-process -> published
                           \-> failed-but-will-retry -> published
                                                  \-> quarantined
```

Statuses:

- `never-published`
- `to-republish`
- `in-process`
- `published`
- `failed-but-will-retry`
- `quarantined`

## Kysely migration helper

```ts
import type { Kysely } from "kysely";

export async function up(db: Kysely<unknown>): Promise<void> {
  await db.schema
    .createTable("events")
    .addColumn("id", "text", (col) => col.primaryKey())
    .addColumn("topic", "text", (col) => col.notNull())
    .addColumn("payload", "jsonb", (col) => col.notNull())
    .addColumn("context", "jsonb")
    .addColumn("status", "text", (col) => col.notNull())
    .addColumn("triggeredByActor", "jsonb", (col) => col.notNull())
    .addColumn("flowId", "text")
    .addColumn("causedByEventId", "text")
    .addColumn("occurredAt", "timestamptz", (col) => col.notNull())
    .addColumn("publications", "jsonb", (col) => col.notNull().defaultTo("[]"))
    .addColumn("priority", "integer")
    .execute();
}
```

## Effect v4 support

Effect-native ports and adapters are available via dedicated subpath exports. Non-Effect users are not impacted — `effect` is an optional peer dependency.

```bash
bun add @l-etabli/events effect
```

### Setup

```ts
import { Effect } from "effect";
import type { InferEventsFromDefinitions } from "@l-etabli/events";
import {
  createEffectInMemoryEventBus,
  createEffectInMemoryEventRepositoryAndQueries,
  createEffectEventCrawler,
} from "@l-etabli/events/effect";

const { eventQueries, withUow } =
  createEffectInMemoryEventRepositoryAndQueries<AppEvents>();

const { eventBus, createNewEvent } = createEffectInMemoryEventBus(withUow, {
  eventDefinitions,
});

const crawler = createEffectEventCrawler({
  withUow,
  eventQueries,
  eventBus,
});
```

### Subscribe

```ts
eventBus.subscribe({
  topic: "ProjectCreated",
  subscriptionId: "send-project-created-email",
  callBack: (event) =>
    Effect.promise(() => emailService.send(event.payload.project.id)),
});
```

### Emit

```ts
await Effect.runPromise(
  withUow((uow) =>
    uow.eventRepository.saveNewEventsBatch([
      createNewEvent({
        topic: "ProjectCreated",
        payload: { project },
        context: { projectId: project.id },
        triggeredByActor: createUserActor(currentUserId),
      }),
    ]),
  ),
);
```

### Process

```ts
// Traditional server
crawler.start();

// Serverless / on-demand
await Effect.runPromise(crawler.triggerProcessing());
```

### Kysely adapters

For PostgreSQL with Effect-native Kysely:

```ts
import {
  createEffectKyselyEventRepository,
  createEffectKyselyEventQueries,
} from "@l-etabli/events/effect-kysely";

const eventRepository = createEffectKyselyEventRepository<AppEvents, Db>(db);
const eventQueries = createEffectKyselyEventQueries<AppEvents, Db>(db);
```

These adapters wrap Kysely queries with `Effect.promise()` — they work with any standard Kysely instance.

### Subpath exports

| Import path | What | Requires |
|---|---|---|
| `@l-etabli/events` | Promise-based ports, in-memory adapters, crawler | — |
| `@l-etabli/events/kysely` | Promise-based Kysely adapters | `kysely` |
| `@l-etabli/events/effect` | Effect-native ports, in-memory adapters, crawler | `effect` |
| `@l-etabli/events/effect-kysely` | Effect-native Kysely adapters | `effect`, `kysely` |

## Examples

See [`examples/`](./examples/) for Kysely integration, cascading events and serverless usage.