BkndWebhooks配置Skill bknd-webhooks

---

名称: bknd-webhooks 描述: 用于在Bknd中配置webhook集成。涵盖通过HTTP触发器接收传入webhooks、使用FetchTask发送传出webhooks、数据更改时的事件触发webhooks、签名验证、重试模式和异步处理。

Webhooks

配置webhook集成以接收外部事件并在数据更改时发送通知。

先决条件

• 运行Bknd实例
• 理解HTTP和webhooks概念
• 熟悉Bknd流（见bknd-custom-endpoint）

何时使用UI模式

Webhook配置需要代码。没有UI方法可用。

何时使用代码模式

• 从外部服务接收webhooks（Stripe、GitHub等）
• 在数据更改时发送通知
• 与第三方服务集成
• 构建事件驱动架构

Webhook类型

类型	描述	方法
传入	从外部服务接收webhooks	HTTP触发器 + 流
传出	当事件发生时发送webhooks	事件触发器 + FetchTask

---

接收传入Webhooks

步骤1：基本Webhook接收器

import { App, Flow, HttpTrigger, Task } from "bknd";
import { s } from "bknd/utils";

class WebhookReceiverTask extends Task<typeof WebhookReceiverTask.schema> {
  override type = "webhook-receiver";
  static override schema = s.strictObject({});

  override async execute(input: Request) {
    const body = await input.json();
    const eventType = input.headers.get("x-event-type");

    console.log(`Received webhook: ${eventType}`, body);

    return { received: true, event: eventType };
  }
}

const receiverTask = new WebhookReceiverTask("receive", {});

const webhookFlow = new Flow("incoming-webhook", [receiverTask]);
webhookFlow.setRespondingTask(receiverTask);

webhookFlow.setTrigger(
  new HttpTrigger({
    path: "/webhooks/external",
    method: "POST",
    mode: "async",  // 立即返回200
  })
);

const app = new App({
  flows: { flows: [webhookFlow] },
});

步骤2：带签名验证的Webhook

import { createHmac, timingSafeEqual } from "crypto";

class SecureWebhookTask extends Task<typeof SecureWebhookTask.schema> {
  override type = "secure-webhook";

  static override schema = s.strictObject({
    secret: s.string(),
  });

  override async execute(input: Request) {
    const signature = input.headers.get("x-webhook-signature");
    const body = await input.text();

    // 验证签名
    if (!this.verifySignature(body, signature)) {
      throw this.error("无效签名", { signature });
    }

    const data = JSON.parse(body);
    return { verified: true, data };
  }

  private verifySignature(payload: string, signature: string | null): boolean {
    if (!signature) return false;

    const expected = createHmac("sha256", this.params.secret)
      .update(payload)
      .digest("hex");

    const sig = Buffer.from(signature);
    const exp = Buffer.from(`sha256=${expected}`);

    return sig.length === exp.length && timingSafeEqual(sig, exp);
  }
}

const secureTask = new SecureWebhookTask("verify", {
  secret: process.env.WEBHOOK_SECRET!,
});

const secureFlow = new Flow("secure-webhook", [secureTask]);
secureFlow.setRespondingTask(secureTask);
secureFlow.setTrigger(
  new HttpTrigger({
    path: "/webhooks/secure",
    method: "POST",
  })
);

步骤3：Stripe Webhook接收器

import Stripe from "stripe";

class StripeWebhookTask extends Task<typeof StripeWebhookTask.schema> {
  override type = "stripe-webhook";

  static override schema = s.strictObject({
    webhookSecret: s.string(),
  });

  override async execute(input: Request) {
    const stripe = new Stripe(process.env.STRIPE_SECRET_KEY!);
    const sig = input.headers.get("stripe-signature")!;
    const body = await input.text();

    let event: Stripe.Event;

    try {
      event = stripe.webhooks.constructEvent(
        body,
        sig,
        this.params.webhookSecret
      );
    } catch (err) {
      throw this.error("Webhook验证失败", { err });
    }

    // 处理事件类型
    switch (event.type) {
      case "checkout.session.completed":
        const session = event.data.object;
        // 处理成功支付...
        break;

      case "customer.subscription.deleted":
        // 处理订阅取消...
        break;
    }

    return { received: true, type: event.type };
  }
}

步骤4：GitHub Webhook接收器

class GitHubWebhookTask extends Task<typeof GitHubWebhookTask.schema> {
  override type = "github-webhook";

  static override schema = s.strictObject({
    secret: s.string(),
  });

