diff --git a/src/main.zig b/src/main.zig index c5fc007..8862643 100644 --- a/src/main.zig +++ b/src/main.zig @@ -5,9 +5,12 @@ //! numbers, and rebroadcasts to downstream consumers over WebSocket. //! //! endpoints: -//! /xrpc/com.atproto.sync.subscribeRepos — firehose WebSocket (supports ?cursor=N) -//! /_health — HTTP 200 JSON health check -//! /_stats — HTTP 200 JSON with relay statistics +//! /xrpc/com.atproto.sync.subscribeRepos — firehose WebSocket (supports ?cursor=N) +//! /xrpc/com.atproto.sync.listRepos — paginated account listing +//! /xrpc/com.atproto.sync.getRepoStatus — single account status +//! /xrpc/com.atproto.sync.getLatestCommit — latest commit CID + rev +//! /xrpc/com.atproto.sync.requestCrawl — request PDS crawl (POST) +//! /_health, /_stats, /metrics — health, stats, prometheus const std = @import("std"); const websocket = @import("websocket"); @@ -214,7 +217,7 @@ fn handleHttpConn(stream: std.net.Stream, stats: *broadcaster.Stats, persist: *e const body: []const u8 = if (header_end) |he| request[he + 4 ..] else ""; if (std.mem.eql(u8, method, "GET")) { - handleGet(stream, path, stats); + handleGet(stream, path, stats, persist); } else if (std.mem.eql(u8, method, "POST")) { handlePost(stream, path, request[0 .. header_end orelse n], body, persist, sub); } else { @@ -222,7 +225,12 @@ fn handleHttpConn(stream: std.net.Stream, stats: *broadcaster.Stats, persist: *e } } -fn handleGet(stream: std.net.Stream, path: []const u8, stats: *broadcaster.Stats) void { +fn handleGet(stream: std.net.Stream, full_path: []const u8, stats: *broadcaster.Stats, persist: *event_log_mod.DiskPersist) void { + // split path from query string + const qmark = std.mem.indexOfScalar(u8, full_path, '?'); + const path = full_path[0..(qmark orelse full_path.len)]; + const query = if (qmark) |q| full_path[q + 1 ..] else ""; + if (std.mem.eql(u8, path, "/_health") or std.mem.eql(u8, path, "/xrpc/_health")) { httpRespond(stream, "200 OK", "application/json", "{\"status\":\"ok\"}"); } else if (std.mem.eql(u8, path, "/_stats")) { @@ -235,6 +243,12 @@ fn handleGet(stream: std.net.Stream, path: []const u8, stats: *broadcaster.Stats var resp_buf: [8192]u8 = undefined; const response = std.fmt.bufPrint(&resp_buf, "HTTP/1.1 200 OK\r\nContent-Type: text/plain; version=0.0.4; charset=utf-8\r\nContent-Length: {d}\r\nConnection: close\r\n\r\n{s}", .{ body.len, body }) catch return; _ = stream.write(response) catch {}; + } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listRepos")) { + handleListRepos(stream, query, persist); + } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getRepoStatus")) { + handleGetRepoStatus(stream, query, persist); + } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getLatestCommit")) { + handleGetLatestCommit(stream, query, persist); } else if (std.mem.eql(u8, path, "/")) { httpRespond(stream, "200 OK", "text/plain", \\ _ @@ -333,6 +347,237 @@ fn handleRequestCrawl(stream: std.net.Stream, body: []const u8, sub: *subscriber httpRespond(stream, "200 OK", "application/json", "{\"success\":true}"); } +// --- XRPC endpoint handlers --- + +fn handleListRepos(stream: std.net.Stream, query: []const u8, persist: *event_log_mod.DiskPersist) void { + const cursor_str = queryParam(query, "cursor") orelse "0"; + const limit_str = queryParam(query, "limit") orelse "500"; + + const cursor_val = std.fmt.parseInt(i64, cursor_str, 10) catch { + httpRespondJson(stream, "400 Bad Request", "{\"error\":\"BadRequest\",\"message\":\"invalid cursor\"}"); + return; + }; + if (cursor_val < 0) { + httpRespondJson(stream, "400 Bad Request", "{\"error\":\"BadRequest\",\"message\":\"cursor must be >= 0\"}"); + return; + } + + const limit = std.fmt.parseInt(i64, limit_str, 10) catch { + httpRespondJson(stream, "400 Bad Request", "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}"); + return; + }; + if (limit < 1 or limit > 1000) { + httpRespondJson(stream, "400 Bad Request", "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..1000\"}"); + return; + } + + // query accounts with repo state, paginated by UID + var result = persist.db.query( + \\SELECT a.uid, a.did, a.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 + , .{ cursor_val, limit }) catch { + httpRespondJson(stream, "500 Internal Server Error", "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); + return; + }; + defer result.deinit(); + + // build JSON response into a buffer + var buf: [65536]u8 = undefined; + var fbs = std.io.fixedBufferStream(&buf); + const w = fbs.writer(); + + 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 status = row.get([]const u8, 2); + const rev = row.get([]const u8, 3); + const head = row.get([]const u8, 4); + + const active = std.mem.eql(u8, status, "active"); + + w.writeAll("{\"did\":\"") catch return; + w.writeAll(did) catch return; + w.writeAll("\"") catch return; + + if (head.len > 0) { + w.writeAll(",\"head\":\"") catch return; + w.writeAll(head) catch return; + w.writeAll("\"") catch return; + } + if (rev.len > 0) { + 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; + + // include cursor if we got a full page + if (count >= limit and count >= 2) { + std.fmt.format(w, ",\"cursor\":\"{d}\"", .{last_uid}) catch return; + } + + w.writeByte('}') catch return; + + const body = fbs.getWritten(); + httpRespondJson(stream, "200 OK", body); +} + +fn handleGetRepoStatus(stream: std.net.Stream, query: []const u8, persist: *event_log_mod.DiskPersist) void { + const did = queryParam(query, "did") orelse { + httpRespondJson(stream, "400 Bad Request", "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}"); + return; + }; + + // basic DID syntax check + if (!std.mem.startsWith(u8, did, "did:")) { + httpRespondJson(stream, "400 Bad Request", "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}"); + return; + } + + // look up account + var row = (persist.db.rowUnsafe( + "SELECT a.uid, a.status, COALESCE(r.rev, '') FROM account a LEFT JOIN account_repo r ON a.uid = r.uid WHERE a.did = $1", + .{did}, + ) catch { + httpRespondJson(stream, "500 Internal Server Error", "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); + return; + }) orelse { + httpRespondJson(stream, "404 Not Found", "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}"); + return; + }; + defer row.deinit() catch {}; + + const status = row.get([]const u8, 1); + const rev = row.get([]const u8, 2); + const active = std.mem.eql(u8, status, "active"); + + var buf: [4096]u8 = undefined; + var fbs = std.io.fixedBufferStream(&buf); + const w = fbs.writer(); + + 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; + httpRespondJson(stream, "200 OK", fbs.getWritten()); +} + +fn handleGetLatestCommit(stream: std.net.Stream, query: []const u8, persist: *event_log_mod.DiskPersist) void { + const did = queryParam(query, "did") orelse { + httpRespondJson(stream, "400 Bad Request", "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}"); + return; + }; + + if (!std.mem.startsWith(u8, did, "did:")) { + httpRespondJson(stream, "400 Bad Request", "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}"); + return; + } + + // look up account + repo state + var row = (persist.db.rowUnsafe( + "SELECT a.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 { + httpRespondJson(stream, "500 Internal Server Error", "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); + return; + }) orelse { + httpRespondJson(stream, "404 Not Found", "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}"); + return; + }; + defer row.deinit() catch {}; + + const status = row.get([]const u8, 0); + const rev = row.get([]const u8, 1); + const cid = row.get([]const u8, 2); + + // check account status (match Go relay behavior) + if (std.mem.eql(u8, status, "takendown") or std.mem.eql(u8, status, "suspended")) { + httpRespondJson(stream, "403 Forbidden", "{\"error\":\"RepoTakendown\",\"message\":\"account has been taken down\"}"); + return; + } else if (std.mem.eql(u8, status, "deactivated")) { + httpRespondJson(stream, "403 Forbidden", "{\"error\":\"RepoDeactivated\",\"message\":\"account is deactivated\"}"); + return; + } else if (std.mem.eql(u8, status, "deleted")) { + httpRespondJson(stream, "403 Forbidden", "{\"error\":\"RepoDeleted\",\"message\":\"account is deleted\"}"); + return; + } else if (!std.mem.eql(u8, status, "active")) { + httpRespondJson(stream, "403 Forbidden", "{\"error\":\"RepoInactive\",\"message\":\"account is not active\"}"); + return; + } + + if (rev.len == 0 or cid.len == 0) { + httpRespondJson(stream, "404 Not Found", "{\"error\":\"RepoNotSynchronized\",\"message\":\"relay has no repo data for this account\"}"); + return; + } + + var buf: [4096]u8 = undefined; + var fbs = std.io.fixedBufferStream(&buf); + const w = fbs.writer(); + + w.writeAll("{\"cid\":\"") catch return; + w.writeAll(cid) catch return; + w.writeAll("\",\"rev\":\"") catch return; + w.writeAll(rev) catch return; + w.writeAll("\"}") catch return; + + httpRespondJson(stream, "200 OK", fbs.getWritten()); +} + +// --- query string helpers --- + +fn queryParam(query: []const u8, name: []const u8) ?[]const u8 { + if (query.len == 0) return null; + var iter = std.mem.splitScalar(u8, query, '&'); + while (iter.next()) |pair| { + const eq = std.mem.indexOfScalar(u8, pair, '=') orelse continue; + if (std.mem.eql(u8, pair[0..eq], name)) { + return pair[eq + 1 ..]; + } + } + return null; +} + +fn httpRespondJson(stream: std.net.Stream, status: []const u8, body: []const u8) void { + httpRespond(stream, status, "application/json", body); +} + fn findHeader(headers: []const u8, name: []const u8) ?[]const u8 { var iter = std.mem.splitScalar(u8, headers, '\n'); while (iter.next()) |line| {