@atmo-dev/contrail
尔时前. 期待打破变化.
一个用于索引和查询公开的AT协议记录的软件包.
基本用途
import { Contrail } from "@atmo-dev/contrail";
const contrail = new Contrail({
namespace: "com.example",
db,
collections: {
event: {
collection: "community.lexicon.calendar.event",
queryable: {
mode: {},
startsAt: { type: "range" },
},
searchable: ["name", "description"],
relations: {
rsvps: {
collection: "rsvp",
field: "subject.uri",
groupBy: "status",
groups: {
going: "community.lexicon.calendar.rsvp#going",
},
},
},
},
rsvp: {
collection: "community.lexicon.calendar.rsvp",
references: {
event: { collection: "event", field: "subject.uri" },
},
},
},
});
await contrail.init();
await contrail.backfillAll({ concurrency: 100 });
仓库范围收藏
设定 repos 在要回填的集合中, 仅索引明确的出版
listItem: {
collection: "app.bsky.graph.listitem",
repos: ["did:plc:yourbot"],
queryable: { list: {}, subject: {} },
}
与 discover: false (来自已知储存库的可靠记录)
独家出版商的许可列表 带有自己的精确回填对。 这需要
中,不能与 discover,并适用于摄入,
后补并通知。 更改注册的许可列表需要投影
重建 。 见 仓库范围配置
用于生命周期、运输和新鲜度保障.
查询
const result = await contrail.query("event", {
filters: { mode: "in-person" },
sort: { countType: "rsvp", direction: "desc" },
limit: 20,
});
HTTP线路暴露出相同的管道,包括关系/参考水分,计数,剖析,搜索和自定义查询.
保持记录的当前
await contrail.ingest(); // bounded Jetstream cycle
// Optional per-cycle overrides (defaults: 25s, 250 candidates,
// 250 distinct identity updates, and 4 MiB).
await contrail.ingest({
maxDrainMs: 20_000,
maxCandidates: 200,
maxIdentityUpdates: 200,
maxSerializedBytes: 3 * 1024 * 1024,
});
await contrail.runPersistent({
signal: abortController.signal,
batchSize: 50,
flushIntervalMs: 5_000,
});
在给用户的PDS写完之后, contrail.notify(uri) 可以立即获取权威记录。 只有权威的未找到的响应才会删除本地状态;速率限制,服务器出错,超时,反应不整齐,网络故障也让其保持不变. 公共HTTP业务的认证和滥用控制仍在设计中.
Contrail存储源事件时间,寄存器修订,源光标,CID,以及本地索引时间与记录/应用时间分开. 现场摄入能说出Jetstream v2 并检查它独特的单调 seq; RFC 3339事件时间仍为显示/源元数据,从未作为恢复坐标. 预定的收集在录取前会降低精确的观察,计出UTF-8记录机构外加固定的512字节元数据备抵,并在停止前保留了入门候选人. 身份更新有一个独立的保留数量限制,由DODE合并,并在数据库批次中承诺;超量的全球身份流量被记入但省去,因此无法饿死记录进度. 来源过滤的观测结果既不占用候选数据,也不占用字节限制,但其产生的后继数据仍然入账。 持久的行为者范围来自身份、中继发现和可见的可发现记录。 已存在的v1分时戳光标会跨出一个有边界的重叠桥并一次从原子上向v2后域过渡. 如果一个时间戳或以下光标已经掉入了主机活性保留窗口之外,摄取失败,而不是接受服务器的时间戳夹并默默地跳过一个缺口;从新边界重建/回填,直到添加了存档重放. 永恒的墓碑拒绝死灰复燃 现场投影承诺同一笔交易中 准确的分出 定期模式和持久性模式都需要一种固定的Jetstream v2服务,因为以下模式是例-局部的。 持续的摄入量保留了流出分批的生命周期,不会继承预定的计数/字节缺省. 历史获取仍然使用现有的PDS或可选的冲积道;Jetstream v2存档回放尚未被使用. 一个成功的PDS listRecords 页面是当前权威的观察,因此它取代了较旧的耐用状态而不读取冗余的版本;其版本写作和页面光标仍然在原子上承诺. 墓碑被无限期地保留;权威的重建/保留工具另行规划.
升级现有数据库时,点 jetstreams 在一个 v2 服务和更改中 orderedSource.epoch 确定新的连续性历史。 正常的服务来源在时间戳到seq的过渡之前是永久地被约束的,因为以下各段是实例-局部的;以后改变它是一个困难的错误。 初始化只允许在持久光标仍然位于遗留时间戳域时更改被命令源的时代;第一个完全处理的v2事件然后单向安装下个域. 一旦一个数据库在以下域中,一个时代的错配仍是一个硬错误. 一个没有持久服务约束的已存在seq域数据库也被拒绝,因为其起源实例无法安全推断.
投影胜者选择在写入交易中再次被守护 。 重叠的克龙、持续、通知或回填写器不能执行一条坚固的条条、所衍生的投影、墓碑或源检出点;改变后的前身将整个尝试卷回,从新的持久状态重复.
交易变更日志(试行)
新鲜的空后代可以选择一个紧凑的交易预测变化日志:
const config = {
// ...
changes: {
consumers: {
search: {
collections: ["community.lexicon.calendar.event"],
phases: ["historical", "live"],
initial: "current",
requiredForActivation: true,
},
},
},
} satisfies ContrailConfig;
静态的定义没有处理器,URL,客户端,证书,或秘密. 对比用随机生成数据库的ID和收集/阶段覆盖分类账进行登记。 一个中选的逻辑 put/ 删除附件会在与Cononical和衍生状态以及源检查点相同的交易中,引用一个紧凑的 URI/版本引用. 重复, Stale, 同型CID, 缺席-删除, 拒绝, 和回滚突变附加什么。 记录体被后期的交付层从当前状态水分而来,而不是被复制入日志.
Cloudflare工人与静态政策分开
export default createWorker(config, {
deliveries: {
search: async (batch, { env, signal }) => {
await updateSearch(env.SEARCH_KEY, batch, signal);
},
},
changeBootstraps: {
search: {
snapshot: async (page, { env, signal }) => {
await writeCandidateIndex(env.SEARCH_KEY, page, signal);
},
activate: async (activation, { env, signal }) => {
await activateCandidateIdempotently(
env.SEARCH_KEY,
activation.bootstrapToken,
signal,
);
},
},
},
});
被安排的行刑会进行摄取,经过适当的历史重试,然后限定公平交付的回合. 一个消费者的失败一直伴随着退缩,并且不会停止摄取,或者另一个消费者. 成功通知请求时间表 ExecutionContext;投影成功永远不取决于它. 持续部署 contrail.runPersistentDeliveries() 并行 runPersistent()。 。 。 .
低水平交付使用有限度的租约和比较和分期确认:
const claim = await contrail.changes.claim("webhooks", {
maxBatches: 20,
maxChanges: 500,
maxBytes: 512_000,
leaseMs: 30_000,
});
if (claim) {
try {
const batch = await contrail.changes.hydrate(claim);
await deliverIdempotently(batch);
await contrail.changes.ack(claim);
} catch {
await contrail.changes.fail(claim, {
code: "destination_unavailable",
nextAttemptAt: Date.now() + 30_000,
});
}
}
主张结合了重复的URI,水合物在以设定为方向的收集查询中,并解决从最新犬科状态删除/再生种族的问题. 消费者独立租赁和进步;不相干的位置范围提前,而不要求处理者。 交货是有意的,至少一次——目的地成功,然后是确认崩溃,造成重复交货。 处理器必须用稳定的记录/文件密钥来处理.
initial: "current" 使用持久的快照加尾注协调员,必须同时遵守 historical 财务报告和审定财务报表 live 因此,每次突变赛车都保留了按键集扫描进行调节。 一再声称并一意孤行 contrail.changes.claimSnapshotPage(),然后排出 claimBootstrapChanges() 通过其固定目标使用普通水合物/ack. 抓图和靴子尾巴的交付带有候选目的地标志;正常的激活后交付则不会。 最终宣称稳定出产范围活化符号 claimActivation(),并调用 completeActivation()。崩溃会重放相同的URI页面、尾端范围或激活符。 记录更新或被删除的同时,按键集扫描比赛由锚尾校正.
当一个消费者的收集/相对已全部覆盖时,可以将其添加到人所居住的日志中;其当前/未来的锚地是原子电流头,而历史始于所保留的地层。 在没有新一代或明确的老作家安静边界的情况下,扩大覆盖范围仍然无法关闭. contrail changes status, (中文(简体) ), retry, (中文(简体) ), prune,并明确确认 skip 曝光SQLite或Wrangler D1部署的私人行动. 状况包括保守的预测/头/批/背书计划。 Pruning以最慢的耐用消费者/薄荷锚为界。 跳过记录一个有限度的私人审计理由,从不隐含发生. 损坏或删除已存在的日志仍为已关闭 。 没有配置的消费者, 则不存在更改日志表格或附件写入 .
备份保留了数据库的更改日志生成ID,位置,租赁,靴子纹章,以及消费者的进步. 恢复作为同一代人的备份是可恢复的,并可能在期满后重新交付飞机租赁。 在重用目的地光标时不要将其复制为新的部署代号: 新建投影数据库并运行当前状态的靴子陷阱 。 恢复没有相匹配的变换表的投影表是一个无法恢复的交付缺口,必须输入重设/陷阱,而不是默默地继续数字位置.
地方发展
从验证记录Lexicons的种子开始一个项目,然后开始一个完整的本地服务:
contrail init my-appview \
--prefix community.lexicon.calendar. \
--namespace com.example.calendar
cd my-appview
contrail dev
前缀与源 Lexicon NSIDs 字面匹配. 包含线索
. 以选择命名空间边界. --namespace 独立:它控制
生成的 Appview XRPC 方法 ID 和默认 com.example。 。 。 .
| 输入选项 | 目的 |
|---|---|
--prefix <prefix> | 导入匹配 NSID 前缀的已核实记录 Lexicons |
--namespace <namespace> | 设置生成的 Appview XRPC 命名空间 |
--lexicon-api <url> | 选择一个登记册( 默认 ): https://lex.atmo.tools(中文(简体) ) |
--timeout <seconds> | 设置准备/依赖性预算( 默认 ): 60(中文(简体) ) |
--allow-partial | 明确接受不完整的前缀发现 |
--no-interactive | 未配置模糊的引用 |
前缀初始化需要一个稳定、完全核实和索引化的目录
缩写、标出每个根和中转依赖
lexicons/pinned/,写出来源 lexicons/pinned.lock生成
在所有支持的数据库中安全的平等/距离字段。 失踪
需要的依赖性总是失败,包括 --allow-partial。 。 。 .
在交互式终端中,可连接到一个
导入目标收藏和可选暴露为反向关系. 这些
选项永远不会从描述中猜出. --no-interactive 离开所有
模糊的参考文献未配置 。 强Ref水合能解决水流
由 URI 编辑的索引记录; 它不检索由
内置的CID.
运行 contrail init my-appview 无 --prefix 创建原始离线
没有网络请求的示例配置。 在没有目录的情况下运行任一窗体
写入当前目录。 命令拒绝替换配置或
在任何标准自有位置上被绑定的导入 .
没有Wrangler的配置,它创建或恢复 .contrail/dev.sqlite,解决AT协议网络中缺少配置的记录/ref Lexicons,而不覆盖项目拥有的系统,运行PDS回填,为完整的公众Contrail发现/XRPC/Lexicon表面服务 http://127.0.0.1:8787,并运行有界喷气流摄入每分钟。 它从不创建或改变部署提供者锁. 添加 .contrail/ 到工程忽略文件。 已存在的Wrangler项目自动保留了之前的D1开发行为.
当配置的 Lexicon 尚未存在或无法解决时, 交互 contrail dev 并配置后缀 contrail connect 警告并询问是否继续使用不完整的本地捆绑 。 未解析的文档和依赖于它们的计划省略,直接受影响的记录值产生为 unknown; 使用 validate: true 仍然需要他们完全的Lexicon关闭。 即时默认为否,非交互使用失败,除非 --yes 通过.
有用的 SQLite 选项包括:
contrail dev --fresh # reset local SQLite first
contrail dev --temporary # delete SQLite when stopped
contrail dev --sqlite ./tmp/dev.sqlite # explicit durable path
contrail dev --alluvium --allow-partial # fast base/archive bootstrap
contrail dev --no-backfill # serve existing state only
直接从自有配置或其包含目录生成输入的消费者 API :
contrail connect ./contrail.config.ts
contrail connect ../api
一个配置源生成 Lexicons, Atcute 类型, contrailApi, (中文(简体) ), createContrailClient(),以及 createLocalContrailClient() 不创建或更改 contrail.lock.json。如果生产锁已经存在,生成的默认客户端保留该部署目标,而本地工厂则使用当前源 API. contrail dev --config ../api/src/contrail.config.ts 然后可以在当地运行相同的配置.
SQLite dev模式仅在回放时曝光所配置的保护方法,而不会继承一个制作服务-auth的受众. 何时 notify 被省略,它也允许只打开回放 notifyOfUpdate。 。 。 . createLocalContrailClient() 没有 OAuth 范围,只允许对已验证的回路端使用平面 HTTP .
新生代(实验)
PdsSnapshotSource, (中文(简体) ), JetstreamChangeSource, (中文(简体) ), DatabaseBootstrapTarget,以及 bootstrapFreshProjection() 建立一个未发布的数据库,其中包含抓取-首回播放。 捕捉标记在接力发现开始前是可持久的;PDS分区光标和Jetstream检查点用他们的记录来承诺. 完成重建了被推迟的预测,验证了总记录/版本的一致性,并只存储了被限定的失败类别.
Jetstream 生成回放需要操作员拥有的连续性时代和保留保证. 它使用真实的溪流事件作为标记,从壁上钟或静静的套接字. 可选的繁忙水印收藏可以证明进展而无需预测.
一个可选的实验性Alnuvium适配器消耗了现有的版本一的收集量表,验证了不可变的压缩对象,并且完全在存档的源边界上结束离线的靴子圈. CLI 支持 Wrangler/ D1 或明确的 SQLite 文件 :
contrail backfill --sqlite ./data/contrail.sqlite --alluvium \
--alluvium-epoch my-source-continuity-epoch \
--alluvium-source-url wss://jetstream1.us-east.bsky.network \
--allow-partial
由于Alnuvium Protocol v1没有发布一个,所以刻意要求时间. --alluvium-source-url 将上面显示的列表和默认信息所公布的确切遗留的 v1 来源命名为主机东部来源; 有意将其与 config.jetstreams,它选择靴后使用的 v2 直播服务. --allow-partial 只有在选定清单明确报告历史遗漏时才需要。 命令拒绝使用一个人口稠密的数据库,除非它恢复了持久的冲积胶靴状态。 相应的方案API是:
import { DatabaseBootstrapTarget } from "@atmo-dev/contrail";
import {
createAlluviumBootstrapSources,
createAlluviumLiveCursor,
} from "@atmo-dev/contrail/alluvium";
const sourceIdentity = {
id: "jetstream-us-east",
// Alluvium v1 does not publish an epoch yet; the operator owns this value.
epoch: "jetstream-us-east-2026-08",
url: "wss://jetstream1.us-east.bsky.network",
};
const source = createAlluviumBootstrapSources(config, {
endpoint: "https://alluvium-v0.atmo.tools",
source: sourceIdentity,
jetstream: {
retentionUs: 72 * 60 * 60 * 1_000_000,
watermarkCollections: ["app.bsky.feed.post"],
},
});
const target = new DatabaseBootstrapTarget(db, config, {
deferDerivedProjections: true,
// Commit the archived v1 time_us as the one-time v2 live bridge cursor.
liveCursor: createAlluviumLiveCursor(sourceIdentity),
});
await bootstrapFreshProjection({
collections: ["community.lexicon.calendar.event"],
...source,
target,
});
离线调用负载只包含基部外加已公布的存档尾部. 它不会重放直接喷气流到新标记的头部。 目标在解剖学上持续了 Alnuvium 最终存档 v1 time_us; 下一个v2活观察过渡,桥会协调到服务下行域,然后由下行恢复Cron引用. 所配置的活 v2 服务和连续性时代仍然是操作员拥有的,因为 Alnuvium 协议 v1 无法发布 v2 交接坐标 .
这个第一个适配器只使用一代,而且刻意严格:选定的清单必须暴露一个共享的基底和归档边界,源ID/URL必须匹配配置,已知的捕获漏洞失败,部分历史覆盖需要 allowPartial: true它缓冲每个被绑定的被压缩对象,因此在预测出任何一行之前将校验和字节长度。 由于版本一显示缺乏全局多收集送出步骤,它也以有界的10万兆克尾部缓冲器来恢复源令;较大的尾部需要计划中的捆绑协议. 它不会默默地回到购买PDS;省略的账号修复和依赖性简介/之后的浓缩仍然是单独核算的PDS工作.
一个单独的控制数据库可以使用 DatabaseGenerationRegistry 以存储不可移动 (code, definition, database, generation) tu. activate(candidate, expectedActive) 切换一个单通指针与比和相接相接,保留先前的已准备好的倒推。 故意没有百分率的交通分化API;平台的出行路线必须解决活性拖拉.
生成克星
从 Contrail 配置生成 XRPC 查询 Lexicons, 并检查它们是否漂移 :
pnpm contrail lexicons generate
pnpm contrail lexicons check
contrail lexicons all 也拉出引用的记录 Lexicons并运行 Atcute TypeScript 生成. 使用 --public 仅生成远程服务合同所公布的方法,包括明确保护的服务-自动方法。 已生成 lex.config.js 文件带有所有权标记;现有的用户拥有的Atcute配置从未被替换。 过 --no-atcute-config 来管理这个文件 生成器也从 @atmo-dev/contrail/lexicons 供方案使用.
公共读取服务
export default createWorker(config, {
lexicons,
publicService: { endpoint: "https://api.example.com" },
});
发现在 /.well-known/contrail 发布一个经过验证的API描述和内容地址Lexicon捆绑的广告. 匿名收集读取,剖析,提供,以及作者自定义查询,可以获取公开的AT协议数据并改进回复后的缓存.
个性化饲料和 notifyOfUpdate 反之,可以要求有方法约束的AT 礼宾服务标志:
const config = {
notify: true,
serviceAuth: {
audience: "did:web:api.example.com#contrail",
methods: ["getFeed", "notifyOfUpdate"],
},
// ...
};
服务描述广告是基础服务DOD,精确的分散受众,canonic OAuth范围,以及每个受保护的查询/程序与匿名方法分开. contrail connect 生成 src/contrail/index.ts;匿名呼叫直接到所配置的终点,同时被保护的呼叫懒散地发现并验证所有服务-自动字段. 它的出口 contrail.scope 包含一个排序 lxm 每个保护方法的参数,例如 rpc?aud=did:web:api.example.com%23contrail&lxm=<namespace>.getFeed&lxm=<namespace>.notifyOfUpdate。 。 。 . contrail.authenticated(authenticatedClient) 返回一个 Atcute 客户端: 向 Contrail 公示的提供者方法路由、 PDS 的普通方法路由以及成功的跟踪 createRecord, (中文(简体) ), putRecord,以及 deleteRecord 电话自动通知 Contrail。 Handle-form 删除被解析为 Cononical DID URI 。 被保护方法使用被缓存的精确方法约束符,而瞬态发现失败仍然可以重试. 饲料演员必须决心给象征性的发售人,每一个被通知的AT URI必须属于发售人. 当基地解决了公共终点 /.well-known/did.json,"工人"发表一份DD文件,其服务条目ID是准确的受众.
公共服务模式需要主命令源,所以 getCursor 能够暴露其坚定的立场。 明确配置连续性身份; 不公开部署则只显示未扫描的数字活光标为 { cursor }数字 :
const config = {
orderedSource: {
source: "jetstream",
epoch: "primary-2026", // change whenever cursor continuity changes
},
// ...
};
命令源光标不透明 。 比较完整 { source, epoch, cursor } 等值; 永远不要命令不同时代的光标。 消费者可以在查询前后读取位置,如果更改后重新试取,再作为回溯/验证信号进行投票.
将独立的消费者与 contrail connect <https-origin>. 版本2提供者锁记录了部署和准确的Lexicon捆绑,但生成的客户端并没有在运行时给提供者的完整方法设定下钉. 因此,当提供者增加方法时,现有的匿名方法继续发挥作用。 重复连接同一端点和提供者拥有的产出根需要 --update; 提供者文件和锁被上演并互换,而不删除消费者拥有的Lexicons. 版本一锁必须去掉并重新连接.
运行时记录验证
将其索引政策公布后, 将文件集进行校验 :
const config: ContrailConfig = {
namespace: "com.example",
collections: {
event: {
collection: "community.example.event",
validate: true,
},
legacy: {
collection: "community.example.legacy",
// Omitted or false: compatibility mode, no runtime validation.
},
},
validation: {
strict: true, // default: enforce blob size/MIME constraints too
verifyCid: true, // default: canonical DAG-CBOR CID verification
},
};
createWorker(config, { lexicons }) 将每个选入的收藏绑定在部署时已经运出的确切生成/绑定的捆绑上. contrail dev 解析并绑定其精确的临时捆绑,同时 CLI 回填装标准检查生成的捆绑。 没有第二个验证特定的Lexicon阵列.
直接运行时间在政策和材料之间使用同样的明确区分:
const contrail = new Contrail({ ...config, db, lexicons });
验证同样适用于喷气流、PDS回填、通知、点取浓缩、冲积和直接 ingestRecords() 打电话 当绑定捆绑中没有选择的收藏或中转引用时启动失败 。 权威来源必须提供匹配的CID;本地和星座合成记录默认可能是无CID. 覆盖 allowCidlessSources 仅供明确信任的合成适配器使用.
私人总和拒绝计数器的可用性不暴露DDE、URI、错误或记录机构。 同时期的批量回填会将这些被捆绑的计数器堆积在内存中并冲出一次,因此诊断不能在每一个源页上变成热D1行:
const diagnostics = await contrail.diagnostics();
HTTP 密码
import { createHandler } from "@atmo-dev/contrail/server";
const handle = createHandler(contrail);
const response = await handle(request, db);
对于工人,使用 createWorker 从 @atmo-dev/contrail/worker。 。 。 .
适配器
import { createSqliteDatabase } from "@atmo-dev/contrail/sqlite";
import { createPostgresDatabase } from "@atmo-dev/contrail/postgres";
import { createPgliteDatabase } from "@atmo-dev/contrail/pglite";
Contrail需要22.js节点或更新. SQLite 适配器使用内置 node:sqlite 模块; D1 直接执行 Contrail 的数据库接口.
PGlite 读取
运行嵌入式的 PostgreSQL 没有服务器或容器. 安装可选对等依赖 :
pnpm add @electric-sql/pglite@^0.5.11
import { PGlite } from "@electric-sql/pglite";
import { Contrail } from "@atmo-dev/contrail";
import { createPgliteDatabase } from "@atmo-dev/contrail/pglite";
const pglite = new PGlite("./data/contrail"); // omit the path for an in-memory database
const contrail = new Contrail({
namespace: "com.example",
db: createPgliteDatabase(pglite),
collections: {
event: { collection: "community.lexicon.calendar.event" },
},
});
await contrail.init();
// After stopping ingestion and finishing pending queries:
await pglite.close();
PGlite使用相同的PostgreSQL方言,JSON处理和全文搜索作为Postgres适配器. Batches是原子,PGlite在其单连接上将交易序列化;它不是一个多连接的PostgreSQL服务器. 呼叫者拥有PGlite实例及其存储生命周期. 这是一个程序适配器; contrail dev 仍然使用 SQLite.