diff --git a/README.md b/README.md index a34287f..571d8ec 100644 --- a/README.md +++ b/README.md @@ -8,7 +8,7 @@ an [AT Protocol](https://atproto.com/) relay in zig. subscribes to every PDS on - **direct PDS crawl** — the bootstrap relay is called once at startup for the host list via `listHosts`, then all data flows directly from each PDS. -- **optimistic signature validation** — on signing key cache miss, the frame passes through immediately and the DID is queued for background resolution. all subsequent commits are verified against the cached key. the cache caps at a configurable size and evicts the oldest 10% by resolve time when full. +- **optimistic signature validation** — on signing key cache miss, the frame passes through immediately and the DID is queued for background resolution. all subsequent commits are verified against the cached key. the cache caps at a configurable size and evicts the least recently used entry when full. - **inline collection index** — indexes `(DID, collection)` pairs in the event processing pipeline using RocksDB. serves `listReposByCollection` from the relay process — no sidecar. the index design draws on [fig](https://tangled.org/microcosm.blue)'s work on [lightrail](https://tangled.org/microcosm.blue/lightrail). @@ -16,7 +16,9 @@ an [AT Protocol](https://atproto.com/) relay in zig. subscribes to every PDS on ## spec compliance -implements the [AT Protocol sync spec](https://atproto.com/specs/sync) — `subscribeRepos`, `listRepos`, `getRepoStatus`, `getLatestCommit`, `listReposByCollection`, `listHosts`, `getHostStatus`, and `requestCrawl`. +implements the relay endpoints from the [AT Protocol sync spec](https://atproto.com/specs/sync): `subscribeRepos`, `listRepos`, `getRepoStatus`, `listHosts`, `getHostStatus`, and `requestCrawl`. + +also implements `getLatestCommit`, `listReposByCollection`, and `getRepo` (302 redirect to PDS) for compatibility with the [indigo relay](https://github.com/bluesky-social/indigo) reference implementation. ## dependencies diff --git a/src/api/http.zig b/src/api/http.zig index 643d716..deb8145 100644 --- a/src/api/http.zig +++ b/src/api/http.zig @@ -28,6 +28,12 @@ pub fn respondText(conn: *Conn, status: http.Status, body: []const u8) void { httpRespond(conn, status, "text/plain", body); } +pub fn respondRedirect(conn: *Conn, location: []const u8) void { + var buf: [1024]u8 = undefined; + const header = std.fmt.bufPrint(&buf, "HTTP/1.1 302 Found\r\nLocation: {s}\r\nContent-Length: 0\r\nConnection: close\r\nServer: zlay\r\n\r\n", .{location}) catch return; + conn.writeFramed(header) catch return; +} + pub fn httpStatusLine(status: http.Status) []const u8 { return switch (status) { .ok => "200 OK", diff --git a/src/api/router.zig b/src/api/router.zig index 38a0cd9..6ea7330 100644 --- a/src/api/router.zig +++ b/src/api/router.zig @@ -73,6 +73,8 @@ fn handleGet(conn: *websocket.Conn, path: []const u8, query: []const u8, headers xrpc.handleListRepos(conn, query, ctx.persist); } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getRepoStatus")) { xrpc.handleGetRepoStatus(conn, query, ctx.persist); + } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getRepo")) { + xrpc.handleGetRepo(conn, query, ctx.persist); } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getLatestCommit")) { xrpc.handleGetLatestCommit(conn, query, ctx.persist); } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listReposByCollection")) { diff --git a/src/api/xrpc.zig b/src/api/xrpc.zig index 08282c5..02c8b72 100644 --- a/src/api/xrpc.zig +++ b/src/api/xrpc.zig @@ -1,7 +1,7 @@ //! XRPC endpoint handlers for the AT Protocol sync API. //! //! implements com.atproto.sync.* lexicon endpoints: -//! listRepos, getRepoStatus, getLatestCommit, listReposByCollection, +//! listRepos, getRepo, getRepoStatus, getLatestCommit, listReposByCollection, //! listHosts, getHostStatus, requestCrawl const std = @import("std"); @@ -169,6 +169,40 @@ pub fn handleGetRepoStatus(conn: *h.Conn, query: []const u8, persist: *event_log h.respondJson(conn, .ok, fbs.getWritten()); } +pub fn handleGetRepo(conn: *h.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) 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; + } + + // look up the PDS hostname for this account + var row = (persist.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 { + h.respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); + return; + }) orelse { + h.respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}"); + return; + }; + defer row.deinit() catch {}; + + const hostname = row.get([]const u8, 0); + + // build redirect URL: https://{hostname}/xrpc/com.atproto.sync.getRepo?did={did} + var url_buf: [512]u8 = undefined; + const url = std.fmt.bufPrint(&url_buf, "https://{s}/xrpc/com.atproto.sync.getRepo?did={s}", .{ hostname, did }) catch return; + + h.respondRedirect(conn, url); +} + pub fn handleGetLatestCommit(conn: *h.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void { var did_buf: [256]u8 = undefined; const did = h.queryParamDecoded(query, "did", &did_buf) orelse {