atproto relay in zig zlay.waow.tech
relay zig atproto
Something went wrong. Try again.
Zig
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705//! 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 ListReposReq = 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 a.uid, a.did, a.status, a.upstream_status, COALESCE(r.rev, ''), COALESCE(r.commit_data_cid, '') \\FROM account a LEFT JOIN account_repo r ON a.uid = r.uid \\WHERE a.uid > $1 ORDER BY a.uid ASC LIMIT $2 , .{ self.cursor_val, self.limit }) catch { self.db_err = true; return; }; defer result.deinit();
var count: i64 = 0; var last_uid: i64 = 0;
w.writeAll("{\"repos\":[") catch return;
while (result.nextUnsafe() catch null) |row| { if (count > 0) w.writeByte(',') catch return;
const uid = row.get(i64, 0); const did = row.get([]const u8, 1); const local_status = row.get([]const u8, 2); const upstream_status = row.get([]const u8, 3); const rev = row.get([]const u8, 4); const head = row.get([]const u8, 5);
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;
w.writeAll("{\"did\":\"") catch return; w.writeAll(did) catch return; w.writeAll("\",\"head\":\"") catch return; w.writeAll(head) catch return; w.writeAll("\",\"rev\":\"") catch return; w.writeAll(rev) 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; }
w.writeByte('}') catch return; last_uid = uid; count += 1; }
w.writeByte(']') catch return; if (count >= self.limit and count >= 2) { w.print(",\"cursor\":\"{d}\"", .{last_uid}) catch return; } w.writeByte('}') catch return; self.json_len = w.end; }};
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 }; 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]);}
// --- 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_data_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; const ci_result = ci.listReposByCollection(collection, limit, cursor_did, &did_buf) catch { h.respondJson(conn, .internal_server_error, "{\"error\":\"InternalError\",\"message\":\"index scan failed\"}"); return; };
var buf: [65536]u8 = undefined; var w: Io.Writer = .fixed(&buf);
w.writeAll("{\"repos\":[") catch return; for (0..ci_result.count) |i| { if (i > 0) w.writeByte(',') catch return; w.writeAll("{\"did\":\"") catch return; w.writeAll(ci_result.getDid(i)) catch return; w.writeAll("\"}") catch return; } w.writeByte(']') catch return;
if (ci_result.last_did) |last| { if (ci_result.count >= limit) { w.writeAll(",\"cursor\":\"") catch return; w.writeAll(last) catch return; w.writeAll("\"") catch return; } }
w.writeByte('}') catch return; h.respondJson(conn, .ok, w.buffered());}
// --- 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}");}