atproto relay in zig zlay.waow.tech
relay zig atproto
Something went wrong. Try again.
Zig
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170//! HTTP route dispatch and static content.//!//! defines HttpContext (the shared context plumbed from main.zig) and the//! top-level request router that delegates to xrpc and admin handler modules.
const std = @import("std");const Io = std.Io;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/index.zig");const backfill_mod = @import("../collection_index/backfill.zig");const cleaner_mod = @import("../collection_index/cleaner.zig");const resync_mod = @import("../collection_index/resync.zig");const host_ops_mod = @import("../host_ops.zig");const ingest_liveness_mod = @import("../ingest_liveness.zig");const ws_ping_mod = @import("../ws_ping.zig");const util = @import("../util/util.zig");const h = @import("http.zig");const xrpc = @import("xrpc.zig");const admin = @import("admin.zig");
/// 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, cleaner: *cleaner_mod.Cleaner, resyncer: *resync_mod.Resyncer, bc: *broadcaster.Broadcaster, validator: *validator_mod.Validator, host_ops: *host_ops_mod.HostOpsQueue, pool_io: Io, io: Io, db_queue: *event_log_mod.DbRequestQueue, shutdown: *std.atomic.Value(bool), ingest_liveness: *ingest_liveness_mod.IngestLiveness, ws_ping: *ws_ping_mod.WsPing,};
/// top-level HTTP request router — installed as bc.http_fallbackpub 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 { h.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, "/_healthz")) { // liveness: the process is up AND process-wide ingest is not stalled const v = ctx.ingest_liveness.verdict(util.milliTimestamp(ctx.io)); if (v.stalled) { var buf: [512]u8 = undefined; h.respondJson(conn, .service_unavailable, ingest_liveness_mod.formatHealthzStalled(&buf, v)); return; } h.respondJson(conn, .ok, "{\"status\":\"ok\"}"); } else if (std.mem.eql(u8, path, "/_readyz") or std.mem.eql(u8, path, "/_health") or std.mem.eql(u8, path, "/xrpc/_health")) { // readiness — use atomic health flag (pg.Pool runs on Threaded, HTTP handlers are Evented) if (!ctx.persist.isDbHealthy()) { h.respondJson(conn, .internal_server_error, "{\"status\":\"error\",\"msg\":\"database unavailable\"}"); return; } h.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, ctx.bc.io); h.respondJson(conn, .ok, body); } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listRepos")) { xrpc.handleListRepos(conn, query, ctx); } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getRepoStatus")) { xrpc.handleGetRepoStatus(conn, query, ctx); } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getRepo")) { xrpc.handleGetRepo(conn, query, ctx); } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getLatestCommit")) { xrpc.handleGetLatestCommit(conn, query, ctx); } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listReposByCollection")) { xrpc.handleListReposByCollection(conn, query, ctx.collection_index); } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.listHosts")) { xrpc.handleListHosts(conn, query, ctx); } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getHostStatus")) { xrpc.handleGetHostStatus(conn, query, ctx); } else if (std.mem.eql(u8, path, "/admin/hosts")) { admin.handleAdminListHosts(conn, headers, ctx); } else if (std.mem.eql(u8, path, "/admin/backfill-collections")) { admin.handleAdminBackfillStatus(conn, headers, ctx.backfiller); } else if (std.mem.eql(u8, path, "/admin/cleanup-collection-index")) { admin.handleCleanupStatus(conn, headers, ctx.cleaner); } else if (std.mem.eql(u8, path, "/admin/resync")) { admin.handleResyncStatus(conn, headers, ctx.resyncer); } else if (std.mem.eql(u8, path, "/admin/ingest-stall")) { admin.handleIngestStallStatus(conn, headers, ctx); } else if (std.mem.eql(u8, path, "/admin/ws-ping")) { admin.handleWsPingStatus(conn, headers, ctx); } else if (std.mem.eql(u8, path, "/")) { h.respondText(conn, .ok, \\ _ \\ ___| | __ _ _ _ \\|_ / |/ _` | | | | \\ / /| | (_| | |_| | \\/___|_|\__,_|\__, | \\ |___/ \\ \\This is an atproto [https://atproto.com] relay instance, \\running the zlay codebase [https://tangled.org/zat.dev/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")) { h.httpRespond(conn, .ok, "image/svg+xml", \\<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 32 32"> \\<rect width="32" height="32" rx="6" fill="#1a1a2e"/> \\<text x="16" y="24" font-family="monospace" font-size="22" font-weight="bold" fill="#e94560" text-anchor="middle">Z</text> \\</svg> ); } else { h.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")) { admin.handleBan(conn, body, headers, ctx); } else if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.requestCrawl")) { xrpc.handleRequestCrawl(conn, body, ctx); } else if (std.mem.eql(u8, path, "/admin/hosts/block")) { admin.handleAdminBlockHost(conn, body, headers, ctx); } else if (std.mem.eql(u8, path, "/admin/hosts/unblock")) { admin.handleAdminUnblockHost(conn, body, headers, ctx); } else if (std.mem.eql(u8, path, "/admin/hosts/reconnect")) { admin.handleAdminReconnectHost(conn, query, body, headers, ctx); } else if (std.mem.eql(u8, path, "/admin/hosts/changeLimits")) { admin.handleAdminChangeLimits(conn, body, headers, ctx); } else if (std.mem.eql(u8, path, "/admin/backfill-collections")) { admin.handleAdminBackfillTrigger(conn, query, headers, ctx.backfiller); } else if (std.mem.eql(u8, path, "/admin/cleanup-collection-index")) { admin.handleCleanupTrigger(conn, headers, ctx.cleaner); } else if (std.mem.eql(u8, path, "/admin/resync")) { admin.handleResyncTrigger(conn, body, headers, ctx.resyncer); } else if (std.mem.eql(u8, path, "/admin/ingest-stall")) { admin.handleIngestStallUpdate(conn, body, headers, ctx); } else if (std.mem.eql(u8, path, "/admin/ws-ping")) { admin.handleWsPingUpdate(conn, body, headers, ctx); } else { h.respondText(conn, .not_found, "not found"); }}