企业官网建设流程全解析

在 Flue 中接入 Salesforce Marketing Cloud Engagement ENS 渠道：签名校验、批量事件与租户绑定 Fetch 客户端实战

【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flue

Salesforce Marketing Cloud Engagement 的 Event Notification Service（ENS）是官方用于把邮件发送、打开、点击、退订等事件批量推送到业务系统的事件通知服务。本文以 Flue 仓库中的 Salesforce Marketing Cloud 渠道文档为主线，完整讲解如何通过@flue/salesforce在现有 Flue 项目中接入“经过验证的”ENS 事件入口（ingress）和“租户绑定的”窄范围 Fetch 客户端，包括环境变量配置、事件批量分发、回调验证、签名机制、响应语义与 Cloudflare Workers 部署。读完本文，你将能够独立把 Marketing Cloud Engagement 的邮件生命周期事件安全地接入 Flue Agent，并让 Agent 通过受信任的工具检索 ENS 回调配置。

适用前提：本文集成面向Marketing Cloud Engagement ENS，并非通用的 Salesforce API。回调注册、OAuth、令牌刷新等流程由应用自己持有（application-owned），@flue/salesforce只负责签名验证与事件入口。

快速开始（Quickstart）

在已有 Flue 项目（Node 或 Cloudflare Workers 目标均可）的终端或编码 Agent 中执行：

flue add channel salesforce-marketing-cloud

该命令会安装@flue/salesforce包，并基于 channel--salesforce-marketing-cloud 蓝图 在项目中生成以下文件：

生成的示例项目可参考 examples/salesforce-marketing-cloud-channel，其中包含完整可运行的app.ts、channel、client、email 身份与 Agent 绑定实现。

整体架构与职责边界

从 blueprints/channel--salesforce-marketing-cloud.md 可以看出，这条集成的核心是清晰的职责切分：

• Flue 负责：精确字节（exact-body）签名验证、请求体与批量大小限制、最小公共字段校验、响应序列化。
• 项目负责：回调注册、POST /platform/v1/ens-verify、OAuth、令牌存储与刷新、订阅生命周期、事件族校验、去重、持久化、Agent 路由策略以及所有出站操作。

因此该集成不依赖 Salesforce SDK，@salesforce/core也不需要安装；入口与窄范围 REST 操作在 Node 与 Cloudflare Workers 中统一使用标准 Fetch 与 Web Crypto API。

渠道模块概览（精简版）

src/channels/salesforce-marketing-cloud.ts的核心结构如下：

