Skip to content

Generic transactional outbox event processor with graceful shutdown and horizontal scalability

License

Notifications You must be signed in to change notification settings

dillonstreator/txob

Repository files navigation

txob

A generic transactional outbox event processor with graceful shutdown and horizontal scalability

Description

txob does not prescribe a storage layer implementation.
txob does prescribe a base event storage data model that enables a high level of visibility into event handler processing outcomes.

  • id string
  • timestamp Date
  • type (enum/string)
  • data json
  • correlation_id string
  • handler_results json
  • errors number
  • backoff_until Date nullable
  • processed_at Date nullable

txob exposes an optionally configurable interface into event processing with control over maximum allowed errors, backoff calculation on error, event update retrying, and logging.

As per the 'transactional outbox specification', you should ensure your events are transactionally persisted alongside their related data mutations.

The processor handles graceful shutdown and is horizontally scalable by default with the native client implementatations for pg and mongodb.

Installation

(npm|yarn) (install|add) txob

Examples

Let's look at an example of an HTTP API that allows a user to be invited where an SMTP request must be sent as a side-effect of the invite.

import http from "node:http";
import { randomUUID } from "node:crypto";
import { Client } from "pg";
import gracefulShutdown from "http-graceful-shutdown";
import { EventProcessor } from "txob";
import { createProcessorClient } from "txob/pg";

const eventTypes = {
  UserInvited: "UserInvited",
  // other event types
} as const;

type EventType = keyof typeof eventTypes;

const client = new Client({
  user: process.env.POSTGRES_USER,
  password: process.env.POSTGRES_PASSWORD,
  database: process.env.POSTGRES_DB,
});
await client.connect();

const HTTP_PORT = process.env.PORT || 3000;

const processor = EventProcessor(
  createProcessorClient<EventType>(client),
  {
    UserInvited: {
      sendEmail: async (event, { signal }) => {
        // find user by event.data.userId to use relevant user data in email sending

        // email sending logic

        // use the AbortSignal `signal` to perform quick cleanup
        // during graceful shutdown enabling the processor to
        // save handler result updates to the event ASAP
      },
      publish: async (event) => {
        // publish to event bus
      },
      // other handler that should be executed when a `UserInvited` event is saved
    },
    // other event types
  }
)
processor.start();

const server = http.createServer(async (req, res) => {
  if (req.url  !== "/invite") return;

  // invite user endpoint

  const correlationId = randomUUID(); // or some value on the incoming request such as a request id

  try {
    await client.query("BEGIN");

    const userId = randomUUID();
    // save user with userId
    await client.query(`INSERT INTO users (id, email) VALUES ($1, $2)`, [userId, req.body.email]);

    // save event to `events` table
    await client.query(
      `INSERT INTO events (id, type, data, correlation_id) VALUES ($1, $2, $3, $4)`,
      [
        randomUUID(),
        eventTypes.UserInvited,
        { userId }, // other relevant data
        correlationId,
      ],
    );

    await client.query("COMMIT");
  } catch (error) {
    await client.query("ROLLBACK").catch(() => {});
  }
}).listen(HTTP_PORT, () => console.log(`listening on port ${HTTP_PORT}`));

gracefulShutdown(server, {
  onShutdown: async () => {
    // allow any actively running event handlers to finish
    // and the event processor to save the results
    await processor.stop();
  }
});

other examples

About

Generic transactional outbox event processor with graceful shutdown and horizontal scalability

Topics

Resources

License

Stars

Watchers

Forks

Releases

No releases published

Packages

No packages published