diff --git a/build.zig b/build.zig index 82609f6..ed3424f 100644 --- a/build.zig +++ b/build.zig @@ -51,6 +51,7 @@ pub fn build(b: *std.Build) void { // tests const test_step = b.step("test", "run unit tests"); const test_files = .{ + "src/api.zig", "src/broadcaster.zig", "src/validator.zig", "src/subscriber.zig", diff --git a/docs/design.md b/docs/design.md new file mode 100644 index 0000000..ffb3d7e --- /dev/null +++ b/docs/design.md @@ -0,0 +1,179 @@ +# zlay — system design + +an AT Protocol relay that crawls PDS instances directly, validates commit +signatures, and rebroadcasts to downstream consumers over WebSocket. + +## data flow + +``` +PDS instances (N hosts) + │ + ▼ +Subscriber (one OS thread per host) + │ decodes CBOR frames, tracks cursor, rate-limits per host + │ resolves DID → numeric UID via postgres + ▼ +Validator + │ cache lookup: DID → signing key (secp256k1 / p256) + │ cache hit → verify commit signature (zat SDK) + │ cache miss → skip, queue background resolution + ▼ +DiskPersist + │ append to event log (28-byte LE header + CBOR payload) + │ assign relay sequence number (monotonic, relay-scoped) + │ write postgres metadata (account state, host cursor) + ▼ +Broadcaster + │ resequence frame with relay seq + │ fan out to all connected consumers via SharedFrame (ref-counted) + │ per-consumer ring buffer (8,192 frames) + write thread + ▼ +Downstream consumers (WebSocket) +``` + +additionally: +- **collection index** (RocksDB): subscriber calls `trackCommitOps` on each + validated commit; stores `(collection, did)` pairs for `listReposByCollection` +- **event log**: append-only files rotated every 10K events, 72h retention. + supports cursor replay — disk first, then in-memory ring buffer (50K frames) +- **slurper**: orchestrates subscribers. bootstraps host list from seed relay's + `listHosts` API, spawns/stops workers, processes `requestCrawl` requests + +## threading model + +| thread type | count at ~2,750 PDS | stack size | responsibility | +|---|---|---|---| +| subscriber workers | ~2,750 | 2 MB | one per host — WebSocket read loop, CBOR decode, validation call | +| resolver threads | 4–8 (env: `RESOLVER_THREADS`) | 2 MB | DID document resolution, signing key extraction, cache population | +| consumer write threads | 1 per downstream consumer | 2 MB | drain ring buffer → WebSocket write, ping/pong keepalive | +| flush thread | 1 | 2 MB | batched fsync of event log (100ms or 400 events) | +| GC thread | 1 | 2 MB | event log file cleanup every 10 minutes | +| crawl queue thread | 1 | 2 MB | process `requestCrawl` — validate hostname, describeServer, spawn worker | +| metrics server | 1 | 2 MB | HTTP on internal port, prometheus scrape | +| main thread | 1 | default | signal handling, shutdown coordination | + +total: ~2,760 + consumers. each subscriber thread runs a blocking WebSocket +read loop — simple, no async runtime, no event loop. this works because each +thread does minimal work per frame (CBOR decode + optional signature verify) +and spends most of its time blocked in `recv()`. + +the 2 MB stack size (vs zig's 16 MB default) is the key to fitting ~3K threads +in memory. actual stack usage is far below 2 MB — the deepest path is CBOR +decode → CAR parse → ECDSA verify, which uses ~50 KB of stack at peak. + +## memory model + +**allocator**: `std.heap.c_allocator` (libc malloc). glibc has per-thread +arenas and `madvise`-based page return. the general-purpose allocator (GPA) +is a debug allocator that never returns freed pages — unsuitable for +long-running servers. + +**arena per frame**: each subscriber creates a `std.heap.ArenaAllocator` per +WebSocket message. all CBOR decode temporaries, CAR parse buffers, and MST +nodes live in this arena. freed in bulk after the frame is processed. this +prevents fragmentation from per-field allocations. + +**shared frames**: the broadcaster creates one `SharedFrame` per broadcast. +consumers acquire references; the frame is freed when the last consumer +releases. this avoids copying frame bytes per consumer. + +**validator cache**: `StringHashMap(CachedKey)` — DID string → 75-byte +fixed-size struct (key type + 33-byte compressed pubkey + resolve timestamp). +capped at 500K entries (env: `VALIDATOR_CACHE_SIZE`), LRU-ish eviction of +oldest 10% when full. ~37 MB at capacity. + +**ring buffer**: 50K-entry in-memory frame history for cursor replay when +disk replay isn't available. entries are `(seq, data)` pairs with data duped +from broadcast. + +## persistence + +### event log (append-only files) + +format matches indigo's `diskpersist`: +``` +[4B flags LE] [4B kind LE] [4B payload_len LE] [8B uid LE] [8B seq LE] [payload] +``` + +files named `evts-{startSeq}`, rotated every 10K events. buffered writes +flushed every 100ms or 400 events (whichever comes first). GC deletes files +older than `RELAY_RETENTION_HOURS` (default: 72h). + +cursor replay: `playback(cursor)` binary-searches log files for the starting +seq, then streams entries forward. the broadcaster tries disk first, falls +back to in-memory ring buffer. + +### postgres + +tables: +- `account` — uid, did, status, upstream_status, host_id +- `account_repo` — uid, rev, commit_data_cid (latest repo state) +- `host` — id, hostname, status, last_seq, failed_attempts +- `log_file_refs` — seq→file mapping for cursor binary search +- `domain_ban` — banned domain suffixes +- `backfill_progress` — collection backfill cursor tracking + +connection pool: 5 connections (hardcoded in pg.zig pool init). + +### RocksDB (collection index) + +two column families: +- `rbc`: `\0` → `()` — prefix scan by collection +- `cbr`: `\0` → `()` — per-repo deletion + +populated live from firehose commits. backfill from source relay's +`listReposByCollection` for historical data. + +## scaling limits + +current deployment: ~2,780 PDS hosts, running on a 32 GB / 16 CPU node. +steady-state memory: ~3.5 GiB. postgres alongside at ~240 MiB. + +| component | current (~2,750 PDS) | at 10x (~27,500 PDS) | status | +|---|---|---|---| +| thread stacks | ~5.5 GB virtual (2,750 × 2 MB) | ~55 GB virtual | **breaks** — exceeds 32 GB node | +| pg pool | 5 connections (hardcoded) | 5 connections | **breaks** — saturates under concurrent UID lookups | +| resolver queue | unbounded `ArrayList` | unbounded | **risk** — backlog grows if resolvers can't keep up | +| validator cache | 500K entries, ~37 MB | same (capped) | **degrades** — miss rate climbs with more unique DIDs | +| broadcaster | O(n consumers) under mutex | same | **risk** — lock contention at high consumer count | +| RocksDB | manageable write rate | ~1.4M writes/sec projected | **needs** compaction tuning | +| event log | buffered, 100ms flush | fine — sequential I/O | ok | +| kernel threads | ~2,800 (below 30K default) | ~28,000 (near default max) | **breaks** without `sysctl` tuning | +| RSS | ~3.5 GiB | ~15–20 GiB projected (malloc overhead scales sublinearly) | **tight** — needs larger node | + +### what breaks first + +1. **thread count**: linux default `kernel.threads-max` is ~30K. at 27,500 + subscriber threads + resolver + consumer + system threads, we hit the wall. + virtual address space for stacks alone is ~55 GB. + +2. **postgres pool**: 5 connections shared across ~2,750 subscriber threads + works because UID lookups are fast (~0.5ms) and only happen on new DIDs. + at 10x, queue contention becomes the bottleneck — every frame touches + `uidForDidFromHost`. + +3. **validator cache miss rate**: 500K cache with 60M+ DIDs means ~99% miss + rate for cold starts. resolver threads (4–8) can't keep up with the + resolution queue at 10x ingest rate. + +## migration path + +### near-term (no architecture change) +- expose `PG_POOL_SIZE` env var, increase from 5 to 20–50 +- expose `VALIDATOR_CACHE_SIZE`, increase to 2M+ (costs ~150 MB) +- tune `RESOLVER_THREADS` to 16–32 for higher resolution throughput +- `sysctl kernel.threads-max=65536` on deploy node + +### mid-term (thread pool) +- replace one-thread-per-host with a thread pool of N workers (N = CPU cores × 2) +- each worker runs an epoll/kqueue loop over multiple host connections +- subscriber becomes a state machine: connect → read → decode → validate → persist +- reduces thread count from O(hosts) to O(cores), eliminates the stack memory wall +- websocket.zig would need a non-blocking client mode or replacement + +### long-term (async I/O) +- zig 0.16 introduces `Io` (io_uring on linux, kqueue on darwin) +- single-threaded event loop with coroutines for all I/O +- eliminates thread overhead entirely, scales to 100K+ hosts per process +- requires rewriting subscriber, resolver, and consumer write paths +- pg.zig and websocket.zig would need async-compatible forks diff --git a/src/api.zig b/src/api.zig new file mode 100644 index 0000000..c63dcfb --- /dev/null +++ b/src/api.zig @@ -0,0 +1,926 @@ +//! HTTP API handlers for the relay +//! +//! serves XRPC endpoints, admin endpoints, health/stats, and the root banner +//! via the websocket server's httpFallback mechanism. all handlers write raw +//! HTTP responses to the websocket connection. + +const std = @import("std"); +const http = std.http; +const websocket = @import("websocket"); +const broadcaster = @import("broadcaster.zig"); +const validator_mod = @import("validator.zig"); +const slurper_mod = @import("slurper.zig"); +const event_log_mod = @import("event_log.zig"); +const collection_index_mod = @import("collection_index.zig"); +const backfill_mod = @import("backfill.zig"); + +const log = std.log.scoped(.relay); + +/// context for HTTP fallback handlers (passed as opaque pointer through broadcaster) +pub const HttpContext = struct { + stats: *broadcaster.Stats, + persist: *event_log_mod.DiskPersist, + slurper: *slurper_mod.Slurper, + collection_index: *collection_index_mod.CollectionIndex, + backfiller: *backfill_mod.Backfiller, + bc: *broadcaster.Broadcaster, + validator: *validator_mod.Validator, +}; + +/// top-level HTTP request router — installed as bc.http_fallback +pub fn handleHttpRequest( + conn: *websocket.Conn, + method: []const u8, + url: []const u8, + body: []const u8, + headers: *const websocket.Handshake.KeyValue, + opaque_ctx: ?*anyopaque, +) void { + const ctx: *HttpContext = @ptrCast(@alignCast(opaque_ctx orelse return)); + + const qmark = std.mem.indexOfScalar(u8, url, '?'); + const path = url[0..(qmark orelse url.len)]; + const query = if (qmark) |q| url[q + 1 ..] else ""; + + if (std.mem.eql(u8, method, "GET")) { + handleGet(conn, path, query, headers, ctx); + } else if (std.mem.eql(u8, method, "POST")) { + handlePost(conn, path, query, body, headers, ctx); + } else { + respondText(conn, .method_not_allowed, "method not allowed"); + } +} + +fn handleGet(conn: *websocket.Conn, path: []const u8, query: []const u8, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void { + if (std.mem.eql(u8, path, "/_health") or std.mem.eql(u8, path, "/xrpc/_health")) { + respondJson(conn, .ok, "{\"status\":\"ok\"}"); + } else if (std.mem.eql(u8, path, "/_stats")) { + var stats_buf: [4096]u8 = undefined; + const body = broadcaster.formatStatsResponse(ctx.stats, &stats_buf); + respondJson(conn, .ok, body); + } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listRepos")) { + handleListRepos(conn, query, ctx.persist); + } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getRepoStatus")) { + handleGetRepoStatus(conn, query, ctx.persist); + } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getLatestCommit")) { + handleGetLatestCommit(conn, query, ctx.persist); + } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listReposByCollection")) { + handleListReposByCollection(conn, query, ctx.collection_index); + } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listHosts")) { + handleListHosts(conn, query, ctx.persist); + } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getHostStatus")) { + handleGetHostStatus(conn, query, ctx.persist); + } else if (std.mem.eql(u8, path, "/admin/hosts")) { + handleAdminListHosts(conn, headers, ctx); + } else if (std.mem.eql(u8, path, "/admin/backfill-collections")) { + handleAdminBackfillStatus(conn, headers, ctx); + } else if (std.mem.eql(u8, path, "/")) { + respondText(conn, .ok, + \\ _ + \\ ___| | __ _ _ _ + \\|_ / |/ _` | | | | + \\ / /| | (_| | |_| | + \\/___|_|\__,_|\__, | + \\ |___/ + \\ + \\This is an atproto [https://atproto.com] relay instance, + \\running the zlay codebase [https://tangled.org/zzstoatzz.io/zlay] + \\ + \\The firehose WebSocket path is at: /xrpc/com.atproto.sync.subscribeRepos + \\ + ); + } else if (std.mem.eql(u8, path, "/favicon.svg") or std.mem.eql(u8, path, "/favicon.ico")) { + httpRespond(conn, .ok, "image/svg+xml", + \\ + \\ + \\Z + \\ + ); + } else { + respondText(conn, .not_found, "not found"); + } +} + +fn handlePost(conn: *websocket.Conn, path: []const u8, query: []const u8, body: []const u8, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void { + if (std.mem.eql(u8, path, "/admin/repo/ban")) { + handleBan(conn, body, headers, ctx); + } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.requestCrawl")) { + handleRequestCrawl(conn, body, ctx.slurper); + } else if (std.mem.eql(u8, path, "/admin/hosts/block")) { + handleAdminBlockHost(conn, body, headers, ctx.persist); + } else if (std.mem.eql(u8, path, "/admin/hosts/unblock")) { + handleAdminUnblockHost(conn, body, headers, ctx.persist); + } else if (std.mem.eql(u8, path, "/admin/backfill-collections")) { + handleAdminBackfillTrigger(conn, query, headers, ctx.backfiller); + } else { + respondText(conn, .not_found, "not found"); + } +} + +fn handleBan(conn: *websocket.Conn, body: []const u8, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void { + if (!checkAdmin(conn, headers)) return; + + const parsed = std.json.parseFromSlice(struct { did: []const u8 }, ctx.persist.allocator, body, .{ .ignore_unknown_fields = true }) catch { + respondJson(conn, .bad_request, "{\"error\":\"invalid JSON, expected {\\\"did\\\":\\\"...\\\"}\"}"); + return; + }; + defer parsed.deinit(); + const did = parsed.value.did; + + // resolve DID → UID and take down + const uid = ctx.persist.uidForDid(did) catch { + respondJson(conn, .internal_server_error, "{\"error\":\"failed to resolve DID\"}"); + return; + }; + ctx.persist.takeDownUser(uid) catch { + respondJson(conn, .internal_server_error, "{\"error\":\"takedown failed\"}"); + return; + }; + + // emit #account event so downstream consumers see the takedown + if (buildAccountFrame(ctx.persist.allocator, did)) |frame_bytes| { + if (ctx.persist.persist(.account, uid, frame_bytes)) |relay_seq| { + ctx.bc.stats.relay_seq.store(relay_seq, .release); + const broadcast_data = broadcaster.resequenceFrame(ctx.persist.allocator, frame_bytes, relay_seq) orelse frame_bytes; + ctx.bc.broadcast(relay_seq, broadcast_data); + log.info("admin: emitted #account takedown event for {s} (seq={d})", .{ did, relay_seq }); + } else |err| { + log.warn("admin: failed to persist #account takedown event: {s}", .{@errorName(err)}); + } + } + + log.info("admin: banned {s} (uid={d})", .{ did, uid }); + respondJson(conn, .ok, "{\"success\":true}"); +} + +fn handleRequestCrawl(conn: *websocket.Conn, body: []const u8, slurper: *slurper_mod.Slurper) void { + const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, slurper.allocator, body, .{ .ignore_unknown_fields = true }) catch { + respondJson(conn, .bad_request, "{\"error\":\"invalid JSON, expected {\\\"hostname\\\":\\\"...\\\"}\"}"); + return; + }; + defer parsed.deinit(); + + // fast validation: hostname format (Go relay does this synchronously in handler) + const hostname = slurper_mod.validateHostname(slurper.allocator, parsed.value.hostname) catch |err| { + log.warn("requestCrawl rejected '{s}': {s}", .{ parsed.value.hostname, @errorName(err) }); + respondJson(conn, .bad_request, switch (err) { + error.EmptyHostname => "{\"error\":\"empty hostname\"}", + error.InvalidCharacter => "{\"error\":\"hostname contains invalid characters\"}", + error.InvalidLabel => "{\"error\":\"hostname has invalid label\"}", + error.TooFewLabels => "{\"error\":\"hostname must have at least two labels (e.g. pds.example.com)\"}", + error.LooksLikeIpAddress => "{\"error\":\"IP addresses not allowed, use a hostname\"}", + error.PortNotAllowed => "{\"error\":\"port numbers not allowed\"}", + error.LocalhostNotAllowed => "{\"error\":\"localhost not allowed\"}", + else => "{\"error\":\"invalid hostname\"}", + }); + return; + }; + defer slurper.allocator.free(hostname); + + // fast validation: domain ban check + if (slurper.persist.isDomainBanned(hostname)) { + log.warn("requestCrawl rejected '{s}': domain banned", .{hostname}); + respondJson(conn, .bad_request, "{\"error\":\"domain is banned\"}"); + return; + } + + // enqueue for async processing (describeServer check happens in crawl processor) + slurper.addCrawlRequest(hostname) catch { + respondJson(conn, .internal_server_error, "{\"error\":\"failed to store crawl request\"}"); + return; + }; + + log.info("crawl requested: {s}", .{hostname}); + respondJson(conn, .ok, "{\"success\":true}"); +} + +// --- admin host management --- + +fn handleAdminListHosts(conn: *websocket.Conn, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void { + if (!checkAdmin(conn, headers)) return; + + const persist = ctx.persist; + const hosts = persist.listAllHosts(persist.allocator) catch { + respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); + return; + }; + defer { + for (hosts) |h| { + persist.allocator.free(h.hostname); + persist.allocator.free(h.status); + } + persist.allocator.free(hosts); + } + + var list: std.ArrayListUnmanaged(u8) = .{}; + defer list.deinit(persist.allocator); + const w = list.writer(persist.allocator); + + w.writeAll("{\"hosts\":[") catch return; + + for (hosts, 0..) |host, i| { + if (i > 0) w.writeByte(',') catch return; + std.fmt.format(w, "{{\"id\":{d},\"hostname\":\"{s}\",\"status\":\"{s}\",\"last_seq\":{d},\"failed_attempts\":{d}}}", .{ + host.id, + host.hostname, + host.status, + host.last_seq, + host.failed_attempts, + }) catch return; + } + + std.fmt.format(w, "],\"active_workers\":{d}}}", .{ctx.slurper.workerCount()}) catch return; + respondJson(conn, .ok, list.items); +} + +fn handleAdminBlockHost(conn: *websocket.Conn, body: []const u8, headers: *const websocket.Handshake.KeyValue, persist: *event_log_mod.DiskPersist) void { + if (!checkAdmin(conn, headers)) return; + + const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, persist.allocator, body, .{ .ignore_unknown_fields = true }) catch { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid JSON\"}"); + return; + }; + defer parsed.deinit(); + + const host_info = persist.getOrCreateHost(parsed.value.hostname) catch { + respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"host lookup failed\"}"); + return; + }; + + persist.updateHostStatus(host_info.id, "blocked") catch { + respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"status update failed\"}"); + return; + }; + + log.info("admin: blocked host {s} (id={d})", .{ parsed.value.hostname, host_info.id }); + respondJson(conn, .ok, "{\"success\":true}"); +} + +fn handleAdminUnblockHost(conn: *websocket.Conn, body: []const u8, headers: *const websocket.Handshake.KeyValue, persist: *event_log_mod.DiskPersist) void { + if (!checkAdmin(conn, headers)) return; + + const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, persist.allocator, body, .{ .ignore_unknown_fields = true }) catch { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid JSON\"}"); + return; + }; + defer parsed.deinit(); + + const host_info = persist.getOrCreateHost(parsed.value.hostname) catch { + respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"host lookup failed\"}"); + return; + }; + + persist.updateHostStatus(host_info.id, "active") catch { + respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"status update failed\"}"); + return; + }; + persist.resetHostFailures(host_info.id) catch {}; + + log.info("admin: unblocked host {s} (id={d})", .{ parsed.value.hostname, host_info.id }); + respondJson(conn, .ok, "{\"success\":true}"); +} + +/// check admin auth via headers, send error response if not authorized. returns true if authorized. +fn checkAdmin(conn: *websocket.Conn, headers: ?*const websocket.Handshake.KeyValue) bool { + const admin_pw = std.posix.getenv("RELAY_ADMIN_PASSWORD") orelse { + respondJson(conn, .forbidden, "{\"error\":\"admin endpoint not configured\"}"); + return false; + }; + + const kv = headers orelse { + respondJson(conn, .unauthorized, "{\"error\":\"missing authorization header\"}"); + return false; + }; + + // handshake parser lowercases all header names + const auth_value = kv.get("authorization") orelse { + respondJson(conn, .unauthorized, "{\"error\":\"missing authorization header\"}"); + return false; + }; + + const bearer_prefix = "Bearer "; + if (!std.mem.startsWith(u8, auth_value, bearer_prefix)) { + respondJson(conn, .unauthorized, "{\"error\":\"invalid authorization scheme\"}"); + return false; + } + const token = auth_value[bearer_prefix.len..]; + if (!std.mem.eql(u8, token, admin_pw)) { + respondJson(conn, .unauthorized, "{\"error\":\"invalid token\"}"); + return false; + } + return true; +} + +// --- XRPC endpoint handlers --- + +fn handleListRepos(conn: *websocket.Conn, 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 { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid cursor\"}"); + return; + }; + if (cursor_val < 0) { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"cursor must be >= 0\"}"); + return; + } + + const limit = std.fmt.parseInt(i64, limit_str, 10) catch { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}"); + return; + }; + if (limit < 1 or limit > 1000) { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..1000\"}"); + return; + } + + // query accounts with repo state, paginated by UID + // includes both local status and upstream_status for combined active check + var result = persist.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 + , .{ cursor_val, limit }) catch { + respondJson(conn, .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 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); + + // Go relay: Account.IsActive() — both local AND upstream must be active + 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; + // Go relay: Account.AccountStatus() — local takes priority + const status = if (!local_ok) local_status else upstream_status; + + 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; + + respondJson(conn, .ok, fbs.getWritten()); +} + +fn handleGetRepoStatus(conn: *websocket.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void { + var did_buf: [256]u8 = undefined; + const did = queryParamDecoded(query, "did", &did_buf) orelse { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}"); + return; + }; + + // basic DID syntax check + if (!std.mem.startsWith(u8, did, "did:")) { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}"); + return; + } + + // look up account (includes both local and upstream status) + var row = (persist.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 { + respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); + return; + }) orelse { + respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}"); + 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); + // Go relay: Account.IsActive() / AccountStatus() + 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 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; + respondJson(conn, .ok, fbs.getWritten()); +} + +fn handleGetLatestCommit(conn: *websocket.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void { + var did_buf: [256]u8 = undefined; + const did = queryParamDecoded(query, "did", &did_buf) orelse { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}"); + return; + }; + + if (!std.mem.startsWith(u8, did, "did:")) { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}"); + return; + } + + // look up account + repo state (includes both local and upstream status) + var row = (persist.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 { + respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); + return; + }) orelse { + respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}"); + 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); + + // combined status: local takes priority (Go relay: AccountStatus()) + const status = if (!std.mem.eql(u8, local_status, "active")) local_status else upstream_status; + + // check account status (match Go relay behavior) + if (std.mem.eql(u8, status, "takendown") or std.mem.eql(u8, status, "suspended")) { + respondJson(conn, .forbidden, "{\"error\":\"RepoTakendown\",\"message\":\"account has been taken down\"}"); + return; + } else if (std.mem.eql(u8, status, "deactivated")) { + respondJson(conn, .forbidden, "{\"error\":\"RepoDeactivated\",\"message\":\"account is deactivated\"}"); + return; + } else if (std.mem.eql(u8, status, "deleted")) { + respondJson(conn, .forbidden, "{\"error\":\"RepoDeleted\",\"message\":\"account is deleted\"}"); + return; + } else if (!std.mem.eql(u8, status, "active")) { + respondJson(conn, .forbidden, "{\"error\":\"RepoInactive\",\"message\":\"account is not active\"}"); + return; + } + + if (rev.len == 0 or cid.len == 0) { + respondJson(conn, .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; + + respondJson(conn, .ok, fbs.getWritten()); +} + +fn handleListReposByCollection(conn: *websocket.Conn, query: []const u8, ci: *collection_index_mod.CollectionIndex) void { + const collection = queryParam(query, "collection") orelse { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"collection parameter required\"}"); + return; + }; + + if (collection.len == 0 or !std.mem.containsAtLeast(u8, collection, 1, ".")) { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid collection NSID\"}"); + return; + } + + const limit_str = queryParam(query, "limit") orelse "500"; + const limit = std.fmt.parseInt(usize, limit_str, 10) catch { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}"); + return; + }; + if (limit < 1 or limit > 1000) { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..1000\"}"); + return; + } + + var cursor_buf: [256]u8 = undefined; + const cursor_did = queryParamDecoded(query, "cursor", &cursor_buf); + + // scan collection index + var did_buf: [65536]u8 = undefined; + const ci_result = ci.listReposByCollection(collection, limit, cursor_did, &did_buf) catch { + respondJson(conn, .internal_server_error, "{\"error\":\"InternalError\",\"message\":\"index scan failed\"}"); + return; + }; + + // build JSON response + var buf: [65536]u8 = undefined; + var fbs = std.io.fixedBufferStream(&buf); + const w = fbs.writer(); + + 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; + respondJson(conn, .ok, fbs.getWritten()); +} + +fn handleListHosts(conn: *websocket.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void { + const cursor_str = queryParam(query, "cursor") orelse "0"; + const limit_str = queryParam(query, "limit") orelse "200"; + + const cursor_val = std.fmt.parseInt(i64, cursor_str, 10) catch { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid cursor\"}"); + return; + }; + if (cursor_val < 0) { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"cursor must be >= 0\"}"); + return; + } + + const limit = std.fmt.parseInt(i64, limit_str, 10) catch { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}"); + return; + }; + if (limit < 1 or limit > 1000) { + respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..1000\"}"); + return; + } + + var result = persist.db.query( + "SELECT id, hostname, status, last_seq FROM host WHERE id > $1 AND last_seq > 0 ORDER BY id ASC LIMIT $2", + .{ cursor_val, limit }, + ) catch { + respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); + return; + }; + defer result.deinit(); + + var buf: [65536]u8 = undefined; + var fbs = std.io.fixedBufferStream(&buf); + const w = fbs.writer(); + + 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; + std.fmt.format(w, ",\"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 >= limit and count > 1) { + std.fmt.format(w, ",\"cursor\":\"{d}\"", .{last_id}) catch return; + } + + w.writeByte('}') catch return; + respondJson(conn, .ok, fbs.getWritten()); +} + +fn handleGetHostStatus(conn: *websocket.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void { + var hostname_buf: [256]u8 = undefined; + const hostname = queryParamDecoded(query, "hostname", &hostname_buf) orelse { + respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"hostname parameter required\"}"); + return; + }; + + // look up host + var row = (persist.db.rowUnsafe( + "SELECT id, hostname, status, last_seq FROM host WHERE hostname = $1", + .{hostname}, + ) catch { + respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); + return; + }) orelse { + respondJson(conn, .bad_request, "{\"error\":\"HostNotFound\",\"message\":\"host not found\"}"); + 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); + + // map internal status to lexicon hostStatus values + const status = if (std.mem.eql(u8, raw_status, "blocked")) + "banned" + else if (std.mem.eql(u8, raw_status, "exhausted")) + "offline" + else + raw_status; // active, idle pass through + + // count accounts on this host + const account_count: i64 = if (persist.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 buf: [4096]u8 = undefined; + var fbs = std.io.fixedBufferStream(&buf); + const w = fbs.writer(); + + w.writeAll("{\"hostname\":\"") catch return; + w.writeAll(host_name) catch return; + w.writeAll("\"") catch return; + std.fmt.format(w, ",\"seq\":{d},\"accountCount\":{d}", .{ seq, account_count }) catch return; + w.writeAll(",\"status\":\"") catch return; + w.writeAll(status) catch return; + w.writeAll("\"}") catch return; + + respondJson(conn, .ok, fbs.getWritten()); +} + +/// build a CBOR #account frame for a takedown event. +/// header: {op: 1, t: "#account"}, payload: {seq: 0, did: "...", time: "...", active: false, status: "takendown"} +fn buildAccountFrame(allocator: std.mem.Allocator, did: []const u8) ?[]const u8 { + const zat = @import("zat"); + const cbor = zat.cbor; + + const header: cbor.Value = .{ .map = &.{ + .{ .key = "op", .value = .{ .unsigned = 1 } }, + .{ .key = "t", .value = .{ .text = "#account" } }, + } }; + + var time_buf: [24]u8 = undefined; + const time_str = formatTimestamp(&time_buf); + + const payload: cbor.Value = .{ .map = &.{ + .{ .key = "seq", .value = .{ .unsigned = 0 } }, + .{ .key = "did", .value = .{ .text = did } }, + .{ .key = "time", .value = .{ .text = time_str } }, + .{ .key = "active", .value = .{ .boolean = false } }, + .{ .key = "status", .value = .{ .text = "takendown" } }, + } }; + + const header_bytes = cbor.encodeAlloc(allocator, header) catch return null; + const payload_bytes = cbor.encodeAlloc(allocator, payload) catch { + allocator.free(header_bytes); + return null; + }; + + var frame = allocator.alloc(u8, header_bytes.len + payload_bytes.len) catch { + allocator.free(header_bytes); + allocator.free(payload_bytes); + return null; + }; + @memcpy(frame[0..header_bytes.len], header_bytes); + @memcpy(frame[header_bytes.len..], payload_bytes); + + allocator.free(header_bytes); + allocator.free(payload_bytes); + + return frame; +} + +/// format current UTC time as ISO 8601 (YYYY-MM-DDTHH:MM:SSZ) +fn formatTimestamp(buf: *[24]u8) []const u8 { + const ts: u64 = @intCast(std.time.timestamp()); + const es = std.time.epoch.EpochSeconds{ .secs = ts }; + const day = es.getEpochDay(); + const yd = day.calculateYearDay(); + const md = yd.calculateMonthDay(); + const ds = es.getDaySeconds(); + + return std.fmt.bufPrint(buf, "{d:0>4}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}Z", .{ + yd.year, + @as(u32, @intFromEnum(md.month)) + 1, + @as(u32, md.day_index) + 1, + ds.getHoursIntoDay(), + ds.getMinutesIntoHour(), + ds.getSecondsIntoMinute(), + }) catch "1970-01-01T00:00:00Z"; +} + +// --- backfill handlers --- + +fn handleAdminBackfillTrigger(conn: *websocket.Conn, query: []const u8, headers: *const websocket.Handshake.KeyValue, backfiller: *backfill_mod.Backfiller) void { + if (!checkAdmin(conn, headers)) return; + + const source = queryParam(query, "source") orelse "bsky.network"; + + backfiller.start(source) catch |err| { + switch (err) { + error.AlreadyRunning => { + respondJson(conn, .conflict, "{\"error\":\"backfill already in progress\"}"); + }, + else => { + respondJson(conn, .internal_server_error, "{\"error\":\"failed to start backfill\"}"); + }, + } + return; + }; + + var buf: [256]u8 = undefined; + const body = std.fmt.bufPrint(&buf, "{{\"status\":\"started\",\"source\":\"{s}\"}}", .{source}) catch { + respondJson(conn, .ok, "{\"status\":\"started\"}"); + return; + }; + respondJson(conn, .ok, body); +} + +fn handleAdminBackfillStatus(conn: *websocket.Conn, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void { + if (!checkAdmin(conn, headers)) return; + + const body = ctx.backfiller.getStatus(ctx.backfiller.allocator) catch { + respondJson(conn, .internal_server_error, "{\"error\":\"failed to query backfill status\"}"); + return; + }; + defer ctx.backfiller.allocator.free(body); + + respondJson(conn, .ok, body); +} + +// --- 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; +} + +/// like queryParam but percent-decodes the value into buf. +/// returns null if the param is missing, or a slice into buf with the decoded value. +fn queryParamDecoded(query: []const u8, name: []const u8, buf: []u8) ?[]const u8 { + const raw = queryParam(query, name) orelse return null; + var i: usize = 0; + var out: usize = 0; + while (i < raw.len) { + if (raw[i] == '%' and i + 2 < raw.len) { + const hi = hexVal(raw[i + 1]) orelse { + if (out >= buf.len) return null; + buf[out] = raw[i]; + out += 1; + i += 1; + continue; + }; + const lo = hexVal(raw[i + 2]) orelse { + if (out >= buf.len) return null; + buf[out] = raw[i]; + out += 1; + i += 1; + continue; + }; + if (out >= buf.len) return null; + buf[out] = (@as(u8, hi) << 4) | @as(u8, lo); + out += 1; + i += 3; + } else if (raw[i] == '+') { + if (out >= buf.len) return null; + buf[out] = ' '; + out += 1; + i += 1; + } else { + if (out >= buf.len) return null; + buf[out] = raw[i]; + out += 1; + i += 1; + } + } + return buf[0..out]; +} + +fn hexVal(c: u8) ?u4 { + return switch (c) { + '0'...'9' => @intCast(c - '0'), + 'a'...'f' => @intCast(c - 'a' + 10), + 'A'...'F' => @intCast(c - 'A' + 10), + else => null, + }; +} + +// --- response helpers (write raw HTTP to websocket.Conn) --- + +fn httpRespond(conn: *websocket.Conn, status: http.Status, content_type: []const u8, body: []const u8) void { + var buf: [512]u8 = undefined; + const header = std.fmt.bufPrint(&buf, "HTTP/1.1 {s}\r\nContent-Type: {s}\r\nContent-Length: {d}\r\nConnection: close\r\nServer: zlay\r\n\r\n", .{ + httpStatusLine(status), + content_type, + body.len, + }) catch return; + conn.writeFramed(header) catch return; + if (body.len > 0) conn.writeFramed(body) catch return; +} + +fn respondJson(conn: *websocket.Conn, status: http.Status, body: []const u8) void { + httpRespond(conn, status, "application/json", body); +} + +fn respondText(conn: *websocket.Conn, status: http.Status, body: []const u8) void { + httpRespond(conn, status, "text/plain", body); +} + +fn httpStatusLine(status: http.Status) []const u8 { + return switch (status) { + .ok => "200 OK", + .bad_request => "400 Bad Request", + .unauthorized => "401 Unauthorized", + .forbidden => "403 Forbidden", + .not_found => "404 Not Found", + .method_not_allowed => "405 Method Not Allowed", + .conflict => "409 Conflict", + .internal_server_error => "500 Internal Server Error", + else => "500 Internal Server Error", + }; +} diff --git a/src/main.zig b/src/main.zig index 3f4125c..a3e88b2 100644 --- a/src/main.zig +++ b/src/main.zig @@ -31,6 +31,7 @@ const slurper_mod = @import("slurper.zig"); const event_log_mod = @import("event_log.zig"); const collection_index_mod = @import("collection_index.zig"); const backfill_mod = @import("backfill.zig"); +const api = @import("api.zig"); const log = std.log.scoped(.relay); @@ -41,17 +42,6 @@ pub const default_stack_size = 2 * 1024 * 1024; var shutdown_flag: std.atomic.Value(bool) = .{ .raw = false }; -/// context for HTTP fallback handlers (passed as opaque pointer through broadcaster) -const HttpContext = struct { - stats: *broadcaster.Stats, - persist: *event_log_mod.DiskPersist, - slurper: *slurper_mod.Slurper, - collection_index: *collection_index_mod.CollectionIndex, - backfiller: *backfill_mod.Backfiller, - bc: *broadcaster.Broadcaster, - validator: *validator_mod.Validator, -}; - /// metrics-only server on the internal port const MetricsServer = struct { server: std.net.Server, @@ -169,7 +159,7 @@ pub fn main() !void { const gc_thread = try std.Thread.spawn(.{ .stack_size = default_stack_size }, gcLoop, .{&dp}); // wire HTTP fallback into broadcaster (all API endpoints served on WS port) - var http_context = HttpContext{ + var http_context = api.HttpContext{ .stats = &bc.stats, .persist = &dp, .slurper = &slurper, @@ -178,7 +168,7 @@ pub fn main() !void { .bc = &bc, .validator = &val, }; - bc.http_fallback = handleHttpRequest; + bc.http_fallback = api.handleHttpRequest; bc.http_fallback_ctx = @ptrCast(&http_context); // start metrics-only server (internal port) @@ -269,905 +259,6 @@ fn installSignalHandlers() void { std.posix.sigaction(std.posix.SIG.PIPE, &ignore_act, null); } -// --- HTTP fallback handler (called from broadcaster via websocket httpFallback) --- - -fn handleHttpRequest( - conn: *websocket.Conn, - method: []const u8, - url: []const u8, - body: []const u8, - headers: *const websocket.Handshake.KeyValue, - opaque_ctx: ?*anyopaque, -) void { - const ctx: *HttpContext = @ptrCast(@alignCast(opaque_ctx orelse return)); - - const qmark = std.mem.indexOfScalar(u8, url, '?'); - const path = url[0..(qmark orelse url.len)]; - const query = if (qmark) |q| url[q + 1 ..] else ""; - - if (std.mem.eql(u8, method, "GET")) { - handleGet(conn, path, query, headers, ctx); - } else if (std.mem.eql(u8, method, "POST")) { - handlePost(conn, path, query, body, headers, ctx); - } else { - respondText(conn, .method_not_allowed, "method not allowed"); - } -} - -fn handleGet(conn: *websocket.Conn, path: []const u8, query: []const u8, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void { - if (std.mem.eql(u8, path, "/_health") or std.mem.eql(u8, path, "/xrpc/_health")) { - respondJson(conn, .ok, "{\"status\":\"ok\"}"); - } else if (std.mem.eql(u8, path, "/_stats")) { - var stats_buf: [4096]u8 = undefined; - const body = broadcaster.formatStatsResponse(ctx.stats, &stats_buf); - respondJson(conn, .ok, body); - } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listRepos")) { - handleListRepos(conn, query, ctx.persist); - } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getRepoStatus")) { - handleGetRepoStatus(conn, query, ctx.persist); - } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getLatestCommit")) { - handleGetLatestCommit(conn, query, ctx.persist); - } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listReposByCollection")) { - handleListReposByCollection(conn, query, ctx.collection_index); - } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listHosts")) { - handleListHosts(conn, query, ctx.persist); - } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getHostStatus")) { - handleGetHostStatus(conn, query, ctx.persist); - } else if (std.mem.eql(u8, path, "/admin/hosts")) { - handleAdminListHosts(conn, headers, ctx); - } else if (std.mem.eql(u8, path, "/admin/backfill-collections")) { - handleAdminBackfillStatus(conn, headers, ctx); - } else if (std.mem.eql(u8, path, "/")) { - respondText(conn, .ok, - \\ _ - \\ ___| | __ _ _ _ - \\|_ / |/ _` | | | | - \\ / /| | (_| | |_| | - \\/___|_|\__,_|\__, | - \\ |___/ - \\ - \\This is an atproto [https://atproto.com] relay instance, - \\running the zlay codebase [https://tangled.org/zzstoatzz.io/zlay] - \\ - \\The firehose WebSocket path is at: /xrpc/com.atproto.sync.subscribeRepos - \\ - ); - } else if (std.mem.eql(u8, path, "/favicon.svg") or std.mem.eql(u8, path, "/favicon.ico")) { - httpRespond(conn, .ok, "image/svg+xml", - \\ - \\ - \\Z - \\ - ); - } else { - respondText(conn, .not_found, "not found"); - } -} - -fn handlePost(conn: *websocket.Conn, path: []const u8, query: []const u8, body: []const u8, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void { - if (std.mem.eql(u8, path, "/admin/repo/ban")) { - handleBan(conn, body, headers, ctx); - } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.requestCrawl")) { - handleRequestCrawl(conn, body, ctx.slurper); - } else if (std.mem.eql(u8, path, "/admin/hosts/block")) { - handleAdminBlockHost(conn, body, headers, ctx.persist); - } else if (std.mem.eql(u8, path, "/admin/hosts/unblock")) { - handleAdminUnblockHost(conn, body, headers, ctx.persist); - } else if (std.mem.eql(u8, path, "/admin/backfill-collections")) { - handleAdminBackfillTrigger(conn, query, headers, ctx.backfiller); - } else { - respondText(conn, .not_found, "not found"); - } -} - -fn handleBan(conn: *websocket.Conn, body: []const u8, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void { - if (!checkAdmin(conn, headers)) return; - - const parsed = std.json.parseFromSlice(struct { did: []const u8 }, ctx.persist.allocator, body, .{ .ignore_unknown_fields = true }) catch { - respondJson(conn, .bad_request, "{\"error\":\"invalid JSON, expected {\\\"did\\\":\\\"...\\\"}\"}"); - return; - }; - defer parsed.deinit(); - const did = parsed.value.did; - - // resolve DID → UID and take down - const uid = ctx.persist.uidForDid(did) catch { - respondJson(conn, .internal_server_error, "{\"error\":\"failed to resolve DID\"}"); - return; - }; - ctx.persist.takeDownUser(uid) catch { - respondJson(conn, .internal_server_error, "{\"error\":\"takedown failed\"}"); - return; - }; - - // emit #account event so downstream consumers see the takedown - if (buildAccountFrame(ctx.persist.allocator, did)) |frame_bytes| { - if (ctx.persist.persist(.account, uid, frame_bytes)) |relay_seq| { - ctx.bc.stats.relay_seq.store(relay_seq, .release); - const broadcast_data = broadcaster.resequenceFrame(ctx.persist.allocator, frame_bytes, relay_seq) orelse frame_bytes; - ctx.bc.broadcast(relay_seq, broadcast_data); - log.info("admin: emitted #account takedown event for {s} (seq={d})", .{ did, relay_seq }); - } else |err| { - log.warn("admin: failed to persist #account takedown event: {s}", .{@errorName(err)}); - } - } - - log.info("admin: banned {s} (uid={d})", .{ did, uid }); - respondJson(conn, .ok, "{\"success\":true}"); -} - -fn handleRequestCrawl(conn: *websocket.Conn, body: []const u8, slurper: *slurper_mod.Slurper) void { - const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, slurper.allocator, body, .{ .ignore_unknown_fields = true }) catch { - respondJson(conn, .bad_request, "{\"error\":\"invalid JSON, expected {\\\"hostname\\\":\\\"...\\\"}\"}"); - return; - }; - defer parsed.deinit(); - - // fast validation: hostname format (Go relay does this synchronously in handler) - const hostname = slurper_mod.validateHostname(slurper.allocator, parsed.value.hostname) catch |err| { - log.warn("requestCrawl rejected '{s}': {s}", .{ parsed.value.hostname, @errorName(err) }); - respondJson(conn, .bad_request, switch (err) { - error.EmptyHostname => "{\"error\":\"empty hostname\"}", - error.InvalidCharacter => "{\"error\":\"hostname contains invalid characters\"}", - error.InvalidLabel => "{\"error\":\"hostname has invalid label\"}", - error.TooFewLabels => "{\"error\":\"hostname must have at least two labels (e.g. pds.example.com)\"}", - error.LooksLikeIpAddress => "{\"error\":\"IP addresses not allowed, use a hostname\"}", - error.PortNotAllowed => "{\"error\":\"port numbers not allowed\"}", - error.LocalhostNotAllowed => "{\"error\":\"localhost not allowed\"}", - else => "{\"error\":\"invalid hostname\"}", - }); - return; - }; - defer slurper.allocator.free(hostname); - - // fast validation: domain ban check - if (slurper.persist.isDomainBanned(hostname)) { - log.warn("requestCrawl rejected '{s}': domain banned", .{hostname}); - respondJson(conn, .bad_request, "{\"error\":\"domain is banned\"}"); - return; - } - - // enqueue for async processing (describeServer check happens in crawl processor) - slurper.addCrawlRequest(hostname) catch { - respondJson(conn, .internal_server_error, "{\"error\":\"failed to store crawl request\"}"); - return; - }; - - log.info("crawl requested: {s}", .{hostname}); - respondJson(conn, .ok, "{\"success\":true}"); -} - -// --- admin host management --- - -fn handleAdminListHosts(conn: *websocket.Conn, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void { - if (!checkAdmin(conn, headers)) return; - - const persist = ctx.persist; - const hosts = persist.listAllHosts(persist.allocator) catch { - respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); - return; - }; - defer { - for (hosts) |h| { - persist.allocator.free(h.hostname); - persist.allocator.free(h.status); - } - persist.allocator.free(hosts); - } - - var list: std.ArrayListUnmanaged(u8) = .{}; - defer list.deinit(persist.allocator); - const w = list.writer(persist.allocator); - - w.writeAll("{\"hosts\":[") catch return; - - for (hosts, 0..) |host, i| { - if (i > 0) w.writeByte(',') catch return; - std.fmt.format(w, "{{\"id\":{d},\"hostname\":\"{s}\",\"status\":\"{s}\",\"last_seq\":{d},\"failed_attempts\":{d}}}", .{ - host.id, - host.hostname, - host.status, - host.last_seq, - host.failed_attempts, - }) catch return; - } - - std.fmt.format(w, "],\"active_workers\":{d}}}", .{ctx.slurper.workerCount()}) catch return; - respondJson(conn, .ok, list.items); -} - -fn handleAdminBlockHost(conn: *websocket.Conn, body: []const u8, headers: *const websocket.Handshake.KeyValue, persist: *event_log_mod.DiskPersist) void { - if (!checkAdmin(conn, headers)) return; - - const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, persist.allocator, body, .{ .ignore_unknown_fields = true }) catch { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid JSON\"}"); - return; - }; - defer parsed.deinit(); - - const host_info = persist.getOrCreateHost(parsed.value.hostname) catch { - respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"host lookup failed\"}"); - return; - }; - - persist.updateHostStatus(host_info.id, "blocked") catch { - respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"status update failed\"}"); - return; - }; - - log.info("admin: blocked host {s} (id={d})", .{ parsed.value.hostname, host_info.id }); - respondJson(conn, .ok, "{\"success\":true}"); -} - -fn handleAdminUnblockHost(conn: *websocket.Conn, body: []const u8, headers: *const websocket.Handshake.KeyValue, persist: *event_log_mod.DiskPersist) void { - if (!checkAdmin(conn, headers)) return; - - const parsed = std.json.parseFromSlice(struct { hostname: []const u8 }, persist.allocator, body, .{ .ignore_unknown_fields = true }) catch { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid JSON\"}"); - return; - }; - defer parsed.deinit(); - - const host_info = persist.getOrCreateHost(parsed.value.hostname) catch { - respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"host lookup failed\"}"); - return; - }; - - persist.updateHostStatus(host_info.id, "active") catch { - respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"status update failed\"}"); - return; - }; - persist.resetHostFailures(host_info.id) catch {}; - - log.info("admin: unblocked host {s} (id={d})", .{ parsed.value.hostname, host_info.id }); - respondJson(conn, .ok, "{\"success\":true}"); -} - -/// check admin auth via headers, send error response if not authorized. returns true if authorized. -fn checkAdmin(conn: *websocket.Conn, headers: ?*const websocket.Handshake.KeyValue) bool { - const admin_pw = std.posix.getenv("RELAY_ADMIN_PASSWORD") orelse { - respondJson(conn, .forbidden, "{\"error\":\"admin endpoint not configured\"}"); - return false; - }; - - const kv = headers orelse { - respondJson(conn, .unauthorized, "{\"error\":\"missing authorization header\"}"); - return false; - }; - - // handshake parser lowercases all header names - const auth_value = kv.get("authorization") orelse { - respondJson(conn, .unauthorized, "{\"error\":\"missing authorization header\"}"); - return false; - }; - - const bearer_prefix = "Bearer "; - if (!std.mem.startsWith(u8, auth_value, bearer_prefix)) { - respondJson(conn, .unauthorized, "{\"error\":\"invalid authorization scheme\"}"); - return false; - } - const token = auth_value[bearer_prefix.len..]; - if (!std.mem.eql(u8, token, admin_pw)) { - respondJson(conn, .unauthorized, "{\"error\":\"invalid token\"}"); - return false; - } - return true; -} - -// --- XRPC endpoint handlers --- - -fn handleListRepos(conn: *websocket.Conn, 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 { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid cursor\"}"); - return; - }; - if (cursor_val < 0) { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"cursor must be >= 0\"}"); - return; - } - - const limit = std.fmt.parseInt(i64, limit_str, 10) catch { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}"); - return; - }; - if (limit < 1 or limit > 1000) { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..1000\"}"); - return; - } - - // query accounts with repo state, paginated by UID - // includes both local status and upstream_status for combined active check - var result = persist.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 - , .{ cursor_val, limit }) catch { - respondJson(conn, .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 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); - - // Go relay: Account.IsActive() — both local AND upstream must be active - 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; - // Go relay: Account.AccountStatus() — local takes priority - const status = if (!local_ok) local_status else upstream_status; - - 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; - - respondJson(conn, .ok, fbs.getWritten()); -} - -fn handleGetRepoStatus(conn: *websocket.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void { - var did_buf: [256]u8 = undefined; - const did = queryParamDecoded(query, "did", &did_buf) orelse { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}"); - return; - }; - - // basic DID syntax check - if (!std.mem.startsWith(u8, did, "did:")) { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}"); - return; - } - - // look up account (includes both local and upstream status) - var row = (persist.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 { - respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); - return; - }) orelse { - respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}"); - 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); - // Go relay: Account.IsActive() / AccountStatus() - 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 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; - respondJson(conn, .ok, fbs.getWritten()); -} - -fn handleGetLatestCommit(conn: *websocket.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void { - var did_buf: [256]u8 = undefined; - const did = queryParamDecoded(query, "did", &did_buf) orelse { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"did parameter required\"}"); - return; - }; - - if (!std.mem.startsWith(u8, did, "did:")) { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid DID\"}"); - return; - } - - // look up account + repo state (includes both local and upstream status) - var row = (persist.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 { - respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); - return; - }) orelse { - respondJson(conn, .not_found, "{\"error\":\"RepoNotFound\",\"message\":\"account not found\"}"); - 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); - - // combined status: local takes priority (Go relay: AccountStatus()) - const status = if (!std.mem.eql(u8, local_status, "active")) local_status else upstream_status; - - // check account status (match Go relay behavior) - if (std.mem.eql(u8, status, "takendown") or std.mem.eql(u8, status, "suspended")) { - respondJson(conn, .forbidden, "{\"error\":\"RepoTakendown\",\"message\":\"account has been taken down\"}"); - return; - } else if (std.mem.eql(u8, status, "deactivated")) { - respondJson(conn, .forbidden, "{\"error\":\"RepoDeactivated\",\"message\":\"account is deactivated\"}"); - return; - } else if (std.mem.eql(u8, status, "deleted")) { - respondJson(conn, .forbidden, "{\"error\":\"RepoDeleted\",\"message\":\"account is deleted\"}"); - return; - } else if (!std.mem.eql(u8, status, "active")) { - respondJson(conn, .forbidden, "{\"error\":\"RepoInactive\",\"message\":\"account is not active\"}"); - return; - } - - if (rev.len == 0 or cid.len == 0) { - respondJson(conn, .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; - - respondJson(conn, .ok, fbs.getWritten()); -} - -fn handleListReposByCollection(conn: *websocket.Conn, query: []const u8, ci: *collection_index_mod.CollectionIndex) void { - const collection = queryParam(query, "collection") orelse { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"collection parameter required\"}"); - return; - }; - - if (collection.len == 0 or !std.mem.containsAtLeast(u8, collection, 1, ".")) { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid collection NSID\"}"); - return; - } - - const limit_str = queryParam(query, "limit") orelse "500"; - const limit = std.fmt.parseInt(usize, limit_str, 10) catch { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}"); - return; - }; - if (limit < 1 or limit > 1000) { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..1000\"}"); - return; - } - - var cursor_buf: [256]u8 = undefined; - const cursor_did = queryParamDecoded(query, "cursor", &cursor_buf); - - // scan collection index - var did_buf: [65536]u8 = undefined; - const result = ci.listReposByCollection(collection, limit, cursor_did, &did_buf) catch { - respondJson(conn, .internal_server_error, "{\"error\":\"InternalError\",\"message\":\"index scan failed\"}"); - return; - }; - - // build JSON response - var buf: [65536]u8 = undefined; - var fbs = std.io.fixedBufferStream(&buf); - const w = fbs.writer(); - - w.writeAll("{\"repos\":[") catch return; - for (0..result.count) |i| { - if (i > 0) w.writeByte(',') catch return; - w.writeAll("{\"did\":\"") catch return; - w.writeAll(result.getDid(i)) catch return; - w.writeAll("\"}") catch return; - } - w.writeByte(']') catch return; - - if (result.last_did) |last| { - if (result.count >= limit) { - w.writeAll(",\"cursor\":\"") catch return; - w.writeAll(last) catch return; - w.writeAll("\"") catch return; - } - } - - w.writeByte('}') catch return; - respondJson(conn, .ok, fbs.getWritten()); -} - -fn handleListHosts(conn: *websocket.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void { - const cursor_str = queryParam(query, "cursor") orelse "0"; - const limit_str = queryParam(query, "limit") orelse "200"; - - const cursor_val = std.fmt.parseInt(i64, cursor_str, 10) catch { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid cursor\"}"); - return; - }; - if (cursor_val < 0) { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"cursor must be >= 0\"}"); - return; - } - - const limit = std.fmt.parseInt(i64, limit_str, 10) catch { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"invalid limit\"}"); - return; - }; - if (limit < 1 or limit > 1000) { - respondJson(conn, .bad_request, "{\"error\":\"BadRequest\",\"message\":\"limit must be 1..1000\"}"); - return; - } - - var result = persist.db.query( - "SELECT id, hostname, status, last_seq FROM host WHERE id > $1 AND last_seq > 0 ORDER BY id ASC LIMIT $2", - .{ cursor_val, limit }, - ) catch { - respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); - return; - }; - defer result.deinit(); - - var buf: [65536]u8 = undefined; - var fbs = std.io.fixedBufferStream(&buf); - const w = fbs.writer(); - - 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; - std.fmt.format(w, ",\"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 >= limit and count > 1) { - std.fmt.format(w, ",\"cursor\":\"{d}\"", .{last_id}) catch return; - } - - w.writeByte('}') catch return; - respondJson(conn, .ok, fbs.getWritten()); -} - -fn handleGetHostStatus(conn: *websocket.Conn, query: []const u8, persist: *event_log_mod.DiskPersist) void { - var hostname_buf: [256]u8 = undefined; - const hostname = queryParamDecoded(query, "hostname", &hostname_buf) orelse { - respondJson(conn, .bad_request, "{\"error\":\"InvalidRequest\",\"message\":\"hostname parameter required\"}"); - return; - }; - - // look up host - var row = (persist.db.rowUnsafe( - "SELECT id, hostname, status, last_seq FROM host WHERE hostname = $1", - .{hostname}, - ) catch { - respondJson(conn, .internal_server_error, "{\"error\":\"DatabaseError\",\"message\":\"query failed\"}"); - return; - }) orelse { - respondJson(conn, .bad_request, "{\"error\":\"HostNotFound\",\"message\":\"host not found\"}"); - 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); - - // map internal status to lexicon hostStatus values - const status = if (std.mem.eql(u8, raw_status, "blocked")) - "banned" - else if (std.mem.eql(u8, raw_status, "exhausted")) - "offline" - else - raw_status; // active, idle pass through - - // count accounts on this host - const account_count: i64 = if (persist.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 buf: [4096]u8 = undefined; - var fbs = std.io.fixedBufferStream(&buf); - const w = fbs.writer(); - - w.writeAll("{\"hostname\":\"") catch return; - w.writeAll(host_name) catch return; - w.writeAll("\"") catch return; - std.fmt.format(w, ",\"seq\":{d},\"accountCount\":{d}", .{ seq, account_count }) catch return; - w.writeAll(",\"status\":\"") catch return; - w.writeAll(status) catch return; - w.writeAll("\"}") catch return; - - respondJson(conn, .ok, fbs.getWritten()); -} - -/// build a CBOR #account frame for a takedown event. -/// header: {op: 1, t: "#account"}, payload: {seq: 0, did: "...", time: "...", active: false, status: "takendown"} -fn buildAccountFrame(allocator: std.mem.Allocator, did: []const u8) ?[]const u8 { - const zat = @import("zat"); - const cbor = zat.cbor; - - const header: cbor.Value = .{ .map = &.{ - .{ .key = "op", .value = .{ .unsigned = 1 } }, - .{ .key = "t", .value = .{ .text = "#account" } }, - } }; - - var time_buf: [24]u8 = undefined; - const time_str = formatTimestamp(&time_buf); - - const payload: cbor.Value = .{ .map = &.{ - .{ .key = "seq", .value = .{ .unsigned = 0 } }, - .{ .key = "did", .value = .{ .text = did } }, - .{ .key = "time", .value = .{ .text = time_str } }, - .{ .key = "active", .value = .{ .boolean = false } }, - .{ .key = "status", .value = .{ .text = "takendown" } }, - } }; - - const header_bytes = cbor.encodeAlloc(allocator, header) catch return null; - const payload_bytes = cbor.encodeAlloc(allocator, payload) catch { - allocator.free(header_bytes); - return null; - }; - - var frame = allocator.alloc(u8, header_bytes.len + payload_bytes.len) catch { - allocator.free(header_bytes); - allocator.free(payload_bytes); - return null; - }; - @memcpy(frame[0..header_bytes.len], header_bytes); - @memcpy(frame[header_bytes.len..], payload_bytes); - - allocator.free(header_bytes); - allocator.free(payload_bytes); - - return frame; -} - -/// format current UTC time as ISO 8601 (YYYY-MM-DDTHH:MM:SSZ) -fn formatTimestamp(buf: *[24]u8) []const u8 { - const ts: u64 = @intCast(std.time.timestamp()); - const es = std.time.epoch.EpochSeconds{ .secs = ts }; - const day = es.getEpochDay(); - const yd = day.calculateYearDay(); - const md = yd.calculateMonthDay(); - const ds = es.getDaySeconds(); - - return std.fmt.bufPrint(buf, "{d:0>4}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}Z", .{ - yd.year, - @as(u32, @intFromEnum(md.month)) + 1, - @as(u32, md.day_index) + 1, - ds.getHoursIntoDay(), - ds.getMinutesIntoHour(), - ds.getSecondsIntoMinute(), - }) catch "1970-01-01T00:00:00Z"; -} - -// --- backfill handlers --- - -fn handleAdminBackfillTrigger(conn: *websocket.Conn, query: []const u8, headers: *const websocket.Handshake.KeyValue, backfiller: *backfill_mod.Backfiller) void { - if (!checkAdmin(conn, headers)) return; - - const source = queryParam(query, "source") orelse "bsky.network"; - - backfiller.start(source) catch |err| { - switch (err) { - error.AlreadyRunning => { - respondJson(conn, .conflict, "{\"error\":\"backfill already in progress\"}"); - }, - else => { - respondJson(conn, .internal_server_error, "{\"error\":\"failed to start backfill\"}"); - }, - } - return; - }; - - var buf: [256]u8 = undefined; - const body = std.fmt.bufPrint(&buf, "{{\"status\":\"started\",\"source\":\"{s}\"}}", .{source}) catch { - respondJson(conn, .ok, "{\"status\":\"started\"}"); - return; - }; - respondJson(conn, .ok, body); -} - -fn handleAdminBackfillStatus(conn: *websocket.Conn, headers: *const websocket.Handshake.KeyValue, ctx: *HttpContext) void { - if (!checkAdmin(conn, headers)) return; - - const body = ctx.backfiller.getStatus(ctx.backfiller.allocator) catch { - respondJson(conn, .internal_server_error, "{\"error\":\"failed to query backfill status\"}"); - return; - }; - defer ctx.backfiller.allocator.free(body); - - respondJson(conn, .ok, body); -} - -// --- 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; -} - -/// like queryParam but percent-decodes the value into buf. -/// returns null if the param is missing, or a slice into buf with the decoded value. -fn queryParamDecoded(query: []const u8, name: []const u8, buf: []u8) ?[]const u8 { - const raw = queryParam(query, name) orelse return null; - var i: usize = 0; - var out: usize = 0; - while (i < raw.len) { - if (raw[i] == '%' and i + 2 < raw.len) { - const hi = hexVal(raw[i + 1]) orelse { - if (out >= buf.len) return null; - buf[out] = raw[i]; - out += 1; - i += 1; - continue; - }; - const lo = hexVal(raw[i + 2]) orelse { - if (out >= buf.len) return null; - buf[out] = raw[i]; - out += 1; - i += 1; - continue; - }; - if (out >= buf.len) return null; - buf[out] = (@as(u8, hi) << 4) | @as(u8, lo); - out += 1; - i += 3; - } else if (raw[i] == '+') { - if (out >= buf.len) return null; - buf[out] = ' '; - out += 1; - i += 1; - } else { - if (out >= buf.len) return null; - buf[out] = raw[i]; - out += 1; - i += 1; - } - } - return buf[0..out]; -} - -fn hexVal(c: u8) ?u4 { - return switch (c) { - '0'...'9' => @intCast(c - '0'), - 'a'...'f' => @intCast(c - 'a' + 10), - 'A'...'F' => @intCast(c - 'A' + 10), - else => null, - }; -} - -// --- response helpers (write raw HTTP to websocket.Conn) --- - -fn httpRespond(conn: *websocket.Conn, status: http.Status, content_type: []const u8, body: []const u8) void { - var buf: [512]u8 = undefined; - const header = std.fmt.bufPrint(&buf, "HTTP/1.1 {s}\r\nContent-Type: {s}\r\nContent-Length: {d}\r\nConnection: close\r\nServer: zlay\r\n\r\n", .{ - httpStatusLine(status), - content_type, - body.len, - }) catch return; - conn.writeFramed(header) catch return; - if (body.len > 0) conn.writeFramed(body) catch return; -} - -fn respondJson(conn: *websocket.Conn, status: http.Status, body: []const u8) void { - httpRespond(conn, status, "application/json", body); -} - -fn respondText(conn: *websocket.Conn, status: http.Status, body: []const u8) void { - httpRespond(conn, status, "text/plain", body); -} - -fn httpStatusLine(status: http.Status) []const u8 { - return switch (status) { - .ok => "200 OK", - .bad_request => "400 Bad Request", - .unauthorized => "401 Unauthorized", - .forbidden => "403 Forbidden", - .not_found => "404 Not Found", - .method_not_allowed => "405 Method Not Allowed", - .conflict => "409 Conflict", - .internal_server_error => "500 Internal Server Error", - else => "500 Internal Server Error", - }; -} - fn parseEnvInt(comptime T: type, key: []const u8, default: T) T { const val = std.posix.getenv(key) orelse return default; return std.fmt.parseInt(T, val, 10) catch default;