import { createSalesforceMarketingCloudChannel } from '@flue/salesforce'; import { dispatch, useModel } from '@flue/runtime'; import { Assistant } from '../agents/assistant.ts'; import { createSalesforceMarketingCloudClient } from '../salesforce-marketing-cloud-client.ts'; import { emailEventInstanceId, emailRefFromEvent } from '../salesforce-marketing-cloud-email.ts'; const callbackId = process.env.SALESFORCE_MARKETING_CLOUD_CALLBACK_ID!; export const client = createSalesforceMarketingCloudClient({ restBaseUrl: process.env.SALESFORCE_MARKETING_CLOUD_REST_BASE_URL!, accessToken: process.env.SALESFORCE_MARKETING_CLOUD_ACCESS_TOKEN!, }); export const channel = createSalesforceMarketingCloudChannel({ signatureKey: process.env.SALESFORCE_MARKETING_CLOUD_SIGNATURE_KEY!, callbackId, async events({ c, batch }) { const usefulEvents = []; for (const event of batch.events) { if (event.eventCategoryType !== 'EngagementEvents.EmailOpen') continue; const ref = emailRefFromEvent(callbackId, event); if (!ref) return c.json({ error: 'Expected a supported email event.' }, 400); usefulEvents.push({ event, ref }); } for (const { event, ref } of usefulEvents) { await dispatch(Assistant, { id: emailEventInstanceId(ref), message: { kind: 'signal', type: `salesforce-marketing-cloud.${event.eventCategoryType}`, // `info` 携带事件族专属字段；参与类事件没有自然消息文本。 body: JSON.stringify(event.info ?? {}), attributes: { ...(typeof event.timestampUTC === 'string' ? { occurredAt: event.timestampUTC } : {}), callbackId: ref.callbackId, mid: ref.mid, eid: ref.eid, jobId: ref.jobId, batchId: ref.batchId, listId: ref.listId, subscriberId: ref.subscriberId, }, }, }); } return c.body(null, 204); }, });

每个通过签名验证且被选中的邮件事件，都会以“信号（signal）”消息分发到绑定在对应回调与邮件跟踪身份上的 Agent，随后整个批量统一返回204。完整生成的模块会额外处理发送类与参与类（send / engagement）事件族，并允许绑定的 Agent 检索所配置的回调。回调注册、OAuth、令牌刷新以及一次性/ens-verify调用全部由应用自持，Node 与 Cloudflare 目标共用同一套 Fetch 与 Web Crypto 实现。

挂载渠道（Mount the channel）

渠道只会在app.ts显式挂载它的地方提供 HTTP 路由。把模块具名的channel导出挂载到应用路由：

import { channel as salesforceMarketingCloud } from './channels/salesforce-marketing-cloud.ts'; app.route('/channels/salesforce-marketing-cloud', salesforceMarketingCloud.route());

channel.route()是一个纯路由器工厂（pure router factory），以挂载路径为基准提供渠道声明的路由。从源码看，packages/salesforce-marketing-cloud/src/index.ts 中createSalesforceMarketingCloudChannel()只声明了一条固定路由：

const routes: readonly ChannelRouteDefinition<E>[] = [ { method: 'POST', path: '/events', handler: createSalesforceMarketingCloudEventsHandler(options), }, ];

也就是说路由固定为POST /events。本文的所有 webhook 路径都假设使用约定俗成的/channels/salesforce-marketing-cloud挂载点；如果换了挂载路径，所有提供方 URL 会随之平移。

被dispatch的 Agent 模块带有'use agent'指令——正是这个指令完成了注册，所以仅用于 dispatch 的 Agent 不需要自己的 HTTP 挂载。只有当 Agent 需要被 HTTP 直接访问时，才需要额外添加app.route('/agents/<name>', createAgentRouter(Assistant))（来自@flue/runtime/routing），可参考 examples/salesforce-marketing-cloud-channel/src/app.ts。

环境变量配置

四个环境变量全部必填：

参考 examples/salesforce-marketing-cloud-channel/README.md 中的示例值：

SALESFORCE_MARKETING_CLOUD_REST_BASE_URL=https://tenant-subdomain.rest.marketingcloudapis.com SALESFORCE_MARKETING_CLOUD_ACCESS_TOKEN=... SALESFORCE_MARKETING_CLOUD_CALLBACK_ID=... SALESFORCE_MARKETING_CLOUD_SIGNATURE_KEY=...

注册完整回调 URL（以约定挂载点为例）：

https://example.com/channels/salesforce-marketing-cloud/events

需要特别强调的是：签名密钥与出站访问令牌是两个独立凭证。Salesforce 只在回调创建时返回一次签名密钥，因此它必须与 REST 访问令牌分开存放。回调注册、OAuth、令牌刷新与令牌存储都保持为应用自持。

完整的渠道模块实现

完整的src/channels/salesforce-marketing-cloud.ts如下（与 examples/salesforce-marketing-cloud-channel/src/channels/salesforce-marketing-cloud.ts 对应）：

import { createSalesforceMarketingCloudChannel, type SalesforceMarketingCloudEvent, } from '@flue/salesforce'; import { defineTool, dispatch, useModel } from '@flue/runtime'; import { Assistant } from '../agents/assistant.ts'; import { createSalesforceMarketingCloudClient } from '../salesforce-marketing-cloud-client.ts'; import { emailEventInstanceId, emailRefFromEvent, type SalesforceMarketingCloudEmailRef, } from '../salesforce-marketing-cloud-email.ts'; const callbackId = requiredEnv('SALESFORCE_MARKETING_CLOUD_CALLBACK_ID'); export const client = createSalesforceMarketingCloudClient({ restBaseUrl: requiredEnv('SALESFORCE_MARKETING_CLOUD_REST_BASE_URL'), accessToken: requiredEnv('SALESFORCE_MARKETING_CLOUD_ACCESS_TOKEN'), }); export const channel = createSalesforceMarketingCloudChannel({ signatureKey: requiredEnv('SALESFORCE_MARKETING_CLOUD_SIGNATURE_KEY'), callbackId, // Path: /channels/salesforce-marketing-cloud/events async events({ c, batch }) { const usefulEvents: Array<{ event: SalesforceMarketingCloudEvent; ref: SalesforceMarketingCloudEmailRef; }> = []; for (const event of batch.events) { switch (event.eventCategoryType) { case 'TransactionalSendEvents.EmailSent': case 'TransactionalSendEvents.EmailNotSent': case 'TransactionalSendEvents.EmailBounced': case 'EngagementEvents.EmailOpen': case 'EngagementEvents.EmailClick': case 'EngagementEvents.EmailUnsubscribe': { const ref = emailRefFromEvent(callbackId, event); if (!ref) { return c.json({ error: 'Expected a supported Marketing Cloud email event.' }, 400); } usefulEvents.push({ event, ref }); break; } default: break; } } for (const { event, ref } of usefulEvents) { await dispatch(Assistant, { id: emailEventInstanceId(ref), message: { kind: 'signal', type: `salesforce-marketing-cloud.${event.eventCategoryType}`, // `info` 携带事件族专属字段；参与类事件没有自然消息文本。 body: JSON.stringify(event.info ?? {}), attributes: { ...(typeof event.timestampUTC === 'string' ? { occurredAt: event.timestampUTC } : {}), callbackId: ref.callbackId, mid: ref.mid, eid: ref.eid, jobId: ref.jobId, batchId: ref.batchId, listId: ref.listId, subscriberId: ref.subscriberId, }, }, }); } return c.body(null, 204); }, }); export function retrieveCallback(ref: SalesforceMarketingCloudEmailRef) { if (ref.callbackId !== callbackId) { throw new TypeError('Expected the configured Marketing Cloud callback.'); } return defineTool({ name: 'retrieve_salesforce_marketing_cloud_callback', description: 'Retrieve the Marketing Cloud ENS callback bound to this agent.', async run() { return { output: await client.getCallback(callbackId) }; }, }); } function requiredEnv(name: string): string { const value = process.env[name]; if (!value) throw new Error(`${name} is required.`); return value; }

该模块把示例中的事件族分组开放，保留了 ENS 事件分类法（taxonomy）的开放性。emailRefFromEvent()是应用层代码，负责校验mid、eid以及所选事件族在event.composite下的跟踪字段，把它们与callbackId一起规范化为本地 Agent id，并拒绝畸形事件。

为什么不能用compositeId做通用主键

ENS 不提供通用的投递（delivery）或会话（conversation）id。这个邮件身份只对应用实际校验过的事件族有效。compositeId是展平后的跟踪 id，且对事务邮件而言已废弃（deprecated），因此不要把它当作通用键使用。正如 packages/salesforce-marketing-cloud/src/index.ts 中的注释所述，ENS 事件族并不共享封闭的 schema：eventCategoryType是开放分类法，事件族专属字段都放在info中，任何未建模的认证字段都会被原样转发。

本地邮件身份的实现细节

examples/salesforce-marketing-cloud-channel/src/salesforce-marketing-cloud-email.ts 给出了完整实现：

• emailRefFromEvent(callbackId, event)：校验callbackId、正十进制mid/eid，以及composite.jobId / batchId / listId / subscriberId均为正十进制字符串，返回SalesforceMarketingCloudEmailRef | undefined；
• emailEventInstanceId(ref)：把规范化的引用序列化为salesforce-marketing-cloud-email:前缀的 URI 编码 JSON，生成稳定的本地 Agent id；
• parseEmailEventInstanceId(id)：解析并反向校验——非规范 id 直接抛TypeError。

实现中用“正十进制字符串”统一了number与string两种表示（Marketing Cloud 的mid、eid在部分事件族里是数字、在另一些里是字符串），保证身份一致性。需要重申：这是应用为所选邮件事件族自定义的身份，不是通用的 ENS 身份，更不是授权凭证。

项目自有的窄范围客户端

建议使用一个窄范围 Fetch 客户端，并在附加 Bearer 令牌之前校验租户源（origin）：

export function createSalesforceMarketingCloudClient({ restBaseUrl, accessToken, fetcher = globalThis.fetch, }: { restBaseUrl: string; accessToken: string; fetcher?: typeof globalThis.fetch; }) { const origin = salesforceMarketingCloudRestOrigin(restBaseUrl); if (!accessToken || accessToken.trim() !== accessToken) { throw new TypeError('Marketing Cloud access token must be non-empty and trimmed.'); } return { async getCallback(callbackId: string) { if (!callbackId || callbackId.trim() !== callbackId) { throw new TypeError('Marketing Cloud callback id must be non-empty and trimmed.'); } const response = await fetcher( `${origin}/platform/v1/ens-callbacks/${encodeURIComponent(callbackId)}`, { method: 'GET', headers: { accept: 'application/json', authorization: `Bearer ${accessToken}`, }, }, ); if (!response.ok) { throw new Error(`Marketing Cloud API request failed with ${response.status}.`); } const value: unknown = await response.json(); if (!value || typeof value !== 'object' || Array.isArray(value)) { throw new TypeError('Marketing Cloud returned an invalid callback response.'); } return value; }, }; } function salesforceMarketingCloudRestOrigin(value: string): string { const url = new URL(value); const suffix = '.rest.marketingcloudapis.com'; if ( url.protocol !== 'https:' || url.username !== '' || url.password !== '' || url.port !== '' || url.pathname !== '/' || url.search !== '' || url.hash !== '' || !url.hostname.endsWith(suffix) || url.hostname.length === suffix.length ) { throw new TypeError('Expected an HTTPS tenant origin ending in .rest.marketingcloudapis.com.'); } return url.origin; }

salesforceMarketingCloudRestOrigin做了 fail-closed 校验：只接受https:、无用户名/密码、无端口、无查询串、无 hash、根路径、主机名以.rest.marketingcloudapis.com结尾且不是裸域名。示例实现 examples/salesforce-marketing-cloud-channel/src/salesforce-marketing-cloud-client.ts 还额外做了 DNS 名称逐标签校验（isDnsName）与响应体 schema 校验（isCallback）。

客户端最终只执行这一条受限操作：

GET /platform/v1/ens-callbacks/{callbackId} Authorization: Bearer <access token>

不要从模型或事件中接受任意 API 源、回调 id 或令牌——上面示例中的工具把三者全部绑定在受信任的应用代码中。无需任何 Salesforce SDK；回调注册、OAuth、令牌刷新、订阅生命周期、令牌存储与更广泛的出站 API 行为都由应用自持。

绑定 Agent 与回调检索工具

Agent 模块示例（对应 examples/salesforce-marketing-cloud-channel/src/agents/assistant.ts）：

'use agent'; import { type AgentProps, useModel, useTool } from '@flue/runtime'; import { retrieveCallback } from '../channels/salesforce-marketing-cloud.ts'; import { parseEmailEventInstanceId } from '../salesforce-marketing-cloud-email.ts'; export function Assistant({ id }: AgentProps) { useModel('anthropic/claude-haiku-4-5'); const email = parseEmailEventInstanceId(id); useTool(retrieveCallback(email)); return 'Review the inbound Salesforce Marketing Cloud email lifecycle event. Retrieve the configured ENS callback when callback status or delivery configuration is relevant.'; }

要点：

• 'use agent'是模块第一行，负责向应用注册该 Agent，因此 dispatch 不需要额外挂载；
• 工具retrieve_callback不接受来自模型的租户源、回调 id、访问令牌或资源 id——模型只提供解析后的本地身份；
• 解析出的本地 id只是标识符，不是授权；工具在挑选凭证前会再次核对callbackId；
• 蓝图版本中还展示了Assistant.initialData静态 schema +useInitialData()的写法：dispatch 时通过initialData携带校验过的邮件引用字段（仅实例创建时记录一次），Agent 用useInitialData()读取，而不是解析实例 id。两种方式都行，但务必保持“先校验、后使用”的原则。

回调验证（Callback verification）

在回调设置阶段，Marketing Cloud 会发送一个未签名的 JSON 请求体，内容恰好是：

{ "callbackId": "provider-callback-id", "verificationKey": "one-time-verification-key" }

处理逻辑（见 packages/salesforce-marketing-cloud/src/webhook.ts 的createSalesforceMarketingCloudEventsHandler）：

• 未签名的设置请求仅当channel 提供了verification处理器时才会被接受；
• 应用代码应限制callbackId、调用POST /platform/v1/ens-verify，并在设置完成后禁用该处理器；
• 没有该处理器时，未签名请求一律返回401；校验到callbackId不匹配时返回403。

Flue 负责校验请求体形状，并在处理器完成后返回所需的空200。它不会自动注册回调、获取令牌或调用验证 API。这个设置调用应与上文只读（GET-only）的客户端保持分离，除非应用明确需要。

签名机制与事件批量（Signatures and event batches）

已签名的通知携带如下请求头：

x-sfmc-ens-signature: <base64 HMAC-SHA256 digest>

关键事实（可从 packages/salesforce-marketing-cloud/src/webhook.ts 源码逐一验证）：

• Marketing Cloud 对精确请求体字节签名；signatureKey是回调创建时返回的不透明字符串，直接作为 UTF-8 HMAC 密钥材料导入（crypto.subtle.importKey('raw', encoder.encode(signatureKey), ...)），不要对其 base64 解码；只有签名头需要 base64 解码；
• 签名验证发生在 UTF-8 解码与 JSON 解析之前（verifySignature直接对原始字节做 HMAC 校验）；
• 签名解析用正则/^[A-Za-z0-9+/]{43}=$/校验 base64 形状，解码后长度必须为 32 字节，否则拒绝。

已签名载荷是有序、非空、至多 1000 个事件的数组（常量MAX_BATCH_SIZE = 1000）。每个事件都保留 Marketing Cloud 自己的字段名与嵌套结构原样透传——没有raw包装、没有字段投影。入口只需要每个事件有一个非空eventCategoryType（isMinimalEvent），这一字段是批量可转发（forwardable）的唯一前提。其余字段按 ENS 原样投递：

• timestampUTC：提供方 UTC 纪元毫秒，原样转发且不做校验（部分事件族会省略或用不同表示）；
• composite（{ jobId, batchId, listId, … }）、definitionKey、definitionId：在携带它们的邮件发送与参与事件族上原样透传；
• info：事件族专属细节；
• mid与eid：在部分事件族以number到达、另一些以string到达；
• compositeId：展平跟踪 id，对事务邮件已废弃。

SalesforceMarketingCloudEvent类型带有顶层索引签名，任何通过认证但类型未建模的字段都会被转发。批量对象还暴露rawBody——签名验证后的精确 UTF-8 请求体。该包不封闭事件分类法，也不推断通用资源、行为者、投递或会话身份——请围绕eventCategoryType收窄处理，并校验你读取的每个事件族字段。

请求体与批量限制

webhook.ts中的防护（均可通过createSalesforceMarketingCloudChannel的bodyLimit选项配置，默认1 MiB）：

• content-type不是application/json→415；
• content-length非数字或超过bodyLimit→400/413；
• 流式读取超过bodyLimit→413；
• 读取失败、UTF-8 解码失败（TextDecoder('utf-8', { fatal: true })）、JSON 解析失败、空数组或超过 1000 个事件 →400。

响应与投递语义

来自events处理器的返回值决定了响应：

• 返回undefined（什么都不返回）→ 空200；
• 返回 JSON 兼容值 → JSON200；
• 返回普通 Hono 或 FetchResponse→ 原样透传。

ENS 只承认200–204之间的状态码。channel 失败或返回不可序列化的值会产生500。自定义Response落在确认区间之外会被透传，并可能引发重新投递。

超时与重试

• Flue 不施加路由超时：处理器会被 await，其结果被序列化；
• 唯一的 ENS 截止时间在设置阶段：未签名的验证 POST 必须在 30 秒内以200应答，否则回调创建失败；
• 稳态投递没有单请求截止时间，但 ENS 会对任何未确认的批量进行重试。

ENS 投递是至少一次（at-least-once），重试最长可能持续七天。因此务必：

1. 快速接收持久性工作——先dispatch再返回，不要在处理器里阻塞慢操作；
2. 依赖幂等性——channel不会去重或持久化事件，请在非幂等工作之前，使用应用自有的持久化状态与事件族合适的键先“认领”数据。

示例 examples/salesforce-marketing-cloud-channel/README.md 的做法是：先校验所有被选事件、按提供方顺序 dispatch，最后统一返回一个204；需要恰好一次（exactly-once）效果的应用，必须在 dispatch 前把提供方事件数据写入持久化存储。

Cloudflare Workers 部署与测试

入口和项目自有客户端都使用标准的 Fetch、URL 与 Web Crypto API，在 Flue 规范的nodejs_compat配置下运行于 workerd；包内的 workerd 测试覆盖了精确字节的 HMAC 验证路径。示例提供了双构建配置：vite.config.ts（Node）与 vite.config.cloudflare.ts，配合 wrangler.jsonc 部署。

测试规范（详见 channel--salesforce-marketing-cloud 蓝图 的 “Test without Salesforce” 一节）：

• 入口侧：用本地密钥构造原始合成事件批量；序列化一次、对未变更字节做 HMAC-SHA256 签名、仅对摘要做 base64。覆盖：改一个字节即拒绝；缺失/畸形/错误签名；证明signatureKey不被 base64 解码；未签名{ callbackId, verificationKey }精确形状；无 verification 处理器时未签名请求被拒；callbackId 不匹配；1 个与 1000 个事件的有序批量以及空/超大批量；公共字段必需性与可选/未建模字段的逐字转发；畸形 UTF-8 与 JSON、媒体类型、声明与流式体限制；无值 / JSON / 普通Response三种结果与确认边界；应用失败返回500；
• 客户端侧：在 Node 与 workerd 中注入 fail-closed Fetch 测试真实客户端，断言精确的受信任租户源、GET /platform/v1/ens-callbacks/{callbackId}、Bearer 认证、/ens-verify的精确请求体，以及 HTTP/凭证/端口/路径/非*.rest.marketingcloudapis.com主机的拒绝，并确认没有意外网络目标被触达。

对应命令（参考 examples/salesforce-marketing-cloud-channel/package.json）：

pnpm run check:types pnpm run test pnpm run build pnpm run build:cloudflare

严禁在实现或测试期间注册/修改真实回调、订阅真实事件、执行 OAuth、向 Salesforce 请求真实令牌或调用/ens-verify——只用原始合成签名事件与 fake Fetch 传输。

小结

把 Salesforce Marketing Cloud Engagement ENS 接入 Flue 的正确姿势可以概括为四句话：

1. 入口交给 Flue：createSalesforceMarketingCloudChannel负责精确字节的 HMAC-SHA256 校验、体大小与批量限制、最小字段校验和响应序列化，路由固定为POST /events；
2. 出站自持且窄化：用受信任代码绑定租户源与访问令牌，只暴露GET /platform/v1/ens-callbacks/{callbackId}这类最小操作，绝不接受模型提供的源、回调 id 或令牌；
3. 身份应用自定义：基于mid、eid与composite跟踪字段构造本地 Agent id，先校验后使用；不要用废弃的compositeId当通用主键；
4. 承认至少一次语义：快速 dispatch 后返回、用持久化状态做幂等，回调验证与/ens-verify等设置流程与应用生命周期保持分离。

后续可继续阅读仓库内的 @flue/salesforce 包说明、渠道蓝图 与 完整示例 深入对照实现。

【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flue