╭─────────────╮ │scroll to top│ ╰─────────────╯ ╭ event.zig ────────────────────────────────────────────────────╮ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ╰───────────────────────────────────────────────────────────────╯
🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 ╭4f2b39fb..╮ 4f2b39fb.. see-em: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 │files │ commits refs discuss issues new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────────╯ (372) user
╭────╮ ╭ clone ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ╮ │http│ ssh │ │ ╰────╯ ╰ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ╯
.. embed/ event/ test/ ui/ command.zig diff3.zig ╭─────────────────────────────────╮ │event.zig │ ╰─────────────────────────────────╯ lib.zig main.zig main_wasm.zig serve.zig serve_common.zig serve_http.zig serve_ssh.zig serve_ssh_protocol.zig test.zig testnet.zig try.zig ui.zig web.zig
16 pub const MergePolicy = enum { 17 field_conflicts, 18 target_wins, 19 }; 20 21 // the most bytes an event's serialized form may hold 22 pub const max_event_size: usize = 100 * 1024; 23 24 // the branch haxy events are committed to before being consumed 25 pub const events_ref: rf.Ref = .{ .kind = .head, .name = "haxy/events" }; 26 27 // options + db type for *the admin repo* — the single event store that holds 28 // users, repos, issues, etc. the functions below stay parameterized over 29 // repo_opts because they also operate on individual repos in the repos dir, 30 // which may use different options 31 pub const admin_repo_opts: rp.RepoOpts(.xit) = .{}; 32 pub const AdminDB = rp.Repo(.xit, admin_repo_opts).DB; 33 34 // the xitdb type events are stored in. it only depends on the hash kind, so 35 // this matches the db of any xit repo using that hash kind. 36 pub fn EventDB(comptime hash_kind: hash.HashKind) type { 37 return rp.Repo(.xit, .{ .hash = hash_kind }).DB; 38 } 39 40 // the email between a commit author line's angle brackets, or null 41 pub fn authorEmail(author_line: []const u8) ?[]const u8 { 42 const open_bracket = std.mem.indexOfScalar(u8, author_line, '<') orelse return null; 43 const close_bracket = std.mem.indexOfScalarPos(u8, author_line, open_bracket + 1, '>') orelse return null; 44 return author_line[open_bracket + 1 .. close_bracket]; 45 } 46 47 // build a `T` by copying its fields, by name, out of `source` 48 pub fn project(comptime T: type, source: anytype) T { 49 var result: T = undefined; 50 inline for (std.meta.fields(T)) |field| { 51 @field(result, field.name) = @field(source, field.name); 52 } 53 return result; 54 } 55 56 pub const EventKind = enum { 57 user, 58 repo, 59 issue, 60 comment, 61 }; 62 63 // a null payload deletes the record 64 pub const Event = union(EventKind) { 65 user: ?User, 66 repo: ?Repo, 67 issue: ?Issue, 68 comment: ?Comment, 69 }; 70 71 pub const EventWithId = struct { 72 id: [event_id_size * 2]u8, 73 event: Event, 74 timestamp: u64 = 0, // not serialized, because it comes from the commit timestamp 75 author_email: []const u8, // not serialized, because it goes in the commit author 76 77 pub fn jsonStringify(self: EventWithId, jw: anytype) !void { 78 try jw.beginObject(); 79 try jw.objectField("id"); 80 try jw.write(self.id); 81 try jw.objectField("kind"); 82 try jw.write(@tagName(self.event)); 83 try jw.objectField("data"); 84 switch (self.event) { 85 inline else => |event_maybe| try jw.write(event_maybe), 86 } 87 try jw.endObject(); 88 } 89 90 fn fromString(arena: *std.heap.ArenaAllocator, message: []const u8) !EventWithId { 91 const JsonEvent = struct { 92 id: [event_id_size * 2]u8, 93 kind: EventKind, 94 data: ?std.json.Value = null, 95 }; 96 const json_event = try std.json.parseFromSliceLeaky(JsonEvent, arena.allocator(), message, .{ .ignore_unknown_fields = true }); 97 return .{ 98 .id = json_event.id, 99 // unused when reading; the author lives in the commit 100 .author_email = "", 101 .event = switch (json_event.kind) { 102 .user => .{ 103 .user = if (json_event.data) |value| 104 try std.json.parseFromValueLeaky(User, arena.allocator(), value, .{ .ignore_unknown_fields = true }) 105 else 106 null, 107 }, 108 .repo => .{ 109 .repo = if (json_event.data) |value| 110 try std.json.parseFromValueLeaky(Repo, arena.allocator(), value, .{ .ignore_unknown_fields = true }) 111 else 112 null, 113 }, 114 .issue => .{ 115 .issue = if (json_event.data) |value| 116 try std.json.parseFromValueLeaky(Issue, arena.allocator(), value, .{ .ignore_unknown_fields = true }) 117 else 118 null, 119 }, 120 .comment => .{ 121 .comment = if (json_event.data) |value| 122 try std.json.parseFromValueLeaky(Comment, arena.allocator(), value, .{ .ignore_unknown_fields = true }) 123 else 124 null, 125 }, 126 }, 127 }; 128 } 129 130 pub fn randomId(random: std.Random) [event_id_size]u8 { 131 var id_bytes: [event_id_size]u8 = undefined; 132 random.bytes(&id_bytes); 133 return id_bytes; 134 } 135 }; 136 137 // commit `events` (if any) as JSON commit messages on `ref`, then consume the 138 // events on `ref` into the db the repo's views read: the repo's own db for a 139 // xit repo, or the standalone event db next to a git repo. for a xit repo the 140 // commits and the consume run in one transaction. 141 pub fn consume( 142 comptime repo_kind: rp.RepoKind, 143 comptime repo_opts: rp.RepoOpts(repo_kind), 144 io: std.Io, 145 allocator: std.mem.Allocator, 146 repo: *rp.Repo(repo_kind, repo_opts), 147 ref: rf.Ref, 148 events: []const EventWithId, 149 ) !void { 150 var resolved_remote_name: ?[]u8 = null; 151 defer if (resolved_remote_name) |name| allocator.free(name); 152 153 // a branch that only exists on a remote is consumed from its remote-tracking 154 // ref, while commits go on the local branch, continuing from the remote tip 155 // so a cloned repo's local branch stays connected to the server history 156 const resolved_ref: rf.Ref, const first_parent_oids: ?[1][hash.hexLen(repo_opts.hash)]u8 = blk: { 157 if (null != try repo.readRef(io, ref)) break :blk .{ ref, null }; 158 var remotes = try repo.listRemotes(io, allocator); 159 defer remotes.deinit(); 160 for (remotes.sections.keys()) |remote_name| { 161 const remote_ref: rf.Ref = .{ .kind = .{ .remote = remote_name }, .name = ref.name }; 162 if (try repo.readRef(io, remote_ref)) |oid| { 163 if (events.len > 0) break :blk .{ ref, .{oid} }; 164 const name = try allocator.dupe(u8, remote_name); 165 resolved_remote_name = name; 166 break :blk .{ rf.Ref{ .kind = .{ .remote = name }, .name = ref.name }, null }; 167 } 168 } 169 if (events.len > 0) break :blk .{ ref, null }; 170 // the branch is gone (or never existed): drop any db from a previous 171 // sync rather than serving its issues indefinitely 172 if (repo_kind == .git) try LocalEventDB(repo_opts.hash).delete(io, repo.core.repo_dir); 173 return; 174 }; 175 176 switch (repo_kind) { 177 .git => { 178 try commitEvents(.git, repo_opts, .{ .core = &repo.core, .extra = .{} }, io, allocator, ref, events, first_parent_oids); 179 180 var event_db = try LocalEventDB(repo_opts.hash).open(io, allocator, repo.core.repo_dir); 181 defer event_db.deinit(io, allocator); 182 183 try event_db.consume(repo_kind, repo_opts, io, allocator, repo, resolved_ref); 184 }, 185 .xit => { 186 const DB = rp.Repo(.xit, repo_opts).DB; 187 const State = rp.Repo(.xit, repo_opts).State; 188 189 const Ctx = struct { 190 core: *rp.Repo(.xit, repo_opts).Core, 191 io: std.Io, 192 allocator: std.mem.Allocator, 193 ref: rf.Ref, 194 events: []const EventWithId, 195 first_parent_oids: ?[1][hash.hexLen(repo_opts.hash)]u8, 196 197 pub fn run(ctx: @This(), cursor: *DB.Cursor(.read_write)) !void { 198 var moment = try DB.HashMap(.read_write).init(cursor.*); 199 const state = State(.read_write){ .core = ctx.core, .extra = .{ .moment = &moment } }; 200 201 try commitEvents(.xit, repo_opts, state, ctx.io, ctx.allocator, ctx.ref, ctx.events, ctx.first_parent_oids); 202 if (!try consumeInTransaction(.xit, repo_opts, state.readOnly(), &ctx.core.db, &moment, ctx.io, ctx.allocator, ctx.ref)) return error.CancelTransaction; 203 try xit.undo.writeMessage(repo_opts, state, .{ .custom = "event" }); 204 205 // fsync the chunk store, so any chunks written by the 206 // commits are durable before the transaction commits 207 try ctx.core.chunk_store_file.sync(ctx.io); 208 } 209 }; 210 211 try repo.core.db_file.lock(io, .exclusive); 212 defer repo.core.db_file.unlock(io); 213 214 const history = try DB.ArrayList(.read_write).init(repo.core.db.rootCursor()); 215 history.appendContext(.{ .slot = try history.getSlot(-1) }, Ctx{ 216 .core = &repo.core, 217 .io = io, 218 .allocator = allocator, 219 .ref = resolved_ref, 220 .events = events, 221 .first_parent_oids = first_parent_oids, 222 }) catch |err| switch (err) { 223 error.CancelTransaction => {}, 224 else => |e| return e, 225 }; 226 }, 227 } 228 } 229 230 // commit each event as a JSON commit message on `ref` through `state`, so a 231 // xit repo can write them inside an already-open transaction 232 fn commitEvents( 233 comptime repo_kind: rp.RepoKind, 234 comptime repo_opts: rp.RepoOpts(repo_kind), 235 state: rp.Repo(repo_kind, repo_opts).State(.read_write), 236 io: std.Io, 237 allocator: std.mem.Allocator, 238 ref: rf.Ref, 239 events: []const EventWithId, 240 first_parent_oids: ?[1][hash.hexLen(repo_opts.hash)]u8, 241 ) !void { 242 var parent_oids: ?[]const [hash.hexLen(repo_opts.hash)]u8 = if (first_parent_oids) |*oids| oids else null; 243 244 var json: std.Io.Writer.Allocating = .init(allocator); 245 defer json.deinit(); 246 247 for (events) |event| { 248 json.clearRetainingCapacity(); 249 try std.json.Stringify.value(event, .{}, &json.writer); 250 if (json.written().len > max_event_size) return error.EventTooLarge; 251 const author = try std.fmt.allocPrint(allocator, "haxy <{s}>", .{event.author_email}); 252 defer allocator.free(author); 253 _ = try obj.writeCommit(repo_kind, repo_opts, state, io, allocator, .{ .author = author, .message = json.written(), .timestamp = event.timestamp, .parent_oids = parent_oids }, null, ref); 254 // later events parent on the ref's new tip 255 parent_oids = null; 256 } 257 } 258 259 // serve a receive-pack and consume any events it pushed to the events branch, 260 // all in one transaction: the push and the views derived from it commit 261 // atomically, and a failed consume cancels the push 262 pub fn receivePackAndConsume( 263 comptime repo_opts: rp.RepoOpts(.xit), 264 io: std.Io, 265 allocator: std.mem.Allocator, 266 repo: *rp.Repo(.xit, repo_opts), 267 reader: *std.Io.Reader, 268 writer: *std.Io.Writer, 269 options: xit.net_server_receive_pack.Options, 270 ) !void { 271 const DB = rp.Repo(.xit, repo_opts).DB; 272 const State = rp.Repo(.xit, repo_opts).State; 273 274 const Ctx = struct { 275 core: *rp.Repo(.xit, repo_opts).Core, 276 io: std.Io, 277 allocator: std.mem.Allocator, 278 reader: *std.Io.Reader, 279 writer: *std.Io.Writer, 280 options: xit.net_server_receive_pack.Options, 281 282 pub fn run(ctx: @This(), cursor: *DB.Cursor(.read_write)) !void { 283 var moment = try DB.HashMap(.read_write).init(cursor.*); 284 const state = State(.read_write){ .core = ctx.core, .extra = .{ .moment = &moment } }; 285 try xit.net_server_receive_pack.run(.xit, repo_opts, state, ctx.io, ctx.allocator, ctx.reader, ctx.writer, ctx.options); 286 try xit.undo.writeMessage(repo_opts, state, .push); 287 288 // a repo without an events branch has no events to consume. a 289 // no-op consume must not cancel here, since the push shares the 290 // transaction and must commit regardless. 291 if (null == try rf.readRecur(.xit, repo_opts, state.readOnly(), ctx.io, .{ .ref = events_ref })) return; 292 _ = try consumeInTransaction(.xit, repo_opts, state.readOnly(), &ctx.core.db, &moment, ctx.io, ctx.allocator, events_ref); 293 } 294 }; 295 296 try repo.core.db_file.lock(io, .exclusive); 297 defer repo.core.db_file.unlock(io); 298 299 const history = try DB.ArrayList(.read_write).init(repo.core.db.rootCursor()); 300 history.appendContext( 301 .{ .slot = try history.getSlot(-1) }, 302 Ctx{ .core = &repo.core, .io = io, .allocator = allocator, .reader = reader, .writer = writer, .options = options }, 303 ) catch |err| switch (err) { 304 error.CancelTransaction => {}, 305 else => |e| return e, 306 }; 307 308 // pkt-line writes are buffered until the transaction settles, so the 309 // report-status reaches the client only after the push truly committed 310 try writer.flush(); 311 } 312 313 // a standalone event db holding events consumed from a local repo's events 314 // branch. it lives inside the repo dir but is not part of the repo's own data, 315 // so viewing a repo never mutates it. 316 pub fn LocalEventDB(comptime hash_kind: hash.HashKind) type { 317 return struct { 318 file: std.Io.File, 319 buffer: *std.Io.Writer.Allocating, 320 db: *EventDB(hash_kind), 321 322 const db_name = "haxy"; 323 const Self = @This(); 324 325 // open the db in `repo_dir` for writing, creating it if it doesn't exist 326 pub fn open(io: std.Io, allocator: std.mem.Allocator, repo_dir: std.Io.Dir) !Self { 327 const file = try repo_dir.createFile(io, db_name, .{ .truncate = false, .read = true, .lock = .exclusive }); 328 return fromFile(io, allocator, file); 329 } 330 331 // open the db in `repo_dir` for reading, or null if it doesn't exist 332 pub fn openReadOnly(io: std.Io, allocator: std.mem.Allocator, repo_dir: std.Io.Dir) !?Self { 333 const file = repo_dir.openFile(io, db_name, .{ .mode = .read_write, .lock = .shared }) catch |err| switch (err) { 334 error.FileNotFound => return null, 335 else => |e| return e, 336 }; 337 return try fromFile(io, allocator, file); 338 } 339 340 // remove the db from `repo_dir`, so a repo whose events branch 341 // disappeared doesn't keep showing stale issues 342 pub fn delete(io: std.Io, repo_dir: std.Io.Dir) !void { 343 repo_dir.deleteFile(io, db_name) catch |err| switch (err) { 344 error.FileNotFound => {}, 345 else => |e| return e, 346 }; 347 } 348 349 fn fromFile(io: std.Io, allocator: std.mem.Allocator, file: std.Io.File) !Self { 350 errdefer file.close(io); 351 352 const buffer = try allocator.create(std.Io.Writer.Allocating); 353 errdefer allocator.destroy(buffer); 354 buffer.* = std.Io.Writer.Allocating.init(allocator); 355 errdefer buffer.deinit(); 356 357 const db = try allocator.create(EventDB(hash_kind)); 358 errdefer allocator.destroy(db); 359 db.* = try EventDB(hash_kind).init(.{ .io = io, .file = file, .buffer = buffer }); 360 361 return .{ .file = file, .buffer = buffer, .db = db }; 362 } 363 364 pub fn deinit(self: *Self, io: std.Io, allocator: std.mem.Allocator) void { 365 self.file.close(io); 366 self.buffer.deinit(); 367 allocator.destroy(self.buffer); 368 allocator.destroy(self.db); 369 } 370 371 // consume the events on `repo`'s `ref` into this db. the db must come 372 // from `open`, whose handle holds the exclusive lock guarding the 373 // transaction. 374 pub fn consume( 375 self: *Self, 376 comptime repo_kind: rp.RepoKind, 377 comptime repo_opts: rp.RepoOpts(repo_kind), 378 io: std.Io, 379 allocator: std.mem.Allocator, 380 repo: *rp.Repo(repo_kind, repo_opts), 381 ref: rf.Ref, 382 ) !void { 383 comptime if (repo_opts.hash != hash_kind) @compileError("the repo must use the db's hash kind"); 384 const DB = EventDB(hash_kind); 385 386 const Ctx = struct { 387 repo: *rp.Repo(repo_kind, repo_opts), 388 db: *DB, 389 io: std.Io, 390 allocator: std.mem.Allocator, 391 ref: rf.Ref, 392 393 pub fn run(ctx: @This(), cursor: *DB.Cursor(.read_write)) !void { 394 var moment = try DB.HashMap(.read_write).init(cursor.*); 395 var repo_moment = try ctx.repo.core.latestMoment(); 396 const read_state = rp.Repo(repo_kind, repo_opts).State(.read_only){ .core = &ctx.repo.core, .extra = .{ .moment = &repo_moment } }; 397 if (!try consumeInTransaction(repo_kind, repo_opts, read_state, ctx.db, &moment, ctx.io, ctx.allocator, ctx.ref)) return error.CancelTransaction; 398 } 399 }; 400 401 const history = try DB.ArrayList(.read_write).init(self.db.rootCursor()); 402 history.appendContext(.{ .slot = try history.getSlot(-1) }, Ctx{ 403 .repo = repo, 404 .db = self.db, 405 .io = io, 406 .allocator = allocator, 407 .ref = ref, 408 }) catch |err| switch (err) { 409 error.CancelTransaction => {}, 410 else => |e| return e, 411 }; 412 } 413 }; 414 } 415 416 // consume the events on `ref` into the db `moment` writes to, returning false 417 // if there was nothing to do. events are read from `read_state`'s repo, which 418 // needn't be the repo backing the db — that's how a local (possibly 419 // git-backed) repo's events land in a standalone db. 420 pub fn consumeInTransaction( 421 comptime repo_kind: rp.RepoKind, 422 comptime repo_opts: rp.RepoOpts(repo_kind), 423 read_state: rp.Repo(repo_kind, repo_opts).State(.read_only), 424 db: *EventDB(repo_opts.hash), 425 moment: *EventDB(repo_opts.hash).HashMap(.read_write), 426 io: std.Io, 427 allocator: std.mem.Allocator, 428 ref: rf.Ref, 429 ) !bool { 430 const DB = EventDB(repo_opts.hash); 431 432 // the last_object_id represents the object id that was last consumed 433 var last_object_id_maybe: ?[hash.byteLen(repo_opts.hash)]u8 = null; 434 if (try moment.getCursor(hash.hashInt(repo_opts.hash, "haxy-last-object-id"))) |last_object_id_cursor| { 435 var last_object_id_buffer: [hash.byteLen(repo_opts.hash)]u8 = undefined; 436 _ = try last_object_id_cursor.readBytes(&last_object_id_buffer); 437 last_object_id_maybe = last_object_id_buffer; 438 } 439 440 // the tip was already consumed, so there is nothing to do 441 if (last_object_id_maybe) |*last_object_id| { 442 const tip_hex = (try rf.readRecur(repo_kind, repo_opts, read_state, io, .{ .ref = ref })) orelse return error.OidNotFound; 443 var tip: [hash.byteLen(repo_opts.hash)]u8 = undefined; 444 _ = try std.fmt.hexToBytes(&tip, &tip_hex); 445 if (std.mem.eql(u8, last_object_id, &tip)) return false; 446 } 447 448 // the list with all of haxy's state, including materialized views. 449 // the reason it is a list is so we can keep every previous haxy 450 // state, making it easy to revert to an older state if the user 451 // rebases some of the past commits. 452 const haxy_cursor = try moment.putCursor(hash.hashInt(repo_opts.hash, "haxy")); 453 const haxy = try DB.ArrayList(.read_write).init(haxy_cursor); 454 455 // add a new item to the haxy list created above. 456 // we call the item haxy_moments. it is a map of object id to haxy moment. 457 // in other words, it maps each object id to a hash map containing the 458 // state that the database was in when that object id was consumed. 459 var haxy_moments_cursor = try haxy.appendCursor(); 460 // use the previous haxy_moments as the basis for this one if it exists 461 if (try haxy.getCursor(-2)) |last_haxy_moments_cursor| { 462 try haxy_moments_cursor.write(.{ .slot = last_haxy_moments_cursor.slot() }); 463 } 464 var haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor); 465 466 var commit_iter = try CommitIterator(repo_kind, repo_opts).init(read_state, io, allocator, haxy_moments.readOnly(), ref); 467 defer commit_iter.deinit(io, allocator); 468 469 // if there are no events to process, look at the oid at HEAD and update 470 // the last_object_id to point to it. this is important in situations 471 // where we do a merge and then force-push to remove the merge. see the 472 // merge test for an example. 473 if (commit_iter.newest_object_id == null) { 474 const head_oid_hex = (try rf.readRecur(repo_kind, repo_opts, read_state, io, .{ .ref = ref })) orelse return error.OidNotFound; 475 var head_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 476 _ = try std.fmt.hexToBytes(&head_oid, &head_oid_hex); 477 478 const haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &head_oid)) orelse return error.CursorNotFound; 479 const haxy_moment = try DB.HashMap(.read_only).init(haxy_moment_cursor); 480 481 const moment_index_cursor = try haxy_moment.getCursor(hash.hashInt(repo_opts.hash, "moment-index")) orelse return error.CursorNotFound; 482 const moment_index = try moment_index_cursor.readUint(); 483 484 try haxy.slice(moment_index + 1); 485 try moment.put(hash.hashInt(repo_opts.hash, "haxy-last-object-id"), .{ .bytes = &head_oid }); 486 return true; 487 } 488 489 // if this branch was rebased and force pushed, we need to detect that and 490 // properly revert the haxy state to the last valid state. we detect this 491 // by simply asking if the last event is a descendent of the last event 492 // we consumed. if it isn't, then we know the haxy state needs to be reverted. 493 if (last_object_id_maybe) |*last_object_id| { 494 var is_rebased = false; 495 const newest_object_id = commit_iter.newest_object_id orelse return error.OidNotFound; 496 497 _ = mrg.getDescendent( 498 repo_kind, 499 repo_opts, 500 read_state, 501 io, 502 allocator, 503 &std.fmt.bytesToHex(last_object_id, .lower), 504 &std.fmt.bytesToHex(&newest_object_id, .lower), 505 ) catch |err| switch (err) { 506 error.DescendentNotFound => is_rebased = true, 507 else => |e| return e, 508 }; 509 510 if (is_rebased) { 511 const oldest_object_id = commit_iter.oldest_object_id orelse return error.OidNotFound; 512 var oldest_object = try obj.Object(repo_kind, repo_opts).init(read_state, io, allocator, &std.fmt.bytesToHex(oldest_object_id, .lower)); 513 defer oldest_object.deinit(); 514 515 const parent_oids = oldest_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound; 516 switch (parent_oids.len) { 517 0 => { 518 // the branch was rebased all the way to the very beginning. 519 // we have a repo event with no parent, which means it is now 520 // the very first event. all we need to do is set the haxy list 521 // to be empty and make a new haxy_moments map to work with. 522 523 try haxy.slice(0); 524 haxy_moments_cursor = try haxy.appendCursor(); 525 haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor); 526 527 last_object_id_maybe = null; 528 }, 529 1 => { 530 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 531 _ = try std.fmt.hexToBytes(&oid, &parent_oids[0]); 532 533 const old_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &oid)) orelse return error.CursorNotFound; 534 const old_moment = try DB.HashMap(.read_only).init(old_moment_cursor); 535 536 const old_moment_index_cursor = try old_moment.getCursor(hash.hashInt(repo_opts.hash, "moment-index")) orelse return error.CursorNotFound; 537 const old_moment_index = try old_moment_index_cursor.readUint(); 538 539 // resize the haxy list so we truncate all the moments that were 540 // created after the parent_oid was consumed 541 try haxy.slice(old_moment_index + 1); 542 543 // make a new haxy moment and set its initial value to the last haxy moment 544 haxy_moments_cursor = try haxy.appendCursor(); 545 const old_haxy_moments_cursor = try haxy.getCursor(old_moment_index) orelse return error.CursorNotFound; 546 try haxy_moments_cursor.write(.{ .slot = old_haxy_moments_cursor.slot() }); 547 haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor); 548 549 last_object_id_maybe = oid; 550 }, 551 else => return error.UnexpectedParentCount, 552 } 553 } 554 } 555 556 while (try commit_iter.next()) |repo_event_oid| { 557 var commit_object = try obj.Object(repo_kind, repo_opts).init(read_state, io, allocator, &std.fmt.bytesToHex(repo_event_oid, .lower)); 558 defer commit_object.deinit(); 559 560 const parent_oids = commit_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound; 561 562 // create a moment for this object id 563 var haxy_moment_cursor = try haxy_moments.putCursor(hash.bytesToInt(repo_opts.hash, &repo_event_oid)); 564 565 // if there was a previous object id, set this haxy moment's initial value to it. 566 // this efficiently "clones" the map so we make further modifications based on it. 567 if (parent_oids.len > 0) { 568 var first_parent_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 569 _ = try std.fmt.hexToBytes(&first_parent_oid, &parent_oids[0]); 570 571 const first_parent_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &first_parent_oid)) orelse return error.CursorNotFound; 572 try haxy_moment_cursor.write(.{ .slot = first_parent_haxy_moment_cursor.slot() }); 573 } 574 575 const haxy_moment = try DB.HashMap(.read_write).init(haxy_moment_cursor); 576 577 // merge changes from every parent after the first parent. the common 578 // ancestor is the baseline, so a later parent only contributes records 579 // it changed relative to the merge base. only the record maps are 580 // merged; every other view is derived from them, and each kind's 581 // `consume` re-derives its own as the winning records are written. 582 if (parent_oids.len > 1) { 583 for (parent_oids[1..]) |*parent_oid| { 584 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 585 _ = try std.fmt.hexToBytes(&oid, parent_oid); 586 587 const parent_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &oid)) orelse return error.CursorNotFound; 588 const parent_haxy_moment = try DB.HashMap(.read_only).init(parent_haxy_moment_cursor); 589 590 const baseline_oid_hex = try mrg.commonAncestor(repo_kind, repo_opts, read_state, io, allocator, &parent_oids[0], parent_oid); 591 var baseline_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 592 _ = try std.fmt.hexToBytes(&baseline_oid, &baseline_oid_hex); 593 const baseline_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &baseline_oid)) orelse return error.CursorNotFound; 594 const baseline_haxy_moment = try DB.HashMap(.read_only).init(baseline_haxy_moment_cursor); 595 596 inline for (std.meta.fields(Event)) |field| { 597 const T = @typeInfo(field.type).optional.child; 598 try merge(T, DB, repo_opts.hash, allocator, haxy_moment, parent_haxy_moment, baseline_haxy_moment); 599 } 600 } 601 } 602 603 // associate this moment with the index it will first appear at in the haxy list. 604 // this will be important later so we can truncate that list if the user ever 605 // rebases starting at this object id. 606 try haxy_moment.put(hash.hashInt(repo_opts.hash, "moment-index"), .{ .uint = try haxy.count() - 1 }); 607 608 // consume the event unless it's a merge commit 609 if (parent_oids.len <= 1) { 610 var arena = std.heap.ArenaAllocator.init(allocator); 611 defer arena.deinit(); 612 613 // read the message from the commit 614 var message: std.ArrayList(u8) = .empty; 615 try commit_object.readMessage(arena.allocator(), &message, .limited(max_event_size)); 616 617 const event_with_id = try EventWithId.fromString(&arena, message.items); 618 619 // get the id of the current event as bytes 620 var current_event_id: [event_id_size]u8 = undefined; 621 _ = try std.fmt.hexToBytes(&current_event_id, &event_with_id.id); 622 623 // wrap the payload into the record `consume` stores, with its 624 // commit-derived fields; on update, `consume` preserves the 625 // existing record's 626 const created_ts = commit_object.content.commit.metadata.timestamp; 627 switch (event_with_id.event) { 628 .user => |event_maybe| { 629 const record_maybe: ?User.Record = if (event_maybe) |event| .{ .event = event, .created_ts = created_ts } else null; 630 try User.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 631 }, 632 .repo => |event_maybe| { 633 const record_maybe: ?Repo.Record = if (event_maybe) |event| .{ .event = event, .created_ts = created_ts } else null; 634 try Repo.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 635 }, 636 .issue => |event_maybe| { 637 const record_maybe: ?Issue.Record = if (event_maybe) |event| .{ 638 .event = event, 639 .author_email = authorEmail(commit_object.content.commit.metadata.author orelse ""), 640 .created_ts = created_ts, 641 } else null; 642 try Issue.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 643 }, 644 .comment => |event_maybe| { 645 const record_maybe: ?Comment.Record = if (event_maybe) |event| .{ 646 .event = event, 647 .author_email = authorEmail(commit_object.content.commit.metadata.author orelse ""), 648 .created_ts = created_ts, 649 } else null; 650 try Comment.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 651 }, 652 } 653 } 654 655 // the current object id is now the last one 656 last_object_id_maybe = repo_event_oid; 657 658 // prevent any of the data created above from being mutated by future iterations of this loop 659 try db.freeze(); 660 } 661 662 if (last_object_id_maybe) |*last_object_id| { 663 try moment.put(hash.hashInt(repo_opts.hash, "haxy-last-object-id"), .{ .bytes = last_object_id }); 664 } 665 666 return true; 667 } 668 669 // the conflicted records of one kind, keyed by the same order key its id set 670 // uses, so this doubles as the index the conflicts view lists and counts. every 671 // value under an entry is a slot reference to a structure that already exists: 672 // 673 // <order key> -> conflict entry 674 // conflicted-fields the conflicting field names, space separated 675 // base-record the merge base's whole record 676 // their-record the merged-in side's whole record 677 // their-field->oid that side's field name to commit map 678 // 679 // our own value is the live record and our commit is in the kind's oid map, so 680 // neither is repeated here. 681 pub const conflicted_fields_key = "conflicted-fields"; 682 pub const base_record_key = "base-record"; 683 pub const their_record_key = "their-record"; 684 pub const their_field_to_oid_key = "their-field->oid"; 685 686 // record the commit against every field this event changes. a merge passes no 687 // oid, since each field's winner may come from a different side. 688 pub fn writeOid( 689 comptime T: type, 690 comptime DB: type, 691 comptime hash_kind: hash.HashKind, 692 id_to_field_to_oid: DB.HashMap(.read_write), 693 record_key: hash.HashInt(hash_kind), 694 existing_maybe: ?T, 695 event: T, 696 event_oid: ?[]const u8, 697 ) !void { 698 const oid = event_oid orelse return; 699 700 // created only once a field actually changes 701 var field_oids: ?DB.SortedMap(.read_write) = null; 702 703 inline for (std.meta.fields(T)) |field| { 704 const changed = if (existing_maybe) |existing| 705 !fieldEqual(field.type, @field(existing, field.name), @field(event, field.name)) 706 else 707 true; 708 if (changed) { 709 if (field_oids == null) { 710 field_oids = try DB.SortedMap(.read_write).init(try id_to_field_to_oid.putCursor(record_key)); 711 } 712 if (field_oids) |map| try map.put(field.name, .{ .bytes = oid }); 713 } 714 } 715 } 716 717 // the whole field list, space separated, is the longest it can get 718 fn conflictedFieldsMaxLen(comptime T: type) usize { 719 var len: usize = 0; 720 for (std.meta.fields(T)) |field| len += field.name.len + 1; 721 return len; 722 } 723 724 // what a three-way merge did with a field 725 const FieldMerge = enum { 726 // the target's value stands, because neither side changed it, both made the 727 // same change, or only the target changed it 728 kept, 729 parent, 730 conflicted, 731 }; 732 733 // three-way merge each field. one only the parent changed takes the parent's 734 // value; one both sides changed differently keeps the target's. a missing 735 // baseline makes every differing field a conflict. 736 fn mergeFields( 737 comptime T: type, 738 baseline_maybe: ?T, 739 target: T, 740 parent: T, 741 outcome: *[std.meta.fields(T).len]FieldMerge, 742 ) T { 743 var merged = target; 744 745 inline for (std.meta.fields(T), 0..) |field, i| { 746 const target_value = @field(target, field.name); 747 const parent_value = @field(parent, field.name); 748 749 if (!fieldEqual(field.type, target_value, parent_value)) { 750 outcome[i] = if (baseline_maybe) |baseline| blk: { 751 const baseline_value = @field(baseline, field.name); 752 if (fieldEqual(field.type, target_value, baseline_value)) { 753 @field(merged, field.name) = parent_value; 754 break :blk .parent; 755 } 756 // only the target changed it 757 if (fieldEqual(field.type, parent_value, baseline_value)) break :blk .kept; 758 break :blk .conflicted; 759 } else .conflicted; 760 } 761 } 762 763 return merged; 764 } 765 766 // three-way merge of one kind's records from a merge parent. kinds that expose 767 // conflicts merge field by field; the others take one whole record. the result 768 // goes through the kind's `consume`, which re-derives the indexes from it. a 769 // deletion carries over unless the other side changed the record, so a merge 770 // never fails. 771 pub fn merge( 772 comptime T: type, 773 comptime DB: type, 774 comptime hash_kind: hash.HashKind, 775 allocator: std.mem.Allocator, 776 haxy_moment: DB.HashMap(.read_write), 777 parent_moment: DB.HashMap(.read_only), 778 baseline_moment: DB.HashMap(.read_only), 779 ) !void { 780 const parent_set_cursor = try parent_moment.getCursor(hash.hashInt(hash_kind, T.id_set_key)) orelse return; 781 const parent_set = try DB.SortedSet(.read_only).init(parent_set_cursor); 782 783 const parent_records_cursor = try parent_moment.getCursor(hash.hashInt(hash_kind, T.record_map_key)) orelse return; 784 const parent_records = try DB.HashMap(.read_only).init(parent_records_cursor); 785 786 var baseline_records: ?DB.HashMap(.read_only) = null; 787 if (try baseline_moment.getCursor(hash.hashInt(hash_kind, T.record_map_key))) |cursor| { 788 baseline_records = try DB.HashMap(.read_only).init(cursor); 789 } 790 791 var parent_conflicts: ?DB.SortedMap(.read_only) = null; 792 var baseline_conflicts: ?DB.SortedMap(.read_only) = null; 793 var parent_id_to_field_to_oid: ?DB.HashMap(.read_only) = null; 794 switch (comptime T.merge_policy) { 795 .field_conflicts => { 796 if (try parent_moment.getCursor(hash.hashInt(hash_kind, T.conflicts_key))) |cursor| { 797 parent_conflicts = try DB.SortedMap(.read_only).init(cursor); 798 } 799 if (try baseline_moment.getCursor(hash.hashInt(hash_kind, T.conflicts_key))) |cursor| { 800 baseline_conflicts = try DB.SortedMap(.read_only).init(cursor); 801 } 802 if (try parent_moment.getCursor(hash.hashInt(hash_kind, T.id_to_field_to_oid_key))) |cursor| { 803 parent_id_to_field_to_oid = try DB.HashMap(.read_only).init(cursor); 804 } 805 }, 806 .target_wins => {}, 807 } 808 809 var arena = std.heap.ArenaAllocator.init(allocator); 810 defer arena.deinit(); 811 812 var iter = try parent_set.iteratorFromIndex(0); 813 while (try iter.next()) |kv_pair_cursor| { 814 const event_id = try readOrderKeyId(DB, kv_pair_cursor); 815 const record_key = hash.hashInt(hash_kind, &event_id); 816 817 _ = arena.reset(.retain_capacity); 818 819 const parent_record_cursor = try parent_records.getCursor(record_key) orelse continue; 820 const parent_slot = parent_record_cursor.slot(); 821 const baseline_slot = if (baseline_records) |map| try map.getSlot(record_key) else null; 822 823 // the parent left this record alone 824 if (baseline_slot) |slot| { 825 if (slot.eql(parent_slot)) continue; 826 } 827 828 const parent_record = try read(T.Record, DB, hash_kind, &arena, try DB.HashMap(.read_only).init(parent_record_cursor)); 829 830 // re-derived every iteration, since `consume` writes through its own 831 // handles 832 const target_records_cursor = try haxy_moment.putCursor(hash.hashInt(hash_kind, T.record_map_key)); 833 const target_records = try DB.HashMap(.read_write).init(target_records_cursor); 834 835 // a record the target doesn't have arrives wholesale from the parent 836 var outcome: [std.meta.fields(T).len]FieldMerge = @splat(.parent); 837 var merged = parent_record; 838 839 if (try target_records.getCursor(record_key)) |target_record_cursor| { 840 // both sides share the change 841 if (target_record_cursor.slot().eql(parent_slot)) continue; 842 843 const target_record = try read(T.Record, DB, hash_kind, &arena, try DB.HashMap(.read_only).init(target_record_cursor)); 844 845 var baseline_record: ?T.Record = null; 846 if (baseline_records) |map| { 847 if (try map.getCursor(record_key)) |baseline_record_cursor| { 848 baseline_record = try read(T.Record, DB, hash_kind, &arena, try DB.HashMap(.read_only).init(baseline_record_cursor)); 849 } 850 } 851 852 switch (comptime T.merge_policy) { 853 .field_conflicts => { 854 outcome = @splat(.kept); 855 merged = target_record; 856 merged.event = mergeFields(T, if (baseline_record) |baseline| baseline.event else null, target_record.event, parent_record.event, &outcome); 857 858 // merge the tombstone separately from serialized fields. a deletion 859 // carries over an unchanged record, while an edit or restoration 860 // keeps it active. 861 merged.deleted = if (target_record.deleted == parent_record.deleted) 862 target_record.deleted 863 else if (baseline_record) |baseline| blk: { 864 const active = if (!target_record.deleted) target_record else parent_record; 865 if (active.deleted != baseline.deleted) break :blk false; 866 if (!fieldsEqual(T, active.event, baseline.event)) break :blk false; 867 break :blk true; 868 } else false; 869 }, 870 .target_wins => { 871 // an uncontested parent change carries over. if both sides 872 // changed the record, the target wins as a whole, except that 873 // an edit or restoration wins over a deletion. 874 const target_changed = if (baseline_slot) |slot| 875 !target_record_cursor.slot().eql(slot) 876 else 877 true; 878 merged = if (!target_changed) 879 parent_record 880 else if (target_record.deleted != parent_record.deleted) 881 if (target_record.deleted) parent_record else target_record 882 else 883 target_record; 884 }, 885 } 886 } 887 888 // the index is unique, so a key the target gave to a different record 889 // leaves this one out rather than stranding that one behind a name it 890 // no longer owns 891 if (!merged.deleted and @hasDecl(T, "name_index_key")) { 892 const index_key = try merged.indexKey(arena.allocator()); 893 const name_index_cursor = try haxy_moment.putCursor(hash.hashInt(hash_kind, T.name_index_key)); 894 const name_index = try DB.HashMap(.read_write).init(name_index_cursor); 895 if (try name_index.getCursor(hash.hashInt(hash_kind, index_key))) |owner_cursor| { 896 var owner_id: [event_id_size]u8 = undefined; 897 _ = try owner_cursor.readBytes(&owner_id); 898 if (!std.mem.eql(u8, &owner_id, &event_id)) continue; 899 } 900 } 901 902 // `consume` sets no oids, since each field's winner may come from 903 // a different side 904 try T.consume(DB, hash_kind, haxy_moment, &event_id, merged, &arena, null); 905 906 switch (comptime T.merge_policy) { 907 .field_conflicts => {}, 908 .target_wins => continue, 909 } 910 911 // tombstones do not carry field conflicts 912 if (merged.deleted) continue; 913 914 const conflicts_cursor = try haxy_moment.putCursor(hash.hashInt(hash_kind, T.conflicts_key)); 915 const conflicts = try DB.SortedMap(.read_write).init(conflicts_cursor); 916 const id_to_field_to_oid_cursor = try haxy_moment.putCursor(hash.hashInt(hash_kind, T.id_to_field_to_oid_key)); 917 const id_to_field_to_oid = try DB.HashMap(.read_write).init(id_to_field_to_oid_cursor); 918 const merged_key = orderKeyDesc(merged.created_ts, &event_id); 919 920 var parent_field_oids: ?DB.SortedMap(.read_only) = null; 921 if (parent_id_to_field_to_oid) |map| { 922 if (try map.getCursor(record_key)) |field_oids_cursor| { 923 parent_field_oids = try DB.SortedMap(.read_only).init(field_oids_cursor); 924 } 925 } 926 // created only once a field actually comes from the parent 927 var field_oids: ?DB.SortedMap(.read_write) = null; 928 929 // the fields both sides changed differently, space separated 930 var conflicted_fields: [conflictedFieldsMaxLen(T)]u8 = undefined; 931 var conflicted_len: usize = 0; 932 933 inline for (std.meta.fields(T), 0..) |field, i| { 934 switch (outcome[i]) { 935 // the target's value stands, so its oid does too 936 .kept => {}, 937 // the parent set this value, so its commit takes the field. a 938 // parent with no oid took its own value from a merge, leaving 939 // the field unattributed rather than crediting the target's 940 // commit for a value it no longer holds. 941 .parent => { 942 if (field_oids == null) { 943 field_oids = try DB.SortedMap(.read_write).init(try id_to_field_to_oid.putCursor(record_key)); 944 } 945 const map = field_oids orelse unreachable; 946 const parent_oid = if (parent_field_oids) |parent_map| try parent_map.getCursor(field.name) else null; 947 if (parent_oid) |oid_cursor| { 948 try map.put(field.name, .{ .slot = oid_cursor.slot() }); 949 } else { 950 _ = try map.remove(field.name); 951 } 952 }, 953 .conflicted => { 954 if (conflicted_len > 0) { 955 conflicted_fields[conflicted_len] = ' '; 956 conflicted_len += 1; 957 } 958 @memcpy(conflicted_fields[conflicted_len..][0..field.name.len], field.name); 959 conflicted_len += field.name.len; 960 }, 961 } 962 } 963 964 if (conflicted_len > 0) { 965 // each side's whole record is stored by reference, so the view can 966 // show any of them or take theirs wholesale 967 _ = try conflicts.remove(&merged_key); 968 const conflict = try DB.HashMap(.read_write).init(try conflicts.putCursor(&merged_key)); 969 try conflict.put(hash.hashInt(hash_kind, conflicted_fields_key), .{ .bytes = conflicted_fields[0..conflicted_len] }); 970 try conflict.put(hash.hashInt(hash_kind, base_record_key), .{ .slot = baseline_slot }); 971 try conflict.put(hash.hashInt(hash_kind, their_record_key), .{ .slot = parent_slot }); 972 if (parent_field_oids) |map| { 973 try conflict.put(hash.hashInt(hash_kind, their_field_to_oid_key), .{ .slot = map.slot() }); 974 } 975 continue; 976 } 977 978 // created_ts never changes, so the parent keys this record's conflict 979 // identically, and it carries over when the target has none, unless the 980 // parent's entry is unchanged from the baseline, meaning the target 981 // inherited it too and resolving it removed it here. 982 if (parent_conflicts) |map| { 983 if (null == try conflicts.getCursor(&merged_key)) { 984 if (try map.getSlot(&merged_key)) |conflict_slot| { 985 const baseline_conflict_slot = if (baseline_conflicts) |baseline_map| try baseline_map.getSlot(&merged_key) else null; 986 const resolved = if (baseline_conflict_slot) |slot| slot.eql(conflict_slot) else false; 987 if (!resolved) try conflicts.put(&merged_key, .{ .slot = conflict_slot }); 988 } 989 } 990 } 991 } 992 } 993 994 pub fn currentMoment( 995 comptime repo_opts: rp.RepoOpts(.xit), 996 repo: *rp.Repo(.xit, repo_opts), 997 ) !rp.Repo(.xit, repo_opts).DB.HashMap(.read_only) { 998 return currentMomentFromDb(repo_opts.hash, &repo.core.db); 999 } 1000 1001 // the oid of the last event commit consumed into `db`, or null before any 1002 // consume completed 1003 fn lastConsumedOid( 1004 comptime hash_kind: hash.HashKind, 1005 db: *EventDB(hash_kind), 1006 ) !?[hash.byteLen(hash_kind)]u8 { 1007 const DB = EventDB(hash_kind); 1008 const history = try DB.ArrayList(.read_only).init(db.rootCursor().readOnly()); 1009 const moment_cursor = try history.getCursor(-1) orelse return null; 1010 const moment = try DB.HashMap(.read_only).init(moment_cursor); 1011 const last_object_id_cursor = try moment.getCursor(hash.hashInt(hash_kind, "haxy-last-object-id")) orelse return null; 1012 var last_object_id: [hash.byteLen(hash_kind)]u8 = undefined; 1013 _ = try last_object_id_cursor.readBytes(&last_object_id); 1014 return last_object_id; 1015 } 1016 1017 pub fn currentMomentFromDb( 1018 comptime hash_kind: hash.HashKind, 1019 db: *EventDB(hash_kind), 1020 ) !EventDB(hash_kind).HashMap(.read_only) { 1021 const DB = EventDB(hash_kind); 1022 const last_object_id = (try lastConsumedOid(hash_kind, db)) orelse return error.NotFound; 1023 const history = try DB.ArrayList(.read_only).init(db.rootCursor().readOnly()); 1024 const moment_cursor = try history.getCursor(-1) orelse return error.NotFound; 1025 const moment = try DB.HashMap(.read_only).init(moment_cursor); 1026 const haxy_cursor = try moment.getCursor(hash.hashInt(hash_kind, "haxy")) orelse return error.NotFound; 1027 const haxy = try DB.ArrayList(.read_only).init(haxy_cursor); 1028 const haxy_moments_cursor = try haxy.getCursor(-1) orelse return error.NotFound; 1029 const haxy_moments = try DB.HashMap(.read_only).init(haxy_moments_cursor); 1030 const haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(hash_kind, &last_object_id)) orelse return error.NotFound; 1031 return try DB.HashMap(.read_only).init(haxy_moment_cursor); 1032 } 1033 1034 // build the key for SortedSets sorted by timestamp. the big-endian timestamp 1035 // makes byte order match creation order; the event id breaks ties and keeps 1036 // keys unique within the same timestamp. 1037 pub fn orderKey(timestamp: u64, event_id: *const [event_id_size]u8) [@sizeOf(u64) + event_id_size]u8 { 1038 var key: [@sizeOf(u64) + event_id_size]u8 = undefined; 1039 std.mem.writeInt(u64, key[0..@sizeOf(u64)], timestamp, .big); 1040 @memcpy(key[@sizeOf(u64)..], event_id); 1041 return key; 1042 } 1043 1044 pub fn orderKeyDesc(timestamp: u64, event_id: *const [event_id_size]u8) [@sizeOf(u64) + event_id_size]u8 { 1045 return orderKey(std.math.maxInt(u64) - timestamp, event_id); 1046 } 1047 1048 // the event id embedded in a set entry's order key 1049 pub fn readOrderKeyId(comptime DB: type, kv_pair_cursor: DB.Cursor(.read_only)) ![event_id_size]u8 { 1050 var cursor = kv_pair_cursor; 1051 const kv_pair = try cursor.readKeyValuePair(); 1052 var order_key: [@sizeOf(u64) + event_id_size]u8 = undefined; 1053 _ = try kv_pair.key_cursor.readBytes(&order_key); 1054 return order_key[@sizeOf(u64)..][0..event_id_size].*; 1055 } 1056 1057 // split a pushed "<owner>/<repo>" path into its two components, or null if it 1058 // isn't exactly two non-empty segments. 1059 pub fn parseOwnerRepoPath(path: []const u8) ?struct { owner: []const u8, name: []const u8 } { 1060 var it = std.mem.splitScalar(u8, path, '/'); 1061 const owner = it.next() orelse return null; 1062 const name = it.next() orelse return null; 1063 if (it.next() != null) return null; 1064 if (owner.len == 0 or name.len == 0) return null; 1065 1066 return .{ .owner = owner, .name = name }; 1067 } 1068 1069 // resolve a pushed `<owner>/<repo>` to the hex event id that names its on-disk 1070 // directory under the repos dir 1071 pub fn resolveOrCreateRepo( 1072 io: std.Io, 1073 allocator: std.mem.Allocator, 1074 admin_repo_path: []const u8, 1075 owner_name: []const u8, 1076 repo_name: []const u8, 1077 create_if_missing: bool, 1078 ) !?[event_id_size * 2]u8 { 1079 var repo = rp.Repo(.xit, admin_repo_opts).open(io, allocator, .{ .path = admin_repo_path }) catch |err| switch (err) { 1080 error.RepoNotFound => return null, 1081 else => |e| return e, 1082 }; 1083 defer repo.deinit(io, allocator); 1084 1085 var arena = std.heap.ArenaAllocator.init(allocator); 1086 defer arena.deinit(); 1087 1088 const moment = try currentMoment(admin_repo_opts, &repo); 1089 1090 // an already-registered repo reuses its event id, so a re-push (or a clone) 1091 // lands in the same repo 1092 if (try Repo.readByOwnerAndName(AdminDB, admin_repo_opts.hash, moment, &arena, owner_name, repo_name)) |found| { 1093 return std.fmt.bytesToHex(found.event_id, .lower); 1094 } 1095 1096 if (!create_if_missing) return null; 1097 1098 // the new repo is owned by the named user, read from the name index; an 1099 // unknown owner can't own a repo, so the push is rejected 1100 const name_to_user_id_cursor = try moment.getCursor(hash.hashInt(admin_repo_opts.hash, "name->user-id")) orelse return null; 1101 const name_to_user_id = try AdminDB.HashMap(.read_only).init(name_to_user_id_cursor); 1102 const owner_id_cursor = try name_to_user_id.getCursor(hash.hashInt(admin_repo_opts.hash, owner_name)) orelse return null; 1103 var owner_user_id: [event_id_size]u8 = undefined; 1104 _ = try owner_id_cursor.readBytes(&owner_user_id); 1105 1106 const owner = (try User.readById(AdminDB, admin_repo_opts.hash, moment, &arena, &owner_user_id)) orelse unreachable; 1107 1108 var id_bytes: [event_id_size]u8 = undefined; 1109 io.random(&id_bytes); 1110 const event_id_hex = std.fmt.bytesToHex(id_bytes, .lower); 1111 1112 try consume(.xit, admin_repo_opts, io, allocator, &repo, events_ref, &[_]EventWithId{.{ 1113 .id = event_id_hex, 1114 .timestamp = @intCast(std.Io.Timestamp.now(io, .real).toSeconds()), 1115 .author_email = owner.event.email, 1116 .event = .{ .repo = .{ 1117 .user_id = &owner_user_id, 1118 .name = repo_name, 1119 .description = "", 1120 } }, 1121 }}); 1122 1123 return event_id_hex; 1124 } 1125 1126 // 1127 // reading from xitdb 1128 // 1129 1130 // a struct field's fields share its container's map, so their names must not 1131 // collide with the container's or two fields would silently share a key 1132 fn validateFlattenedFields(comptime T: type) void { 1133 comptime { 1134 for (std.meta.fields(T)) |field| { 1135 if (@typeInfo(field.type) != .@"struct") continue; 1136 for (std.meta.fields(field.type)) |nested| { 1137 if (@hasField(T, nested.name)) 1138 @compileError("flattened field name collision in " ++ @typeName(T) ++ ": " ++ nested.name); 1139 } 1140 } 1141 } 1142 } 1143 1144 pub fn read( 1145 comptime T: type, 1146 comptime DB: type, 1147 comptime hash_kind: hash.HashKind, 1148 arena: *std.heap.ArenaAllocator, 1149 map: DB.HashMap(.read_only), 1150 ) !T { 1151 var event: T = undefined; 1152 1153 switch (@typeInfo(T)) { 1154 .@"struct" => |struct_info| { 1155 comptime validateFlattenedFields(T); 1156 inline for (struct_info.fields) |field| { 1157 switch (@typeInfo(field.type)) { 1158 .pointer => |pointer_info| { 1159 if (pointer_info.size == .slice and pointer_info.child == u8 and pointer_info.is_const) { 1160 @field(event, field.name) = try readBytes(DB, hash_kind, arena.allocator(), map, field.name); 1161 } else { 1162 @compileError("unsupported read field type: " ++ @typeName(field.type)); 1163 } 1164 }, 1165 .array => |array_info| { 1166 if (array_info.child == u8) { 1167 const bytes = try readBytes(DB, hash_kind, arena.allocator(), map, field.name); 1168 if (bytes.len != array_info.len) return error.InvalidByteArrayLength; 1169 @memcpy(@field(event, field.name)[0..], bytes); 1170 } else { 1171 @compileError("unsupported read field type: " ++ @typeName(field.type)); 1172 } 1173 }, 1174 .bool => @field(event, field.name) = try readBool(DB, hash_kind, map, field.name), 1175 .int => |int_info| switch (int_info.signedness) { 1176 .unsigned => @field(event, field.name) = @intCast(try readUint(DB, hash_kind, map, field.name)), 1177 .signed => @field(event, field.name) = @intCast(try readInt(DB, hash_kind, map, field.name)), 1178 }, 1179 .@"enum" => { 1180 const bytes = try readBytes(DB, hash_kind, arena.allocator(), map, field.name); 1181 @field(event, field.name) = std.meta.stringToEnum(field.type, bytes) orelse return error.InvalidEnumTag; 1182 }, 1183 .optional => |optional_info| { 1184 if (optional_info.child != []const u8) @compileError("unsupported read field type: " ++ @typeName(field.type)); 1185 // a missing key is null 1186 @field(event, field.name) = if (try map.getCursor(hash.hashInt(hash_kind, field.name))) |cursor| 1187 try cursor.readBytesAlloc(arena.allocator(), null) 1188 else 1189 null; 1190 }, 1191 // a struct field's own fields are read from the same map 1192 .@"struct" => @field(event, field.name) = try read(field.type, DB, hash_kind, arena, map), 1193 else => @compileError("unsupported read field type: " ++ @typeName(field.type)), 1194 } 1195 } 1196 }, 1197 else => @compileError("read expects a struct"), 1198 } 1199 1200 return event; 1201 } 1202 1203 // whether two events carry the same data 1204 fn fieldsEqual(comptime T: type, a: T, b: T) bool { 1205 switch (@typeInfo(T)) { 1206 .@"struct" => |struct_info| { 1207 inline for (struct_info.fields) |field| { 1208 if (!fieldEqual(field.type, @field(a, field.name), @field(b, field.name))) return false; 1209 } 1210 }, 1211 else => @compileError("fieldsEqual expects a struct"), 1212 } 1213 1214 return true; 1215 } 1216 1217 pub fn fieldEqual(comptime Field: type, a: Field, b: Field) bool { 1218 switch (@typeInfo(Field)) { 1219 .pointer => |pointer_info| { 1220 if (pointer_info.size == .slice and pointer_info.child == u8) { 1221 return std.mem.eql(u8, a, b); 1222 } else { 1223 @compileError("unsupported field type: " ++ @typeName(Field)); 1224 } 1225 }, 1226 .array => |array_info| { 1227 if (array_info.child == u8) { 1228 return std.mem.eql(u8, &a, &b); 1229 } else { 1230 @compileError("unsupported field type: " ++ @typeName(Field)); 1231 } 1232 }, 1233 .bool, .int, .@"enum" => return a == b, 1234 else => @compileError("unsupported field type: " ++ @typeName(Field)), 1235 } 1236 } 1237 1238 fn readBytes( 1239 comptime DB: type, 1240 comptime hash_kind: hash.HashKind, 1241 allocator: std.mem.Allocator, 1242 map: DB.HashMap(.read_only), 1243 field_name: []const u8, 1244 ) ![]const u8 { 1245 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 1246 return try cursor.readBytesAlloc(allocator, null); 1247 } 1248 1249 fn readUint( 1250 comptime DB: type, 1251 comptime hash_kind: hash.HashKind, 1252 map: DB.HashMap(.read_only), 1253 field_name: []const u8, 1254 ) !u64 { 1255 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 1256 return try cursor.readUint(); 1257 } 1258 1259 fn readInt( 1260 comptime DB: type, 1261 comptime hash_kind: hash.HashKind, 1262 map: DB.HashMap(.read_only), 1263 field_name: []const u8, 1264 ) !i64 { 1265 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 1266 return try cursor.readInt(); 1267 } 1268 1269 fn readBool( 1270 comptime DB: type, 1271 comptime hash_kind: hash.HashKind, 1272 map: DB.HashMap(.read_only), 1273 field_name: []const u8, 1274 ) !bool { 1275 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 1276 var buffer: [5]u8 = undefined; 1277 const bytes = try cursor.readBytesObject(&buffer); 1278 const format_tag = bytes.format_tag orelse return error.InvalidFormatTag; 1279 if (!std.mem.eql(u8, &format_tag, "bl")) return error.InvalidFormatTag; 1280 if (std.mem.eql(u8, bytes.value, "true")) return true; 1281 if (std.mem.eql(u8, bytes.value, "false")) return false; 1282 return error.InvalidBool; 1283 } 1284 1285 // 1286 // writing to xitdb 1287 // 1288 1289 pub fn upsert( 1290 comptime T: type, 1291 comptime DB: type, 1292 comptime hash_kind: hash.HashKind, 1293 map: DB.HashMap(.read_write), 1294 event: T, 1295 ) !void { 1296 switch (@typeInfo(T)) { 1297 .@"struct" => |struct_info| { 1298 comptime validateFlattenedFields(T); 1299 inline for (struct_info.fields) |field| { 1300 try upsertField(DB, hash_kind, map, field.name, field.type, @field(event, field.name)); 1301 } 1302 }, 1303 else => @compileError("upsert expects a struct"), 1304 } 1305 } 1306 1307 fn upsertField( 1308 comptime DB: type, 1309 comptime hash_kind: hash.HashKind, 1310 map: DB.HashMap(.read_write), 1311 comptime field_name: []const u8, 1312 comptime Field: type, 1313 value: Field, 1314 ) !void { 1315 const key = hash.hashInt(hash_kind, field_name); 1316 1317 switch (@typeInfo(Field)) { 1318 .pointer => |pointer_info| { 1319 if (pointer_info.size == .slice and pointer_info.child == u8) { 1320 try upsertBytes(DB, hash_kind, map, key, value); 1321 } else { 1322 @compileError("unsupported upsert field type: " ++ @typeName(Field)); 1323 } 1324 }, 1325 .array => |array_info| { 1326 if (array_info.child == u8) { 1327 try upsertBytes(DB, hash_kind, map, key, &value); 1328 } else { 1329 @compileError("unsupported upsert field type: " ++ @typeName(Field)); 1330 } 1331 }, 1332 .int => |int_info| switch (int_info.signedness) { 1333 .unsigned => { 1334 if (try map.getCursor(key)) |value_cursor| { 1335 if (try value_cursor.readUint() == value) { 1336 return; 1337 } 1338 } 1339 1340 try map.put(key, .{ .uint = value }); 1341 }, 1342 .signed => { 1343 if (try map.getCursor(key)) |value_cursor| { 1344 if (try value_cursor.readInt() == value) { 1345 return; 1346 } 1347 } 1348 1349 try map.put(key, .{ .int = value }); 1350 }, 1351 }, 1352 .bool => { 1353 const bytes = if (value) "true" else "false"; 1354 if (try map.getCursor(key)) |value_cursor| { 1355 var buffer: [5]u8 = undefined; 1356 const existing = try value_cursor.readBytesObject(&buffer); 1357 if (existing.format_tag) |format_tag| { 1358 if (std.mem.eql(u8, &format_tag, "bl") and std.mem.eql(u8, existing.value, bytes)) { 1359 return; 1360 } 1361 } 1362 } 1363 1364 try map.put(key, .{ .bytes_object = .{ .value = bytes, .format_tag = "bl".* } }); 1365 }, 1366 .@"enum" => try upsertBytes(DB, hash_kind, map, key, @tagName(value)), 1367 .optional => |optional_info| { 1368 if (optional_info.child != []const u8) @compileError("unsupported upsert field type: " ++ @typeName(Field)); 1369 // a missing key is null 1370 if (value) |bytes| { 1371 try upsertBytes(DB, hash_kind, map, key, bytes); 1372 } else { 1373 _ = try map.remove(key); 1374 } 1375 }, 1376 // a struct field's own fields are stored in the same map 1377 .@"struct" => try upsert(Field, DB, hash_kind, map, value), 1378 else => @compileError("unsupported upsert field type: " ++ @typeName(Field)), 1379 } 1380 } 1381 1382 fn upsertBytes( 1383 comptime DB: type, 1384 comptime hash_kind: hash.HashKind, 1385 map: DB.HashMap(.read_write), 1386 key: hash.HashInt(hash_kind), 1387 value: []const u8, 1388 ) !void { 1389 var existing_cursor_maybe = try map.getCursor(key); 1390 if (existing_cursor_maybe) |*existing_cursor| { 1391 if (try bytesEqual(DB, existing_cursor, value)) { 1392 return; 1393 } 1394 } 1395 1396 var value_cursor = try map.putCursor(key); 1397 var write_buffer: [1024]u8 = undefined; 1398 var writer = try value_cursor.writer(&write_buffer); 1399 try writer.interface.writeAll(value); 1400 try writer.finish(); 1401 } 1402 1403 fn bytesEqual( 1404 comptime DB: type, 1405 cursor: *DB.Cursor(.read_only), 1406 value: []const u8, 1407 ) !bool { 1408 var read_buffer: [1024]u8 = undefined; 1409 var reader = try cursor.reader(&read_buffer); 1410 if (reader.size != value.len) { 1411 return false; 1412 } 1413 1414 var chunk_buffer: [1024]u8 = undefined; 1415 var offset: usize = 0; 1416 while (offset < value.len) { 1417 const chunk_len = @min(chunk_buffer.len, value.len - offset); 1418 try reader.interface.readSliceAll(chunk_buffer[0..chunk_len]); 1419 if (!std.mem.eql(u8, chunk_buffer[0..chunk_len], value[offset .. offset + chunk_len])) { 1420 return false; 1421 } 1422 offset += chunk_len; 1423 } 1424 1425 return true; 1426 } 1427 1428 // 1429 // CommitIterator 1430 // 1431 // differs from xit's ObjectIterator in a few ways: 1432 // 1433 // 1. iterates parents first, rather than children first 1434 // 2. `next` only returns the object id, not the full Object 1435 // 3. stores its temporary state in a file on disk to avoid OOMs 1436 // 1437 1438 pub fn CommitIterator(comptime repo_kind: rp.RepoKind, comptime repo_opts: rp.RepoOpts(repo_kind)) type { 1439 return struct { 1440 const DB = EventDB(repo_opts.hash); 1441 const db_name = "haxy-repo-events.db"; 1442 1443 repo_dir: std.Io.Dir, 1444 db_file: std.Io.File, 1445 db: *DB, 1446 // map of each object id to its children 1447 parent_to_children: DB.HashMap(.read_write), 1448 // tracking pending parents is necessary because a child can only 1449 // be returned after all of its unconsumed parents have been returned 1450 child_to_pending_parents: DB.HashMap(.read_write), 1451 // queue of object ids that are ready to be returned from `next` 1452 ready_queue: DB.ArrayList(.read_write), 1453 ready_queue_index: u64, 1454 // the object id at the tip of this branch 1455 newest_object_id: ?[hash.byteLen(repo_opts.hash)]u8, 1456 // the object id furthest back in the history that hasn't been consumed 1457 oldest_object_id: ?[hash.byteLen(repo_opts.hash)]u8, 1458 1459 pub fn init( 1460 state: rp.Repo(repo_kind, repo_opts).State(.read_only), 1461 io: std.Io, 1462 allocator: std.mem.Allocator, 1463 consumed_object_ids: DB.HashMap(.read_only), 1464 ref: rf.Ref, 1465 ) !CommitIterator(repo_kind, repo_opts) { 1466 const db_file = try state.core.repo_dir.createFile(io, db_name, .{ .truncate = true, .lock = .exclusive, .read = true }); 1467 errdefer { 1468 db_file.close(io); 1469 state.core.repo_dir.deleteFile(io, db_name) catch {}; 1470 } 1471 1472 const buffer_ptr = try allocator.create(std.Io.Writer.Allocating); 1473 errdefer allocator.destroy(buffer_ptr); 1474 1475 buffer_ptr.* = std.Io.Writer.Allocating.init(allocator); 1476 errdefer buffer_ptr.deinit(); 1477 1478 const db_ptr = try allocator.create(DB); 1479 errdefer allocator.destroy(db_ptr); 1480 // the db is scratch state that gets deleted after iteration, so 1481 // there is nothing worth fsyncing 1482 db_ptr.* = try DB.init(.{ .io = io, .file = db_file, .buffer = buffer_ptr, .fsync = false }); 1483 1484 const map = try DB.HashMap(.read_write).init(db_ptr.rootCursor()); 1485 1486 const parent_to_children_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "parent->children")); 1487 const parent_to_children = try DB.HashMap(.read_write).init(parent_to_children_cursor); 1488 1489 const child_to_pending_parents_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "child->pending-parents")); 1490 const child_to_pending_parents = try DB.HashMap(.read_write).init(child_to_pending_parents_cursor); 1491 1492 const ready_queue_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "ready-queue")); 1493 const ready_queue = try DB.ArrayList(.read_write).init(ready_queue_cursor); 1494 1495 var self = CommitIterator(repo_kind, repo_opts){ 1496 .repo_dir = state.core.repo_dir, 1497 .db_file = db_file, 1498 .db = db_ptr, 1499 .parent_to_children = parent_to_children, 1500 .child_to_pending_parents = child_to_pending_parents, 1501 .ready_queue = ready_queue, 1502 .ready_queue_index = 0, 1503 .newest_object_id = null, 1504 .oldest_object_id = null, 1505 }; 1506 errdefer self.deinit(io, allocator); 1507 1508 try self.collect(state, io, allocator, consumed_object_ids, ref); 1509 1510 return self; 1511 } 1512 1513 pub fn deinit(self: *CommitIterator(repo_kind, repo_opts), io: std.Io, allocator: std.mem.Allocator) void { 1514 self.db_file.close(io); 1515 self.db.core.memory.buffer.deinit(); 1516 allocator.destroy(self.db.core.memory.buffer); 1517 self.repo_dir.deleteFile(io, db_name) catch {}; 1518 allocator.destroy(self.db); 1519 } 1520 1521 pub fn next(self: *CommitIterator(repo_kind, repo_opts)) !?[hash.byteLen(repo_opts.hash)]u8 { 1522 if (self.ready_queue_index >= try self.ready_queue.count()) return null; 1523 1524 const oid = try readOidFromList(self.ready_queue.readOnly(), self.ready_queue_index); 1525 self.ready_queue_index += 1; 1526 1527 try self.enqueueChildren(&oid); 1528 return oid; 1529 } 1530 1531 fn collect( 1532 self: *CommitIterator(repo_kind, repo_opts), 1533 state: rp.Repo(repo_kind, repo_opts).State(.read_only), 1534 io: std.Io, 1535 allocator: std.mem.Allocator, 1536 consumed_object_ids: DB.HashMap(.read_only), 1537 ref: rf.Ref, 1538 ) !void { 1539 const map = try DB.HashMap(.read_write).init(self.db.rootCursor()); 1540 1541 const walk_queue_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "walk-queue")); 1542 const walk_queue = try DB.ArrayList(.read_write).init(walk_queue_cursor); 1543 var walk_queue_index: u64 = 0; 1544 1545 const seen_object_ids_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "seen-object-ids")); 1546 const seen_object_ids = try DB.HashSet(.read_write).init(seen_object_ids_cursor); 1547 1548 const head_oid_hex = (try rf.readRecur(repo_kind, repo_opts, state, io, .{ .ref = ref })) orelse return error.OidNotFound; 1549 var head_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 1550 _ = try std.fmt.hexToBytes(&head_oid, &head_oid_hex); 1551 try walk_queue.append(.{ .bytes = &head_oid }); 1552 1553 while (walk_queue_index < try walk_queue.count()) { 1554 const oid = try readOidFromList(walk_queue.readOnly(), walk_queue_index); 1555 const oid_int = hash.bytesToInt(repo_opts.hash, &oid); 1556 1557 walk_queue_index += 1; 1558 1559 // if we've seen this object id, skip it 1560 if (null != try seen_object_ids.getCursor(oid_int)) { 1561 continue; 1562 } 1563 1564 try seen_object_ids.put(oid_int, .{ .bytes = &oid }); 1565 1566 var commit_object = try obj.Object(repo_kind, repo_opts).init(state, io, allocator, &std.fmt.bytesToHex(oid, .lower)); 1567 defer commit_object.deinit(); 1568 1569 // if this object id has already been consumed, skip it 1570 if (null != try consumed_object_ids.getCursor(oid_int)) { 1571 continue; 1572 } 1573 1574 if (self.newest_object_id == null) self.newest_object_id = oid; 1575 self.oldest_object_id = oid; 1576 1577 const parent_oids = commit_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound; 1578 var pending_parent_count: u64 = 0; 1579 for (parent_oids) |*parent_oid| { 1580 var parent_oid_bytes: [hash.byteLen(repo_opts.hash)]u8 = undefined; 1581 _ = try std.fmt.hexToBytes(&parent_oid_bytes, parent_oid); 1582 const parent_oid_int = hash.bytesToInt(repo_opts.hash, &parent_oid_bytes); 1583 1584 // if this object id has already been consumed, skip it 1585 if (null != try consumed_object_ids.getCursor(parent_oid_int)) { 1586 continue; 1587 } 1588 1589 // if this object id hasn't already been seen, add it to the walk queue 1590 if (null == try seen_object_ids.getCursor(parent_oid_int)) { 1591 try walk_queue.append(.{ .bytes = &parent_oid_bytes }); 1592 } 1593 1594 const children_cursor = try self.parent_to_children.putCursor(parent_oid_int); 1595 const children = try DB.HashSet(.read_write).init(children_cursor); 1596 try children.put(oid_int, .{ .bytes = &oid }); 1597 pending_parent_count += 1; 1598 } 1599 1600 if (pending_parent_count == 0) { 1601 try self.ready_queue.append(.{ .bytes = &oid }); 1602 } else { 1603 try self.child_to_pending_parents.put(oid_int, .{ .uint = pending_parent_count }); 1604 } 1605 } 1606 } 1607 1608 fn enqueueChildren(self: *CommitIterator(repo_kind, repo_opts), oid: *const [hash.byteLen(repo_opts.hash)]u8) !void { 1609 const oid_int = hash.bytesToInt(repo_opts.hash, oid); 1610 const children_cursor = try self.parent_to_children.getCursor(oid_int) orelse return; 1611 const children = try DB.HashSet(.read_only).init(children_cursor); 1612 var children_iter = try children.iterator(); 1613 1614 while (try children_iter.next()) |*child_cursor| { 1615 const kv_pair = try child_cursor.readKeyValuePair(); 1616 const child_oid_int = kv_pair.hash; 1617 const pending_parent_cursor = try self.child_to_pending_parents.getCursor(child_oid_int) orelse return error.CursorNotFound; 1618 const pending_parent_count = try pending_parent_cursor.readUint(); 1619 1620 if (pending_parent_count <= 1) { 1621 _ = try self.child_to_pending_parents.remove(child_oid_int); 1622 const child_oid = hash.intToBytes(hash.HashInt(repo_opts.hash), child_oid_int); 1623 try self.ready_queue.append(.{ .bytes = &child_oid }); 1624 } else { 1625 try self.child_to_pending_parents.put(child_oid_int, .{ .uint = pending_parent_count - 1 }); 1626 } 1627 } 1628 } 1629 1630 fn readOidFromList(list: DB.ArrayList(.read_only), index: u64) ![hash.byteLen(repo_opts.hash)]u8 { 1631 const oid_cursor = try list.getCursor(index) orelse return error.CursorNotFound; 1632 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 1633 _ = try oid_cursor.readBytes(&oid); 1634 return oid; 1635 } 1636 }; 1637 } 1638