@nxgt / janus-webhooks

@nxgt/janus 用户事件, 作为签名的webhook交付 标准的人 规格:每 用户事件被发送到你的端点，用HMAC-SHA256签名，重试 失败——通过一个可能比流程更持久的队列——和每次交付 放弃了就报告，放弃了就不吭声。接收方检查 签名： verifyWebhook，或使用任何标准Webhooks库。

import { createMemoryStores, janus, scryptHasher } from '@nxgt/janus';
import { webhooks } from '@nxgt/janus-webhooks';
import { z } from 'zod';

const secret = process.env.WEBHOOK_SECRET; // whsec_…, from mintWebhookSecret()
if (!secret) throw new Error('WEBHOOK_SECRET is not set');

const listener = webhooks({
    endpoints: [{ url: 'https://crm.example.com/hooks/janus', secrets: [secret] }],
});

export const auth = janus({
    user: z.object({ email: z.email() }),
    password: { login: 'email' },
    store: createMemoryStores(),
    hasher: scryptHasher(),
    events: listener, // every user.created, user.emailVerified, user.passwordReset, user.passwordChanged, user.emailChanged, user.secondFactorEnabled, user.secondFactorDisabled, user.recoveryCodesRegenerated, user.recoveryCodeUsed, user.newDeviceSignedIn, user.avatarChanged, user.avatarRemoved, user.deleted
});

process.on('SIGTERM', async () => {
    await listener.close(); // waits for requests in flight, gives up the deliveries waiting for a retry
    process.exit(0);
});

0. x。 一个小版本仍然可能改变表面；更新日志告诉我们该怎么做。

安装

bun add @nxgt/janus-webhooks @nxgt/janus
bun add zod # the schema of the examples; any Standard Schema library will do
bun add -d typescript

两个对等体都需要： @nxgt/janus 0.17 -确切的范围是在 peerDependencies 和 typescript （6或7）。 @nxgt/janus 是一个 同行，所以它的一个副本定义 UserEvent。就像 @nxgt/janus，它期望 "moduleResolution": "bundler". 它使用 node:crypto，全球 fetch 和 process.emitWarning：包子或 节点。

子路径

适配器的规范是在每个情况下对一个新的空队列运行套件：

import { describe, it } from 'bun:test';
import { describeWebhookQueues } from '@nxgt/janus-webhooks/conformance';

describeWebhookQueues({
    name: 'my queue',
    runner: { describe, it }, // under bun test, pass them: they are not on globalThis
    harness: {
        async open() {
            const db = await openEmptyDatabase();
            return {
                queue: createMyWebhookQueue(db),
                faults: { fail: async (method) => db.failEvery(method) }, // leave out, and the outage cases pass with a JANUS_CONFORMANCE_SKIPPED warning
                close: () => db.drop(),
            };
        },
    },
});

这个 队列指南 每一个案例 失败的是什么？

使用

发送, webhooks()

webhooks({ endpoints }) 听众回答道 janus({ events }) 需要, 与一个 close() 为关闭。它将每个用户事件发送到每个端点 谁的 types 包括它-所有13种类型 types 缺席:

import { webhooks } from '@nxgt/janus-webhooks';

const listener = webhooks({
    endpoints: [
        { url: 'https://crm.example.com/hooks/janus', secrets: [crmSecret] },
        { url: 'https://search.example.com/hooks', secrets: [searchSecret], types: ['user.deleted'] },
    ],
    retries: ['5s', '5m', '30m', '2h', '5h', '10h', '10h'], // the default: at once, then these
    timeout: '10s', // the default, per request
    onGivingUp: async (delivery, reason) => {
        // reason.why: 'retriesRanOut' | 'closed' | 'endpointRemoved'; delivery.attempts: how many requests were sent
        await deadLetters.insert({ eventId: delivery.event.id, endpoint: delivery.endpoint, ...reason });
    },
});

这个 发送指南 所有的选择，重试 时间表，何时放弃交付以及为什么放弃，电汇格式和测试。

〇在重启中幸存下来 queue

没有一个 queue，交付在进程的内存中等待：崩溃会丢失 重试仍在等待中。通过一项 WebhookQueue 每个进程共享，a 一个进程失败的重试被下一个进程发送：

import { createMemoryWebhookQueue, webhooks } from '@nxgt/janus-webhooks';

const secret = process.env.WEBHOOK_SECRET; // whsec_…, from mintWebhookSecret()
if (!secret) throw new Error('WEBHOOK_SECRET is not set');

// The reference, in memory: swap in a durable adapter of the port — such as
// createRedisWebhookQueue(redis) from @nxgt/janus-webhooks-redis — for a
// queue that survives a restart.
const queue = createMemoryWebhookQueue();

