zzstoatzz.io
zig pds
Add SQLite-backed PDS record store
author
zzstoatzz
date
(May 20, 2026, 12:06 AM -0500)
commit
f80c18e8
f80c18e80ca1513554200b889f632c2a4c410750
parent
1dd86740
1dd8674069978533c8ee1a35d2c568213ca07c6d
+729
-91
+3
.gitignore
+3
.gitignore
+8
-1
build.zig
+8
-1
build.zig
···
8
8
.target = target,
9
9
.optimize = optimize,
10
10
});
11
+
const zqlite = b.dependency("zqlite", .{
12
+
.target = target,
13
+
.optimize = optimize,
14
+
});
11
15
12
16
const mod = b.addModule("zds", .{
13
17
.root_source_file = b.path("src/root.zig"),
14
18
.target = target,
15
19
.optimize = optimize,
16
-
.imports = &.{.{ .name = "zat", .module = zat.module("zat") }},
20
+
.imports = &.{
21
+
.{ .name = "zat", .module = zat.module("zat") },
22
+
.{ .name = "zqlite", .module = zqlite.module("zqlite") },
23
+
},
17
24
});
18
25
19
26
const exe = b.addExecutable(.{
+4
build.zig.zon
+4
build.zig.zon
···
7
7
.zat = .{
8
8
.path = "../../zat.dev/zat",
9
9
},
10
+
.zqlite = .{
11
+
.url = "git+https://github.com/karlseguin/zqlite.zig?ref=master#05a88d6758753e1c63fdd45b211dde2057094b0c",
12
+
.hash = "zqlite-0.0.1-RWLaYz6bmAAT7E_jxopXf-j5Ea8VQldnxsd6TU8sa0Bb",
13
+
},
10
14
},
11
15
.paths = .{
12
16
"build.zig",
+11
-1
src/main.zig
+11
-1
src/main.zig
···
3
3
4
4
pub fn main(init: std.process.Init) !void {
5
5
var port: u16 = 2583;
6
+
var db_path: []const u8 = "dev/zds.sqlite3";
6
7
7
8
var args = std.process.Args.Iterator.init(init.minimal.args);
8
9
_ = args.next();
···
16
17
port = try std.fmt.parseInt(u16, value, 10);
17
18
continue;
18
19
}
20
+
if (std.mem.eql(u8, arg, "--db")) {
21
+
db_path = args.next() orelse return error.MissingDatabasePath;
22
+
continue;
23
+
}
19
24
if (std.mem.startsWith(u8, arg, "--port=")) {
20
25
port = try std.fmt.parseInt(u16, arg["--port=".len..], 10);
21
26
continue;
22
27
}
28
+
if (std.mem.startsWith(u8, arg, "--db=")) {
29
+
db_path = arg["--db=".len..];
30
+
continue;
31
+
}
23
32
return error.UnknownArgument;
24
33
}
25
34
35
+
try zds.store.init(init.io, db_path);
26
36
try zds.server.listen(init.io, .{ .port = port });
27
37
}
28
38
29
39
fn usage() void {
30
40
std.debug.print(
31
-
\\usage: zds [--port PORT]
41
+
\\usage: zds [--port PORT] [--db PATH]
32
42
\\
33
43
\\Runs a local PDS-shaped HTTP server.
34
44
\\
+368
src/server.zig
+368
src/server.zig
···
25
25
repo_get_record,
26
26
repo_list_records,
27
27
repo_delete_record,
28
+
repo_apply_writes,
29
+
repo_upload_blob,
30
+
identity_resolve_handle,
31
+
actor_get_profile,
32
+
actor_get_profiles,
33
+
actor_get_preferences,
34
+
actor_put_preferences,
35
+
graph_get_follows,
36
+
feed_get_timeline,
37
+
feed_get_feed,
38
+
feed_get_author_feed,
39
+
draft_get_drafts,
40
+
age_assurance_get_state,
41
+
notification_list,
42
+
unspecced_get_config,
43
+
labeler_get_services,
44
+
chat_get_log,
45
+
chat_list_convos,
28
46
not_found,
29
47
};
30
48
···
61
79
if (method == .POST and std.mem.eql(u8, path, "/xrpc/com.atproto.repo.deleteRecord")) {
62
80
return .repo_delete_record;
63
81
}
82
+
if (method == .POST and std.mem.eql(u8, path, "/xrpc/com.atproto.repo.applyWrites")) {
83
+
return .repo_apply_writes;
84
+
}
85
+
if (method == .POST and std.mem.eql(u8, path, "/xrpc/com.atproto.repo.uploadBlob")) {
86
+
return .repo_upload_blob;
87
+
}
88
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/com.atproto.identity.resolveHandle")) {
89
+
return .identity_resolve_handle;
90
+
}
91
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.actor.getProfile")) {
92
+
return .actor_get_profile;
93
+
}
94
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.actor.getProfiles")) {
95
+
return .actor_get_profiles;
96
+
}
97
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.actor.getPreferences")) {
98
+
return .actor_get_preferences;
99
+
}
100
+
if (method == .POST and std.mem.eql(u8, path, "/xrpc/app.bsky.actor.putPreferences")) {
101
+
return .actor_put_preferences;
102
+
}
103
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.graph.getFollows")) {
104
+
return .graph_get_follows;
105
+
}
106
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.feed.getTimeline")) {
107
+
return .feed_get_timeline;
108
+
}
109
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.feed.getFeed")) {
110
+
return .feed_get_feed;
111
+
}
112
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.feed.getAuthorFeed")) {
113
+
return .feed_get_author_feed;
114
+
}
115
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.draft.getDrafts")) {
116
+
return .draft_get_drafts;
117
+
}
118
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.ageassurance.getState")) {
119
+
return .age_assurance_get_state;
120
+
}
121
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.notification.listNotifications")) {
122
+
return .notification_list;
123
+
}
124
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.unspecced.getConfig")) {
125
+
return .unspecced_get_config;
126
+
}
127
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.labeler.getServices")) {
128
+
return .labeler_get_services;
129
+
}
130
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/chat.bsky.convo.getLog")) {
131
+
return .chat_get_log;
132
+
}
133
+
if (method == .GET and std.mem.eql(u8, path, "/xrpc/chat.bsky.convo.listConvos")) {
134
+
return .chat_list_convos;
135
+
}
64
136
return .not_found;
65
137
}
66
138
···
122
194
.repo_get_record => try repoGetRecord(request),
123
195
.repo_list_records => try repoListRecords(request),
124
196
.repo_delete_record => try repoDeleteRecord(request),
197
+
.repo_apply_writes => try repoApplyWrites(request),
198
+
.repo_upload_blob => try repoUploadBlob(request),
199
+
.identity_resolve_handle => try identityResolveHandle(request),
200
+
.actor_get_profile => try actorGetProfile(request),
201
+
.actor_get_profiles => try actorGetProfiles(request),
202
+
.actor_get_preferences => try actorGetPreferences(request),
203
+
.actor_put_preferences => try actorPutPreferences(request),
204
+
.graph_get_follows => try authenticatedEmpty(request, "{\"follows\":[]}"),
205
+
.feed_get_timeline => try authenticatedEmpty(request, "{\"feed\":[]}"),
206
+
.feed_get_feed => try authenticatedEmpty(request, "{\"feed\":[]}"),
207
+
.feed_get_author_feed => try feedGetAuthorFeed(request),
208
+
.draft_get_drafts => try authenticatedEmpty(request, "{\"drafts\":[]}"),
209
+
.age_assurance_get_state => try ageAssuranceGetState(request),
210
+
.notification_list => try authenticatedEmpty(request, "{\"notifications\":[],\"seenAt\":\"2026-05-20T00:00:00.000Z\"}"),
211
+
.unspecced_get_config => try json(request, .ok, "{\"checkEmailConfirmed\":false,\"liveNow\":{},\"featureFlags\":{}}"),
212
+
.labeler_get_services => try json(request, .ok, "{\"views\":[]}"),
213
+
.chat_get_log => try authenticatedEmpty(request, "{\"logs\":[],\"cursor\":\"\"}"),
214
+
.chat_list_convos => try authenticatedEmpty(request, "{\"convos\":[]}"),
125
215
.not_found => try xrpcError(request, .not_found, "MethodNotImplemented", "Method not implemented"),
126
216
}
127
217
}
···
293
383
};
294
384
295
385
const record = store.get(account.did, collection, rkey) orelse {
386
+
if (std.mem.eql(u8, collection, "chat.bsky.actor.declaration") and std.mem.eql(u8, rkey, "self")) {
387
+
var body_buf: [512]u8 = undefined;
388
+
const uri = try std.fmt.bufPrint(
389
+
body_buf[0..160],
390
+
"at://{s}/chat.bsky.actor.declaration/self",
391
+
.{account.did},
392
+
);
393
+
const body = try std.fmt.bufPrint(
394
+
body_buf[160..],
395
+
"{{\"uri\":{f},\"cid\":\"bafyreie5cvv4h45feadgeuwhbcutmh6t2ceseocckahdoe6uat64zmz454\",\"value\":{{\"$type\":\"chat.bsky.actor.declaration\",\"allowIncoming\":\"all\"}}}}",
396
+
.{std.json.fmt(uri, .{})},
397
+
);
398
+
return json(request, .ok, body);
399
+
}
296
400
return xrpcError(request, .not_found, "RecordNotFound", "Record not found");
297
401
};
298
402
const body = try store.writeRecordJson(allocator, record);
···
348
452
return json(request, .ok, "{}");
349
453
}
350
454
455
+
fn repoApplyWrites(request: *http.Server.Request) !void {
456
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
457
+
defer arena.deinit();
458
+
const allocator = arena.allocator();
459
+
460
+
const account = requireBearerAccount(request, allocator) catch |err| switch (err) {
461
+
error.AuthRequired => return xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"),
462
+
error.InvalidToken => return xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"),
463
+
};
464
+
465
+
var body_buf: [131072]u8 = undefined;
466
+
const body = try readBody(request, &body_buf);
467
+
const parsed = try parseJsonBody(request, allocator, body);
468
+
const writes = switch (parsed.value) {
469
+
.object => |object| object.get("writes") orelse return xrpcError(request, .bad_request, "InvalidRequest", "Missing writes"),
470
+
else => return xrpcError(request, .bad_request, "InvalidRequest", "Expected object"),
471
+
};
472
+
if (writes != .array) return xrpcError(request, .bad_request, "InvalidRequest", "writes must be an array");
473
+
474
+
var out: std.Io.Writer.Allocating = .init(allocator);
475
+
defer out.deinit();
476
+
try out.writer.writeAll("{\"commit\":{\"cid\":\"bafyreie5cvv4h45feadgeuwhbcutmh6t2ceseocckahdoe6uat64zmz454\",\"rev\":\"zds-dev\"},\"results\":[");
477
+
for (writes.array.items, 0..) |write, idx| {
478
+
if (write != .object) return xrpcError(request, .bad_request, "InvalidRequest", "Invalid write");
479
+
const type_name = valueString(write, "$type") orelse {
480
+
return xrpcError(request, .bad_request, "InvalidRequest", "Missing write $type");
481
+
};
482
+
const collection = valueString(write, "collection") orelse {
483
+
return xrpcError(request, .bad_request, "InvalidRequest", "Missing collection");
484
+
};
485
+
if (idx != 0) try out.writer.writeByte(',');
486
+
487
+
if (std.mem.eql(u8, type_name, "com.atproto.repo.applyWrites#create")) {
488
+
const record_value = write.object.get("value") orelse {
489
+
return xrpcError(request, .bad_request, "InvalidRequest", "Missing value");
490
+
};
491
+
const record = store.create(allocator, account, collection, valueString(write, "rkey"), record_value) catch |err| switch (err) {
492
+
error.InvalidCollection => return xrpcError(request, .bad_request, "InvalidRequest", "Invalid collection"),
493
+
error.InvalidRecordKey => return xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"),
494
+
else => return err,
495
+
};
496
+
const uri = try record.uri(allocator);
497
+
try out.writer.print(
498
+
"{{\"$type\":\"com.atproto.repo.applyWrites#createResult\",\"uri\":{f},\"cid\":{f},\"validationStatus\":\"unknown\"}}",
499
+
.{ std.json.fmt(uri, .{}), std.json.fmt(record.cid, .{}) },
500
+
);
501
+
} else if (std.mem.eql(u8, type_name, "com.atproto.repo.applyWrites#update")) {
502
+
const rkey = valueString(write, "rkey") orelse {
503
+
return xrpcError(request, .bad_request, "InvalidRequest", "Missing rkey");
504
+
};
505
+
const record_value = write.object.get("value") orelse {
506
+
return xrpcError(request, .bad_request, "InvalidRequest", "Missing value");
507
+
};
508
+
const record = store.put(account, collection, rkey, record_value) catch |err| switch (err) {
509
+
error.InvalidCollection => return xrpcError(request, .bad_request, "InvalidRequest", "Invalid collection"),
510
+
error.InvalidRecordKey => return xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"),
511
+
else => return err,
512
+
};
513
+
const uri = try record.uri(allocator);
514
+
try out.writer.print(
515
+
"{{\"$type\":\"com.atproto.repo.applyWrites#updateResult\",\"uri\":{f},\"cid\":{f},\"validationStatus\":\"unknown\"}}",
516
+
.{ std.json.fmt(uri, .{}), std.json.fmt(record.cid, .{}) },
517
+
);
518
+
} else if (std.mem.eql(u8, type_name, "com.atproto.repo.applyWrites#delete")) {
519
+
const rkey = valueString(write, "rkey") orelse {
520
+
return xrpcError(request, .bad_request, "InvalidRequest", "Missing rkey");
521
+
};
522
+
store.delete(account.did, collection, rkey);
523
+
try out.writer.writeAll("{\"$type\":\"com.atproto.repo.applyWrites#deleteResult\"}");
524
+
} else {
525
+
return xrpcError(request, .bad_request, "InvalidRequest", "Unknown write operation type");
526
+
}
527
+
}
528
+
try out.writer.writeAll("]}");
529
+
return json(request, .ok, out.written());
530
+
}
531
+
532
+
fn repoUploadBlob(request: *http.Server.Request) !void {
533
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
534
+
defer arena.deinit();
535
+
const allocator = arena.allocator();
536
+
537
+
const account = requireBearerAccount(request, allocator) catch |err| switch (err) {
538
+
error.AuthRequired => return xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"),
539
+
error.InvalidToken => return xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"),
540
+
};
541
+
542
+
var body_buf: [2 * 1024 * 1024]u8 = undefined;
543
+
const body = try readBody(request, &body_buf);
544
+
const mime_type = headerValue(request, "content-type") orelse "application/octet-stream";
545
+
const cid = try store.putBlob(allocator, account, body, mime_type);
546
+
const body_out = try std.fmt.allocPrint(
547
+
allocator,
548
+
"{{\"blob\":{{\"$type\":\"blob\",\"ref\":{{\"$link\":{f}}},\"mimeType\":{f},\"size\":{d}}}}}",
549
+
.{ std.json.fmt(cid, .{}), std.json.fmt(mime_type, .{}), body.len },
550
+
);
551
+
return json(request, .ok, body_out);
552
+
}
553
+
554
+
fn identityResolveHandle(request: *http.Server.Request) !void {
555
+
var handle_buf: [256]u8 = undefined;
556
+
const handle = queryParam(request.head.target, "handle", &handle_buf) orelse {
557
+
return xrpcError(request, .bad_request, "InvalidRequest", "Missing handle");
558
+
};
559
+
const did = if (auth.findAccount(handle)) |account|
560
+
account.did
561
+
else if (std.ascii.eqlIgnoreCase(handle, "mod-authority.test"))
562
+
"did:plc:ar7c4by46qjdydhdevvrndac"
563
+
else
564
+
return xrpcError(request, .bad_request, "InvalidRequest", "Handle not found");
565
+
566
+
var body_buf: [320]u8 = undefined;
567
+
const body = try std.fmt.bufPrint(&body_buf, "{{\"did\":{f}}}", .{std.json.fmt(did, .{})});
568
+
return json(request, .ok, body);
569
+
}
570
+
571
+
fn actorGetProfile(request: *http.Server.Request) !void {
572
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
573
+
defer arena.deinit();
574
+
const allocator = arena.allocator();
575
+
576
+
var actor_buf: [256]u8 = undefined;
577
+
const actor = queryParam(request.head.target, "actor", &actor_buf) orelse {
578
+
return xrpcError(request, .bad_request, "InvalidRequest", "Missing actor");
579
+
};
580
+
const account = auth.findAccount(actor) orelse {
581
+
return xrpcError(request, .not_found, "ActorNotFound", "Actor not found");
582
+
};
583
+
584
+
const body = try actorProfileJson(allocator, account);
585
+
return json(request, .ok, body);
586
+
}
587
+
588
+
fn actorGetProfiles(request: *http.Server.Request) !void {
589
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
590
+
defer arena.deinit();
591
+
const allocator = arena.allocator();
592
+
593
+
var actor_buf: [256]u8 = undefined;
594
+
const actor = queryParam(request.head.target, "actors", &actor_buf) orelse {
595
+
return xrpcError(request, .bad_request, "InvalidRequest", "Missing actors");
596
+
};
597
+
const account = auth.findAccount(actor) orelse {
598
+
return xrpcError(request, .not_found, "ActorNotFound", "Actor not found");
599
+
};
600
+
const profile = try actorProfileJson(allocator, account);
601
+
const body = try std.fmt.allocPrint(allocator, "{{\"profiles\":[{s}]}}", .{profile});
602
+
return json(request, .ok, body);
603
+
}
604
+
605
+
fn actorPutPreferences(request: *http.Server.Request) !void {
606
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
607
+
defer arena.deinit();
608
+
const allocator = arena.allocator();
609
+
const account = requireBearerAccount(request, allocator) catch |err| switch (err) {
610
+
error.AuthRequired => return xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"),
611
+
error.InvalidToken => return xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"),
612
+
};
613
+
var body_buf: [65536]u8 = undefined;
614
+
const body = try readBody(request, &body_buf);
615
+
const parsed = try parseJsonBody(request, allocator, body);
616
+
const preferences = switch (parsed.value) {
617
+
.object => |object| object.get("preferences") orelse return xrpcError(request, .bad_request, "InvalidRequest", "Missing preferences"),
618
+
else => return xrpcError(request, .bad_request, "InvalidRequest", "Expected object"),
619
+
};
620
+
try store.setPreferences(account, preferences);
621
+
return json(request, .ok, "{}");
622
+
}
623
+
624
+
fn feedGetAuthorFeed(request: *http.Server.Request) !void {
625
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
626
+
defer arena.deinit();
627
+
const allocator = arena.allocator();
628
+
_ = requireBearerAccount(request, allocator) catch |err| switch (err) {
629
+
error.AuthRequired => return xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"),
630
+
error.InvalidToken => return xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"),
631
+
};
632
+
633
+
var actor_buf: [256]u8 = undefined;
634
+
const actor = queryParam(request.head.target, "actor", &actor_buf) orelse {
635
+
return xrpcError(request, .bad_request, "InvalidRequest", "Missing actor");
636
+
};
637
+
const account = auth.findAccount(actor) orelse {
638
+
return xrpcError(request, .not_found, "ActorNotFound", "Actor not found");
639
+
};
640
+
var limit_buf: [16]u8 = undefined;
641
+
const limit = if (queryParam(request.head.target, "limit", &limit_buf)) |raw|
642
+
std.fmt.parseInt(usize, raw, 10) catch 50
643
+
else
644
+
50;
645
+
const body = try store.writeAuthorFeedJson(allocator, account.did, account.handle, @min(limit, 100));
646
+
return json(request, .ok, body);
647
+
}
648
+
649
+
fn actorGetPreferences(request: *http.Server.Request) !void {
650
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
651
+
defer arena.deinit();
652
+
const allocator = arena.allocator();
653
+
const account = requireBearerAccount(request, allocator) catch |err| switch (err) {
654
+
error.AuthRequired => return xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"),
655
+
error.InvalidToken => return xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"),
656
+
};
657
+
const body = try std.fmt.allocPrint(allocator, "{{\"preferences\":{s}}}", .{store.getPreferences(account)});
658
+
return json(request, .ok, body);
659
+
}
660
+
661
+
fn ageAssuranceGetState(request: *http.Server.Request) !void {
662
+
return authenticatedEmpty(
663
+
request,
664
+
"{\"state\":{\"status\":\"assured\",\"access\":\"full\",\"lastInitiatedAt\":null},\"metadata\":{\"accountCreatedAt\":null}}",
665
+
);
666
+
}
667
+
668
+
fn authenticatedEmpty(request: *http.Server.Request, body: []const u8) !void {
669
+
var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
670
+
defer arena.deinit();
671
+
_ = requireBearerAccount(request, arena.allocator()) catch |err| switch (err) {
672
+
error.AuthRequired => return xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"),
673
+
error.InvalidToken => return xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"),
674
+
};
675
+
return json(request, .ok, body);
676
+
}
677
+
351
678
fn requireBearerAccount(request: *const http.Server.Request, allocator: std.mem.Allocator) !auth.Account {
352
679
const auth_header = headerValue(request, "authorization") orelse {
353
680
return error.AuthRequired;
···
374
701
};
375
702
}
376
703
704
+
fn valueString(value: std.json.Value, key: []const u8) ?[]const u8 {
705
+
return switch (value) {
706
+
.object => |object| switch (object.get(key) orelse return null) {
707
+
.string => |string| string,
708
+
else => null,
709
+
},
710
+
else => null,
711
+
};
712
+
}
713
+
377
714
fn writeRecordRef(request: *http.Server.Request, allocator: std.mem.Allocator, record: store.Record) !void {
378
715
const uri = try record.uri(allocator);
379
716
const body_out = try std.fmt.allocPrint(
···
382
719
.{ std.json.fmt(uri, .{}), std.json.fmt(record.cid, .{}) },
383
720
);
384
721
return json(request, .ok, body_out);
722
+
}
723
+
724
+
fn actorProfileJson(allocator: std.mem.Allocator, account: auth.Account) ![]const u8 {
725
+
const display_name, const description = profileText(allocator, account) catch .{ account.handle, "" };
726
+
const posts_count = store.count(account.did, "app.bsky.feed.post");
727
+
const follows_count = store.count(account.did, "app.bsky.graph.follow");
728
+
const followers_count = store.countSubject("app.bsky.graph.follow", account.did);
729
+
return std.fmt.allocPrint(
730
+
allocator,
731
+
"{{\"did\":{f},\"handle\":{f},\"displayName\":{f},\"description\":{f},\"postsCount\":{d},\"followersCount\":{d},\"followsCount\":{d},\"viewer\":{{\"muted\":false,\"blockedBy\":false}}}}",
732
+
.{
733
+
std.json.fmt(account.did, .{}),
734
+
std.json.fmt(account.handle, .{}),
735
+
std.json.fmt(display_name, .{}),
736
+
std.json.fmt(description, .{}),
737
+
posts_count,
738
+
followers_count,
739
+
follows_count,
740
+
},
741
+
);
742
+
}
743
+
744
+
fn profileText(allocator: std.mem.Allocator, account: auth.Account) !struct { []const u8, []const u8 } {
745
+
const profile = store.get(account.did, "app.bsky.actor.profile", "self") orelse return .{ account.handle, "" };
746
+
const parsed = try std.json.parseFromSlice(std.json.Value, allocator, profile.value_json, .{});
747
+
const display_name = zat.json.getString(parsed.value, "displayName") orelse account.handle;
748
+
const description = zat.json.getString(parsed.value, "description") orelse "";
749
+
return .{ display_name, description };
385
750
}
386
751
387
752
fn readBody(request: *http.Server.Request, buf: []u8) ![]const u8 {
···
494
859
try std.testing.expectEqual(Route.get_service_auth, route(.GET, "/xrpc/com.atproto.server.getServiceAuth?aud=did%3Aplc%3Aservice"));
495
860
try std.testing.expectEqual(Route.repo_list_records, route(.GET, "/xrpc/com.atproto.repo.listRecords?repo=alice.test&collection=app.bsky.feed.post"));
496
861
try std.testing.expectEqual(Route.repo_create_record, route(.POST, "/xrpc/com.atproto.repo.createRecord"));
862
+
try std.testing.expectEqual(Route.repo_upload_blob, route(.POST, "/xrpc/com.atproto.repo.uploadBlob"));
863
+
try std.testing.expectEqual(Route.age_assurance_get_state, route(.GET, "/xrpc/app.bsky.ageassurance.getState?countryCode=US"));
864
+
try std.testing.expectEqual(Route.identity_resolve_handle, route(.GET, "/xrpc/com.atproto.identity.resolveHandle?handle=alice.test"));
497
865
}
498
866
499
867
test "does not route wrong methods" {
+335
-89
src/store.zig
+335
-89
src/store.zig
···
1
1
const std = @import("std");
2
2
const auth = @import("auth.zig");
3
3
const zat = @import("zat");
4
-
5
-
const max_records = 4096;
4
+
const zqlite = @import("zqlite");
5
+
const Io = std.Io;
6
6
7
7
pub const Error = error{
8
8
InvalidCollection,
9
9
InvalidRecordKey,
10
10
MissingRecord,
11
-
StoreFull,
11
+
StoreNotInitialized,
12
12
};
13
13
14
14
pub const Record = struct {
15
-
active: bool = false,
16
-
did: []const u8 = "",
17
-
collection: []const u8 = "",
18
-
rkey: []const u8 = "",
19
-
cid: []const u8 = "",
20
-
value_json: []const u8 = "",
21
-
seq: u64 = 0,
15
+
did: []const u8,
16
+
collection: []const u8,
17
+
rkey: []const u8,
18
+
cid: []const u8,
19
+
value_json: []const u8,
20
+
seq: u64,
22
21
23
22
pub fn uri(self: Record, allocator: std.mem.Allocator) ![]const u8 {
24
23
return std.fmt.allocPrint(allocator, "at://{s}/{s}/{s}", .{ self.did, self.collection, self.rkey });
25
24
}
26
25
};
27
26
28
-
var records: [max_records]Record = [_]Record{.{}} ** max_records;
29
-
var next_seq: u64 = 1;
27
+
var conn: zqlite.Conn = undefined;
28
+
var initialized = false;
29
+
var store_io: Io = undefined;
30
+
var mutex: Io.Mutex = .init;
31
+
32
+
pub fn init(io: Io, path: []const u8) !void {
33
+
if (initialized) return;
34
+
store_io = io;
35
+
if (!std.mem.eql(u8, path, ":memory:")) {
36
+
if (std.fs.path.dirname(path)) |dir| try Io.Dir.createDirPath(.cwd(), io, dir);
37
+
}
38
+
39
+
const path_z = try std.heap.page_allocator.dupeZ(u8, path);
40
+
conn = try zqlite.open(path_z.ptr, zqlite.OpenFlags.Create | zqlite.OpenFlags.ReadWrite);
41
+
initialized = true;
42
+
errdefer close();
43
+
44
+
try conn.busyTimeout(5000);
45
+
try conn.execNoArgs("PRAGMA journal_mode=WAL");
46
+
try conn.execNoArgs("PRAGMA foreign_keys=ON");
47
+
try migrate();
48
+
try seedAccounts();
49
+
}
50
+
51
+
pub fn close() void {
52
+
if (!initialized) return;
53
+
conn.close();
54
+
initialized = false;
55
+
}
30
56
31
57
pub fn resolveRepo(repo: []const u8) ?auth.Account {
32
58
return auth.findAccount(repo);
···
53
79
if (zat.Rkey.parse(rkey) == null) return Error.InvalidRecordKey;
54
80
55
81
const value_json = try stringifyValue(std.heap.page_allocator, value);
56
-
const cid = try std.fmt.allocPrint(std.heap.page_allocator, "bafyreidzds{d}", .{next_seq});
57
82
58
-
if (findIndex(account.did, collection, rkey)) |idx| {
59
-
records[idx] = .{
60
-
.active = true,
61
-
.did = records[idx].did,
62
-
.collection = records[idx].collection,
63
-
.rkey = records[idx].rkey,
64
-
.cid = cid,
65
-
.value_json = value_json,
66
-
.seq = next_seq,
67
-
};
68
-
next_seq += 1;
69
-
return records[idx];
70
-
}
83
+
mutex.lockUncancelable(store_io);
84
+
defer mutex.unlock(store_io);
85
+
try requireInitialized();
86
+
87
+
const seq = try nextSeqLocked();
88
+
const cid = try cidForJson(std.heap.page_allocator, value_json);
89
+
const uri = try std.fmt.allocPrint(std.heap.page_allocator, "at://{s}/{s}/{s}", .{ account.did, collection, rkey });
90
+
const rev = try std.fmt.allocPrint(std.heap.page_allocator, "zds-{d}", .{seq});
91
+
92
+
try conn.exclusiveTransaction();
93
+
errdefer conn.rollback();
94
+
try conn.exec(
95
+
\\INSERT INTO records (did, collection, rkey, uri, cid, value_json, rev, seq)
96
+
\\VALUES (?, ?, ?, ?, ?, ?, ?, ?)
97
+
\\ON CONFLICT(did, collection, rkey) DO UPDATE SET
98
+
\\ uri = excluded.uri,
99
+
\\ cid = excluded.cid,
100
+
\\ value_json = excluded.value_json,
101
+
\\ rev = excluded.rev,
102
+
\\ seq = excluded.seq
103
+
, .{ account.did, collection, rkey, uri, cid, value_json, rev, @as(i64, @intCast(seq)) });
104
+
try conn.exec(
105
+
\\INSERT INTO commits (seq, did, cid, rev, prev)
106
+
\\VALUES (?, ?, ?, ?, (SELECT cid FROM commits WHERE did = ? ORDER BY seq DESC LIMIT 1))
107
+
, .{ @as(i64, @intCast(seq)), account.did, cid, rev, account.did });
108
+
try conn.commit();
71
109
72
-
const idx = freeIndex() orelse return Error.StoreFull;
73
-
records[idx] = .{
74
-
.active = true,
110
+
return .{
75
111
.did = try std.heap.page_allocator.dupe(u8, account.did),
76
112
.collection = try std.heap.page_allocator.dupe(u8, collection),
77
113
.rkey = try std.heap.page_allocator.dupe(u8, rkey),
78
114
.cid = cid,
79
115
.value_json = value_json,
80
-
.seq = next_seq,
116
+
.seq = seq,
81
117
};
82
-
next_seq += 1;
83
-
return records[idx];
84
118
}
85
119
86
120
pub fn delete(did: []const u8, collection: []const u8, rkey: []const u8) void {
87
-
if (findIndex(did, collection, rkey)) |idx| {
88
-
records[idx].active = false;
89
-
}
121
+
mutex.lockUncancelable(store_io);
122
+
defer mutex.unlock(store_io);
123
+
requireInitialized() catch return;
124
+
conn.exec("DELETE FROM records WHERE did = ? AND collection = ? AND rkey = ?", .{ did, collection, rkey }) catch {};
90
125
}
91
126
92
127
pub fn get(did: []const u8, collection: []const u8, rkey: []const u8) ?Record {
93
-
if (findIndex(did, collection, rkey)) |idx| return records[idx];
94
-
return null;
128
+
mutex.lockUncancelable(store_io);
129
+
defer mutex.unlock(store_io);
130
+
requireInitialized() catch return null;
131
+
132
+
const row = conn.row(
133
+
\\SELECT did, collection, rkey, cid, value_json, seq
134
+
\\FROM records
135
+
\\WHERE did = ? AND collection = ? AND rkey = ?
136
+
, .{ did, collection, rkey }) catch return null;
137
+
if (row == null) return null;
138
+
defer row.?.deinit();
139
+
return recordFromRow(row.?, std.heap.page_allocator) catch null;
140
+
}
141
+
142
+
pub fn count(did: []const u8, collection: []const u8) usize {
143
+
mutex.lockUncancelable(store_io);
144
+
defer mutex.unlock(store_io);
145
+
requireInitialized() catch return 0;
146
+
const row = conn.row("SELECT count(*) FROM records WHERE did = ? AND collection = ?", .{ did, collection }) catch return 0;
147
+
if (row == null) return 0;
148
+
defer row.?.deinit();
149
+
return @intCast(row.?.int(0));
150
+
}
151
+
152
+
pub fn countSubject(collection: []const u8, subject_did: []const u8) usize {
153
+
mutex.lockUncancelable(store_io);
154
+
defer mutex.unlock(store_io);
155
+
requireInitialized() catch return 0;
156
+
const row = conn.row("SELECT count(*) FROM records WHERE collection = ? AND instr(value_json, ?) > 0", .{ collection, subject_did }) catch return 0;
157
+
if (row == null) return 0;
158
+
defer row.?.deinit();
159
+
return @intCast(row.?.int(0));
160
+
}
161
+
162
+
pub fn setPreferences(account: auth.Account, value: std.json.Value) !void {
163
+
const value_json = try stringifyValue(std.heap.page_allocator, value);
164
+
mutex.lockUncancelable(store_io);
165
+
defer mutex.unlock(store_io);
166
+
try requireInitialized();
167
+
try conn.exec(
168
+
\\INSERT INTO preferences (did, value_json)
169
+
\\VALUES (?, ?)
170
+
\\ON CONFLICT(did) DO UPDATE SET value_json = excluded.value_json
171
+
, .{ account.did, value_json });
172
+
}
173
+
174
+
pub fn getPreferences(account: auth.Account) []const u8 {
175
+
mutex.lockUncancelable(store_io);
176
+
defer mutex.unlock(store_io);
177
+
requireInitialized() catch return "[]";
178
+
const row = conn.row("SELECT value_json FROM preferences WHERE did = ?", .{account.did}) catch return "[]";
179
+
if (row == null) return "[]";
180
+
defer row.?.deinit();
181
+
return std.heap.page_allocator.dupe(u8, row.?.text(0)) catch "[]";
182
+
}
183
+
184
+
pub fn putBlob(
185
+
allocator: std.mem.Allocator,
186
+
account: auth.Account,
187
+
data: []const u8,
188
+
mime_type: []const u8,
189
+
) ![]const u8 {
190
+
const cid = try cidForBlob(allocator, data);
191
+
192
+
mutex.lockUncancelable(store_io);
193
+
defer mutex.unlock(store_io);
194
+
try requireInitialized();
195
+
try conn.exec(
196
+
\\INSERT INTO blobs (cid, did, mime_type, size)
197
+
\\VALUES (?, ?, ?, ?)
198
+
\\ON CONFLICT(cid) DO UPDATE SET
199
+
\\ did = excluded.did,
200
+
\\ mime_type = excluded.mime_type,
201
+
\\ size = excluded.size
202
+
, .{ cid, account.did, mime_type, @as(i64, @intCast(data.len)) });
203
+
return cid;
95
204
}
96
205
97
206
pub fn writeRecordJson(allocator: std.mem.Allocator, record: Record) ![]const u8 {
···
113
222
collection: []const u8,
114
223
limit: usize,
115
224
) ![]const u8 {
225
+
mutex.lockUncancelable(store_io);
226
+
defer mutex.unlock(store_io);
227
+
try requireInitialized();
228
+
229
+
var rows = try conn.rows(
230
+
\\SELECT did, collection, rkey, cid, value_json, seq
231
+
\\FROM records
232
+
\\WHERE did = ? AND collection = ?
233
+
\\ORDER BY seq DESC
234
+
\\LIMIT ?
235
+
, .{ did, collection, @as(i64, @intCast(if (limit == 0) 100 else limit)) });
236
+
defer rows.deinit();
237
+
116
238
var out: std.Io.Writer.Allocating = .init(allocator);
117
239
defer out.deinit();
118
-
119
240
try out.writer.writeAll("{\"records\":[");
120
-
var remaining = if (limit == 0) max_records else limit;
121
241
var first = true;
122
-
var before: ?u64 = null;
123
-
while (remaining > 0) {
124
-
const maybe_record = newestBefore(did, collection, before);
125
-
const record = maybe_record orelse break;
242
+
while (rows.next()) |row| {
243
+
const record = try recordFromRow(row, std.heap.page_allocator);
126
244
if (!first) try out.writer.writeByte(',');
127
245
first = false;
128
-
129
246
const uri = try record.uri(allocator);
130
247
defer allocator.free(uri);
131
248
try out.writer.print(
132
249
"{{\"uri\":{f},\"cid\":{f},\"value\":{s}}}",
133
250
.{ std.json.fmt(uri, .{}), std.json.fmt(record.cid, .{}), record.value_json },
134
251
);
135
-
before = record.seq;
136
-
remaining -= 1;
137
252
}
253
+
if (rows.err) |err| return err;
138
254
try out.writer.writeAll("]}");
139
255
return out.toOwnedSlice();
140
256
}
141
257
142
-
fn stringifyValue(allocator: std.mem.Allocator, value: std.json.Value) ![]const u8 {
258
+
pub fn writeAuthorFeedJson(
259
+
allocator: std.mem.Allocator,
260
+
did: []const u8,
261
+
handle: []const u8,
262
+
limit: usize,
263
+
) ![]const u8 {
264
+
mutex.lockUncancelable(store_io);
265
+
defer mutex.unlock(store_io);
266
+
try requireInitialized();
267
+
268
+
var rows = try conn.rows(
269
+
\\SELECT did, collection, rkey, cid, value_json, seq
270
+
\\FROM records
271
+
\\WHERE did = ? AND collection = 'app.bsky.feed.post'
272
+
\\ORDER BY seq DESC
273
+
\\LIMIT ?
274
+
, .{ did, @as(i64, @intCast(if (limit == 0) 100 else limit)) });
275
+
defer rows.deinit();
276
+
143
277
var out: std.Io.Writer.Allocating = .init(allocator);
144
278
defer out.deinit();
145
-
try out.writer.print("{f}", .{std.json.fmt(value, .{})});
279
+
try out.writer.writeAll("{\"feed\":[");
280
+
var first = true;
281
+
while (rows.next()) |row| {
282
+
const record = try recordFromRow(row, std.heap.page_allocator);
283
+
if (!first) try out.writer.writeByte(',');
284
+
first = false;
285
+
const uri = try record.uri(allocator);
286
+
defer allocator.free(uri);
287
+
try out.writer.print(
288
+
"{{\"post\":{{\"uri\":{f},\"cid\":{f},\"author\":{{\"did\":{f},\"handle\":{f},\"displayName\":{f}}},\"record\":{s},\"replyCount\":0,\"repostCount\":0,\"likeCount\":0,\"quoteCount\":0,\"indexedAt\":\"2026-05-20T00:00:00.000Z\",\"viewer\":{{}}}}}}",
289
+
.{
290
+
std.json.fmt(uri, .{}),
291
+
std.json.fmt(record.cid, .{}),
292
+
std.json.fmt(did, .{}),
293
+
std.json.fmt(handle, .{}),
294
+
std.json.fmt(handle, .{}),
295
+
record.value_json,
296
+
},
297
+
);
298
+
}
299
+
if (rows.err) |err| return err;
300
+
try out.writer.writeAll("]}");
146
301
return out.toOwnedSlice();
147
302
}
148
303
149
-
fn nextRkey(allocator: std.mem.Allocator) ![]const u8 {
150
-
return std.fmt.allocPrint(allocator, "3zds{d}", .{next_seq});
304
+
fn migrate() !void {
305
+
inline for (schema_statements) |sql| try conn.execNoArgs(sql);
151
306
}
152
307
153
-
fn findIndex(did: []const u8, collection: []const u8, rkey: []const u8) ?usize {
154
-
for (records, 0..) |record, idx| {
155
-
if (!record.active) continue;
156
-
if (std.mem.eql(u8, record.did, did) and
157
-
std.mem.eql(u8, record.collection, collection) and
158
-
std.mem.eql(u8, record.rkey, rkey))
159
-
{
160
-
return idx;
161
-
}
308
+
fn seedAccounts() !void {
309
+
for (auth.accounts) |account| {
310
+
try conn.exec(
311
+
\\INSERT INTO accounts (did, handle, password_hash)
312
+
\\VALUES (?, ?, ?)
313
+
\\ON CONFLICT(did) DO UPDATE SET
314
+
\\ handle = excluded.handle,
315
+
\\ password_hash = excluded.password_hash
316
+
, .{ account.did, account.handle, account.password });
162
317
}
163
-
return null;
164
318
}
165
319
166
-
fn freeIndex() ?usize {
167
-
for (records, 0..) |record, idx| {
168
-
if (!record.active) return idx;
169
-
}
170
-
return null;
320
+
fn nextSeqLocked() !u64 {
321
+
const row = try conn.row("SELECT COALESCE(MAX(seq), 0) + 1 FROM commits", .{});
322
+
if (row == null) return 1;
323
+
defer row.?.deinit();
324
+
return @intCast(row.?.int(0));
171
325
}
172
326
173
-
fn newestBefore(did: []const u8, collection: []const u8, before: ?u64) ?Record {
174
-
var best: ?Record = null;
175
-
for (records) |record| {
176
-
if (!record.active) continue;
177
-
if (!std.mem.eql(u8, record.did, did)) continue;
178
-
if (!std.mem.eql(u8, record.collection, collection)) continue;
179
-
if (before) |seq| {
180
-
if (record.seq >= seq) continue;
181
-
}
182
-
if (best == null or record.seq > best.?.seq) best = record;
183
-
}
184
-
return best;
327
+
fn nextRkey(allocator: std.mem.Allocator) ![]const u8 {
328
+
mutex.lockUncancelable(store_io);
329
+
defer mutex.unlock(store_io);
330
+
try requireInitialized();
331
+
return std.fmt.allocPrint(allocator, "3zds{d}", .{try nextSeqLocked()});
185
332
}
186
333
187
-
test "stores and lists records newest first" {
188
-
const alice = auth.findAccount("alice.test").?;
189
-
const one = try std.json.parseFromSlice(std.json.Value, std.testing.allocator, "{\"text\":\"one\"}", .{});
190
-
defer one.deinit();
191
-
const two = try std.json.parseFromSlice(std.json.Value, std.testing.allocator, "{\"text\":\"two\"}", .{});
192
-
defer two.deinit();
334
+
fn recordFromRow(row: zqlite.Row, allocator: std.mem.Allocator) !Record {
335
+
return .{
336
+
.did = try allocator.dupe(u8, row.text(0)),
337
+
.collection = try allocator.dupe(u8, row.text(1)),
338
+
.rkey = try allocator.dupe(u8, row.text(2)),
339
+
.cid = try allocator.dupe(u8, row.text(3)),
340
+
.value_json = try allocator.dupe(u8, row.text(4)),
341
+
.seq = @intCast(row.int(5)),
342
+
};
343
+
}
193
344
194
-
_ = try create(std.testing.allocator, alice, "app.bsky.feed.post", "a", one.value);
195
-
_ = try create(std.testing.allocator, alice, "app.bsky.feed.post", "b", two.value);
196
-
const body = try writeListJson(std.testing.allocator, alice.did, "app.bsky.feed.post", 10);
197
-
defer std.testing.allocator.free(body);
345
+
fn stringifyValue(allocator: std.mem.Allocator, value: std.json.Value) ![]const u8 {
346
+
var out: std.Io.Writer.Allocating = .init(allocator);
347
+
defer out.deinit();
348
+
try out.writer.print("{f}", .{std.json.fmt(value, .{})});
349
+
return out.toOwnedSlice();
350
+
}
351
+
352
+
fn cidForJson(allocator: std.mem.Allocator, json: []const u8) ![]const u8 {
353
+
const cid = try zat.cbor.Cid.forDagCbor(allocator, json);
354
+
defer allocator.free(cid.raw);
355
+
return zat.multibase.base32lower.encode(allocator, cid.raw);
356
+
}
357
+
358
+
fn cidForBlob(allocator: std.mem.Allocator, data: []const u8) ![]const u8 {
359
+
const cid = try zat.cbor.Cid.create(allocator, 1, 0x55, 0x12, data);
360
+
defer allocator.free(cid.raw);
361
+
return zat.multibase.base32lower.encode(allocator, cid.raw);
362
+
}
363
+
364
+
fn requireInitialized() !void {
365
+
if (!initialized) return Error.StoreNotInitialized;
366
+
}
198
367
199
-
try std.testing.expect(std.mem.indexOf(u8, body, "\"text\":\"two\"").? <
200
-
std.mem.indexOf(u8, body, "\"text\":\"one\"").?);
368
+
const schema_statements = [_][*:0]const u8{
369
+
\\CREATE TABLE IF NOT EXISTS accounts (
370
+
\\ did TEXT PRIMARY KEY,
371
+
\\ handle TEXT NOT NULL UNIQUE,
372
+
\\ password_hash TEXT NOT NULL,
373
+
\\ created_at INTEGER NOT NULL DEFAULT (unixepoch())
374
+
\\)
375
+
,
376
+
\\CREATE TABLE IF NOT EXISTS records (
377
+
\\ did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE,
378
+
\\ collection TEXT NOT NULL,
379
+
\\ rkey TEXT NOT NULL,
380
+
\\ uri TEXT NOT NULL UNIQUE,
381
+
\\ cid TEXT NOT NULL,
382
+
\\ value_json BLOB NOT NULL,
383
+
\\ rev TEXT NOT NULL,
384
+
\\ seq INTEGER NOT NULL,
385
+
\\ PRIMARY KEY (did, collection, rkey)
386
+
\\)
387
+
,
388
+
"CREATE INDEX IF NOT EXISTS records_collection_idx ON records (did, collection, seq DESC)",
389
+
"CREATE INDEX IF NOT EXISTS records_cid_idx ON records (cid)",
390
+
\\CREATE TABLE IF NOT EXISTS commits (
391
+
\\ seq INTEGER PRIMARY KEY,
392
+
\\ did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE,
393
+
\\ cid TEXT NOT NULL,
394
+
\\ rev TEXT NOT NULL,
395
+
\\ prev TEXT
396
+
\\)
397
+
,
398
+
"CREATE INDEX IF NOT EXISTS commits_did_idx ON commits (did, seq DESC)",
399
+
\\CREATE TABLE IF NOT EXISTS seq_events (
400
+
\\ seq INTEGER PRIMARY KEY,
401
+
\\ did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE,
402
+
\\ commit_cid TEXT NOT NULL,
403
+
\\ evt BLOB NOT NULL
404
+
\\)
405
+
,
406
+
\\CREATE TABLE IF NOT EXISTS blocks (
407
+
\\ cid TEXT PRIMARY KEY,
408
+
\\ data BLOB NOT NULL
409
+
\\)
410
+
,
411
+
\\CREATE TABLE IF NOT EXISTS blobs (
412
+
\\ cid TEXT PRIMARY KEY,
413
+
\\ did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE,
414
+
\\ mime_type TEXT NOT NULL,
415
+
\\ size INTEGER NOT NULL,
416
+
\\ storage TEXT NOT NULL DEFAULT 'local',
417
+
\\ created_at INTEGER NOT NULL DEFAULT (unixepoch())
418
+
\\)
419
+
,
420
+
\\CREATE TABLE IF NOT EXISTS record_blobs (
421
+
\\ blob_cid TEXT NOT NULL REFERENCES blobs(cid) ON DELETE CASCADE,
422
+
\\ record_uri TEXT NOT NULL REFERENCES records(uri) ON DELETE CASCADE,
423
+
\\ PRIMARY KEY (blob_cid, record_uri)
424
+
\\)
425
+
,
426
+
\\CREATE TABLE IF NOT EXISTS preferences (
427
+
\\ did TEXT PRIMARY KEY REFERENCES accounts(did) ON DELETE CASCADE,
428
+
\\ value_json BLOB NOT NULL
429
+
\\)
430
+
,
431
+
};
432
+
433
+
test "persists records in sqlite" {
434
+
try init(std.Options.debug_io, ":memory:");
435
+
defer close();
436
+
437
+
const account = auth.findAccount("alice.test").?;
438
+
const parsed = try std.json.parseFromSlice(std.json.Value, std.testing.allocator, "{\"text\":\"hello\"}", .{});
439
+
defer parsed.deinit();
440
+
441
+
const record = try create(std.testing.allocator, account, "app.bsky.feed.post", "3ztest", parsed.value);
442
+
try std.testing.expectEqualStrings("3ztest", record.rkey);
443
+
444
+
const fetched = get(account.did, "app.bsky.feed.post", "3ztest").?;
445
+
try std.testing.expectEqualStrings(record.cid, fetched.cid);
446
+
try std.testing.expect(std.mem.indexOf(u8, fetched.value_json, "hello") != null);
201
447
}