open-source, lexicon-agnostic PDS for AI agents. welcome-mat enrollment, AT Proto federation.
agents
atprotocol
pds
cloudflare
Merge branch 'hopper-fdscfzmz-l1-storage-layer'
# Conflicts:
# .gitignore
+605
+3
README.md
+3
README.md
+88
src/cbor-compat.ts
+88
src/cbor-compat.ts
···
1
+
import {
2
+
encode as atcuteEncode,
3
+
decode as atcuteDecode,
4
+
toCIDLink,
5
+
toBytes,
6
+
fromBytes,
7
+
type Bytes,
8
+
type CIDLink,
9
+
} from "@atcute/cbor";
10
+
import { parse } from "@atcute/cid";
11
+
import type { CID } from "@atproto/lex-data";
12
+
13
+
function isAtprotoCid(value: unknown): value is CID {
14
+
if (value === null || typeof value !== "object") {
15
+
return false;
16
+
}
17
+
const obj = value as Record<string | symbol, unknown>;
18
+
return "asCID" in obj && obj[Symbol.toStringTag] === "CID";
19
+
}
20
+
21
+
function isBytes(value: unknown): value is Bytes {
22
+
return value !== null && typeof value === "object" && "$bytes" in value;
23
+
}
24
+
25
+
function atprotoCidToCidLink(cid: CID): CIDLink {
26
+
return toCIDLink(parse(cid.toString()));
27
+
}
28
+
29
+
function convertCidsForEncode(value: unknown): unknown {
30
+
if (value === null || value === undefined) {
31
+
return value;
32
+
}
33
+
if (typeof value !== "object") {
34
+
return value;
35
+
}
36
+
if (ArrayBuffer.isView(value) && value instanceof Uint8Array) {
37
+
return toBytes(value);
38
+
}
39
+
if (isAtprotoCid(value)) {
40
+
return atprotoCidToCidLink(value);
41
+
}
42
+
if (Array.isArray(value)) {
43
+
return (value as unknown[]).map(convertCidsForEncode);
44
+
}
45
+
const obj = value as object;
46
+
if (obj.constructor === Object) {
47
+
const result: Record<string, unknown> = {};
48
+
for (const [key, val] of Object.entries(obj)) {
49
+
result[key] = convertCidsForEncode(val);
50
+
}
51
+
return result;
52
+
}
53
+
return value;
54
+
}
55
+
56
+
export function encode(value: unknown): Uint8Array {
57
+
const converted = convertCidsForEncode(value);
58
+
return atcuteEncode(converted);
59
+
}
60
+
61
+
function convertWrappersForDecode(value: unknown): unknown {
62
+
if (value === null || value === undefined) {
63
+
return value;
64
+
}
65
+
if (typeof value !== "object") {
66
+
return value;
67
+
}
68
+
if (isBytes(value)) {
69
+
return fromBytes(value);
70
+
}
71
+
if (Array.isArray(value)) {
72
+
return (value as unknown[]).map(convertWrappersForDecode);
73
+
}
74
+
const obj = value as object;
75
+
if (obj.constructor === Object) {
76
+
const result: Record<string, unknown> = {};
77
+
for (const [key, val] of Object.entries(obj)) {
78
+
result[key] = convertWrappersForDecode(val);
79
+
}
80
+
return result;
81
+
}
82
+
return value;
83
+
}
84
+
85
+
export function decode(bytes: Uint8Array): unknown {
86
+
const decoded = atcuteDecode(bytes);
87
+
return convertWrappersForDecode(decoded);
88
+
}
+33
src/config.ts
+33
src/config.ts
···
1
+
export interface Config {
2
+
hostname: string;
3
+
handleDomain: string;
4
+
plcUrl: string;
5
+
dbPath: string;
6
+
port: number;
7
+
}
8
+
9
+
export function loadConfig(): Config {
10
+
const missing = [
11
+
"ROOKERY_HOSTNAME",
12
+
"ROOKERY_HANDLE_DOMAIN",
13
+
"ROOKERY_PLC_URL",
14
+
].filter((name) => !process.env[name]);
15
+
16
+
if (missing.length > 0) {
17
+
throw new Error(`Missing required environment variables: ${missing.join(", ")}`);
18
+
}
19
+
20
+
const portRaw = process.env.PORT ?? "3000";
21
+
const port = Number.parseInt(portRaw, 10);
22
+
if (Number.isNaN(port)) {
23
+
throw new Error(`Invalid PORT: ${portRaw}`);
24
+
}
25
+
26
+
return {
27
+
hostname: process.env.ROOKERY_HOSTNAME!,
28
+
handleDomain: process.env.ROOKERY_HANDLE_DOMAIN!,
29
+
plcUrl: process.env.ROOKERY_PLC_URL!,
30
+
dbPath: process.env.ROOKERY_DB_PATH ?? "./rookery.db",
31
+
port,
32
+
};
33
+
}
+61
src/db.ts
+61
src/db.ts
···
1
+
import Database from "better-sqlite3";
2
+
3
+
export function initDatabase(dbPath: string): Database.Database {
4
+
const db = new Database(dbPath);
5
+
db.pragma("journal_mode = WAL");
6
+
7
+
db.exec(`
8
+
CREATE TABLE IF NOT EXISTS accounts (
9
+
id INTEGER PRIMARY KEY AUTOINCREMENT,
10
+
did TEXT UNIQUE NOT NULL,
11
+
handle TEXT,
12
+
jwk_thumbprint TEXT,
13
+
signing_key_hex TEXT,
14
+
signing_key_pub TEXT,
15
+
rotation_key_hex TEXT,
16
+
rotation_key_pub TEXT,
17
+
root_cid TEXT,
18
+
rev TEXT,
19
+
prev_data_cid TEXT,
20
+
active INTEGER DEFAULT 1,
21
+
created_at TEXT DEFAULT (datetime('now'))
22
+
);
23
+
24
+
CREATE TABLE IF NOT EXISTS blocks (
25
+
id INTEGER PRIMARY KEY AUTOINCREMENT,
26
+
account_id INTEGER NOT NULL REFERENCES accounts(id),
27
+
cid TEXT NOT NULL,
28
+
bytes BLOB NOT NULL,
29
+
rev TEXT NOT NULL,
30
+
UNIQUE(account_id, cid)
31
+
);
32
+
CREATE INDEX IF NOT EXISTS idx_blocks_account_rev ON blocks(account_id, rev);
33
+
34
+
CREATE TABLE IF NOT EXISTS collections (
35
+
account_id INTEGER NOT NULL REFERENCES accounts(id),
36
+
collection TEXT NOT NULL,
37
+
PRIMARY KEY (account_id, collection)
38
+
);
39
+
40
+
CREATE TABLE IF NOT EXISTS blobs (
41
+
id INTEGER PRIMARY KEY AUTOINCREMENT,
42
+
account_id INTEGER NOT NULL REFERENCES accounts(id),
43
+
cid TEXT NOT NULL,
44
+
mime_type TEXT,
45
+
size INTEGER,
46
+
created_at TEXT DEFAULT (datetime('now')),
47
+
UNIQUE(account_id, cid)
48
+
);
49
+
50
+
CREATE TABLE IF NOT EXISTS firehose_events (
51
+
seq INTEGER PRIMARY KEY AUTOINCREMENT,
52
+
did TEXT NOT NULL,
53
+
event_type TEXT NOT NULL,
54
+
payload BLOB NOT NULL,
55
+
created_at TEXT DEFAULT (datetime('now'))
56
+
);
57
+
CREATE INDEX IF NOT EXISTS idx_firehose_did ON firehose_events(did);
58
+
`);
59
+
60
+
return db;
61
+
}
+15
src/index.ts
+15
src/index.ts
···
1
+
import { Hono } from "hono";
2
+
import { serve } from "@hono/node-server";
3
+
import { loadConfig } from "./config.js";
4
+
import { initDatabase } from "./db.js";
5
+
6
+
const config = loadConfig();
7
+
const db = initDatabase(config.dbPath);
8
+
9
+
const app = new Hono();
10
+
11
+
app.get("/", (c) => c.json({ status: "ok" }));
12
+
13
+
serve({ fetch: app.fetch, port: config.port }, (info) => {
14
+
console.log(`rookery listening on port ${info.port}`);
15
+
});
+141
src/storage.ts
+141
src/storage.ts
···
1
+
import { CID } from "@atproto/lex-data";
2
+
import {
3
+
BlockMap,
4
+
ReadableBlockstore,
5
+
cborToLex,
6
+
type CommitData,
7
+
type RepoStorage,
8
+
} from "@atproto/repo";
9
+
import type Database from "better-sqlite3";
10
+
11
+
export 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
+
}
+245
test/storage.test.ts
+245
test/storage.test.ts
···
1
+
import { beforeEach, describe, expect, it } from "vitest";
2
+
import Database from "better-sqlite3";
3
+
import { CID } from "@atproto/lex-data";
4
+
import { BlockMap, Repo, WriteOpAction } from "@atproto/repo";
5
+
import { Secp256k1Keypair } from "@atproto/crypto";
6
+
import { initDatabase } from "../src/db.js";
7
+
import { SqliteRepoStorage } from "../src/storage.js";
8
+
9
+
function createAccount(
10
+
db: Database.Database,
11
+
did: string,
12
+
): { id: number; storage: SqliteRepoStorage } {
13
+
const info = db.prepare("INSERT INTO accounts (did) VALUES (?)").run(did);
14
+
const id = Number(info.lastInsertRowid);
15
+
return { id, storage: new SqliteRepoStorage(db, id) };
16
+
}
17
+
18
+
describe("SqliteRepoStorage", () => {
19
+
let db: Database.Database;
20
+
21
+
beforeEach(() => {
22
+
db = initDatabase(":memory:");
23
+
});
24
+
25
+
it("getRoot returns null before any commit", async () => {
26
+
const { storage } = createAccount(db, "did:plc:test123");
27
+
28
+
await expect(storage.getRoot()).resolves.toBeNull();
29
+
});
30
+
31
+
it("Repo.create() succeeds and persists root/rev", async () => {
32
+
const did = "did:plc:test123";
33
+
const { id, storage } = createAccount(db, did);
34
+
const keypair = await Secp256k1Keypair.create();
35
+
36
+
const repo = await Repo.create(storage, did, keypair);
37
+
const root = await storage.getRoot();
38
+
const rev = await storage.getRev();
39
+
const row = db
40
+
.prepare("SELECT root_cid, rev FROM accounts WHERE id = ?")
41
+
.get(id) as { root_cid: string | null; rev: string | null };
42
+
43
+
expect(repo.cid.toString()).toBe(root?.toString());
44
+
expect(root).not.toBeNull();
45
+
expect(rev).not.toBeNull();
46
+
expect(row.root_cid).toBe(root?.toString());
47
+
expect(row.rev).toBe(rev);
48
+
});
49
+
50
+
it("repo.applyWrites() persists blocks and updates root", async () => {
51
+
const did = "did:plc:test123";
52
+
const { id, storage } = createAccount(db, did);
53
+
const keypair = await Secp256k1Keypair.create();
54
+
const repo = await Repo.create(storage, did, keypair);
55
+
const initialRoot = await storage.getRoot();
56
+
const initialRev = await storage.getRev();
57
+
58
+
const updated = await repo.applyWrites(
59
+
{
60
+
action: WriteOpAction.Create,
61
+
collection: "app.bsky.feed.post",
62
+
rkey: "test1",
63
+
record: {
64
+
text: "hello world",
65
+
createdAt: new Date().toISOString(),
66
+
$type: "app.bsky.feed.post",
67
+
},
68
+
},
69
+
keypair,
70
+
);
71
+
72
+
const nextRoot = await storage.getRoot();
73
+
const nextRev = await storage.getRev();
74
+
const blockCount = db
75
+
.prepare("SELECT COUNT(*) AS count FROM blocks WHERE account_id = ?")
76
+
.get(id) as { count: number };
77
+
78
+
expect(initialRoot?.toString()).not.toBe(nextRoot?.toString());
79
+
expect(initialRev).not.toBe(nextRev);
80
+
expect(nextRoot?.toString()).toBe(updated.cid.toString());
81
+
expect(blockCount.count).toBeGreaterThan(0);
82
+
});
83
+
84
+
it("prev_data_cid is null before first commit, set after", async () => {
85
+
const did = "did:plc:test123";
86
+
const { id, storage } = createAccount(db, did);
87
+
const before = db
88
+
.prepare("SELECT prev_data_cid FROM accounts WHERE id = ?")
89
+
.get(id) as { prev_data_cid: string | null };
90
+
const keypair = await Secp256k1Keypair.create();
91
+
92
+
const repo = await Repo.create(storage, did, keypair);
93
+
const afterCreate = db
94
+
.prepare("SELECT prev_data_cid FROM accounts WHERE id = ?")
95
+
.get(id) as { prev_data_cid: string | null };
96
+
97
+
const updated = await repo.applyWrites(
98
+
{
99
+
action: WriteOpAction.Create,
100
+
collection: "app.bsky.feed.post",
101
+
rkey: "test1",
102
+
record: {
103
+
text: "hello world",
104
+
createdAt: new Date().toISOString(),
105
+
$type: "app.bsky.feed.post",
106
+
},
107
+
},
108
+
keypair,
109
+
);
110
+
111
+
const afterWrite = db
112
+
.prepare("SELECT prev_data_cid FROM accounts WHERE id = ?")
113
+
.get(id) as { prev_data_cid: string | null };
114
+
115
+
expect(before.prev_data_cid).toBeNull();
116
+
expect(afterCreate.prev_data_cid).not.toBeNull();
117
+
expect(afterCreate.prev_data_cid).toBe(repo.data.pointer.toString());
118
+
expect(afterWrite.prev_data_cid).not.toBeNull();
119
+
expect(afterWrite.prev_data_cid).toBe(updated.data.pointer.toString());
120
+
expect(afterWrite.prev_data_cid).not.toBe(afterCreate.prev_data_cid);
121
+
});
122
+
123
+
it("multi-tenant isolation", async () => {
124
+
const alice = createAccount(db, "did:plc:alice");
125
+
const bob = createAccount(db, "did:plc:bob");
126
+
const keypair = await Secp256k1Keypair.create();
127
+
const repo = await Repo.create(alice.storage, "did:plc:alice", keypair);
128
+
const updated = await repo.applyWrites(
129
+
{
130
+
action: WriteOpAction.Create,
131
+
collection: "app.bsky.feed.post",
132
+
rkey: "test1",
133
+
record: {
134
+
text: "hello world",
135
+
createdAt: new Date().toISOString(),
136
+
$type: "app.bsky.feed.post",
137
+
},
138
+
},
139
+
keypair,
140
+
);
141
+
const aliceBlocks = db
142
+
.prepare("SELECT cid FROM blocks WHERE account_id = ? ORDER BY cid")
143
+
.all(alice.id) as { cid: string }[];
144
+
145
+
expect(await bob.storage.getRoot()).toBeNull();
146
+
expect(await bob.storage.has(updated.cid)).toBe(false);
147
+
expect(await bob.storage.has(updated.data.pointer)).toBe(false);
148
+
expect(aliceBlocks.length).toBeGreaterThan(0);
149
+
for (const row of aliceBlocks) {
150
+
expect(await bob.storage.getBytes(CID.parse(row.cid))).toBeNull();
151
+
}
152
+
});
153
+
154
+
it("addCollection and getCollections", () => {
155
+
const alice = createAccount(db, "did:plc:alice");
156
+
const bob = createAccount(db, "did:plc:bob");
157
+
158
+
alice.storage.addCollection("app.bsky.graph.follow");
159
+
alice.storage.addCollection("app.bsky.feed.post");
160
+
alice.storage.addCollection("app.bsky.feed.post");
161
+
162
+
expect(alice.storage.getCollections()).toEqual([
163
+
"app.bsky.feed.post",
164
+
"app.bsky.graph.follow",
165
+
]);
166
+
expect(bob.storage.getCollections()).toEqual([]);
167
+
});
168
+
169
+
it("getRev returns rev after commit", async () => {
170
+
const did = "did:plc:test123";
171
+
const { id, storage } = createAccount(db, did);
172
+
const keypair = await Secp256k1Keypair.create();
173
+
174
+
await Repo.create(storage, did, keypair);
175
+
176
+
const rev = await storage.getRev();
177
+
const row = db
178
+
.prepare("SELECT rev FROM accounts WHERE id = ?")
179
+
.get(id) as { rev: string | null };
180
+
181
+
expect(rev).not.toBeNull();
182
+
expect(rev).toBe(row.rev);
183
+
});
184
+
185
+
it("applyCommit is transactional", async () => {
186
+
const did = "did:plc:test123";
187
+
const { id, storage } = createAccount(db, did);
188
+
const keypair = await Secp256k1Keypair.create();
189
+
const repo = await Repo.create(storage, did, keypair);
190
+
const goodCommit = await repo.formatCommit(
191
+
{
192
+
action: WriteOpAction.Create,
193
+
collection: "app.bsky.feed.post",
194
+
rkey: "test1",
195
+
record: {
196
+
text: "hello world",
197
+
createdAt: new Date().toISOString(),
198
+
$type: "app.bsky.feed.post",
199
+
},
200
+
},
201
+
keypair,
202
+
);
203
+
204
+
const originalRootRow = db
205
+
.prepare("SELECT root_cid, rev, prev_data_cid FROM accounts WHERE id = ?")
206
+
.get(id) as {
207
+
root_cid: string | null;
208
+
rev: string | null;
209
+
prev_data_cid: string | null;
210
+
};
211
+
const originalBlockCount = (
212
+
db.prepare("SELECT COUNT(*) AS count FROM blocks WHERE account_id = ?").get(id) as {
213
+
count: number;
214
+
}
215
+
).count;
216
+
217
+
const corruptedBlocks = new BlockMap();
218
+
for (const [cid, bytes] of goodCommit.newBlocks) {
219
+
corruptedBlocks.set(cid, cid.equals(goodCommit.cid) ? Buffer.from([0xff]) : bytes);
220
+
}
221
+
222
+
await expect(
223
+
storage.applyCommit({
224
+
...goodCommit,
225
+
newBlocks: corruptedBlocks,
226
+
}),
227
+
).rejects.toThrow();
228
+
229
+
const finalRootRow = db
230
+
.prepare("SELECT root_cid, rev, prev_data_cid FROM accounts WHERE id = ?")
231
+
.get(id) as {
232
+
root_cid: string | null;
233
+
rev: string | null;
234
+
prev_data_cid: string | null;
235
+
};
236
+
const finalBlockCount = (
237
+
db.prepare("SELECT COUNT(*) AS count FROM blocks WHERE account_id = ?").get(id) as {
238
+
count: number;
239
+
}
240
+
).count;
241
+
242
+
expect(finalRootRow).toEqual(originalRootRow);
243
+
expect(finalBlockCount).toBe(originalBlockCount);
244
+
});
245
+
});
+16
tsconfig.json
+16
tsconfig.json
···
1
+
{
2
+
"compilerOptions": {
3
+
"target": "es2022",
4
+
"module": "nodenext",
5
+
"moduleResolution": "nodenext",
6
+
"outDir": "dist",
7
+
"rootDir": "src",
8
+
"strict": true,
9
+
"skipLibCheck": true,
10
+
"declaration": true,
11
+
"esModuleInterop": true,
12
+
"forceConsistentCasingInFileNames": true
13
+
},
14
+
"include": ["src"],
15
+
"exclude": ["node_modules", "dist", "test"]
16
+
}