216 // the very first event. all we need to do is set the haxy list
217 // to be empty and make a new haxy_moments map to work with.
218
219 try haxy.slice(0);
220 haxy_moments_cursor = try haxy.appendCursor();
221 haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor);
222
223 last_object_id_maybe = null;
224 },
225 1 => {
226 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined;
227 _ = try std.fmt.hexToBytes(&oid, &parent_oids[0]);
228
229 const old_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &oid)) orelse return error.CursorNotFound;
230 const old_moment = try DB.HashMap(.read_only).init(old_moment_cursor);
231
232 const old_moment_index_cursor = try old_moment.getCursor(hash.hashInt(repo_opts.hash, "moment-index")) orelse return error.CursorNotFound;
233 const old_moment_index = try old_moment_index_cursor.readUint();
234
235 // resize the haxy list so we truncate all the moments that were
236 // created after the parent_oid was consumed
237 try haxy.slice(old_moment_index + 1);
238
239 // make a new haxy moment and set its initial value to the last haxy moment
240 haxy_moments_cursor = try haxy.appendCursor();
241 const old_haxy_moments_cursor = try haxy.getCursor(old_moment_index) orelse return error.CursorNotFound;
242 try haxy_moments_cursor.write(.{ .slot = old_haxy_moments_cursor.slot() });
243 haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor);
244
245 last_object_id_maybe = oid;
246 },
247 else => return error.UnexpectedParentCount,
248 }
249 }
250 }
251
252 while (try commit_iter.next()) |repo_event_oid| {
253 var commit_object = try obj.Object(.xit, repo_opts).init(state.readOnly(), io, allocator, &std.fmt.bytesToHex(repo_event_oid, .lower));
254 defer commit_object.deinit();
255
256 const parent_oids = commit_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound;
257
258 // create a moment for this object id
259 var haxy_moment_cursor = try haxy_moments.putCursor(hash.bytesToInt(repo_opts.hash, &repo_event_oid));
260
261 // if there was a previous object id, set this haxy moment's initial value to it.
262 // this efficiently "clones" the map so we make further modifications based on it.
263 if (parent_oids.len > 0) {
264 var first_parent_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined;
265 _ = try std.fmt.hexToBytes(&first_parent_oid, &parent_oids[0]);
266
267 const first_parent_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &first_parent_oid)) orelse return error.CursorNotFound;
268 try haxy_moment_cursor.write(.{ .slot = first_parent_haxy_moment_cursor.slot() });
269 }
270
271 const haxy_moment = try DB.HashMap(.read_write).init(haxy_moment_cursor);
272
273 // merge changes from every parent after the first parent. the common
274 // ancestor is the baseline, so a later parent only contributes values
275 // it changed relative to the merge base.
276 if (parent_oids.len > 1) {
277 for (parent_oids[1..]) |*parent_oid| {
278 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined;
279 _ = try std.fmt.hexToBytes(&oid, parent_oid);
280
281 const parent_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &oid)) orelse return error.CursorNotFound;
282 const parent_haxy_moment = try DB.HashMap(.read_only).init(parent_haxy_moment_cursor);
283
284 const baseline_oid_hex = try mrg.commonAncestor(.xit, repo_opts, state.readOnly(), io, allocator, &parent_oids[0], parent_oid);
285 var baseline_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined;
286 _ = try std.fmt.hexToBytes(&baseline_oid, &baseline_oid_hex);
287 const baseline_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &baseline_oid)) orelse return error.CursorNotFound;
288 const baseline_haxy_moment = try DB.HashMap(.read_only).init(baseline_haxy_moment_cursor);
289
290 try mergeChangedMapEntries(DB, allocator, haxy_moment, parent_haxy_moment, baseline_haxy_moment, true);
291 }
292 }
293
294 // associate this moment with the index it will first appear at in the haxy list.
295 // this will be important later so we can truncate that list if the user ever
296 // rebases starting at this object id.
297 try haxy_moment.put(hash.hashInt(repo_opts.hash, "moment-index"), .{ .uint = try haxy.count() - 1 });
298
299 // consume the event unless it's a merge commit
300 if (parent_oids.len <= 1) {
301 var arena = std.heap.ArenaAllocator.init(allocator);
302 defer arena.deinit();
303
304 // read the message from the commit
305 try commit_object.object_reader.seekTo(commit_object.content.commit.message_position);
306 const message = try commit_object.object_reader.interface.allocRemaining(arena.allocator(), .unlimited);
307
308 const event_with_id = try EventWithId.fromString(&arena, message);
309
310 // get the id of the current event as bytes
311 var current_event_id: [event_id_size]u8 = undefined;
312 _ = try std.fmt.hexToBytes(¤t_event_id, &event_with_id.id);
313
314 const created_ts = commit_object.content.commit.metadata.timestamp;
315
316 switch (event_with_id.event) {
317 .user => |event_maybe| try User.consume(DB, repo_opts.hash, haxy_moment, ¤t_event_id, event_maybe, &arena, created_ts),
318 .repo => |event_maybe| try Repo.consume(DB, repo_opts.hash, haxy_moment, ¤t_event_id, event_maybe, &arena, created_ts),
319 .issue => |event_maybe| try Issue.consume(DB, repo_opts.hash, haxy_moment, ¤t_event_id, event_maybe, &arena, created_ts),
320 }
321 }
322
323 // the current object id is now the last one
324 last_object_id_maybe = repo_event_oid;
325
326 // prevent any of the data created above from being mutated by future iterations of this loop
327 try state.core.db.freeze();
328 }
329
330 if (last_object_id_maybe) |*last_object_id| {
331 try state.extra.moment.put(hash.hashInt(repo_opts.hash, "haxy-last-object-id"), .{ .bytes = last_object_id });
332 }
333 }
334
335 pub fn currentMoment(
336 comptime repo_opts: rp.RepoOpts(.xit),
337 repo: *rp.Repo(.xit, repo_opts),
338 ) !rp.Repo(.xit, repo_opts).DB.HashMap(.read_only) {
339 const DB = rp.Repo(.xit, repo_opts).DB;
340 const history = try DB.ArrayList(.read_only).init(repo.core.db.rootCursor().readOnly());
341 const moment_cursor = try history.getCursor(-1) orelse return error.NotFound;
342 const moment = try DB.HashMap(.read_only).init(moment_cursor);
343 const last_object_id_cursor = try moment.getCursor(hash.hashInt(repo_opts.hash, "haxy-last-object-id")) orelse return error.NotFound;
344 var last_object_id: [hash.byteLen(repo_opts.hash)]u8 = undefined;
345 _ = try last_object_id_cursor.readBytes(&last_object_id);
346 const haxy_cursor = try moment.getCursor(hash.hashInt(repo_opts.hash, "haxy")) orelse return error.NotFound;
347 const haxy = try DB.ArrayList(.read_only).init(haxy_cursor);
348 const haxy_moments_cursor = try haxy.getCursor(-1) orelse return error.NotFound;
349 const haxy_moments = try DB.HashMap(.read_only).init(haxy_moments_cursor);
350 const haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &last_object_id)) orelse return error.NotFound;
351 return try DB.HashMap(.read_only).init(haxy_moment_cursor);
352 }
353
354 pub fn commitAndConsume(
355 comptime repo_opts: rp.RepoOpts(.xit),
356 io: std.Io,
357 allocator: std.mem.Allocator,
358 repo: *rp.Repo(.xit, repo_opts),
359 ref: rf.Ref,
360 events: []const EventWithId,
361 ) !void {
362 var json: std.Io.Writer.Allocating = .init(allocator);
363 defer json.deinit();
364
365 for (events) |event| {
366 json.clearRetainingCapacity();
367 try std.json.Stringify.value(event, .{}, &json.writer);
368 _ = try repo.commitAtRef(io, allocator, .{ .message = json.written(), .timestamp = event.timestamp }, null, ref);
369 }
370
371 try consume(repo_opts, io, allocator, repo, ref);
372 }
373
374 // build the key for SortedSets sorted by timestamp. the big-endian timestamp
375 // makes byte order match creation order; the event id breaks ties and keeps
376 // keys unique within the same timestamp.
377 pub fn orderKey(timestamp: u64, event_id: *const [event_id_size]u8) [@sizeOf(u64) + event_id_size]u8 {
378 var key: [@sizeOf(u64) + event_id_size]u8 = undefined;
379 std.mem.writeInt(u64, key[0..@sizeOf(u64)], timestamp, .big);
380 @memcpy(key[@sizeOf(u64)..], event_id);
381 return key;
382 }
383
384 // split a pushed "<owner>/<repo>" path into its two components, or null if it
385 // isn't exactly two non-empty segments.
386 pub fn parseOwnerRepoPath(path: []const u8) ?struct { owner: []const u8, name: []const u8 } {
387 var it = std.mem.splitScalar(u8, path, '/');
388 const owner = it.next() orelse return null;
389 const name = it.next() orelse return null;
390 if (it.next() != null) return null;
391 if (owner.len == 0 or name.len == 0) return null;
392
393 return .{ .owner = owner, .name = name };
394 }
395
396 // resolve a pushed `<owner>/<repo>` to the hex event id that names its on-disk
397 // directory under the repos dir
398 pub fn resolveOrCreateRepo(
399 io: std.Io,
400 allocator: std.mem.Allocator,
401 admin_repo_path: []const u8,
402 owner_name: []const u8,
403 repo_name: []const u8,
404 create_if_missing: bool,
405 ) !?[event_id_size * 2]u8 {
406 var repo = rp.Repo(.xit, admin_repo_opts).open(io, allocator, .{ .path = admin_repo_path }) catch |err| switch (err) {
407 error.RepoNotFound => return null,
408 else => |e| return e,
409 };
410 defer repo.deinit(io, allocator);
411
412 var arena = std.heap.ArenaAllocator.init(allocator);
413 defer arena.deinit();
414
415 const moment = try currentMoment(admin_repo_opts, &repo);
416
417 // an already-registered repo reuses its event id, so a re-push (or a clone)
418 // lands in the same repo
419 if (try Repo.readByOwnerAndName(AdminDB, admin_repo_opts.hash, moment, &arena, owner_name, repo_name)) |found| {
420 return std.fmt.bytesToHex(found.event_id, .lower);
421 }
422
423 if (!create_if_missing) return null;
424
425 // the new repo is owned by the named user, read from the name index; an
426 // unknown owner can't own a repo, so the push is rejected
427 const name_to_user_id_cursor = try moment.getCursor(hash.hashInt(admin_repo_opts.hash, "name->user-id")) orelse return null;
428 const name_to_user_id = try AdminDB.HashMap(.read_only).init(name_to_user_id_cursor);
429 const owner_id_cursor = try name_to_user_id.getCursor(hash.hashInt(admin_repo_opts.hash, owner_name)) orelse return null;
430 var owner_user_id: [event_id_size]u8 = undefined;
431 _ = try owner_id_cursor.readBytes(&owner_user_id);
432
433 var id_bytes: [event_id_size]u8 = undefined;
434 io.random(&id_bytes);
435 const event_id_hex = std.fmt.bytesToHex(id_bytes, .lower);
436
437 try commitAndConsume(admin_repo_opts, io, allocator, &repo, events_ref, &[_]EventWithId{.{
438 .id = event_id_hex,
439 .timestamp = @intCast(std.Io.Timestamp.now(io, .real).toSeconds()),
440 .event = .{ .repo = .{
441 .user_id = &owner_user_id,
442 .name = repo_name,
443 .description = "",
444 .enable_issue = true,
445 } },
446 }});
447
448 return event_id_hex;
449 }
450
451 //
452 // reading from xitdb
453 //
454
455 pub fn read(
456 comptime T: type,
457 comptime DB: type,
458 comptime hash_kind: hash.HashKind,
459 arena: *std.heap.ArenaAllocator,
460 map: DB.HashMap(.read_only),
461 ) !T {
462 var event: T = undefined;
463
464 switch (@typeInfo(T)) {
465 .@"struct" => |struct_info| {
466 inline for (struct_info.fields) |field| {
467 switch (@typeInfo(field.type)) {
468 .pointer => |pointer_info| {
469 if (pointer_info.size == .slice and pointer_info.child == u8 and pointer_info.is_const) {
470 @field(event, field.name) = try readBytes(DB, hash_kind, arena.allocator(), map, field.name);
471 } else {
472 @compileError("unsupported read field type: " ++ @typeName(field.type));
473 }
474 },
475 .array => |array_info| {
476 if (array_info.child == u8) {
477 const bytes = try readBytes(DB, hash_kind, arena.allocator(), map, field.name);
478 if (bytes.len != array_info.len) return error.InvalidByteArrayLength;
479 @memcpy(@field(event, field.name)[0..], bytes);
480 } else {
481 @compileError("unsupported read field type: " ++ @typeName(field.type));
482 }
483 },
484 .bool => @field(event, field.name) = try readBool(DB, hash_kind, map, field.name),
485 .int => |int_info| switch (int_info.signedness) {
486 .unsigned => @field(event, field.name) = @intCast(try readUint(DB, hash_kind, map, field.name)),
487 .signed => @field(event, field.name) = @intCast(try readInt(DB, hash_kind, map, field.name)),
488 },
489 else => @compileError("unsupported read field type: " ++ @typeName(field.type)),
490 }
491 }
492 },
493 else => @compileError("read expects a struct"),
494 }
495
496 return event;
497 }
498
499 fn readBytes(
500 comptime DB: type,
501 comptime hash_kind: hash.HashKind,
502 allocator: std.mem.Allocator,
503 map: DB.HashMap(.read_only),
504 field_name: []const u8,
505 ) ![]const u8 {
506 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound;
507 return try cursor.readBytesAlloc(allocator, null);
508 }
509
510 fn readUint(
511 comptime DB: type,
512 comptime hash_kind: hash.HashKind,
513 map: DB.HashMap(.read_only),
514 field_name: []const u8,
515 ) !u64 {
516 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound;
517 return try cursor.readUint();
518 }
519
520 fn readInt(
521 comptime DB: type,
522 comptime hash_kind: hash.HashKind,
523 map: DB.HashMap(.read_only),
524 field_name: []const u8,
525 ) !i64 {
526 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound;
527 return try cursor.readInt();
528 }
529
530 fn readBool(
531 comptime DB: type,
532 comptime hash_kind: hash.HashKind,
533 map: DB.HashMap(.read_only),
534 field_name: []const u8,
535 ) !bool {
536 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound;
537 var buffer: [5]u8 = undefined;
538 const bytes = try cursor.readBytesObject(&buffer);
539 const format_tag = bytes.format_tag orelse return error.InvalidFormatTag;
540 if (!std.mem.eql(u8, &format_tag, "bl")) return error.InvalidFormatTag;
541 if (std.mem.eql(u8, bytes.value, "true")) return true;
542 if (std.mem.eql(u8, bytes.value, "false")) return false;
543 return error.InvalidBool;
544 }
545
546 //
547 // writing to xitdb
548 //
549
550 pub fn upsert(
551 comptime T: type,
552 comptime DB: type,
553 comptime hash_kind: hash.HashKind,
554 map: DB.HashMap(.read_write),
555 event: T,
556 ) !void {
557 switch (@typeInfo(T)) {
558 .@"struct" => |struct_info| {
559 inline for (struct_info.fields) |field| {
560 try upsertField(DB, hash_kind, map, field.name, field.type, @field(event, field.name));
561 }
562 },
563 else => @compileError("upsert expects a struct"),
564 }
565 }
566
567 fn mergeChangedMapEntries(
568 comptime DB: type,
569 allocator: std.mem.Allocator,
570 target: DB.HashMap(.read_write),
571 parent: DB.HashMap(.read_only),
572 baseline: DB.HashMap(.read_only),
573 comptime is_top_level: bool,
574 ) !void {
575 var parent_iter = try parent.iterator();
576 while (try parent_iter.next()) |kv_pair_cursor| {
577 const kv_pair = try kv_pair_cursor.readKeyValuePair();
578
579 const baseline_value_cursor = try baseline.getCursor(kv_pair.hash) orelse {
580 if ((try target.getCursor(kv_pair.hash)) != null) {
581 return error.MergeConflict;
582 }
583
584 try target.put(kv_pair.hash, .{ .slot = kv_pair.value_cursor.slot() });
585 continue;
586 };
587
588 const tag = kv_pair.value_cursor.slot().tag;
589
590 if (tag != baseline_value_cursor.slot().tag) {
591 return error.UnexpectedTag;
592 }
593
594 if (kv_pair.value_cursor.slot().value == baseline_value_cursor.slot().value) {
595 continue;
596 }
597
598 if (tag == .hash_map) {
599 const target_existing_cursor = try target.getCursor(kv_pair.hash) orelse return error.MergeConflict;
600 if (target_existing_cursor.slot().tag != .hash_map) {
601 return error.UnexpectedTag;
602 }
603
604 const target_child_cursor = try target.putCursor(kv_pair.hash);
605 const target_child = try DB.HashMap(.read_write).init(target_child_cursor);
606 const parent_child = try DB.HashMap(.read_only).init(kv_pair.value_cursor);
607 const baseline_child = try DB.HashMap(.read_only).init(baseline_value_cursor);
608 try mergeChangedMapEntries(DB, allocator, target_child, parent_child, baseline_child, false);
609 } else if (tag == .sorted_set) {
610 const target_existing_cursor = try target.getCursor(kv_pair.hash) orelse return error.MergeConflict;
611 if (target_existing_cursor.slot().tag != .sorted_set) {
612 return error.UnexpectedTag;
613 }
614
615 const target_child_cursor = try target.putCursor(kv_pair.hash);
616 const target_child = try DB.SortedSet(.read_write).init(target_child_cursor);
617 const parent_child = try DB.SortedSet(.read_only).init(kv_pair.value_cursor);
618 const baseline_child = try DB.SortedSet(.read_only).init(baseline_value_cursor);
619 try mergeChangedSetEntries(DB, allocator, target_child, parent_child, baseline_child);
620 } else if (tag == .sorted_map) {
621 const target_existing_cursor = try target.getCursor(kv_pair.hash) orelse return error.MergeConflict;
622 if (target_existing_cursor.slot().tag != .sorted_map) {
623 return error.UnexpectedTag;
624 }
625
626 const target_child_cursor = try target.putCursor(kv_pair.hash);
627 const target_child = try DB.SortedMap(.read_write).init(target_child_cursor);
628 const parent_child = try DB.SortedMap(.read_only).init(kv_pair.value_cursor);
629 const baseline_child = try DB.SortedMap(.read_only).init(baseline_value_cursor);
630 try mergeChangedSortedMapEntries(DB, allocator, target_child, parent_child, baseline_child);
631 } else if (is_top_level) {
632 // the moment-index is a top-level key in the haxy moment
633 // and is not a map. we don't want to do any merge of this.
634 continue;
635 } else {
636 const target_value_cursor = try target.getCursor(kv_pair.hash) orelse return error.MergeConflict;
637 if (target_value_cursor.slot().tag != baseline_value_cursor.slot().tag) {
638 return error.UnexpectedTag;
639 }
640 if (target_value_cursor.slot().value != baseline_value_cursor.slot().value) {
641 return error.MergeConflict;
642 }
643
644 try target.put(kv_pair.hash, .{ .slot = kv_pair.value_cursor.slot() });
645 }
646 }
647 }
648
649 // three-way merge of a sorted map keyed by byte strings, like
650 // mergeChangedMapEntries but recursing into sorted-set values.
651 fn mergeChangedSortedMapEntries(
652 comptime DB: type,
653 allocator: std.mem.Allocator,
654 target: DB.SortedMap(.read_write),
655 parent: DB.SortedMap(.read_only),
656 baseline: DB.SortedMap(.read_only),
657 ) !void {
658 var parent_iter = try parent.iterator();
659 while (try parent_iter.next()) |kv_pair_cursor| {
660 const kv_pair = try kv_pair_cursor.readKeyValuePair();
661 const key = try kv_pair.key_cursor.readBytesAlloc(allocator, null);
662 defer allocator.free(key);
663
664 const baseline_value_cursor = try baseline.getCursor(key) orelse {
665 if (try target.getCursor(key)) |target_existing_cursor| {
666 // both parents created this key. sets union (they hold
667 // independently added members); anything else conflicts.
668 const parent_tag = kv_pair.value_cursor.slot().tag;
669 if (target_existing_cursor.slot().tag != parent_tag) {
670 return error.UnexpectedTag;
671 }
672 if (parent_tag != .sorted_set) {
673 return error.MergeConflict;
674 }
675
676 const target_child_cursor = try target.putCursor(key);
677 const target_child = try DB.SortedSet(.read_write).init(target_child_cursor);
678 const parent_child = try DB.SortedSet(.read_only).init(kv_pair.value_cursor);
679 var set_iter = try parent_child.iterator();
680 while (try set_iter.next()) |set_kv_cursor| {
681 const set_kv = try set_kv_cursor.readKeyValuePair();
682 const set_key = try set_kv.key_cursor.readBytesAlloc(allocator, null);
683 defer allocator.free(set_key);
684 try target_child.put(set_key);
685 }
686 continue;
687 }
688
689 try target.put(key, .{ .slot = kv_pair.value_cursor.slot() });
690 continue;
691 };
692
693 const tag = kv_pair.value_cursor.slot().tag;
694
695 if (tag != baseline_value_cursor.slot().tag) {
696 return error.UnexpectedTag;
697 }
698
699 if (kv_pair.value_cursor.slot().value == baseline_value_cursor.slot().value) {
700 continue;
701 }
702
703 if (tag == .sorted_set) {
704 const target_existing_cursor = try target.getCursor(key) orelse return error.MergeConflict;
705 if (target_existing_cursor.slot().tag != .sorted_set) {
706 return error.UnexpectedTag;
707 }
708
709 const target_child_cursor = try target.putCursor(key);
710 const target_child = try DB.SortedSet(.read_write).init(target_child_cursor);
711 const parent_child = try DB.SortedSet(.read_only).init(kv_pair.value_cursor);
712 const baseline_child = try DB.SortedSet(.read_only).init(baseline_value_cursor);
713 try mergeChangedSetEntries(DB, allocator, target_child, parent_child, baseline_child);
714 } else {
715 const target_value_cursor = try target.getCursor(key) orelse return error.MergeConflict;
716 if (target_value_cursor.slot().tag != baseline_value_cursor.slot().tag) {
717 return error.UnexpectedTag;
718 }
719 if (target_value_cursor.slot().value != baseline_value_cursor.slot().value) {
720 return error.MergeConflict;
721 }
722
723 try target.put(key, .{ .slot = kv_pair.value_cursor.slot() });
724 }
725 }
726 }
727
728 // three-way merge of a sorted set: apply the parent's changes relative to the
729 // baseline. keys have no values, so additions and removals can't conflict.
730 fn mergeChangedSetEntries(
731 comptime DB: type,
732 allocator: std.mem.Allocator,
733 target: DB.SortedSet(.read_write),
734 parent: DB.SortedSet(.read_only),
735 baseline: DB.SortedSet(.read_only),
736 ) !void {
737 var parent_iter = try parent.iterator();
738 while (try parent_iter.next()) |kv_pair_cursor| {
739 const kv_pair = try kv_pair_cursor.readKeyValuePair();
740 const key = try kv_pair.key_cursor.readBytesAlloc(allocator, null);
741 defer allocator.free(key);
742 if (!try baseline.contains(key)) try target.put(key);
743 }
744
745 var baseline_iter = try baseline.iterator();
746 while (try baseline_iter.next()) |kv_pair_cursor| {
747 const kv_pair = try kv_pair_cursor.readKeyValuePair();
748 const key = try kv_pair.key_cursor.readBytesAlloc(allocator, null);
749 defer allocator.free(key);
750 if (!try parent.contains(key)) _ = try target.remove(key);
751 }
752 }
753
754 fn upsertField(
755 comptime DB: type,
756 comptime hash_kind: hash.HashKind,
757 map: DB.HashMap(.read_write),
758 comptime field_name: []const u8,
759 comptime Field: type,
760 value: Field,
761 ) !void {
762 const key = hash.hashInt(hash_kind, field_name);
763
764 switch (@typeInfo(Field)) {
765 .pointer => |pointer_info| {
766 if (pointer_info.size == .slice and pointer_info.child == u8) {
767 try upsertBytes(DB, hash_kind, map, key, value);
768 } else {
769 @compileError("unsupported upsert field type: " ++ @typeName(Field));
770 }
771 },
772 .array => |array_info| {
773 if (array_info.child == u8) {
774 try upsertBytes(DB, hash_kind, map, key, &value);
775 } else {
776 @compileError("unsupported upsert field type: " ++ @typeName(Field));
777 }
778 },
779 .int => |int_info| switch (int_info.signedness) {
780 .unsigned => {
781 if (try map.getCursor(key)) |value_cursor| {
782 if (try value_cursor.readUint() == value) {
783 return;
784 }
785 }
786
787 try map.put(key, .{ .uint = value });
788 },
789 .signed => {
790 if (try map.getCursor(key)) |value_cursor| {
791 if (try value_cursor.readInt() == value) {
792 return;
793 }
794 }
795
796 try map.put(key, .{ .int = value });
797 },
798 },
799 .bool => {
800 const bytes = if (value) "true" else "false";
801 if (try map.getCursor(key)) |value_cursor| {
802 var buffer: [5]u8 = undefined;
803 const existing = try value_cursor.readBytesObject(&buffer);
804 if (existing.format_tag) |format_tag| {
805 if (std.mem.eql(u8, &format_tag, "bl") and std.mem.eql(u8, existing.value, bytes)) {
806 return;
807 }
808 }
809 }
810
811 try map.put(key, .{ .bytes_object = .{ .value = bytes, .format_tag = "bl".* } });
812 },
813 else => @compileError("unsupported upsert field type: " ++ @typeName(Field)),
814 }
815 }
816
817 fn upsertBytes(
818 comptime DB: type,
819 comptime hash_kind: hash.HashKind,
820 map: DB.HashMap(.read_write),
821 key: hash.HashInt(hash_kind),
822 value: []const u8,
823 ) !void {
824 var existing_cursor_maybe = try map.getCursor(key);
825 if (existing_cursor_maybe) |*existing_cursor| {
826 if (try bytesEqual(DB, existing_cursor, value)) {
827 return;
828 }
829 }
830
831 var value_cursor = try map.putCursor(key);
832 var write_buffer: [1024]u8 = undefined;
833 var writer = try value_cursor.writer(&write_buffer);
834 try writer.interface.writeAll(value);
835 try writer.finish();
836 }
837
838 fn bytesEqual(
839 comptime DB: type,
840 cursor: *DB.Cursor(.read_only),
841 value: []const u8,
842 ) !bool {
843 var read_buffer: [1024]u8 = undefined;
844 var reader = try cursor.reader(&read_buffer);
845 if (reader.size != value.len) {
846 return false;
847 }
848
849 var chunk_buffer: [1024]u8 = undefined;
850 var offset: usize = 0;
851 while (offset < value.len) {
852 const chunk_len = @min(chunk_buffer.len, value.len - offset);
853 try reader.interface.readSliceAll(chunk_buffer[0..chunk_len]);
854 if (!std.mem.eql(u8, chunk_buffer[0..chunk_len], value[offset .. offset + chunk_len])) {
855 return false;
856 }
857 offset += chunk_len;
858 }
859
860 return true;
861 }
862
863 //
864 // CommitIterator
865 //
866 // differs from xit's ObjectIterator in a few ways:
867 //
868 // 1. iterates parents first, rather than children first
869 // 2. `next` only returns the object id, not the full Object
870 // 3. stores its temporary state in a file on disk to avoid OOMs
871 //
872
873 pub fn CommitIterator(comptime repo_opts: rp.RepoOpts(.xit)) type {
874 return struct {
875 const DB = rp.Repo(.xit, repo_opts).DB;
876 const db_name = "haxy-repo-events.db";
877
878 repo_dir: std.Io.Dir,
879 db_file: std.Io.File,
880 db: *DB,
881 // map of each object id to its children
882 parent_to_children: DB.HashMap(.read_write),
883 // tracking pending parents is necessary because a child can only
884 // be returned after all of its unconsumed parents have been returned
885 child_to_pending_parents: DB.HashMap(.read_write),
886 // queue of object ids that are ready to be returned from `next`
887 ready_queue: DB.ArrayList(.read_write),
888 ready_queue_index: u64,
889 // the object id at the tip of this branch
890 newest_object_id: ?[hash.byteLen(repo_opts.hash)]u8,
891 // the object id furthest back in the history that hasn't been consumed
892 oldest_object_id: ?[hash.byteLen(repo_opts.hash)]u8,
893
894 pub fn init(
895 state: rp.Repo(.xit, repo_opts).State(.read_only),
896 io: std.Io,
897 allocator: std.mem.Allocator,
898 consumed_object_ids: DB.HashMap(.read_only),
899 ref: rf.Ref,
900 ) !CommitIterator(repo_opts) {
901 const db_file = try state.core.repo_dir.createFile(io, db_name, .{ .truncate = true, .lock = .exclusive, .read = true });
902 errdefer {
903 db_file.close(io);
904 state.core.repo_dir.deleteFile(io, db_name) catch {};
905 }
906
907 const buffer_ptr = try allocator.create(std.Io.Writer.Allocating);
908 errdefer allocator.destroy(buffer_ptr);
909
910 buffer_ptr.* = std.Io.Writer.Allocating.init(allocator);
911 errdefer buffer_ptr.deinit();
912
913 const db_ptr = try allocator.create(DB);
914 errdefer allocator.destroy(db_ptr);
915 db_ptr.* = try DB.init(.{ .io = io, .file = db_file, .buffer = buffer_ptr });
916
917 const map = try DB.HashMap(.read_write).init(db_ptr.rootCursor());
918
919 const parent_to_children_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "parent->children"));
920 const parent_to_children = try DB.HashMap(.read_write).init(parent_to_children_cursor);
921
922 const child_to_pending_parents_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "child->pending-parents"));
923 const child_to_pending_parents = try DB.HashMap(.read_write).init(child_to_pending_parents_cursor);
924
925 const ready_queue_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "ready-queue"));
926 const ready_queue = try DB.ArrayList(.read_write).init(ready_queue_cursor);
927
928 var self = CommitIterator(repo_opts){
929 .repo_dir = state.core.repo_dir,
930 .db_file = db_file,
931 .db = db_ptr,
932 .parent_to_children = parent_to_children,
933 .child_to_pending_parents = child_to_pending_parents,
934 .ready_queue = ready_queue,
935 .ready_queue_index = 0,
936 .newest_object_id = null,
937 .oldest_object_id = null,
938 };
939 errdefer self.deinit(io, allocator);
940
941 try self.collect(state, io, allocator, consumed_object_ids, ref);
942
943 return self;
944 }
945
946 pub fn deinit(self: *CommitIterator(repo_opts), io: std.Io, allocator: std.mem.Allocator) void {
947 self.db_file.close(io);
948 self.db.core.memory.buffer.deinit();
949 allocator.destroy(self.db.core.memory.buffer);
950 self.repo_dir.deleteFile(io, db_name) catch {};
951 allocator.destroy(self.db);
952 }
953
954 pub fn next(self: *CommitIterator(repo_opts)) !?[hash.byteLen(repo_opts.hash)]u8 {
955 if (self.ready_queue_index >= try self.ready_queue.count()) return null;
956
957 const oid = try readOidFromList(self.ready_queue.readOnly(), self.ready_queue_index);
958 self.ready_queue_index += 1;
959
960 try self.enqueueChildren(&oid);
961 return oid;
962 }
963
964 fn collect(
965 self: *CommitIterator(repo_opts),
966 state: rp.Repo(.xit, repo_opts).State(.read_only),
967 io: std.Io,
968 allocator: std.mem.Allocator,
969 consumed_object_ids: DB.HashMap(.read_only),
970 ref: rf.Ref,
971 ) !void {
972 const map = try DB.HashMap(.read_write).init(self.db.rootCursor());
973
974 const walk_queue_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "walk-queue"));
975 const walk_queue = try DB.ArrayList(.read_write).init(walk_queue_cursor);
976 var walk_queue_index: u64 = 0;
977
978 const seen_object_ids_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "seen-object-ids"));
979 const seen_object_ids = try DB.HashSet(.read_write).init(seen_object_ids_cursor);
980
981 const head_oid_hex = (try rf.readRecur(.xit, repo_opts, state, io, .{ .ref = ref })) orelse return error.OidNotFound;
982 var head_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined;
983 _ = try std.fmt.hexToBytes(&head_oid, &head_oid_hex);
984 try walk_queue.append(.{ .bytes = &head_oid });
985
986 while (walk_queue_index < try walk_queue.count()) {
987 const oid = try readOidFromList(walk_queue.readOnly(), walk_queue_index);
988 const oid_int = hash.bytesToInt(repo_opts.hash, &oid);
989
990 walk_queue_index += 1;
991
992 // if we've seen this object id, skip it
993 if (null != try seen_object_ids.getCursor(oid_int)) {
994 continue;
995 }
996
997 try seen_object_ids.put(oid_int, .{ .bytes = &oid });
998
999 var commit_object = try obj.Object(.xit, repo_opts).init(state, io, allocator, &std.fmt.bytesToHex(oid, .lower));
1000 defer commit_object.deinit();
1001
1002 // if this object id has already been consumed, skip it
1003 if (null != try consumed_object_ids.getCursor(oid_int)) {
1004 continue;
1005 }
1006
1007 if (self.newest_object_id == null) self.newest_object_id = oid;
1008 self.oldest_object_id = oid;
1009
1010 const parent_oids = commit_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound;
1011 var pending_parent_count: u64 = 0;
1012 for (parent_oids) |*parent_oid| {
1013 var parent_oid_bytes: [hash.byteLen(repo_opts.hash)]u8 = undefined;
1014 _ = try std.fmt.hexToBytes(&parent_oid_bytes, parent_oid);
1015 const parent_oid_int = hash.bytesToInt(repo_opts.hash, &parent_oid_bytes);
1016
1017 // if this object id has already been consumed, skip it
1018 if (null != try consumed_object_ids.getCursor(parent_oid_int)) {
1019 continue;
1020 }
1021
1022 // if this object id hasn't already been seen, add it to the walk queue
1023 if (null == try seen_object_ids.getCursor(parent_oid_int)) {
1024 try walk_queue.append(.{ .bytes = &parent_oid_bytes });
1025 }
1026
1027 const children_cursor = try self.parent_to_children.putCursor(parent_oid_int);
1028 const children = try DB.HashSet(.read_write).init(children_cursor);
1029 try children.put(oid_int, .{ .bytes = &oid });
1030 pending_parent_count += 1;
1031 }
1032
1033 if (pending_parent_count == 0) {
1034 try self.ready_queue.append(.{ .bytes = &oid });
1035 } else {
1036 try self.child_to_pending_parents.put(oid_int, .{ .uint = pending_parent_count });
1037 }
1038 }
1039 }
1040
1041 fn enqueueChildren(self: *CommitIterator(repo_opts), oid: *const [hash.byteLen(repo_opts.hash)]u8) !void {
1042 const oid_int = hash.bytesToInt(repo_opts.hash, oid);
1043 const children_cursor = try self.parent_to_children.getCursor(oid_int) orelse return;
1044 const children = try DB.HashSet(.read_only).init(children_cursor);
1045 var children_iter = try children.iterator();
1046
1047 while (try children_iter.next()) |*child_cursor| {
1048 const kv_pair = try child_cursor.readKeyValuePair();
1049 const child_oid_int = kv_pair.hash;
1050 const pending_parent_cursor = try self.child_to_pending_parents.getCursor(child_oid_int) orelse return error.CursorNotFound;
1051 const pending_parent_count = try pending_parent_cursor.readUint();
1052
1053 if (pending_parent_count <= 1) {
1054 _ = try self.child_to_pending_parents.remove(child_oid_int);
1055 const child_oid = hash.intToBytes(hash.HashInt(repo_opts.hash), child_oid_int);
1056 try self.ready_queue.append(.{ .bytes = &child_oid });
1057 } else {
1058 try self.child_to_pending_parents.put(child_oid_int, .{ .uint = pending_parent_count - 1 });
1059 }
1060 }
1061 }
1062
1063 fn readOidFromList(list: DB.ArrayList(.read_only), index: u64) ![hash.byteLen(repo_opts.hash)]u8 {
1064 const oid_cursor = try list.getCursor(index) orelse return error.CursorNotFound;
1065 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined;
1066 _ = try oid_cursor.readBytes(&oid);
1067 return oid;
1068 }
1069 };
1070 }
1071