const listener = webhooks({
    endpoints: [{ id: 'crm', url: 'https://crm.example.com/hooks/janus', secrets: [secret] }],
    queue, // the listener now awaits the insert — one call to the queue, never an endpoint
});

process.on('SIGTERM', async () => {
    await listener.close(); // gives nothing up: what waits is sent by the next process
    process.exit(0);
});

createMemoryWebhookQueue() 是港口的参考，和 @nxgt/janus-webhooks/conformance 适配器运行的套件。这个 队列指南 有合同，租约，端点id， 以及如何编写和测试适配器。

收到, verifyWebhook()

verifyWebhook 回答了 UserEvent 请求携带或 null 当它 没有一个端点的秘密被签名。在一个fetch风格的处理程序中：

import { verifyWebhook } from '@nxgt/janus-webhooks';

export async function receive(request: Request): Promise<Response> {
    const event = verifyWebhook({
        secrets: [secret],
        headers: request.headers,
        body: await request.text(), // the raw body: never request.json()
    });
    if (event === null) return new Response(null, { status: 401 }); // forged, altered, replayed late, or not a user event
    if (await handled.has(event.id)) return new Response(null, { status: 204 }); // a retry of one already handled

    await onUserEvent(event); // { id, type, occurredAt, userId, userType }
    await handled.add(event.id);
    return new Response(null, { status: 204 });
}

这个 接受指导 有选择的理由吗 用于 null， Hono和Node处理程序，处理一次事件，以及一个测试。

秘密, mintWebhookSecret()

import { mintWebhookSecret } from '@nxgt/janus-webhooks';

mintWebhookSecret(); // 'whsec_…': 32 random bytes, base64 — give the same one to the sender and the receiver

来自另一个标准Webhooks库的秘密被接受 它可以保存24到64字节 whsec_，范围规格成套。

API

陷阱

重试在内存中等待，除非传递 queue. 没有它，就会崩溃， 或者一个没有 close()，失去了每一次重试，仍然等待，一无所获 稍后发送。与一个 queue 每个进程共享一个进程留下的东西 等待-重试，请求被崩溃打断-下一个发送。

至少一次，即使是跨重启。 在A项下要求交货 在请求中途死亡的进程不会损失任何东西：一旦该请求过期 失效后，另一个进程会再次发送它——所以接收器可能会看到两次， 而坠机只需要一次尝试。接收方重复数据删除 webhook-id，因为他们已经必须进行重试。一个 lease 至少是 timeout 再加上一秒钟：工人在报告失败之前扩展了它， 所以慢一点 onGivingUp 不将交付交给另一个进程。

呼叫 close() 在关闭。 没有一个 queue，它首先发送每个 交货期： close() - - - - - - concurrency 一次，所以一个事件发生了一会儿 每个繁忙的槽位仍然得到它的尝试-等待请求进入 并放弃等待重试的交付，每个报告为 closed ——没有它，他们一言不发地消失了。有一个，它等待 请求在飞行中，不放弃任何东西。 process.on('SIGTERM', () => listener.close()).

侦听器只等待插入操作。 每次送货都要经过 队列;侦听器等待插入- a Map 在内存中，一个调用 queue -然后返回，第一个请求立即开始，所以 signUp 从不等待端点。与一个 queue，插入失败被拒绝； 和 janus 报告为 JANUS_EVENT_FAILED 使用事件的id。重试的 定时器不保持进程打开结束一个脚本 await listener.close().

端点的id是队列知道它的依据。 没有一个 id，这是一个 URL的散列：在等待交付的情况下更改端点的URL 让他们没有终点，之后 orphanGrace （24小时） 作为 endpointRemoved, delivery.url null。给每个端点一个 id 当你第一次通过 queue：稍后添加id也会改变id。

重定向是失败的，而且只有 2xx 是成功的。 重定向不是 遵循:点 url 在最后的演讲中， https:// - - - - - - http:// 只 localhost.

在进行任何解析之前，验证原始正文。 签名覆盖字节数 发送: await request.text(),从来没有 request.json() 也不是主体解析器 重新连载了它。

容差内的重播进行验证。 请求被复制并再次发送 在 toleranceSeconds （300）通过，您已经重试的一个也通过 已处理：保持id已处理 event.id, webhook-id 标题- and 用a回答第二题 2xx什么都不做。

用四个步骤旋转一个秘密。 双方签字 （secrets: [old, next])，让接收者都接受，然后丢掉旧的 来自发送者，最后来自接收者。

secret的长度为24 ~ 64字节。 过去的64年, webhooks() 和 verifyWebhook() 扔一个 TypeError -从0.4.0开始早期版本的长度为24。 以前的较长的秘密在升级时被拒绝，所以要远离它 首先，在运行mint的版本上 mintWebhookSecret()，旋转到 它在上面的四个步骤，然后升级。

