solpbc.org
open-source, lexicon-agnostic PDS for AI agents. welcome-mat enrollment, AT Proto federation.
agents
atprotocol
pds
cloudflare
1import { CID } from "@atproto/lex-data";
2import {
3 BlockMap,
4 ReadableBlockstore,
5 cborToLex,
6 type CommitData,
7 type RepoStorage,
8} from "@atproto/repo";
9import type Database from "better-sqlite3";
10
11export class SqliteRepoStorage
12 extends ReadableBlockstore
13 implements RepoStorage
14{
15 constructor(
16 private db: Database.Database,
17 private accountId: number,
18 ) {
19 super();
20 }
21
22 async getBytes(cid: CID): Promise<Uint8Array | null> {
23 const row = this.db
24 .prepare("SELECT bytes FROM blocks WHERE account_id = ? AND cid = ?")
25 .get(this.accountId, cid.toString()) as { bytes: Buffer } | undefined;
26 return row?.bytes ?? null;
27 }
28
29 async has(cid: CID): Promise<boolean> {
30 const row = this.db
31 .prepare(
32 "SELECT 1 FROM blocks WHERE account_id = ? AND cid = ? LIMIT 1",
33 )
34 .get(this.accountId, cid.toString());
35 return !!row;
36 }
37
38 async getBlocks(cids: CID[]): Promise<{ blocks: BlockMap; missing: CID[] }> {
39 const blocks = new BlockMap();
40 const missing: CID[] = [];
41 for (const cid of cids) {
42 const bytes = await this.getBytes(cid);
43 if (bytes) {
44 blocks.set(cid, bytes);
45 } else {
46 missing.push(cid);
47 }
48 }
49 return { blocks, missing };
50 }
51
52 async getRoot(): Promise<CID | null> {
53 const row = this.db
54 .prepare("SELECT root_cid FROM accounts WHERE id = ?")
55 .get(this.accountId) as { root_cid: string | null } | undefined;
56 if (!row?.root_cid) return null;
57 return CID.parse(row.root_cid);
58 }
59
60 async getRev(): Promise<string | null> {
61 const row = this.db
62 .prepare("SELECT rev FROM accounts WHERE id = ?")
63 .get(this.accountId) as { rev: string | null } | undefined;
64 return row?.rev ?? null;
65 }
66
67 async putBlock(cid: CID, block: Uint8Array, rev: string): Promise<void> {
68 this.db
69 .prepare(
70 "INSERT OR REPLACE INTO blocks (account_id, cid, bytes, rev) VALUES (?, ?, ?, ?)",
71 )
72 .run(this.accountId, cid.toString(), block, rev);
73 }
74
75 async putMany(blocks: BlockMap, rev: string): Promise<void> {
76 const stmt = this.db.prepare(
77 "INSERT OR REPLACE INTO blocks (account_id, cid, bytes, rev) VALUES (?, ?, ?, ?)",
78 );
79 for (const [cid, bytes] of blocks) {
80 stmt.run(this.accountId, cid.toString(), bytes, rev);
81 }
82 }
83
84 async updateRoot(cid: CID, rev: string): Promise<void> {
85 this.db
86 .prepare("UPDATE accounts SET root_cid = ?, rev = ? WHERE id = ?")
87 .run(cid.toString(), rev, this.accountId);
88 }
89
90 async applyCommit(commit: CommitData): Promise<void> {
91 const doCommit = this.db.transaction(() => {
92 const insertBlock = this.db.prepare(
93 "INSERT OR REPLACE INTO blocks (account_id, cid, bytes, rev) VALUES (?, ?, ?, ?)",
94 );
95 for (const [cid, bytes] of commit.newBlocks) {
96 insertBlock.run(this.accountId, cid.toString(), bytes, commit.rev);
97 }
98
99 const deleteBlock = this.db.prepare(
100 "DELETE FROM blocks WHERE account_id = ? AND cid = ?",
101 );
102 for (const cid of commit.removedCids) {
103 deleteBlock.run(this.accountId, cid.toString());
104 }
105
106 this.db
107 .prepare(
108 "UPDATE accounts SET root_cid = ?, rev = ? WHERE id = ?",
109 )
110 .run(commit.cid.toString(), commit.rev, this.accountId);
111
112 // Extract MST data root CID from the commit block for prev_data_cid
113 const commitBytes = commit.newBlocks.get(commit.cid);
114 if (!commitBytes) {
115 throw new Error(`Missing commit block for CID ${commit.cid.toString()}`);
116 }
117 const commitObj = cborToLex(commitBytes) as { data: CID };
118 this.db
119 .prepare("UPDATE accounts SET prev_data_cid = ? WHERE id = ?")
120 .run(commitObj.data.toString(), this.accountId);
121 });
122 doCommit();
123 }
124
125 addCollection(collection: string): void {
126 this.db
127 .prepare(
128 "INSERT OR IGNORE INTO collections (account_id, collection) VALUES (?, ?)",
129 )
130 .run(this.accountId, collection);
131 }
132
133 getCollections(): string[] {
134 const rows = this.db
135 .prepare(
136 "SELECT collection FROM collections WHERE account_id = ? ORDER BY collection",
137 )
138 .all(this.accountId) as { collection: string }[];
139 return rows.map((r) => r.collection);
140 }
141}