prefect-server / src / api / flows.zig

const std = @import("std");
const zap = @import("zap");
const mem = std.mem;
const json = std.json;

const db = @import("../db/sqlite.zig");
const uuid_util = @import("../utilities/uuid.zig");
const time_util = @import("../utilities/time.zig");
const json_util = @import("../utilities/json.zig");

// POST /flows/ - create or get flow by name
// POST /flows/filter - list flows
// GET /flows/{id} - get flow by id
// PATCH /flows/{id} - update flow
// DELETE /flows/{id} - delete flow
pub fn handle(r: zap.Request) !void {
    const target = r.path orelse "/";
    const method = r.method orelse "GET";

    // POST /flows/filter - list
    if (mem.eql(u8, method, "POST") and mem.endsWith(u8, target, "/filter")) {
        try filter(r);
        return;
    }

    // extract id from path for GET/PATCH/DELETE
    const prefix = if (mem.startsWith(u8, target, "/api/flows/")) "/api/flows/" else "/flows/";
    const has_id = target.len > prefix.len;
    const flow_id = if (has_id) target[prefix.len..] else "";

    if (mem.eql(u8, method, "POST") and (mem.eql(u8, target, "/flows/") or mem.eql(u8, target, "/api/flows/"))) {
        try createFlow(r);
    } else if (mem.eql(u8, method, "GET") and has_id) {
        try getFlow(r, flow_id);
    } else if (mem.eql(u8, method, "PATCH") and has_id) {
        try patchFlow(r, flow_id);
    } else if (mem.eql(u8, method, "DELETE") and has_id) {
        try deleteFlow(r, flow_id);
    } else if (mem.eql(u8, method, "GET") or mem.eql(u8, method, "PATCH") or mem.eql(u8, method, "DELETE")) {
        json_util.sendStatus(r, "{\"detail\":\"flow id required\"}", .bad_request);
    } else {
        json_util.sendStatus(r, "{\"detail\":\"not implemented\"}", .not_implemented);
    }
}

fn createFlow(r: zap.Request) !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const alloc = arena.allocator();

    const body = r.body orelse {
        json_util.sendStatus(r, "{\"detail\":\"failed to read body\"}", .bad_request);
        return;
    };

    const parsed = json.parseFromSlice(json.Value, alloc, body, .{}) catch {
        json_util.sendStatus(r, "{\"detail\":\"invalid json\"}", .bad_request);
        return;
    };

    const name = parsed.value.object.get("name") orelse {
        json_util.sendStatus(r, "{\"detail\":\"name required\"}", .bad_request);
        return;
    };
    const name_str = name.string;

    // try to get existing flow first
    if (db.getFlowByName(alloc, name_str) catch null) |flow| {
        const resp = writeFlow(alloc, flow) catch {
            json_util.sendStatus(r, "{\"detail\":\"serialize error\"}", .internal_server_error);
            return;
        };
        json_util.send(r, resp);
        return;
    }

    // create new flow
    var new_id_buf: [36]u8 = undefined;
    const new_id = uuid_util.generate(&new_id_buf);

    db.insertFlow(new_id, name_str) catch {
        json_util.sendStatus(r, "{\"detail\":\"insert failed\"}", .internal_server_error);
        return;
    };

    var ts_buf: [32]u8 = undefined;
    const now = time_util.timestamp(&ts_buf);

    const flow = db.FlowRow{
        .id = new_id,
        .created = now,
        .updated = now,
        .name = name_str,
        .tags = "[]",
    };

    const resp = writeFlow(alloc, flow) catch {
        json_util.sendStatus(r, "{\"detail\":\"serialize error\"}", .internal_server_error);
        return;
    };
    json_util.sendStatus(r, resp, .created);
}

fn getFlow(r: zap.Request, flow_id: []const u8) !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const alloc = arena.allocator();

    if (db.getFlowById(alloc, flow_id) catch null) |flow| {
        const resp = writeFlow(alloc, flow) catch {
            json_util.sendStatus(r, "{\"detail\":\"serialize error\"}", .internal_server_error);
            return;
        };
        json_util.send(r, resp);
    } else {
        json_util.sendStatus(r, "{\"detail\":\"flow not found\"}", .not_found);
    }
}

fn patchFlow(r: zap.Request, flow_id: []const u8) !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const alloc = arena.allocator();

    const body = r.body orelse {
        json_util.sendStatus(r, "{\"detail\":\"failed to read body\"}", .bad_request);
        return;
    };

    const parsed = json.parseFromSlice(json.Value, alloc, body, .{}) catch {
        json_util.sendStatus(r, "{\"detail\":\"invalid json\"}", .bad_request);
        return;
    };

    // extract tags if provided
    var tags_json: ?[]const u8 = null;
    if (parsed.value.object.get("tags")) |tags_val| {
        var output: std.Io.Writer.Allocating = .init(alloc);
        var jw: json.Stringify = .{ .writer = &output.writer };
        jw.write(tags_val) catch {
            json_util.sendStatus(r, "{\"detail\":\"serialize error\"}", .internal_server_error);
            return;
        };
        tags_json = output.toOwnedSlice() catch null;
    }

    const updated = db.updateFlow(flow_id, tags_json) catch false;
    if (!updated) {
        json_util.sendStatus(r, "{\"detail\":\"flow not found\"}", .not_found);
        return;
    }

    json_util.sendStatus(r, "", .no_content);
}

fn deleteFlow(r: zap.Request, flow_id: []const u8) !void {
    const deleted = db.deleteFlow(flow_id) catch false;
    if (!deleted) {
        json_util.sendStatus(r, "{\"detail\":\"flow not found\"}", .not_found);
        return;
    }

    json_util.sendStatus(r, "", .no_content);
}

fn filter(r: zap.Request) !void {
    var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator);
    defer arena.deinit();
    const alloc = arena.allocator();

    const flows = db.listFlows(alloc, 50) catch {
        json_util.sendStatus(r, "{\"detail\":\"database error\"}", .internal_server_error);
        return;
    };

    var output: std.Io.Writer.Allocating = .init(alloc);
    var jw: json.Stringify = .{ .writer = &output.writer };

    jw.beginArray() catch {
        json_util.sendStatus(r, "{\"detail\":\"serialize error\"}", .internal_server_error);
        return;
    };

    for (flows) |flow| {
        writeFlowObject(&jw, flow) catch continue;
    }

    jw.endArray() catch {};

    json_util.send(r, output.toOwnedSlice() catch "[]");
}

fn writeFlow(alloc: std.mem.Allocator, flow: db.FlowRow) ![]const u8 {
    var output: std.Io.Writer.Allocating = .init(alloc);
    var jw: json.Stringify = .{ .writer = &output.writer };
    try writeFlowObject(&jw, flow);
    return output.toOwnedSlice();
}

fn writeFlowObject(jw: *json.Stringify, flow: db.FlowRow) !void {
    try jw.beginObject();

    try jw.objectField("id");
    try jw.write(flow.id);

    try jw.objectField("created");
    try jw.write(flow.created);

    try jw.objectField("updated");
    try jw.write(flow.updated);

    try jw.objectField("name");
    try jw.write(flow.name);

    try jw.objectField("tags");
    try jw.beginWriteRaw();
    try jw.writer.writeAll(flow.tags);
    jw.endWriteRaw();

    try jw.endObject();
}