先升级接收方，再升级发送方。 一个接收器 verifyWebhook 答案 null 对于它不知道的类型，所以传递失败，直到它知道 〇放弃 @nxgt/janus-webhooks 在0.3.0之前，这两者都不知道 user.secondFactorEnabled 也不 user.secondFactorDisabled，以及0.4.0之前 既不 user.recoveryCodesRegenerated 也不 user.recoveryCodeUsed, 在0.6.0之前 user.passwordChanged 也不 user.emailChanged, 在0.7.0之前不 user.newDeviceSignedIn，以及0.8.0之前的版本 user.avatarChanged 也不 user.avatarRemoved。在每个接收器都升级之前 给它的端点 types 它知道:

webhooks({
    endpoints: [{ url, secrets: [secret], types: ['user.created', 'user.emailVerified', 'user.passwordReset', 'user.deleted'] }],
});

该列表适用于0.3.0之前的接收器。0.3分之一。X，它知道 第二个因素的事件，而不是恢复代码，加上这两个：

types: ['user.created', 'user.emailVerified', 'user.passwordReset', 'user.secondFactorEnabled', 'user.secondFactorDisabled', 'user.deleted'],

1对0.4。X或0.5。X，它知道恢复代码的事件，但不知道 变化事件，加上这两个

types: ['user.created', 'user.emailVerified', 'user.passwordReset', 'user.secondFactorEnabled', 'user.secondFactorDisabled', 'user.recoveryCodesRegenerated', 'user.recoveryCodeUsed', 'user.deleted'],

1 / 0.6。X知道变化事件，但不知道新设备的变化， 加上这两个：

types: ['user.created', 'user.emailVerified', 'user.passwordReset', 'user.passwordChanged', 'user.emailChanged', 'user.secondFactorEnabled', 'user.secondFactorDisabled', 'user.recoveryCodesRegenerated', 'user.recoveryCodeUsed', 'user.deleted'],

user.passwordChanged 张贴的时候没有 first （在没有密码的帐户上设置的第一个密码）：接收方无法区分它和更改。

user.emailChanged 张贴的时候没有 formerEmail. @nxgt/janus 将用户之前的地址交给听者，这样通知就可以到达 收件箱; webhooks() 在队列之前丢弃它，因此没有端点、队列或 onGivingUp 报告中没有地址。接收器通过读取用户 userId 对于任何类型；把通知寄到原来的地址 janus({ events }) 而不是听者，旁边 webhooks()：

const auth = janus({
    ...config,
    async events(event) {
        await listener(event); // webhooks(): every type, formerEmail dropped
        if (event.type === 'user.emailChanged' && event.formerEmail != null) {
            const user = await auth.get(event.userId);
            await mail.emailChanged({ name: user.name, formerEmail: event.formerEmail, newEmail: user.email });
        }
    },
});

user.avatarChanged 和 user.avatarRemoved 张贴没有他们的 formerAvatarId，出于同样的原因：图像id保留在进程中。 中删除前一个图像 janus({ events }) 侦听器。

user.newDeviceSignedIn 张贴的时候没有 sessionId，为了同样的目的 原因：会话id还在进程中。发送新签到通知 这个 janus({ events }) 听者，或从签到者自己的回答，其 newDevice 所以说。

警告会指定URL的来源，而不是URL。 JANUS_WEBHOOK_GAVE_UP 写 https://crm.example.com 因为路径或查询可能包含的令牌 接收机的; onGivingUp 接收全部 delivery.url，所以不要登录 就像现在这样。队列根本不保存URL——只保存端点的URL id。

症状和修复程序已经出现 故障排除.

类型安全，计数

11个貌似合理的错误，11个在编译时被拒绝。 test/types/webhooks.ts 拥有一个 @ts-expect-error 每个错误，旁边的 必须保持编译的连接(janus({ events: webhooks(…) })，与 没有一个 queue)：没有秘密的端点，从 环境及未检查(string | undefined)，一种类型 janus 从来没有 发送一个重试，该重试不是 Duration，一个没有秘密的接收器，读一个 已验证事件，然后再检查是否为 null，缺少一个方法的队列 concurrency 写成字符串，端点 id 写成数字a switch 在 reason.why ,忘记 endpointRemoved，还有 onGivingUp ,上面写着 delivery.url 没有检查 null - - - 不再配置端点的URL。

文档

• 指南 -发送，接收和队列，每个选项的例子
• 故障排除 -通过您看到的警告或错误
• 路线图 -下一步是什么，什么是没有计划的

执照

麻省理工学院