zzstoatzz.io
prefect server in zig
1const std = @import("std");
2const Allocator = std.mem.Allocator;
3
4const backend = @import("backend.zig");
5const log = @import("../logging.zig");
6
7pub const FlowRow = struct {
8 id: []const u8,
9 created: []const u8,
10 updated: []const u8,
11 name: []const u8,
12 tags: []const u8,
13};
14
15pub fn getByName(alloc: Allocator, name: []const u8) !?FlowRow {
16 var rows = backend.db.query(
17 "SELECT id, created, updated, name, tags FROM flow WHERE name = ?",
18 .{name},
19 ) catch return null;
20 defer rows.deinit();
21
22 if (rows.next()) |row| {
23 return rowToFlow(alloc, row);
24 }
25 return null;
26}
27
28pub fn getById(alloc: Allocator, id: []const u8) !?FlowRow {
29 var rows = backend.db.query(
30 "SELECT id, created, updated, name, tags FROM flow WHERE id = ?",
31 .{id},
32 ) catch return null;
33 defer rows.deinit();
34
35 if (rows.next()) |row| {
36 return rowToFlow(alloc, row);
37 }
38 return null;
39}
40
41pub fn insert(id: []const u8, name: []const u8) !void {
42 backend.db.exec(
43 "INSERT INTO flow (id, name) VALUES (?, ?)",
44 .{ id, name },
45 ) catch |err| {
46 log.err("database", "insert flow error: {}", .{err});
47 return err;
48 };
49}
50
51pub fn list(alloc: Allocator, limit: usize) ![]FlowRow {
52 var results = std.ArrayListUnmanaged(FlowRow){};
53 errdefer results.deinit(alloc);
54
55 var rows = backend.db.query(
56 "SELECT id, created, updated, name, tags FROM flow ORDER BY created DESC LIMIT ?",
57 .{@as(i64, @intCast(limit))},
58 ) catch |err| {
59 log.err("database", "list flows error: {}", .{err});
60 return err;
61 };
62 defer rows.deinit();
63
64 while (rows.next()) |row| {
65 try results.append(alloc, rowToFlow(alloc, row));
66 }
67
68 return results.toOwnedSlice(alloc);
69}
70
71pub fn update(id: []const u8, tags: ?[]const u8) !bool {
72 // check if flow exists
73 var rows = backend.db.query(
74 "SELECT id FROM flow WHERE id = ?",
75 .{id},
76 ) catch return false;
77 defer rows.deinit();
78
79 if (rows.next() == null) {
80 return false;
81 }
82
83 if (tags) |t| {
84 backend.db.exec(
85 "UPDATE flow SET tags = ?, updated = CURRENT_TIMESTAMP WHERE id = ?",
86 .{ t, id },
87 ) catch |err| {
88 log.err("database", "update flow error: {}", .{err});
89 return false;
90 };
91 }
92
93 return true;
94}
95
96pub fn delete(id: []const u8) !bool {
97 // check if flow exists
98 var rows = backend.db.query(
99 "SELECT id FROM flow WHERE id = ?",
100 .{id},
101 ) catch return false;
102 defer rows.deinit();
103
104 if (rows.next() == null) {
105 return false;
106 }
107
108 backend.db.exec(
109 "DELETE FROM flow WHERE id = ?",
110 .{id},
111 ) catch |err| {
112 log.err("database", "delete flow error: {}", .{err});
113 return false;
114 };
115
116 return true;
117}
118
119fn rowToFlow(alloc: Allocator, row: anytype) FlowRow {
120 return .{
121 .id = alloc.dupe(u8, row.text(0)) catch "",
122 .created = alloc.dupe(u8, row.text(1)) catch "",
123 .updated = alloc.dupe(u8, row.text(2)) catch "",
124 .name = alloc.dupe(u8, row.text(3)) catch "",
125 .tags = alloc.dupe(u8, row.text(4)) catch "[]",
126 };
127}