集成

集成是一个定义：一个驱动及其有类型的 configuration 与 credential、把行写入集合的 feeds，以及发送或提交记录的 outbound 规则。它位于 src/integration/<name>/+integration.ts，写作 integration('<name>', { … })；文件夹即其名称。一个外部账户——一个 WhatsApp 号码、一个邮箱、一个机器人、一个数据库、一个日历账户——是一个实例：管理员在“设置 → 集成”中配对的一行 sys_integration。凭据在宿主侧密封保存，从不进入访客代码。

声明集成

// src/integration/kingdee/+integration.ts (templates/trade-operations, excerpt)
import { integration } from '@norbital-ai/bolt';

const kingdee = integration('kingdee', {
  description: 'Accounting SQL Server CDC and approved order submissions.',
  driver: 'sqlserver',
  credential: { connectionString: { kind: 'text', label: 'SQL Server connection string' } },
  configuration: { maxLagMinutes: { kind: 'int', min: 1, default: 15 }, … },
  feeds: {
    products: { into: 'products', key: ['external_code'], cron: '*/15 * * * *', removed: { active: false } },
    …
  },
  outbound: {
    orders: { from: 'orders', on: ['create', 'update'], when: { status: { eq: 'confirmed' } }, via: 'ledger', version: 'seal_hash' }
  }
});
export default kingdee;

// one page per run: { changes, cursor, more }
kingdee.feed('products', {
  read: function* (cursor, ctx) {
    return yield* readCdc(projections.products, 'external_code', cursor, ctx, …);
  }
});

// the committed order's payload (null skips it), then the command under its idempotency key
kingdee.outbound('orders', {
  build: (row) => (row.status !== 'confirmed' || !row.seal_hash ? null : { order_id: String(row.id), … }),
  run: function* (payload, ctx) {
    yield* verifyOrder(payload, ctx);
    return { id: yield* submitOrder(payload, ctx.key, ctx) };
  }
});

driver 为 http、sqlserver、calendar、whatsapp、email、telegram、slack、discord 或 wechat 之一。configuration 存放安全、可在运行时编辑的输入字段；credential（仅 http 与 sqlserver）存放管理员配对时填写的机密字段。http 定义声明其 baseUrl 与 auth——以凭据字段作为 bearer、header 或 basic——或是以密钥命名的定义级 oauth 客户端。函数体在字面量之后按名称附加：i.feed('<feed>', …)、i.outbound('<rule>', …)，以及可选的 i.test(…)。会话驱动自带同名的内置定义；只有在需要附加数据源或出站规则时，才在其上声明你自己的定义。

数据源

数据源把行写入 (into) 一个集合，以其远端 key 字段与实例为键，实例由引擎写入。read 数据源按其 cron 运行并附加 read(cursor, ctx)，每次返回一页：其变更、下一个游标与 more。driver 数据源映射实例链接推送的内容（日历事件，每个日历一个游标），messages 数据源映射实例已存储的消息。一页的行、标记与游标在一条语句中提交，未变化的页面不会增加修订号。removed 规定远端删除时如何处理：删除该行（默认）或修补它。数据源目标集合不暴露 create、delete 或操作；其 update 输入中的列是它的本地字段，只由你这一侧写入，数据源从不写入。

出站

出站规则监视一个集合（from）中符合 when 的创建或更新，并在记录自身的提交中为每条记录排入一次投递，前提是任何审批已使其生效。via 指定连接，或记录中保存其实例的字段；version 在该字段变化时再次发送。投递时，build(row) 依据已提交的行返回负载，返回 null 则跳过。在会话驱动上，负载是一条消息，events 依据投递回报（sent、delivered、opened、bounced、failed、replied）修补该记录；在 http 或 sqlserver 上，run(payload, ctx) 以 ctx.key（不可变的幂等键）执行命令。读取该集合时，$outbound 携带每条规则的状态。

实例与连接

// src/+workspace.ts (templates/on-demand-services, excerpt): the instances source addresses, by slot
connections: {
  customer_mail: { integration: 'customer_mail_delivery' },
  whatsapp: { integration: 'whatsapp' },
  google_routes: { integration: 'google_routes' }
}

// src/integration/google_routes/+integration.ts: the API key is sealed when an administrator pairs the instance
export default integration('google_routes', {
  description: 'Google Routes for drive times and traffic.',
  driver: 'http',
  credential: { apiKey: { kind: 'text', label: 'Google Routes API key' } },
  http: {
    baseUrl: 'https://routes.googleapis.com/',
    auth: { header: { name: 'X-Goog-Api-Key', value: 'apiKey' } }
  }
});

// in an automation's generator run (excerpt): absence is allowed, other failures propagate
import { Effect } from 'effect';
const answer = yield* ctx.request('google_routes', {
  method: 'POST',
  path: 'distanceMatrix/v2:computeRouteMatrix',
  body: { origins, destinations, travelMode: 'DRIVE' },
  output: { kind: 'list', of: { kind: 'object', fields: { duration: { kind: 'text', optional: true } } } }
}).pipe(Effect.catchTag('Unavailable', () => Effect.succeed(null)));
yield* ctx.send('whatsapp', { to: helper.phone, text: 'A warning has been filed on your record. …' });

实例是一行 sys_integration：它的定义、名称、active、配置，以及在会话实例上所部署的 Envoy。+workspace.ts 中的 connections 为源码要引用的实例命名——connections: { mail: { integration: 'email' } } 会创建一个以该槽位为 id 的实例；其余实例在运行时创建。记录通过指向 sys_integration 的关系引用实例，并以 where: { integration: { eq: '<definition>' } } 固定其定义。自动化用 ctx.send(<连接或实例>, …) 发送消息，用 ctx.request(<连接或实例>, operation) 调用 http 或 sqlserver 实例，凭据由宿主解析。实例未配对时，ctx.request 以 Unavailable 类型失败；用 Effect.catchTag 处理该失败。

运行

数据源以 <definition>.<feed> 的形式在每个活跃实例上运行，且只在正式环境中运行；一个失败的实例不会阻塞其他实例，其健康状况会写入该实例的状态。集合工具栏启动的也是同一个运行。消息投递有次数上限的重试；命令会一直重试，直到被确认。实例依据 sys_integration 上的普通授权进行管理。