  override async execute(input: Request) {
    const event = input.headers.get("x-github-event");
    const delivery = input.headers.get("x-github-delivery");
    const signature = input.headers.get("x-hub-signature-256");
    const body = await input.text();

    // 验证GitHub签名
    const expected = createHmac("sha256", this.params.secret)
      .update(body)
      .digest("hex");

    if (signature !== `sha256=${expected}`) {
      throw this.error("无效GitHub签名");
    }

    const payload = JSON.parse(body);

    switch (event) {
      case "push":
        console.log(`Push to ${payload.ref} by ${payload.pusher.name}`);
        break;

      case "pull_request":
        console.log(`PR ${payload.action}: ${payload.pull_request.title}`);
        break;

      case "issues":
        console.log(`Issue ${payload.action}: ${payload.issue.title}`);
        break;
    }

    return { event, delivery };
  }
}

步骤5：基于插件的Webhook接收器

对于更简单的情况，使用插件路由：

import { createPlugin } from "bknd";
import { Hono } from "hono";

const webhooksPlugin = createPlugin({
  name: "webhooks",

  onServerInit: (server) => {
    const webhooks = new Hono();

    // Stripe
    webhooks.post("/stripe", async (c) => {
      const sig = c.req.header("stripe-signature");
      const body = await c.req.text();
      // 验证和处理...
      return c.json({ received: true });
    });

    // GitHub
    webhooks.post("/github", async (c) => {
      const event = c.req.header("x-github-event");
      const body = await c.req.json();
      // 处理...
      return c.json({ received: true });
    });

    // 通用
    webhooks.post("/:source", async (c) => {
      const source = c.req.param("source");
      const body = await c.req.json();
      console.log(`Webhook from ${source}:`, body);
      return c.json({ received: true });
    });

    server.route("/webhooks", webhooks);
  },
});

---

发送传出Webhooks

使用带有事件触发器的流在数据更改时发送webhooks。

步骤1：基本传出Webhook

import { App, Flow, FetchTask, EventTrigger } from "bknd";

// 任务发送webhook
const sendWebhook = new FetchTask("send-webhook", {
  url: "https://example.com/webhook",
  method: "POST",
  headers: [
    { key: "Content-Type", value: "application/json" },
    { key: "X-Webhook-Source", value: "my-app" },
  ],
  body: "{{JSON.stringify(input)}}",  // 转发事件数据
});

const webhookFlow = new Flow("outgoing-webhook", [sendWebhook]);

// 在数据事件上触发
webhookFlow.setTrigger(
  new EventTrigger({
    event: "mutator-insert-after",  // 记录创建后
    mode: "async",
  })
);

const app = new App({
  flows: { flows: [webhookFlow] },
});

步骤2：实体特定Webhook

import { App, Flow, FetchTask, Task, EventTrigger } from "bknd";
import { s } from "bknd/utils";

// 过滤任务检查实体
class EntityFilterTask extends Task<typeof EntityFilterTask.schema> {
  override type = "entity-filter";

  static override schema = s.strictObject({
    targetEntity: s.string(),
  });

  override async execute(input: any) {
    if (input.entity?.name !== this.params.targetEntity) {
      throw this.error("跳过 - 错误实体");
    }
    return input;
  }
}

const filterTask = new EntityFilterTask("filter", {
  targetEntity: "orders",
});

const sendWebhook = new FetchTask("send", {
  url: "https://api.example.com/orders/webhook",
  method: "POST",
  headers: [{ key: "Content-Type", value: "application/json" }],
  body: "{{JSON.stringify({ event: 'order.created', data: input.changed })}}",
});

const flow = new Flow("order-webhook", [filterTask, sendWebhook]);
flow.task(filterTask).asInputFor(sendWebhook);

flow.setTrigger(
  new EventTrigger({
    event: "mutator-insert-after",
    mode: "async",
  })
);

步骤3：多目的地Webhooks

import { Flow, FetchTask, EventTrigger, Condition } from "bknd";

// 发送到多个端点
const sendSlack = new FetchTask("slack", {
  url: "https://hooks.slack.com/services/XXX/YYY/ZZZ",
  method: "POST",
  headers: [{ key: "Content-Type", value: "application/json" }],
  body: '{"text": "New order: {{input.changed.id}}"}',
});

const sendDiscord = new FetchTask("discord", {
  url: "https://discord.com/api/webhooks/XXX/YYY",
  method: "POST",
  headers: [{ key: "Content-Type", value: "application/json" }],
  body: '{"content": "New order: {{input.changed.id}}"}',
});

