@evumi/berth

Berth 项目 Queues（队列）和 Alarms（定时器）的客户端，用法照 Cloudflare Queues 和 Durable Object alarms。也负责处理平台推给应用的请求（队列批次、定时器、Cron HTTP 目标），并校验签名。主入口零运行时依赖，支持 Node ≥ 18、Bun、Deno 和 edge 运行时。文档数据库是独立子路径，驱动按需安装。

项目开了 queues 或 alarms 能力后，Berth 会给容器注入 BERTH_CONFIG_URL、BERTH_CONFIG_TOKEN 和 BERTH_SIGNING_SECRET。环境是请求来源的那个环境：预览环境只会用到自己的队列和定时器。

此 SDK 用于服务端（包括 API 路由和 edge Worker）。平台 Token、签名密钥、Cron 密钥和 KV 连接凭据都不能下发到浏览器。前端通过应用后端操作，后端负责鉴权并只返回业务结果。不要把这些变量复制到 NEXT_PUBLIC_*、VITE_*、打包器环境变量映射或客户端 props，也不要把处理器收到的 env 快照返回给客户端。

发消息（producer）

import { queue } from '@evumi/berth';

const EMAILS = queue('emails'); // 放在模块顶层即可：环境变量到发请求时才读

await EMAILS.send({ to: '[email protected]', template: 'welcome' });
await EMAILS.send({ to: '[email protected]' }, { delaySeconds: 600, dedupId: 'welcome:42' }); // 24 小时内同一 dedupId 只存一次
await EMAILS.send('plain text', { contentType: 'text' });
await EMAILS.send(new Uint8Array(png)); // 自动用 bytes

const { ids } = await EMAILS.sendBatch([{ body: { n: 1 } }, { body: { n: 2 }, delaySeconds: 30 }], { delaySeconds: 5 });

contentType 是 json（默认，任意 JSON 值）、text（字符串）或 bytes（Uint8Array / ArrayBuffer）。send 返回 { id, deduplicated }。

自己建客户端：

import { createClient } from '@evumi/berth';

const berth = createClient({ url: process.env.BERTH_CONFIG_URL, token: process.env.BERTH_CONFIG_TOKEN, timeoutMs: 10_000 });
const EMAILS = berth.queue('emails');
const trialEnd = berth.alarm('trial-end:42');

接收推送（consumer）

队列的消费者（push 模式）、定时器的目标服务和 Cron HTTP 目标都指向同一个处理器即可，靠 Berth-Source 区分。createHandler 会先用原始请求体校验 Berth-Signature（默认读 BERTH_SIGNING_SECRET，失败回 401），再调用对应函数：

import { createHandler } from '@evumi/berth';

export const handler = createHandler({
  async queue(batch, env) {
    for (const m of batch.messages) {
      try {
        await sendEmail(m.body);   // m.id, m.timestamp (Date), m.attempts, m.contentType
        m.ack();
      } catch {
        m.retry({ delaySeconds: 60 }); // 不传则按队列退避
      }
    }
    // 没 ack 也没 retry 的消息在函数正常返回后算确认；也可以 batch.ackAll() / batch.retryAll({ delaySeconds })
  },
  async alarm(info) {
    // info.key, info.scheduledTime (ms), info.data, info.retryCount, info.isRetry
    await endTrial(info.data);
  },
  async scheduled(controller) {
    // controller.cron, controller.name, controller.scheduledTime (ms)
    await cleanup();
  },
});

函数抛异常时回 500，平台会重试：队列是整批重试，定时器最多重试 6 次。某个来源没有对应函数时回 404。env 是 process.env 的快照；第三个参数 ctx.waitUntil(promise) 会让响应等这些 promise 结束。

Next.js（App Router）、Hono、Bun、Deno：处理器就是 (request: Request) => Promise<Response>：

// app/api/berth/route.ts
import { handler } from '@/lib/berth';
export const POST = handler;

Node http / Express：用 toNodeHandler。它直接读原始请求流，不依赖 body parser。前面如果已经用了 express.raw()，它会用 req.body（Buffer 或字符串）。不要把它挂在 express.json() 后面：

