在 Flue 中接入 Salesforce Marketing Cloud Engagement ENS 通道签名校验、事件分发与无 Salesforce 测试指南【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flue本文以 Flue 官方蓝图 blueprints/channel--salesforce-marketing-cloud.md 为骨架结合仓库中flue/salesforce包源码packages/salesforce-marketing-cloud/src/index.ts、packages/salesforce-marketing-cloud/src/webhook.ts与完整可运行示例examples/salesforce-marketing-cloud-channel详细讲解如何为 Flue 项目新增 Salesforce Marketing Cloud Engagement Event Notification ServiceENS入口与一个极窄的应用自有 REST 客户端。读完本文你将掌握通道客户端与channel的完整创建流程、app.ts中的挂载方式、邮件事件家族身份instance id的构造与校验、agent 的use agent注册机制、端点配置与 HMAC 签名细节以及完全不接触真实 Salesforce 的测试策略。文中所有配置项、代码与命令均以当前仓库实际内容为准。一、理解职责边界Flue 负责什么项目负责什么蓝图开篇就强调这不是一个通用的 Salesforce 集成而是面向 ENS 事件通知的窄接入。职责划分非常明确Flueflue/salesforce包负责精确的请求体签名校验HMAC-SHA256Web Crypto 实现请求体大小限制与批量大小限制最小公共字段校验仅要求eventCategoryType非空响应序列化空值 → 空200JSON 兼容值 → JSON200Response原样透传。项目应用代码负责回调callback注册/platform/v1/ens-verify验证调用OAuth 与 token 存储、刷新订阅生命周期管理事件家族event family校验去重、持久化agent 路由策略以及所有出站操作。从 packages/salesforce-marketing-cloud/src/index.ts 的createSalesforceMarketingCloudChannel可以看出通道只声明一条固定路由POST /events源码 L138-L144并通过validateOptionsL151-L173对signatureKey、callbackId、verification、events做启动期校验。换句话说Flue 不替你做回调注册、OAuth 或验证 API 调用这些一律是应用层职责。二、创建窄 REST 客户端首先安装flue/salesforce。蓝图明确不要安装salesforce/core因为本蓝图中的入口和窄 REST 操作在 Node 与 Cloudflare Workers 中都使用标准 Fetch 与 Web Crypto。在源码根目录创建salesforce-marketing-cloud-client.ts源码根目录的选择顺序为root/.flue/→root/src/→root/将租户 origin 与 access token 从可信配置绑定export interface SalesforceMarketingCloudClientOptions { restBaseUrl: string; accessToken: string; fetcher?: typeof globalThis.fetch; } interface SalesforceMarketingCloudCallback { callbackId: string; callbackName: string; url: string; maxBatchSize: number; status: string; statusReason: string; } export function createSalesforceMarketingCloudClient({ restBaseUrl, accessToken, fetcher globalThis.fetch, }: SalesforceMarketingCloudClientOptions) { const origin salesforceMarketingCloudRestOrigin(restBaseUrl); if (!accessToken || accessToken.trim() ! accessToken) { throw new TypeError( Salesforce Marketing Cloud access token must be non-empty and trimmed., ); } if (typeof fetcher ! function) { throw new TypeError(Salesforce Marketing Cloud Fetch must be callable.); } return { async getCallback(callbackId: string): PromiseSalesforceMarketingCloudCallback { validateCallbackId(callbackId); 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( Salesforce Marketing Cloud request failed with status ${response.status}., ); } let value: unknown; try { value await response.json(); } catch { throw new TypeError(Salesforce Marketing Cloud returned invalid JSON.); } if (!isCallback(value) || value.callbackId ! callbackId) { throw new TypeError( Salesforce Marketing Cloud returned an invalid callback response., ); } return value; }, }; }salesforceMarketingCloudRestOrigin是本客户端的核心安全闸门它只接受以.rest.marketingcloudapis.com结尾的 HTTPS 租户 origin拒绝 HTTP、带用户名密码、带端口、带非根路径、带 query/hash 以及非 DNS 名称的主机。完整实现见 examples/salesforce-marketing-cloud-channel/src/salesforce-marketing-cloud-client.tsexport function salesforceMarketingCloudRestOrigin(value: string): string { let url: URL; try { url new URL(value); } catch { throw new TypeError( Salesforce Marketing Cloud REST base URL must be a valid URL., ); } const suffix .rest.marketingcloudapis.com; if ( url.protocol ! https: || url.username ! || url.password ! || url.port ! || url.search ! || url.hash ! || url.pathname ! / || !url.hostname.endsWith(suffix) || url.hostname.length suffix.length || !isDnsName(url.hostname) ) { throw new TypeError( Salesforce Marketing Cloud REST base URL must be an HTTPS tenant origin ending in .rest.marketingcloudapis.com., ); } return url.origin; }响应校验同样严格isCallback要求callbackId、callbackName、url、status、statusReason均为非空字符串maxBatchSize为正的安全整数并且返回的callbackId必须与请求的一致防串号。需要特别注意的安全原则不接受来自事件、模型、工具参数或未签名 setup 请求的任意 base URL——租户 origin 只能来自可信配置access token 与租户 origin 必须与 ENS 回调签名密钥signature key分离存放该客户端刻意只演示回调查询GET /platform/v1/ens-callbacks/{callbackId}。回调创建、回调验证、订阅管理、OAuth、token 刷新和更广泛的 Marketing Cloud API 都不要放进这个客户端——这是窄客户端设计的本意。三、创建通道channel创建channels/salesforce-marketing-cloud.ts将环境变量读取、选定的事件家族和派发消息适配到具体项目// flue-blueprint: channel/salesforce-marketing-cloud1 import { createSalesforceMarketingCloudChannel, type SalesforceMarketingCloudEvent, } from flue/salesforce; import { defineTool, dispatch, type JsonValue } 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), // Recorded once when this event creates the instance; ignored after. initialData: { callbackId: ref.callbackId, mid: ref.mid, eid: ref.eid, jobId: ref.jobId, batchId: ref.batchId, listId: ref.listId, subscriberId: ref.subscriberId, }, message: { kind: signal, type: salesforce-marketing-cloud.${event.eventCategoryType}, // info carries the family-specific event fields; there is no // natural message text for an engagement event. 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)) as unknown as JsonValue }; }, }); } function requiredEnv(name: string): string { const value process.env[name]; if (!value) throw new Error(${name} is required.); return value; }完整可运行版本见 examples/salesforce-marketing-cloud-channel/src/channels/salesforce-marketing-cloud.ts。要点解读事件家族过滤示例选择了 6 个当前邮件家族——TransactionalSendEvents.EmailSent / EmailNotSent / EmailBounced与EngagementEvents.EmailOpen / EmailClick / EmailUnsubscribe。不支持的eventCategoryType直接跳过default: break先校验再派发每个被选中的事件都必须能解析出合法的邮件引用emailRefFromEvent否则返回400全部解析成功后按供应商顺序逐条dispatch最后统一返回一个204作为整批确认initialData语义它是实例的创建数据仅在事件创建实例时记录一次、之后被忽略因此通道在每次派发时都传递它。它携带经过校验的邮件引用字段——agent 通过useInitialData()读取而不是解析 instance id。每条消息级别的事实放在 signal 的attributes上如occurredAt时间戳工具不暴露凭据retrieveCallback工具没有模型可控参数租户 origin、access token 与 callback id 都在可信代码中绑定完成。该工具还校验ref.callbackId callbackId防止跨回调混用。通道选项与默认值源码级createSalesforceMarketingCloudChannel的完整选项定义在 packages/salesforce-marketing-cloud/src/index.ts下表为各选项的语义与默认值选项类型说明signatureKeystring必填回调专属签名密钥回调创建时一次性返回。Marketing Cloud 将该不透明字符串直接作为 HMAC keyUTF-8只有签名头做 base64 解码。不要对它做 base64 解码callbackIdstring可选未签名 setup 挑战的 callback-id 限制非空且必须 trim 后不变bodyLimitnumber可选请求体大小上限字节默认1 MiB1024 * 1024见 webhook.ts必须是正整数安全整数verification(input) void \| Promisevoid可选仅 setup 阶段使用的未签名回调验证处理器。省略时未签名请求一律被拒events(input) HandlerResult必填接收每一个通过认证的 ENS 通知批挂载通道channel.route()通道只有在app.ts中挂载后才对外提供 HTTP 路由。channel.route()是一个纯路由工厂返回可在挂载点之下提供相对路径服务的 Hono 子应用// app.ts import { Hono } from hono; import { channel } from ./channels/salesforce-marketing-cloud.ts; const app new Hono(); app.route(/channels/salesforce-marketing-cloud, channel.route()); export default app;// Path:注释假定了约定挂载路径/channels/salesforce-marketing-cloud更换挂载路径会相应地移动每个供应商 URL。真实示例 examples/salesforce-marketing-cloud-channel/src/app.ts 还额外挂载了createAgentRouter(Assistant)来自flue/runtime/routing用于让 agent 也通过 HTTP 直接可达。四、验证处理器unsigned verification与/events路由语义createSalesforceMarketingCloudEventsHandlerwebhook.ts完整实现了请求处理流水线顺序如下内容类型content-type必须以application/json开头否则返回415Content-Length 预检非数字返回400超过bodyLimit返回413流式读取readBody边读边累计字节数超过限制即取消读取并返回413读取异常返回400实现见 L165-L197UTF-8 严格解码使用fatal: true的TextDecoder非法 UTF-8 返回400签名分支无x-sfmc-ens-signature头若未配置verification处理器 →401否则解析{ callbackId, verificationKey }未签名 setup 挑战callbackId与配置不一致返回403校验通过后调用verification处理器并返回空200有签名头解析 base64 签名正则^[A-Za-z0-9/]{43}$且解码后恰好 32 字节L141-L150用crypto.subtle.verify对原始请求体字节做 HMAC-SHA256 校验失败返回401批解析parseBatch要求 JSON 是有序的非空数组、至多 1000 个事件每个事件只需是非空对象且eventCategoryType为非空字符串L83-L108。注意单个缺少eventCategoryType的项是唯一会让整个批在事件形状上失败的情况——这正是最小公共字段校验的含义。未签名验证setup 工作流未签名的回调验证仅在配置了verification处理器时被接受请求体必须恰好是{ callbackId: provider-callback-id, verificationKey: one-time-verification-key }启用规则只在拥有该回调的 setup 工作流中启用该处理器检查配置的callbackId从应用代码调用/platform/v1/ens-verify验证完成后禁用未签名验证。蓝图明确告诫不要把verification加进常规事件服务配置中上文通道示例就没有。当 setup 明确在范围内时应用自有的验证调用必须遵循与查询客户端相同的可信租户 origin 与 Bearer token 规则并且只能通过注入的 fake Fetch 测试。五、构造家族身份family identity创建salesforce-marketing-cloud-email.ts为选定的邮件家族构造本地身份。要点校验mid、eid以及event.composite下的家族专属字段jobId、batchId、listId、subscriberId将校验后的字段与callbackId一起序列化进一个规范化的本地 agent id提供配套函数示例实现见 examples/salesforce-marketing-cloud-channel/src/salesforce-marketing-cloud-email.tsemailRefFromEvent( callbackId: string, event: SalesforceMarketingCloudEvent, ): SalesforceMarketingCloudEmailRef | undefined; emailEventInstanceId(ref: SalesforceMarketingCloudEmailRef): string; parseEmailEventInstanceId(id: string): SalesforceMarketingCloudEmailRef;实现细节值得注意正十进制字符串规范化decimalId同时接受正安全整数number与符合/^[1-9]\d*$/的字符串统一转为十进制字符串使数字 id 与字符串 id 归一化一致拒绝畸形与非规范 idparseEmailEventInstanceId先验证前缀salesforce-marketing-cloud-email:再解码 JSON、逐字段校验最后要求emailEventInstanceId(ref) id回环一致L57-L86防止任意注入本地身份不是通用身份这是为选定邮件家族应用定义的标识不是通用的 ENS 身份不要使用已废弃的compositeId扁平化字段用于事务性邮件——应使用分解后的composite对象。六、创建 agent 并注册创建agents/assistant.ts。use agent指令模块第一个语句负责把 agent 注册进应用——通道回调中的dispatch(...)不需要app.ts挂载use agent; import { useInitialData, useModel, useTool } from flue/runtime; import * as v from valibot; import { retrieveCallback } from ../channels/salesforce-marketing-cloud.ts; const initialDataSchema v.object({ callbackId: v.string(), mid: v.string(), eid: v.string(), jobId: v.string(), batchId: v.string(), listId: v.string(), subscriberId: v.string(), }); export function Assistant() { useModel(anthropic/claude-haiku-4-5); const data useInitialDatav.InferOutputtypeof initialDataSchema(); if (!data) { throw new Error( This agent is created by the Salesforce Marketing Cloud channel dispatch., ); } useTool(retrieveCallback(data)); return Review the inbound Salesforce Marketing Cloud email lifecycle event. Retrieve the configured ENS callback when callback status or delivery configuration is relevant.; } Assistant.initialData initialDataSchema;要点示例见 examples/salesforce-marketing-cloud-channel/src/agents/assistant.tsAssistant.initialData静态属性用 Valibot schema 在实例创建时校验派发的initialDatauseInitialData()在每次渲染时返回解析后的值无initialData即报错该 agent 只能由通道派发创建不允许以空数据进入工具零参数模型无法传入租户 origin、callback id 或 access token。ENS 本身不提供通用 delivery id 或 conversation id应用定义的邮件 id 只有在选定家族字段通过校验后才有效且它只是标识符而非授权凭据HTTP 直连可选只有当 agent 需要直接通过 HTTP 可达时才在app.ts中追加app.route(/agents/name, createAgentRouter(Assistant))来自flue/runtime/routing。七、配置 Marketing Cloud 端点在 Marketing Cloud Engagement 中注册完整的 HTTPS 回调 URL——即app.ts中通道的挂载路径加上路由后缀。按约定挂载时即为https://example.com/channels/salesforce-marketing-cloud/events更换挂载路径会相应地改变该 URL。签名头与批格式Marketing Cloud 发送的签名通知批携带x-sfmc-ens-signature: base64 HMAC-SHA256 digestHMAC 输入是精确的请求体字节原始字节webhook.ts在 UTF-8 解码与 JSON 解析之前完成签名校验配置的signatureKey直接作为 UTF-8 密钥材料使用importSigningKey用TextEncoder编码后crypto.subtle.importKey(raw, ..., { name: HMAC, hash: SHA-256 }, ...)只有x-sfmc-ens-signature头做 base64 解码。每个签名 payload 是一个有序、非空、至多 1000 个事件的 JSON 数组。入口只要求每个事件有非空eventCategoryType其余所有字段——timestampUTC、compositeId、composite、definitionKey、definitionId、mid、eid、info及任何未来字段——都按 Marketing Cloud 交付的原样转发保留其自有命名与嵌套。timestampUTC是供应商的 UTC 毫秒时间戳不做校验。应用应围绕eventCategoryType收窄只读取预期家族字段。batch还暴露精确解码后的rawBody。因为不存在通用的 ENS delivery id、resource id、actor id 或 conversation id应用必须从每个订阅事件家族的文档化字段中校验并组装自己的应用身份。八、响应与投递行为serializeHandlerResultwebhook.ts的规则如下处理器返回值响应什么都不返回undefined空200JSON 兼容值JSON200普通 Hono 或 FetchResponse原样透传回调抛异常或结果不可序列化500关键投递语义ENS 将200~204之外的任何状态视为投递失败并重试所以只在有意触发重投时使用范围外的透传响应Marketing Cloud 期望快速确认如果验证 POST 未在30 秒内得到200应答回调创建就会失败未快速确认的投递会被重试。因此应快速受理持久化工作派发给 agent 或入队后立即返回不要在响应前阻塞于慢操作。Flue 不施加自己的路由超时ENS 投递是至少一次at-least-once语义重试最长可持续7 天且通道不去重。因此在执行非幂等工作之前必须先用家族合适的应用键持久化claim否则失败的批可能导致已完成的工作被重复执行。九、不接触 Salesforce 的本地测试运行项目的严格类型检查、针对配置目标执行vite build以及真实的 workerd 测试。Flue 的规范 Cloudflare 环境启用nodejs_compat而本入口与客户端只使用标准 Fetch、URL 与 Web Crypto API因此 Node 与 workerd 均可运行。示例的可用命令见 examples/salesforce-marketing-cloud-channel/package.jsonpnpm run check:types pnpm run test pnpm run build pnpm run build:cloudflare测试数据必须使用原创合成 ENS 批与本地密钥将每个请求体序列化一次用不透明 UTF-8 密钥对未改动字节做 HMAC-SHA256 签名然后只对摘要做 base64 编码。入口ingress测试清单精确字节有效、改动一个字节后被拒绝缺失、畸形与错误的签名证明signatureKey不被base64 解码精确的未签名{ callbackId, verificationKey }setup 形状未配置验证处理器时未签名请求被拒callback-id 不匹配1 个与 1000 个事件的有序批以及空批、超大批必填公共字段以及可选/未建模的家族依赖字段按供应商命名与嵌套原样转发畸形 UTF-8 与 JSON、媒体类型、声明的与流式读取的 body 上限无值、JSON 与普通Response结果包括确认边界应用失败返回500。客户端测试清单Node 与 workerd注入 fail-closed Fetch精确的可信租户 originGET /platform/v1/ens-callbacks/{callbackId}请求路径Bearer 授权头启用 setup 验证时POST /platform/v1/ens-verify的精确请求体拒绝 HTTP、凭据、端口、路径与*.rest.marketingcloudapis.com之外的主机断言没有到达任何意外网络目标。绝对禁止在实现或测试期间注册/修改真实回调、订阅真实事件、执行 OAuth、请求真实 token、对 Salesforce 调用/ens-verify或联系任何 Salesforce API——只能使用原创合成签名事件与 fake Fetch 传输。十、环境变量与升级说明示例要求以下环境变量examples/salesforce-marketing-cloud-channel/README.mdSALESFORCE_MARKETING_CLOUD_REST_BASE_URLhttps://tenant-subdomain.rest.marketingcloudapis.com SALESFORCE_MARKETING_CLOUD_ACCESS_TOKEN... SALESFORCE_MARKETING_CLOUD_CALLBACK_ID... SALESFORCE_MARKETING_CLOUD_SIGNATURE_KEY...其中REST_BASE_URL必须是安装包时返回的可信租户 REST origin客户端会拒绝 HTTP、凭据、端口、query、fragment、非根路径与*.rest.marketingcloudapis.com之外的主机。getCallback()只能在/platform/v1/ens-callbacks/下追加编码后的 callback id并且始终以 Bearer 方式发送配置的 access token。SIGNATURE_KEY只在回调创建时返回应与 REST access token 分开存储。当更新既有集成时先检查并对照本完整蓝图应用所有相关变更并保留自定义内容然后在主标记文件中新增或更新flue-blueprint标记——标记缺失时这一步是必须的。当前蓝图版本为Version 12026-06-14初始版本见 blueprints/channel--salesforce-marketing-cloud.md 的 Upgrade Guide。通用通道规范可进一步参考 blueprints/channel.md其中解释了路由后缀约定/webhook、/events、供应商原生名与createChannelRouter挂载方式。【免费下载链接】flueThe sandbox agent framework.项目地址: https://gitcode.com/GitHub_Trending/flue1/flue创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考