@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 本地持久化（用于 Berth 之外的 Bun 进程；Berth Worker 没有本地卷，请使用 libSQL）：
import { openBunDocumentStore } from '@evumi/berth/documents/bun';
const db = await openBunDocumentStore('./app.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的并发检查