relay-eval: reshape phi API → /api/relays/* + events log

+312 -34

2 changed files

expand all

relay-eval

src

server.zig

store.zig

+179 -29

relay-eval/src/server.zig

··· 75 75 } else if (std.mem.eql(u8, path, "/api/trend")) { 76 76 const limit = parseLimitParam(query, trend_limit, store); 77 77 try serveTrend(allocator, stream, store, limit); 78 - } else if (std.mem.eql(u8, path, "/api/phi/monitors")) { 79 - try servePhiMonitors(allocator, stream, store); 80 - } else if (std.mem.eql(u8, path, "/api/phi/history")) { 81 - try servePhiHistory(allocator, stream, store, query); 78 + } else if (std.mem.eql(u8, path, "/api/relays")) { 79 + try serveRelays(allocator, stream, store); 80 + } else if (std.mem.eql(u8, path, "/api/relays/history")) { 81 + try serveRelaysHistory(allocator, stream, store, query); 82 + } else if (std.mem.eql(u8, path, "/api/relays/events")) { 83 + try serveRelaysEvents(allocator, stream, store, query); 82 84 } else if (std.mem.eql(u8, path, "/og")) { 83 85 try serveOgPng(allocator, stream, og_png_path); 84 86 } else if (std.mem.eql(u8, path, "/og.svg")) { ··· 459 461 } 460 462 } 461 463 462 - fn servePhiMonitors(allocator: std.mem.Allocator, stream: std.net.Stream, store: *Store) !void { 464 + fn serveRelays(allocator: std.mem.Allocator, stream: std.net.Stream, store: *Store) !void { 463 465 const short = try store.getPerHostStats(allocator, short_window_runs); 464 466 defer { 465 467 for (short) |s| allocator.free(s.host); ··· 479 481 480 482 try json.append(allocator, '['); 481 483 484 + // reusable scratch for building headlines (rebuilt per host) 485 + var headline_buf: std.ArrayList(u8) = .empty; 486 + defer headline_buf.deinit(allocator); 487 + 482 488 for (short, 0..) |host_short, i| { 483 489 if (i > 0) try json.append(allocator, ','); 484 490 485 491 const host_baseline = findSnapshot(baseline, host_short.host); 486 492 const status = classifyMonitor(host_short, host_baseline); 487 493 488 - // look up prior monitor_state; if status differs, update last_changed. 489 - // on disagreement OR no prior state, last_changed = now. 494 + // build the headline into a scratch buffer so it can be both appended 495 + // to the response AND used as the headline on any transition row. 496 + headline_buf.clearRetainingCapacity(); 497 + try writeHeadline(&headline_buf, allocator, host_short.host, status, host_short, host_baseline); 498 + const headline = headline_buf.items; 499 + 500 + // look up prior monitor_state. three cases: 501 + // 1. no prior state → seed, no transition logged 502 + // 2. prior state == current → keep existing last_changed 503 + // 3. prior state != current → update last_changed, log transition 490 504 var last_changed_buf: [32]u8 = undefined; 491 505 const last_changed: []const u8 = blk: { 492 506 const prior = try store.getMonitorState(allocator, host_short.host); ··· 500 514 const copy = std.fmt.bufPrint(&last_changed_buf, "{s}", .{p.last_changed}) catch p.last_changed; 501 515 break :blk copy; 502 516 } 503 - // status transitioned — bump to checked_at 517 + // status transitioned — append to event log AND update state 518 + try store.insertTransition(checked_at, host_short.host, p.status, status.text(), headline); 504 519 try store.upsertMonitorState(host_short.host, status.text(), checked_at); 505 520 const copy = std.fmt.bufPrint(&last_changed_buf, "{s}", .{checked_at}) catch checked_at; 506 521 break :blk copy; 507 522 } else { 508 - // first time seeing this host — seed with current state 523 + // first time seeing this host — seed without logging a transition 524 + // (there's no "from" state to report) 509 525 try store.upsertMonitorState(host_short.host, status.text(), checked_at); 510 526 const copy = std.fmt.bufPrint(&last_changed_buf, "{s}", .{checked_at}) catch checked_at; 511 527 break :blk copy; ··· 513 529 }; 514 530 515 531 // emit monitor object 516 - try json.print(allocator, "{{\"name\":\"{s}\",\"status\":\"{s}\",\"headline\":\"", .{ host_short.host, status.text() }); 517 - try writeHeadline(&json, allocator, host_short.host, status, host_short, host_baseline); 518 - try json.appendSlice(allocator, "\",\"metrics\":{"); 532 + try json.print(allocator, "{{\"name\":\"{s}\",\"status\":\"{s}\",\"headline\":\"{s}\",\"metrics\":{{", .{ 533 + host_short.host, status.text(), headline, 534 + }); 519 535 520 536 // coverage values are fractions (0..~1 for normal runs, can exceed 1 521 537 // transiently during replay). emit as percent with 2 decimals: a ··· 538 554 try sendResponse(stream, "200 OK", "application/json", json.items); 539 555 } 540 556 541 - // --- phi history endpoint --- 557 + // --- relays history endpoint --- 542 558 // 543 - // GET /api/phi/history?name=<host>&limit=<n> 559 + // GET /api/relays/history?name=<host>&limit=<n> 560 + // GET /api/relays/history?name=<host>&since=<iso>&until=<iso> 544 561 // 545 562 // returns per-run coverage history for a single host, oldest → newest, with 546 - // precomputed coverage_pct so phi doesn't have to do math. includes a summary 547 - // block with mean/min/max coverage + connected-run count for the window. 563 + // precomputed coverage_pct. includes a summary block with mean/min/max 564 + // coverage + connected-run count for the window. 565 + // 566 + // when since/until are set, returns every point in the inclusive range 567 + // (no cap). when they're absent, returns the last N runs (default 288 568 + // ≈ 24h, max 2016 ≈ 7d). limit is ignored when since/until are set. 548 569 // 549 - // coverage semantics match /api/phi/monitors: 570 + // coverage semantics match /api/relays: 550 571 // coverage_pct = unique_dids / max(unique_dids in that run) × 100 551 572 // self-normalizing — replay inflates only the replaying host's own numbers. 552 573 ··· 566 587 return null; 567 588 } 568 589 569 - fn servePhiHistory(allocator: std.mem.Allocator, stream: std.net.Stream, store: *Store, query: []const u8) !void { 590 + // permissive check that a query value looks like an ISO 8601 UTC timestamp 591 + // (e.g. "2026-04-17T06:36:30Z"). string-comparable against stored 592 + // `runs.timestamp` values which use the same format. 593 + fn isValidIsoTimestamp(s: []const u8) bool { 594 + if (s.len == 0 or s.len > 32) return false; 595 + for (s) |ch| { 596 + const ok = (ch >= '0' and ch <= '9') or ch == '-' or ch == ':' or 597 + ch == 'T' or ch == 'Z' or ch == '.' or ch == '+'; 598 + if (!ok) return false; 599 + } 600 + return true; 601 + } 602 + 603 + // hostnames we might plausibly track: alphanumerics, dots, hyphens. 604 + // cheap validation before binding into SQL. 605 + fn isValidHostname(s: []const u8) bool { 606 + if (s.len == 0 or s.len > 253) return false; 607 + for (s) |ch| { 608 + const ok = (ch >= 'a' and ch <= 'z') or (ch >= 'A' and ch <= 'Z') or 609 + (ch >= '0' and ch <= '9') or ch == '.' or ch == '-'; 610 + if (!ok) return false; 611 + } 612 + return true; 613 + } 614 + 615 + fn serveRelaysHistory(allocator: std.mem.Allocator, stream: std.net.Stream, store: *Store, query: []const u8) !void { 570 616 const name = parseQueryString(query, "name") orelse { 571 617 try sendResponse(stream, "400 Bad Request", "application/json", "{\"error\":\"missing required query parameter: name\"}"); 572 618 return; 573 619 }; 620 + if (!isValidHostname(name)) { 621 + try sendResponse(stream, "400 Bad Request", "application/json", "{\"error\":\"invalid name\"}"); 622 + return; 623 + } 574 624 575 - // validate name: only hostnames we might plausibly track (alphanumerics, 576 - // dots, hyphens). cheap prevention against weird injection attempts. 577 - for (name) |ch| { 578 - const ok = (ch >= 'a' and ch <= 'z') or (ch >= 'A' and ch <= 'Z') or 579 - (ch >= '0' and ch <= '9') or ch == '.' or ch == '-'; 580 - if (!ok) { 581 - try sendResponse(stream, "400 Bad Request", "application/json", "{\"error\":\"invalid name\"}"); 582 - return; 583 - } 625 + const since_raw = parseQueryString(query, "since") orelse ""; 626 + const until_raw = parseQueryString(query, "until") orelse ""; 627 + if (since_raw.len > 0 and !isValidIsoTimestamp(since_raw)) { 628 + try sendResponse(stream, "400 Bad Request", "application/json", "{\"error\":\"invalid since (expected ISO 8601)\"}"); 629 + return; 630 + } 631 + if (until_raw.len > 0 and !isValidIsoTimestamp(until_raw)) { 632 + try sendResponse(stream, "400 Bad Request", "application/json", "{\"error\":\"invalid until (expected ISO 8601)\"}"); 633 + return; 584 634 } 585 635 586 636 const limit: u32 = if (parseQueryString(query, "limit")) |l| ··· 588 638 else 589 639 default_history_limit; 590 640 591 - const points = try store.getHostHistory(allocator, name, limit); 641 + const points = try store.getHostHistory(allocator, name, limit, since_raw, until_raw); 592 642 defer { 593 643 for (points) |p| allocator.free(p.timestamp); 594 644 allocator.free(points); ··· 617 667 var json: std.ArrayList(u8) = .empty; 618 668 defer json.deinit(allocator); 619 669 620 - try json.print(allocator, "{{\"name\":\"{s}\",\"limit\":{d},\"points\":[", .{ name, limit }); 670 + try json.print(allocator, "{{\"name\":\"{s}\"", .{name}); 671 + if (since_raw.len > 0 or until_raw.len > 0) { 672 + try json.print(allocator, ",\"since\":\"{s}\",\"until\":\"{s}\"", .{ since_raw, until_raw }); 673 + } else { 674 + try json.print(allocator, ",\"limit\":{d}", .{limit}); 675 + } 676 + try json.appendSlice(allocator, ",\"points\":["); 677 + 621 678 for (points, 0..) |p, i| { 622 679 if (i > 0) try json.append(allocator, ','); 623 680 const pct_hundredths: u32 = @intFromFloat(std.math.clamp(p.coverage, 0, 10.0) * 10000.0 + 0.5); ··· 640 697 connected_count, points.len, 641 698 }, 642 699 ); 700 + 701 + try sendResponse(stream, "200 OK", "application/json", json.items); 702 + } 703 + 704 + // --- relays events endpoint --- 705 + // 706 + // GET /api/relays/events 707 + // GET /api/relays/events?since=<iso>&until=<iso>&name=<host> 708 + // 709 + // returns status-transition log rows, oldest → newest. complements the 710 + // snapshot (/api/relays) and per-host timeseries (/api/relays/history) by 711 + // exposing the "what changed, when" axis directly. headlines are stored 712 + // at transition time and returned verbatim so callers can quote them. 713 + // 714 + // default window is last 24h. name filter is optional; omitted = all relays. 715 + 716 + fn formatIsoEpoch(epoch: i64, buf: []u8) []const u8 { 717 + const es = std.time.epoch.EpochSeconds{ .secs = @intCast(epoch) }; 718 + const day = es.getEpochDay(); 719 + const yd = day.calculateYearDay(); 720 + const md = yd.calculateMonthDay(); 721 + const ds = es.getDaySeconds(); 722 + return std.fmt.bufPrint(buf, "{d}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}Z", .{ 723 + yd.year, 724 + @as(u32, @intFromEnum(md.month)), 725 + @as(u32, md.day_index) + 1, 726 + ds.getHoursIntoDay(), 727 + ds.getMinutesIntoHour(), 728 + ds.getSecondsIntoMinute(), 729 + }) catch "1970-01-01T00:00:00Z"; 730 + } 731 + 732 + fn serveRelaysEvents(allocator: std.mem.Allocator, stream: std.net.Stream, store: *Store, query: []const u8) !void { 733 + var since_buf: [32]u8 = undefined; 734 + var until_buf: [32]u8 = undefined; 735 + 736 + const since = blk: { 737 + if (parseQueryString(query, "since")) |raw| { 738 + if (!isValidIsoTimestamp(raw)) { 739 + try sendResponse(stream, "400 Bad Request", "application/json", "{\"error\":\"invalid since (expected ISO 8601)\"}"); 740 + return; 741 + } 742 + break :blk raw; 743 + } 744 + // default: 24h ago 745 + break :blk formatIsoEpoch(std.time.timestamp() - 86400, &since_buf); 746 + }; 747 + 748 + const until = blk: { 749 + if (parseQueryString(query, "until")) |raw| { 750 + if (!isValidIsoTimestamp(raw)) { 751 + try sendResponse(stream, "400 Bad Request", "application/json", "{\"error\":\"invalid until (expected ISO 8601)\"}"); 752 + return; 753 + } 754 + break :blk raw; 755 + } 756 + // default: now 757 + break :blk formatIsoEpoch(std.time.timestamp(), &until_buf); 758 + }; 759 + 760 + const name = parseQueryString(query, "name") orelse ""; 761 + if (name.len > 0 and !isValidHostname(name)) { 762 + try sendResponse(stream, "400 Bad Request", "application/json", "{\"error\":\"invalid name\"}"); 763 + return; 764 + } 765 + 766 + const transitions = try store.getTransitions(allocator, since, until, name); 767 + defer { 768 + for (transitions) |t| { 769 + allocator.free(t.ts); 770 + allocator.free(t.name); 771 + if (t.from_status) |fs| allocator.free(fs); 772 + allocator.free(t.to_status); 773 + allocator.free(t.headline); 774 + } 775 + allocator.free(transitions); 776 + } 777 + 778 + var json: std.ArrayList(u8) = .empty; 779 + defer json.deinit(allocator); 780 + 781 + try json.append(allocator, '['); 782 + for (transitions, 0..) |t, i| { 783 + if (i > 0) try json.append(allocator, ','); 784 + try json.print(allocator, "{{\"ts\":\"{s}\",\"name\":\"{s}\",", .{ t.ts, t.name }); 785 + if (t.from_status) |fs| { 786 + try json.print(allocator, "\"from_status\":\"{s}\",", .{fs}); 787 + } else { 788 + try json.appendSlice(allocator, "\"from_status\":null,"); 789 + } 790 + try json.print(allocator, "\"to_status\":\"{s}\",\"headline\":\"{s}\"}}", .{ t.to_status, t.headline }); 791 + } 792 + try json.append(allocator, ']'); 643 793 644 794 try sendResponse(stream, "200 OK", "application/json", json.items); 645 795 }

+133 -5

relay-eval/src/store.zig

··· 61 61 \\ status TEXT NOT NULL, 62 62 \\ last_changed TEXT NOT NULL 63 63 \\); 64 + \\CREATE TABLE IF NOT EXISTS monitor_transitions ( 65 + \\ id INTEGER PRIMARY KEY, 66 + \\ ts TEXT NOT NULL, 67 + \\ name TEXT NOT NULL, 68 + \\ from_status TEXT, 69 + \\ to_status TEXT NOT NULL, 70 + \\ headline TEXT NOT NULL 71 + \\); 72 + \\CREATE INDEX IF NOT EXISTS idx_monitor_transitions_ts ON monitor_transitions(ts); 73 + \\CREATE INDEX IF NOT EXISTS idx_monitor_transitions_name_ts ON monitor_transitions(name, ts); 64 74 ; 65 75 try self.exec(sql); 66 76 } ··· 448 458 connected: bool, 449 459 }; 450 460 451 - /// get per-run history for a single host, oldest → newest, limited to 452 - /// the last N valid runs. used by /api/phi/history. 453 - pub fn getHostHistory(self: *Store, allocator: std.mem.Allocator, host: []const u8, last_n: u32) ![]HostHistoryPoint { 454 - const sql = 461 + /// get per-run history for a single host, oldest → newest. 462 + /// when since/until are non-empty, filters by timestamp range and 463 + /// ignores last_n (returns every point in range, no cap). 464 + /// when since/until are empty, returns the last_n most recent points. 465 + pub fn getHostHistory( 466 + self: *Store, 467 + allocator: std.mem.Allocator, 468 + host: []const u8, 469 + last_n: u32, 470 + since: []const u8, 471 + until: []const u8, 472 + ) ![]HostHistoryPoint { 473 + const use_range = since.len > 0 or until.len > 0; 474 + const sql = if (use_range) 475 + \\SELECT r.id, r.timestamp, rs.unique_dids, rs.events, rs.connected, 476 + \\ CAST(rs.unique_dids AS REAL) / NULLIF(mpr.m, 0) AS coverage 477 + \\ FROM runs r 478 + \\ JOIN relay_stats rs ON rs.run_id = r.id 479 + \\ JOIN ( 480 + \\ SELECT run_id, MAX(unique_dids) AS m 481 + \\ FROM relay_stats 482 + \\ GROUP BY run_id 483 + \\ ) mpr ON mpr.run_id = r.id 484 + \\ WHERE rs.host = ? 485 + \\ AND r.timestamp >= ? 486 + \\ AND r.timestamp <= ? 487 + \\ AND r.id IN (SELECT run_id FROM relay_stats GROUP BY run_id HAVING SUM(connected) > COUNT(*) / 2) 488 + \\ ORDER BY r.id ASC 489 + else 455 490 \\SELECT r.id, r.timestamp, rs.unique_dids, rs.events, rs.connected, 456 491 \\ CAST(rs.unique_dids AS REAL) / NULLIF(mpr.m, 0) AS coverage 457 492 \\ FROM runs r ··· 469 504 \\ ) 470 505 \\ ORDER BY r.id ASC 471 506 ; 507 + 472 508 var stmt: ?*c.sqlite3_stmt = null; 473 509 if (c.sqlite3_prepare_v2(self.db, sql, -1, &stmt, null) != c.SQLITE_OK) { 474 510 return self.sqlError(); ··· 476 512 defer _ = c.sqlite3_finalize(stmt); 477 513 478 514 _ = c.sqlite3_bind_text(stmt, 1, host.ptr, @intCast(host.len), SQLITE_STATIC); 479 - _ = c.sqlite3_bind_int(stmt, 2, @intCast(last_n)); 515 + if (use_range) { 516 + const s = if (since.len > 0) since else "0000"; 517 + const u = if (until.len > 0) until else "9999"; 518 + _ = c.sqlite3_bind_text(stmt, 2, s.ptr, @intCast(s.len), SQLITE_STATIC); 519 + _ = c.sqlite3_bind_text(stmt, 3, u.ptr, @intCast(u.len), SQLITE_STATIC); 520 + } else { 521 + _ = c.sqlite3_bind_int(stmt, 2, @intCast(last_n)); 522 + } 480 523 481 524 var out: std.ArrayList(HostHistoryPoint) = .empty; 482 525 while (c.sqlite3_step(stmt) == c.SQLITE_ROW) { ··· 487 530 .events = @intCast(c.sqlite3_column_int64(stmt, 3)), 488 531 .connected = c.sqlite3_column_int(stmt, 4) != 0, 489 532 .coverage = c.sqlite3_column_double(stmt, 5), 533 + }); 534 + } 535 + return out.toOwnedSlice(allocator); 536 + } 537 + 538 + pub const Transition = struct { 539 + ts: []const u8, 540 + name: []const u8, 541 + from_status: ?[]const u8, 542 + to_status: []const u8, 543 + headline: []const u8, 544 + }; 545 + 546 + /// append a status-transition row. called from the relays endpoint 547 + /// whenever a monitor's status changes (not on first observation). 548 + pub fn insertTransition( 549 + self: *Store, 550 + ts: []const u8, 551 + name: []const u8, 552 + from_status: ?[]const u8, 553 + to_status: []const u8, 554 + headline: []const u8, 555 + ) !void { 556 + const sql = "INSERT INTO monitor_transitions (ts, name, from_status, to_status, headline) VALUES (?, ?, ?, ?, ?)"; 557 + var stmt: ?*c.sqlite3_stmt = null; 558 + if (c.sqlite3_prepare_v2(self.db, sql, -1, &stmt, null) != c.SQLITE_OK) { 559 + return self.sqlError(); 560 + } 561 + defer _ = c.sqlite3_finalize(stmt); 562 + 563 + _ = c.sqlite3_bind_text(stmt, 1, ts.ptr, @intCast(ts.len), SQLITE_STATIC); 564 + _ = c.sqlite3_bind_text(stmt, 2, name.ptr, @intCast(name.len), SQLITE_STATIC); 565 + if (from_status) |f| { 566 + _ = c.sqlite3_bind_text(stmt, 3, f.ptr, @intCast(f.len), SQLITE_STATIC); 567 + } else { 568 + _ = c.sqlite3_bind_null(stmt, 3); 569 + } 570 + _ = c.sqlite3_bind_text(stmt, 4, to_status.ptr, @intCast(to_status.len), SQLITE_STATIC); 571 + _ = c.sqlite3_bind_text(stmt, 5, headline.ptr, @intCast(headline.len), SQLITE_STATIC); 572 + 573 + if (c.sqlite3_step(stmt) != c.SQLITE_DONE) { 574 + return self.sqlError(); 575 + } 576 + } 577 + 578 + /// fetch transitions in [since, until], ordered ascending by ts. 579 + /// when name is empty, returns all relays; otherwise filters to that relay. 580 + pub fn getTransitions( 581 + self: *Store, 582 + allocator: std.mem.Allocator, 583 + since: []const u8, 584 + until: []const u8, 585 + name: []const u8, 586 + ) ![]Transition { 587 + const sql = 588 + \\SELECT ts, name, from_status, to_status, headline 589 + \\ FROM monitor_transitions 590 + \\ WHERE ts >= ? 591 + \\ AND ts <= ? 592 + \\ AND (? = '' OR name = ?) 593 + \\ ORDER BY ts ASC, id ASC 594 + ; 595 + var stmt: ?*c.sqlite3_stmt = null; 596 + if (c.sqlite3_prepare_v2(self.db, sql, -1, &stmt, null) != c.SQLITE_OK) { 597 + return self.sqlError(); 598 + } 599 + defer _ = c.sqlite3_finalize(stmt); 600 + 601 + _ = c.sqlite3_bind_text(stmt, 1, since.ptr, @intCast(since.len), SQLITE_STATIC); 602 + _ = c.sqlite3_bind_text(stmt, 2, until.ptr, @intCast(until.len), SQLITE_STATIC); 603 + _ = c.sqlite3_bind_text(stmt, 3, name.ptr, @intCast(name.len), SQLITE_STATIC); 604 + _ = c.sqlite3_bind_text(stmt, 4, name.ptr, @intCast(name.len), SQLITE_STATIC); 605 + 606 + var out: std.ArrayList(Transition) = .empty; 607 + while (c.sqlite3_step(stmt) == c.SQLITE_ROW) { 608 + const from_status: ?[]const u8 = if (c.sqlite3_column_type(stmt, 2) == c.SQLITE_NULL) 609 + null 610 + else 611 + try self.colText(allocator, stmt, 2); 612 + try out.append(allocator, .{ 613 + .ts = try self.colText(allocator, stmt, 0), 614 + .name = try self.colText(allocator, stmt, 1), 615 + .from_status = from_status, 616 + .to_status = try self.colText(allocator, stmt, 3), 617 + .headline = try self.colText(allocator, stmt, 4), 490 618 }); 491 619 } 492 620 return out.toOwnedSlice(allocator);

Configure Feed

Configure Feed