const sendCustom = new FetchTask("custom", {
  url: process.env.CUSTOM_WEBHOOK_URL!,
  method: "POST",
  headers: [{ key: "Content-Type", value: "application/json" }],
  body: "{{JSON.stringify(input)}}",
});

const flow = new Flow("multi-webhook", [sendSlack, sendDiscord, sendCustom]);

// 所有任务并行运行（无连接）
flow.setTrigger(
  new EventTrigger({
    event: "mutator-insert-after",
    mode: "async",
  })
);

步骤4：带重试逻辑的Webhook

import { Flow, FetchTask, Task, Condition, EventTrigger } from "bknd";
import { s } from "bknd/utils";

class RetryTask extends Task<typeof RetryTask.schema> {
  override type = "retry-webhook";

  static override schema = s.strictObject({
    url: s.string(),
    maxRetries: s.number({ default: 3 }),
    delayMs: s.number({ default: 1000 }),
  });

  override async execute(input: any) {
    let lastError: Error | null = null;

    for (let attempt = 1; attempt <= this.params.maxRetries; attempt++) {
      try {
        const response = await fetch(this.params.url, {
          method: "POST",
          headers: { "Content-Type": "application/json" },
          body: JSON.stringify(input),
        });

        if (response.ok) {
          return { success: true, attempt };
        }

        lastError = new Error(`HTTP ${response.status}`);
      } catch (err) {
        lastError = err as Error;
      }

      // 重试前等待（指数退避）
      if (attempt < this.params.maxRetries) {
        await new Promise((r) => setTimeout(r, this.params.delayMs * attempt));
      }
    }

    throw this.error("所有重试失败", { lastError: lastError?.message });
  }
}

---

可用事件

Bknd发出这些事件，可以触发webhooks：

数据事件

事件Slug	描述	负载
mutator-insert-before	记录创建前	{ entity, data }
mutator-insert-after	记录创建后	{ entity, data, changed }
mutator-update-before	记录更新前	{ entity, entityId, data }
mutator-update-after	记录更新后	{ entity, entityId, data, changed }
mutator-delete-before	记录删除前	{ entity, entityId }
mutator-delete-after	记录删除后	{ entity, entityId, data }

媒体事件

事件Slug	描述	负载
file-uploaded	文件上传	{ name, meta, etag, file, state }
file-deleted	文件删除	{ name }
file-access	文件访问	{ name }

示例：使用事件负载

// 事件负载结构，用于mutator-insert-after
interface InsertAfterPayload {
  entity: {
    name: string;       // 实体名称，例如"orders"
    fields: Field[];    // 实体字段
  };
  data: Record<string, any>;     // 原始输入数据
  changed: Record<string, any>;  // 结果记录带ID
}

class ProcessEventTask extends Task {
  override async execute(input: InsertAfterPayload) {
    const entityName = input.entity.name;
    const recordId = input.changed.id;
    const recordData = input.changed;

    // 发送带结构化数据的webhook
    await fetch("https://api.example.com/webhook", {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({
        event: `${entityName}.created`,
        timestamp: new Date().toISOString(),
        data: recordData,
      }),
    });

    return { sent: true };
  }
}

---

完整示例：订单通知系统

import { App, em, entity, text, number, Flow, FetchTask, Task, EventTrigger, Condition } from "bknd";
import { s } from "bknd/utils";

// 模式
const schema = em({
  orders: entity({
    customer_email: text().required(),
    total: number().required(),
    status: text().default("pending"),
  }),
});

// 过滤订单
class OrderFilterTask extends Task<typeof OrderFilterTask.schema> {
  override type = "order-filter";
  static override schema = s.strictObject({});

  override async execute(input: any) {
    if (input.entity?.name !== "orders") {
      throw this.error("不是订单");
    }
    return input.changed;  // 传递订单数据
  }
}

// 格式化webhook负载
class FormatWebhookTask extends Task<typeof FormatWebhookTask.schema> {
  override type = "format-webhook";
  static override schema = s.strictObject({});

  override async execute(order: any) {
    return {
      event: "order.created",
      timestamp: new Date().toISOString(),
      order: {
        id: order.id,
        email: order.customer_email,
        total: order.total,
        status: order.status,
      },
    };
  }
}

const filterTask = new OrderFilterTask("filter", {});
const formatTask = new FormatWebhookTask("format", {});

// 发送到多个目的地
const sendSlack = new FetchTask("slack", {
  url: process.env.SLACK_WEBHOOK_URL!,
  method: "POST",
  headers: [{ key: "Content-Type", value: "application/json" }],
  body: '{"text": "New order #{{input.order.id}} - ${{input.order.total}}"}',
});

