@alxia/events

An event bus for an application on @softistx/mongo and @softistx/redis, built on a transactional outbox. A module writes an event in the same MongoDB transaction as the change that caused it, so the two stand or fall together. A relay hands each event to the modules that subscribed, through queues; each subscriber applies it once, however many times it is delivered.

bun add @alxia/events @alxia/modules @alxia/tenancy @softistx/mongo @softistx/redis mongodb zod
bun add -d typescript

@softistx/mongo and @softistx/redis are private, on GitHub Packages. Map the scope in your project's bunfig.toml and export GITHUB_TOKEN, a token with read:packages:

[install.scopes]
"@softistx" = { url = "https://npm.pkg.github.com/", token = "$GITHUB_TOKEN" }

MongoDB must run as a replica set (transactions need one). See troubleshooting.

In 30 seconds

import { collectionsOf, defineModule } from '@alxia/modules';
import {
    createEvents,
    eventsModule,
    type EventEnvelope,
    type SubscriberContext,
} from '@alxia/events';
import { single } from '@alxia/tenancy';
import { connectMongo, defineMongo, openMongo } from '@softistx/mongo';

// `Tenant` and `Core` are your tenant and core collection wirings
// (see the guide, "Write a subscriber").

// 1. Declare what a module emits and what it subscribes to.
export const shop = defineModule({
    name: 'shop',
    emits: ['shop.order.placed'],
});

export const billing = defineModule({
    name: 'billing',
    subscribers: {
        'shop.order.placed': async (
            event: EventEnvelope,
            context: SubscriberContext<'billing', Tenant, Core>,
        ) => {
            // `context.mongo` is in the transaction that records the inbox row,
            // and narrowed to billing's collections and those of its dependsOn.
            const { mongo } = context;
            await mongo.db.invoices.create({ orderId: event.aggregate.id });
        },
    },
});

// 2. Put the kernel module in the registry you give the tenant store,
//    then build the bus from that store.
const modules = [eventsModule, shop, billing];
const { client } = await connectMongo(process.env.MONGO_URI!);
const mongo = await openMongo(
    defineMongo({ client, database: 'app', collections: collectionsOf(modules) }),
);
const tenancy = single({
    mongo,
    modules,
    tenant: { id: 'main', slug: 'main', name: 'Main' },
}); // or perOrganization(...), see @alxia/tenancy
const events = createEvents({ store: tenancy });
// `request` and `viewer` are the HTTP request and its signed-in account
const scope = await tenancy.scope(request, viewer);

// 3. Emit inside the transaction of the change.
await scope.mongo.transaction(async (tx) => {
    const order = await tx.db.orders.create({ label: 'Ada' });
    await events.emit(
        { tenant: scope.tenant, mongo: tx },
        {
            type: 'shop.order.placed',
            aggregate: { type: 'order', id: order.id },
            payload: { label: 'Ada' },
        },
    );
});

Delivery

In production, a relay and workers run beside the app. Each module that subscribes gets a queue, events.<module>:

import { bindQueue, connectRedis } from '@softistx/redis';
import { createEvents, eventsQueue } from '@alxia/events';

const redis = await connectRedis(process.env.REDIS_URL!);
const queues = { billing: bindQueue(redis.client, eventsQueue('billing')) };
const events = createEvents({ store: tenancy, queues }); // `tenancy` from above

const relay = events.relay({ redis: redis.client }); // one leader per deployment
relay.start();
const workers = events.work();

// on shutdown
await relay.stop();
await workers.stop();

Without Redis (dev, tests), run the subscribers in your process, in order:

await events.drain(scope.tenant.id); // omit the tenant to drain every tenant

Guarantees

• Atomic emit. The event is in _outbox if and only if the transaction that emitted it committed.
• At least once. The relay enqueues a job per (event, subscriber) with jobId = <eventId>:<module>:<type>, then marks the event dispatched. A crash in between enqueues the same ids again, which the queue ignores.
• Applied once. A subscriber's effects and its _inbox row commit in one transaction; a second delivery finds the row and does nothing.
• Order. Per tenant, the relay sends events in _id order. There is no global order, and a subscriber must tolerate reordering.
• Same on every store. Emit, relay and drain run unchanged on perOrganization and single (one database, one fixed tenant).

API

Peers: @alxia/modules, @alxia/tenancy, @softistx/mongo, @softistx/redis, mongodb, zod, typescript.

Documentation

• Guide: the envelope, the outbox, subscribers, the relay and workers, drain, dead letters, tenants, testing.
• Troubleshooting: an error message, and what to do about it.
• Roadmap: what is coming, and what is not planned.

License

MIT