@krizaka /发件箱
每个Krizaka产品的事务处理箱。一个事件是 写在调用者的事务中,因此它恰好在工作提交时存在，并且是 在提交之后提交，至少一次，由工人租用一批。无代理：PostgreSQL和 FOR UPDATE SKIP LOCKED 在实例之间共享工作。它取代了Akinaza的 mail_outbox (其 flushOutbox 没有人打电话)，分析人员将产品发送出去；它是的TypeScript版本 krizaka-messaging的 krizaka_outbox (信封是 id， type， occurredAt， payload）。
保证是 至少一次． 消费者在消息id上必须是幂等的。 在租赁有效期间，债权是排他性的。一个工人交付，然后崩溃之前
complete让租约到期，并传递消息 再次。outbox从不承诺 “一次”。
import { emitMigration, transaction } from "@krizaka/db";
import { consumeOnce, createOutbox, outboxSchema, runWorker } from "@krizaka/outbox";
import { readOutboxSettings } from "@krizaka/outbox/settings";
emitMigration(outboxSchema, 1, { tables: {} }); // commit the SQL once, in your migrations: outbox.message and outbox.processed
const outbox = createOutbox({ clock: { now: () => new Date() }, settings: () => readOutboxSettings((key) => settings.get(key)) });
// 1. in the use case: the message commits or rolls back with the work
await transaction(connect, async (tx) => {
const redemption = await coupons.redeem(tx, { code, accountId });
await outbox.append(tx, { topic: "credits", type: "evt.coupon.redeemed.v1", payload: { redemptionId: redemption.id }, id: redemption.eventId });
});
// 2. a worker the product starts (Next.js instrumentation.ts); the package has no timer, you pass `sleep`
const stop = new AbortController();
void runWorker({
outbox, query: pool.query.bind(pool) /* a pool-level Query */, owner: `${hostname}:${process.pid}`, topic: "credits", signal: stop.signal,
sleep: (seconds) => new Promise((resolve) => setTimeout(resolve, seconds * 1000)),
deliver: async (message) => { /* throw to retry */ },
});
// 3. a consumer with an effect in the database applies it once per message id, in the effect's transaction
await transaction(connect, (tx) => consumeOnce(tx, "credits", message.id, clock.now(), () => credits.grant(tx, …)));
append(tx, { topic, type, payload, id?, availableAt? })取交易的Query首先。一个给定的id使追加幂等 (同一事实两次是一个消息)。type是路由密钥，evt.<module>.<fact>.v1像Java的。claim(query, { owner, topic? })是UPDATE … WHERE id IN (SELECT … FOR UPDATE SKIP LOCKED)并设定租约。complete和fail(error)只为租约的所有者行事 (失去租约的工人不能覆盖新的持有者)。fail后退-availableAt = now + backoffSeconds × 2^(attempts−1)封顶和封顶后maxAttempts消息是 死了 (保留，连同它最后的错误；retryDead赋予它另一种生命)。一个消息 最后 允许尝试的租约到期，没有答案也死了，没有第九次交付。runWorker循环drainOnce(要求一批，按顺序交付，记录每个结果)，通过休眠sleep你通过的时候一轮没有发现任何东西，结束的时候信号中止。运行drainOnce从一个调度程序，而不是如果你喜欢cron一个循环。- 设置 （
@krizaka/outbox/settings,需要@krizaka/settings和zod):outbox.batchSize(50，1-500)，outbox.leaseSeconds（60,10 - 3600：比你最慢的交货时间长）outbox.maxAttempts1-50(8日),outbox.backoffSeconds(30，1-3600)，outbox.backoffMaxSeconds(3600)，outbox.retentionDays(7)。purge删除早于保留时间的已发送和失效消息: 从产品的作业中调用它。 - 表 （
outboxSchema、Postgres模式outbox):message（id, topic, event_type, payload, created_at, available_at, attempts, leased_until, lease_owner, delivered_at, dead_at, last_error）和processed(使用者，message_id)，其中consumeOnce使用: 标记和效果在同一个事务中提交，因此崩溃不会留下任何一个，并且重新交付会找到标记。 outboxStoreContract(从@krizaka/outbox/testkit)是每个商店的合同；memoryOutboxStore()用于单元测试。