const sendExternal = new FetchTask("external", {
  url: process.env.EXTERNAL_WEBHOOK_URL!,
  method: "POST",
  headers: [
    { key: "Content-Type", value: "application/json" },
    { key: "X-API-Key", value: process.env.EXTERNAL_API_KEY! },
  ],
  body: "{{JSON.stringify(input)}}",
});

// 构建流
const orderWebhookFlow = new Flow("order-notifications", [
  filterTask,
  formatTask,
  sendSlack,
  sendExternal,
]);

// 连接：filter -> format -> [slack, external]（并行）
orderWebhookFlow.task(filterTask).asInputFor(formatTask);
orderWebhookFlow.task(formatTask).asInputFor(sendSlack);
orderWebhookFlow.task(formatTask).asInputFor(sendExternal);

// 在新订单上触发
orderWebhookFlow.setTrigger(
  new EventTrigger({
    event: "mutator-insert-after",
    mode: "async",
  })
);

const app = new App({
  data: { schema },
  flows: { flows: [orderWebhookFlow] },
});

---

测试Webhooks

测试传入Webhook

# 基本测试
curl -X POST http://localhost:7654/webhooks/external \
  -H "Content-Type: application/json" \
  -H "X-Event-Type: test" \
  -d '{"test": true}'

# 带签名（HMAC-SHA256）
PAYLOAD='{"test":true}'
SECRET="your-secret"
SIGNATURE=$(echo -n "$PAYLOAD" | openssl dgst -sha256 -hmac "$SECRET" | cut -d' ' -f2)

curl -X POST http://localhost:7654/webhooks/secure \
  -H "Content-Type: application/json" \
  -H "X-Webhook-Signature: sha256=$SIGNATURE" \
  -d "$PAYLOAD"

测试传出Webhook

使用webhook.site或类似：

// 临时指向测试URL
const sendWebhook = new FetchTask("send", {
  url: "https://webhook.site/your-unique-id",
  method: "POST",
  body: "{{JSON.stringify(input)}}",
});

然后创建记录：

curl -X POST http://localhost:7654/api/data/orders \
  -H "Content-Type: application/json" \
  -d '{"customer_email": "test@example.com", "total": 99.99}'

---

常见陷阱

Webhook未接收数据

问题： 传入webhook返回200但不处理

修复： 检查模式 - 异步立即返回：

// 异步模式在后台处理
new HttpTrigger({ mode: "async" });

// 调试用，使用同步
new HttpTrigger({ mode: "sync" });

签名验证失败

问题： 有效webhooks被拒绝

修复： 确保在解析前读取原始正文：

// 错误 - 正文已解析
const body = await input.json();
const sig = verify(JSON.stringify(body), signature);

// 正确 - 先读取原始文本
const bodyText = await input.text();
const verified = verify(bodyText, signature);
const body = JSON.parse(bodyText);

传出Webhook未触发

问题： 事件触发器流不运行

修复： 检查事件名称是否完全匹配：

// 可用事件（使用精确slug）
"mutator-insert-after"   // 不是"data:entity:created"
"mutator-update-after"   // 不是"data:entity:updated"
"mutator-delete-after"   // 不是"data:entity:deleted"

所有实体触发Webhook

问题： Webhook为每个实体触发，不仅目标

修复： 添加实体过滤器任务：

class EntityFilter extends Task {
  async execute(input) {
    if (input.entity?.name !== "orders") {
      throw this.error("跳过");  // 停止流
    }
    return input;
  }
}

FetchTask正文未插值

问题： {{input}} 在正文中字面出现

修复： 使用正确的模板语法：

// 错误
body: "{ data: {{input}} }"

// 正确
body: "{{JSON.stringify({ data: input })}}"

---

注意事项

做：

• 对传入webhooks使用mode: "async"（快速返回200）
• 对安全敏感的webhooks验证签名
• 对目标传出webhooks使用实体过滤器任务
• 对关键传出webhooks实现重试逻辑
• 记录webhook事件以进行调试
• 使用环境变量存储webhook URL和秘密

不做：

• 在传入webhooks上阻塞（外部服务有超时）
• 未经验证信任传入数据
• 在代码中硬编码webhook秘密
• 忘记优雅处理webhook失败
• 在不加密的情况下在webhook负载中发送敏感数据
• 在没有速率限制的情况下暴露webhook端点

---

相关技能

• bknd-custom-endpoint - 创建自定义API端点（HTTP触发器）
• bknd-protect-endpoint - 保护webhook端点
• bknd-api-discovery - 探索可用端点
• bknd-client-setup - 从前端调用webhooks