+2

dev/atproto-smoke-dual.json

Reviewed

··· 1 1 { 2 2 "pdsUrl": "http://localhost:2583", 3 3 "pdsHost": "localhost:2583", 4 4 + "publicApiUrl": "http://localhost:2583", 5 5 + "publicCheckTimeoutMs": 15000, 4 6 "artifactsDir": "/tmp/zds-reference-study/atproto-smoke/.tmp/zds-artifacts", 5 7 "browserExecutablePath": "/Users/nate/tangled.org/zzstoatzz.io/zds/dev/chromium-insecure.sh", 6 8 "headless": true,

+197 -18

src/server.zig

Reviewed

··· 27 27 repo_delete_record, 28 28 repo_apply_writes, 29 29 repo_upload_blob, 30 30 + sync_get_blob, 30 31 identity_resolve_handle, 31 32 actor_get_profile, 32 33 actor_get_profiles, 34 34 + actor_search, 35 35 + actor_search_typeahead, 33 36 actor_get_preferences, 34 37 actor_put_preferences, 35 38 graph_get_follows, 39 39 + graph_get_lists, 40 40 + graph_get_list, 41 41 + graph_get_suggested_follows_by_actor, 36 42 feed_get_timeline, 37 43 feed_get_feed, 38 44 feed_get_author_feed, ··· 40 46 age_assurance_get_state, 41 47 notification_list, 42 48 unspecced_get_config, 49 49 + unspecced_get_trending_topics, 50 50 + unspecced_get_post_thread_v2, 43 51 labeler_get_services, 44 52 chat_get_log, 45 53 chat_list_convos, 54 54 + chat_get_convo_availability, 46 55 not_found, 47 56 }; 48 57 ··· 85 94 if (method == .POST and std.mem.eql(u8, path, "/xrpc/com.atproto.repo.uploadBlob")) { 86 95 return .repo_upload_blob; 87 96 } 97 97 + if (method == .GET and std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getBlob")) { 98 98 + return .sync_get_blob; 99 99 + } 88 100 if (method == .GET and std.mem.eql(u8, path, "/xrpc/com.atproto.identity.resolveHandle")) { 89 101 return .identity_resolve_handle; 90 102 } ··· 94 106 if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.actor.getProfiles")) { 95 107 return .actor_get_profiles; 96 108 } 109 109 + if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.actor.searchActors")) { 110 110 + return .actor_search; 111 111 + } 112 112 + if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.actor.searchActorsTypeahead")) { 113 113 + return .actor_search_typeahead; 114 114 + } 97 115 if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.actor.getPreferences")) { 98 116 return .actor_get_preferences; 99 117 } ··· 102 120 } 103 121 if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.graph.getFollows")) { 104 122 return .graph_get_follows; 123 123 + } 124 124 + if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.graph.getLists")) { 125 125 + return .graph_get_lists; 126 126 + } 127 127 + if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.graph.getList")) { 128 128 + return .graph_get_list; 129 129 + } 130 130 + if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.graph.getSuggestedFollowsByActor")) { 131 131 + return .graph_get_suggested_follows_by_actor; 105 132 } 106 133 if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.feed.getTimeline")) { 107 134 return .feed_get_timeline; ··· 124 151 if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.unspecced.getConfig")) { 125 152 return .unspecced_get_config; 126 153 } 154 154 + if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.unspecced.getTrendingTopics")) { 155 155 + return .unspecced_get_trending_topics; 156 156 + } 157 157 + if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.unspecced.getPostThreadV2")) { 158 158 + return .unspecced_get_post_thread_v2; 159 159 + } 127 160 if (method == .GET and std.mem.eql(u8, path, "/xrpc/app.bsky.labeler.getServices")) { 128 161 return .labeler_get_services; 129 162 } ··· 132 165 } 133 166 if (method == .GET and std.mem.eql(u8, path, "/xrpc/chat.bsky.convo.listConvos")) { 134 167 return .chat_list_convos; 168 168 + } 169 169 + if (method == .GET and std.mem.eql(u8, path, "/xrpc/chat.bsky.convo.getConvoAvailability")) { 170 170 + return .chat_get_convo_availability; 135 171 } 136 172 return .not_found; 137 173 } ··· 196 232 .repo_delete_record => try repoDeleteRecord(request), 197 233 .repo_apply_writes => try repoApplyWrites(request), 198 234 .repo_upload_blob => try repoUploadBlob(request), 235 235 + .sync_get_blob => try syncGetBlob(request), 199 236 .identity_resolve_handle => try identityResolveHandle(request), 200 237 .actor_get_profile => try actorGetProfile(request), 201 238 .actor_get_profiles => try actorGetProfiles(request), 239 239 + .actor_search => try actorSearch(request), 240 240 + .actor_search_typeahead => try actorSearch(request), 202 241 .actor_get_preferences => try actorGetPreferences(request), 203 242 .actor_put_preferences => try actorPutPreferences(request), 204 243 .graph_get_follows => try authenticatedEmpty(request, "{\"follows\":[]}"), 244 244 + .graph_get_lists => try graphGetLists(request), 245 245 + .graph_get_list => try graphGetList(request), 246 246 + .graph_get_suggested_follows_by_actor => try authenticatedEmpty(request, "{\"suggestions\":[]}"), 205 247 .feed_get_timeline => try authenticatedEmpty(request, "{\"feed\":[]}"), 206 248 .feed_get_feed => try authenticatedEmpty(request, "{\"feed\":[]}"), 207 249 .feed_get_author_feed => try feedGetAuthorFeed(request), ··· 209 251 .age_assurance_get_state => try ageAssuranceGetState(request), 210 252 .notification_list => try authenticatedEmpty(request, "{\"notifications\":[],\"seenAt\":\"2026-05-20T00:00:00.000Z\"}"), 211 253 .unspecced_get_config => try json(request, .ok, "{\"checkEmailConfirmed\":false,\"liveNow\":{},\"featureFlags\":{}}"), 254 254 + .unspecced_get_trending_topics => try authenticatedEmpty(request, "{\"topics\":[]}"), 255 255 + .unspecced_get_post_thread_v2 => try authenticatedEmpty(request, "{\"thread\":{\"type\":\"thread\",\"items\":[]}}"), 212 256 .labeler_get_services => try json(request, .ok, "{\"views\":[]}"), 213 257 .chat_get_log => try authenticatedEmpty(request, "{\"logs\":[],\"cursor\":\"\"}"), 214 258 .chat_list_convos => try authenticatedEmpty(request, "{\"convos\":[]}"), 259 259 + .chat_get_convo_availability => try authenticatedEmpty(request, "{\"canChat\":false}"), 215 260 .not_found => try xrpcError(request, .not_found, "MethodNotImplemented", "Method not implemented"), 216 261 } 217 262 } ··· 473 518 474 519 var out: std.Io.Writer.Allocating = .init(allocator); 475 520 defer out.deinit(); 476 476 - try out.writer.writeAll("{\"commit\":{\"cid\":\"bafyreie5cvv4h45feadgeuwhbcutmh6t2ceseocckahdoe6uat64zmz454\",\"rev\":\"zds-dev\"},\"results\":["); 521 521 + try out.writer.writeByte('['); 522 522 + var latest_seq: u64 = 0; 477 523 for (writes.array.items, 0..) |write, idx| { 478 524 if (write != .object) return xrpcError(request, .bad_request, "InvalidRequest", "Invalid write"); 479 525 const type_name = valueString(write, "$type") orelse { ··· 493 539 error.InvalidRecordKey => return xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"), 494 540 else => return err, 495 541 }; 542 542 + latest_seq = @max(latest_seq, record.seq); 496 543 const uri = try record.uri(allocator); 497 544 try out.writer.print( 498 545 "{{\"$type\":\"com.atproto.repo.applyWrites#createResult\",\"uri\":{f},\"cid\":{f},\"validationStatus\":\"unknown\"}}", ··· 510 557 error.InvalidRecordKey => return xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"), 511 558 else => return err, 512 559 }; 560 560 + latest_seq = @max(latest_seq, record.seq); 513 561 const uri = try record.uri(allocator); 514 562 try out.writer.print( 515 563 "{{\"$type\":\"com.atproto.repo.applyWrites#updateResult\",\"uri\":{f},\"cid\":{f},\"validationStatus\":\"unknown\"}}", ··· 525 573 return xrpcError(request, .bad_request, "InvalidRequest", "Unknown write operation type"); 526 574 } 527 575 } 528 528 - try out.writer.writeAll("]}"); 529 529 - return json(request, .ok, out.written()); 576 576 + try out.writer.writeByte(']'); 577 577 + const rev = try store.revForSeq(allocator, if (latest_seq == 0) 1 else latest_seq); 578 578 + const body_out = try std.fmt.allocPrint( 579 579 + allocator, 580 580 + "{{\"commit\":{{\"cid\":\"bafyreie5cvv4h45feadgeuwhbcutmh6t2ceseocckahdoe6uat64zmz454\",\"rev\":{f}}},\"results\":{s}}}", 581 581 + .{ std.json.fmt(rev, .{}), out.written() }, 582 582 + ); 583 583 + return json(request, .ok, body_out); 530 584 } 531 585 532 586 fn repoUploadBlob(request: *http.Server.Request) !void { ··· 539 593 error.InvalidToken => return xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"), 540 594 }; 541 595 596 596 + const mime_type = headerValue(request, "content-type") orelse "application/octet-stream"; 542 597 var body_buf: [2 * 1024 * 1024]u8 = undefined; 543 598 const body = try readBody(request, &body_buf); 544 544 - const mime_type = headerValue(request, "content-type") orelse "application/octet-stream"; 545 599 const cid = try store.putBlob(allocator, account, body, mime_type); 546 600 const body_out = try std.fmt.allocPrint( 547 601 allocator, ··· 551 605 return json(request, .ok, body_out); 552 606 } 553 607 608 608 + fn syncGetBlob(request: *http.Server.Request) !void { 609 609 + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); 610 610 + defer arena.deinit(); 611 611 + const allocator = arena.allocator(); 612 612 + 613 613 + var did_buf: [256]u8 = undefined; 614 614 + const did = queryParam(request.head.target, "did", &did_buf) orelse { 615 615 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing did"); 616 616 + }; 617 617 + var cid_buf: [256]u8 = undefined; 618 618 + const cid = queryParam(request.head.target, "cid", &cid_buf) orelse { 619 619 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing cid"); 620 620 + }; 621 621 + const blob = store.getBlob(allocator, did, cid) orelse { 622 622 + return xrpcError(request, .not_found, "BlobNotFound", "Blob not found"); 623 623 + }; 624 624 + const headers = [_]http.Header{ 625 625 + .{ .name = "content-type", .value = blob.mime_type }, 626 626 + .{ .name = "cache-control", .value = "public, max-age=31536000, immutable" }, 627 627 + .{ .name = "x-content-type-options", .value = "nosniff" }, 628 628 + .{ .name = "access-control-allow-origin", .value = "*" }, 629 629 + .{ .name = "access-control-allow-private-network", .value = "true" }, 630 630 + .{ .name = "connection", .value = "close" }, 631 631 + }; 632 632 + try request.respond(blob.data, .{ 633 633 + .status = .ok, 634 634 + .extra_headers = &headers, 635 635 + }); 636 636 + } 637 637 + 554 638 fn identityResolveHandle(request: *http.Server.Request) !void { 555 639 var handle_buf: [256]u8 = undefined; 556 640 const handle = queryParam(request.head.target, "handle", &handle_buf) orelse { ··· 602 686 return json(request, .ok, body); 603 687 } 604 688 689 689 + fn actorSearch(request: *http.Server.Request) !void { 690 690 + var query_buf: [128]u8 = undefined; 691 691 + const query = queryParam(request.head.target, "q", &query_buf) orelse 692 692 + queryParam(request.head.target, "term", &query_buf) orelse ""; 693 693 + const limit = queryLimit(request.head.target, 25); 694 694 + 695 695 + var out: std.Io.Writer.Allocating = .init(std.heap.page_allocator); 696 696 + defer out.deinit(); 697 697 + try out.writer.writeAll("{\"actors\":["); 698 698 + var emitted: usize = 0; 699 699 + for (auth.accounts) |account| { 700 700 + if (emitted >= limit) break; 701 701 + if (query.len != 0 and 702 702 + std.ascii.indexOfIgnoreCase(account.handle, query) == null and 703 703 + std.ascii.indexOfIgnoreCase(account.did, query) == null) continue; 704 704 + if (emitted != 0) try out.writer.writeByte(','); 705 705 + emitted += 1; 706 706 + try out.writer.print( 707 707 + "{{\"did\":{f},\"handle\":{f},\"displayName\":{f},\"viewer\":{{\"muted\":false,\"blockedBy\":false}}}}", 708 708 + .{ std.json.fmt(account.did, .{}), std.json.fmt(account.handle, .{}), std.json.fmt(account.handle, .{}) }, 709 709 + ); 710 710 + } 711 711 + try out.writer.writeAll("]}"); 712 712 + return json(request, .ok, out.written()); 713 713 + } 714 714 + 605 715 fn actorPutPreferences(request: *http.Server.Request) !void { 606 716 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); 607 717 defer arena.deinit(); ··· 619 729 }; 620 730 try store.setPreferences(account, preferences); 621 731 return json(request, .ok, "{}"); 732 732 + } 733 733 + 734 734 + fn graphGetLists(request: *http.Server.Request) !void { 735 735 + _ = requireBearerAccount(request, std.heap.page_allocator) catch |err| switch (err) { 736 736 + error.AuthRequired => return xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"), 737 737 + error.InvalidToken => return xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"), 738 738 + }; 739 739 + var actor_buf: [256]u8 = undefined; 740 740 + const actor = queryParam(request.head.target, "actor", &actor_buf) orelse { 741 741 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing actor"); 742 742 + }; 743 743 + const account = store.resolveRepo(actor) orelse { 744 744 + return xrpcError(request, .bad_request, "InvalidRequest", "Actor not found"); 745 745 + }; 746 746 + const limit = queryLimit(request.head.target, 50); 747 747 + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); 748 748 + defer arena.deinit(); 749 749 + const body = try store.writeGraphListsJson(arena.allocator(), account, limit); 750 750 + return json(request, .ok, body); 751 751 + } 752 752 + 753 753 + fn graphGetList(request: *http.Server.Request) !void { 754 754 + _ = requireBearerAccount(request, std.heap.page_allocator) catch |err| switch (err) { 755 755 + error.AuthRequired => return xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"), 756 756 + error.InvalidToken => return xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"), 757 757 + }; 758 758 + var list_buf: [512]u8 = undefined; 759 759 + const list_uri = queryParam(request.head.target, "list", &list_buf) orelse { 760 760 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing list"); 761 761 + }; 762 762 + const list = store.getByUri(list_uri) orelse { 763 763 + return xrpcError(request, .not_found, "NotFound", "List not found"); 764 764 + }; 765 765 + const limit = queryLimit(request.head.target, 50); 766 766 + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); 767 767 + defer arena.deinit(); 768 768 + const body = try store.writeGraphListJson(arena.allocator(), list, limit); 769 769 + return json(request, .ok, body); 622 770 } 623 771 624 772 fn feedGetAuthorFeed(request: *http.Server.Request) !void { ··· 722 870 } 723 871 724 872 fn actorProfileJson(allocator: std.mem.Allocator, account: auth.Account) ![]const u8 { 725 725 - const display_name, const description = profileText(allocator, account) catch .{ account.handle, "" }; 873 873 + const profile = profileDetails(allocator, account) catch ProfileDetails{}; 726 874 const posts_count = store.count(account.did, "app.bsky.feed.post"); 727 875 const follows_count = store.count(account.did, "app.bsky.graph.follow"); 728 876 const followers_count = store.countSubject("app.bsky.graph.follow", account.did); 729 729 - return std.fmt.allocPrint( 730 730 - allocator, 731 731 - "{{\"did\":{f},\"handle\":{f},\"displayName\":{f},\"description\":{f},\"postsCount\":{d},\"followersCount\":{d},\"followsCount\":{d},\"viewer\":{{\"muted\":false,\"blockedBy\":false}}}}", 877 877 + var out: std.Io.Writer.Allocating = .init(allocator); 878 878 + defer out.deinit(); 879 879 + try out.writer.print( 880 880 + "{{\"did\":{f},\"handle\":{f},\"displayName\":{f},\"description\":{f}", 732 881 .{ 733 882 std.json.fmt(account.did, .{}), 734 883 std.json.fmt(account.handle, .{}), 735 735 - std.json.fmt(display_name, .{}), 736 736 - std.json.fmt(description, .{}), 737 737 - posts_count, 738 738 - followers_count, 739 739 - follows_count, 884 884 + std.json.fmt(profile.display_name orelse account.handle, .{}), 885 885 + std.json.fmt(profile.description orelse "", .{}), 740 886 }, 741 887 ); 888 888 + if (profile.avatar_cid) |avatar_cid| { 889 889 + try out.writer.print( 890 890 + ",\"avatar\":\"http://localhost:2583/xrpc/com.atproto.sync.getBlob?did={s}&cid={s}\"", 891 891 + .{ account.did, avatar_cid }, 892 892 + ); 893 893 + } 894 894 + try out.writer.print( 895 895 + ",\"postsCount\":{d},\"followersCount\":{d},\"followsCount\":{d},\"viewer\":{{\"muted\":false,\"blockedBy\":false}}}}", 896 896 + .{ posts_count, followers_count, follows_count }, 897 897 + ); 898 898 + return out.toOwnedSlice(); 742 899 } 743 900 744 744 - fn profileText(allocator: std.mem.Allocator, account: auth.Account) !struct { []const u8, []const u8 } { 745 745 - const profile = store.get(account.did, "app.bsky.actor.profile", "self") orelse return .{ account.handle, "" }; 901 901 + const ProfileDetails = struct { 902 902 + display_name: ?[]const u8 = null, 903 903 + description: ?[]const u8 = null, 904 904 + avatar_cid: ?[]const u8 = null, 905 905 + }; 906 906 + 907 907 + fn profileDetails(allocator: std.mem.Allocator, account: auth.Account) !ProfileDetails { 908 908 + const profile = store.get(account.did, "app.bsky.actor.profile", "self") orelse return .{}; 746 909 const parsed = try std.json.parseFromSlice(std.json.Value, allocator, profile.value_json, .{}); 747 747 - const display_name = zat.json.getString(parsed.value, "displayName") orelse account.handle; 748 748 - const description = zat.json.getString(parsed.value, "description") orelse ""; 749 749 - return .{ display_name, description }; 910 910 + const avatar_cid = if (parsed.value == .object) blk: { 911 911 + const avatar = parsed.value.object.get("avatar") orelse break :blk null; 912 912 + if (avatar != .object) break :blk null; 913 913 + const ref = avatar.object.get("ref") orelse break :blk null; 914 914 + if (ref != .object) break :blk null; 915 915 + const link = ref.object.get("$link") orelse break :blk null; 916 916 + break :blk if (link == .string) link.string else null; 917 917 + } else null; 918 918 + return .{ 919 919 + .display_name = zat.json.getString(parsed.value, "displayName"), 920 920 + .description = zat.json.getString(parsed.value, "description"), 921 921 + .avatar_cid = avatar_cid, 922 922 + }; 750 923 } 751 924 752 925 fn readBody(request: *http.Server.Request, buf: []u8) ![]const u8 { ··· 775 948 return percentDecode(param[eq + 1 ..], out) catch null; 776 949 } 777 950 return null; 951 951 + } 952 952 + 953 953 + fn queryLimit(target: []const u8, default: usize) usize { 954 954 + var buf: [16]u8 = undefined; 955 955 + const raw = queryParam(target, "limit", &buf) orelse return default; 956 956 + return std.fmt.parseInt(usize, raw, 10) catch default; 778 957 } 779 958 780 959 fn percentDecode(input: []const u8, out: []u8) ![]const u8 {

+187 -5

src/store.zig

Reviewed

··· 1 1 const std = @import("std"); 2 2 + const atid = @import("atid.zig"); 2 3 const auth = @import("auth.zig"); 3 4 const zat = @import("zat"); 4 5 const zqlite = @import("zqlite"); ··· 22 23 pub fn uri(self: Record, allocator: std.mem.Allocator) ![]const u8 { 23 24 return std.fmt.allocPrint(allocator, "at://{s}/{s}/{s}", .{ self.did, self.collection, self.rkey }); 24 25 } 26 26 + }; 27 27 + 28 28 + pub const BlobRecord = struct { 29 29 + mime_type: []const u8, 30 30 + data: []const u8, 25 31 }; 26 32 27 33 var conn: zqlite.Conn = undefined; ··· 87 93 const seq = try nextSeqLocked(); 88 94 const cid = try cidForJson(std.heap.page_allocator, value_json); 89 95 const uri = try std.fmt.allocPrint(std.heap.page_allocator, "at://{s}/{s}/{s}", .{ account.did, collection, rkey }); 90 90 - const rev = try std.fmt.allocPrint(std.heap.page_allocator, "zds-{d}", .{seq}); 96 96 + const rev = try revForSeq(std.heap.page_allocator, seq); 91 97 92 98 try conn.exclusiveTransaction(); 93 99 errdefer conn.rollback(); ··· 117 123 }; 118 124 } 119 125 126 126 + pub fn revForSeq(allocator: std.mem.Allocator, seq: u64) ![]const u8 { 127 127 + const timestamp_us = 1_704_067_200_000_000 + seq; 128 128 + const tid = try atid.encode(timestamp_us, @intCast(seq % 1024)); 129 129 + return allocator.dupe(u8, &tid); 130 130 + } 131 131 + 120 132 pub fn delete(did: []const u8, collection: []const u8, rkey: []const u8) void { 121 133 mutex.lockUncancelable(store_io); 122 134 defer mutex.unlock(store_io); ··· 139 151 return recordFromRow(row.?, std.heap.page_allocator) catch null; 140 152 } 141 153 154 154 + pub fn getByUri(uri: []const u8) ?Record { 155 155 + mutex.lockUncancelable(store_io); 156 156 + defer mutex.unlock(store_io); 157 157 + requireInitialized() catch return null; 158 158 + 159 159 + const row = conn.row( 160 160 + \\SELECT did, collection, rkey, cid, value_json, seq 161 161 + \\FROM records 162 162 + \\WHERE uri = ? 163 163 + , .{uri}) catch return null; 164 164 + if (row == null) return null; 165 165 + defer row.?.deinit(); 166 166 + return recordFromRow(row.?, std.heap.page_allocator) catch null; 167 167 + } 168 168 + 142 169 pub fn count(did: []const u8, collection: []const u8) usize { 143 170 mutex.lockUncancelable(store_io); 144 171 defer mutex.unlock(store_io); 145 172 requireInitialized() catch return 0; 146 173 const row = conn.row("SELECT count(*) FROM records WHERE did = ? AND collection = ?", .{ did, collection }) catch return 0; 174 174 + if (row == null) return 0; 175 175 + defer row.?.deinit(); 176 176 + return @intCast(row.?.int(0)); 177 177 + } 178 178 + 179 179 + pub fn countListItems(list_uri: []const u8) usize { 180 180 + mutex.lockUncancelable(store_io); 181 181 + defer mutex.unlock(store_io); 182 182 + requireInitialized() catch return 0; 183 183 + return countListItemsLocked(list_uri); 184 184 + } 185 185 + 186 186 + fn countListItemsLocked(list_uri: []const u8) usize { 187 187 + const row = conn.row( 188 188 + \\SELECT count(*) 189 189 + \\FROM records 190 190 + \\WHERE collection = 'app.bsky.graph.listitem' 191 191 + \\ AND instr(value_json, ?) > 0 192 192 + , .{list_uri}) catch return 0; 147 193 if (row == null) return 0; 148 194 defer row.?.deinit(); 149 195 return @intCast(row.?.int(0)); ··· 193 239 defer mutex.unlock(store_io); 194 240 try requireInitialized(); 195 241 try conn.exec( 196 196 - \\INSERT INTO blobs (cid, did, mime_type, size) 197 197 - \\VALUES (?, ?, ?, ?) 242 242 + \\INSERT INTO blobs (cid, did, mime_type, size, data) 243 243 + \\VALUES (?, ?, ?, ?, ?) 198 244 \\ON CONFLICT(cid) DO UPDATE SET 199 245 \\ did = excluded.did, 200 246 \\ mime_type = excluded.mime_type, 201 201 - \\ size = excluded.size 202 202 - , .{ cid, account.did, mime_type, @as(i64, @intCast(data.len)) }); 247 247 + \\ size = excluded.size, 248 248 + \\ data = excluded.data 249 249 + , .{ cid, account.did, mime_type, @as(i64, @intCast(data.len)), zqlite.blob(data) }); 203 250 return cid; 251 251 + } 252 252 + 253 253 + pub fn getBlob( 254 254 + allocator: std.mem.Allocator, 255 255 + did: []const u8, 256 256 + cid: []const u8, 257 257 + ) ?BlobRecord { 258 258 + mutex.lockUncancelable(store_io); 259 259 + defer mutex.unlock(store_io); 260 260 + requireInitialized() catch return null; 261 261 + const row = conn.row( 262 262 + \\SELECT mime_type, data 263 263 + \\FROM blobs 264 264 + \\WHERE did = ? AND cid = ? 265 265 + , .{ did, cid }) catch return null; 266 266 + if (row == null) return null; 267 267 + defer row.?.deinit(); 268 268 + const data = row.?.nullableBlob(1) orelse return null; 269 269 + return .{ 270 270 + .mime_type = allocator.dupe(u8, row.?.text(0)) catch return null, 271 271 + .data = allocator.dupe(u8, data) catch return null, 272 272 + }; 204 273 } 205 274 206 275 pub fn writeRecordJson(allocator: std.mem.Allocator, record: Record) ![]const u8 { ··· 255 324 return out.toOwnedSlice(); 256 325 } 257 326 327 327 + pub fn writeGraphListsJson( 328 328 + allocator: std.mem.Allocator, 329 329 + account: auth.Account, 330 330 + limit: usize, 331 331 + ) ![]const u8 { 332 332 + mutex.lockUncancelable(store_io); 333 333 + defer mutex.unlock(store_io); 334 334 + try requireInitialized(); 335 335 + 336 336 + var rows = try conn.rows( 337 337 + \\SELECT did, collection, rkey, cid, value_json, seq 338 338 + \\FROM records 339 339 + \\WHERE did = ? AND collection = 'app.bsky.graph.list' 340 340 + \\ORDER BY seq DESC 341 341 + \\LIMIT ? 342 342 + , .{ account.did, @as(i64, @intCast(if (limit == 0) 50 else limit)) }); 343 343 + defer rows.deinit(); 344 344 + 345 345 + var out: std.Io.Writer.Allocating = .init(allocator); 346 346 + defer out.deinit(); 347 347 + try out.writer.writeAll("{\"lists\":["); 348 348 + var first = true; 349 349 + while (rows.next()) |row| { 350 350 + const record = try recordFromRow(row, std.heap.page_allocator); 351 351 + if (!first) try out.writer.writeByte(','); 352 352 + first = false; 353 353 + try writeListView(&out.writer, allocator, record); 354 354 + } 355 355 + if (rows.err) |err| return err; 356 356 + try out.writer.writeAll("]}"); 357 357 + return out.toOwnedSlice(); 358 358 + } 359 359 + 360 360 + pub fn writeGraphListJson( 361 361 + allocator: std.mem.Allocator, 362 362 + list: Record, 363 363 + limit: usize, 364 364 + ) ![]const u8 { 365 365 + const list_uri = try list.uri(allocator); 366 366 + 367 367 + mutex.lockUncancelable(store_io); 368 368 + defer mutex.unlock(store_io); 369 369 + try requireInitialized(); 370 370 + 371 371 + var rows = try conn.rows( 372 372 + \\SELECT did, collection, rkey, cid, value_json, seq 373 373 + \\FROM records 374 374 + \\WHERE collection = 'app.bsky.graph.listitem' 375 375 + \\ AND instr(value_json, ?) 376 376 + \\ORDER BY seq DESC 377 377 + \\LIMIT ? 378 378 + , .{ list_uri, @as(i64, @intCast(if (limit == 0) 50 else limit)) }); 379 379 + defer rows.deinit(); 380 380 + 381 381 + var out: std.Io.Writer.Allocating = .init(allocator); 382 382 + defer out.deinit(); 383 383 + try out.writer.writeAll("{\"list\":"); 384 384 + try writeListView(&out.writer, allocator, list); 385 385 + try out.writer.writeAll(",\"items\":["); 386 386 + var first = true; 387 387 + while (rows.next()) |row| { 388 388 + const item = try recordFromRow(row, std.heap.page_allocator); 389 389 + const parsed = try std.json.parseFromSlice(std.json.Value, allocator, item.value_json, .{}); 390 390 + const subject_did = zat.json.getString(parsed.value, "subject") orelse continue; 391 391 + const item_uri = try item.uri(allocator); 392 392 + if (!first) try out.writer.writeByte(','); 393 393 + first = false; 394 394 + try out.writer.print("{{\"uri\":{f},\"subject\":", .{std.json.fmt(item_uri, .{})}); 395 395 + try writeProfileView(&out.writer, subject_did); 396 396 + try out.writer.writeByte('}'); 397 397 + } 398 398 + if (rows.err) |err| return err; 399 399 + try out.writer.writeAll("]}"); 400 400 + return out.toOwnedSlice(); 401 401 + } 402 402 + 258 403 pub fn writeAuthorFeedJson( 259 404 allocator: std.mem.Allocator, 260 405 did: []const u8, ··· 303 448 304 449 fn migrate() !void { 305 450 inline for (schema_statements) |sql| try conn.execNoArgs(sql); 451 451 + conn.execNoArgs("ALTER TABLE blobs ADD COLUMN data BLOB") catch {}; 306 452 } 307 453 308 454 fn seedAccounts() !void { ··· 342 488 }; 343 489 } 344 490 491 491 + fn writeListView( 492 492 + writer: *std.Io.Writer, 493 493 + allocator: std.mem.Allocator, 494 494 + record: Record, 495 495 + ) !void { 496 496 + const parsed = try std.json.parseFromSlice(std.json.Value, allocator, record.value_json, .{}); 497 497 + const uri = try record.uri(allocator); 498 498 + const name = zat.json.getString(parsed.value, "name") orelse "List"; 499 499 + const purpose = zat.json.getString(parsed.value, "purpose") orelse "app.bsky.graph.defs#curatelist"; 500 500 + const description = zat.json.getString(parsed.value, "description"); 501 501 + try writer.print( 502 502 + "{{\"uri\":{f},\"cid\":{f},\"creator\":", 503 503 + .{ std.json.fmt(uri, .{}), std.json.fmt(record.cid, .{}) }, 504 504 + ); 505 505 + try writeProfileView(writer, record.did); 506 506 + try writer.print( 507 507 + ",\"name\":{f},\"purpose\":{f}", 508 508 + .{ std.json.fmt(name, .{}), std.json.fmt(purpose, .{}) }, 509 509 + ); 510 510 + if (description) |text| try writer.print(",\"description\":{f}", .{std.json.fmt(text, .{})}); 511 511 + try writer.print( 512 512 + ",\"listItemCount\":{d},\"viewer\":{{}},\"indexedAt\":\"2026-05-20T00:00:00.000Z\"}}", 513 513 + .{countListItemsLocked(uri)}, 514 514 + ); 515 515 + } 516 516 + 517 517 + fn writeProfileView(writer: *std.Io.Writer, did: []const u8) !void { 518 518 + const account = auth.findAccount(did); 519 519 + const handle = if (account) |found| found.handle else did; 520 520 + try writer.print( 521 521 + "{{\"did\":{f},\"handle\":{f},\"displayName\":{f},\"viewer\":{{\"muted\":false,\"blockedBy\":false}}}}", 522 522 + .{ std.json.fmt(did, .{}), std.json.fmt(handle, .{}), std.json.fmt(handle, .{}) }, 523 523 + ); 524 524 + } 525 525 + 345 526 fn stringifyValue(allocator: std.mem.Allocator, value: std.json.Value) ![]const u8 { 346 527 var out: std.Io.Writer.Allocating = .init(allocator); 347 528 defer out.deinit(); ··· 413 594 \\ did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE, 414 595 \\ mime_type TEXT NOT NULL, 415 596 \\ size INTEGER NOT NULL, 597 597 + \\ data BLOB NOT NULL, 416 598 \\ storage TEXT NOT NULL DEFAULT 'local', 417 599 \\ created_at INTEGER NOT NULL DEFAULT (unixepoch()) 418 600 \\)