From f64340579e50dc1e14c15c5c591bd2f93dd0c237 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Mon, 2 Mar 2026 05:20:24 -0600 Subject: [PATCH] refactor: extract HTTP API from main.zig, add design doc main.zig was 1,174 lines doing env parsing, component init, signal handling, and the entire HTTP API. now main.zig is 265 lines (entry point + wiring) and api.zig has all handlers, response helpers, and query string parsing. adds docs/design.md covering data flow, threading model, memory model, persistence, and an honest scaling limits table for 10x growth. Co-Authored-By: Claude Opus 4.6 --- build.zig | 1 + docs/design.md | 179 ++++++++++ src/api.zig | 926 +++++++++++++++++++++++++++++++++++++++++++++++++ src/main.zig | 915 +----------------------------------------------- 4 files changed, 1109 insertions(+), 912 deletions(-) create mode 100644 docs/design.md create mode 100644 src/api.zig 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; -- 2.51.2