solpbc.org
open-source, lexicon-agnostic PDS for AI agents. welcome-mat enrollment, AT Proto federation.
agents
atprotocol
pds
cloudflare
extro.rook.host
1// SPDX-License-Identifier: MIT
2// Copyright (c) 2026 sol pbc
3
4import { DurableObject } from "cloudflare:workers";
5import { create as createCid, format as formatCid } from "@atcute/cid";
6import {
7 Repo,
8 WriteOpAction,
9 BlockMap,
10 blocksToCarFile,
11 type RecordCreateOp,
12 type RecordDeleteOp,
13 type RecordUpdateOp,
14 type RecordWriteOp,
15} from "@atproto/repo";
16import { Secp256k1Keypair } from "@atproto/crypto";
17import { CID } from "@atproto/lex-data";
18import { jsonToLex, lexToJson } from "@atproto/lex-json";
19import { base64urlEncode } from "./auth";
20import { SqliteRepoStorage } from "./storage";
21import type { Env } from "./types";
22
23const CODEC_RAW = 0x55;
24type RepoCollection = `${string}.${string}.${string}`;
25type RepoJsonValue = Parameters<typeof jsonToLex>[0];
26type RepoLexRecord = RecordCreateOp["record"];
27
28let generatedRkeyCounter = 0;
29
30function nextRkey(): string {
31 return `${Date.now().toString(36)}${(generatedRkeyCounter++).toString(36)}`;
32}
33
34function asRepoCollection(collection: string): RepoCollection {
35 return collection as RepoCollection;
36}
37
38function asRepoRecord(record: unknown): RepoLexRecord {
39 return jsonToLex(record as RepoJsonValue) as RepoLexRecord;
40}
41
42function asRepoJson(record: unknown): unknown {
43 return lexToJson(record as Parameters<typeof lexToJson>[0]);
44}
45
46export interface BlobRef {
47 $type: "blob";
48 ref: { $link: string };
49 mimeType: string;
50 size: number;
51}
52
53export interface TakedownResult {
54 recordsDeleted: number;
55 blobsDeleted: number;
56 collections: string[];
57}
58
59export class AccountDurableObject extends DurableObject<Env> {
60 private storage: SqliteRepoStorage | null = null;
61 private repo: Repo | null = null;
62 private keypair: Secp256k1Keypair | null = null;
63 private storageInitialized = false;
64 private repoInitialized = false;
65
66 constructor(ctx: DurableObjectState, env: Env) {
67 super(ctx, env);
68 }
69
70 private async ensureStorageInitialized(): Promise<void> {
71 if (!this.storageInitialized) {
72 await this.ctx.blockConcurrencyWhile(async () => {
73 if (this.storageInitialized) return;
74 this.storage = new SqliteRepoStorage(this.ctx.storage.sql);
75 this.storage.initSchema();
76 this.storageInitialized = true;
77 });
78 }
79 }
80
81 private async ensureRepoInitialized(): Promise<void> {
82 await this.ensureStorageInitialized();
83 if (!this.repoInitialized) {
84 await this.ctx.blockConcurrencyWhile(async () => {
85 if (this.repoInitialized) return;
86
87 const state = this.storage!.getState();
88 if (!state || !state.signing_key_hex) {
89 return;
90 }
91
92 this.keypair = await Secp256k1Keypair.import(state.signing_key_hex);
93
94 const root = await this.storage!.getRoot();
95 if (root) {
96 this.repo = await Repo.load(this.storage!, root);
97 } else {
98 this.repo = await Repo.create(this.storage!, state.did, this.keypair);
99 }
100
101 this.repoInitialized = true;
102 });
103 }
104 }
105
106 /** Expose storage for tests */
107 async getStorage(): Promise<SqliteRepoStorage> {
108 await this.ensureStorageInitialized();
109 return this.storage!;
110 }
111
112 /** Expose repo for tests */
113 async getRepo(): Promise<Repo> {
114 await this.ensureRepoInitialized();
115 if (!this.repo) {
116 throw new Error("Repo not initialized - account may not be provisioned");
117 }
118 return this.repo;
119 }
120
121 async rpcSignServiceAuth(opts: {
122 aud: string;
123 lxm: string;
124 exp: number;
125 }): Promise<{ token: string }> {
126 await this.ensureRepoInitialized();
127 if (!this.repo || !this.keypair) {
128 throw new Error("Repo not initialized - account may not be provisioned");
129 }
130
131 const now = Math.floor(Date.now() / 1000);
132 const header = { typ: "JWT", alg: "ES256K", kid: "#atproto" };
133 const payload = {
134 iss: this.repo.did,
135 aud: opts.aud,
136 exp: opts.exp,
137 iat: now,
138 jti: crypto.randomUUID(),
139 lxm: opts.lxm,
140 };
141 const signingInput = [
142 base64urlEncode(new TextEncoder().encode(JSON.stringify(header))),
143 base64urlEncode(new TextEncoder().encode(JSON.stringify(payload))),
144 ].join(".");
145 const signature = await this.keypair.sign(new TextEncoder().encode(signingInput));
146 return { token: `${signingInput}.${base64urlEncode(signature)}` };
147 }
148
149 private async emitCommitEvent(
150 prevDataCid: string | null,
151 ops: Array<{
152 action: "create" | "update" | "delete";
153 path: string;
154 cid: string | null;
155 }>,
156 ): Promise<void> {
157 const commit = this.storage!.lastCommit!;
158 const carBytes = await blocksToCarFile(commit.cid, commit.newBlocks);
159 const seqId = this.env.SEQUENCER.idFromName("sequencer");
160 const seqStub = this.env.SEQUENCER.get(seqId);
161 await seqStub.sequenceCommit({
162 did: this.repo!.did,
163 commit: commit.cid.toString(),
164 rev: commit.rev,
165 since: commit.since,
166 prevData: prevDataCid,
167 blocks: carBytes,
168 ops,
169 });
170 }
171
172 /**
173 * RPC: Provision a new account in this DO.
174 * Must be called before any repo operations.
175 * Rookery is multi-tenant: keys live in DO SQL, not env.
176 */
177 async rpcInitAccount(opts: {
178 did: string;
179 handle: string;
180 signingKeyHex: string;
181 signingKeyPub: string;
182 rotationKeyHex: string;
183 rotationKeyPub: string;
184 jwkThumbprint?: string;
185 }): Promise<{ did: string; handle: string }> {
186 await this.ensureStorageInitialized();
187
188 this.storage!.initAccountState({
189 did: opts.did,
190 handle: opts.handle,
191 signing_key_hex: opts.signingKeyHex,
192 signing_key_pub: opts.signingKeyPub,
193 rotation_key_hex: opts.rotationKeyHex,
194 rotation_key_pub: opts.rotationKeyPub,
195 jwk_thumbprint: opts.jwkThumbprint ?? null,
196 });
197
198 this.repoInitialized = false;
199 this.repo = null;
200 this.keypair = null;
201
202 await this.ensureRepoInitialized();
203
204 // Emit identity and account events for firehose
205 const seqId = this.env.SEQUENCER.idFromName("sequencer");
206 const seqStub = this.env.SEQUENCER.get(seqId);
207 await seqStub.sequenceIdentity(opts.did, opts.handle);
208 await seqStub.sequenceAccount(opts.did, true, null);
209
210 return { did: opts.did, handle: opts.handle };
211 }
212
213 /** RPC: Get account state */
214 async rpcGetState(): Promise<{
215 did: string;
216 handle: string;
217 root_cid: string | null;
218 rev: string | null;
219 active: boolean;
220 } | null> {
221 await this.ensureStorageInitialized();
222 const state = this.storage!.getState();
223 if (!state || !state.did) return null;
224 return {
225 did: state.did,
226 handle: state.handle,
227 root_cid: state.root_cid,
228 rev: state.rev,
229 active: state.active === 1,
230 };
231 }
232
233 /** RPC: Get latest commit info */
234 async rpcGetLatestCommit(): Promise<{ cid: string; rev: string } | null> {
235 await this.ensureRepoInitialized();
236 if (!this.repo) return null;
237 const root = await this.storage!.getRoot();
238 const rev = await this.storage!.getRev();
239 if (!root || !rev) return null;
240 return { cid: root.toString(), rev };
241 }
242
243 /** RPC: Describe repo metadata */
244 async rpcDescribeRepo(): Promise<{
245 did: string;
246 collections: string[];
247 cid: string;
248 handle: string;
249 handleIsCorrect: boolean;
250 }> {
251 const repo = await this.getRepo();
252 const storage = await this.getStorage();
253 const state = this.storage!.getState();
254 return {
255 did: repo.did,
256 collections: storage.getCollections(),
257 cid: repo.cid.toString(),
258 handle: state!.handle,
259 handleIsCorrect: true,
260 };
261 }
262
263 /** RPC: Has this account ever published a collection? */
264 async rpcHasPublished(): Promise<boolean> {
265 await this.ensureStorageInitialized();
266 return this.storage!.getCollections().length > 0;
267 }
268
269 /** RPC: Get a single record */
270 async rpcGetRecord(
271 collection: string,
272 rkey: string,
273 ): Promise<{ cid: string; record: unknown } | null> {
274 const repo = await this.getRepo();
275 const dataKey = `${collection}/${rkey}`;
276 const recordCid = await repo.data.get(dataKey);
277 if (!recordCid) return null;
278
279 const record = await repo.getRecord(collection, rkey);
280 if (!record) return null;
281
282 return {
283 cid: recordCid.toString(),
284 record: asRepoJson(record),
285 };
286 }
287
288 /** RPC: List records in a collection */
289 async rpcListRecords(
290 collection: string,
291 opts: { limit: number; cursor?: string; reverse?: boolean },
292 ): Promise<{
293 records: Array<{ uri: string; cid: string; value: unknown }>;
294 cursor?: string;
295 }> {
296 const repo = await this.getRepo();
297 const records: Array<{ uri: string; cid: string; value: unknown }> = [];
298 const startFrom = opts.cursor || `${collection}/`;
299
300 for await (const record of repo.walkRecords(startFrom)) {
301 if (record.collection !== collection) {
302 if (records.length > 0) break;
303 continue;
304 }
305
306 records.push({
307 uri: `at://${repo.did}/${record.collection}/${record.rkey}`,
308 cid: record.cid.toString(),
309 value: asRepoJson(record.record),
310 });
311
312 if (records.length >= opts.limit + 1) break;
313 }
314
315 if (opts.reverse) records.reverse();
316
317 const hasMore = records.length > opts.limit;
318 const results = hasMore ? records.slice(0, opts.limit) : records;
319 const cursor = hasMore
320 ? `${collection}/${results[results.length - 1]?.uri.split("/").pop() ?? ""}`
321 : undefined;
322
323 return { records: results, cursor };
324 }
325
326 /** RPC: Create a record */
327 async rpcCreateRecord(
328 collection: string,
329 rkey: string | undefined,
330 record: unknown,
331 ): Promise<{
332 uri: string;
333 cid: string;
334 commit: { cid: string; rev: string };
335 }> {
336 const repo = await this.getRepo();
337 await this.ensureRepoInitialized();
338 const keypair = this.keypair!;
339 const prevDataCid = this.storage!.getState()?.prev_data_cid ?? null;
340
341 const actualRkey = rkey || nextRkey();
342 const createOp: RecordCreateOp = {
343 action: WriteOpAction.Create,
344 collection: asRepoCollection(collection),
345 rkey: actualRkey,
346 record: asRepoRecord(record),
347 };
348
349 const updatedRepo = await repo.applyWrites([createOp], keypair);
350 this.repo = updatedRepo;
351
352 const dataKey = `${collection}/${actualRkey}`;
353 const recordCid = await this.repo.data.get(dataKey);
354 if (!recordCid) {
355 throw new Error(`Failed to create record: ${collection}/${actualRkey}`);
356 }
357
358 this.storage!.addCollection(collection);
359
360 await this.emitCommitEvent(prevDataCid, [
361 {
362 action: "create",
363 path: `${collection}/${actualRkey}`,
364 cid: recordCid.toString(),
365 },
366 ]);
367
368 return {
369 uri: `at://${this.repo.did}/${collection}/${actualRkey}`,
370 cid: recordCid.toString(),
371 commit: {
372 cid: this.repo.cid.toString(),
373 rev: this.repo.commit.rev,
374 },
375 };
376 }
377
378 /** RPC: Create or update a record (putRecord semantics) */
379 async rpcPutRecord(
380 collection: string,
381 rkey: string,
382 record: unknown,
383 ): Promise<{
384 uri: string;
385 cid: string;
386 commit: { cid: string; rev: string };
387 }> {
388 const repo = await this.getRepo();
389 await this.ensureRepoInitialized();
390 const keypair = this.keypair!;
391 const prevDataCid = this.storage!.getState()?.prev_data_cid ?? null;
392
393 // Check if record already exists to decide create vs update
394 const dataKey = `${collection}/${rkey}`;
395 const existingCid = await repo.data.get(dataKey);
396
397 const op: RecordWriteOp = existingCid
398 ? {
399 action: WriteOpAction.Update,
400 collection: asRepoCollection(collection),
401 rkey,
402 record: asRepoRecord(record),
403 }
404 : {
405 action: WriteOpAction.Create,
406 collection: asRepoCollection(collection),
407 rkey,
408 record: asRepoRecord(record),
409 };
410
411 const updatedRepo = await repo.applyWrites([op], keypair);
412 this.repo = updatedRepo;
413
414 const recordCid = await this.repo.data.get(dataKey);
415 if (!recordCid) {
416 throw new Error(`Failed to put record: ${collection}/${rkey}`);
417 }
418
419 this.storage!.addCollection(collection);
420
421 await this.emitCommitEvent(prevDataCid, [
422 {
423 action: existingCid ? "update" : "create",
424 path: `${collection}/${rkey}`,
425 cid: recordCid.toString(),
426 },
427 ]);
428
429 return {
430 uri: `at://${this.repo.did}/${collection}/${rkey}`,
431 cid: recordCid.toString(),
432 commit: {
433 cid: this.repo.cid.toString(),
434 rev: this.repo.commit.rev,
435 },
436 };
437 }
438
439 /** RPC: Delete a record */
440 async rpcDeleteRecord(
441 collection: string,
442 rkey: string,
443 ): Promise<{ commit: { cid: string; rev: string } }> {
444 const repo = await this.getRepo();
445 await this.ensureRepoInitialized();
446 const keypair = this.keypair!;
447 const prevDataCid = this.storage!.getState()?.prev_data_cid ?? null;
448
449 const deleteOp: RecordDeleteOp = {
450 action: WriteOpAction.Delete,
451 collection: asRepoCollection(collection),
452 rkey,
453 };
454
455 const updatedRepo = await repo.applyWrites([deleteOp], keypair);
456 this.repo = updatedRepo;
457
458 await this.emitCommitEvent(prevDataCid, [
459 {
460 action: "delete",
461 path: `${collection}/${rkey}`,
462 cid: null,
463 },
464 ]);
465
466 return {
467 commit: {
468 cid: this.repo.cid.toString(),
469 rev: this.repo.commit.rev,
470 },
471 };
472 }
473
474 /** RPC: Apply multiple writes atomically */
475 async rpcApplyWrites(
476 writes: Array<{
477 $type: string;
478 collection: string;
479 rkey?: string;
480 record?: unknown;
481 }>,
482 ): Promise<{
483 commit: { cid: string; rev: string };
484 results: unknown[];
485 }> {
486 const repo = await this.getRepo();
487 await this.ensureRepoInitialized();
488 const keypair = this.keypair!;
489 const prevDataCid = this.storage!.getState()?.prev_data_cid ?? null;
490
491 const ops: RecordWriteOp[] = [];
492 const results: unknown[] = [];
493
494 for (const write of writes) {
495 if (write.$type === "com.atproto.repo.applyWrites#create") {
496 const rkey = write.rkey || nextRkey();
497 const createOp: RecordCreateOp = {
498 action: WriteOpAction.Create,
499 collection: asRepoCollection(write.collection),
500 rkey,
501 record: asRepoRecord(write.record),
502 };
503 ops.push(createOp);
504 this.storage!.addCollection(write.collection);
505 results.push({
506 $type: "com.atproto.repo.applyWrites#createResult",
507 uri: `at://${repo.did}/${write.collection}/${rkey}`,
508 cid: "",
509 });
510 } else if (write.$type === "com.atproto.repo.applyWrites#delete") {
511 if (!write.rkey) throw new Error("Delete requires rkey");
512 const deleteOp: RecordDeleteOp = {
513 action: WriteOpAction.Delete,
514 collection: asRepoCollection(write.collection),
515 rkey: write.rkey,
516 };
517 ops.push(deleteOp);
518 results.push({
519 $type: "com.atproto.repo.applyWrites#deleteResult",
520 });
521 } else if (write.$type === "com.atproto.repo.applyWrites#update") {
522 if (!write.rkey) throw new Error("Update requires rkey");
523 const updateOp: RecordUpdateOp = {
524 action: WriteOpAction.Update,
525 collection: asRepoCollection(write.collection),
526 rkey: write.rkey,
527 record: asRepoRecord(write.record),
528 };
529 ops.push(updateOp);
530 this.storage!.addCollection(write.collection);
531 results.push({
532 $type: "com.atproto.repo.applyWrites#updateResult",
533 uri: `at://${repo.did}/${write.collection}/${write.rkey}`,
534 cid: "",
535 });
536 }
537 }
538
539 const updatedRepo = await repo.applyWrites(ops, keypair);
540 this.repo = updatedRepo;
541
542 // Build firehose ops from the already-constructed ops array
543 const firehoseOps: Array<{
544 action: "create" | "update" | "delete";
545 path: string;
546 cid: string | null;
547 }> = [];
548 for (const op of ops) {
549 if (op.action === WriteOpAction.Create) {
550 const createOp = op as RecordCreateOp;
551 const recordCid = await this.repo.data.get(
552 `${createOp.collection}/${createOp.rkey}`,
553 );
554 firehoseOps.push({
555 action: "create",
556 path: `${createOp.collection}/${createOp.rkey}`,
557 cid: recordCid?.toString() ?? null,
558 });
559 } else if (op.action === WriteOpAction.Delete) {
560 const deleteOp = op as RecordDeleteOp;
561 firehoseOps.push({
562 action: "delete",
563 path: `${deleteOp.collection}/${deleteOp.rkey}`,
564 cid: null,
565 });
566 } else if (op.action === WriteOpAction.Update) {
567 const updateOp = op as RecordUpdateOp;
568 const recordCid = await this.repo.data.get(
569 `${updateOp.collection}/${updateOp.rkey}`,
570 );
571 firehoseOps.push({
572 action: "update",
573 path: `${updateOp.collection}/${updateOp.rkey}`,
574 cid: recordCid?.toString() ?? null,
575 });
576 }
577 }
578 await this.emitCommitEvent(prevDataCid, firehoseOps);
579
580 return {
581 commit: {
582 cid: this.repo.cid.toString(),
583 rev: this.repo.commit.rev,
584 },
585 results,
586 };
587 }
588
589 /** RPC: Permanently remove an account's repo, blobs, and DO storage. */
590 async rpcTakedown(did: string): Promise<TakedownResult> {
591 await this.ensureRepoInitialized();
592
593 const records: Array<{ collection: string; rkey: string }> = [];
594 const collectionSet = new Set<string>();
595 if (this.repo) {
596 for await (const record of this.repo.walkRecords()) {
597 records.push({ collection: record.collection, rkey: record.rkey });
598 collectionSet.add(record.collection);
599 }
600 }
601
602 for (let offset = 0; offset < records.length; offset += 200) {
603 const chunk = records.slice(offset, offset + 200);
604 const prevDataCid = this.storage!.getState()?.prev_data_cid ?? null;
605 const deleteOps: RecordDeleteOp[] = chunk.map((record) => ({
606 action: WriteOpAction.Delete,
607 collection: asRepoCollection(record.collection),
608 rkey: record.rkey,
609 }));
610
611 this.repo = await this.repo!.applyWrites(deleteOps, this.keypair!);
612 await this.emitCommitEvent(
613 prevDataCid,
614 chunk.map((record) => ({
615 action: "delete" as const,
616 path: `${record.collection}/${record.rkey}`,
617 cid: null,
618 })),
619 );
620 }
621
622 const seqId = this.env.SEQUENCER.idFromName("sequencer");
623 const seqStub = this.env.SEQUENCER.get(seqId);
624 await seqStub.sequenceAccount(did, false, "deleted");
625
626 const prefix = `${did}/`;
627 let blobsDeleted = 0;
628 let cursor: string | undefined;
629 do {
630 const listed = await this.env.BLOBS.list({ prefix, cursor });
631 const keys = listed.objects.map((object) => object.key);
632 if (keys.length > 0) {
633 await this.env.BLOBS.delete(keys);
634 blobsDeleted += keys.length;
635 }
636 if (listed.truncated && !listed.cursor) {
637 throw new Error("R2 blob listing was truncated without a cursor");
638 }
639 cursor = listed.truncated ? listed.cursor : undefined;
640 } while (cursor);
641
642 await this.ctx.storage.deleteAll();
643 this.storage = null;
644 this.repo = null;
645 this.keypair = null;
646 this.storageInitialized = false;
647 this.repoInitialized = false;
648
649 return {
650 recordsDeleted: records.length,
651 blobsDeleted,
652 collections: Array.from(collectionSet).sort(),
653 };
654 }
655
656 /** RPC: Get repo status */
657 async rpcGetRepoStatus(): Promise<{
658 did: string;
659 active: boolean;
660 status: string;
661 rev: string | null;
662 } | null> {
663 await this.ensureStorageInitialized();
664 const state = this.storage!.getState();
665 if (!state || !state.did) return null;
666 return {
667 did: state.did,
668 active: state.active === 1,
669 status: state.active === 1 ? "active" : "deactivated",
670 rev: state.rev,
671 };
672 }
673
674 /** RPC: Export repo as CAR bytes */
675 async rpcExportRepo(): Promise<Uint8Array> {
676 await this.ensureRepoInitialized();
677 const root = await this.storage!.getRoot();
678 if (!root) throw new Error("Repo has no root");
679
680 const allBlocks = new BlockMap();
681 const rows = this.storage!.getAllBlocks();
682 for (const row of rows) {
683 allBlocks.set(CID.parse(row.cid), new Uint8Array(row.bytes as ArrayBuffer));
684 }
685
686 return blocksToCarFile(root, allBlocks);
687 }
688
689 /** RPC: Upload a blob to R2 and track in metadata */
690 async rpcUploadBlob(bytes: Uint8Array, mimeType: string): Promise<BlobRef> {
691 await this.ensureStorageInitialized();
692 const state = this.storage!.getState();
693 if (!state?.did) throw new Error("Account not provisioned");
694
695 const cidObj = await createCid(CODEC_RAW, bytes);
696 const cidStr = formatCid(cidObj);
697
698 const key = `${state.did}/${cidStr}`;
699 await this.env.BLOBS.put(key, bytes, {
700 httpMetadata: { contentType: mimeType },
701 });
702
703 this.storage!.insertBlob(cidStr, mimeType, bytes.length);
704
705 return {
706 $type: "blob",
707 ref: { $link: cidStr },
708 mimeType,
709 size: bytes.length,
710 };
711 }
712
713 /** RPC: List blob CIDs for this account */
714 async rpcListBlobs(
715 opts?: { limit?: number; cursor?: string },
716 ): Promise<{ cids: string[]; cursor?: string }> {
717 await this.ensureStorageInitialized();
718 return this.storage!.listBlobs(opts);
719 }
720}