╭ event.zig ────────────────────────────────────────────────────╮ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ╰───────────────────────────────────────────────────────────────╯
🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 ╭0a9c9e35..╮ 0a9c9e35.. see-em: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 │files │ commits refs discuss issues new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────────╯ (123) user
╭────╮ ╭ clone ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ╮ │http│ ssh │ │ ╰────╯ ╰ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ╯
.. embed/ event/ test/ ui/ command.zig ╭─────────────────────────────────╮ │event.zig │ ╰─────────────────────────────────╯ lib.zig main.zig main_wasm.zig serve.zig serve_ssh.zig serve_ssh_protocol.zig test.zig testnet.zig try.zig ui.zig web.zig
1 const std = @import("std"); 2 const xit = @import("xit"); 3 const rp = xit.repo; 4 const hash = xit.hash; 5 const mrg = xit.merge; 6 const obj = xit.object; 7 const rf = xit.ref; 8 9 pub const User = @import("event/User.zig"); 10 pub const Repo = @import("event/Repo.zig"); 11 pub const Issue = @import("event/Issue.zig"); 12 13 pub const event_id_size: usize = 32; 14 15 pub const EventKind = enum { 16 user, 17 repo, 18 issue, 19 }; 20 21 pub const EventWithId = struct { 22 id: [event_id_size * 2]u8, 23 event: union(EventKind) { 24 user: ?User, 25 repo: ?Repo, 26 issue: ?Issue, 27 }, 28 29 pub fn jsonStringify(self: EventWithId, jw: anytype) !void { 30 try jw.beginObject(); 31 try jw.objectField("id"); 32 try jw.write(self.id); 33 try jw.objectField("kind"); 34 try jw.write(@tagName(self.event)); 35 try jw.objectField("data"); 36 switch (self.event) { 37 inline else => |event_maybe| try jw.write(event_maybe), 38 } 39 try jw.endObject(); 40 } 41 42 fn fromString(arena: *std.heap.ArenaAllocator, message: []const u8) !EventWithId { 43 const JsonEvent = struct { 44 id: [event_id_size * 2]u8, 45 kind: EventKind, 46 data: ?std.json.Value = null, 47 }; 48 const json_event = try std.json.parseFromSliceLeaky(JsonEvent, arena.allocator(), message, .{}); 49 return .{ 50 .id = json_event.id, 51 .event = switch (json_event.kind) { 52 .user => .{ 53 .user = if (json_event.data) |value| 54 try std.json.parseFromValueLeaky(User, arena.allocator(), value, .{}) 55 else 56 null, 57 }, 58 .repo => .{ 59 .repo = if (json_event.data) |value| 60 try std.json.parseFromValueLeaky(Repo, arena.allocator(), value, .{}) 61 else 62 null, 63 }, 64 .issue => .{ 65 .issue = if (json_event.data) |value| 66 try std.json.parseFromValueLeaky(Issue, arena.allocator(), value, .{}) 67 else 68 null, 69 }, 70 }, 71 }; 72 } 73 74 pub fn randomId(random: std.Random) [event_id_size]u8 { 75 var id_bytes: [event_id_size]u8 = undefined; 76 random.bytes(&id_bytes); 77 return id_bytes; 78 } 79 }; 80 81 pub fn consume( 82 comptime repo_opts: rp.RepoOpts(.xit), 83 io: std.Io, 84 allocator: std.mem.Allocator, 85 repo: *rp.Repo(.xit, repo_opts), 86 ref: rf.Ref, 87 ) !void { 88 const DB = rp.Repo(.xit, repo_opts).DB; 89 const State = rp.Repo(.xit, repo_opts).State; 90 91 const Ctx = struct { 92 core: *rp.Repo(.xit, repo_opts).Core, 93 io: std.Io, 94 allocator: std.mem.Allocator, 95 ref: rf.Ref, 96 97 pub fn run(ctx: @This(), cursor: *DB.Cursor(.read_write)) !void { 98 var moment = try DB.HashMap(.read_write).init(cursor.*); 99 const state = State(.read_write){ .core = ctx.core, .extra = .{ .moment = &moment } }; 100 try consumeInTransaction(repo_opts, state, ctx.io, ctx.allocator, ctx.ref); 101 } 102 }; 103 104 try repo.core.db_file.lock(io, .exclusive); 105 defer repo.core.db_file.unlock(io); 106 107 const history = try DB.ArrayList(.read_write).init(repo.core.db.rootCursor()); 108 try history.appendContext(.{ .slot = try history.getSlot(-1) }, Ctx{ 109 .core = &repo.core, 110 .io = io, 111 .allocator = allocator, 112 .ref = ref, 113 }); 114 } 115 116 pub fn consumeInTransaction( 117 comptime repo_opts: rp.RepoOpts(.xit), 118 state: rp.Repo(.xit, repo_opts).State(.read_write), 119 io: std.Io, 120 allocator: std.mem.Allocator, 121 ref: rf.Ref, 122 ) !void { 123 const DB = rp.Repo(.xit, repo_opts).DB; 124 125 // the last_object_id represents the object id that was last consumed 126 var last_object_id_maybe: ?[hash.byteLen(repo_opts.hash)]u8 = null; 127 if (try state.extra.moment.getCursor(hash.hashInt(repo_opts.hash, "haxy-last-object-id"))) |last_object_id_cursor| { 128 var last_object_id_buffer: [hash.byteLen(repo_opts.hash)]u8 = undefined; 129 _ = try last_object_id_cursor.readBytes(&last_object_id_buffer); 130 last_object_id_maybe = last_object_id_buffer; 131 } 132 133 // the list with all of haxy's state, including materialized views. 134 // the reason it is a list is so we can keep every previous haxy 135 // state, making it easy to revert to an older state if the user 136 // rebases some of the past commits. 137 const haxy_cursor = try state.extra.moment.putCursor(hash.hashInt(repo_opts.hash, "haxy")); 138 const haxy = try DB.ArrayList(.read_write).init(haxy_cursor); 139 140 // add a new item to the haxy list created above. 141 // we call the item haxy_moments. it is a map of object id to haxy moment. 142 // in other words, it maps each object id to a hash map containing the 143 // state that the database was in when that object id was consumed. 144 var haxy_moments_cursor = try haxy.appendCursor(); 145 // use the previous haxy_moments as the basis for this one if it exists 146 if (try haxy.getCursor(-2)) |last_haxy_moments_cursor| { 147 try haxy_moments_cursor.write(.{ .slot = last_haxy_moments_cursor.slot() }); 148 } 149 var haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor); 150 151 var commit_iter = try CommitIterator(repo_opts).init(state.readOnly(), io, allocator, haxy_moments.readOnly(), ref); 152 defer commit_iter.deinit(io, allocator); 153 154 // if there are no events to process, look at the oid at HEAD and update 155 // the last_object_id to point to it. this is important in situations 156 // where we do a merge and then force-push to remove the merge. see the 157 // merge test for an example. 158 if (commit_iter.newest_object_id == null) { 159 const head_oid_hex = (try rf.readRecur(.xit, repo_opts, state.readOnly(), io, .{ .ref = ref })) orelse return error.OidNotFound; 160 var head_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 161 _ = try std.fmt.hexToBytes(&head_oid, &head_oid_hex); 162 163 const haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &head_oid)) orelse return error.CursorNotFound; 164 const haxy_moment = try DB.HashMap(.read_only).init(haxy_moment_cursor); 165 166 const moment_index_cursor = try haxy_moment.getCursor(hash.hashInt(repo_opts.hash, "moment-index")) orelse return error.CursorNotFound; 167 const moment_index = try moment_index_cursor.readUint(); 168 169 try haxy.slice(moment_index + 1); 170 try state.extra.moment.put(hash.hashInt(repo_opts.hash, "haxy-last-object-id"), .{ .bytes = &head_oid }); 171 return; 172 } 173 174 // if this branch was rebased and force pushed, we need to detect that and 175 // properly revert the haxy state to the last valid state. we detect this 176 // by simply asking if the last event is a descendent of the last event 177 // we consumed. if it isn't, then we know the haxy state needs to be reverted. 178 if (last_object_id_maybe) |*last_object_id| { 179 var is_rebased = false; 180 const newest_object_id = commit_iter.newest_object_id orelse return error.OidNotFound; 181 182 _ = mrg.getDescendent( 183 .xit, 184 repo_opts, 185 state.readOnly(), 186 io, 187 allocator, 188 &std.fmt.bytesToHex(last_object_id, .lower), 189 &std.fmt.bytesToHex(&newest_object_id, .lower), 190 ) catch |err| switch (err) { 191 error.DescendentNotFound => is_rebased = true, 192 else => |e| return e, 193 }; 194 195 if (is_rebased) { 196 const oldest_object_id = commit_iter.oldest_object_id orelse return error.OidNotFound; 197 var oldest_object = try obj.Object(.xit, repo_opts, .full).init(state.readOnly(), io, allocator, &std.fmt.bytesToHex(oldest_object_id, .lower)); 198 defer oldest_object.deinit(); 199 200 const parent_oids = oldest_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound; 201 switch (parent_oids.len) { 202 0 => { 203 // the branch was rebased all the way to the very beginning. 204 // we have a repo event with no parent, which means it is now 205 // the very first event. all we need to do is set the haxy list 206 // to be empty and make a new haxy_moments map to work with. 207 208 try haxy.slice(0); 209 haxy_moments_cursor = try haxy.appendCursor(); 210 haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor); 211 212 last_object_id_maybe = null; 213 }, 214 1 => { 215 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 216 _ = try std.fmt.hexToBytes(&oid, &parent_oids[0]); 217 218 const old_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &oid)) orelse return error.CursorNotFound; 219 const old_moment = try DB.HashMap(.read_only).init(old_moment_cursor); 220 221 const old_moment_index_cursor = try old_moment.getCursor(hash.hashInt(repo_opts.hash, "moment-index")) orelse return error.CursorNotFound; 222 const old_moment_index = try old_moment_index_cursor.readUint(); 223 224 // resize the haxy list so we truncate all the moments that were 225 // created after the parent_oid was consumed 226 try haxy.slice(old_moment_index + 1); 227 228 // make a new haxy moment and set its initial value to the last haxy moment 229 haxy_moments_cursor = try haxy.appendCursor(); 230 const old_haxy_moments_cursor = try haxy.getCursor(old_moment_index) orelse return error.CursorNotFound; 231 try haxy_moments_cursor.write(.{ .slot = old_haxy_moments_cursor.slot() }); 232 haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor); 233 234 last_object_id_maybe = oid; 235 }, 236 else => return error.UnexpectedParentCount, 237 } 238 } 239 } 240 241 while (try commit_iter.next()) |repo_event_oid| { 242 var commit_object = try obj.Object(.xit, repo_opts, .full).init(state.readOnly(), io, allocator, &std.fmt.bytesToHex(repo_event_oid, .lower)); 243 defer commit_object.deinit(); 244 245 const parent_oids = commit_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound; 246 247 // create a moment for this object id 248 var haxy_moment_cursor = try haxy_moments.putCursor(hash.bytesToInt(repo_opts.hash, &repo_event_oid)); 249 250 // if there was a previous object id, set this haxy moment's initial value to it. 251 // this efficiently "clones" the map so we make further modifications based on it. 252 if (parent_oids.len > 0) { 253 var first_parent_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 254 _ = try std.fmt.hexToBytes(&first_parent_oid, &parent_oids[0]); 255 256 const first_parent_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &first_parent_oid)) orelse return error.CursorNotFound; 257 try haxy_moment_cursor.write(.{ .slot = first_parent_haxy_moment_cursor.slot() }); 258 } 259 260 const haxy_moment = try DB.HashMap(.read_write).init(haxy_moment_cursor); 261 262 // merge changes from every parent after the first parent. the common 263 // ancestor is the baseline, so a later parent only contributes values 264 // it changed relative to the merge base. 265 if (parent_oids.len > 1) { 266 for (parent_oids[1..]) |*parent_oid| { 267 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 268 _ = try std.fmt.hexToBytes(&oid, parent_oid); 269 270 const parent_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &oid)) orelse return error.CursorNotFound; 271 const parent_haxy_moment = try DB.HashMap(.read_only).init(parent_haxy_moment_cursor); 272 273 const baseline_oid_hex = try mrg.commonAncestor(.xit, repo_opts, state.readOnly(), io, allocator, &parent_oids[0], parent_oid); 274 var baseline_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 275 _ = try std.fmt.hexToBytes(&baseline_oid, &baseline_oid_hex); 276 const baseline_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &baseline_oid)) orelse return error.CursorNotFound; 277 const baseline_haxy_moment = try DB.HashMap(.read_only).init(baseline_haxy_moment_cursor); 278 279 try mergeChangedMapEntries(DB, haxy_moment, parent_haxy_moment, baseline_haxy_moment, true); 280 } 281 } 282 283 // associate this moment with the index it will first appear at in the haxy list. 284 // this will be important later so we can truncate that list if the user ever 285 // rebases starting at this object id. 286 try haxy_moment.put(hash.hashInt(repo_opts.hash, "moment-index"), .{ .uint = try haxy.count() - 1 }); 287 288 // consume the event unless it's a merge commit 289 if (parent_oids.len <= 1) { 290 var arena = std.heap.ArenaAllocator.init(allocator); 291 defer arena.deinit(); 292 293 // read the message from the commit 294 try commit_object.object_reader.seekTo(commit_object.content.commit.message_position); 295 const message = try commit_object.object_reader.interface.allocRemaining(arena.allocator(), .unlimited); 296 297 const event_with_id = try EventWithId.fromString(&arena, message); 298 299 // get the id of the current event as bytes 300 var current_event_id: [event_id_size]u8 = undefined; 301 _ = try std.fmt.hexToBytes(&current_event_id, &event_with_id.id); 302 303 switch (event_with_id.event) { 304 .user => |event_maybe| try User.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, event_maybe, &arena), 305 .repo => |event_maybe| try Repo.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, event_maybe, &arena), 306 .issue => |event_maybe| try Issue.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, event_maybe), 307 } 308 } 309 310 // the current object id is now the last one 311 last_object_id_maybe = repo_event_oid; 312 313 // prevent any of the data created above from being mutated by future iterations of this loop 314 try state.core.db.freeze(); 315 } 316 317 if (last_object_id_maybe) |*last_object_id| { 318 try state.extra.moment.put(hash.hashInt(repo_opts.hash, "haxy-last-object-id"), .{ .bytes = last_object_id }); 319 } 320 } 321 322 // 323 // reading from xitdb 324 // 325 326 pub fn read( 327 comptime T: type, 328 comptime DB: type, 329 comptime hash_kind: hash.HashKind, 330 arena: *std.heap.ArenaAllocator, 331 map: DB.HashMap(.read_only), 332 ) !T { 333 var event: T = undefined; 334 335 switch (@typeInfo(T)) { 336 .@"struct" => |struct_info| { 337 inline for (struct_info.fields) |field| { 338 switch (@typeInfo(field.type)) { 339 .pointer => |pointer_info| { 340 if (pointer_info.size == .slice and pointer_info.child == u8 and pointer_info.is_const) { 341 @field(event, field.name) = try readBytes(DB, hash_kind, arena.allocator(), map, field.name); 342 } else { 343 @compileError("unsupported read field type: " ++ @typeName(field.type)); 344 } 345 }, 346 .array => |array_info| { 347 if (array_info.child == u8) { 348 const bytes = try readBytes(DB, hash_kind, arena.allocator(), map, field.name); 349 if (bytes.len != array_info.len) return error.InvalidByteArrayLength; 350 @memcpy(@field(event, field.name)[0..], bytes); 351 } else { 352 @compileError("unsupported read field type: " ++ @typeName(field.type)); 353 } 354 }, 355 .bool => @field(event, field.name) = try readBool(DB, hash_kind, map, field.name), 356 else => @compileError("unsupported read field type: " ++ @typeName(field.type)), 357 } 358 } 359 }, 360 else => @compileError("read expects a struct"), 361 } 362 363 return event; 364 } 365 366 fn readBytes( 367 comptime DB: type, 368 comptime hash_kind: hash.HashKind, 369 allocator: std.mem.Allocator, 370 map: DB.HashMap(.read_only), 371 field_name: []const u8, 372 ) ![]const u8 { 373 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 374 return try cursor.readBytesAlloc(allocator, null); 375 } 376 377 fn readBool( 378 comptime DB: type, 379 comptime hash_kind: hash.HashKind, 380 map: DB.HashMap(.read_only), 381 field_name: []const u8, 382 ) !bool { 383 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 384 var buffer: [5]u8 = undefined; 385 const bytes = try cursor.readBytesObject(&buffer); 386 const format_tag = bytes.format_tag orelse return error.InvalidFormatTag; 387 if (!std.mem.eql(u8, &format_tag, "bl")) return error.InvalidFormatTag; 388 if (std.mem.eql(u8, bytes.value, "true")) return true; 389 if (std.mem.eql(u8, bytes.value, "false")) return false; 390 return error.InvalidBool; 391 } 392 393 // 394 // writing to xitdb 395 // 396 397 pub fn upsert( 398 comptime T: type, 399 comptime DB: type, 400 comptime hash_kind: hash.HashKind, 401 map: DB.HashMap(.read_write), 402 event: T, 403 ) !void { 404 switch (@typeInfo(T)) { 405 .@"struct" => |struct_info| { 406 inline for (struct_info.fields) |field| { 407 try upsertField(DB, hash_kind, map, field.name, field.type, @field(event, field.name)); 408 } 409 }, 410 else => @compileError("upsert expects a struct"), 411 } 412 } 413 414 fn mergeChangedMapEntries( 415 comptime DB: type, 416 target: DB.HashMap(.read_write), 417 parent: DB.HashMap(.read_only), 418 baseline: DB.HashMap(.read_only), 419 comptime is_top_level: bool, 420 ) !void { 421 var parent_iter = try parent.iterator(); 422 while (try parent_iter.next()) |kv_pair_cursor| { 423 const kv_pair = try kv_pair_cursor.readKeyValuePair(); 424 425 const baseline_value_cursor = try baseline.getCursor(kv_pair.hash) orelse { 426 if ((try target.getCursor(kv_pair.hash)) != null) { 427 return error.MergeConflict; 428 } 429 430 try target.put(kv_pair.hash, .{ .slot = kv_pair.value_cursor.slot() }); 431 continue; 432 }; 433 434 const tag = kv_pair.value_cursor.slot().tag; 435 436 if (tag != baseline_value_cursor.slot().tag) { 437 return error.UnexpectedTag; 438 } 439 440 if (kv_pair.value_cursor.slot().value == baseline_value_cursor.slot().value) { 441 continue; 442 } 443 444 if (tag == .hash_map) { 445 const target_existing_cursor = try target.getCursor(kv_pair.hash) orelse return error.MergeConflict; 446 if (target_existing_cursor.slot().tag != .hash_map) { 447 return error.MergeConflict; 448 } 449 450 const target_child_cursor = try target.putCursor(kv_pair.hash); 451 const target_child = try DB.HashMap(.read_write).init(target_child_cursor); 452 const parent_child = try DB.HashMap(.read_only).init(kv_pair.value_cursor); 453 const baseline_child = try DB.HashMap(.read_only).init(baseline_value_cursor); 454 try mergeChangedMapEntries(DB, target_child, parent_child, baseline_child, false); 455 } else if (is_top_level) { 456 // the moment-index is a top-level key in the haxy moment 457 // and is not a map. we don't want to do any merge of this. 458 continue; 459 } else { 460 const target_value_cursor = try target.getCursor(kv_pair.hash) orelse return error.MergeConflict; 461 if (target_value_cursor.slot().value != baseline_value_cursor.slot().value) { 462 return error.MergeConflict; 463 } 464 465 try target.put(kv_pair.hash, .{ .slot = kv_pair.value_cursor.slot() }); 466 } 467 } 468 } 469 470 fn upsertField( 471 comptime DB: type, 472 comptime hash_kind: hash.HashKind, 473 map: DB.HashMap(.read_write), 474 comptime field_name: []const u8, 475 comptime Field: type, 476 value: Field, 477 ) !void { 478 const key = hash.hashInt(hash_kind, field_name); 479 480 switch (@typeInfo(Field)) { 481 .pointer => |pointer_info| { 482 if (pointer_info.size == .slice and pointer_info.child == u8) { 483 try upsertBytes(DB, hash_kind, map, key, value); 484 } else { 485 @compileError("unsupported upsert field type: " ++ @typeName(Field)); 486 } 487 }, 488 .array => |array_info| { 489 if (array_info.child == u8) { 490 try upsertBytes(DB, hash_kind, map, key, &value); 491 } else { 492 @compileError("unsupported upsert field type: " ++ @typeName(Field)); 493 } 494 }, 495 .int => |int_info| switch (int_info.signedness) { 496 .unsigned => { 497 if (try map.getCursor(key)) |value_cursor| { 498 if (try value_cursor.readUint() == value) { 499 return; 500 } 501 } 502 503 try map.put(key, .{ .uint = value }); 504 }, 505 .signed => { 506 if (try map.getCursor(key)) |value_cursor| { 507 if (try value_cursor.readInt() == value) { 508 return; 509 } 510 } 511 512 try map.put(key, .{ .int = value }); 513 }, 514 }, 515 .bool => { 516 const bytes = if (value) "true" else "false"; 517 if (try map.getCursor(key)) |value_cursor| { 518 var buffer: [5]u8 = undefined; 519 const existing = try value_cursor.readBytesObject(&buffer); 520 if (existing.format_tag) |format_tag| { 521 if (std.mem.eql(u8, &format_tag, "bl") and std.mem.eql(u8, existing.value, bytes)) { 522 return; 523 } 524 } 525 } 526 527 try map.put(key, .{ .bytes_object = .{ .value = bytes, .format_tag = "bl".* } }); 528 }, 529 else => @compileError("unsupported upsert field type: " ++ @typeName(Field)), 530 } 531 } 532 533 fn upsertBytes( 534 comptime DB: type, 535 comptime hash_kind: hash.HashKind, 536 map: DB.HashMap(.read_write), 537 key: hash.HashInt(hash_kind), 538 value: []const u8, 539 ) !void { 540 var existing_cursor_maybe = try map.getCursor(key); 541 if (existing_cursor_maybe) |*existing_cursor| { 542 if (try bytesEqual(DB, existing_cursor, value)) { 543 return; 544 } 545 } 546 547 var value_cursor = try map.putCursor(key); 548 var write_buffer: [1024]u8 = undefined; 549 var writer = try value_cursor.writer(&write_buffer); 550 try writer.interface.writeAll(value); 551 try writer.finish(); 552 } 553 554 fn bytesEqual( 555 comptime DB: type, 556 cursor: *DB.Cursor(.read_only), 557 value: []const u8, 558 ) !bool { 559 var read_buffer: [1024]u8 = undefined; 560 var reader = try cursor.reader(&read_buffer); 561 if (reader.size != value.len) { 562 return false; 563 } 564 565 var chunk_buffer: [1024]u8 = undefined; 566 var offset: usize = 0; 567 while (offset < value.len) { 568 const chunk_len = @min(chunk_buffer.len, value.len - offset); 569 try reader.interface.readSliceAll(chunk_buffer[0..chunk_len]); 570 if (!std.mem.eql(u8, chunk_buffer[0..chunk_len], value[offset .. offset + chunk_len])) { 571 return false; 572 } 573 offset += chunk_len; 574 } 575 576 return true; 577 } 578 579 // 580 // CommitIterator 581 // 582 // differs from xit's ObjectIterator in a few ways: 583 // 584 // 1. iterates parents first, rather than children first 585 // 2. `next` only returns the object id, not the full Object 586 // 3. stores its temporary state in a file on disk to avoid OOMs 587 // 588 589 pub fn CommitIterator(comptime repo_opts: rp.RepoOpts(.xit)) type { 590 return struct { 591 const DB = rp.Repo(.xit, repo_opts).DB; 592 const db_name = "haxy-repo-events.db"; 593 594 repo_dir: std.Io.Dir, 595 db_file: std.Io.File, 596 db: *DB, 597 // map of each object id to its children 598 parent_to_children: DB.HashMap(.read_write), 599 // tracking pending parents is necessary because a child can only 600 // be returned after all of its unconsumed parents have been returned 601 child_to_pending_parents: DB.HashMap(.read_write), 602 // queue of object ids that are ready to be returned from `next` 603 ready_queue: DB.ArrayList(.read_write), 604 ready_queue_index: u64, 605 // the object id at the tip of this branch 606 newest_object_id: ?[hash.byteLen(repo_opts.hash)]u8, 607 // the object id furthest back in the history that hasn't been consumed 608 oldest_object_id: ?[hash.byteLen(repo_opts.hash)]u8, 609 610 pub fn init( 611 state: rp.Repo(.xit, repo_opts).State(.read_only), 612 io: std.Io, 613 allocator: std.mem.Allocator, 614 consumed_object_ids: DB.HashMap(.read_only), 615 ref: rf.Ref, 616 ) !CommitIterator(repo_opts) { 617 const db_file = try state.core.repo_dir.createFile(io, db_name, .{ .truncate = true, .lock = .exclusive, .read = true }); 618 errdefer { 619 db_file.close(io); 620 state.core.repo_dir.deleteFile(io, db_name) catch {}; 621 } 622 623 const buffer_ptr = try allocator.create(std.Io.Writer.Allocating); 624 errdefer allocator.destroy(buffer_ptr); 625 626 buffer_ptr.* = std.Io.Writer.Allocating.init(allocator); 627 errdefer buffer_ptr.deinit(); 628 629 const db_ptr = try allocator.create(DB); 630 errdefer allocator.destroy(db_ptr); 631 db_ptr.* = try DB.init(.{ .io = io, .file = db_file, .buffer = buffer_ptr }); 632 633 const map = try DB.HashMap(.read_write).init(db_ptr.rootCursor()); 634 635 const parent_to_children_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "parent->children")); 636 const parent_to_children = try DB.HashMap(.read_write).init(parent_to_children_cursor); 637 638 const child_to_pending_parents_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "child->pending-parents")); 639 const child_to_pending_parents = try DB.HashMap(.read_write).init(child_to_pending_parents_cursor); 640 641 const ready_queue_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "ready-queue")); 642 const ready_queue = try DB.ArrayList(.read_write).init(ready_queue_cursor); 643 644 var self = CommitIterator(repo_opts){ 645 .repo_dir = state.core.repo_dir, 646 .db_file = db_file, 647 .db = db_ptr, 648 .parent_to_children = parent_to_children, 649 .child_to_pending_parents = child_to_pending_parents, 650 .ready_queue = ready_queue, 651 .ready_queue_index = 0, 652 .newest_object_id = null, 653 .oldest_object_id = null, 654 }; 655 errdefer self.deinit(io, allocator); 656 657 try self.collect(state, io, allocator, consumed_object_ids, ref); 658 659 return self; 660 } 661 662 pub fn deinit(self: *CommitIterator(repo_opts), io: std.Io, allocator: std.mem.Allocator) void { 663 self.db_file.close(io); 664 self.db.core.memory.buffer.deinit(); 665 allocator.destroy(self.db.core.memory.buffer); 666 self.repo_dir.deleteFile(io, db_name) catch {}; 667 allocator.destroy(self.db); 668 } 669 670 pub fn next(self: *CommitIterator(repo_opts)) !?[hash.byteLen(repo_opts.hash)]u8 { 671 if (self.ready_queue_index >= try self.ready_queue.count()) return null; 672 673 const oid = try readOidFromList(self.ready_queue.readOnly(), self.ready_queue_index); 674 self.ready_queue_index += 1; 675 676 try self.enqueueChildren(&oid); 677 return oid; 678 } 679 680 fn collect( 681 self: *CommitIterator(repo_opts), 682 state: rp.Repo(.xit, repo_opts).State(.read_only), 683 io: std.Io, 684 allocator: std.mem.Allocator, 685 consumed_object_ids: DB.HashMap(.read_only), 686 ref: rf.Ref, 687 ) !void { 688 const map = try DB.HashMap(.read_write).init(self.db.rootCursor()); 689 690 const walk_queue_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "walk-queue")); 691 const walk_queue = try DB.ArrayList(.read_write).init(walk_queue_cursor); 692 var walk_queue_index: u64 = 0; 693 694 const seen_object_ids_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "seen-object-ids")); 695 const seen_object_ids = try DB.HashSet(.read_write).init(seen_object_ids_cursor); 696 697 const head_oid_hex = (try rf.readRecur(.xit, repo_opts, state, io, .{ .ref = ref })) orelse return error.OidNotFound; 698 var head_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 699 _ = try std.fmt.hexToBytes(&head_oid, &head_oid_hex); 700 try walk_queue.append(.{ .bytes = &head_oid }); 701 702 while (walk_queue_index < try walk_queue.count()) { 703 const oid = try readOidFromList(walk_queue.readOnly(), walk_queue_index); 704 const oid_int = hash.bytesToInt(repo_opts.hash, &oid); 705 706 walk_queue_index += 1; 707 708 // if we've seen this object id, skip it 709 if (null != try seen_object_ids.getCursor(oid_int)) { 710 continue; 711 } 712 713 try seen_object_ids.put(oid_int, .{ .bytes = &oid }); 714 715 var commit_object = try obj.Object(.xit, repo_opts, .full).init(state, io, allocator, &std.fmt.bytesToHex(oid, .lower)); 716 defer commit_object.deinit(); 717 718 // if this object id has already been consumed, skip it 719 if (null != try consumed_object_ids.getCursor(oid_int)) { 720 continue; 721 } 722 723 if (self.newest_object_id == null) self.newest_object_id = oid; 724 self.oldest_object_id = oid; 725 726 const parent_oids = commit_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound; 727 var pending_parent_count: u64 = 0; 728 for (parent_oids) |*parent_oid| { 729 var parent_oid_bytes: [hash.byteLen(repo_opts.hash)]u8 = undefined; 730 _ = try std.fmt.hexToBytes(&parent_oid_bytes, parent_oid); 731 const parent_oid_int = hash.bytesToInt(repo_opts.hash, &parent_oid_bytes); 732 733 // if this object id has already been consumed, skip it 734 if (null != try consumed_object_ids.getCursor(parent_oid_int)) { 735 continue; 736 } 737 738 // if this object id hasn't already been seen, add it to the walk queue 739 if (null == try seen_object_ids.getCursor(parent_oid_int)) { 740 try walk_queue.append(.{ .bytes = &parent_oid_bytes }); 741 } 742 743 const children_cursor = try self.parent_to_children.putCursor(parent_oid_int); 744 const children = try DB.HashSet(.read_write).init(children_cursor); 745 try children.put(oid_int, .{ .bytes = &oid }); 746 pending_parent_count += 1; 747 } 748 749 if (pending_parent_count == 0) { 750 try self.ready_queue.append(.{ .bytes = &oid }); 751 } else { 752 try self.child_to_pending_parents.put(oid_int, .{ .uint = pending_parent_count }); 753 } 754 } 755 } 756 757 fn enqueueChildren(self: *CommitIterator(repo_opts), oid: *const [hash.byteLen(repo_opts.hash)]u8) !void { 758 const oid_int = hash.bytesToInt(repo_opts.hash, oid); 759 const children_cursor = try self.parent_to_children.getCursor(oid_int) orelse return; 760 const children = try DB.HashSet(.read_only).init(children_cursor); 761 var children_iter = try children.iterator(); 762 763 while (try children_iter.next()) |*child_cursor| { 764 const kv_pair = try child_cursor.readKeyValuePair(); 765 const child_oid_int = kv_pair.hash; 766 const pending_parent_cursor = try self.child_to_pending_parents.getCursor(child_oid_int) orelse return error.CursorNotFound; 767 const pending_parent_count = try pending_parent_cursor.readUint(); 768 769 if (pending_parent_count <= 1) { 770 _ = try self.child_to_pending_parents.remove(child_oid_int); 771 const child_oid = hash.intToBytes(hash.HashInt(repo_opts.hash), child_oid_int); 772 try self.ready_queue.append(.{ .bytes = &child_oid }); 773 } else { 774 try self.child_to_pending_parents.put(child_oid_int, .{ .uint = pending_parent_count - 1 }); 775 } 776 } 777 } 778 779 fn readOidFromList(list: DB.ArrayList(.read_only), index: u64) ![hash.byteLen(repo_opts.hash)]u8 { 780 const oid_cursor = try list.getCursor(index) orelse return error.CursorNotFound; 781 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 782 _ = try oid_cursor.readBytes(&oid); 783 return oid; 784 } 785 }; 786 } 787