zzstoatzz.io
This repository has no description
event feed: served directly from the canonical log
POST /events/filter and /events/count-by/{countable} are real now —
and they read straight from the repo's io.prefect.v0.event collection:
TID-keyed records decode back into prefect's ReceivedEvent shape,
filtered by occurred range / event prefix / resource id, newest-first,
with EventPage and EventCount envelopes. the histogram buckets by hour
and carries each bucket's min/max occurred as start/end.
this is the page where the architecture stops hiding: the feed the ui
renders IS the signed event log, no derived table in between. (at
scale it wants an appview index — collection scan per request is honest
about being v0.) replaces the empty-envelope fallback that the graph
fix introduced.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
author
zzstoatzz
co-author
Claude Fable 5
date
(Jun 12, 2026, 3:34 PM -0500)
commit
b8a2ad57
b8a2ad571ff5bbeef329f1c73f3fa74e8add8580
parent
30702ee2
30702ee29661f10dbeb96736ac6b23b56a270c4b
+213
-4
+213
-4
src/api.zig
+213
-4
src/api.zig
···
147
147
if (setStateId(path, "/task_runs/")) |id| return self.setState(req, res, id, .task_run);
148
148
if (std.mem.eql(u8, path, "/ui/flow_runs/count-task-runs")) return self.countTaskRunsByFlowRun(req, res);
149
149
if (std.mem.eql(u8, path, "/logs/")) return self.createLogs(req, res);
150
+
if (std.mem.eql(u8, path, "/events/filter")) return self.eventsFilter(req, res);
151
+
if (std.mem.startsWith(u8, path, "/events/count-by/"))
152
+
return self.eventsCountBy(req, res, path["/events/count-by/".len..]);
150
153
if (std.mem.eql(u8, path, "/events")) {
151
154
res.setStatus(.no_content);
152
155
res.body = "";
···
160
163
// render as empty states instead of error toasts. shapes weir
161
164
// actually implements are matched above.
162
165
if (req.method == .POST and std.mem.startsWith(u8, req.url.path, "/api/")) {
163
-
if (std.mem.eql(u8, path, "/events/filter")) {
164
-
res.body = "{\"events\":[],\"total\":0,\"next_page\":null}";
165
-
return;
166
-
}
167
166
if (std.mem.endsWith(u8, path, "/filter") or std.mem.endsWith(u8, path, "/history")) {
168
167
res.body = "[]";
169
168
return;
···
1398
1397
var jw: std.json.Stringify = .{ .writer = &out.writer };
1399
1398
try json_cbor.cborToJson(&jw, sub);
1400
1399
return out.written();
1400
+
}
1401
+
1402
+
// ------------------------------------------------------------------
1403
+
// the event feed — served directly from the canonical log. every
1404
+
// io.prefect.v0.event record decodes back into prefect's ReceivedEvent
1405
+
// shape; TID rkeys make the walk chronological before we reverse it.
1406
+
// ------------------------------------------------------------------
1407
+
1408
+
const FeedEvent = struct {
1409
+
occurred: []const u8,
1410
+
name: []const u8,
1411
+
resource_id: []const u8,
1412
+
resource_json: []const u8,
1413
+
related_json: []const u8,
1414
+
payload_json: []const u8,
1415
+
event_id: []const u8,
1416
+
};
1417
+
1418
+
/// normalize "+00:00" suffixes to "Z" so ISO strings compare correctly
1419
+
fn normTime(ts: []const u8) []const u8 {
1420
+
if (std.mem.endsWith(u8, ts, "+00:00")) return ts[0 .. ts.len - 6];
1421
+
return ts;
1422
+
}
1423
+
1424
+
fn collectFeedEvents(self: *Api, arena: std.mem.Allocator, filter: std.json.Value) ![]FeedEvent {
1425
+
const since: ?[]const u8, const until: ?[]const u8 = blk: {
1426
+
if (filter != .object) break :blk .{ null, null };
1427
+
const occ = filter.object.get("occurred") orelse break :blk .{ null, null };
1428
+
if (occ != .object) break :blk .{ null, null };
1429
+
break :blk .{ getString(occ, "since"), getString(occ, "until") };
1430
+
};
1431
+
const prefixes: ?[]std.json.Value = filterAny(filter, "event", null) orelse blk: {
1432
+
if (filter != .object) break :blk null;
1433
+
const ev = filter.object.get("event") orelse break :blk null;
1434
+
if (ev != .object) break :blk null;
1435
+
const pfx = ev.object.get("prefix") orelse break :blk null;
1436
+
if (pfx != .array) break :blk null;
1437
+
break :blk pfx.array.items;
1438
+
};
1439
+
const resource_ids: ?[]std.json.Value = blk: {
1440
+
if (filter != .object) break :blk null;
1441
+
const any = filter.object.get("any_resource") orelse break :blk null;
1442
+
if (any != .object) break :blk null;
1443
+
const ids = any.object.get("id") orelse break :blk null;
1444
+
if (ids != .array) break :blk null;
1445
+
break :blk ids.array.items;
1446
+
};
1447
+
1448
+
self.lock.lock(self.io) catch return error.LockFailed;
1449
+
defer self.lock.unlock(self.io);
1450
+
1451
+
var out: std.ArrayList(FeedEvent) = .empty;
1452
+
for (try self.repo.listRecords(arena, event_collection)) |rec| {
1453
+
const v = zat.cbor.decodeAll(arena, rec.cbor_data) catch continue;
1454
+
const occurred = v.getString("occurred") orelse continue;
1455
+
const name = v.getString("event") orelse continue;
1456
+
const resource = v.get("resource") orelse continue;
1457
+
const resource_id = resource.getString("prefect.resource.id") orelse "";
1458
+
1459
+
if (since) |t| if (std.mem.order(u8, normTime(occurred), normTime(t)) == .lt) continue;
1460
+
if (until) |t| if (std.mem.order(u8, normTime(occurred), normTime(t)) == .gt) continue;
1461
+
if (prefixes) |pfx| {
1462
+
var hit = false;
1463
+
for (pfx) |pv| {
1464
+
if (pv == .string and std.mem.startsWith(u8, name, pv.string)) hit = true;
1465
+
}
1466
+
if (!hit) continue;
1467
+
}
1468
+
if (resource_ids) |ids| {
1469
+
var hit = false;
1470
+
for (ids) |iv| {
1471
+
if (iv == .string and std.mem.eql(u8, resource_id, iv.string)) hit = true;
1472
+
}
1473
+
if (!hit) continue;
1474
+
}
1475
+
1476
+
try out.append(arena, .{
1477
+
.occurred = occurred,
1478
+
.name = name,
1479
+
.resource_id = resource_id,
1480
+
.resource_json = try cborFieldJson(arena, v, "resource", "{}"),
1481
+
.related_json = try cborFieldJson(arena, v, "related", "[]"),
1482
+
.payload_json = try cborFieldJson(arena, v, "payload", "{}"),
1483
+
.event_id = v.getString("eventId") orelse "",
1484
+
});
1485
+
}
1486
+
return out.items;
1487
+
}
1488
+
1489
+
fn eventsFilter(self: *Api, req: *httpz.Request, res: *httpz.Response) !void {
1490
+
const arena = req.arena;
1491
+
const parsed: std.json.Value = blk: {
1492
+
const body = req.body() orelse break :blk .null;
1493
+
if (body.len == 0) break :blk .null;
1494
+
break :blk std.json.parseFromSliceLeaky(std.json.Value, arena, body, .{}) catch .null;
1495
+
};
1496
+
const filter: std.json.Value = if (parsed == .object)
1497
+
parsed.object.get("filter") orelse .null
1498
+
else
1499
+
.null;
1500
+
const limit: usize = if (parsed == .object)
1501
+
if (parsed.object.get("limit")) |l| (if (l == .integer and l.integer > 0) @intCast(l.integer) else 50) else 50
1502
+
else
1503
+
50;
1504
+
1505
+
const events = try self.collectFeedEvents(arena, filter);
1506
+
std.mem.sort(FeedEvent, events, {}, struct {
1507
+
fn newerFirst(_: void, a: FeedEvent, b: FeedEvent) bool {
1508
+
return std.mem.order(u8, a.occurred, b.occurred) == .gt;
1509
+
}
1510
+
}.newerFirst);
1511
+
1512
+
const n = @min(limit, events.len);
1513
+
var out: std.Io.Writer.Allocating = .init(arena);
1514
+
var jw: std.json.Stringify = .{ .writer = &out.writer };
1515
+
try jw.beginObject();
1516
+
try jw.objectField("events");
1517
+
try jw.beginArray();
1518
+
for (events[0..n]) |e| {
1519
+
try jw.beginObject();
1520
+
try jw.objectField("id");
1521
+
try jw.write(e.event_id);
1522
+
try jw.objectField("occurred");
1523
+
try jw.write(e.occurred);
1524
+
try jw.objectField("event");
1525
+
try jw.write(e.name);
1526
+
try jw.objectField("resource");
1527
+
try writeRaw(&jw, e.resource_json);
1528
+
try jw.objectField("related");
1529
+
try writeRaw(&jw, e.related_json);
1530
+
try jw.objectField("payload");
1531
+
try writeRaw(&jw, e.payload_json);
1532
+
try jw.objectField("received");
1533
+
try jw.write(e.occurred);
1534
+
try jw.objectField("follows");
1535
+
try jw.write(null);
1536
+
try jw.endObject();
1537
+
}
1538
+
try jw.endArray();
1539
+
try jw.objectField("total");
1540
+
try jw.write(events.len);
1541
+
try jw.objectField("next_page");
1542
+
try jw.write(null);
1543
+
try jw.endObject();
1544
+
res.body = out.written();
1545
+
}
1546
+
1547
+
/// POST /events/count-by/{countable}: histogram + group counts.
1548
+
/// buckets carry their min/max occurred as start/end.
1549
+
fn eventsCountBy(self: *Api, req: *httpz.Request, res: *httpz.Response, countable: []const u8) !void {
1550
+
const arena = req.arena;
1551
+
const parsed: std.json.Value = blk: {
1552
+
const body = req.body() orelse break :blk .null;
1553
+
if (body.len == 0) break :blk .null;
1554
+
break :blk std.json.parseFromSliceLeaky(std.json.Value, arena, body, .{}) catch .null;
1555
+
};
1556
+
const filter: std.json.Value = if (parsed == .object)
1557
+
parsed.object.get("filter") orelse .null
1558
+
else
1559
+
.null;
1560
+
1561
+
const events = try self.collectFeedEvents(arena, filter);
1562
+
1563
+
const Bucket = struct {
1564
+
label: []const u8,
1565
+
count: usize,
1566
+
min: []const u8,
1567
+
max: []const u8,
1568
+
};
1569
+
var order: std.ArrayList([]const u8) = .empty;
1570
+
var buckets: std.StringHashMapUnmanaged(Bucket) = .empty;
1571
+
1572
+
const by_time = std.mem.eql(u8, countable, "time") or std.mem.eql(u8, countable, "day");
1573
+
for (events) |e| {
1574
+
const key: []const u8 = if (by_time)
1575
+
e.occurred[0..@min(13, e.occurred.len)] // hour bucket
1576
+
else if (std.mem.eql(u8, countable, "resource"))
1577
+
e.resource_id
1578
+
else
1579
+
e.name;
1580
+
const entry = try buckets.getOrPut(arena, key);
1581
+
if (!entry.found_existing) {
1582
+
entry.value_ptr.* = .{ .label = key, .count = 0, .min = e.occurred, .max = e.occurred };
1583
+
try order.append(arena, key);
1584
+
}
1585
+
entry.value_ptr.count += 1;
1586
+
if (std.mem.order(u8, e.occurred, entry.value_ptr.min) == .lt) entry.value_ptr.min = e.occurred;
1587
+
if (std.mem.order(u8, e.occurred, entry.value_ptr.max) == .gt) entry.value_ptr.max = e.occurred;
1588
+
}
1589
+
1590
+
var out: std.Io.Writer.Allocating = .init(arena);
1591
+
var jw: std.json.Stringify = .{ .writer = &out.writer };
1592
+
try jw.beginArray();
1593
+
for (order.items) |key| {
1594
+
const b = buckets.get(key).?;
1595
+
try jw.beginObject();
1596
+
try jw.objectField("value");
1597
+
try jw.write(key);
1598
+
try jw.objectField("label");
1599
+
try jw.write(b.label);
1600
+
try jw.objectField("count");
1601
+
try jw.write(b.count);
1602
+
try jw.objectField("start_time");
1603
+
try jw.write(b.min);
1604
+
try jw.objectField("end_time");
1605
+
try jw.write(b.max);
1606
+
try jw.endObject();
1607
+
}
1608
+
try jw.endArray();
1609
+
res.body = out.written();
1401
1610
}
1402
1611
1403
1612
// ------------------------------------------------------------------