import express from 'express';
import { toNodeHandler } from '@evumi/berth';
import { handler } from './berth.js';

const app = express();
app.post('/berth', toNodeHandler(handler)); // 放在 app.use(express.json()) 之前

只想自己校验签名时用 verifySignature(secret, header, rawBody, { toleranceSeconds = 300 })。它是同步的，返回布尔值，不抛异常。checkSignature(...) 返回失败原因：'missing' | 'malformed' | 'expired' | 'mismatch'，通过时返回 null。

拉取（pull 队列）

和 Cloudflare 的 HTTP pull 一样：

import { queue } from '@evumi/berth';

const JOBS = queue('jobs');
for (;;) {
  const { messages } = await JOBS.pull({ batchSize: 10, visibilityTimeoutMs: 60_000 });
  if (messages.length === 0) {
    await new Promise((r) => setTimeout(r, 1000));
    continue;
  }
  const acks: string[] = [];
  const retries: { leaseId: string; delaySeconds?: number }[] = [];
  for (const m of messages) {
    try {
      await handle(m.body); // m.id, m.leaseId, m.attempts, m.timestamp (ms), m.contentType
      acks.push(m.leaseId);
    } catch {
      retries.push({ leaseId: m.leaseId, delaySeconds: 30 });
    }
  }
  await JOBS.ack({ acks, retries }); // → { acked, retried, dead }
}

visibilityTimeoutMs 内没 ack 的消息会换一个新 leaseId 再发一次。过期的 lease 会被忽略。

定时器（Alarms）

每个 key 一个，作用范围是整个环境，和 Durable Object 的 storage.setAlarm 一样：再设一次就是改时间，删掉就是取消。

import { alarm } from '@evumi/berth';

const trialEnd = alarm(`trial-end:${userId}`);
await trialEnd.setAlarm(Date.now() + 14 * 86_400_000, { data: { userId } }); // ms 或 Date；最多 365 天后
const when = await trialEnd.getAlarm();   // ms，没有则为 null
const pending = await trialEnd.get();     // { key, scheduledTime, data } 或 null
await trialEnd.deleteAlarm();             // 没有也不报错

错误

请求失败抛 BerthError，带 status、code、message（429 时还有 retryAfterMs）。code 取服务器返回的值：unauthorized、capability_disabled、not_found、bad_request、conflict、too_large、rate_limited、quota_exceeded、unavailable、queue_full。客户端自己的 code：timeout、network、invalid_response、missing_env。错误信息里不含令牌。

文档数据库（SQLite / libSQL）

@evumi/berth/documents 提供一个刻意收小的 Mongo 风格接口，不用定义业务表或 schema。底层是一张 JSON 文档表；不是 MongoDB 协议、Mongoose 或完整 MongoDB 兼容层。

Bun 本地持久化（Worker 默认路径是 /data/worker.sqlite）：

import { openBunDocumentStore } from '@evumi/berth/documents/bun';

const db = await openBunDocumentStore(Bun.env.BERTH_SQLITE_PATH ?? '/data/worker.sqlite');
const notes = db.collection<{ title: string; views: number; author?: { name: string } }>('notes');
const { insertedId } = await notes.insertOne({ title: 'hello', views: 0 });
await notes.updateOne({ _id: insertedId }, { $inc: { views: 1 } });
const page = await notes.find({ views: { $gte: 1 } }).sort({ views: -1 }).limit(20).skip(0).toArray();
await notes.createIndex({ views: -1 });
await db.close(); // 通常在应用关闭时调用

远程 libSQL 复用 Berth 已有的绑定变量：

pnpm add @libsql/client

import { openLibSQLDocumentStore } from '@evumi/berth/documents/libsql';
const db = await openLibSQLDocumentStore({
  url: Bun.env.LIBSQL_URL!,
  authToken: Bun.env.LIBSQL_AUTH_TOKEN,
});

也能注入已有的官方 libSQL 客户端或自定义 adapter：

