@alxia/活动

用于应用程序的事件总线 @softistx/mongo 财务报告和审定财务报表 @softistx/redis,基于 交易发件箱。 模块在同一 MongoDB 交易中写入事件 作为导致它的变化, 所以两个站在一起或跌倒在一起。 一个接力手 每个事件通过队列订阅的模块; 每个订阅者 施用一次,无论交付多次.

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

@softistx/mongo 财务报告和审定财务报表 @softistx/redis 是私人的,在 GitHub 包。 绘制范围 项目的 bunfig.toml 和导出 GITHUB_TOKEN,与 read:packages数字 :

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

蒙戈DB必须运行为 复制集 (交易需要一个). 见 排除出问题。 。 。 .

30秒内

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' },
        },
    );
});

交付

在生产中,一个继电器和工人运行在应用程序旁边. 每个模块 订阅者得到一个队列, 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();

没有Redis(dev,测试),您会按以下顺序运行您的进程订阅者:

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

保证

• 原子发射. 该活动是在 _outbox 如果且只有在交易中 释放了它.
• 起码一次. 中继器为每个任务( 事件, 订阅者) 排入 jobId = <eventId>:<module>:<type>,然后标记已发送的事件。 撞车 在再次排入同一编号之间,队列忽略了这些编号.
• 申请一次. 用户的效果及其 _inbox 单行执行 交易; 第二次发送找到行并什么都不做.
• 肃静. 每个租户,中继发送事件 _id 秩序。 秩序。 没有 全球秩序,一个用户必须容忍重订顺序.
• 每家店都一样. 释放、中继和 drain 运行不变 perOrganization 财务报告和审定财务报表 single (一个数据库,一个固定租户).

API 密码

端点 : @alxia/modules, (中文(简体) ), @alxia/tenancy, (中文(简体) ), @softistx/mongo, (中文(简体) ), @softistx/redis, (中文(简体) ), mongodb, (中文(简体) ), zod, (中文(简体) ), typescript。 。 。 .

文档

• 指南信封、发包箱、订户、继电器和工人, drain死信 房客 测试.
• 解决问题:一个错误消息,以及该如何应对.
• 路线图:即将到来的,以及没有计划的.

许可证

麻省理工学院