zzstoatzz.io
This repository has no description
1//! minimal prefect 3.x REST surface over the atproto repo + appview cache.
2//!
3//! the write path is the whole idea:
4//! api mutation -> record into the MST -> signed commit -> CAR on disk
5//! -> appview (burner-redis) updated for reads
6//!
7//! orchestration policy is intentionally absent: every proposed state is
8//! ACCEPTed. this server demonstrates the storage shape, not the rule engine.
9
10const std = @import("std");
11const httpz = @import("httpz");
12const zat = @import("zat");
13
14const repo_mod = @import("repo.zig");
15const cache_mod = @import("cache.zig");
16const json_cbor = @import("json_cbor.zig");
17const util = @import("util.zig");
18
19const log = std.log.scoped(.api);
20
21const flow_collection = "io.prefect.v0.flow";
22const flow_run_collection = "io.prefect.v0.flowRun";
23const task_run_collection = "io.prefect.v0.taskRun";
24const event_collection = "io.prefect.v0.event";
25const log_batch_collection = "io.prefect.v0.logBatch";
26
27pub const Api = struct {
28 allocator: std.mem.Allocator,
29 io: std.Io,
30 repo: *repo_mod.Repo,
31 cache: *cache_mod.Cache,
32 lock: std.Io.Mutex = .init,
33 /// commit cadence: sign+persist every N record writes. 1 = every write
34 /// is durably committed before its response (strictest). higher values
35 /// amortize commit cost at the price of a durability window — the same
36 /// lever as the cloud prototype's delta-flush cadence.
37 commit_every: usize = 1,
38 dirty: usize = 0,
39
40 pub fn handle(self: *Api, req: *httpz.Request, res: *httpz.Response) void {
41 self.dispatch(req, res) catch |err| {
42 log.err("{s} {s}: {}", .{ @tagName(req.method), req.url.path, err });
43 res.setStatus(.internal_server_error);
44 res.content_type = .JSON;
45 res.body = "{\"detail\":\"internal error\"}";
46 };
47 }
48
49 fn dispatch(self: *Api, req: *httpz.Request, res: *httpz.Response) !void {
50 var path = req.url.path;
51 if (std.mem.startsWith(u8, path, "/api/")) path = path[4..];
52
53 res.content_type = .JSON;
54
55 switch (req.method) {
56 .GET => {
57 if (std.mem.eql(u8, path, "/health")) return self.sendRaw(res, .ok, "true");
58 if (std.mem.eql(u8, path, "/version") or std.mem.eql(u8, path, "/admin/version"))
59 return self.sendRaw(res, .ok, "\"3.7.2\"");
60 if (std.mem.eql(u8, path, "/csrf-token")) return csrfToken(req, res);
61 if (std.mem.eql(u8, path, "/admin/settings"))
62 return self.sendRaw(res, .ok, "{\"server\":{\"analytics_enabled\":false}}");
63 if (std.mem.eql(u8, path, "/ready")) return self.sendRaw(res, .ok, "true");
64
65 if (pathId(path, "/flows/")) |id| return self.getEntity(req, res, "flow", id);
66 if (pathId(path, "/flow_runs/")) |id| return self.getFlowRun(req, res, id);
67 if (pathId(path, "/task_runs/")) |id| return self.getTaskRun(req, res, id);
68
69 if (std.mem.eql(u8, path, "/xrpc/com.atproto.sync.getRepo")) return self.getRepoCar(res);
70 if (std.mem.eql(u8, path, "/xrpc/com.atproto.repo.describeRepo")) return self.describeRepo(req, res);
71 if (std.mem.eql(u8, path, "/xrpc/com.atproto.repo.listRecords")) return self.listRecords(req, res);
72 },
73 .POST => {
74 if (std.mem.eql(u8, path, "/flows/")) return self.createFlow(req, res);
75 if (std.mem.eql(u8, path, "/flow_runs/")) return self.createFlowRun(req, res);
76 if (std.mem.eql(u8, path, "/task_runs/")) return self.createTaskRun(req, res);
77 if (std.mem.eql(u8, path, "/flows/filter")) return self.filterFlows(req, res);
78 if (std.mem.eql(u8, path, "/flow_runs/filter")) return self.filterRuns(req, res, .flow_run, false);
79 if (std.mem.eql(u8, path, "/flow_runs/count")) return self.filterRuns(req, res, .flow_run, true);
80 if (std.mem.eql(u8, path, "/task_runs/filter")) return self.filterRuns(req, res, .task_run, false);
81 if (std.mem.eql(u8, path, "/task_runs/count")) return self.filterRuns(req, res, .task_run, true);
82 if (setStateId(path, "/flow_runs/")) |id| return self.setState(req, res, id, .flow_run);
83 if (setStateId(path, "/task_runs/")) |id| return self.setState(req, res, id, .task_run);
84 if (std.mem.eql(u8, path, "/logs/")) return self.createLogs(req, res);
85 if (std.mem.eql(u8, path, "/events")) {
86 res.setStatus(.no_content);
87 res.body = "";
88 return;
89 }
90 },
91 else => {},
92 }
93
94 res.setStatus(.not_found);
95 res.body = "{\"detail\":\"not found\"}";
96 }
97
98 fn sendRaw(_: *Api, res: *httpz.Response, status: std.http.Status, body: []const u8) !void {
99 res.setStatus(status);
100 res.body = body;
101 }
102
103 fn csrfToken(req: *httpz.Request, res: *httpz.Response) !void {
104 const client = (try req.query()).get("client") orelse "unknown";
105 try res.json(.{
106 .token = "prefect-pds-csrf-token",
107 .client = client,
108 .expiration = "2099-01-01T00:00:00Z",
109 }, .{});
110 }
111
112 /// "/flows/{id}" -> id, rejecting deeper paths
113 fn pathId(path: []const u8, comptime prefix: []const u8) ?[]const u8 {
114 if (!std.mem.startsWith(u8, path, prefix)) return null;
115 const rest = path[prefix.len..];
116 if (rest.len == 0 or std.mem.indexOfScalar(u8, rest, '/') != null) return null;
117 return rest;
118 }
119
120 /// "/flow_runs/{id}/set_state" -> id
121 fn setStateId(path: []const u8, comptime prefix: []const u8) ?[]const u8 {
122 if (!std.mem.startsWith(u8, path, prefix)) return null;
123 const rest = path[prefix.len..];
124 const suffix = "/set_state";
125 if (!std.mem.endsWith(u8, rest, suffix)) return null;
126 const id = rest[0 .. rest.len - suffix.len];
127 if (id.len == 0 or std.mem.indexOfScalar(u8, id, '/') != null) return null;
128 return id;
129 }
130
131 // ------------------------------------------------------------------
132 // rows: the appview's internal storage format. raw_* fields hold
133 // verbatim JSON from the client, spliced back raw into responses.
134 // ------------------------------------------------------------------
135
136 const RunRow = struct {
137 id: []const u8,
138 created: []const u8,
139 updated: []const u8,
140 name: []const u8,
141 flow_id: []const u8 = "",
142 flow_run_id: ?[]const u8 = null,
143 task_key: []const u8 = "",
144 dynamic_key: []const u8 = "",
145 state_id: []const u8 = "",
146 state_type: []const u8 = "PENDING",
147 state_name: []const u8 = "Pending",
148 state_timestamp: []const u8 = "",
149 state_message: ?[]const u8 = null,
150 raw_state_details: []const u8 = "{}",
151 raw_parameters: []const u8 = "{}",
152 raw_tags: []const u8 = "[]",
153 raw_empirical_policy: []const u8 = "{}",
154 raw_job_variables: []const u8 = "{}",
155 raw_task_inputs: []const u8 = "{}",
156 run_count: i64 = 0,
157 total_run_time: f64 = 0,
158 expected_start_time: ?[]const u8 = null,
159 start_time: ?[]const u8 = null,
160 end_time: ?[]const u8 = null,
161 parent_task_run_id: ?[]const u8 = null,
162 idempotency_key: ?[]const u8 = null,
163 cache_key: ?[]const u8 = null,
164 cache_expiration: ?[]const u8 = null,
165 task_version: ?[]const u8 = null,
166 };
167
168 fn cacheKey(arena: std.mem.Allocator, kind: []const u8, id: []const u8) ![]const u8 {
169 return std.fmt.allocPrint(arena, "{s}:{s}", .{ kind, id });
170 }
171
172 fn loadRow(self: *Api, arena: std.mem.Allocator, kind: []const u8, id: []const u8) !?RunRow {
173 const key = try cacheKey(arena, kind, id);
174 const raw = self.cache.get(arena, key) orelse return null;
175 return try std.json.parseFromSliceLeaky(RunRow, arena, raw, .{ .ignore_unknown_fields = true });
176 }
177
178 fn storeRow(self: *Api, arena: std.mem.Allocator, kind: []const u8, row: RunRow) !void {
179 var out: std.Io.Writer.Allocating = .init(arena);
180 var jw: std.json.Stringify = .{ .writer = &out.writer };
181 try jw.write(row);
182 const key = try cacheKey(arena, kind, row.id);
183 try self.cache.set(key, out.written());
184 }
185
186 // ------------------------------------------------------------------
187 // repo writes
188 // ------------------------------------------------------------------
189
190 /// build a record from (key, json string) pairs of mixed raw/plain
191 /// values, write it to the repo, and commit.
192 fn writeRecord(
193 self: *Api,
194 arena: std.mem.Allocator,
195 collection: []const u8,
196 rkey: []const u8,
197 record_json: []const u8,
198 ) !void {
199 const parsed = try std.json.parseFromSliceLeaky(std.json.Value, arena, record_json, .{});
200 const cb = try json_cbor.jsonToCbor(arena, parsed);
201 const encoded = try zat.cbor.encodeAlloc(arena, cb);
202 _ = try self.repo.putRecord(collection, rkey, encoded);
203 self.dirty += 1;
204 if (self.dirty >= self.commit_every) {
205 _ = try self.repo.commit();
206 self.dirty = 0;
207 }
208 }
209
210 // ------------------------------------------------------------------
211 // flows
212 // ------------------------------------------------------------------
213
214 fn createFlow(self: *Api, req: *httpz.Request, res: *httpz.Response) !void {
215 const arena = req.arena;
216 const body = req.body() orelse return badRequest(res, "body required");
217 const parsed = std.json.parseFromSliceLeaky(std.json.Value, arena, body, .{}) catch
218 return badRequest(res, "invalid json");
219 const name = getString(parsed, "name") orelse return badRequest(res, "name required");
220 const tags = try rawField(arena, parsed, "tags", "[]");
221
222 self.lock.lock(self.io) catch return error.LockFailed;
223 defer self.lock.unlock(self.io);
224
225 const name_key = try std.fmt.allocPrint(arena, "flow:name:{s}", .{name});
226 if (self.cache.get(arena, name_key)) |existing_id| {
227 if (self.cache.get(arena, try cacheKey(arena, "flow", existing_id))) |row| {
228 res.setStatus(.ok);
229 res.body = row;
230 return;
231 }
232 }
233
234 var id_buf: [36]u8 = undefined;
235 const id = util.uuid4(self.io, &id_buf);
236 var ts_buf: [32]u8 = undefined;
237 const now = util.timestamp(self.io, &ts_buf);
238
239 const flow_json = try std.fmt.allocPrint(arena,
240 \\{{"id":"{s}","created":"{s}","updated":"{s}","name":"{s}","tags":{s}}}
241 , .{ id, now, now, name, tags });
242
243 const record_json = try std.fmt.allocPrint(arena,
244 \\{{"$type":"{s}","id":"{s}","created":"{s}","name":"{s}","tags":{s}}}
245 , .{ flow_collection, id, now, name, tags });
246 try self.writeRecord(arena, flow_collection, id, record_json);
247
248 try self.cache.set(try cacheKey(arena, "flow", id), flow_json);
249 try self.cache.set(name_key, id);
250 try self.cache.sadd("s:flow:all", id);
251
252 res.setStatus(.created);
253 res.body = try arena.dupe(u8, flow_json);
254 }
255
256 fn getEntity(self: *Api, req: *httpz.Request, res: *httpz.Response, kind: []const u8, id: []const u8) !void {
257 const arena = req.arena;
258 if (self.cache.get(arena, try cacheKey(arena, kind, id))) |row| {
259 res.body = row;
260 } else {
261 res.setStatus(.not_found);
262 res.body = "{\"detail\":\"not found\"}";
263 }
264 }
265
266 // ------------------------------------------------------------------
267 // flow runs + task runs
268 // ------------------------------------------------------------------
269
270 const RunKind = enum { flow_run, task_run };
271
272 fn createFlowRun(self: *Api, req: *httpz.Request, res: *httpz.Response) !void {
273 const arena = req.arena;
274 const body = req.body() orelse return badRequest(res, "body required");
275 const parsed = std.json.parseFromSliceLeaky(std.json.Value, arena, body, .{}) catch
276 return badRequest(res, "invalid json");
277
278 var id_buf: [36]u8 = undefined;
279 const id = util.uuid4(self.io, &id_buf);
280 var ts_buf: [32]u8 = undefined;
281 const now = try arena.dupe(u8, util.timestamp(self.io, &ts_buf));
282
283 var row = RunRow{
284 .id = try arena.dupe(u8, id),
285 .created = now,
286 .updated = now,
287 .name = getString(parsed, "name") orelse try runName(arena, id),
288 .flow_id = getString(parsed, "flow_id") orelse return badRequest(res, "flow_id required"),
289 .raw_parameters = try rawField(arena, parsed, "parameters", "{}"),
290 .raw_tags = try rawField(arena, parsed, "tags", "[]"),
291 .raw_empirical_policy = try rawField(arena, parsed, "empirical_policy", "{}"),
292 .raw_job_variables = try rawField(arena, parsed, "job_variables", "{}"),
293 .parent_task_run_id = getString(parsed, "parent_task_run_id"),
294 .idempotency_key = getString(parsed, "idempotency_key"),
295 };
296
297 self.lock.lock(self.io) catch return error.LockFailed;
298 defer self.lock.unlock(self.io);
299
300 try self.applyInitialState(arena, &row, parsed, now, .flow_run);
301
302 const record_json = try std.fmt.allocPrint(arena,
303 \\{{"$type":"{s}","id":"{s}","created":"{s}","name":"{s}","flowId":"{s}","parameters":{s},"tags":{s}}}
304 , .{ flow_run_collection, row.id, now, row.name, row.flow_id, row.raw_parameters, row.raw_tags });
305 try self.writeRecord(arena, flow_run_collection, row.id, record_json);
306
307 try self.storeRow(arena, "flow_run", row);
308 try self.indexRun(arena, "flow_run", row);
309
310 res.setStatus(.created);
311 try self.writeRunResponse(res, .flow_run, row);
312 }
313
314 fn createTaskRun(self: *Api, req: *httpz.Request, res: *httpz.Response) !void {
315 const arena = req.arena;
316 const body = req.body() orelse return badRequest(res, "body required");
317 const parsed = std.json.parseFromSliceLeaky(std.json.Value, arena, body, .{}) catch
318 return badRequest(res, "invalid json");
319
320 var id_buf: [36]u8 = undefined;
321 const id = util.uuid4(self.io, &id_buf);
322 var ts_buf: [32]u8 = undefined;
323 const now = try arena.dupe(u8, util.timestamp(self.io, &ts_buf));
324
325 var row = RunRow{
326 .id = try arena.dupe(u8, id),
327 .created = now,
328 .updated = now,
329 .name = getString(parsed, "name") orelse try runName(arena, id),
330 .flow_run_id = getString(parsed, "flow_run_id"),
331 .task_key = getString(parsed, "task_key") orelse "",
332 .dynamic_key = getString(parsed, "dynamic_key") orelse "",
333 .cache_key = getString(parsed, "cache_key"),
334 .cache_expiration = getString(parsed, "cache_expiration"),
335 .task_version = getString(parsed, "task_version"),
336 .raw_tags = try rawField(arena, parsed, "tags", "[]"),
337 .raw_empirical_policy = try rawField(arena, parsed, "empirical_policy", "{}"),
338 .raw_task_inputs = try rawField(arena, parsed, "task_inputs", "{}"),
339 };
340
341 self.lock.lock(self.io) catch return error.LockFailed;
342 defer self.lock.unlock(self.io);
343
344 try self.applyInitialState(arena, &row, parsed, now, .task_run);
345
346 const record_json = try std.fmt.allocPrint(arena,
347 \\{{"$type":"{s}","id":"{s}","created":"{s}","name":"{s}","flowRunId":{s},"taskKey":"{s}","dynamicKey":"{s}"}}
348 , .{
349 task_run_collection, row.id,
350 now, row.name,
351 try jsonOptString(arena, row.flow_run_id), row.task_key,
352 row.dynamic_key,
353 });
354 try self.writeRecord(arena, task_run_collection, row.id, record_json);
355
356 try self.storeRow(arena, "task_run", row);
357 try self.indexRun(arena, "task_run", row);
358
359 res.setStatus(.created);
360 try self.writeRunResponse(res, .task_run, row);
361 }
362
363 /// secondary indexes: membership sets per state / per parent, mirroring
364 /// the cloud prototype's redis run-state indices. derived, rebuildable.
365 fn indexRun(self: *Api, arena: std.mem.Allocator, kind_str: []const u8, row: RunRow) !void {
366 try self.cache.sadd(try std.fmt.allocPrint(arena, "s:{s}:all", .{kind_str}), row.id);
367 if (row.state_id.len > 0) {
368 try self.cache.sadd(try std.fmt.allocPrint(arena, "s:{s}:state:{s}", .{ kind_str, row.state_type }), row.id);
369 }
370 if (row.flow_id.len > 0) {
371 try self.cache.sadd(try std.fmt.allocPrint(arena, "s:{s}:flow:{s}", .{ kind_str, row.flow_id }), row.id);
372 }
373 if (row.flow_run_id) |frid| {
374 try self.cache.sadd(try std.fmt.allocPrint(arena, "s:{s}:flowrun:{s}", .{ kind_str, frid }), row.id);
375 }
376 }
377
378 fn moveStateIndex(self: *Api, arena: std.mem.Allocator, kind_str: []const u8, id: []const u8, prior: ?[]const u8, new_type: []const u8) !void {
379 if (prior) |p| {
380 try self.cache.srem(try std.fmt.allocPrint(arena, "s:{s}:state:{s}", .{ kind_str, p }), id);
381 }
382 try self.cache.sadd(try std.fmt.allocPrint(arena, "s:{s}:state:{s}", .{ kind_str, new_type }), id);
383 }
384
385 /// runs are created with an optional initial state (the client sends
386 /// Pending). recorded like any other transition: an event.
387 fn applyInitialState(self: *Api, arena: std.mem.Allocator, row: *RunRow, parsed: std.json.Value, now: []const u8, kind: RunKind) !void {
388 const state = parsed.object.get("state") orelse return;
389 if (state != .object) return;
390
391 var sid_buf: [36]u8 = undefined;
392 const state_id = try arena.dupe(u8, util.uuid4(self.io, &sid_buf));
393 const state_type = if (state.object.get("type")) |t| (if (t == .string) t.string else "PENDING") else "PENDING";
394
395 row.state_id = state_id;
396 row.state_type = try arena.dupe(u8, state_type);
397 row.state_name = if (state.object.get("name")) |n|
398 (if (n == .string) try arena.dupe(u8, n.string) else defaultStateName(state_type))
399 else
400 defaultStateName(state_type);
401 row.state_timestamp = now;
402 row.raw_state_details = try rawField(arena, state, "state_details", "{}");
403
404 try self.writeTransitionEvent(arena, kind, row.*, null);
405 }
406
407 fn setState(self: *Api, req: *httpz.Request, res: *httpz.Response, id: []const u8, kind: RunKind) !void {
408 const arena = req.arena;
409 const body = req.body() orelse return badRequest(res, "body required");
410 const parsed = std.json.parseFromSliceLeaky(std.json.Value, arena, body, .{}) catch
411 return badRequest(res, "invalid json");
412 const state = parsed.object.get("state") orelse return badRequest(res, "state required");
413 if (state != .object) return badRequest(res, "state must be an object");
414
415 const kind_str = switch (kind) {
416 .flow_run => "flow_run",
417 .task_run => "task_run",
418 };
419
420 self.lock.lock(self.io) catch return error.LockFailed;
421 defer self.lock.unlock(self.io);
422
423 var row = (try self.loadRow(arena, kind_str, id)) orelse {
424 res.setStatus(.not_found);
425 res.body = "{\"detail\":\"run not found\"}";
426 return;
427 };
428
429 var ts_buf: [32]u8 = undefined;
430 const now = try arena.dupe(u8, util.timestamp(self.io, &ts_buf));
431 var sid_buf: [36]u8 = undefined;
432 const state_id = try arena.dupe(u8, util.uuid4(self.io, &sid_buf));
433
434 const state_type = if (state.object.get("type")) |t| (if (t == .string) t.string else "PENDING") else "PENDING";
435 const state_name = if (state.object.get("name")) |n|
436 (if (n == .string and n.string.len > 0) n.string else defaultStateName(state_type))
437 else
438 defaultStateName(state_type);
439 const message: ?[]const u8 = if (state.object.get("message")) |m|
440 (if (m == .string) m.string else null)
441 else
442 null;
443
444 const prior_type: ?[]const u8 = if (row.state_id.len > 0)
445 try arena.dupe(u8, row.state_type)
446 else
447 null;
448
449 row.state_id = state_id;
450 row.state_type = try arena.dupe(u8, state_type);
451 row.state_name = try arena.dupe(u8, state_name);
452 row.state_timestamp = now;
453 row.state_message = message;
454 row.raw_state_details = try rawField(arena, state, "state_details", "{}");
455 row.updated = now;
456 if (std.mem.eql(u8, state_type, "RUNNING")) {
457 row.run_count += 1;
458 if (row.start_time == null) row.start_time = now;
459 }
460 if (isTerminal(state_type)) row.end_time = now;
461
462 try self.writeTransitionEvent(arena, kind, row, prior_type);
463
464 try self.storeRow(arena, kind_str, row);
465 try self.moveStateIndex(arena, kind_str, row.id, prior_type, row.state_type);
466
467 // OrchestrationResult: state + ACCEPT
468 var out: std.Io.Writer.Allocating = .init(arena);
469 var jw: std.json.Stringify = .{ .writer = &out.writer };
470 try jw.beginObject();
471 try jw.objectField("state");
472 try writeStateObject(&jw, row, true);
473 try jw.objectField("status");
474 try jw.write("ACCEPT");
475 try jw.objectField("details");
476 try jw.beginObject();
477 try jw.objectField("type");
478 try jw.write("accept_details");
479 try jw.endObject();
480 try jw.endObject();
481
482 res.setStatus(.created);
483 res.body = out.written();
484 }
485
486 /// state transitions are io.prefect.v0.event records — there is no
487 /// state collection. mirrors prefect's own convention: event name
488 /// `prefect.flow-run.{StateName}`, resource labels, state in payload.
489 fn writeTransitionEvent(self: *Api, arena: std.mem.Allocator, kind: RunKind, row: RunRow, prior_type: ?[]const u8) !void {
490 const entity = switch (kind) {
491 .flow_run => "flow-run",
492 .task_run => "task-run",
493 };
494 const record_json = try std.fmt.allocPrint(arena,
495 \\{{"$type":"{s}","occurred":"{s}","event":"prefect.{s}.{s}","resource":{{"prefect.resource.id":"prefect.{s}.{s}","prefect.resource.name":"{s}"}},"payload":{{"intended":{{"from":{s},"to":"{s}"}},"validated_state":{{"type":"{s}","name":"{s}","timestamp":"{s}","message":{s},"stateDetails":{s}}}}},"eventId":"{s}"}}
496 , .{
497 event_collection,
498 row.state_timestamp,
499 entity,
500 row.state_name,
501 entity,
502 row.id,
503 row.name,
504 try jsonOptString(arena, prior_type),
505 row.state_type,
506 row.state_type,
507 row.state_name,
508 row.state_timestamp,
509 try jsonOptString(arena, row.state_message),
510 row.raw_state_details,
511 row.state_id,
512 });
513
514 // rkey: a fresh TID past the commit rev clock so events sort
515 // chronologically within the collection
516 const now_us: u64 = @intCast(@max(0, util.nowMicros(self.io)));
517 var tid = zat.Tid.fromTimestamp(@max(now_us, self.repo.last_rev_us + 1), 1);
518 const rkey = try arena.dupe(u8, tid.str());
519 try self.writeRecord(arena, event_collection, rkey, record_json);
520 }
521
522 fn getFlowRun(self: *Api, req: *httpz.Request, res: *httpz.Response, id: []const u8) !void {
523 const arena = req.arena;
524 const row = (try self.loadRow(arena, "flow_run", id)) orelse {
525 res.setStatus(.not_found);
526 res.body = "{\"detail\":\"not found\"}";
527 return;
528 };
529 try self.writeRunResponse(res, .flow_run, row);
530 }
531
532 fn getTaskRun(self: *Api, req: *httpz.Request, res: *httpz.Response, id: []const u8) !void {
533 const arena = req.arena;
534 const row = (try self.loadRow(arena, "task_run", id)) orelse {
535 res.setStatus(.not_found);
536 res.body = "{\"detail\":\"not found\"}";
537 return;
538 };
539 try self.writeRunResponse(res, .task_run, row);
540 }
541
542 // ------------------------------------------------------------------
543 // response serialization (prefect shapes)
544 // ------------------------------------------------------------------
545
546 fn writeRunResponse(self: *Api, res: *httpz.Response, kind: RunKind, row: RunRow) !void {
547 _ = self;
548 var out: std.Io.Writer.Allocating = .init(res.arena);
549 var jw: std.json.Stringify = .{ .writer = &out.writer };
550 try writeRunObject(&jw, kind, row);
551 res.body = out.written();
552 }
553
554 fn writeRunObject(jw: *std.json.Stringify, kind: RunKind, row: RunRow) !void {
555 try jw.beginObject();
556 try jw.objectField("id");
557 try jw.write(row.id);
558 try jw.objectField("created");
559 try jw.write(row.created);
560 try jw.objectField("updated");
561 try jw.write(row.updated);
562 try jw.objectField("name");
563 try jw.write(row.name);
564
565 switch (kind) {
566 .flow_run => {
567 try jw.objectField("flow_id");
568 try jw.write(row.flow_id);
569 try jw.objectField("parameters");
570 try writeRaw(jw, row.raw_parameters);
571 try jw.objectField("job_variables");
572 try writeRaw(jw, row.raw_job_variables);
573 try jw.objectField("deployment_id");
574 try jw.write(null);
575 try jw.objectField("deployment_version");
576 try jw.write(null);
577 try jw.objectField("work_queue_name");
578 try jw.write(null);
579 try jw.objectField("work_queue_id");
580 try jw.write(null);
581 try jw.objectField("auto_scheduled");
582 try jw.write(false);
583 try jw.objectField("parent_task_run_id");
584 try jw.write(row.parent_task_run_id);
585 try jw.objectField("idempotency_key");
586 try jw.write(row.idempotency_key);
587 },
588 .task_run => {
589 try jw.objectField("flow_run_id");
590 try jw.write(row.flow_run_id);
591 try jw.objectField("task_key");
592 try jw.write(row.task_key);
593 try jw.objectField("dynamic_key");
594 try jw.write(row.dynamic_key);
595 try jw.objectField("cache_key");
596 try jw.write(row.cache_key);
597 try jw.objectField("cache_expiration");
598 try jw.write(row.cache_expiration);
599 try jw.objectField("task_version");
600 try jw.write(row.task_version);
601 try jw.objectField("task_inputs");
602 try writeRaw(jw, row.raw_task_inputs);
603 },
604 }
605
606 try jw.objectField("tags");
607 try writeRaw(jw, row.raw_tags);
608 try jw.objectField("empirical_policy");
609 try writeRaw(jw, row.raw_empirical_policy);
610 try jw.objectField("run_count");
611 try jw.write(row.run_count);
612 try jw.objectField("expected_start_time");
613 try jw.write(row.expected_start_time);
614 try jw.objectField("start_time");
615 try jw.write(row.start_time);
616 try jw.objectField("end_time");
617 try jw.write(row.end_time);
618 try jw.objectField("total_run_time");
619 try jw.write(row.total_run_time);
620 try jw.objectField("state_type");
621 try jw.write(if (row.state_id.len > 0) row.state_type else null);
622 try jw.objectField("state_name");
623 try jw.write(if (row.state_id.len > 0) row.state_name else null);
624 try jw.objectField("state");
625 if (row.state_id.len > 0) {
626 try writeStateObject(jw, row, kind == .task_run);
627 } else {
628 try jw.write(null);
629 }
630 try jw.endObject();
631 }
632
633 fn writeStateObject(jw: *std.json.Stringify, row: RunRow, include_data: bool) !void {
634 try jw.beginObject();
635 try jw.objectField("id");
636 try jw.write(row.state_id);
637 try jw.objectField("type");
638 try jw.write(row.state_type);
639 try jw.objectField("name");
640 try jw.write(row.state_name);
641 try jw.objectField("timestamp");
642 try jw.write(row.state_timestamp);
643 try jw.objectField("message");
644 try jw.write(row.state_message);
645 try jw.objectField("state_details");
646 try writeRaw(jw, row.raw_state_details);
647 if (include_data) {
648 try jw.objectField("data");
649 try jw.write(null);
650 }
651 try jw.endObject();
652 }
653
654 // ------------------------------------------------------------------
655 // filtered reads — served entirely from the appview. candidate ids
656 // come from membership sets; rows sort by ISO timestamp (string order
657 // == time order); limit/offset applied after sort.
658 // ------------------------------------------------------------------
659
660 fn filterRuns(self: *Api, req: *httpz.Request, res: *httpz.Response, kind: RunKind, count_only: bool) !void {
661 const arena = req.arena;
662 const kind_str = switch (kind) {
663 .flow_run => "flow_run",
664 .task_run => "task_run",
665 };
666
667 const parsed: std.json.Value = blk: {
668 const body = req.body() orelse break :blk .null;
669 if (body.len == 0) break :blk .null;
670 break :blk std.json.parseFromSliceLeaky(std.json.Value, arena, body, .{}) catch .null;
671 };
672
673 const runs_filter: std.json.Value = blk: {
674 if (parsed != .object) break :blk .null;
675 const key = switch (kind) {
676 .flow_run => "flow_runs",
677 .task_run => "task_runs",
678 };
679 break :blk parsed.object.get(key) orelse .null;
680 };
681
682 // candidates: state filter -> union of state sets, else the all-set
683 var candidates: std.ArrayList([]const u8) = .empty;
684 const state_types = filterAny(runs_filter, "state", "type");
685 if (state_types) |types| {
686 for (types) |t| {
687 if (t != .string) continue;
688 const members = try self.cache.smembers(arena, try std.fmt.allocPrint(arena, "s:{s}:state:{s}", .{ kind_str, t.string }));
689 try candidates.appendSlice(arena, members);
690 }
691 } else {
692 const members = try self.cache.smembers(arena, try std.fmt.allocPrint(arena, "s:{s}:all", .{kind_str}));
693 try candidates.appendSlice(arena, members);
694 }
695
696 // intersect with flow-run scope for task runs (the engine's
697 // "task runs of this flow run" read)
698 if (kind == .task_run) {
699 if (filterAny(runs_filter, "flow_run_id", null)) |frids| {
700 if (frids.len > 0 and frids[0] == .string) {
701 const scoped = try self.cache.smembers(arena, try std.fmt.allocPrint(arena, "s:task_run:flowrun:{s}", .{frids[0].string}));
702 candidates.items = try intersect(arena, candidates.items, scoped);
703 }
704 }
705 }
706
707 if (count_only) {
708 res.body = try std.fmt.allocPrint(arena, "{d}", .{candidates.items.len});
709 return;
710 }
711
712 // load rows, sort newest-first, slice
713 var rows: std.ArrayList(RunRow) = .empty;
714 for (candidates.items) |id| {
715 if (try self.loadRow(arena, kind_str, id)) |row| try rows.append(arena, row);
716 }
717 std.mem.sort(RunRow, rows.items, {}, struct {
718 fn newerFirst(_: void, a: RunRow, b: RunRow) bool {
719 return std.mem.order(u8, a.created, b.created) == .gt;
720 }
721 }.newerFirst);
722
723 const offset: usize = if (parsed == .object)
724 if (parsed.object.get("offset")) |o| (if (o == .integer and o.integer > 0) @intCast(o.integer) else 0) else 0
725 else
726 0;
727 const limit: usize = if (parsed == .object)
728 if (parsed.object.get("limit")) |l| (if (l == .integer and l.integer > 0) @intCast(l.integer) else 200) else 200
729 else
730 200;
731
732 const start = @min(offset, rows.items.len);
733 const end = @min(start + limit, rows.items.len);
734
735 var out: std.Io.Writer.Allocating = .init(arena);
736 var jw: std.json.Stringify = .{ .writer = &out.writer };
737 try jw.beginArray();
738 for (rows.items[start..end]) |row| try writeRunObject(&jw, kind, row);
739 try jw.endArray();
740 res.body = out.written();
741 }
742
743 fn filterFlows(self: *Api, req: *httpz.Request, res: *httpz.Response) !void {
744 const arena = req.arena;
745 const ids = try self.cache.smembers(arena, "s:flow:all");
746
747 const Named = struct { name: []const u8, raw: []const u8 };
748 var flows: std.ArrayList(Named) = .empty;
749 for (ids) |id| {
750 const raw = self.cache.get(arena, try cacheKey(arena, "flow", id)) orelse continue;
751 const v = std.json.parseFromSliceLeaky(std.json.Value, arena, raw, .{}) catch continue;
752 const name = getString(v, "name") orelse "";
753 try flows.append(arena, .{ .name = name, .raw = raw });
754 }
755 std.mem.sort(Named, flows.items, {}, struct {
756 fn byName(_: void, a: Named, b: Named) bool {
757 return std.mem.order(u8, a.name, b.name) == .lt;
758 }
759 }.byName);
760
761 var out: std.Io.Writer.Allocating = .init(arena);
762 var jw: std.json.Stringify = .{ .writer = &out.writer };
763 try jw.beginArray();
764 for (flows.items) |f| {
765 try jw.beginWriteRaw();
766 try jw.writer.writeAll(f.raw);
767 jw.endWriteRaw();
768 }
769 try jw.endArray();
770 res.body = out.written();
771 }
772
773 /// navigate filter json like flow_runs.state.type.any_ -> []Value
774 fn filterAny(filter: std.json.Value, key: []const u8, sub: ?[]const u8) ?[]std.json.Value {
775 if (filter != .object) return null;
776 var node = filter.object.get(key) orelse return null;
777 if (sub) |s| {
778 if (node != .object) return null;
779 node = node.object.get(s) orelse return null;
780 }
781 if (node != .object) return null;
782 const any = node.object.get("any_") orelse return null;
783 if (any != .array) return null;
784 return any.array.items;
785 }
786
787 fn intersect(arena: std.mem.Allocator, a: []const []const u8, b: [][]u8) ![][]const u8 {
788 var keep: std.StringHashMapUnmanaged(void) = .empty;
789 for (b) |m| try keep.put(arena, m, {});
790 var out: std.ArrayList([]const u8) = .empty;
791 for (a) |m| if (keep.contains(m)) try out.append(arena, m);
792 return out.items;
793 }
794
795 // ------------------------------------------------------------------
796 // hydration: rebuild the entire appview by replaying the repo.
797 // entities first, then the event log in TID order. this is the
798 // "delete the cache, replay the truth" property, exercised on boot.
799 // ------------------------------------------------------------------
800
801 pub fn hydrate(self: *Api, arena: std.mem.Allocator) !usize {
802 var applied: usize = 0;
803
804 for (try self.repo.listRecords(arena, flow_collection)) |rec| {
805 const v = zat.cbor.decodeAll(arena, rec.cbor_data) catch continue;
806 const id = rec.key[flow_collection.len + 1 ..];
807 const name = v.getString("name") orelse continue;
808 const created = v.getString("created") orelse "";
809 const flow_json = try std.fmt.allocPrint(arena,
810 \\{{"id":"{s}","created":"{s}","updated":"{s}","name":"{s}","tags":{s}}}
811 , .{ id, created, created, name, try cborFieldJson(arena, v, "tags", "[]") });
812 try self.cache.set(try cacheKey(arena, "flow", id), flow_json);
813 try self.cache.set(try std.fmt.allocPrint(arena, "flow:name:{s}", .{name}), id);
814 try self.cache.sadd("s:flow:all", id);
815 applied += 1;
816 }
817
818 for (try self.repo.listRecords(arena, flow_run_collection)) |rec| {
819 const v = zat.cbor.decodeAll(arena, rec.cbor_data) catch continue;
820 const id = rec.key[flow_run_collection.len + 1 ..];
821 const row = RunRow{
822 .id = id,
823 .created = v.getString("created") orelse "",
824 .updated = v.getString("created") orelse "",
825 .name = v.getString("name") orelse "",
826 .flow_id = v.getString("flowId") orelse "",
827 .raw_parameters = try cborFieldJson(arena, v, "parameters", "{}"),
828 .raw_tags = try cborFieldJson(arena, v, "tags", "[]"),
829 };
830 try self.storeRow(arena, "flow_run", row);
831 try self.indexRun(arena, "flow_run", row);
832 applied += 1;
833 }
834
835 for (try self.repo.listRecords(arena, task_run_collection)) |rec| {
836 const v = zat.cbor.decodeAll(arena, rec.cbor_data) catch continue;
837 const id = rec.key[task_run_collection.len + 1 ..];
838 const row = RunRow{
839 .id = id,
840 .created = v.getString("created") orelse "",
841 .updated = v.getString("created") orelse "",
842 .name = v.getString("name") orelse "",
843 .flow_run_id = v.getString("flowRunId"),
844 .task_key = v.getString("taskKey") orelse "",
845 .dynamic_key = v.getString("dynamicKey") orelse "",
846 };
847 try self.storeRow(arena, "task_run", row);
848 try self.indexRun(arena, "task_run", row);
849 applied += 1;
850 }
851
852 // replay the event log: MST walk order == TID order == time order
853 for (try self.repo.listRecords(arena, event_collection)) |rec| {
854 const v = zat.cbor.decodeAll(arena, rec.cbor_data) catch continue;
855 const event_name = v.getString("event") orelse continue;
856 const resource = v.get("resource") orelse continue;
857 const resource_id = resource.getString("prefect.resource.id") orelse continue;
858
859 const kind_str: []const u8, const kind: RunKind, const prefix: []const u8 =
860 if (std.mem.indexOf(u8, event_name, "flow-run") != null)
861 .{ "flow_run", .flow_run, "prefect.flow-run." }
862 else if (std.mem.indexOf(u8, event_name, "task-run") != null)
863 .{ "task_run", .task_run, "prefect.task-run." }
864 else
865 continue;
866 _ = kind;
867 if (!std.mem.startsWith(u8, resource_id, prefix)) continue;
868 const run_id = resource_id[prefix.len..];
869
870 const payload = v.get("payload") orelse continue;
871 const state = payload.get("validated_state") orelse continue;
872 const state_type = state.getString("type") orelse continue;
873 const ts = state.getString("timestamp") orelse "";
874
875 var row = (try self.loadRow(arena, kind_str, run_id)) orelse continue;
876 const prior: ?[]const u8 = if (row.state_id.len > 0) try arena.dupe(u8, row.state_type) else null;
877
878 row.state_id = v.getString("eventId") orelse "";
879 row.state_type = try arena.dupe(u8, state_type);
880 row.state_name = try arena.dupe(u8, state.getString("name") orelse defaultStateName(state_type));
881 row.state_timestamp = ts;
882 row.state_message = state.getString("message");
883 row.raw_state_details = try cborFieldJson(arena, state, "stateDetails", "{}");
884 row.updated = ts;
885 if (std.mem.eql(u8, state_type, "RUNNING")) {
886 row.run_count += 1;
887 if (row.start_time == null) row.start_time = ts;
888 }
889 if (isTerminal(state_type)) row.end_time = ts;
890
891 try self.storeRow(arena, kind_str, row);
892 try self.moveStateIndex(arena, kind_str, row.id, prior, row.state_type);
893 applied += 1;
894 }
895
896 return applied;
897 }
898
899 /// serialize a cbor map field back to JSON text (or default)
900 fn cborFieldJson(arena: std.mem.Allocator, v: zat.cbor.Value, key: []const u8, default: []const u8) ![]const u8 {
901 const sub = v.get(key) orelse return default;
902 var out: std.Io.Writer.Allocating = .init(arena);
903 var jw: std.json.Stringify = .{ .writer = &out.writer };
904 try json_cbor.cborToJson(&jw, sub);
905 return out.written();
906 }
907
908 // ------------------------------------------------------------------
909 // logs
910 // ------------------------------------------------------------------
911
912 /// the client already batches logs; one POST = one record
913 fn createLogs(self: *Api, req: *httpz.Request, res: *httpz.Response) !void {
914 const arena = req.arena;
915 const body = req.body() orelse return badRequest(res, "body required");
916
917 self.lock.lock(self.io) catch return error.LockFailed;
918 defer self.lock.unlock(self.io);
919
920 const record_json = try std.fmt.allocPrint(arena,
921 \\{{"$type":"{s}","entries":{s}}}
922 , .{ log_batch_collection, body });
923
924 const now_us: u64 = @intCast(@max(0, util.nowMicros(self.io)));
925 var tid = zat.Tid.fromTimestamp(@max(now_us, self.repo.last_rev_us + 1), 2);
926 const rkey = try arena.dupe(u8, tid.str());
927 try self.writeRecord(arena, log_batch_collection, rkey, record_json);
928
929 res.setStatus(.created);
930 res.body = "null";
931 }
932
933 // ------------------------------------------------------------------
934 // atproto read surface
935 // ------------------------------------------------------------------
936
937 fn getRepoCar(self: *Api, res: *httpz.Response) !void {
938 self.lock.lock(self.io) catch return error.LockFailed;
939 defer self.lock.unlock(self.io);
940
941 const car = self.repo.latest_car orelse {
942 res.setStatus(.not_found);
943 res.body = "{\"error\":\"RepoNotFound\",\"message\":\"no commits yet\"}";
944 return;
945 };
946 res.content_type = .BINARY;
947 res.header("content-type", "application/vnd.ipld.car");
948 res.body = try res.arena.dupe(u8, car);
949 }
950
951 fn describeRepo(self: *Api, req: *httpz.Request, res: *httpz.Response) !void {
952 _ = req;
953 self.lock.lock(self.io) catch return error.LockFailed;
954 defer self.lock.unlock(self.io);
955
956 try res.json(.{
957 .did = self.repo.did,
958 .didDoc = .{},
959 .handle = "prefect.local",
960 .collections = &[_][]const u8{
961 flow_collection,
962 flow_run_collection,
963 task_run_collection,
964 event_collection,
965 log_batch_collection,
966 },
967 .handleIsCorrect = false,
968 }, .{});
969 }
970
971 fn listRecords(self: *Api, req: *httpz.Request, res: *httpz.Response) !void {
972 const arena = req.arena;
973 const collection = (try req.query()).get("collection") orelse
974 return badRequest(res, "collection required");
975
976 self.lock.lock(self.io) catch return error.LockFailed;
977 defer self.lock.unlock(self.io);
978
979 const records = try self.repo.listRecords(arena, collection);
980
981 var out: std.Io.Writer.Allocating = .init(arena);
982 var jw: std.json.Stringify = .{ .writer = &out.writer };
983 try jw.beginObject();
984 try jw.objectField("records");
985 try jw.beginArray();
986 for (records) |rec| {
987 const decoded = zat.cbor.decodeAll(arena, rec.cbor_data) catch continue;
988 try jw.beginObject();
989 try jw.objectField("uri");
990 const uri = try std.fmt.allocPrint(arena, "at://{s}/{s}", .{ self.repo.did, rec.key });
991 try jw.write(uri);
992 try jw.objectField("value");
993 try json_cbor.cborToJson(&jw, decoded);
994 try jw.endObject();
995 }
996 try jw.endArray();
997 try jw.endObject();
998
999 res.body = out.written();
1000 }
1001
1002 // ------------------------------------------------------------------
1003 // helpers
1004 // ------------------------------------------------------------------
1005
1006 fn badRequest(res: *httpz.Response, msg: []const u8) !void {
1007 res.setStatus(.unprocessable_entity);
1008 var out: std.Io.Writer.Allocating = .init(res.arena);
1009 var jw: std.json.Stringify = .{ .writer = &out.writer };
1010 try jw.beginObject();
1011 try jw.objectField("detail");
1012 try jw.write(msg);
1013 try jw.endObject();
1014 res.body = out.written();
1015 }
1016
1017 fn getString(v: std.json.Value, key: []const u8) ?[]const u8 {
1018 if (v != .object) return null;
1019 const field = v.object.get(key) orelse return null;
1020 return if (field == .string) field.string else null;
1021 }
1022
1023 /// re-serialize a json field verbatim (or default if absent/null)
1024 fn rawField(arena: std.mem.Allocator, v: std.json.Value, key: []const u8, default: []const u8) ![]const u8 {
1025 if (v != .object) return default;
1026 const field = v.object.get(key) orelse return default;
1027 if (field == .null) return default;
1028 var out: std.Io.Writer.Allocating = .init(arena);
1029 var jw: std.json.Stringify = .{ .writer = &out.writer };
1030 try jw.write(field);
1031 return out.written();
1032 }
1033
1034 fn jsonOptString(arena: std.mem.Allocator, s: ?[]const u8) ![]const u8 {
1035 const v = s orelse return "null";
1036 return std.fmt.allocPrint(arena, "\"{s}\"", .{v});
1037 }
1038
1039 fn writeRaw(jw: *std.json.Stringify, raw: []const u8) !void {
1040 try jw.beginWriteRaw();
1041 try jw.writer.writeAll(if (raw.len > 0) raw else "null");
1042 jw.endWriteRaw();
1043 }
1044
1045 fn runName(arena: std.mem.Allocator, id: []const u8) ![]const u8 {
1046 return std.fmt.allocPrint(arena, "run-{s}", .{id[0..8]});
1047 }
1048
1049 fn defaultStateName(state_type: []const u8) []const u8 {
1050 const names = std.StaticStringMap([]const u8).initComptime(.{
1051 .{ "PENDING", "Pending" },
1052 .{ "RUNNING", "Running" },
1053 .{ "COMPLETED", "Completed" },
1054 .{ "FAILED", "Failed" },
1055 .{ "SCHEDULED", "Scheduled" },
1056 .{ "CANCELLED", "Cancelled" },
1057 .{ "CANCELLING", "Cancelling" },
1058 .{ "CRASHED", "Crashed" },
1059 .{ "PAUSED", "Paused" },
1060 });
1061 return names.get(state_type) orelse "Unknown";
1062 }
1063
1064 fn isTerminal(state_type: []const u8) bool {
1065 return std.mem.eql(u8, state_type, "COMPLETED") or
1066 std.mem.eql(u8, state_type, "FAILED") or
1067 std.mem.eql(u8, state_type, "CRASHED") or
1068 std.mem.eql(u8, state_type, "CANCELLED");
1069 }
1070};