Something went wrong. Try again.
atproto relay in zig zlay.waow.tech
relay zig atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927//! XRPC endpoint handlers for the AT Protocol sync API.//!//! implements com.atproto.sync.* lexicon endpoints://! listRepos, getRepo, getRepoStatus, getLatestCommit, listReposByCollection,//! listHosts, getHostStatus, requestCrawl//!//! DB-accessing handlers use DbRequest + DbRequestQueue to route queries through//! pool_io workers.
const std = @import("std");const Io = std.Io;const pg = @import("pg");const h = @import("http.zig");const router = @import("router.zig");const event_log_mod = @import("../event_log.zig");const collection_index_mod = @import("../collection_index/index.zig");const slurper_mod = @import("../slurper.zig");
const Allocator = std.mem.Allocator;const HttpContext = router.HttpContext;const DbRequest = event_log_mod.DbRequest;const DiskPersist = event_log_mod.DiskPersist;const log = std.log.scoped(.relay);
// --- listRepos ---
const RepoListEncoder = struct { output: Io.Writer.Allocating, count: i64 = 0, last_uid: i64 = 0,
fn init(allocator: Allocator) !RepoListEncoder { var self: RepoListEncoder = .{ .output = .init(allocator) }; errdefer self.output.deinit(); try self.output.writer.writeAll("{\"repos\":["); return self; }
fn append(self: *RepoListEncoder, uid: i64, did: []const u8, head: []const u8, rev: []const u8, local_status: []const u8, upstream_status: []const u8) !void { if (head.len == 0 or rev.len == 0) return error.MissingRepoCommit; const w = &self.output.writer; if (self.count > 0) try w.writeByte(','); const local_ok = std.mem.eql(u8, local_status, "active"); const active = local_ok and std.mem.eql(u8, upstream_status, "active"); const status = if (!local_ok) local_status else upstream_status; try w.writeAll("{\"did\":\""); try w.writeAll(did); try w.writeAll("\",\"head\":\""); try w.writeAll(head); try w.writeAll("\",\"rev\":\""); try w.writeAll(rev); try w.writeAll("\""); if (active) { try w.writeAll(",\"active\":true"); } else { try w.writeAll(",\"active\":false,\"status\":\""); try w.writeAll(status); try w.writeByte('"'); } try w.writeByte('}'); self.last_uid = uid; self.count += 1; }
fn finish(self: *RepoListEncoder, limit: i64) ![]u8 { const w = &self.output.writer; try w.writeByte(']'); if (self.count >= limit) try w.print(",\"cursor\":\"{d}\"", .{self.last_uid}); try w.writeByte('}'); return self.output.toOwnedSlice(); }};
const ListReposReq = struct { base: DbRequest = .{ .callback = &execute }, cursor_val: i64, limit: i64, body: ?[]u8 = null, db_err: bool = false,
fn execute(b: *DbRequest, dp: *DiskPersist) void { const self: *@This() = @fieldParentPtr("base", b); self.body = self.render(dp) catch { self.db_err = true; return; }; }
fn render(self: *ListReposReq, dp: *DiskPersist) ![]u8 { var output = try RepoListEncoder.init(std.heap.c_allocator); defer output.output.deinit(); var result = try dp.db.query( \\SELECT a.uid, a.did, a.status, a.upstream_status, COALESCE(r.rev, ''), COALESCE(r.commit_cid, '') \\FROM account a INNER JOIN account_repo r ON a.uid = r.uid \\WHERE a.uid > $1 AND r.uid > $1 AND r.rev <> '' AND r.commit_cid <> '' \\ORDER BY a.uid ASC LIMIT $2 , .{ self.cursor_val, self.limit }); defer result.deinit(); while (try result.nextUnsafe()) |row| { try output.append(row.get(i64, 0), row.get([]const u8, 1), row.get([]const u8, 5), row.get([]const u8, 4), row.get([]const u8, 2), row.get([]const u8, 3)); } return output.finish(self.limit); }};
pub fn handleListRepos(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void { const cursor_str = h.queryParam(query, "cursor") orelse "0"; const limit_str = h.queryParam(query, "limit") orelse "500";
const cursor_val = std.fmt.parseInt(i64, cursor_str, 10) catch { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid cursor\"}"); return; }; if (cursor_val < 0) { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"cursor must be >= 0\"}"); return; }
const limit = std.fmt.parseInt(i64, limit_str, 10) catch { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}"); return; }; if (limit < 1 or limit > 1000) { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..1000\"}"); return; }
var req: ListReposReq = .{ .cursor_val = cursor_val, .limit = limit }; defer if (req.body) |body| std.heap.c_allocator.free(body); ctx.db_queue.push(&req.base); req.base.wait(ctx.io, ctx.shutdown);
if (req.base.err != null) { h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}"); return; } if (req.db_err) { h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query or response encoding failed\"}"); return; }
h.respondJson(conn, .ok, req.body.?);}
test "listRepos renders full default and maximum pages beyond 64 KiB" { const allocator = std.testing.allocator; for ([_]i64{ 500, 1000 }) |limit| { var output = try RepoListEncoder.init(allocator); defer output.output.deinit(); for (0..@intCast(limit)) |i| { try output.append(@intCast(i + 1), "did:plc:2222cxntd4c3qttyd7smldsp", "bafyreigvbgb2nn5d7zgiiq3ziot25wfsximdjax2oetzbpgzmobnrjqh7u", "3my46tkfps52u", "active", if (i == 0) "deactivated" else "active"); } const body = try output.finish(limit); defer allocator.free(body); try std.testing.expect(body.len > 65536); const parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); defer parsed.deinit(); const repos = parsed.value.object.get("repos").?.array.items; try std.testing.expectEqual(@as(usize, @intCast(limit)), repos.len); try std.testing.expect(!repos[0].object.get("active").?.bool); try std.testing.expectEqualStrings("deactivated", repos[0].object.get("status").?.string); try std.testing.expect(repos[1].object.get("active").?.bool); try std.testing.expectEqual(limit, try std.fmt.parseInt(i64, parsed.value.object.get("cursor").?.string, 10)); }}
test "listRepos keeps a cursor for a one-repo page and omits it at end" { const allocator = std.testing.allocator; var output = try RepoListEncoder.init(allocator); defer output.output.deinit(); try output.append(42, "did:plc:2222cxntd4c3qttyd7smldsp", "bafyreigvbgb2nn5d7zgiiq3ziot25wfsximdjax2oetzbpgzmobnrjqh7u", "3my46tkfps52u", "taken_down", "active"); const body = try output.finish(1); defer allocator.free(body); const parsed = try std.json.parseFromSlice(std.json.Value, allocator, body, .{}); defer parsed.deinit(); try std.testing.expectEqualStrings("42", parsed.value.object.get("cursor").?.string); try std.testing.expectEqualStrings("taken_down", parsed.value.object.get("repos").?.array.items[0].object.get("status").?.string); var empty = try RepoListEncoder.init(allocator); defer empty.output.deinit(); const end = try empty.finish(1); defer allocator.free(end); try std.testing.expectEqualStrings("{\"repos\":[]}", end);}
fn encodeRepoPageWithAllocationFailures(allocator: Allocator) !void { encodeRepoPage(allocator) catch |err| switch (err) { error.WriteFailed => return error.OutOfMemory, else => return err, };}
test "listRepos rejects incomplete repository commits without emitting a row" { var output = try RepoListEncoder.init(std.testing.allocator); defer output.output.deinit(); try std.testing.expectError(error.MissingRepoCommit, output.append(1, "did:plc:2222cxntd4c3qttyd7smldsp", "", "3my46tkfps52u", "active", "active")); try std.testing.expectError(error.MissingRepoCommit, output.append(2, "did:plc:2222cxntd4c3qttyd7smldsp", "bafyreigvbgb2nn5d7zgiiq3ziot25wfsximdjax2oetzbpgzmobnrjqh7u", "", "active", "active")); const body = try output.finish(500); defer std.testing.allocator.free(body); try std.testing.expectEqualStrings("{\"repos\":[]}", body);}
fn encodeRepoPage(allocator: Allocator) !void { var output = try RepoListEncoder.init(allocator); defer output.output.deinit(); for (0..1000) |i| try output.append(@intCast(i + 1), "did:plc:2222cxntd4c3qttyd7smldsp", "bafyreigvbgb2nn5d7zgiiq3ziot25wfsximdjax2oetzbpgzmobnrjqh7u", "3my46tkfps52u", "active", "active"); const body = try output.finish(1000); defer allocator.free(body);}
test "listRepos releases partial responses on allocation failure" { try std.testing.checkAllAllocationFailures(std.testing.allocator, encodeRepoPageWithAllocationFailures, .{});}
// --- getRepoStatus ---
const GetRepoStatusReq = struct { base: DbRequest = .{ .callback = &execute }, did_buf: [256]u8 = undefined, did_len: usize = 0, json_buf: [4096]u8 = undefined, json_len: usize = 0, not_found: bool = false, db_err: bool = false,
fn execute(b: *DbRequest, dp: *DiskPersist) void { const self: *@This() = @fieldParentPtr("base", b); const did = self.did_buf[0..self.did_len];
var row = (dp.db.rowUnsafe( "SELECT a.uid, a.status, a.upstream_status, COALESCE(r.rev, '') FROM account a LEFT JOIN account_repo r ON a.uid = r.uid WHERE a.did = $1", .{did}, ) catch { self.db_err = true; return; }) orelse { self.not_found = true; return; }; defer row.deinit() catch {};
const local_status = row.get([]const u8, 1); const upstream_status = row.get([]const u8, 2); const rev = row.get([]const u8, 3); const local_ok = std.mem.eql(u8, local_status, "active"); const upstream_ok = std.mem.eql(u8, upstream_status, "active"); const active = local_ok and upstream_ok; const status = if (!local_ok) local_status else upstream_status;
var w: Io.Writer = .fixed(&self.json_buf); w.writeAll("{\"did\":\"") catch return; w.writeAll(did) catch return; w.writeAll("\"") catch return;
if (active) { w.writeAll(",\"active\":true") catch return; } else { w.writeAll(",\"active\":false,\"status\":\"") catch return; w.writeAll(status) catch return; w.writeAll("\"") catch return; }
if (rev.len > 0) { w.writeAll(",\"rev\":\"") catch return; w.writeAll(rev) catch return; w.writeAll("\"") catch return; }
w.writeByte('}') catch return; self.json_len = w.end; }};
pub fn handleGetRepoStatus(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void { var did_buf: [256]u8 = undefined; const did = h.queryParamDecoded(query, "did", &did_buf) orelse { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}"); return; }; if (!std.mem.startsWith(u8, did, "did:")) { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}"); return; }
var req: GetRepoStatusReq = .{}; @memcpy(req.did_buf[0..did.len], did); req.did_len = did.len; ctx.db_queue.push(&req.base); req.base.wait(ctx.io, ctx.shutdown);
if (req.base.err != null) { h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}"); return; } if (req.db_err) { h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); return; } if (req.not_found) { h.respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}"); return; }
h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);}
// --- getRepo ---
const GetRepoReq = struct { base: DbRequest = .{ .callback = &execute }, did_buf: [256]u8 = undefined, did_len: usize = 0, url_buf: [512]u8 = undefined, url_len: usize = 0, not_found: bool = false, db_err: bool = false,
fn execute(b: *DbRequest, dp: *DiskPersist) void { const self: *@This() = @fieldParentPtr("base", b); const did = self.did_buf[0..self.did_len];
var row = (dp.db.rowUnsafe( "SELECT h.hostname FROM account a JOIN host h ON a.host_id = h.id WHERE a.did = $1 AND a.host_id > 0", .{did}, ) catch { self.db_err = true; return; }) orelse { self.not_found = true; return; }; defer row.deinit() catch {};
const hostname = row.get([]const u8, 0); const url = std.fmt.bufPrint(&self.url_buf, "https://{s}/xrpc/com.atproto.sync.getRepo?did={s}", .{ hostname, did }) catch return; self.url_len = url.len; }};
pub fn handleGetRepo(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void { var did_buf: [256]u8 = undefined; const did = h.queryParamDecoded(query, "did", &did_buf) orelse { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}"); return; }; if (!std.mem.startsWith(u8, did, "did:")) { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}"); return; }
var req: GetRepoReq = .{}; @memcpy(req.did_buf[0..did.len], did); req.did_len = did.len; ctx.db_queue.push(&req.base); req.base.wait(ctx.io, ctx.shutdown);
if (req.base.err != null) { h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}"); return; } if (req.db_err) { h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); return; } if (req.not_found) { h.respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}"); return; }
h.respondRedirect(conn, req.url_buf[0..req.url_len]);}
// --- getLatestCommit ---
const GetLatestCommitReq = struct { base: DbRequest = .{ .callback = &execute }, did_buf: [256]u8 = undefined, did_len: usize = 0, json_buf: [4096]u8 = undefined, json_len: usize = 0, not_found: bool = false, db_err: bool = false, status_err: ?[]const u8 = null,
fn execute(b: *DbRequest, dp: *DiskPersist) void { const self: *@This() = @fieldParentPtr("base", b); const did = self.did_buf[0..self.did_len];
var row = (dp.db.rowUnsafe( "SELECT a.status, a.upstream_status, COALESCE(r.rev, ''), COALESCE(r.commit_cid, '') FROM account a LEFT JOIN account_repo r ON a.uid = r.uid WHERE a.did = $1", .{did}, ) catch { self.db_err = true; return; }) orelse { self.not_found = true; return; }; defer row.deinit() catch {};
const local_status = row.get([]const u8, 0); const upstream_status = row.get([]const u8, 1); const rev = row.get([]const u8, 2); const cid = row.get([]const u8, 3);
const status = if (!std.mem.eql(u8, local_status, "active")) local_status else upstream_status;
if (std.mem.eql(u8, status, "takendown") or std.mem.eql(u8, status, "suspended")) { self.status_err = "{\"error\":\"RepoTakendown\",\"message\":\"account has been taken down\"}"; return; } else if (std.mem.eql(u8, status, "deactivated")) { self.status_err = "{\"error\":\"RepoDeactivated\",\"message\":\"account is deactivated\"}"; return; } else if (std.mem.eql(u8, status, "deleted")) { self.status_err = "{\"error\":\"RepoDeleted\",\"message\":\"account is deleted\"}"; return; } else if (!std.mem.eql(u8, status, "active")) { self.status_err = "{\"error\":\"RepoInactive\",\"message\":\"account is not active\"}"; return; }
if (rev.len == 0 or cid.len == 0) { self.not_found = true; return; }
var w: Io.Writer = .fixed(&self.json_buf); w.writeAll("{\"cid\":\"") catch return; w.writeAll(cid) catch return; w.writeAll("\",\"rev\":\"") catch return; w.writeAll(rev) catch return; w.writeAll("\"}") catch return; self.json_len = w.end; }};
pub fn handleGetLatestCommit(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void { var did_buf: [256]u8 = undefined; const did = h.queryParamDecoded(query, "did", &did_buf) orelse { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}"); return; }; if (!std.mem.startsWith(u8, did, "did:")) { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}"); return; }
var req: GetLatestCommitReq = .{}; @memcpy(req.did_buf[0..did.len], did); req.did_len = did.len; ctx.db_queue.push(&req.base); req.base.wait(ctx.io, ctx.shutdown);
if (req.base.err != null) { h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}"); return; } if (req.db_err) { h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); return; } if (req.status_err) |err_body| { h.respondJson(conn, .forbidden, err_body); return; } if (req.not_found) { if (req.json_len == 0) { h.respondJson(conn, .not_found, "{\"error\":\"RepoNotSynchronized\",\"message\":\"relay has no repo data for this account\"}"); } else { h.respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}"); } return; }
h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);}
// --- listReposByCollection (no DB, uses collection index directly) ---
pub fn handleListReposByCollection(conn: *h.Conn, query: []const u8, ci: *collection_index_mod.CollectionIndex) void { const collection = h.queryParam(query, "collection") orelse { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"collection parameter required\"}"); return; };
if (collection.len == 0 or !std.mem.containsAtLeast(u8, collection, 1, ".")) { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid collection NSID\"}"); return; }
const limit_str = h.queryParam(query, "limit") orelse "500"; const limit = std.fmt.parseInt(usize, limit_str, 10) catch { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}"); return; }; if (limit < 1 or limit > 2000) { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..2000\"}"); return; }
var cursor_buf: [256]u8 = undefined; const cursor_did = h.queryParamDecoded(query, "cursor", &cursor_buf);
var did_buf: [65536]u8 = undefined; var did_offsets: [2000]collection_index_mod.Span = undefined; const ci_result = ci.listReposByCollection(collection, limit, cursor_did, &did_buf, &did_offsets) catch { h.respondJson(conn, .internal_server_error, "{\"error\":\"InternalError\",\"message\":\"index scan failed\"}"); return; };
const body = renderListReposByCollection(std.heap.c_allocator, ci_result, limit) catch { h.respondJson(conn, .internal_server_error, "{\"error\":\"InternalError\",\"message\":\"response encoding failed\"}"); return; }; defer std.heap.c_allocator.free(body); h.respondJson(conn, .ok, body);}
fn renderListReposByCollection(allocator: Allocator, result: collection_index_mod.ListResult, limit: usize) ![]u8 { var aw: Io.Writer.Allocating = .init(allocator); errdefer aw.deinit(); const w = &aw.writer;
try w.writeAll("{\"repos\":["); for (0..result.count) |i| { if (i > 0) try w.writeByte(','); try w.writeAll("{\"did\":\""); try w.writeAll(result.getDid(i)); try w.writeAll("\"}"); } try w.writeByte(']');
if (result.last_did) |last| { if (result.count >= limit) { try w.writeAll(",\"cursor\":\""); try w.writeAll(last); try w.writeByte('"'); } }
try w.writeByte('}'); return aw.toOwnedSlice();}
test "listReposByCollection renders paginated RocksDB results" { const alloc = std.testing.allocator; var dir = std.testing.tmpDir(.{}); defer dir.cleanup(); const path = try dir.dir.realPathFileAlloc(std.testing.io, ".", alloc); defer alloc.free(path); var ci = try collection_index_mod.CollectionIndex.open(alloc, path); defer ci.deinit(); const collection = "app.bsky.feed.post"; const dids = [_][]const u8{ "did:plc:2222bxcsksmtrafdqh7hgvxo", "did:web:example.com", "did:web:example.org:users:alice" }; for (dids) |did| try ci.addCollection(did, collection); try ci.addCollection("did:web:other.example", "app.bsky.feed.like");
for ([_]usize{ 1, 2, 2000 }) |limit| { var cursor: ?[]const u8 = null; var seen: usize = 0; var cursor_buf: [256]u8 = undefined; while (true) { var buf: [65536]u8 = undefined; var offsets: [2000]collection_index_mod.Span = undefined; const result = try @call(.never_inline, collection_index_mod.CollectionIndex.listReposByCollection, .{ &ci, collection, limit, cursor, &buf, &offsets }); var other_buf: [256]u8 = undefined; var other_offsets: [1]collection_index_mod.Span = undefined; const other = try @call(.never_inline, collection_index_mod.CollectionIndex.listReposByCollection, .{ &ci, "app.bsky.feed.like", 1, null, &other_buf, &other_offsets }); try std.testing.expectEqual(@as(usize, 1), other.count); const body = try renderListReposByCollection(alloc, result, limit); defer alloc.free(body); const parsed = try std.json.parseFromSlice(std.json.Value, alloc, body, .{}); defer parsed.deinit(); for (parsed.value.object.get("repos").?.array.items) |repo| { try std.testing.expect(seen < dids.len); try std.testing.expectEqualStrings(dids[seen], repo.object.get("did").?.string); seen += 1; } if (parsed.value.object.get("cursor")) |next| { @memcpy(cursor_buf[0..next.string.len], next.string); cursor = cursor_buf[0..next.string.len]; } else break; } try std.testing.expectEqual(dids.len, seen); }}
test "listReposByCollection renders a full page at the 2000 limit" { const alloc = std.testing.allocator; const did = "did:plc:2222cxntd4c3qttyd7smldsp"; const limit = 2000;
const did_buf = try alloc.alloc(u8, did.len * limit); defer alloc.free(did_buf); var offsets: [limit]collection_index_mod.Span = undefined; for (0..limit) |i| { @memcpy(did_buf[i * did.len ..][0..did.len], did); offsets[i] = .{ .start = i * did.len, .len = did.len }; } const result: collection_index_mod.ListResult = .{ .count = limit, .offsets = &offsets, .buf = did_buf, .last_did = did_buf[(limit - 1) * did.len ..][0..did.len], };
const body = try renderListReposByCollection(alloc, result, limit); defer alloc.free(body);
try std.testing.expect(body.len > 65536); const parsed = try std.json.parseFromSlice(std.json.Value, alloc, body, .{}); defer parsed.deinit(); try std.testing.expectEqual(@as(usize, limit), parsed.value.object.get("repos").?.array.items.len); try std.testing.expectEqualStrings(did, parsed.value.object.get("cursor").?.string);}
// --- listHosts ---
const ListHostsReq = struct { base: DbRequest = .{ .callback = &execute }, cursor_val: i64, limit: i64, json_buf: [65536]u8 = undefined, json_len: usize = 0, db_err: bool = false,
fn execute(b: *DbRequest, dp: *DiskPersist) void { const self: *@This() = @fieldParentPtr("base", b); var w: Io.Writer = .fixed(&self.json_buf);
var result = dp.db.query( "SELECT id, hostname, status, last_seq FROM host WHERE id > $1 AND last_seq > 0 ORDER BY id ASC LIMIT $2", .{ self.cursor_val, self.limit }, ) catch { self.db_err = true; return; }; defer result.deinit();
var count: i64 = 0; var last_id: i64 = 0;
w.writeAll("{\"hosts\":[") catch return;
while (result.nextUnsafe() catch null) |row| { if (count > 0) w.writeByte(',') catch return;
const id = row.get(i64, 0); const hostname = row.get([]const u8, 1); const status = row.get([]const u8, 2); const seq = row.get(i64, 3);
w.writeAll("{\"hostname\":\"") catch return; w.writeAll(hostname) catch return; w.writeAll("\"") catch return; w.print(",\"seq\":{d}", .{seq}) catch return; w.writeAll(",\"status\":\"") catch return; w.writeAll(status) catch return; w.writeAll("\"}") catch return;
last_id = id; count += 1; }
w.writeByte(']') catch return; if (count >= self.limit and count > 1) { w.print(",\"cursor\":\"{d}\"", .{last_id}) catch return; } w.writeByte('}') catch return; self.json_len = w.end; }};
pub fn handleListHosts(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void { const cursor_str = h.queryParam(query, "cursor") orelse "0"; const limit_str = h.queryParam(query, "limit") orelse "200";
const cursor_val = std.fmt.parseInt(i64, cursor_str, 10) catch { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid cursor\"}"); return; }; if (cursor_val < 0) { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"cursor must be >= 0\"}"); return; }
const limit = std.fmt.parseInt(i64, limit_str, 10) catch { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}"); return; }; if (limit < 1 or limit > 1000) { h.respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..1000\"}"); return; }
var req: ListHostsReq = .{ .cursor_val = cursor_val, .limit = limit }; ctx.db_queue.push(&req.base); req.base.wait(ctx.io, ctx.shutdown);
if (req.base.err != null) { h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}"); return; } if (req.db_err) { h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); return; }
h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);}
// --- getHostStatus ---
const GetHostStatusReq = struct { base: DbRequest = .{ .callback = &execute }, hostname_buf: [256]u8 = undefined, hostname_len: usize = 0, json_buf: [4096]u8 = undefined, json_len: usize = 0, not_found: bool = false, db_err: bool = false,
fn execute(b: *DbRequest, dp: *DiskPersist) void { const self: *@This() = @fieldParentPtr("base", b); const hostname = self.hostname_buf[0..self.hostname_len];
var row = (dp.db.rowUnsafe( "SELECT id, hostname, status, last_seq FROM host WHERE hostname = $1", .{hostname}, ) catch { self.db_err = true; return; }) orelse { self.not_found = true; return; }; defer row.deinit() catch {};
const host_id = row.get(i64, 0); const host_name = row.get([]const u8, 1); const raw_status = row.get([]const u8, 2); const seq = row.get(i64, 3);
const status = if (std.mem.eql(u8, raw_status, "blocked")) "banned" else if (std.mem.eql(u8, raw_status, "exhausted")) "offline" else raw_status;
// count accounts on this host const account_count: i64 = if (dp.db.rowUnsafe( "SELECT COUNT(*) FROM account WHERE host_id = $1", .{host_id}, ) catch null) |cnt_row| blk: { var r = cnt_row; defer r.deinit() catch {}; break :blk r.get(i64, 0); } else 0;
var w: Io.Writer = .fixed(&self.json_buf); w.writeAll("{\"hostname\":\"") catch return; w.writeAll(host_name) catch return; w.writeAll("\"") catch return; w.print(",\"seq\":{d},\"accountCount\":{d}", .{ seq, account_count }) catch return; w.writeAll(",\"status\":\"") catch return; w.writeAll(status) catch return; w.writeAll("\"}") catch return; self.json_len = w.end; }};
pub fn handleGetHostStatus(conn: *h.Conn, query: []const u8, ctx: *HttpContext) void { var hostname_buf: [256]u8 = undefined; const hostname = h.queryParamDecoded(query, "hostname", &hostname_buf) orelse { h.respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"hostname parameter required\"}"); return; };
var req: GetHostStatusReq = .{}; @memcpy(req.hostname_buf[0..hostname.len], hostname); req.hostname_len = hostname.len; ctx.db_queue.push(&req.base); req.base.wait(ctx.io, ctx.shutdown);
if (req.base.err != null) { h.respondJson(conn, .service_unavailable, "{\"error\":\"ServiceUnavailable\",\"message\":\"database unavailable\"}"); return; } if (req.db_err) { h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); return; } if (req.not_found) { h.respondJson(conn, .not_found, "{\"error\":\"HostNotFound\",\"message\":\"host not found\"}"); return; }
h.respondJson(conn, .ok, req.json_buf[0..req.json_len]);}
// --- requestCrawl ---
pub fn handleRequestCrawl(conn: *h.Conn, body: []const u8, ctx: *HttpContext) void { // log on entry so we can tell from logs whether requests are being received // even if downstream waits wedge. body is small (~32 bytes) so this is cheap. log.info("requestCrawl received (body_len={d})", .{body.len});
const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, ctx.persist.allocator, body, .{ .ignore_unknown_fields = true }) catch { h.respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"invalid JSON, expected {\\\"hostname\\\":\\\"...\\\"}\"}"); return; }; defer parsed.deinit();
// fast validation: hostname format const hostname = slurper_mod.validateHostname(ctx.persist.allocator, parsed.value.hostname) catch |err| { log.warn("requestCrawl rejected '{s}': {s}", .{ parsed.value.hostname, @errorName(err) }); h.respondJson(conn, .bad_request, switch (err) { error.EmptyHostname => "{\"error\":\"InvalidRequest\",\"message\":\"empty hostname\"}", error.InvalidCharacter => "{\"error\":\"InvalidRequest\",\"message\":\"hostname contains invalid characters\"}", error.InvalidLabel => "{\"error\":\"InvalidRequest\",\"message\":\"hostname has invalid label\"}", error.TooFewLabels => "{\"error\":\"InvalidRequest\",\"message\":\"hostname must have at least two labels (e.g. pds.example.com)\"}", error.LooksLikeIpAddress => "{\"error\":\"InvalidRequest\",\"message\":\"IP addresses not allowed, use a hostname\"}", error.PortNotAllowed => "{\"error\":\"InvalidRequest\",\"message\":\"port numbers not allowed\"}", error.LocalhostNotAllowed => "{\"error\":\"InvalidRequest\",\"message\":\"localhost not allowed\"}", else => "{\"error\":\"InvalidRequest\",\"message\":\"invalid hostname\"}", }); return; }; defer ctx.persist.allocator.free(hostname);
// domain ban check via DbRequestQueue const DomainBanReq = struct { base_req: DbRequest = .{ .callback = &execute }, hostname_buf: [256]u8 = undefined, hostname_len: usize = 0, banned: bool = false,
fn execute(b: *DbRequest, dp: *DiskPersist) void { const self: *@This() = @fieldParentPtr("base_req", b); self.banned = dp.isDomainBanned(self.hostname_buf[0..self.hostname_len]); } }; var ban_req: DomainBanReq = .{}; const copy_len = @min(hostname.len, ban_req.hostname_buf.len); @memcpy(ban_req.hostname_buf[0..copy_len], hostname[0..copy_len]); ban_req.hostname_len = copy_len; ctx.db_queue.push(&ban_req.base_req); ban_req.base_req.wait(ctx.io, ctx.shutdown);
if (ban_req.banned) { log.warn("requestCrawl rejected '{s}': domain banned", .{hostname}); h.respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"domain is banned\"}"); return; }
// enqueue for async processing ctx.slurper.addCrawlRequest(hostname) catch { h.respondJson(conn, .internal_server_error, "{\"error\":\"failed to store crawl request\"}"); return; };
log.info("crawl requested: {s}", .{hostname}); h.respondJson(conn, .ok, "{\"success\":true}");}
test "repo endpoints return the signed commit CID and omit legacy roots" { const base_url = @import("../util/util.zig").getenv("DATABASE_URL") orelse return error.SkipZigTest; var tdb = try event_log_mod.TestDb.create(base_url); defer tdb.destroy(); var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup(); var path_buf: [Io.Dir.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(std.testing.io, &path_buf); var dp = try DiskPersist.init(std.testing.allocator, path_buf[0..path_len], tdb.url, 5, std.testing.io); defer dp.deinit(); var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const a = arena.allocator(); const zat = @import("zat"); const did = "did:plc:2222cxntd4c3qttyd7smldsp"; const rev = "3my46tkfps52u"; const kp = try zat.Keypair.fromSecretKey(.p256, @splat(1)); var mst = zat.mst.Mst.init(a); const root = try mst.rootCid(); const signed = try zat.signCommit(a, .{ .did = did, .rev = rev, .data = root }, &kp); const car = try zat.car.writeAlloc(a, .{ .roots = &.{signed.cid}, .blocks = &.{.{ .cid_raw = signed.cid.raw, .data = signed.bytes }} }); const verified = try zat.verifyCommitCar(a, car, .{ .key_type = .p256, .raw = &(try kp.publicKey()) }, .{ .verify_mst = false, .expected_did = did }); const commit_cid = try zat.multibase.encode(a, .base32lower, verified.commit_cid); const data_cid = try zat.multibase.encode(a, .base32lower, root.raw); try std.testing.expect(!std.mem.eql(u8, commit_cid, data_cid));
const legacy_did = "did:plc:legacy"; const legacy_uid = try dp.uidForDid(legacy_did); _ = try dp.db.exec("INSERT INTO account_repo (uid, rev, commit_data_cid) VALUES ($1, $2, $3)", .{ @as(i64, @intCast(legacy_uid)), rev, data_cid }); const uid = try dp.uidForDid(did); try std.testing.expect(try dp.updateAccountState(uid, rev, data_cid, commit_cid));
var list: ListReposReq = .{ .cursor_val = 0, .limit = 1 }; const body = try list.render(&dp); defer std.heap.c_allocator.free(body); const parsed = try std.json.parseFromSlice(std.json.Value, a, body, .{}); const repos = parsed.value.object.get("repos").?.array.items; try std.testing.expectEqual(@as(usize, 1), repos.len); try std.testing.expectEqualStrings(did, repos[0].object.get("did").?.string); try std.testing.expectEqualStrings(commit_cid, repos[0].object.get("head").?.string); try std.testing.expectEqualStrings(rev, repos[0].object.get("rev").?.string); list.cursor_val = @intCast(uid); const end = try list.render(&dp); defer std.heap.c_allocator.free(end); try std.testing.expectEqualStrings("{\"repos\":[]}", end);
var latest: GetLatestCommitReq = .{}; @memcpy(latest.did_buf[0..did.len], did); latest.did_len = did.len; GetLatestCommitReq.execute(&latest.base, &dp); try std.testing.expect(!latest.db_err and !latest.not_found); const head = try std.json.parseFromSlice(std.json.Value, a, latest.json_buf[0..latest.json_len], .{}); try std.testing.expectEqualStrings(commit_cid, head.value.object.get("cid").?.string); try std.testing.expectEqualStrings(rev, head.value.object.get("rev").?.string);
latest = .{}; @memcpy(latest.did_buf[0..legacy_did.len], legacy_did); latest.did_len = legacy_did.len; GetLatestCommitReq.execute(&latest.base, &dp); try std.testing.expect(latest.not_found and !latest.db_err);}