import { createClient } from '@libsql/client';
import { createDocumentStore } from '@evumi/berth/documents';
const client = createClient({ url: 'file:./app.sqlite' });
const db = createDocumentStore(client);
await db.ready(); // 幂等初始化；各 CRUD 也会等待初始化
// 低层工厂不接管 client 的生命周期或改变它的 PRAGMA。

Bun 和本地 file: 便利工厂使用 WAL、synchronous=FULL、busy_timeout=5000。远程工厂服从服务器的持久化设置；它不承诺复制、节点或磁盘丢失后的数据安全。busy_timeout 是 SQLite 锁等待，不会重放已发送的写操作。所有工厂均无透明重试；数据库错误可能表示写入已执行但响应丢失，不能据此直接重发 $inc。

API 和边界：

• insertOne(document) 返回 { insertedId }；可传 1–256 字符的字符串 _id，否则生成 UUID。findOne(filter?) 返回文档或 null；find(filter?) 返回不可变 cursor。updateOne(filter, update) 返回 { matchedCount }，不声称内容一定改变；deleteOne(filter) 返回 { deletedCount }；countDocuments(filter?) 返回匹配总数。
• 查询支持标量等值、$eq/$ne/$gt/$gte/$lt/$lte/$in/$exists、$and/$or。点路径仅遍历对象，例如 author.name；每段为 ASCII 字母或下划线开头，后接字母、数字、下划线。没有数组元素匹配、正则、任意 JS、聚合、join、upsert、unique index 或跨调用事务。
• null 只匹配显式 JSON null；missing 独立，$exists:false 只匹配 missing。$ne 包括 missing。布尔与数值严格区分；范围查询只比较同类的 number 或 string。$in:[] 不匹配任何文档。JSON 数值是 JS number；高精度业务值用字符串保存。
• 排序使用 SQLite JSON 提取值的排序：missing/null 同组，然后 number/boolean（false=0、true=1），然后 text、array/object 的 JSON 文本。字符串使用 SQLite BINARY 顺序；这个跨类型排序不是 MongoDB BSON 顺序。末尾追加 _id ASC 保证稳定结果；未指定排序时按 _id ASC。findOne/updateOne/deleteOne 选匹配中的最小 _id。
• 默认读取 100 条，最大 1000；limit(0) 是空结果。skip 最大 1,000,000，较大 offset 和无索引过滤会扫描；计数不受读取条数上限影响。过滤最多 64 条件、10 层逻辑嵌套、每个 $in 最多 100 项。
• $set 支持 JSON 值，$unset 的值为 true 或 1，$inc 支持有限数。missing 的自增从 0 开始；已有 null、布尔、字符串、对象、数组会让整次更新失败。父路径如果存在但不是对象，也失败；不允许重叠更新路径或修改 _id。每次更新是一条原子 SQL，最多 32 个字段，不会先读取到 JS 再写回。
• 单文档最多 1 MiB（UTF-8 JSON），最多 32 层；仅接受普通 JSON。undefined、NaN/Infinity、Date、BigInt、循环、getter、稀疏数组和自定义原型均拒绝。数组可以存储但没有数组查询语义。
• createIndex({ field: 1|-1 }) 创建最多 4 字段的普通表达式索引，返回名称。物理索引以 collection 为首列，覆盖同库全部 collection；相同规格共享索引，新增索引会影响所有 collection 的写入成本。查询中的 collection 和值始终绑定参数；字段路径经过校验，表达式与索引完全一致。
• collection 是命名空间，不是权限隔离。每个项目/环境应使用自己的数据库文件或凭据；持有同一数据库客户端的业务代码可以访问该库所有 collection。内部 _berth_documents 表不应手动改写。

错误为 DocumentError，code 是 invalid_input、database_error 或 invalid_data。数据库错误不包含 SQL 参数或文档内容。database_error 不说明远端是否已提交；需要业务幂等设计才能安全重试。

真实数据库测试和三条执行路径的复现基准见 bench/README.md。

开发

pnpm install
pnpm typecheck
pnpm test   # 先构建，再跑 vitest（含 sdk/testdata 的共享签名向量）
pnpm test:bun # Bun持久化烟测 + 4个Bun写进程、2个读进程共享SQLite的并发检查