╭─────────────╮ │scroll to top│ ╰─────────────╯ ╭ event.zig ────────────────────────────────────────────────────╮ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ╰───────────────────────────────────────────────────────────────╯
🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 ╭062dbf30..╮ 062dbf30.. see-em: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 │files │ commits refs discuss issues new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────────╯ (45) user
╭────╮ ╭ clone ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ╮ │http│ ssh │ │ ╰────╯ ╰ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ╯
.. event/ test/ command.zig ╭─────────────────────────────────╮ │event.zig │ ╰─────────────────────────────────╯ lib.zig main.zig serve.zig ssh_helper.zig test.zig testnet.zig
50 .user => .{ 51 .user = .{ 52 .name = try readBytes(DB, hash_kind, allocator, map, "name"), 53 .email = try readBytes(DB, hash_kind, allocator, map, "email"), 54 .password_hash = try readBytes(DB, hash_kind, allocator, map, "password_hash"), 55 }, 56 }, 57 .repo => .{ 58 .repo = .{ 59 .user_id = try readBytes(DB, hash_kind, allocator, map, "user_id"), 60 .name = try readBytes(DB, hash_kind, allocator, map, "name"), 61 .enable_issue = try readBool(DB, hash_kind, map, "enable_issue"), 62 }, 63 }, 64 .issue => .{ 65 .issue = .{ 66 .title = try readBytes(DB, hash_kind, allocator, map, "title"), 67 .description = try readBytes(DB, hash_kind, allocator, map, "description"), 68 .tags = try readBytes(DB, hash_kind, allocator, map, "tags"), 69 }, 70 }, 71 }; 72 } 73 74 pub fn fromJson( 75 allocator: std.mem.Allocator, 76 kind: EventKind, 77 source: std.json.Value, 78 ) !EventData { 79 return switch (kind) { 80 .user => .{ 81 .user = try std.json.parseFromValueLeaky(User, allocator, source, .{}), 82 }, 83 .repo => .{ 84 .repo = try std.json.parseFromValueLeaky(Repo, allocator, source, .{}), 85 }, 86 .issue => .{ 87 .issue = try std.json.parseFromValueLeaky(Issue, allocator, source, .{}), 88 }, 89 }; 90 } 91 92 fn readBytes( 93 comptime DB: type, 94 comptime hash_kind: hash.HashKind, 95 allocator: std.mem.Allocator, 96 map: DB.HashMap(.read_only), 97 field_name: []const u8, 98 ) ![]const u8 { 99 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 100 return try cursor.readBytesAlloc(allocator, null); 101 } 102 103 fn readBool( 104 comptime DB: type, 105 comptime hash_kind: hash.HashKind, 106 map: DB.HashMap(.read_only), 107 field_name: []const u8, 108 ) !bool { 109 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 110 var buffer: [5]u8 = undefined; 111 const bytes = try cursor.readBytesObject(&buffer); 112 if (bytes.format_tag == null or !std.mem.eql(u8, &bytes.format_tag.?, "bl")) return error.InvalidFormatTag; 113 if (std.mem.eql(u8, bytes.value, "true")) return true; 114 if (std.mem.eql(u8, bytes.value, "false")) return false; 115 return error.InvalidBool; 116 } 117 }; 118 119 pub const Event = struct { 120 id: [event_id_size * 2]u8, 121 kind: EventKind, 122 data: ?EventData = null, 123 124 pub fn jsonStringify(self: Event, jw: anytype) !void { 125 try jw.beginObject(); 126 try jw.objectField("id"); 127 try jw.write(self.id); 128 try jw.objectField("kind"); 129 try jw.write(self.kind); 130 try jw.objectField("data"); 131 if (self.data) |data_maybe| { 132 switch (data_maybe) { 133 .user => |data| try jw.write(data), 134 .repo => |data| try jw.write(data), 135 .issue => |data| try jw.write(data), 136 } 137 } else { 138 try jw.write(null); 139 } 140 try jw.endObject(); 141 } 142 143 fn fromString(allocator: std.mem.Allocator, message: []const u8) !Event { 144 const JsonEvent = struct { 145 id: [event_id_size * 2]u8, 146 kind: EventKind, 147 data: ?std.json.Value = null, 148 }; 149 const json_event = try std.json.parseFromSliceLeaky(JsonEvent, allocator, message, .{}); 150 return .{ 151 .id = json_event.id, 152 .kind = json_event.kind, 153 .data = if (json_event.data) |data| 154 try EventData.fromJson(allocator, json_event.kind, data) 155 else 156 null, 157 }; 158 } 159 }; 160 161 pub fn randomId(random: std.Random) [event_id_size]u8 { 162 var id_bytes: [event_id_size]u8 = undefined; 163 random.bytes(&id_bytes); 164 return id_bytes; 165 } 166 167 pub fn consume( 168 comptime repo_opts: rp.RepoOpts(.xit), 169 io: std.Io, 170 allocator: std.mem.Allocator, 171 repo: *rp.Repo(.xit, repo_opts), 172 ref: rf.Ref, 173 ) !void { 174 const DB = rp.Repo(.xit, repo_opts).DB; 175 const State = rp.Repo(.xit, repo_opts).State; 176 177 const Ctx = struct { 178 core: *rp.Repo(.xit, repo_opts).Core, 179 io: std.Io, 180 allocator: std.mem.Allocator, 181 ref: rf.Ref, 182 183 pub fn run(ctx: @This(), cursor: *DB.Cursor(.read_write)) !void { 184 var moment = try DB.HashMap(.read_write).init(cursor.*); 185 const state = State(.read_write){ .core = ctx.core, .extra = .{ .moment = &moment } }; 186 try consumeInTransaction(repo_opts, state, ctx.io, ctx.allocator, ctx.ref); 187 } 188 }; 189 190 try repo.core.db_file.lock(io, .exclusive); 191 defer repo.core.db_file.unlock(io); 192 193 const history = try DB.ArrayList(.read_write).init(repo.core.db.rootCursor()); 194 try history.appendContext(.{ .slot = try history.getSlot(-1) }, Ctx{ 195 .core = &repo.core, 196 .io = io, 197 .allocator = allocator, 198 .ref = ref, 199 }); 200 } 201 202 pub fn consumeInTransaction( 203 comptime repo_opts: rp.RepoOpts(.xit), 204 state: rp.Repo(.xit, repo_opts).State(.read_write), 205 io: std.Io, 206 allocator: std.mem.Allocator, 207 ref: rf.Ref, 208 ) !void { 209 const DB = rp.Repo(.xit, repo_opts).DB; 210 211 // the last_object_id represents the object id that was last consumed 212 var last_object_id_maybe: ?[hash.byteLen(repo_opts.hash)]u8 = null; 213 if (try state.extra.moment.getCursor(hash.hashInt(repo_opts.hash, "haxy-last-object-id"))) |last_object_id_cursor| { 214 var last_object_id_buffer: [hash.byteLen(repo_opts.hash)]u8 = undefined; 215 _ = try last_object_id_cursor.readBytes(&last_object_id_buffer); 216 last_object_id_maybe = last_object_id_buffer; 217 } 218 219 // the list with all of haxy's state, including materialized views. 220 // the reason it is a list is so we can keep every previous haxy 221 // state, making it easy to revert to an older state if the user 222 // rebases some of the past commits. 223 const haxy_cursor = try state.extra.moment.putCursor(hash.hashInt(repo_opts.hash, "haxy")); 224 const haxy = try DB.ArrayList(.read_write).init(haxy_cursor); 225 226 // add a new item to the haxy list created above. 227 // we call the item haxy_moments. it is a map of object id to haxy moment. 228 // in other words, it maps each object id to a hash map containing the 229 // state that the database was in when that object id was consumed. 230 var haxy_moments_cursor = try haxy.appendCursor(); 231 // use the previous haxy_moments as the basis for this one if it exists 232 if (try haxy.getCursor(-2)) |last_haxy_moments_cursor| { 233 try haxy_moments_cursor.write(.{ .slot = last_haxy_moments_cursor.slot() }); 234 } 235 var haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor); 236 237 var arena = std.heap.ArenaAllocator.init(allocator); 238 defer arena.deinit(); 239 240 // compute a list of events that haven't been consumed yet 241 const RepoEvent = struct { 242 parent_oids: []const [hash.hexLen(repo_opts.hash)]u8, 243 oid: [hash.byteLen(repo_opts.hash)]u8, 244 }; 245 var repo_events: std.ArrayList(RepoEvent) = .empty; 246 defer repo_events.deinit(allocator); 247 { 248 var commit_iter = try obj.ObjectIterator(.xit, repo_opts, .full).init(state.readOnly(), io, allocator, .{ .kind = .commit }); 249 defer commit_iter.deinit(); 250 251 const head_oid = (try rf.readRecur(.xit, repo_opts, state.readOnly(), io, .{ .ref = ref })) orelse return error.OidNotFound; 252 try commit_iter.include(&head_oid); 253 254 while (try commit_iter.next(arena.allocator())) |commit_object| { 255 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 256 _ = try std.fmt.hexToBytes(&oid, &commit_object.oid); 257 258 // if this object id has already been consumed, skip it 259 if (null != try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &oid))) { 260 continue; 261 } 262 263 const parent_oids = commit_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound; 264 try repo_events.append(allocator, .{ .parent_oids = parent_oids, .oid = oid }); 265 } 266 } 267 268 // if there are no events to process, look at the oid at HEAD and update 269 // the last_object_id to point to it. this is important in situations 270 // where we do a merge and then force-push to remove the merge. see the 271 // merge test for an example. 272 if (repo_events.items.len == 0) { 273 const head_oid_hex = (try rf.readRecur(.xit, repo_opts, state.readOnly(), io, .{ .ref = ref })) orelse return error.OidNotFound; 274 var head_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 275 _ = try std.fmt.hexToBytes(&head_oid, &head_oid_hex); 276 277 const haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &head_oid)) orelse return error.CursorNotFound; 278 const haxy_moment = try DB.HashMap(.read_only).init(haxy_moment_cursor); 279 280 const moment_index_cursor = try haxy_moment.getCursor(hash.hashInt(repo_opts.hash, "moment-index")) orelse return error.CursorNotFound; 281 const moment_index = try moment_index_cursor.readUint(); 282 283 try haxy.slice(moment_index + 1); 284 try state.extra.moment.put(hash.hashInt(repo_opts.hash, "haxy-last-object-id"), .{ .bytes = &head_oid }); 285 return; 286 } 287 288 // if this branch was rebased and force pushed, we need to detect that and 289 // properly revert the haxy state to the last valid state. we detect this 290 // by simply asking if the last event is a descendent of the last event 291 // we consumed. if it isn't, then we know the haxy state needs to be reverted. 292 if (last_object_id_maybe) |*last_object_id| { 293 var is_rebased = false; 294 295 _ = mrg.getDescendent( 296 .xit, 297 repo_opts, 298 state.readOnly(), 299 io, 300 allocator, 301 &std.fmt.bytesToHex(last_object_id, .lower), 302 &std.fmt.bytesToHex(repo_events.items[0].oid, .lower), 303 ) catch |err| switch (err) { 304 error.DescendentNotFound => is_rebased = true, 305 else => |e| return e, 306 }; 307 308 if (is_rebased) { 309 const parent_oids = repo_events.items[repo_events.items.len - 1].parent_oids; 310 switch (parent_oids.len) { 311 0 => { 312 // the branch was rebased all the way to the very beginning. 313 // we have a repo event with no parent, which means it is now 314 // the very first event. all we need to do is set the haxy list 315 // to be empty and make a new haxy_moments map to work with. 316 317 try haxy.slice(0); 318 haxy_moments_cursor = try haxy.appendCursor(); 319 haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor); 320 321 last_object_id_maybe = null; 322 }, 323 1 => { 324 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 325 _ = try std.fmt.hexToBytes(&oid, &parent_oids[0]); 326 327 const old_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &oid)) orelse return error.CursorNotFound; 328 const old_moment = try DB.HashMap(.read_only).init(old_moment_cursor); 329 330 const old_moment_index_cursor = try old_moment.getCursor(hash.hashInt(repo_opts.hash, "moment-index")) orelse return error.CursorNotFound; 331 const old_moment_index = try old_moment_index_cursor.readUint(); 332 333 // resize the haxy list so we truncate all the moments that were 334 // created after the parent_oid was consumed 335 try haxy.slice(old_moment_index + 1); 336 337 // make a new haxy moment and set its initial value to the last haxy moment 338 haxy_moments_cursor = try haxy.appendCursor(); 339 const old_haxy_moments_cursor = try haxy.getCursor(old_moment_index) orelse return error.CursorNotFound; 340 try haxy_moments_cursor.write(.{ .slot = old_haxy_moments_cursor.slot() }); 341 haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor); 342 343 last_object_id_maybe = oid; 344 }, 345 else => return error.UnexpectedParentCount, 346 } 347 } 348 } else { 349 if (repo_events.items[repo_events.items.len - 1].parent_oids.len != 0) { 350 // there is no last_object_id, but this event has a parent. 351 // this is an invalid state. if the event has a parent, that 352 // implies that an event has already been processed, but 353 // if that's true then there would be a last_object_id. 354 return error.UnexpectedParent; 355 } 356 } 357 358 // iterate over the events in reverse order 359 // (they are sorted by most recent events first) 360 for (0..repo_events.items.len) |i| { 361 const repo_event = repo_events.items[repo_events.items.len - i - 1]; 362 363 // create a moment for this object id 364 var haxy_moment_cursor = try haxy_moments.putCursor(hash.bytesToInt(repo_opts.hash, &repo_event.oid)); 365 366 // if there was a previous object id, set this haxy moment's initial value to it. 367 // this efficiently "clones" the map so we make further modifications based on it. 368 if (repo_event.parent_oids.len > 0) { 369 var first_parent_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 370 _ = try std.fmt.hexToBytes(&first_parent_oid, &repo_event.parent_oids[0]); 371 372 const first_parent_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &first_parent_oid)) orelse return error.CursorNotFound; 373 try haxy_moment_cursor.write(.{ .slot = first_parent_haxy_moment_cursor.slot() }); 374 } 375 376 const haxy_moment = try DB.HashMap(.read_write).init(haxy_moment_cursor); 377 378 // merge changes from every parent after the first parent. the common 379 // ancestor is the baseline, so a later parent only contributes values 380 // it changed relative to the merge base. 381 if (repo_event.parent_oids.len > 1) { 382 for (repo_event.parent_oids[1..]) |*parent_oid| { 383 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 384 _ = try std.fmt.hexToBytes(&oid, parent_oid); 385 386 const parent_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &oid)) orelse return error.CursorNotFound; 387 const parent_haxy_moment = try DB.HashMap(.read_only).init(parent_haxy_moment_cursor); 388 389 const baseline_oid_hex = try mrg.commonAncestor(.xit, repo_opts, state.readOnly(), io, allocator, &repo_event.parent_oids[0], parent_oid); 390 var baseline_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 391 _ = try std.fmt.hexToBytes(&baseline_oid, &baseline_oid_hex); 392 const baseline_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &baseline_oid)) orelse return error.CursorNotFound; 393 const baseline_haxy_moment = try DB.HashMap(.read_only).init(baseline_haxy_moment_cursor); 394 395 try mergeChangedMapEntries(DB, haxy_moment, parent_haxy_moment, baseline_haxy_moment, true); 396 } 397 } 398 399 // associate this moment with the index it will first appear at in the haxy list. 400 // this will be important later so we can truncate that list if the user ever 401 // rebases starting at this object id. 402 try haxy_moment.put(hash.hashInt(repo_opts.hash, "moment-index"), .{ .uint = try haxy.count() - 1 }); 403 404 // consume the event unless it's a merge commit 405 if (repo_event.parent_oids.len <= 1) { 406 var commit_object = try obj.Object(.xit, repo_opts, .full).init(state.readOnly(), io, allocator, &std.fmt.bytesToHex(repo_event.oid, .lower)); 407 defer commit_object.deinit(); 408 409 // read the message from the commit 410 try commit_object.object_reader.seekTo(commit_object.content.commit.message_position); 411 const message = try commit_object.object_reader.interface.allocRemaining(arena.allocator(), .unlimited); 412 413 const event = try Event.fromString(arena.allocator(), message); 414 415 // get the id of the current event as bytes 416 var current_event_id: [event_id_size]u8 = undefined; 417 _ = try std.fmt.hexToBytes(&current_event_id, &event.id); 418 419 switch (event.kind) { 420 .user => { 421 const user_key = hash.hashInt(repo_opts.hash, &current_event_id); 422 423 const event_id_to_user_cursor = try haxy_moment.putCursor(hash.hashInt(repo_opts.hash, "event-id->user")); 424 const event_id_to_user = try DB.HashMap(.read_write).init(event_id_to_user_cursor); 425 426 if (event.data) |data| { 427 const user_cursor = try event_id_to_user.putCursor(user_key); 428 const user_map = try DB.HashMap(.read_write).init(user_cursor); 429 try upsert(DB, repo_opts.hash, user_map, EventData.User, data.user); 430 } else { 431 if (!try event_id_to_user.remove(user_key)) return error.EventNotFound; 432 433 const user_id_to_repos_cursor = try haxy_moment.putCursor(hash.hashInt(repo_opts.hash, "user-id->repos")); 434 const user_id_to_repos = try DB.HashMap(.read_write).init(user_id_to_repos_cursor); 435 _ = try user_id_to_repos.remove(user_key); 436 } 437 }, 438 .repo => { 439 const repo_key = hash.hashInt(repo_opts.hash, &current_event_id); 440 441 const event_id_to_repo_cursor = try haxy_moment.putCursor(hash.hashInt(repo_opts.hash, "event-id->repo")); 442 const event_id_to_repo = try DB.HashMap(.read_write).init(event_id_to_repo_cursor); 443 444 if (event.data) |data| { 445 const repo_cursor = try event_id_to_repo.putCursor(repo_key); 446 const repo_map = try DB.HashMap(.read_write).init(repo_cursor); 447 try upsert(DB, repo_opts.hash, repo_map, EventData.Repo, data.repo); 448 449 const user_id_to_repos_cursor = try haxy_moment.putCursor(hash.hashInt(repo_opts.hash, "user-id->repos")); 450 const user_id_to_repos = try DB.HashMap(.read_write).init(user_id_to_repos_cursor); 451 452 const user_repos_cursor = try user_id_to_repos.putCursor(hash.hashInt(repo_opts.hash, data.repo.user_id)); 453 const user_repos = try DB.CountedHashSet(.read_write).init(user_repos_cursor); 454 try user_repos.put(repo_key, .{ .bytes = &current_event_id }); 455 } else { 456 if (try event_id_to_repo.getCursor(repo_key)) |existing_repo_cursor| { 457 const existing_repo_map = try DB.HashMap(.read_only).init(existing_repo_cursor); 458 const existing_repo = try EventData.read(DB, repo_opts.hash, arena.allocator(), existing_repo_map, .repo); 459 const data = existing_repo.repo; 460 461 const user_id_to_repos_cursor = try haxy_moment.putCursor(hash.hashInt(repo_opts.hash, "user-id->repos")); 462 const user_id_to_repos = try DB.HashMap(.read_write).init(user_id_to_repos_cursor); 463 464 const user_key = hash.hashInt(repo_opts.hash, data.user_id); 465 const user_repos_cursor = try user_id_to_repos.putCursor(user_key); 466 const user_repos = try DB.CountedHashSet(.read_write).init(user_repos_cursor); 467 _ = try user_repos.remove(repo_key); 468 } 469 if (!try event_id_to_repo.remove(repo_key)) return error.EventNotFound; 470 } 471 }, 472 .issue => { 473 const issue_key = hash.hashInt(repo_opts.hash, &current_event_id); 474 475 const event_id_to_issue_cursor = try haxy_moment.putCursor(hash.hashInt(repo_opts.hash, "event-id->issue")); 476 const event_id_to_issue = try DB.HashMap(.read_write).init(event_id_to_issue_cursor); 477 478 if (event.data) |data| { 479 const issue_cursor = try event_id_to_issue.putCursor(issue_key); 480 const issue = try DB.HashMap(.read_write).init(issue_cursor); 481 try upsert(DB, repo_opts.hash, issue, EventData.Issue, data.issue); 482 } else { 483 if (!try event_id_to_issue.remove(issue_key)) return error.EventNotFound; 484 } 485 }, 486 } 487 } 488 489 // the current object id is now the last one 490 last_object_id_maybe = repo_event.oid; 491 492 // prevent any of the data created above from being mutated by future iterations of this loop 493 try state.core.db.freeze(); 494 } 495 496 if (last_object_id_maybe) |*last_object_id| { 497 try state.extra.moment.put(hash.hashInt(repo_opts.hash, "haxy-last-object-id"), .{ .bytes = last_object_id }); 498 } 499 } 500 501 fn upsert( 502 comptime DB: type, 503 comptime hash_kind: hash.HashKind, 504 map: DB.HashMap(.read_write), 505 comptime Data: type, 506 data: Data, 507 ) !void { 508 switch (@typeInfo(Data)) { 509 .@"struct" => |struct_info| { 510 inline for (struct_info.fields) |field| { 511 try upsertField(DB, hash_kind, map, field.name, field.type, @field(data, field.name)); 512 } 513 }, 514 else => @compileError("upsert expects a struct"), 515 } 516 } 517 518 fn mergeChangedMapEntries( 519 comptime DB: type, 520 target: DB.HashMap(.read_write), 521 parent: DB.HashMap(.read_only), 522 baseline: DB.HashMap(.read_only), 523 comptime is_top_level: bool, 524 ) !void { 525 var parent_iter = try parent.iterator(); 526 while (try parent_iter.next()) |kv_pair_cursor| { 527 const kv_pair = try kv_pair_cursor.readKeyValuePair(); 528 529 const baseline_value_cursor = try baseline.getCursor(kv_pair.hash) orelse { 530 if ((try target.getCursor(kv_pair.hash)) != null) { 531 return error.MergeConflict; 532 } 533 534 try target.put(kv_pair.hash, .{ .slot = kv_pair.value_cursor.slot() }); 535 continue; 536 }; 537 538 const tag = kv_pair.value_cursor.slot().tag; 539 540 if (tag != baseline_value_cursor.slot().tag) { 541 return error.UnexpectedTag; 542 } 543 544 if (kv_pair.value_cursor.slot().value == baseline_value_cursor.slot().value) { 545 continue; 546 } 547 548 if (tag == .hash_map) { 549 const target_existing_cursor = try target.getCursor(kv_pair.hash) orelse return error.MergeConflict; 550 if (target_existing_cursor.slot().tag != .hash_map) { 551 return error.MergeConflict; 552 } 553 554 const target_child_cursor = try target.putCursor(kv_pair.hash); 555 const target_child = try DB.HashMap(.read_write).init(target_child_cursor); 556 const parent_child = try DB.HashMap(.read_only).init(kv_pair.value_cursor); 557 const baseline_child = try DB.HashMap(.read_only).init(baseline_value_cursor); 558 try mergeChangedMapEntries(DB, target_child, parent_child, baseline_child, false); 559 } else if (is_top_level) { 560 // the moment-index is a top-level key in the haxy moment 561 // and is not a map. we don't want to do any merge of this. 562 continue; 563 } else { 564 const target_value_cursor = try target.getCursor(kv_pair.hash) orelse return error.MergeConflict; 565 if (target_value_cursor.slot().value != baseline_value_cursor.slot().value) { 566 return error.MergeConflict; 567 } 568 569 try target.put(kv_pair.hash, .{ .slot = kv_pair.value_cursor.slot() }); 570 } 571 } 572 } 573 574 fn upsertField( 575 comptime DB: type, 576 comptime hash_kind: hash.HashKind, 577 map: DB.HashMap(.read_write), 578 comptime field_name: []const u8, 579 comptime Field: type, 580 value: Field, 581 ) !void { 582 const key = hash.hashInt(hash_kind, field_name); 583 584 switch (@typeInfo(Field)) { 585 .pointer => |pointer_info| { 586 if (pointer_info.size == .slice and pointer_info.child == u8) { 587 try upsertBytes(DB, hash_kind, map, key, value); 588 } else { 589 @compileError("unsupported upsert field type: " ++ @typeName(Field)); 590 } 591 }, 592 .array => |array_info| { 593 if (array_info.child == u8) { 594 try upsertBytes(DB, hash_kind, map, key, &value); 595 } else { 596 @compileError("unsupported upsert field type: " ++ @typeName(Field)); 597 } 598 }, 599 .int => |int_info| switch (int_info.signedness) { 600 .unsigned => { 601 if (try map.getCursor(key)) |value_cursor| { 602 if (try value_cursor.readUint() == value) { 603 return; 604 } 605 } 606 607 try map.put(key, .{ .uint = value }); 608 }, 609 .signed => { 610 if (try map.getCursor(key)) |value_cursor| { 611 if (try value_cursor.readInt() == value) { 612 return; 613 } 614 } 615 616 try map.put(key, .{ .int = value }); 617 }, 618 }, 619 .bool => { 620 const bytes = if (value) "true" else "false"; 621 if (try map.getCursor(key)) |value_cursor| { 622 var buffer: [5]u8 = undefined; 623 const existing = try value_cursor.readBytesObject(&buffer); 624 if (existing.format_tag != null and 625 std.mem.eql(u8, &existing.format_tag.?, "bl") and 626 std.mem.eql(u8, existing.value, bytes)) 627 { 628 return; 629 } 630 } 631 632 try map.put(key, .{ .bytes_object = .{ .value = bytes, .format_tag = "bl".* } }); 633 }, 634 else => @compileError("unsupported upsert field type: " ++ @typeName(Field)), 635 } 636 } 637 638 fn upsertBytes( 639 comptime DB: type, 640 comptime hash_kind: hash.HashKind, 641 map: DB.HashMap(.read_write), 642 key: hash.HashInt(hash_kind), 643 value: []const u8, 644 ) !void { 645 var existing_cursor_maybe = try map.getCursor(key); 646 if (existing_cursor_maybe) |*existing_cursor| { 647 if (try bytesEqual(DB, existing_cursor, value)) { 648 return; 649 } 650 } 651 652 var value_cursor = try map.putCursor(key); 653 var write_buffer: [1024]u8 = undefined; 654 var writer = try value_cursor.writer(&write_buffer); 655 try writer.interface.writeAll(value); 656 try writer.finish(); 657 } 658 659 fn bytesEqual( 660 comptime DB: type, 661 cursor: *DB.Cursor(.read_only), 662 value: []const u8, 663 ) !bool { 664 var read_buffer: [1024]u8 = undefined; 665 var reader = try cursor.reader(&read_buffer); 666 if (reader.size != value.len) { 667 return false; 668 } 669 670 var chunk_buffer: [1024]u8 = undefined; 671 var offset: usize = 0; 672 while (offset < value.len) { 673 const chunk_len = @min(chunk_buffer.len, value.len - offset); 674 try reader.interface.readSliceAll(chunk_buffer[0..chunk_len]); 675 if (!std.mem.eql(u8, chunk_buffer[0..chunk_len], value[offset .. offset + chunk_len])) { 676 return false; 677 } 678 offset += chunk_len; 679 } 680 681 return true; 682 } 683