zzstoatzz.io
search for standard sites
pub-search.waow.tech
search
zig
blog
atproto
65 kB
1567 lines
1const std = @import("std");
2const Io = std.Io;
3const http = std.http;
4const mem = std.mem;
5const json = std.json;
6const Allocator = mem.Allocator;
7const logfire = @import("logfire");
8const zql = @import("zql");
9const zat = @import("zat");
10const db = @import("db.zig");
11const ingest = @import("ingest.zig");
12const metrics = @import("metrics.zig");
13const search = @import("server/search.zig");
14const documents = @import("server/documents.zig");
15const dashboard = @import("server/dashboard.zig");
16const recommended = @import("server/recommended.zig");
17const curators = @import("server/curators.zig");
18const recommenders = @import("server/recommenders.zig");
19const subscribed = @import("server/subscribed.zig");
20const subscribers = @import("server/subscribers.zig");
21const wrapped_ep = @import("server/wrapped.zig");
22const labeler = @import("labeler.zig");
23const classifier = @import("ingest/classifier.zig");
24const policy = @import("policy.zig");
25
26pub const initRecommendedCache = recommended.init;
27pub const initCuratorsCache = curators.init;
28pub const initSubscribedCache = subscribed.init;
29pub const initDashboardCache = dashboard.initCache;
30pub const initTagsCache = TagsCache.init;
31pub const initPopularCache = PopularCache.init;
32
33const server_cache = @import("server/cache.zig");
34
35/// /tags reads local-then-turso live; both stall during sync write bursts
36/// (soak 2026-06-10). Tag aggregates tolerate minutes of staleness.
37const TagsSlot = enum { all };
38
39fn refreshTags(slot: TagsSlot, alloc: Allocator) anyerror![]const u8 {
40 _ = slot;
41 return try getTags(alloc);
42}
43
44const TagsCache = server_cache.WindowedJsonCache(TagsSlot, .{
45 .name = "tags",
46 .refresh = &refreshTags,
47 .interval_ms = 300_000,
48});
49
50/// /popular aggregates search_events live on Turso per request. search_events
51/// is a write-path table not carried in the frozen replica, so it can't be
52/// served locally — but a 7-day popular-query window tolerates minutes of
53/// staleness, so refresh it out of band like the leaderboards.
54const PopularSlot = enum { all };
55
56fn refreshPopular(slot: PopularSlot, alloc: Allocator) anyerror![]const u8 {
57 _ = slot;
58 return try getPopular(alloc, 5);
59}
60
61const PopularCache = server_cache.WindowedJsonCache(PopularSlot, .{
62 .name = "popular",
63 .refresh = &refreshPopular,
64 .interval_ms = 300_000,
65});
66
67const HTTP_BUF_SIZE = 65536;
68const QUERY_PARAM_BUF_SIZE = 64;
69const SEARCH_MAX_LIMIT: usize = 40;
70const SEARCH_MAX_OFFSET: usize = 1000;
71const HYBRID_MAX_WINDOW: usize = 200;
72
73fn microTimestamp(io: Io) i64 {
74 return Io.Timestamp.now(io, .real).toMicroseconds();
75}
76
77pub fn handleConnection(stream: Io.net.Stream, io: Io, accepted_at: i64) void {
78 defer stream.close(io);
79
80 const queue_us = microTimestamp(io) - accepted_at;
81 if (queue_us > 100_000) { // > 100ms
82 logfire.warn("http.queue slow: {d}ms", .{@divTrunc(queue_us, 1000)});
83 }
84
85 var read_buffer: [HTTP_BUF_SIZE]u8 = undefined;
86 var write_buffer: [HTTP_BUF_SIZE]u8 = undefined;
87
88 var reader = stream.reader(io, &read_buffer);
89 var writer = stream.writer(io, &write_buffer);
90
91 var server = http.Server.init(&reader.interface, &writer.interface);
92
93 while (true) {
94 const recv_start = microTimestamp(io);
95 var request = server.receiveHead() catch |err| {
96 if (err != error.HttpConnectionClosing and err != error.EndOfStream) {
97 logfire.debug("http receive error: {}", .{err});
98 }
99 return;
100 };
101 const recv_us = microTimestamp(io) - recv_start;
102 const target = request.head.target;
103
104 const req_span = logfire.span("http.request", .{
105 .target = target,
106 .queue_ms = @divTrunc(queue_us, 1000),
107 .receive_ms = @divTrunc(recv_us, 1000),
108 });
109
110 handleRequest(&server, &request, io) catch |err| {
111 logfire.err("request error: {}", .{err});
112 req_span.end();
113 return;
114 };
115 req_span.end();
116
117 if (!request.head.keep_alive) return;
118 }
119}
120
121fn handleRequest(server: *http.Server, request: *http.Server.Request, io: Io) !void {
122 _ = server;
123 const target = request.head.target;
124
125 if (request.head.method == .OPTIONS) {
126 try sendCorsHeaders(request, "");
127 return;
128 }
129
130 const path = if (mem.indexOf(u8, target, "?")) |qi| target[0..qi] else target;
131
132 if (mem.startsWith(u8, path, "/search")) {
133 try handleSearch(request, target, io);
134 } else if (mem.eql(u8, path, "/tags")) {
135 try handleTags(request, target, io);
136 } else if (mem.eql(u8, path, "/stats")) {
137 try handleStats(request);
138 } else if (mem.eql(u8, path, "/health")) {
139 try sendJson(request, "{\"status\":\"ok\"}");
140 } else if (mem.eql(u8, path, "/popular")) {
141 try handlePopular(request, target, io);
142 } else if (mem.eql(u8, path, "/dashboard")) {
143 try handleDashboard(request);
144 } else if (mem.eql(u8, path, "/api/dashboard")) {
145 try handleDashboardApi(request, io);
146 } else if (mem.eql(u8, path, "/api/timeline")) {
147 try handleTimelineApi(request, target, io);
148 } else if (mem.eql(u8, path, "/api/latency")) {
149 try handleLatencyApi(request, target);
150 } else if (mem.eql(u8, path, "/recommended")) {
151 try handleRecommended(request, target, io);
152 } else if (mem.eql(u8, path, "/recommended-by-top-authors")) {
153 try handleRecommendedByTopAuthors(request, target, io);
154 } else if (mem.eql(u8, path, "/curators")) {
155 try handleCurators(request, target, io);
156 } else if (mem.eql(u8, path, "/recommenders")) {
157 try handleRecommenders(request, target);
158 } else if (mem.eql(u8, path, "/subscribed")) {
159 try handleSubscribed(request, target, io);
160 } else if (mem.eql(u8, path, "/subscribers")) {
161 try handleSubscribers(request, target);
162 } else if (mem.eql(u8, path, "/wrapped")) {
163 try handleWrapped(request, target, io);
164 } else if (mem.eql(u8, path, "/document")) {
165 try handleDocument(request, target);
166 } else if (mem.startsWith(u8, path, "/similar")) {
167 try handleSimilar(request, target, io);
168 } else if (mem.eql(u8, path, "/activity")) {
169 try handleActivity(request, io);
170 } else if (mem.eql(u8, path, "/admin/backfill")) {
171 try handleBackfill(request, target, io);
172 } else if (mem.eql(u8, path, "/admin/reconcile-document")) {
173 try handleReconcileDocument(request, target, io);
174 } else if (mem.eql(u8, path, "/admin/label")) {
175 try handleLabel(request, target);
176 } else if (mem.eql(u8, path, "/api/labeler")) {
177 try handleLabelerSummary(request);
178 } else if (mem.eql(u8, path, "/snapshot")) {
179 try handleSnapshot(request, io);
180 } else {
181 try sendNotFound(request);
182 }
183}
184
185fn handleSearch(request: *http.Server.Request, target: []const u8, io: Io) !void {
186 const start_time = microTimestamp(io);
187
188 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
189 defer arena.deinit();
190 const alloc = arena.allocator();
191
192 const query = parseQueryParam(alloc, target, "q") catch "";
193 const tag_filter = parseQueryParam(alloc, target, "tag") catch null;
194 const platform_filter = parseQueryParam(alloc, target, "platform") catch null;
195 const since_filter = parseQueryParam(alloc, target, "since") catch null;
196 const author_param = parseQueryParam(alloc, target, "author") catch null;
197 const mode_str = parseQueryParam(alloc, target, "mode") catch null;
198 const mode = search.SearchMode.fromString(mode_str);
199 const format = parseQueryParam(alloc, target, "format") catch "v1";
200 const limit_str = parseQueryParam(alloc, target, "limit") catch null;
201 const offset_str = parseQueryParam(alloc, target, "offset") catch null;
202 const requested_limit = if (limit_str) |s| std.fmt.parseInt(usize, s, 10) catch 20 else 20;
203 const limit = @min(@max(requested_limit, 1), SEARCH_MAX_LIMIT);
204 const offset = if (offset_str) |s| std.fmt.parseInt(usize, s, 10) catch 0 else 0;
205
206 if (offset > SEARCH_MAX_OFFSET) {
207 try sendJson(request, "{\"error\":\"offset exceeds maximum of 1000\"}");
208 return;
209 }
210
211 // resolve author param: if it's a handle (not a DID), resolve via AT Protocol
212 const author_filter: ?[]const u8 = if (author_param) |ap| blk: {
213 if (mem.startsWith(u8, ap, "did:")) break :blk ap;
214 break :blk resolveHandle(alloc, ap, io) catch null;
215 } else null;
216
217 // record per-mode latency
218 const timing_endpoint: metrics.timing.Endpoint = switch (mode) {
219 .keyword => .search_keyword,
220 .semantic => .search_semantic,
221 .hybrid => .search_hybrid,
222 };
223 defer metrics.timing.record(timing_endpoint, start_time);
224
225 // span attributes are now copied internally, safe to use arena strings
226 const span = logfire.span("http.search", .{
227 .query = query,
228 .tag = tag_filter,
229 .platform = platform_filter,
230 .author = author_filter,
231 .mode = @tagName(mode),
232 });
233 defer span.end();
234
235 if (query.len == 0 and tag_filter == null and author_filter == null) {
236 try sendJson(request, "{\"error\":\"enter a search term\"}");
237 return;
238 }
239
240 const labeled_pref = parseQueryParam(alloc, target, "labeled") catch null;
241 const show_labeled = labeled_pref != null and mem.eql(u8, labeled_pref.?, "show");
242
243 // Retrieve the full requested prefix plus one policy-visible row. Paging
244 // is then a pure slice of a stable ranking, and that extra row is the
245 // evidence for hasMore.
246 const result_window = std.math.add(usize, offset, limit + 1) catch {
247 try sendJson(request, "{\"error\":\"pagination window is too large\"}");
248 return;
249 };
250 if (mode == .hybrid and offset + limit > HYBRID_MAX_WINDOW) {
251 try sendJson(request, "{\"error\":\"hybrid search supports the top 200 results\"}");
252 return;
253 }
254 const raw_results = search.search(alloc, query, tag_filter, platform_filter, since_filter, author_filter, mode, .{
255 .max_results = result_window,
256 .show_labeled = show_labeled,
257 }) catch |err| {
258 logfire.err("search failed: {}", .{err});
259 metrics.stats.recordError();
260 return err;
261 };
262 metrics.stats.recordSearch(query);
263 logfire.counter("search.requests", 1);
264
265 // label policy: results from bulk-generated-labeled accounts are hidden by
266 // default; `labeled=show` opts in (they come back annotated so the UI can
267 // badge them). Kept accounts are always shown, annotated. Mirrors the
268 // opt-in model of bsky labeler subscriptions, for our own surface.
269 const results = applyLabelPolicy(alloc, raw_results, show_labeled) catch raw_results;
270
271 if (mem.eql(u8, format, "v2")) {
272 const wrapped = try wrapResponse(alloc, results, query, @tagName(mode), limit, offset, false);
273 try sendJson(request, wrapped);
274 } else {
275 // Always slice the bounded candidate array. The old `limit < 40`
276 // shortcut leaked every candidate when limit was exactly 40 (or larger),
277 // so `limit=40` could return 80+ document/base-path/publication rows.
278 const paginated = try paginateJsonArray(alloc, results, limit, offset);
279 try sendJson(request, paginated);
280 }
281}
282
283/// Label policy over a serialized search-result array: rows from labeled
284/// accounts are dropped unless kept or `show`; survivors are annotated
285/// (labeled/kept) so the UI can badge them. One choke point covers every
286/// mode and format; result sets are ≤tens of rows so the reparse is noise.
287fn applyLabelPolicy(alloc: Allocator, body: []const u8, show: bool) ![]const u8 {
288 const root = try json.parseFromSliceLeaky(json.Value, alloc, body, .{});
289 if (root != .array) return body; // error payloads etc. pass through
290
291 var out = json.Array.init(alloc);
292 for (root.array.items) |item| {
293 var result = item;
294 if (zat.json.getString(result, "did")) |did| {
295 if (classifier.isLabeledDid(did)) {
296 const kept = policy.isKept(did);
297 if (!kept and !show) continue;
298 try result.object.put(alloc, "labeled", .{ .bool = true });
299 try result.object.put(alloc, "kept", .{ .bool = kept });
300 }
301 }
302 try out.append(result);
303 }
304 return json.Stringify.valueAlloc(alloc, json.Value{ .array = out }, .{});
305}
306
307fn handleTags(request: *http.Server.Request, target: []const u8, io: Io) !void {
308 const start_time = microTimestamp(io);
309 defer metrics.timing.record(.tags, start_time);
310
311 const span = logfire.span("http.tags", .{});
312 defer span.end();
313
314 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
315 defer arena.deinit();
316 const alloc = arena.allocator();
317
318 const format = parseQueryParam(alloc, target, "format") catch "v1";
319 // background-refreshed snapshot; live only before the first refresh.
320 const tags = if (TagsCache.snapshot(.all, alloc) catch null) |body|
321 body
322 else
323 try getTags(alloc);
324
325 if (mem.eql(u8, format, "v2")) {
326 const wrapped = try wrapResponse(alloc, tags, "", "tags", 100, 0, true);
327 try sendJson(request, wrapped);
328 } else {
329 try sendJson(request, tags);
330 }
331}
332
333fn handlePopular(request: *http.Server.Request, target: []const u8, io: Io) !void {
334 const start_time = microTimestamp(io);
335 defer metrics.timing.record(.popular, start_time);
336
337 const span = logfire.span("http.popular", .{});
338 defer span.end();
339
340 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
341 defer arena.deinit();
342 const alloc = arena.allocator();
343
344 const format = parseQueryParam(alloc, target, "format") catch "v1";
345 // background-refreshed snapshot; before the first fill, live-query and
346 // degrade to [] on a transient Turso failure (handler-only, never cached).
347 const popular = if (PopularCache.snapshot(.all, alloc) catch null) |body|
348 body
349 else
350 getPopular(alloc, 5) catch "[]";
351
352 if (mem.eql(u8, format, "v2")) {
353 const wrapped = try wrapResponse(alloc, popular, "", "popular", 100, 0, true);
354 try sendJson(request, wrapped);
355 } else {
356 try sendJson(request, popular);
357 }
358}
359
360// --- tags/popular query logic ---
361
362const TagJson = struct { tag: []const u8, count: i64 };
363const PopularJson = struct { query: []const u8, count: i64 };
364
365// Tag row matches both the Turso and local-SQLite query shape (SELECT tag, count …).
366// Using a named struct + zql.Query.fromRow means adding/removing a column
367// becomes a compile error rather than a runtime miscount.
368const TagsQuery = zql.Query(dashboard.TAGS_SQL);
369const TagsRow = struct { tag: []const u8, count: i64 };
370
371fn getTags(alloc: Allocator) ![]const u8 {
372 // try local SQLite first (faster)
373 if (db.getLocalDb()) |local| {
374 if (getTagsLocal(alloc, local)) |result| {
375 return result;
376 } else |_| {}
377 }
378
379 // fall back to Turso
380 const c = db.getClient() orelse return error.NotInitialized;
381
382 var output: std.Io.Writer.Allocating = .init(alloc);
383 errdefer output.deinit();
384
385 var res = c.query(TagsQuery.positional, &.{}) catch {
386 try output.writer.writeAll("{\"error\":\"failed to fetch tags\"}");
387 return try output.toOwnedSlice();
388 };
389 defer res.deinit();
390
391 var jw: json.Stringify = .{ .writer = &output.writer };
392 try jw.beginArray();
393 for (res.rows) |row| {
394 const r = TagsQuery.fromRow(TagsRow, row);
395 try jw.write(TagJson{ .tag = r.tag, .count = r.count });
396 }
397 try jw.endArray();
398 return try output.toOwnedSlice();
399}
400
401fn getTagsLocal(alloc: Allocator, local: *db.LocalDb) ![]const u8 {
402 var output: std.Io.Writer.Allocating = .init(alloc);
403 errdefer output.deinit();
404
405 var rows = try local.query(dashboard.TAGS_SQL, .{});
406 defer rows.deinit();
407
408 var jw: json.Stringify = .{ .writer = &output.writer };
409 try jw.beginArray();
410 while (rows.next()) |row| {
411 const r = TagsQuery.fromRow(TagsRow, row);
412 try jw.write(TagJson{ .tag = r.tag, .count = r.count });
413 }
414 try jw.endArray();
415 return try output.toOwnedSlice();
416}
417
418// Window for the popular-searches aggregation. 7 days strikes a balance:
419// long enough for low-traffic queries to show up, short enough that test/
420// seed traffic from weeks ago doesn't dominate. Seeded historical events
421// fully age out 14 days after migration (see migration 013 for the seed
422// strategy).
423const POPULAR_WINDOW_SECS: i64 = 7 * 24 * 60 * 60;
424
425// Aggregate distinct events in the window. Fast: idx_search_events_at
426// makes the date filter range-scan, and the post-filter set is small
427// (at the current ~50 searches/day rate, 7d ≈ 350 events to group).
428const PopularQuery = zql.Query(
429 \\SELECT query, COUNT(*) AS n
430 \\FROM search_events
431 \\WHERE at >= strftime('%s', 'now') - ?
432 \\GROUP BY query
433 \\ORDER BY n DESC, query
434 \\LIMIT ?
435);
436const PopularRow = struct { query: []const u8, n: i64 };
437
438fn getPopular(alloc: Allocator, limit: usize) ![]const u8 {
439 const c = db.getClient() orelse return error.NotInitialized;
440
441 var output: std.Io.Writer.Allocating = .init(alloc);
442 errdefer output.deinit();
443
444 var lim_buf: [8]u8 = undefined;
445 const limit_str = std.fmt.bufPrint(&lim_buf, "{d}", .{limit}) catch "5";
446 var win_buf: [16]u8 = undefined;
447 const window_str = std.fmt.bufPrint(&win_buf, "{d}", .{POPULAR_WINDOW_SECS}) catch "604800";
448
449 // Propagate query failures: the cache refresh path (refreshPopular) must
450 // see the error so WindowedJsonCache keeps the previous good body rather
451 // than poisoning it with [] for a full interval. Cold-start degradation to
452 // [] is handled at the handler call site, before the first cache fill.
453 var res = try c.query(PopularQuery.positional, &.{ window_str, limit_str });
454 defer res.deinit();
455
456 var jw: json.Stringify = .{ .writer = &output.writer };
457 try jw.beginArray();
458 for (res.rows) |row| {
459 const r = PopularQuery.fromRow(PopularRow, row);
460 try jw.write(PopularJson{ .query = r.query, .count = r.n });
461 }
462 try jw.endArray();
463 return try output.toOwnedSlice();
464}
465
466/// Thin wrapper around `server/recommended.zig`. Parses query params,
467/// pulls a cache snapshot (or live-queries for author-filtered views,
468/// which can't reasonably be cached). Slices for pagination, returns JSON.
469fn handleRecommended(request: *http.Server.Request, target: []const u8, io: Io) !void {
470 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
471 defer arena.deinit();
472 const alloc = arena.allocator();
473
474 const start_time = microTimestamp(io);
475 defer metrics.timing.record(.recommended, start_time);
476
477 const span = logfire.span("http.recommended", .{});
478 defer span.end();
479
480 const limit_str = parseQueryParam(alloc, target, "limit") catch null;
481 const offset_str = parseQueryParam(alloc, target, "offset") catch null;
482 const since_str = parseQueryParam(alloc, target, "since") catch null;
483 const sort_str = parseQueryParam(alloc, target, "sort") catch null;
484 const author_param = parseQueryParam(alloc, target, "author") catch null;
485 const curator_param = parseQueryParam(alloc, target, "curator") catch null;
486 const limit: usize = if (limit_str) |s| std.fmt.parseInt(usize, s, 10) catch 20 else 20;
487 const offset: usize = if (offset_str) |s| std.fmt.parseInt(usize, s, 10) catch 0 else 0;
488 const window = recommended.Window.fromString(since_str);
489 const sort = recommended.Sort.fromString(sort_str);
490 span.setAttribute("window", window.slug());
491 span.setAttribute("sort", sort.slug());
492
493 // resolve author / curator → DID (accept either form, matches search.zig's pattern)
494 const resolveActor = struct {
495 fn run(alloc_: std.mem.Allocator, ap_: []const u8, io_: Io) ?[]const u8 {
496 if (mem.startsWith(u8, ap_, "did:")) return ap_;
497 return resolveHandle(alloc_, ap_, io_) catch null;
498 }
499 }.run;
500 const author_did: ?[]const u8 = if (author_param) |ap| resolveActor(alloc, ap, io) else null;
501 const curator_did: ?[]const u8 = if (curator_param) |cp| resolveActor(alloc, cp, io) else null;
502 if (author_did) |d| span.setAttribute("author", d);
503 if (curator_did) |d| span.setAttribute("curator", d);
504
505 // curator + author both set: curator wins (more specific intent — "what
506 // has X recommended" is narrower than "what has Y written"). Avoids
507 // surprising empty results from intersecting two filters.
508 const filter: recommended.Filter = .{
509 .author_did = if (curator_did != null) null else author_did,
510 .curator_did = curator_did,
511 };
512
513 var body: []u8 = undefined;
514 if (filter.author_did != null or filter.curator_did != null) {
515 // filtered queries bypass the cache — narrow scope means sub-100ms
516 // Turso turnaround, and per-(filter, window, sort) cache slots
517 // would explode the working set.
518 span.setAttribute("cache", "bypass");
519 body = try alloc.dupe(u8, try recommended.fetch(alloc, window, sort, filter));
520 } else {
521 var snapshot = try recommended.snapshot(sort, window, alloc);
522 if (snapshot != null) {
523 span.setAttribute("cache", "hit");
524 } else {
525 // cold fallback — refresh thread hasn't populated this slot yet.
526 span.setAttribute("cache", "cold");
527 snapshot = try alloc.dupe(u8, try recommended.fetch(alloc, window, sort, .{}));
528 }
529 body = snapshot.?;
530 }
531
532 const sliced = try recommended.sliceJson(alloc, body, limit, offset);
533 try sendJson(request, sliced);
534}
535
536/// /recommended-by-top-authors — what have the network's top-N writers
537/// (by all-time recommends received) themselves recommended in `since=`?
538/// A transitive-taste signal distinct from raw popularity. Tunable resolution
539/// via `pool=` (how many authors form the taste-pool — small = sharp focal,
540/// large = broader consensus) and `since=` (day/week/month/year/all).
541fn handleRecommendedByTopAuthors(request: *http.Server.Request, target: []const u8, io: Io) !void {
542 const start_time = microTimestamp(io);
543 defer metrics.timing.record(.recommended_top_authors, start_time);
544
545 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
546 defer arena.deinit();
547 const alloc = arena.allocator();
548
549 const span = logfire.span("http.recommended_by_top_authors", .{});
550 defer span.end();
551
552 const limit_str = parseQueryParam(alloc, target, "limit") catch null;
553 const offset_str = parseQueryParam(alloc, target, "offset") catch null;
554 const since_str = parseQueryParam(alloc, target, "since") catch null;
555 const pool_str = parseQueryParam(alloc, target, "pool") catch null;
556 const limit: usize = if (limit_str) |s| std.fmt.parseInt(usize, s, 10) catch 10 else 10;
557 const offset: usize = if (offset_str) |s| std.fmt.parseInt(usize, s, 10) catch 0 else 0;
558 const window = recommended.Window.fromString(since_str);
559 // Clamp pool to a sane range. <1 is meaningless; >500 is wider than the
560 // active recommender set and just slows things down for no signal change.
561 const pool_raw: i64 = if (pool_str) |s| std.fmt.parseInt(i64, s, 10) catch 10 else 10;
562 const pool: i64 = @max(1, @min(500, pool_raw));
563 span.setAttribute("window", window.slug());
564 span.setAttribute("pool", pool);
565
566 const body = try recommended.fetchTopAuthorCascade(alloc, window, pool);
567 const sliced = try recommended.sliceJson(alloc, body, limit, offset);
568 try sendJson(request, sliced);
569}
570
571/// /curators — leaderboard of recommenders (DIDs), windowed by `since=`.
572/// Same cache + slice + cold-fallback shape as /recommended, minus the
573/// sort + author dimensions (curators has one natural metric).
574fn handleCurators(request: *http.Server.Request, target: []const u8, io: Io) !void {
575 const start_time = microTimestamp(io);
576 defer metrics.timing.record(.curators, start_time);
577
578 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
579 defer arena.deinit();
580 const alloc = arena.allocator();
581
582 const span = logfire.span("http.curators", .{});
583 defer span.end();
584
585 const limit_str = parseQueryParam(alloc, target, "limit") catch null;
586 const offset_str = parseQueryParam(alloc, target, "offset") catch null;
587 const since_str = parseQueryParam(alloc, target, "since") catch null;
588 const limit: usize = if (limit_str) |s| std.fmt.parseInt(usize, s, 10) catch 20 else 20;
589 const offset: usize = if (offset_str) |s| std.fmt.parseInt(usize, s, 10) catch 0 else 0;
590 const window = recommended.Window.fromString(since_str);
591 span.setAttribute("window", window.slug());
592
593 var snapshot = try curators.Cache.snapshot(window, alloc);
594 if (snapshot != null) {
595 span.setAttribute("cache", "hit");
596 } else {
597 span.setAttribute("cache", "cold");
598 snapshot = try alloc.dupe(u8, try curators.fetch(alloc, window));
599 }
600
601 const sliced = try curators.sliceJson(alloc, snapshot.?, limit, offset);
602 try sendJson(request, sliced);
603}
604
605/// /recommenders?document=<at-uri> — the recommender DIDs behind one doc's
606/// count. Opens up the COUNT(DISTINCT did) aggregate so the UI can show who,
607/// not just how many. Recency-ordered, deduped by did. No cache (per-document
608/// keyspace is unbounded; the lookup is an indexed point query).
609fn handleRecommenders(request: *http.Server.Request, target: []const u8) !void {
610 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
611 defer arena.deinit();
612 const alloc = arena.allocator();
613
614 const span = logfire.span("http.recommenders", .{});
615 defer span.end();
616
617 const document = parseQueryParam(alloc, target, "document") catch null;
618 if (document == null or document.?.len == 0) {
619 try sendJson(request, "{\"error\":\"missing document param\"}");
620 return;
621 }
622 span.setAttribute("document", document.?);
623
624 const body = try recommenders.fetch(alloc, document.?);
625 try sendJson(request, body);
626}
627
628/// /subscribed — subscription leaderboards. `view=publications|people`,
629/// windowed by `since=`. Same cache + slice + cold-fallback shape as
630/// /recommended. No author/curator filters — the two views ARE the two axes.
631fn handleSubscribed(request: *http.Server.Request, target: []const u8, io: Io) !void {
632 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
633 defer arena.deinit();
634 const alloc = arena.allocator();
635
636 const start_time = microTimestamp(io);
637 defer metrics.timing.record(.subscribed, start_time);
638
639 const span = logfire.span("http.subscribed", .{});
640 defer span.end();
641
642 const limit_str = parseQueryParam(alloc, target, "limit") catch null;
643 const offset_str = parseQueryParam(alloc, target, "offset") catch null;
644 const since_str = parseQueryParam(alloc, target, "since") catch null;
645 const view_str = parseQueryParam(alloc, target, "view") catch null;
646 const limit: usize = if (limit_str) |s| std.fmt.parseInt(usize, s, 10) catch 20 else 20;
647 const offset: usize = if (offset_str) |s| std.fmt.parseInt(usize, s, 10) catch 0 else 0;
648 const window = subscribed.Window.fromString(since_str);
649 const view = subscribed.View.fromString(view_str);
650 span.setAttribute("window", window.slug());
651 span.setAttribute("view", view.slug());
652
653 var snapshot = try subscribed.snapshot(view, window, alloc);
654 if (snapshot != null) {
655 span.setAttribute("cache", "hit");
656 } else {
657 span.setAttribute("cache", "cold");
658 snapshot = try alloc.dupe(u8, try subscribed.fetch(alloc, view, window));
659 }
660
661 const sliced = try subscribed.sliceJson(alloc, snapshot.?, limit, offset);
662 try sendJson(request, sliced);
663}
664
665/// /subscribers — the subscriber DIDs behind one publication's (or one
666/// owner's) count. `?publication=<at-uri>` or `?did=<owner-did>`. Opens up the
667/// COUNT(DISTINCT did) aggregate so the UI can show who, not just how many —
668/// this is the "who's subscribed to me" surface. No cache (per-scope keyspace
669/// is unbounded; the lookup is an indexed point query).
670fn handleSubscribers(request: *http.Server.Request, target: []const u8) !void {
671 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
672 defer arena.deinit();
673 const alloc = arena.allocator();
674
675 const span = logfire.span("http.subscribers", .{});
676 defer span.end();
677
678 const publication = parseQueryParam(alloc, target, "publication") catch null;
679 const owner = parseQueryParam(alloc, target, "did") catch null;
680
681 const scope: subscribers.Scope = if (publication != null and publication.?.len > 0)
682 .{ .publication = publication.? }
683 else if (owner != null and owner.?.len > 0)
684 .{ .owner = owner.? }
685 else {
686 try sendJson(request, "{\"error\":\"missing publication or did param\"}");
687 return;
688 };
689 switch (scope) {
690 .publication => |v| span.setAttribute("publication", v),
691 .owner => |v| span.setAttribute("owner", v),
692 }
693
694 const body = try subscribers.fetch(alloc, scope);
695 try sendJson(request, body);
696}
697
698/// /wrapped?did=<did> or ?handle=<handle> — one identity's standing across the
699/// standard.site graph (publisher / curator / reader lenses). Local-replica
700/// only; resolves a handle to a DID first. No cache (per-DID keyspace).
701fn handleWrapped(request: *http.Server.Request, target: []const u8, io: Io) !void {
702 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
703 defer arena.deinit();
704 const alloc = arena.allocator();
705
706 const span = logfire.span("http.wrapped", .{});
707 defer span.end();
708
709 const did_param = parseQueryParam(alloc, target, "did") catch null;
710 const handle_param = parseQueryParam(alloc, target, "handle") catch null;
711
712 const did: ?[]const u8 = if (did_param != null and did_param.?.len > 0) blk: {
713 break :blk did_param;
714 } else if (handle_param != null and handle_param.?.len > 0) blk: {
715 if (mem.startsWith(u8, handle_param.?, "did:")) break :blk handle_param;
716 break :blk resolveHandle(alloc, handle_param.?, io) catch null;
717 } else null;
718
719 if (did == null or did.?.len == 0) {
720 try sendJson(request, "{\"error\":\"missing or unresolvable did/handle\"}");
721 return;
722 }
723 span.setAttribute("did", did.?);
724
725 const body = try wrapped_ep.fetch(alloc, did.?);
726 try sendJson(request, body);
727}
728
729fn parseQueryParam(alloc: std.mem.Allocator, target: []const u8, param: []const u8) ![]const u8 {
730 // look for ?param= or ¶m=
731 const patterns = [_][]const u8{ "?", "&" };
732 for (patterns) |prefix| {
733 var search_buf: [QUERY_PARAM_BUF_SIZE]u8 = undefined;
734 const search_str = std.fmt.bufPrint(&search_buf, "{s}{s}=", .{ prefix, param }) catch continue;
735 if (mem.indexOf(u8, target, search_str)) |idx| {
736 const encoded = target[idx + search_str.len ..];
737 const end = mem.indexOf(u8, encoded, "&") orelse encoded.len;
738 const query_encoded = encoded[0..end];
739 const buf = try alloc.dupe(u8, query_encoded);
740 // decode + as space (form-urlencoded), then percent-decode
741 for (buf) |*c| {
742 if (c.* == '+') c.* = ' ';
743 }
744 return std.Uri.percentDecodeInPlace(buf);
745 }
746 }
747 return error.NotFound;
748}
749
750fn handleStats(request: *http.Server.Request) !void {
751 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
752 defer arena.deinit();
753 const alloc = arena.allocator();
754
755 const db_stats = metrics.stats.getStats();
756 const all_timing = metrics.timing.getAllStats();
757
758 var output: std.Io.Writer.Allocating = .init(alloc);
759 errdefer output.deinit();
760
761 var jw: json.Stringify = .{ .writer = &output.writer };
762 try jw.beginObject();
763
764 // db stats
765 try jw.objectField("documents");
766 try jw.write(db_stats.documents);
767 try jw.objectField("publications");
768 try jw.write(db_stats.publications);
769 try jw.objectField("embeddings");
770 try jw.write(db_stats.embeddings);
771 try jw.objectField("searches");
772 try jw.write(db_stats.searches);
773 try jw.objectField("errors");
774 try jw.write(db_stats.errors);
775 try jw.objectField("started_at");
776 try jw.write(db_stats.started_at);
777 try jw.objectField("cache_hits");
778 try jw.write(db_stats.cache_hits);
779 try jw.objectField("cache_misses");
780 try jw.write(db_stats.cache_misses);
781
782 // timing stats per endpoint
783 try jw.objectField("timing");
784 try jw.beginObject();
785 inline for (@typeInfo(metrics.timing.Endpoint).@"enum".fields, 0..) |field, i| {
786 const t = all_timing[i];
787 try jw.objectField(field.name);
788 try jw.beginObject();
789 try jw.objectField("count");
790 try jw.write(t.count);
791 try jw.objectField("avg_ms");
792 try jw.write(t.avg_ms);
793 try jw.objectField("p50_ms");
794 try jw.write(t.p50_ms);
795 try jw.objectField("p95_ms");
796 try jw.write(t.p95_ms);
797 try jw.objectField("p99_ms");
798 try jw.write(t.p99_ms);
799 try jw.objectField("max_ms");
800 try jw.write(t.max_ms);
801 try jw.endObject();
802 }
803 try jw.endObject();
804
805 try jw.endObject();
806
807 try sendJson(request, try output.toOwnedSlice());
808}
809
810/// Only one backfill may run at a time: each one writes to turso, and N
811/// concurrent admin backfills saturating turso is a self-inflicted outage
812/// (the 2026-06-10 purge lesson). Excess requests get 409.
813var backfill_busy = std.atomic.Value(bool).init(false);
814
815fn backfillWorker(io: Io, did: []u8, collection: ?[]u8) void {
816 defer backfill_busy.store(false, .release);
817 defer std.heap.page_allocator.free(did);
818 defer if (collection) |c| std.heap.page_allocator.free(c);
819
820 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
821 defer arena.deinit();
822 _ = ingest.ingester.backfillRepo(arena.allocator(), io, did, collection) catch |err| {
823 logfire.warn("backfill: {s} failed: {s}", .{ did, @errorName(err) });
824 };
825}
826
827/// On-demand backfill of a single repo, bypassing the ingester's serial resync queue.
828/// `POST /admin/backfill?did=<did>[&collection=<nsid>]`. Pulls every record of
829/// our collections straight from the author's PDS through the normal
830/// extract+index path. Guarded by BACKFILL_TOKEN (?token=) when that env is set.
831/// Responds 202 and runs in the background (fly's proxy drops long-held
832/// connections; big repos take minutes). `&sync=1` keeps the old blocking
833/// behavior and returns counts. Completion signal either way is the logfire
834/// line `backfill: <did> done`.
835// Emit (or negate) a labeler account-label. Gated by BACKFILL_TOKEN (same admin
836// secret as /admin/backfill). To retract a label, pass neg=1 with the same
837// did+val — per the atproto spec, consumers stop hydrating the original.
838// GET /admin/label?token=…&did=did:plc:…&val=bulk-mirror&neg=0
839fn handleLabel(request: *http.Server.Request, target: []const u8) !void {
840 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
841 defer arena.deinit();
842 const alloc = arena.allocator();
843
844 const json_hdr: []const http.Header = &.{.{ .name = "content-type", .value = "application/json" }};
845
846 if (std.c.getenv("BACKFILL_TOKEN")) |tok_c| {
847 const provided = parseQueryParam(alloc, target, "token") catch "";
848 if (!mem.eql(u8, provided, std.mem.span(tok_c))) {
849 try request.respond("{\"error\":\"unauthorized\"}", .{ .status = .unauthorized, .extra_headers = json_hdr });
850 return;
851 }
852 }
853
854 const did = parseQueryParam(alloc, target, "did") catch {
855 try request.respond("{\"error\":\"missing did param\"}", .{ .status = .bad_request, .extra_headers = json_hdr });
856 return;
857 };
858 const val = parseQueryParam(alloc, target, "val") catch labeler.LABEL_BULK_GENERATED;
859 const neg = blk: {
860 const v = parseQueryParam(alloc, target, "neg") catch break :blk false;
861 break :blk mem.eql(u8, v, "1") or mem.eql(u8, v, "true");
862 };
863
864 const seq = labeler.emit(did, val, neg) catch |err| {
865 const msg = if (err == error.NotConfigured)
866 "{\"error\":\"labeler not configured (LABELER_DID/LABELER_SECRET_KEY unset)\"}"
867 else
868 "{\"error\":\"emit failed\"}";
869 try request.respond(msg, .{ .status = .internal_server_error, .extra_headers = json_hdr });
870 return;
871 };
872
873 // a negation is the operator overruling the model — keep the classifier's
874 // book in sync so /labels reflects the retraction and the DID never re-flags.
875 if (neg and mem.eql(u8, val, labeler.LABEL_BULK_GENERATED)) classifier.markNegated(did);
876
877 const body = try std.fmt.allocPrint(alloc, "{{\"ok\":true,\"seq\":{d},\"did\":\"{s}\",\"val\":\"{s}\",\"neg\":{}}}", .{ seq, did, val, neg });
878 try request.respond(body, .{ .extra_headers = json_hdr });
879}
880
881// Read-only labeler summary for the /labels heads-up page (counts by state +
882// every decided author with score + title patterns). Public — the data is the
883// labels we already publish.
884fn handleLabelerSummary(request: *http.Server.Request) !void {
885 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
886 defer arena.deinit();
887 const body = classifier.writeSummaryJson(arena.allocator()) catch {
888 try sendJson(request, "{\"counts\":{},\"authors\":[]}");
889 return;
890 };
891 try sendJson(request, body);
892}
893
894fn handleBackfill(request: *http.Server.Request, target: []const u8, io: Io) !void {
895 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
896 defer arena.deinit();
897 const alloc = arena.allocator();
898
899 if (std.c.getenv("BACKFILL_TOKEN")) |tok_c| {
900 const expected = std.mem.span(tok_c);
901 const provided = parseQueryParam(alloc, target, "token") catch "";
902 if (!mem.eql(u8, provided, expected)) {
903 try request.respond("{\"error\":\"unauthorized\"}", .{
904 .status = .unauthorized,
905 .extra_headers = &.{.{ .name = "content-type", .value = "application/json" }},
906 });
907 return;
908 }
909 }
910
911 const did = parseQueryParam(alloc, target, "did") catch {
912 try request.respond("{\"error\":\"missing did param\"}", .{
913 .status = .bad_request,
914 .extra_headers = &.{.{ .name = "content-type", .value = "application/json" }},
915 });
916 return;
917 };
918 const collection: ?[]const u8 = parseQueryParam(alloc, target, "collection") catch null;
919
920 const sync_mode = blk: {
921 const v = parseQueryParam(alloc, target, "sync") catch break :blk false;
922 break :blk mem.eql(u8, v, "1");
923 };
924
925 if (!sync_mode) {
926 if (backfill_busy.cmpxchgStrong(false, true, .acq_rel, .acquire) != null) {
927 try request.respond("{\"error\":\"a backfill is already running\"}", .{
928 .status = .conflict,
929 .extra_headers = &.{.{ .name = "content-type", .value = "application/json" }},
930 });
931 return;
932 }
933
934 const did_owned = std.heap.page_allocator.dupe(u8, did) catch {
935 backfill_busy.store(false, .release);
936 return error.OutOfMemory;
937 };
938 const coll_owned: ?[]u8 = if (collection) |c|
939 std.heap.page_allocator.dupe(u8, c) catch {
940 std.heap.page_allocator.free(did_owned);
941 backfill_busy.store(false, .release);
942 return error.OutOfMemory;
943 }
944 else
945 null;
946
947 const thread = std.Thread.spawn(.{}, backfillWorker, .{ io, did_owned, coll_owned }) catch {
948 std.heap.page_allocator.free(did_owned);
949 if (coll_owned) |c| std.heap.page_allocator.free(c);
950 backfill_busy.store(false, .release);
951 try request.respond("{\"error\":\"failed to start backfill\"}", .{
952 .status = .internal_server_error,
953 .extra_headers = &.{.{ .name = "content-type", .value = "application/json" }},
954 });
955 return;
956 };
957 thread.detach();
958
959 const body = try std.fmt.allocPrint(alloc, "{{\"did\":\"{s}\",\"status\":\"accepted\"}}", .{did});
960 try request.respond(body, .{
961 .status = .accepted,
962 .extra_headers = &.{.{ .name = "content-type", .value = "application/json" }},
963 });
964 return;
965 }
966
967 const counts = ingest.ingester.backfillRepo(alloc, io, did, collection) catch |err| {
968 const body = try std.fmt.allocPrint(alloc, "{{\"error\":\"backfill failed: {s}\"}}", .{@errorName(err)});
969 try request.respond(body, .{
970 .status = .internal_server_error,
971 .extra_headers = &.{.{ .name = "content-type", .value = "application/json" }},
972 });
973 return;
974 };
975
976 const body = try std.fmt.allocPrint(
977 alloc,
978 "{{\"did\":\"{s}\",\"documents\":{d},\"publications\":{d},\"recommends\":{d},\"subscriptions\":{d},\"skipped\":{d}}}",
979 .{ did, counts.documents, counts.publications, counts.recommends, counts.subscriptions, counts.skipped },
980 );
981 try sendJson(request, body);
982}
983
984/// Apply one pre-audited site.standard.document ledger item. This endpoint is
985/// intentionally synchronous and item-scoped: the operator controls pacing,
986/// while the backend re-fetches the source and enforces the expected CID at
987/// the last possible moment before an upsert. Deletes require a fresh,
988/// definitive source 400/404.
989fn handleReconcileDocument(request: *http.Server.Request, target: []const u8, io: Io) !void {
990 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
991 defer arena.deinit();
992 const alloc = arena.allocator();
993 const json_hdr: []const http.Header = &.{.{ .name = "content-type", .value = "application/json" }};
994
995 const tok_c = std.c.getenv("BACKFILL_TOKEN") orelse {
996 try request.respond("{\"error\":\"reconciliation endpoint disabled\"}", .{
997 .status = .service_unavailable,
998 .extra_headers = json_hdr,
999 });
1000 return;
1001 };
1002 const provided = parseQueryParam(alloc, target, "token") catch "";
1003 if (!mem.eql(u8, provided, std.mem.span(tok_c))) {
1004 try request.respond("{\"error\":\"unauthorized\"}", .{ .status = .unauthorized, .extra_headers = json_hdr });
1005 return;
1006 }
1007
1008 const did = parseQueryParam(alloc, target, "did") catch {
1009 try request.respond("{\"error\":\"missing did\"}", .{ .status = .bad_request, .extra_headers = json_hdr });
1010 return;
1011 };
1012 const collection = parseQueryParam(alloc, target, "collection") catch {
1013 try request.respond("{\"error\":\"missing collection\"}", .{ .status = .bad_request, .extra_headers = json_hdr });
1014 return;
1015 };
1016 const rkey = parseQueryParam(alloc, target, "rkey") catch {
1017 try request.respond("{\"error\":\"missing rkey\"}", .{ .status = .bad_request, .extra_headers = json_hdr });
1018 return;
1019 };
1020 const pds = parseQueryParam(alloc, target, "pds") catch {
1021 try request.respond("{\"error\":\"missing pds\"}", .{ .status = .bad_request, .extra_headers = json_hdr });
1022 return;
1023 };
1024 const expected_cid: ?[]const u8 = parseQueryParam(alloc, target, "expected_cid") catch null;
1025 const observe_classifier = blk: {
1026 const value = parseQueryParam(alloc, target, "observe_classifier") catch break :blk false;
1027 break :blk mem.eql(u8, value, "1") or mem.eql(u8, value, "true");
1028 };
1029 const action_text = parseQueryParam(alloc, target, "action") catch "";
1030 const action: ingest.ingester.TargetedAction = if (mem.eql(u8, action_text, "upsert"))
1031 .upsert
1032 else if (mem.eql(u8, action_text, "delete"))
1033 .delete
1034 else {
1035 try request.respond("{\"error\":\"action must be upsert or delete\"}", .{ .status = .bad_request, .extra_headers = json_hdr });
1036 return;
1037 };
1038
1039 const result = ingest.ingester.applyDocumentReconciliation(
1040 alloc, io, did, collection, rkey, pds, expected_cid, action, observe_classifier,
1041 ) catch |err| {
1042 const body = try std.fmt.allocPrint(alloc, "{{\"error\":\"{s}\"}}", .{@errorName(err)});
1043 try request.respond(body, .{ .status = .service_unavailable, .extra_headers = json_hdr });
1044 return;
1045 };
1046 const cid = result.source_cid orelse "";
1047 const body = try std.fmt.allocPrint(
1048 alloc,
1049 "{{\"outcome\":\"{s}\",\"source_cid\":\"{s}\"}}",
1050 .{ @tagName(result.outcome), cid },
1051 );
1052 try request.respond(body, .{ .extra_headers = json_hdr });
1053}
1054
1055/// Serve the live replica's manifest sidecar (build id, sha256, watermark,
1056/// counts). This is the watchdog's snapshot-age signal and a human's
1057/// "what is prod actually serving" answer. 404 until the first adoption.
1058fn handleSnapshot(request: *http.Server.Request, io: Io) !void {
1059 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
1060 defer arena.deinit();
1061 const alloc = arena.allocator();
1062
1063 const live = if (std.c.getenv("LOCAL_DB_PATH")) |p| std.mem.span(p) else "/data/local.db";
1064 const sidecar = try std.fmt.allocPrint(alloc, "{s}.manifest.json", .{live});
1065
1066 const file = Io.Dir.openFileAbsolute(io, sidecar, .{}) catch {
1067 try request.respond("{\"error\":\"no snapshot manifest (pre-adoption replica)\"}", .{
1068 .status = .not_found,
1069 .extra_headers = &.{.{ .name = "content-type", .value = "application/json" }},
1070 });
1071 return;
1072 };
1073 defer file.close(io);
1074
1075 var buf: [16 * 1024]u8 = undefined;
1076 const n = file.readStreaming(io, &.{&buf}) catch |err| switch (err) {
1077 error.EndOfStream => 0,
1078 else => return err,
1079 };
1080 try sendJson(request, try alloc.dupe(u8, buf[0..n]));
1081}
1082
1083fn sendJson(request: *http.Server.Request, body: []const u8) !void {
1084 try request.respond(body, .{
1085 .status = .ok,
1086 .extra_headers = &.{
1087 .{ .name = "content-type", .value = "application/json" },
1088 .{ .name = "access-control-allow-origin", .value = "*" },
1089 .{ .name = "access-control-allow-methods", .value = "GET, OPTIONS" },
1090 .{ .name = "access-control-allow-headers", .value = "content-type" },
1091 },
1092 });
1093}
1094
1095fn sendCorsHeaders(request: *http.Server.Request, body: []const u8) !void {
1096 // public read-only API — wildcard origin, no credentials needed.
1097 try request.respond(body, .{
1098 .status = .no_content,
1099 .extra_headers = &.{
1100 .{ .name = "access-control-allow-origin", .value = "*" },
1101 .{ .name = "access-control-allow-methods", .value = "GET, OPTIONS" },
1102 .{ .name = "access-control-allow-headers", .value = "content-type" },
1103 },
1104 });
1105}
1106
1107fn sendNotFound(request: *http.Server.Request) !void {
1108 try request.respond("{\"error\":\"not found\"}", .{
1109 .status = .not_found,
1110 .extra_headers = &.{
1111 .{ .name = "content-type", .value = "application/json" },
1112 .{ .name = "access-control-allow-origin", .value = "*" },
1113 },
1114 });
1115}
1116
1117fn handleDashboardApi(request: *http.Server.Request, io: Io) !void {
1118 const start_time = microTimestamp(io);
1119 defer metrics.timing.record(.dashboard, start_time);
1120
1121 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
1122 defer arena.deinit();
1123 const alloc = arena.allocator();
1124
1125 // background-refreshed snapshot — the page never waits on a live turso
1126 // query. Falls through to a live fetch only before the first refresh.
1127 if (dashboard.ApiCache.snapshot(.main, alloc) catch null) |body| {
1128 try sendJson(request, body);
1129 return;
1130 }
1131
1132 const data = dashboard.fetch(alloc) catch {
1133 try sendJson(request, "{\"error\":\"failed to fetch dashboard data\"}");
1134 return;
1135 };
1136
1137 const json_response = dashboard.toJson(alloc, data) catch {
1138 try sendJson(request, "{\"error\":\"failed to serialize dashboard data\"}");
1139 return;
1140 };
1141
1142 try sendJson(request, json_response);
1143}
1144
1145fn handleTimelineApi(request: *http.Server.Request, target: []const u8, io: Io) !void {
1146 const start_time = microTimestamp(io);
1147 defer metrics.timing.record(.timeline, start_time);
1148
1149 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
1150 defer arena.deinit();
1151 const alloc = arena.allocator();
1152
1153 const range_str = parseQueryParam(alloc, target, "range") catch "30d";
1154 const range = dashboard.TimelineRange.fromString(range_str);
1155
1156 const field_str = parseQueryParam(alloc, target, "field") catch "indexed";
1157 const field = dashboard.TimelineField.fromString(field_str);
1158
1159 // background-refreshed snapshot; live fetch only before the first refresh.
1160 if (dashboard.TimelineCache.snapshot(dashboard.TimelineSlot.from(range, field), alloc) catch null) |body| {
1161 try sendJson(request, body);
1162 return;
1163 }
1164
1165 const json_response = dashboard.fetchTimeline(alloc, range, field) catch {
1166 try sendJson(request, "{\"error\":\"failed to fetch timeline\"}");
1167 return;
1168 };
1169
1170 try sendJson(request, json_response);
1171}
1172
1173fn handleLatencyApi(request: *http.Server.Request, target: []const u8) !void {
1174 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
1175 defer arena.deinit();
1176 const alloc = arena.allocator();
1177
1178 const range_str = parseQueryParam(alloc, target, "range") catch "24h";
1179 const range = metrics.timing.LatencyRange.fromString(range_str);
1180
1181 const json_response = dashboard.fetchLatency(alloc, range) catch {
1182 try sendJson(request, "{\"error\":\"failed to fetch latency\"}");
1183 return;
1184 };
1185
1186 try sendJson(request, json_response);
1187}
1188
1189fn getDashboardUrl() []const u8 {
1190 return if (std.c.getenv("DASHBOARD_URL")) |p| std.mem.span(p) else "https://pub-search.waow.tech/stats";
1191}
1192
1193fn handleDashboard(request: *http.Server.Request) !void {
1194 const dashboard_url = getDashboardUrl();
1195 try request.respond("", .{
1196 .status = .moved_permanently,
1197 .extra_headers = &.{
1198 .{ .name = "location", .value = dashboard_url },
1199 },
1200 });
1201}
1202
1203fn handleDocument(request: *http.Server.Request, target: []const u8) !void {
1204 const json_hdr: []const http.Header = &.{.{ .name = "content-type", .value = "application/json" }};
1205 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
1206 defer arena.deinit();
1207 const alloc = arena.allocator();
1208
1209 const raw = parseQueryParam(alloc, target, "uri") catch {
1210 try request.respond("{\"error\":\"missing uri parameter (comma-separated AT-URIs)\"}", .{ .status = .bad_request, .extra_headers = json_hdr });
1211 return;
1212 };
1213 const uris = documents.splitUris(alloc, raw) catch {
1214 try request.respond("{\"error\":\"too many uris (max 25)\"}", .{ .status = .bad_request, .extra_headers = json_hdr });
1215 return;
1216 };
1217 if (uris.len == 0) {
1218 try request.respond("{\"error\":\"missing uri parameter (comma-separated AT-URIs)\"}", .{ .status = .bad_request, .extra_headers = json_hdr });
1219 return;
1220 }
1221
1222 const span = logfire.span("http.document", .{ .count = uris.len });
1223 defer span.end();
1224
1225 const body = documents.fetch(alloc, uris) catch {
1226 try request.respond("{\"error\":\"replica not ready, retry shortly\"}", .{ .status = .service_unavailable, .extra_headers = json_hdr });
1227 return;
1228 };
1229 try sendJson(request, body);
1230}
1231
1232fn handleSimilar(request: *http.Server.Request, target: []const u8, io: Io) !void {
1233 const start_time = microTimestamp(io);
1234 defer metrics.timing.record(.similar, start_time);
1235
1236 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
1237 defer arena.deinit();
1238 const alloc = arena.allocator();
1239
1240 const uri = parseQueryParam(alloc, target, "uri") catch {
1241 try sendJson(request, "{\"error\":\"missing uri parameter\"}");
1242 return;
1243 };
1244
1245 const format = parseQueryParam(alloc, target, "format") catch "v1";
1246
1247 // span attributes are copied internally, safe to use arena strings
1248 const span = logfire.span("http.similar", .{ .uri = uri });
1249 defer span.end();
1250
1251 const results = search.findSimilar(alloc, uri, 5) catch {
1252 if (mem.eql(u8, format, "v2")) {
1253 try sendJson(request, "{\"results\":[],\"total\":0,\"hasMore\":false}");
1254 } else {
1255 try sendJson(request, "[]");
1256 }
1257 return;
1258 };
1259
1260 if (mem.eql(u8, format, "v2")) {
1261 const wrapped = try wrapResponse(alloc, results, "", "similar", 20, 0, false);
1262 try sendJson(request, wrapped);
1263 } else {
1264 try sendJson(request, results);
1265 }
1266}
1267
1268/// Wrap a result prefix in the v2 page envelope. Search prefixes pass
1269/// total_known=false rather than presenting the prefix length as a corpus
1270/// count; bounded endpoints such as tags can opt into their exact total.
1271fn wrapResponse(alloc: Allocator, array_json: []const u8, query: []const u8, mode: []const u8, limit: usize, offset: usize, total_known: bool) ![]const u8 {
1272 // parse the array to count items and apply pagination
1273 const parsed = json.parseFromSlice(json.Value, alloc, array_json, .{}) catch {
1274 return array_json; // fallback to raw if parse fails
1275 };
1276 defer parsed.deinit();
1277
1278 const items = switch (parsed.value) {
1279 .array => |arr| arr.items,
1280 else => return array_json,
1281 };
1282
1283 const total = items.len;
1284 const start = @min(offset, total);
1285 const end = @min(start + limit, total);
1286 const has_more = end < total;
1287
1288 var output: std.Io.Writer.Allocating = .init(alloc);
1289 errdefer output.deinit();
1290
1291 var jw: json.Stringify = .{ .writer = &output.writer };
1292 try jw.beginObject();
1293
1294 try jw.objectField("results");
1295 try jw.beginArray();
1296 for (items[start..end]) |item| {
1297 try jw.write(item);
1298 }
1299 try jw.endArray();
1300
1301 try jw.objectField("total");
1302 if (total_known) {
1303 try jw.write(total);
1304 } else {
1305 try jw.write(null);
1306 }
1307
1308 try jw.objectField("hasMore");
1309 try jw.write(has_more);
1310
1311 try jw.objectField("nextOffset");
1312 if (has_more) {
1313 try jw.write(end);
1314 } else {
1315 try jw.write(null);
1316 }
1317
1318 if (query.len > 0) {
1319 try jw.objectField("query");
1320 try jw.write(query);
1321 }
1322
1323 if (mode.len > 0) {
1324 try jw.objectField("mode");
1325 try jw.write(mode);
1326 }
1327
1328 try jw.endObject();
1329 return try output.toOwnedSlice();
1330}
1331
1332/// Apply pagination to a JSON array (for v1 format with limit/offset)
1333fn paginateJsonArray(alloc: Allocator, array_json: []const u8, limit: usize, offset: usize) ![]const u8 {
1334 const parsed = json.parseFromSlice(json.Value, alloc, array_json, .{}) catch {
1335 return array_json;
1336 };
1337 defer parsed.deinit();
1338
1339 const items = switch (parsed.value) {
1340 .array => |arr| arr.items,
1341 else => return array_json,
1342 };
1343
1344 const total = items.len;
1345 const start = @min(offset, total);
1346 const end = @min(start + limit, total);
1347
1348 var output: std.Io.Writer.Allocating = .init(alloc);
1349 errdefer output.deinit();
1350
1351 var jw: json.Stringify = .{ .writer = &output.writer };
1352 try jw.beginArray();
1353 for (items[start..end]) |item| {
1354 try jw.write(item);
1355 }
1356 try jw.endArray();
1357 return try output.toOwnedSlice();
1358}
1359
1360test "wrapResponse uses an extra result for truthful page metadata" {
1361 const page = try wrapResponse(std.testing.allocator, "[0,1,2,3,4,5]", "q", "keyword", 2, 2, false);
1362 defer std.testing.allocator.free(page);
1363
1364 const parsed = try json.parseFromSlice(json.Value, std.testing.allocator, page, .{});
1365 defer parsed.deinit();
1366 const obj = parsed.value.object;
1367 try std.testing.expectEqual(@as(usize, 2), obj.get("results").?.array.items.len);
1368 try std.testing.expect(obj.get("total").? == .null);
1369 try std.testing.expect(obj.get("hasMore").?.bool);
1370 try std.testing.expectEqual(@as(i64, 4), obj.get("nextOffset").?.integer);
1371}
1372
1373test "wrapResponse reports total for known complete arrays" {
1374 const page = try wrapResponse(std.testing.allocator, "[0,1,2,3]", "", "tags", 2, 2, true);
1375 defer std.testing.allocator.free(page);
1376
1377 const parsed = try json.parseFromSlice(json.Value, std.testing.allocator, page, .{});
1378 defer parsed.deinit();
1379 const obj = parsed.value.object;
1380 try std.testing.expectEqual(@as(i64, 4), obj.get("total").?.integer);
1381 try std.testing.expect(!obj.get("hasMore").?.bool);
1382 try std.testing.expect(obj.get("nextOffset").? == .null);
1383}
1384
1385test "paginateJsonArray honors limit equal to the search candidate cap" {
1386 const input =
1387 \\[0,1,2,3,4,5,6,7,8,9,10,11,12,13,14,15,16,17,18,19,20,21,22,23,24,25,26,27,28,29,30,31,32,33,34,35,36,37,38,39,40]
1388 ;
1389 const page = try paginateJsonArray(std.testing.allocator, input, 40, 0);
1390 defer std.testing.allocator.free(page);
1391
1392 const parsed = try json.parseFromSlice(json.Value, std.testing.allocator, page, .{});
1393 defer parsed.deinit();
1394 try std.testing.expectEqual(@as(usize, 40), parsed.value.array.items.len);
1395 try std.testing.expectEqual(@as(i64, 39), parsed.value.array.items[39].integer);
1396}
1397
1398test "paginateJsonArray applies offset with an oversized limit" {
1399 const input =
1400 \\[{"id":1},{"id":2},{"id":3},{"id":4},{"id":5}]
1401 ;
1402 const page = try paginateJsonArray(std.testing.allocator, input, 40, 3);
1403 defer std.testing.allocator.free(page);
1404
1405 const parsed = try json.parseFromSlice(json.Value, std.testing.allocator, page, .{});
1406 defer parsed.deinit();
1407 try std.testing.expectEqual(@as(usize, 2), parsed.value.array.items.len);
1408 try std.testing.expectEqual(@as(i64, 4), parsed.value.array.items[0].object.get("id").?.integer);
1409}
1410
1411/// Resolve an AT Protocol handle to a DID via zat's HandleResolver.
1412/// Tries HTTP .well-known first, falls back to DNS-over-HTTPS.
1413fn resolveHandle(alloc: std.mem.Allocator, handle: []const u8, io: Io) ![]const u8 {
1414 const parsed = zat.Handle.parse(handle) orelse {
1415 logfire.warn("resolveHandle: invalid handle: {s}", .{handle});
1416 return error.InvalidHandle;
1417 };
1418
1419 var resolver = zat.HandleResolver.init(io, alloc);
1420 defer resolver.deinit();
1421
1422 if (resolver.resolve(parsed)) |did| {
1423 return did;
1424 } else |err| {
1425 // zat's DoH parser rejects Cloudflare's `Comment` field (present on
1426 // DNSSEC handles like dholms.at), surfacing as InvalidDnsResponse and
1427 // silently dropping the author filter. Fall back to a lenient DoH
1428 // lookup here so DNS-based handles still resolve. See zat note:
1429 // BUG-doh-comment-field.md. Remove once a fixed zat is pinned.
1430 logfire.warn("resolveHandle: zat failed for {s}: {}, trying fallbacks", .{ handle, err });
1431 // HTTP .well-known first: covers *.bsky.social and any handle without a
1432 // `_atproto` DNS TXT record (zat's own HTTP path is unreliable in our
1433 // env — it falls through to DNS and dies on bsky.social's no-Answer
1434 // DoH body). Then DoH for DNS-based handles.
1435 if (resolveHandleWellKnown(alloc, handle, io)) |did| {
1436 return did;
1437 } else |wk_err| {
1438 logfire.warn("resolveHandle: well-known fallback failed for {s}: {}", .{ handle, wk_err });
1439 }
1440 return resolveHandleDoh(alloc, handle, io) catch |fb_err| {
1441 logfire.warn("resolveHandle: DoH fallback failed for {s}: {}", .{ handle, fb_err });
1442 return error.ResolveFailed;
1443 };
1444 }
1445}
1446
1447// HTTP well-known resolution: GET https://<handle>/.well-known/atproto-did,
1448// body is the DID as plain text. Covers bsky.social handles (no DNS TXT).
1449fn resolveHandleWellKnown(alloc: Allocator, handle: []const u8, io: Io) ![]const u8 {
1450 var url_buf: [256]u8 = undefined;
1451 const url = try std.fmt.bufPrint(&url_buf, "https://{s}/.well-known/atproto-did", .{handle});
1452
1453 var client: http.Client = .{ .allocator = alloc, .io = io };
1454 defer client.deinit();
1455
1456 var response_body: std.Io.Writer.Allocating = .init(alloc);
1457 defer response_body.deinit();
1458
1459 const res = client.fetch(.{
1460 .location = .{ .url = url },
1461 .response_writer = &response_body.writer,
1462 }) catch return error.WellKnownRequestFailed;
1463 if (res.status != .ok) return error.WellKnownRequestFailed;
1464
1465 const did = mem.trim(u8, response_body.written(), &std.ascii.whitespace);
1466 if (!mem.startsWith(u8, did, "did:")) return error.NoDidInWellKnown;
1467 return try alloc.dupe(u8, did);
1468}
1469
1470// Lenient DNS-over-HTTPS resolution of `_atproto.<handle>` TXT → did=.
1471// Mirrors zat's resolveDns but parses with ignore_unknown_fields so the
1472// optional `Comment` field Cloudflare adds for DNSSEC zones doesn't abort
1473// the parse (the root cause of dropped author filters for dholms.at et al).
1474fn resolveHandleDoh(alloc: Allocator, handle: []const u8, io: Io) ![]const u8 {
1475 var url_buf: [256]u8 = undefined;
1476 const url = try std.fmt.bufPrint(
1477 &url_buf,
1478 "https://cloudflare-dns.com/dns-query?name=_atproto.{s}&type=TXT",
1479 .{handle},
1480 );
1481
1482 var client: http.Client = .{ .allocator = alloc, .io = io };
1483 defer client.deinit();
1484
1485 var response_body: std.Io.Writer.Allocating = .init(alloc);
1486 defer response_body.deinit();
1487
1488 const res = client.fetch(.{
1489 .location = .{ .url = url },
1490 .extra_headers = &.{.{ .name = "accept", .value = "application/dns-json" }},
1491 .response_writer = &response_body.writer,
1492 }) catch return error.DohRequestFailed;
1493 if (res.status != .ok) return error.DohRequestFailed;
1494
1495 return didFromDohBody(alloc, response_body.written());
1496}
1497
1498const DohAnswer = struct { data: ?[]const u8 = null };
1499const DohResponse = struct { Answer: ?[]DohAnswer = null };
1500
1501// Parse a Cloudflare DoH JSON body and pull the did= out of the TXT answer.
1502// Parses leniently (ignore_unknown_fields) so the optional `Comment` field
1503// Cloudflare emits for DNSSEC zones doesn't abort the parse.
1504fn didFromDohBody(alloc: Allocator, body: []const u8) ![]const u8 {
1505 const parsed = json.parseFromSlice(DohResponse, alloc, body, .{
1506 .ignore_unknown_fields = true,
1507 }) catch return error.InvalidDohResponse;
1508 defer parsed.deinit();
1509
1510 const answers = parsed.value.Answer orelse return error.NoDnsRecords;
1511 for (answers) |answer| {
1512 var data = answer.data orelse continue;
1513 // TXT data arrives quoted, e.g. "\"did=did:plc:...\""
1514 if (data.len >= 2 and data[0] == '"' and data[data.len - 1] == '"') {
1515 data = data[1 .. data.len - 1];
1516 }
1517 const prefix = "did=";
1518 if (mem.startsWith(u8, data, prefix)) {
1519 return try alloc.dupe(u8, data[prefix.len..]);
1520 }
1521 }
1522 return error.NoDidInTxt;
1523}
1524
1525test "didFromDohBody tolerates Cloudflare Comment field (dholms.at regression)" {
1526 // the literal body Cloudflare returns for _atproto.dholms.at — the trailing
1527 // `Comment` field is what broke zat's strict parser and silently dropped the
1528 // author filter (2026-06-05). lenient parse must still extract the did.
1529 const body =
1530 \\{"Status":0,"TC":false,"RD":true,"RA":true,"AD":false,"CD":false,"Question":[{"name":"_atproto.dholms.at","type":16}],"Answer":[{"name":"_atproto.dholms.at","type":16,"TTL":3600,"data":"\"did=did:plc:yk4dd2qkboz2yv6tpubpc6co\""}],"Comment":["EDE(10): RRSIGs Missing for DNSKEY at., id = 1253"]}
1531 ;
1532 const did = try didFromDohBody(std.testing.allocator, body);
1533 defer std.testing.allocator.free(did);
1534 try std.testing.expectEqualStrings("did:plc:yk4dd2qkboz2yv6tpubpc6co", did);
1535}
1536
1537test "didFromDohBody errors when no TXT answer present" {
1538 const body =
1539 \\{"Status":0,"TC":false,"RD":true,"RA":true,"AD":false,"CD":false}
1540 ;
1541 try std.testing.expectError(error.NoDnsRecords, didFromDohBody(std.testing.allocator, body));
1542}
1543
1544fn handleActivity(request: *http.Server.Request, io: Io) !void {
1545 const start_time = microTimestamp(io);
1546 defer metrics.timing.record(.activity, start_time);
1547
1548 const counts = metrics.activity.getCounts();
1549
1550 // format as JSON array manually into buffer
1551 var buf: [512]u8 = undefined;
1552 var pos: usize = 0;
1553 buf[pos] = '[';
1554 pos += 1;
1555 for (counts, 0..) |c, i| {
1556 if (i > 0) {
1557 buf[pos] = ',';
1558 pos += 1;
1559 }
1560 const written = std.fmt.bufPrint(buf[pos..], "{d}", .{c}) catch return;
1561 pos += written.len;
1562 }
1563 buf[pos] = ']';
1564 pos += 1;
1565
1566 try sendJson(request, buf[0..pos]);
1567}