@elysiumjs/moirai (英语)
采取的行动和查询 极乐世界 应用程序。 一项动作经过验证,在一个工作单位内运行,重复进行瞬间故障,并且可以被一分能. 它的副作用到一个与之相承的发包箱,消费者使用收件箱处理每个消息一次. 聚合物可以是事件源:它们的状态是它们流中事件的折叠,预测和反应堆跟随这些事件.
安装
bun add @elysiumjs/moirai @elysiumjs/mnemosyne
行动和询问
动作是写作: 扩展的类 Action,标记为 @Action.register()。查询是读取:扩展的类 Query,标记为 @Query.register()。 。 。 . Moirai.execute() 运行动作和 Moirai.ask() 运行查询.
import type { Static } from '@elysiumjs/core';
import { DomainError, Service, t } from '@elysiumjs/core';
import { Action, Moirai, Outbox, Query } from '@elysiumjs/moirai';
const TopUpInput = t.Object({ walletId: t.String(), amount: t.Integer({ minimum: 1 }) });
export type TopUpInput = Static<typeof TopUpInput>;
export type TopUpResult = { balance: number };
class WalletClosed extends DomainError.for('CONFLICT', 409) {}
@Action.register({
errors: [WalletClosed],
input: TopUpInput,
idempotent: true,
serializeBy: (input) => `wallet:${input.walletId}`
})
class TopUp extends Action<TopUpInput, TopUpResult, WalletClosed> {
constructor(@Service.inject() private readonly wallets: WalletRepository) {
super();
}
public override async handle(input: TopUpInput): Promise<TopUpResult | WalletClosed> {
const wallet = await this.wallets.find(input.walletId);
if (wallet.closed) return new WalletClosed(`wallet ${input.walletId} is closed`);
const credited = await this.wallets.credit(input.walletId, input.amount);
await Outbox.enqueue({ kind: 'wallet.credited', payload: { walletId: credited.id } });
return { balance: credited.balance };
}
}
const WalletBalanceInput = t.Object({ walletId: t.String() });
export type WalletBalanceInput = Static<typeof WalletBalanceInput>;
@Query.register({ errors: [], input: WalletBalanceInput })
class WalletBalance extends Query<WalletBalanceInput, { balance: number }> {
constructor(@Service.inject() private readonly wallets: WalletRepository) {
super();
}
public override async handle(input: WalletBalanceInput) {
return { balance: await this.wallets.balance(input.walletId) };
}
}
// `Moirai` runs them. In a controller, which returns the outcome: it answers the value, or the problem that refused the call.
return Moirai.execute(TopUp, body, { idempotencyKey: headers['idempotency-key'] });
return Moirai.ask(WalletBalance, { walletId });
Action<TInput, TResult, TErrors>财务报告和审定财务报表Query<TInput, TResult, TErrors>类型handle(): 这需要TInput并给出TResult或其中一个TErrors,它返回的域(默认情况下的任何域),或 aResult他们.Moirai.execute()财务报告和审定财务报表Moirai.ask()取出输入类型并返回类型handle()。 。 。 .- 拒绝是一种价值.
Moirai.execute()财务报告和审定财务报表Moirai.ask()给一个Outcome,则AsyncResult呼叫通知(见Result内@elysiumjs/core等待,这是Result,以处理器的值传递,或拒绝调用域名失败,其类型列出错误。 返回结果的控制器会解答值,或将错误作为路径问题发送,并输入客户端在其状态下查看.
const balance = await Moirai.ask(WalletBalance, { walletId });
if (!balance.check()) return balance.error;
balance.result.balance;
return Moirai.ask(WalletBalance, { walletId }).errorOr((balance) => view.render(balance));
- 与计划不符的输入用422拒绝
VALIDATION_FAILED问题,每个错误一个条目. - 何时
authorize(input)返回时false电话以403拒绝FORBIDDEN问题. - 动作在一个工作单位中运行,或者加入已经打开的动作. 在新的工作单位进行瞬间故障重试,最多3次尝试. 短暂的失败是错误
transient: true,还有Postgres系列化失败和僵局.retry: { attempts, isTransient }设定每个动作,例如重试返回的乐观-货币错误;retry: false从不重复。 超过尝试的瞬态域被还原 . - 接线员拒绝回话
DomainError失败Result,或投掷DomainError。工作单位然后滚回,不为一能而存储,错误被退回给呼叫者。 一个加入一个开放的工作单位的动作将它交接起来:当其开放者也返回时,这个单位会回滚. - 独能行动需要钥匙,否则就用400块拒绝
IDEMPOTENCY_KEY_REQUIRED问题。 重复使用相同的密钥并输入重放所存储的响应, 如 JSON; 重复使用另一个输入, 拒绝使用 422IDEMPOTENCY_KEY_REUSED问题. - 密钥属于其所有者: 默认的当前用户( U)
currentUserId()核部分,其他部分anonymous),或者说什么scope(input)行动或行动Moirai.execute(..., { scope })返回,例如租户和用户。 另一个显示相同密钥的拥有者独自运行动作. serializeBy发出一个接一个地共享一个密钥运行的呼叫,所以一个双回击等待第一个呼叫并重放其响应.errors需要: 它列出每个域handle()给,[]当它不提供,或登记不汇编.- 当呼叫运行时也会检查列表: 它不列出的域名, 如服务所抛出的域名, 或拒绝接通的嵌入式呼叫
orThrow(),是一个错误。 电话响起UndeclaredRefusalError,以拒绝为理由cause,而一条路线回答500而不是它从未宣布的问题。 Moirai本身的问题(无效输入,拒绝)authorize,一能键)总是被宣布. Moirai.lock(key)从处理器内部取出相同的锁,因为只有在处理器读出东西后才知道一个密钥. 它拿着钥匙直到工作结束 然后扔LockOutsideUnitOfWorkError在一个外面 各地按同顺序取出单位工作的锁.- 如果拒绝是错误的, 比如在工作或事件处理者,
orThrow()单独给出值并丢出域名 :const wallet = await Moirai.execute(OpenWallet, input).orThrow()。 。 。 .orError()给出值或域,以及map()财务报告和审定财务报表andThen()将进一步的步骤连起来而不等待第一. - 一个打过电话的处理者 拒绝使用
orPass():它只与被调用处理器的域名产生调用结果,并投出NestedCallRefusedError拒绝(无效输入、拒绝)authorize因为电话接线员建了电话 处理者可以按原样归还:return Moirai.execute(OpenSession, input).orPass()。 。 。 .Moirai.ErrorsOf<typeof Action>名称为处理器的域名,不包含moirai.
路线
result 财务报告和审定财务报表 errors 注册记录中描述的是一个电话给它开通的路线. errors 必须列出每个域 handle() 提供,或登记不汇编:
@Action.register({ input: TopUpInput, idempotent: true, result: Receipt, errors: [WalletClosed] })
class TopUp extends Action<TopUpInput, TopUpResult, WalletClosed> {}
@Http.post('/top-ups', { response: TopUp.result, errors: TopUp.errors })
public topUp(@Http.body(TopUpInput) body: TopUpInput, @Http.header('idempotency-key') key: string) {
return Moirai.execute(TopUp, body, { idempotencyKey: key });
}
静态 result 把计划还给我 静态 errors 列出已声明的类, 然后列出适用 moirai 的问题: 当存在输入方案时输入无效, 拒绝使用 authorize,以及"一能"行动的关键问题.
收发箱和收发箱
class Mailer {
@Outbox.handle('wallet.credited')
public async notify(payload: { walletId: string }) {
/* send the email */
}
}
// From a cron trigger, for rows that failed or were delayed:
await Outbox.sweep();
// In a consumer, inside its unit of work:
await Inbox.once({ messageId: message.id, consumer: 'billing' }, () => apply(message));
处理器也获得行和上下文. context.replacePayload(payload) 一旦处理器成功,就替换该行的有效载荷,在它所发送的标记工作单位中,例如,一旦发送就投放一次性代码。 发包店必须执行 replacePayload()就像记忆和Drizzle商店一样.
工作单位完成后,效果接力。 失败的一行被从5 s到1 h的后退重审,并在8次尝试后被标记为已死亡(onDead 被调用。 交付至少是一次,所以操作者必须容忍重复.
@elysiumjs/moirai/heracles 添加 job(JobClass, args),一旦行动发生,即发出一个Heracles任务:
import { job } from '@elysiumjs/moirai/heracles';
await Outbox.enqueue(job(SendReceiptJob, [order.id]));
@Once() 每一次发送编号: execute() 运行在工作单位中,并有ID的收件匣记录。 它适合一个工作 Scheduler,其发送的id名称 调度器,案件和滴答, 所以重复发送 或滴答处理两次 什么也没有.
效果处理器在模块导入时注册, @elysiumjs/moirai/heracles 包含。 一个转发出箱的过程,如克龙扫描器,会导入它继电器的效果的模块.
活动
一个事件说发生了什么。 它在促成它发生的工作单位中公布,每个操作员一旦该单位犯下,就得到它,从不回滚:
import type { EventMeta } from '@elysiumjs/moirai';
import { Event } from '@elysiumjs/moirai';
declare module '@elysiumjs/moirai' {
interface EventCatalogue {
'wallet.credited': { walletId: string; amount: number };
}
}
// In an action:
await Event.publish('wallet.credited', { walletId: wallet.id, amount: input.amount });
class Receipts {
@Event.on('wallet.credited', { consumer: 'billing.send-receipt', retry: { attempts: 5 } })
public async send(payload: { walletId: string; amount: number }, meta: EventMeta) {
/* ... */
}
}
EventCatalogue类型和有效载荷Event.publish()财务报告和审定财务报表@Event.on(). 虽然是空的,但接受任何名称和有效载荷。 声明一次,在一个文件中,Tsconfig包括,例如src/events.ts:声明适用于整个程序,因此发布或收听的文件不需要导入.- 每个操作员通过发包箱获得自己的交付,因此它会自己重复并死亡(
retry: { attempts }, (中文(简体) ),retry: false为了一次尝试, 否则maxRelayAttempts) (中文(简体) ). 送货至少一次. - 每个处理器将它所处理的事件记录在收件箱中,在自己写作的工作单位中,所以一次重复的发送无所作为.
consumer记录和默认名称Class.method:当方法被重新命名时保留. 两名操作员,一个是消费者DuplicateEventConsumerError。 。 。 . meta给出事件编号,名称,当它出版和尝试.- 事件需要
outbox财务报告和审定财务报表inbox商店。 Handlers在模块被导入时会注册,所以继电器导入的过程.
事件来源
存储事件
存储的事件是记录在一个集合的流中的事实:类扩展 StoredEvent 及其数据,以 @StoredEvent.register()。 。 。 .
import type { Static } from 'typebox';
import { StoredEvent } from '@elysiumjs/moirai';
import { Type as t } from 'typebox';
const Klaps = t
.Codec(t.String({ pattern: '^[0-9]+$' }))
.Decode((value) => BigInt(value))
.Encode((value: bigint) => value.toString());
const KlapsSpentData = t.Object({ spendId: t.String(), sellerId: t.String(), klaps: Klaps });
@StoredEvent.register({ name: 'KlapsSpent', version: 2, schema: KlapsSpentData })
export class KlapsSpent extends StoredEvent<Static<typeof KlapsSpentData>> {
// Version 1 kept the seller in `ref`.
@StoredEvent.upcast(1)
static fromV1({ ref, ...data }: { spendId: string; klaps: string; ref: { sellerId: string } }) {
return { ...data, sellerId: ref.sellerId };
}
}
name是每个事件保持的类型,默认为类名:当类被重新命名时保留。 两个有名字的班DuplicateStreamNameError。 。 。 .schema附加数据时进行编码(代码随后运行,例如将比特作为字符串保存),加载时进行解码;拒绝提供不匹配的数据。 如果没有一个计划,数据将保持原样,而且必须是JSON安全.version是数据版本,默认为 1。 以@StoredEvent.upcast(n)获取版本所存储的数据n并给出版本数据n + 1; 从存储的版本到当前版本的上下架链. 存储的未读类丢出的事件UnknownStoredEventError。 。 。 .- 单位
development财务报告和审定财务报表test数据被冻结.
合计数
集合是一个扩展的类 Aggregate,标记为 @Aggregate.register()。它的公开方法是命令:它们对照当前状态检查规则,然后用 record(),或者返回拒绝命令的域名。 标记方法 @Aggregate.evolve() 给下一个州 从州和事件; 他们可能受到保护或私人.
import { Aggregate } from '@elysiumjs/moirai';
type WalletState = { klaps: bigint };
@Aggregate.register({ name: 'wallet' })
export class Wallet extends Aggregate<WalletState> {
static override initialState: WalletState = { klaps: 0n };
get balance(): bigint {
return this.state.klaps;
}
spend(spendId: string, sellerId: string, klaps: bigint) {
if (klaps > this.state.klaps) return new InsufficientKlaps(klaps - this.state.klaps);
this.record(new KlapsSpent({ spendId, sellerId, klaps }));
}
@Aggregate.evolve(KlapsSpent)
protected spent(state: WalletState, event: KlapsSpent): WalletState {
return { klaps: state.klaps - event.data.klaps };
}
}
- 记录的事件一并被应用,所以记录出两个事件的命令检查第二个事件与第一个事件之后的状态.
- 状态是一个值:进化处理器给出了一个新的值. 单位
development财务报告和审定财务报表test,它被冷冻, 所以一个突变抛出。 没有进化处理器的事件离开状态. name是每个事件保持的集合类型,默认为类名.
Streams.act() 在开放的工作单元中运行一个命令,或者一个新的命令:它加载流,运行命令,附加它在下个版本所记录的事件,然后运行每个事件的内线预测和反应堆. 它给 AsyncResult 指挥系统.
import { Streams } from '@elysiumjs/moirai';
// In an action:
const spent = await Streams.act(Wallet, userId, (wallet) => wallet.spend(spendId, sellerId, 120n));
if (!spent.check()) return spent.error;
return { balance: spent.result.aggregate.balance, spends: spent.result.of(KlapsSpent).length };
- 返回域错误或失败的命令
Result无附件,结果act()以错误失败 . - 当另一个作家自被装入后附在溪流上时,该附件拒绝使用
ConcurrencyConflictError,这是短暂的:一个动作在新的工作单元中重试它,它再次加载了溪流. - 每个事件保存元数据:
eventMetadata选项,然后metadata授予act()。 。 。 . Streams.retrieve(Wallet, id)装入集合读取状态.
Aggregate.given() 运行根据过去事件创建的集合的指令,而不设置存储 :
Wallet.given([new KlapsToppedUp({ topUpId: 't-1', klaps: 100n })])
.when((wallet) => wallet.spend('s-1', 'seller-1', 30n))
.assertRecorded(new KlapsSpent({ spendId: 's-1', sellerId: 'seller-1', klaps: 30n }));
Wallet.given()
.when((wallet) => wallet.spend('s-1', 'seller-1', 30n))
.assertRefused(InsufficientKlaps);
assertRecorded() (一) 进行活动,在他们的班级和数据上进行匹配; assertNothingRecorded() 财务报告和审定财务报表 assertRefused() 完成它,和 recorded 财务报告和审定财务报表 outcome 给剩下的.
预测
投影会从存储的事件中写出读取模型 。 内线投影在附加事件的工作单位中运行,所以在命令看到它做了什么之后右侧读取,它的失败将命令卷回.
import type { StoredEventMeta } from '@elysiumjs/moirai';
import { Service } from '@elysiumjs/core';
import { Projection } from '@elysiumjs/moirai';
@Projection.inline({ name: 'wallet-movements', version: 1, models: [WalletMovementModel] })
export class WalletMovements {
constructor(@Service.inject() private readonly movements: WalletMovementRepository) {}
@Projection.on(KlapsSpent)
protected async spent(event: KlapsSpent, meta: StoredEventMeta) {
await this.movements.insert({
id: event.data.spendId,
userId: meta.aggregateId,
klaps: -event.data.klaps
});
}
}
- 掌上电脑在注册该类的模块中运行,具有其依赖性,并可能受到保护或私人保护.
meta给出流,版本,当事件发生时,它的元数据,以及它是否被重放. Projection.rebuild(WalletMovements)重建一个工作单位的投影:重放models与projections存储,按日志顺序重放其处理的每个事件,并记录其version。在一个交易存储上,读者可以看到以前的读取模式,直到重建完成.Projection.stale()列出预测version和他们最后建造的那个不一样 这个projection:rebuild命令MoiraiModule重建它们,或者用--name无论它是什么版本.Projection.project(WalletMovements, [[event, meta]])运行特定事件的处理器,例如在测试中.
反应堆
一个反应堆将一个存储的事件变成它除了状态之外的原因:为其他模块发布一个事件,为一个任务排队,运行一个动作. 反应堆方法属于任何服务,标记为: @Reactor.on()数字 :
import type { StoredEventMeta } from '@elysiumjs/moirai';
import { Service } from '@elysiumjs/core';
import { Event, Reactor } from '@elysiumjs/moirai';
@Service.register()
export class GiftReactor {
@Reactor.on(KlapsSpent, { when: (event) => event.data.sellerId !== 'house' })
protected async gifted(event: KlapsSpent, meta: StoredEventMeta) {
await Event.publish('KlapsGifted', {
buyerId: meta.aggregateId,
klaps: event.data.klaps.toString()
});
}
}
- 反应器运行在附带事件的工作单位中,在内线预测后,其效应随事件而发生. 当事件被重播时它永远不会运行.
when过滤其反应的事件.- 被存储的事件留在其模块内,并用上声改变形状. 一个发布
Event也就是其他模块如何了解所发生的情况, 其有效载荷是他们的合同.
存储
import { MoiraiModule } from '@elysiumjs/moirai';
import {
DrizzleEventStore,
DrizzleIdempotencyStore,
DrizzleInboxStore,
DrizzleOutboxStore,
DrizzleProjectionStorage,
moiraiTables,
PgAdvisoryLock
} from '@elysiumjs/moirai-drizzle';
export const billing = moiraiTables('billing');
MoiraiModule.with({
idempotency: () => new DrizzleIdempotencyStore({ table: billing.idempotencyKeys }),
outbox: () => new DrizzleOutboxStore({ table: billing.outbox }),
inbox: () => new DrizzleInboxStore({ table: billing.inbox }),
lock: () => new PgAdvisoryLock(),
events: () => new DrizzleEventStore({ table: billing.events }),
projections: () => new DrizzleProjectionStorage({ table: billing.projections }),
// On Cloudflare Workers, answer before the relay that follows a commit:
defer: (work) => Workers.context().waitUntil(work)
});
记忆库MemoryIdempotencyStore, (中文(简体) ), MemoryOutboxStore, (中文(简体) ), MemoryInboxStore, (中文(简体) ), MemoryActionLock, (中文(简体) ), MemoryEventStore 财务报告和审定财务报表 MemoryProjectionStorage)是用于测试和单进程应用. 内存放出箱、收件箱、事件存储和投影存储会像数据库表一样,将卷回的单位工作放下; MemoryProjectionStorage 重置带有它给定的函数的模型 .
模块存储
一个模块可以拥有自己的存储,比如一个域的发件箱和收件箱表格. Moirai.stores() 将它们列为模块的私人服务; 模块中遗漏的内容来自 MoiraiModule商店和选项都一样.
const billing = moiraiTables('billing');
@Module.register({
controllers: [InvoiceController],
services: [
Charge,
Receipts,
...Moirai.stores({
outbox: () => new DrizzleOutboxStore({ table: billing.outbox }),
inbox: () => new DrizzleInboxStore({ table: billing.inbox }),
relayBatch: 20
})
]
})
class BillingModule extends Module {}
- 一个动作运行在注册它的模块中,无论谁称它:它的一能库,它的锁,它的选项,它的效应和事件的发包箱就是该模块的.
- 一个事件通过发布它的模块的发件箱. 一个
@Event.on处理器在登记其类的模块中运行:它所处理的事件被记录在该模块的收件箱中,它本身的效果和事件会到该模块的出件箱去. 事件发件箱的模块中没有模块注册的处理器. - A 类
@Once()任务在注册任务的模块的收件箱中记录其发送id. - 承诺后继的中继器 排出各行.
Outbox.relay({ module })财务报告和审定财务报表Outbox.sweep({ module })中继一个模块的发件箱;Outbox.sweep()中继发件箱MoiraiModule和每个模块中有一个有它自己的, 所以一个克龙触发器覆盖他们全部. - 与任何服务一样,模块存储在输出时会被导入模块的模块所看到.