╭─────────────╮ │scroll to top│ ╰─────────────╯ ╭ event.zig ────────────────────────────────────────────────────╮ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ╰───────────────────────────────────────────────────────────────╯
🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 ╭d9e70c06..╮ d9e70c06.. see-em: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 │files │ commits refs discuss issues new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────────╯ (432) user
╭────╮ ╭ clone ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ╮ │http│ ssh │ │ ╰────╯ ╰ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ╯
.. embed/ event/ test/ ui/ command.zig diff3.zig ╭─────────────────────────────────╮ │event.zig │ ╰─────────────────────────────────╯ fork.zig lib.zig main.zig main_wasm.zig patch.zig serve.zig serve_common.zig serve_http.zig serve_ssh.zig serve_ssh_protocol.zig test.zig testnet.zig try.zig ui.zig web.zig
631 try writer.flush(); 632 } 633 634 // a standalone event db holding events consumed from a local repo's events 635 // branch. it lives inside the repo dir but is not part of the repo's own data, 636 // so viewing a repo never mutates it. 637 pub fn LocalEventDB(comptime hash_kind: hash.HashKind) type { 638 return struct { 639 file: std.Io.File, 640 buffer: *std.Io.Writer.Allocating, 641 db: *EventDB(hash_kind), 642 643 const db_name = "haxy"; 644 const Self = @This(); 645 646 // open the db in `repo_dir` for writing, creating it if it doesn't exist 647 pub fn open(io: std.Io, allocator: std.mem.Allocator, repo_dir: std.Io.Dir) !Self { 648 const file = try repo_dir.createFile(io, db_name, .{ .truncate = false, .read = true, .lock = .exclusive }); 649 return fromFile(io, allocator, file); 650 } 651 652 // open the db in `repo_dir` for reading, or null if it doesn't exist 653 pub fn openReadOnly(io: std.Io, allocator: std.mem.Allocator, repo_dir: std.Io.Dir) !?Self { 654 const file = repo_dir.openFile(io, db_name, .{ .mode = .read_write, .lock = .shared }) catch |err| switch (err) { 655 error.FileNotFound => return null, 656 else => |e| return e, 657 }; 658 return try fromFile(io, allocator, file); 659 } 660 661 // remove the db from `repo_dir`, so a repo whose events branch 662 // disappeared doesn't keep showing stale issues 663 pub fn delete(io: std.Io, repo_dir: std.Io.Dir) !void { 664 repo_dir.deleteFile(io, db_name) catch |err| switch (err) { 665 error.FileNotFound => {}, 666 else => |e| return e, 667 }; 668 } 669 670 fn fromFile(io: std.Io, allocator: std.mem.Allocator, file: std.Io.File) !Self { 671 errdefer file.close(io); 672 673 const buffer = try allocator.create(std.Io.Writer.Allocating); 674 errdefer allocator.destroy(buffer); 675 buffer.* = std.Io.Writer.Allocating.init(allocator); 676 errdefer buffer.deinit(); 677 678 const db = try allocator.create(EventDB(hash_kind)); 679 errdefer allocator.destroy(db); 680 db.* = try EventDB(hash_kind).init(.{ .io = io, .file = file, .buffer = buffer }); 681 682 return .{ .file = file, .buffer = buffer, .db = db }; 683 } 684 685 pub fn deinit(self: *Self, io: std.Io, allocator: std.mem.Allocator) void { 686 self.file.close(io); 687 self.buffer.deinit(); 688 allocator.destroy(self.buffer); 689 allocator.destroy(self.db); 690 } 691 692 // consume the events on `repo`'s `ref` into this db. the db must come 693 // from `open`, whose handle holds the exclusive lock guarding the 694 // transaction. 695 pub fn consume( 696 self: *Self, 697 comptime repo_kind: rp.RepoKind, 698 comptime repo_opts: rp.RepoOpts(repo_kind), 699 io: std.Io, 700 allocator: std.mem.Allocator, 701 repo: *rp.Repo(repo_kind, repo_opts), 702 ref: rf.Ref, 703 ) !void { 704 comptime if (repo_opts.hash != hash_kind) @compileError("the repo must use the db's hash kind"); 705 const DB = EventDB(hash_kind); 706 707 const Ctx = struct { 708 repo: *rp.Repo(repo_kind, repo_opts), 709 db: *DB, 710 io: std.Io, 711 allocator: std.mem.Allocator, 712 ref: rf.Ref, 713 714 pub fn run(ctx: @This(), cursor: *DB.Cursor(.read_write)) !void { 715 var moment = try DB.HashMap(.read_write).init(cursor.*); 716 const state = rp.Repo(repo_kind, repo_opts).State(.read_write){ .core = &ctx.repo.core, .extra = .{} }; 717 if (!try consumeInTransaction(.repo, repo_kind, repo_opts, state, ctx.db, &moment, ctx.io, ctx.allocator, ctx.ref)) return error.CancelTransaction; 718 } 719 }; 720 721 const history = try DB.ArrayList(.read_write).init(self.db.rootCursor()); 722 history.appendContext(.{ .slot = try history.getSlot(-1) }, Ctx{ 723 .repo = repo, 724 .db = self.db, 725 .io = io, 726 .allocator = allocator, 727 .ref = ref, 728 }) catch |err| switch (err) { 729 error.CancelTransaction => {}, 730 else => |e| return e, 731 }; 732 } 733 }; 734 } 735 736 // consume the events on `ref` into the db `moment` writes to, returning false 737 // if there was nothing to do. events are read from `state`'s repo, which 738 // needn't be the repo backing the db — that's how a local (possibly 739 // git-backed) repo's events land in a standalone db. 740 pub fn consumeInTransaction( 741 comptime role: RepoRole, 742 comptime repo_kind: rp.RepoKind, 743 comptime repo_opts: rp.RepoOpts(repo_kind), 744 state: rp.Repo(repo_kind, repo_opts).State(.read_write), 745 db: *EventDB(repo_opts.hash), 746 moment: *EventDB(repo_opts.hash).HashMap(.read_write), 747 io: std.Io, 748 allocator: std.mem.Allocator, 749 ref: rf.Ref, 750 ) !bool { 751 const DB = EventDB(repo_opts.hash); 752 const read_state = state.readOnly(); 753 754 // the last_object_id represents the object id that was last consumed 755 var last_object_id_maybe: ?[hash.byteLen(repo_opts.hash)]u8 = null; 756 if (try moment.getCursor(hash.hashInt(repo_opts.hash, last_object_id_key))) |last_object_id_cursor| { 757 var last_object_id_buffer: [hash.byteLen(repo_opts.hash)]u8 = undefined; 758 _ = try last_object_id_cursor.readBytes(&last_object_id_buffer); 759 last_object_id_maybe = last_object_id_buffer; 760 } 761 762 // the tip was already consumed, so there is nothing to do 763 if (last_object_id_maybe) |*last_object_id| { 764 const tip_hex = (try rf.readRecur(repo_kind, repo_opts, read_state, io, .{ .ref = ref })) orelse return error.OidNotFound; 765 var tip: [hash.byteLen(repo_opts.hash)]u8 = undefined; 766 _ = try std.fmt.hexToBytes(&tip, &tip_hex); 767 if (std.mem.eql(u8, last_object_id, &tip)) return false; 768 } 769 770 // the list with all of haxy's state, including materialized views. 771 // the reason it is a list is so we can keep every previous haxy 772 // state, making it easy to revert to an older state if the user 773 // rebases some of the past commits. 774 const haxy_cursor = try moment.putCursor(hash.hashInt(repo_opts.hash, materialized_key)); 775 const haxy = try DB.ArrayList(.read_write).init(haxy_cursor); 776 777 // add a new item to the haxy list created above. 778 // we call the item haxy_moments. it is a map of object id to haxy moment. 779 // in other words, it maps each object id to a hash map containing the 780 // state that the database was in when that object id was consumed. 781 var haxy_moments_cursor = try haxy.appendCursor(); 782 // use the previous haxy_moments as the basis for this one if it exists 783 if (try haxy.getCursor(-2)) |last_haxy_moments_cursor| { 784 try haxy_moments_cursor.write(.{ .slot = last_haxy_moments_cursor.slot() }); 785 } 786 var haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor); 787 788 var commit_iter = try CommitIterator(repo_kind, repo_opts).init(read_state, io, allocator, haxy_moments.readOnly(), ref); 789 defer commit_iter.deinit(io, allocator); 790 791 // if there are no events to process, look at the oid at HEAD and update 792 // the last_object_id to point to it. this is important in situations 793 // where we do a merge and then force-push to remove the merge. see the 794 // merge test for an example. 795 if (commit_iter.newest_object_id == null) { 796 const head_oid_hex = (try rf.readRecur(repo_kind, repo_opts, read_state, io, .{ .ref = ref })) orelse return error.OidNotFound; 797 var head_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 798 _ = try std.fmt.hexToBytes(&head_oid, &head_oid_hex); 799 800 const haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &head_oid)) orelse return error.CursorNotFound; 801 const haxy_moment = try DB.HashMap(.read_only).init(haxy_moment_cursor); 802 803 const moment_index_cursor = try haxy_moment.getCursor(hash.hashInt(repo_opts.hash, "moment-index")) orelse return error.CursorNotFound; 804 const moment_index = try moment_index_cursor.readUint(); 805 806 try haxy.slice(moment_index + 1); 807 try moment.put(hash.hashInt(repo_opts.hash, last_object_id_key), .{ .bytes = &head_oid }); 808 return true; 809 } 810 811 // if this branch was rebased and force pushed, we need to detect that and 812 // properly revert the haxy state to the last valid state. we detect this 813 // by simply asking if the last event is a descendent of the last event 814 // we consumed. if it isn't, then we know the haxy state needs to be reverted. 815 if (last_object_id_maybe) |*last_object_id| { 816 var is_rebased = false; 817 const newest_object_id = commit_iter.newest_object_id orelse return error.OidNotFound; 818 819 _ = mrg.getDescendent( 820 repo_kind, 821 repo_opts, 822 read_state, 823 io, 824 allocator, 825 &std.fmt.bytesToHex(last_object_id, .lower), 826 &std.fmt.bytesToHex(&newest_object_id, .lower), 827 ) catch |err| switch (err) { 828 error.DescendentNotFound => is_rebased = true, 829 else => |e| return e, 830 }; 831 832 if (is_rebased) { 833 const oldest_object_id = commit_iter.oldest_object_id orelse return error.OidNotFound; 834 var oldest_object = try obj.Object(repo_kind, repo_opts).init(read_state, io, allocator, &std.fmt.bytesToHex(oldest_object_id, .lower)); 835 defer oldest_object.deinit(); 836 837 const parent_oids = oldest_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound; 838 switch (parent_oids.len) { 839 0 => { 840 // the branch was rebased all the way to the very beginning. 841 // we have a repo event with no parent, which means it is now 842 // the very first event. all we need to do is set the haxy list 843 // to be empty and make a new haxy_moments map to work with. 844 845 try haxy.slice(0); 846 haxy_moments_cursor = try haxy.appendCursor(); 847 haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor); 848 849 last_object_id_maybe = null; 850 }, 851 1 => { 852 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 853 _ = try std.fmt.hexToBytes(&oid, &parent_oids[0]); 854 855 const old_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &oid)) orelse return error.CursorNotFound; 856 const old_moment = try DB.HashMap(.read_only).init(old_moment_cursor); 857 858 const old_moment_index_cursor = try old_moment.getCursor(hash.hashInt(repo_opts.hash, "moment-index")) orelse return error.CursorNotFound; 859 const old_moment_index = try old_moment_index_cursor.readUint(); 860 861 // resize the haxy list so we truncate all the moments that were 862 // created after the parent_oid was consumed 863 try haxy.slice(old_moment_index + 1); 864 865 // make a new haxy moment and set its initial value to the last haxy moment 866 haxy_moments_cursor = try haxy.appendCursor(); 867 const old_haxy_moments_cursor = try haxy.getCursor(old_moment_index) orelse return error.CursorNotFound; 868 try haxy_moments_cursor.write(.{ .slot = old_haxy_moments_cursor.slot() }); 869 haxy_moments = try DB.HashMap(.read_write).init(haxy_moments_cursor); 870 871 last_object_id_maybe = oid; 872 }, 873 else => return error.UnexpectedParentCount, 874 } 875 } 876 } 877 878 while (try commit_iter.next()) |repo_event_oid| { 879 var commit_object = try obj.Object(repo_kind, repo_opts).init(read_state, io, allocator, &std.fmt.bytesToHex(repo_event_oid, .lower)); 880 defer commit_object.deinit(); 881 882 const parent_oids = commit_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound; 883 884 // create a moment for this object id 885 var haxy_moment_cursor = try haxy_moments.putCursor(hash.bytesToInt(repo_opts.hash, &repo_event_oid)); 886 887 // if there was a previous object id, set this haxy moment's initial value to it. 888 // this efficiently "clones" the map so we make further modifications based on it. 889 if (parent_oids.len > 0) { 890 var first_parent_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 891 _ = try std.fmt.hexToBytes(&first_parent_oid, &parent_oids[0]); 892 893 const first_parent_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &first_parent_oid)) orelse return error.CursorNotFound; 894 try haxy_moment_cursor.write(.{ .slot = first_parent_haxy_moment_cursor.slot() }); 895 } 896 897 const haxy_moment = try DB.HashMap(.read_write).init(haxy_moment_cursor); 898 899 // graph generation gives every event a causal order independent of 900 // commit timestamps. concurrent events tie-break by id in their index. 901 var event_order: u64 = 1; 902 for (parent_oids) |*parent_oid| { 903 var parent_id: [hash.byteLen(repo_opts.hash)]u8 = undefined; 904 _ = try std.fmt.hexToBytes(&parent_id, parent_oid); 905 const parent_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &parent_id)) orelse return error.CursorNotFound; 906 const parent = try DB.HashMap(.read_only).init(parent_cursor); 907 const parent_order_cursor = try parent.getCursor(hash.hashInt(repo_opts.hash, event_order_key)) orelse return error.CursorNotFound; 908 const parent_order = try parent_order_cursor.readUint(); 909 event_order = @max(event_order, parent_order +| 1); 910 } 911 try haxy_moment.put(hash.hashInt(repo_opts.hash, event_order_key), .{ .uint = event_order }); 912 913 // merge changes from every parent after the first parent. the common 914 // ancestor is the baseline, so a later parent only contributes records 915 // it changed relative to the merge base. only the record maps are 916 // merged; every other view is derived from them, and each kind's 917 // `consume` re-derives its own as the winning records are written. 918 if (parent_oids.len > 1) { 919 for (parent_oids[1..]) |*parent_oid| { 920 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 921 _ = try std.fmt.hexToBytes(&oid, parent_oid); 922 923 const parent_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &oid)) orelse return error.CursorNotFound; 924 const parent_haxy_moment = try DB.HashMap(.read_only).init(parent_haxy_moment_cursor); 925 926 const baseline_oid_hex = try mrg.commonAncestor(repo_kind, repo_opts, read_state, io, allocator, &parent_oids[0], parent_oid); 927 var baseline_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 928 _ = try std.fmt.hexToBytes(&baseline_oid, &baseline_oid_hex); 929 const baseline_haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(repo_opts.hash, &baseline_oid)) orelse return error.CursorNotFound; 930 const baseline_haxy_moment = try DB.HashMap(.read_only).init(baseline_haxy_moment_cursor); 931 932 inline for (std.meta.fields(Event)) |field| { 933 const T = @typeInfo(field.type).optional.child; 934 try merge(T, DB, repo_opts.hash, allocator, haxy_moment, parent_haxy_moment, baseline_haxy_moment); 935 } 936 } 937 } 938 939 // associate this moment with the index it will first appear at in the haxy list. 940 // this will be important later so we can truncate that list if the user ever 941 // rebases starting at this object id. 942 try haxy_moment.put(hash.hashInt(repo_opts.hash, "moment-index"), .{ .uint = try haxy.count() - 1 }); 943 944 // consume the event unless it's a merge commit 945 if (parent_oids.len <= 1) { 946 var arena = std.heap.ArenaAllocator.init(allocator); 947 defer arena.deinit(); 948 949 // read the message from the commit 950 var message: std.ArrayList(u8) = .empty; 951 try commit_object.readMessage(arena.allocator(), &message, .limited(max_event_size)); 952 953 const event_with_id = try EventWithId.fromString(&arena, message.items); 954 if (!role.allows(std.meta.activeTag(event_with_id.event))) return error.EventKindNotAllowed; 955 956 const current_event_id = try parseEventId(&event_with_id.id); 957 958 // wrap the payload into the record `consume` stores, with its 959 // commit-derived fields; on update, `consume` preserves the 960 // existing record's 961 switch (event_with_id.event) { 962 .user => |event_maybe| { 963 const record_maybe: ?User.Record = if (event_maybe) |event| .{ .event = event, .created_order = event_order } else null; 964 try User.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 965 }, 966 .repo => |event_maybe| { 967 const record_maybe: ?Repo.Record = if (event_maybe) |event| .{ .event = event, .created_order = event_order } else null; 968 try Repo.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 969 }, 970 .fork => |event_maybe| { 971 const record_maybe: ?Fork.Record = if (event_maybe) |event| .{ .event = event, .created_order = event_order } else null; 972 try Fork.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 973 }, 974 .issue => |event_maybe| { 975 const record_maybe: ?Issue.Record = if (event_maybe) |event| .{ 976 .event = event, 977 .author_email = authorEmail(commit_object.content.commit.metadata.author orelse ""), 978 .created_order = event_order, 979 } else null; 980 try Issue.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 981 }, 982 .discuss => |event_maybe| { 983 const record_maybe: ?Discussion.Record = if (event_maybe) |event| .{ 984 .event = event, 985 .author_email = authorEmail(commit_object.content.commit.metadata.author orelse ""), 986 .created_order = event_order, 987 } else null; 988 try Discussion.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 989 }, 990 .comment => |event_maybe| { 991 const record_maybe: ?Comment.Record = if (event_maybe) |event| .{ 992 .event = event, 993 .author_email = authorEmail(commit_object.content.commit.metadata.author orelse ""), 994 .created_order = event_order, 995 } else null; 996 try Comment.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 997 }, 998 .attach => |event_maybe| { 999 const record_maybe: ?Attachment.Record = if (event_maybe) |event| blk: { 1000 const entry = try attachedFile(repo_kind, repo_opts, read_state, io, allocator, &arena, &commit_object.content.commit.tree); 1001 break :blk .{ 1002 .event = event, 1003 .author_email = authorEmail(commit_object.content.commit.metadata.author orelse ""), 1004 .created_order = event_order, 1005 .name = entry.name, 1006 .blob_oid = entry.oid, 1007 }; 1008 } else null; 1009 try Attachment.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 1010 }, 1011 .patchrev => |event_maybe| { 1012 const record_maybe: ?PatchRev.Record = if (event_maybe) |event| blk: { 1013 const trees = try PatchRev.readTrees(repo_kind, repo_opts, read_state, io, allocator, &commit_object.content.commit.tree); 1014 const existing_maybe = try PatchRev.readById(DB, repo_opts.hash, haxy_moment.readOnly(), &arena, &current_event_id); 1015 const event_oid = std.fmt.bytesToHex(repo_event_oid, .lower); 1016 1017 var patch_oid: [hash.hexLen(repo_opts.hash)]u8 = undefined; 1018 if (existing_maybe) |existing| { 1019 if (existing.patch_oid.len != patch_oid.len) return error.InvalidPatch; 1020 @memcpy(&patch_oid, existing.patch_oid); 1021 } else { 1022 const author = try commitIdentity(commit_object.content.commit.metadata.author orelse return error.AuthorNotFound); 1023 const committer = try commitIdentity(commit_object.content.commit.metadata.committer orelse author); 1024 patch_oid = try PatchRev.writeSquashCommit( 1025 repo_kind, 1026 repo_opts, 1027 state, 1028 io, 1029 allocator, 1030 event, 1031 &trees.head, 1032 author, 1033 committer, 1034 commit_object.content.commit.metadata.timestamp, 1035 ); 1036 } 1037 1038 break :blk .{ 1039 .event = event, 1040 .author_email = authorEmail(commit_object.content.commit.metadata.author orelse ""), 1041 .created_order = event_order, 1042 .event_oid = &event_oid, 1043 .base_tree_oid = &trees.base, 1044 .head_tree_oid = &trees.head, 1045 .patch_oid = &patch_oid, 1046 }; 1047 } else null; 1048 try PatchRev.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 1049 }, 1050 .patch => |event_maybe| { 1051 const record_maybe: ?Patch.Record = if (event_maybe) |event| .{ 1052 .event = event, 1053 .author_email = authorEmail(commit_object.content.commit.metadata.author orelse ""), 1054 .created_order = event_order, 1055 } else null; 1056 try Patch.consume(DB, repo_opts.hash, haxy_moment, &current_event_id, record_maybe, &arena, &repo_event_oid); 1057 }, 1058 } 1059 } 1060 1061 // the current object id is now the last one 1062 last_object_id_maybe = repo_event_oid; 1063 1064 // prevent any of the data created above from being mutated by future iterations of this loop 1065 try db.freeze(); 1066 } 1067 1068 if (last_object_id_maybe) |*last_object_id| { 1069 try moment.put(hash.hashInt(repo_opts.hash, last_object_id_key), .{ .bytes = last_object_id }); 1070 } 1071 1072 return true; 1073 } 1074 1075 // the conflicted records of one kind, keyed by the same order key its id set 1076 // uses, so this doubles as the index the conflicts view lists and counts. every 1077 // value under an entry is a slot reference to a structure that already exists: 1078 // 1079 // <order key> -> conflict entry 1080 // conflicted-fields the conflicting field names, space separated 1081 // base-record the merge base's whole record 1082 // their-record the merged-in side's whole record 1083 // their-field->oid that side's field name to commit map 1084 // 1085 // our own value is the live record and our commit is in the kind's oid map, so 1086 // neither is repeated here. 1087 pub const conflicted_fields_key = "conflicted-fields"; 1088 pub const base_record_key = "base-record"; 1089 pub const their_record_key = "their-record"; 1090 pub const their_field_to_oid_key = "their-field->oid"; 1091 1092 // record the commit against every field this event changes. a merge passes no 1093 // oid, since each field's winner may come from a different side. 1094 pub fn writeOid( 1095 comptime T: type, 1096 comptime DB: type, 1097 comptime hash_kind: hash.HashKind, 1098 id_to_field_to_oid: DB.HashMap(.read_write), 1099 record_key: hash.HashInt(hash_kind), 1100 existing_maybe: ?T, 1101 event: T, 1102 event_oid: ?[]const u8, 1103 ) !void { 1104 const oid = event_oid orelse return; 1105 1106 // created only once a field actually changes 1107 var field_oids: ?DB.SortedMap(.read_write) = null; 1108 1109 inline for (std.meta.fields(T)) |field| { 1110 const changed = if (existing_maybe) |existing| 1111 !fieldEqual(field.type, @field(existing, field.name), @field(event, field.name)) 1112 else 1113 true; 1114 if (changed) { 1115 if (field_oids == null) { 1116 field_oids = try DB.SortedMap(.read_write).init(try id_to_field_to_oid.putCursor(record_key)); 1117 } 1118 if (field_oids) |map| try map.put(field.name, .{ .bytes = oid }); 1119 } 1120 } 1121 } 1122 1123 // the whole field list, space separated, is the longest it can get 1124 fn conflictedFieldsMaxLen(comptime T: type) usize { 1125 var len: usize = 0; 1126 for (std.meta.fields(T)) |field| len += field.name.len + 1; 1127 return len; 1128 } 1129 1130 pub const MergePolicy = enum { 1131 field_conflicts, 1132 target_wins, 1133 }; 1134 1135 // what a three-way merge did with a field 1136 pub const FieldMerge = enum { 1137 // the target's value stands, because neither side changed it, both made the 1138 // same change, or only the target changed it 1139 kept, 1140 parent, 1141 conflicted, 1142 }; 1143 1144 // three-way merge each field. one only the parent changed takes the parent's 1145 // value; one both sides changed differently keeps the target's. a missing 1146 // baseline makes every differing field a conflict. 1147 fn mergeFields( 1148 comptime T: type, 1149 baseline_maybe: ?T, 1150 target: T, 1151 parent: T, 1152 outcome: *[std.meta.fields(T).len]FieldMerge, 1153 ) T { 1154 var merged = target; 1155 1156 inline for (std.meta.fields(T), 0..) |field, i| { 1157 const target_value = @field(target, field.name); 1158 const parent_value = @field(parent, field.name); 1159 1160 if (!fieldEqual(field.type, target_value, parent_value)) { 1161 outcome[i] = if (baseline_maybe) |baseline| blk: { 1162 const baseline_value = @field(baseline, field.name); 1163 if (fieldEqual(field.type, target_value, baseline_value)) { 1164 @field(merged, field.name) = parent_value; 1165 break :blk .parent; 1166 } 1167 // only the target changed it 1168 if (fieldEqual(field.type, parent_value, baseline_value)) break :blk .kept; 1169 break :blk .conflicted; 1170 } else .conflicted; 1171 } 1172 } 1173 1174 return merged; 1175 } 1176 1177 // three-way merge of one kind's records from a merge parent. kinds that expose 1178 // conflicts merge field by field; the others take one whole record. the result 1179 // goes through the kind's `consume`, which re-derives the indexes from it. a 1180 // removal carries over unless the other side changed the record, so a merge 1181 // never fails. 1182 pub fn merge( 1183 comptime T: type, 1184 comptime DB: type, 1185 comptime hash_kind: hash.HashKind, 1186 allocator: std.mem.Allocator, 1187 haxy_moment: DB.HashMap(.read_write), 1188 parent_moment: DB.HashMap(.read_only), 1189 baseline_moment: DB.HashMap(.read_only), 1190 ) !void { 1191 const parent_set_cursor = try parent_moment.getCursor(hash.hashInt(hash_kind, T.id_set_key)) orelse return; 1192 const parent_set = try DB.SortedSet(.read_only).init(parent_set_cursor); 1193 1194 const parent_records_cursor = try parent_moment.getCursor(hash.hashInt(hash_kind, T.record_map_key)) orelse return; 1195 const parent_records = try DB.HashMap(.read_only).init(parent_records_cursor); 1196 1197 var baseline_records: ?DB.HashMap(.read_only) = null; 1198 if (try baseline_moment.getCursor(hash.hashInt(hash_kind, T.record_map_key))) |cursor| { 1199 baseline_records = try DB.HashMap(.read_only).init(cursor); 1200 } 1201 1202 var parent_conflicts: ?DB.SortedMap(.read_only) = null; 1203 var baseline_conflicts: ?DB.SortedMap(.read_only) = null; 1204 var parent_id_to_field_to_oid: ?DB.HashMap(.read_only) = null; 1205 switch (comptime T.merge_policy) { 1206 .field_conflicts => { 1207 if (try parent_moment.getCursor(hash.hashInt(hash_kind, T.conflicts_key))) |cursor| { 1208 parent_conflicts = try DB.SortedMap(.read_only).init(cursor); 1209 } 1210 if (try baseline_moment.getCursor(hash.hashInt(hash_kind, T.conflicts_key))) |cursor| { 1211 baseline_conflicts = try DB.SortedMap(.read_only).init(cursor); 1212 } 1213 if (try parent_moment.getCursor(hash.hashInt(hash_kind, T.id_to_field_to_oid_key))) |cursor| { 1214 parent_id_to_field_to_oid = try DB.HashMap(.read_only).init(cursor); 1215 } 1216 }, 1217 .target_wins => {}, 1218 } 1219 1220 var arena = std.heap.ArenaAllocator.init(allocator); 1221 defer arena.deinit(); 1222 1223 var iter = try parent_set.iteratorFromIndex(0); 1224 while (try iter.next()) |kv_pair_cursor| { 1225 const event_id = try readOrderKeyId(DB, kv_pair_cursor); 1226 const record_key = hash.hashInt(hash_kind, &event_id); 1227 1228 _ = arena.reset(.retain_capacity); 1229 1230 const parent_record_cursor = try parent_records.getCursor(record_key) orelse continue; 1231 const parent_slot = parent_record_cursor.slot(); 1232 const baseline_slot = if (baseline_records) |map| try map.getSlot(record_key) else null; 1233 1234 // the parent left this record alone 1235 if (baseline_slot) |slot| { 1236 if (slot.eql(parent_slot)) continue; 1237 } 1238 1239 const parent_record = try read(T.Record, DB, hash_kind, &arena, try DB.HashMap(.read_only).init(parent_record_cursor)); 1240 1241 // re-derived every iteration, since `consume` writes through its own 1242 // handles 1243 const target_records_cursor = try haxy_moment.putCursor(hash.hashInt(hash_kind, T.record_map_key)); 1244 const target_records = try DB.HashMap(.read_write).init(target_records_cursor); 1245 1246 // a record the target doesn't have arrives wholesale from the parent 1247 var outcome: [std.meta.fields(T).len]FieldMerge = @splat(.parent); 1248 var merged = parent_record; 1249 1250 if (try target_records.getCursor(record_key)) |target_record_cursor| { 1251 // both sides share the change 1252 if (target_record_cursor.slot().eql(parent_slot)) continue; 1253 1254 const target_record = try read(T.Record, DB, hash_kind, &arena, try DB.HashMap(.read_only).init(target_record_cursor)); 1255 1256 var baseline_record: ?T.Record = null; 1257 if (baseline_records) |map| { 1258 if (try map.getCursor(record_key)) |baseline_record_cursor| { 1259 baseline_record = try read(T.Record, DB, hash_kind, &arena, try DB.HashMap(.read_only).init(baseline_record_cursor)); 1260 } 1261 } 1262 1263 switch (comptime T.merge_policy) { 1264 .field_conflicts => { 1265 outcome = @splat(.kept); 1266 merged = target_record; 1267 merged.event = mergeFields(T, if (baseline_record) |baseline| baseline.event else null, target_record.event, parent_record.event, &outcome); 1268 if (comptime @hasDecl(T, "resolveMerge")) { 1269 T.resolveMerge(target_record.event, parent_record.event, &merged.event, &outcome); 1270 } 1271 1272 // merge the removed state separately from serialized fields. removal 1273 // carries over an unchanged record, while an edit or restoration 1274 // keeps it active. 1275 merged.removed = if (target_record.removed == parent_record.removed) 1276 target_record.removed 1277 else if (baseline_record) |baseline| blk: { 1278 const active = if (!target_record.removed) target_record else parent_record; 1279 if (active.removed != baseline.removed) break :blk false; 1280 if (!fieldsEqual(T, active.event, baseline.event)) break :blk false; 1281 break :blk true; 1282 } else false; 1283 }, 1284 .target_wins => { 1285 // an uncontested parent change carries over. if both sides 1286 // changed the record, the target wins as a whole, except that 1287 // an edit or restoration wins over a removal. 1288 const target_changed = if (baseline_slot) |slot| 1289 !target_record_cursor.slot().eql(slot) 1290 else 1291 true; 1292 merged = if (!target_changed) 1293 parent_record 1294 else if (target_record.removed != parent_record.removed) 1295 if (target_record.removed) parent_record else target_record 1296 else 1297 target_record; 1298 }, 1299 } 1300 } 1301 1302 // the index is unique, so a key the target gave to a different record 1303 // leaves this one out rather than stranding that one behind a name it 1304 // no longer owns 1305 if (!merged.removed and @hasDecl(T, "name_index_key")) { 1306 const index_key = try merged.indexKey(arena.allocator()); 1307 const name_index_cursor = try haxy_moment.putCursor(hash.hashInt(hash_kind, T.name_index_key)); 1308 const name_index = try DB.HashMap(.read_write).init(name_index_cursor); 1309 if (try name_index.getCursor(hash.hashInt(hash_kind, index_key))) |owner_cursor| { 1310 var owner_id: [event_id_size]u8 = undefined; 1311 _ = try owner_cursor.readBytes(&owner_id); 1312 if (!std.mem.eql(u8, &owner_id, &event_id)) continue; 1313 } 1314 } 1315 1316 // `consume` sets no oids, since each field's winner may come from 1317 // a different side 1318 try T.consume(DB, hash_kind, haxy_moment, &event_id, merged, &arena, null); 1319 1320 switch (comptime T.merge_policy) { 1321 .field_conflicts => {}, 1322 .target_wins => continue, 1323 } 1324 1325 // removed records do not carry field conflicts 1326 if (merged.removed) continue; 1327 1328 const conflicts_cursor = try haxy_moment.putCursor(hash.hashInt(hash_kind, T.conflicts_key)); 1329 const conflicts = try DB.SortedMap(.read_write).init(conflicts_cursor); 1330 const id_to_field_to_oid_cursor = try haxy_moment.putCursor(hash.hashInt(hash_kind, T.id_to_field_to_oid_key)); 1331 const id_to_field_to_oid = try DB.HashMap(.read_write).init(id_to_field_to_oid_cursor); 1332 const merged_key = orderKeyDesc(merged.created_order, &event_id); 1333 1334 var parent_field_oids: ?DB.SortedMap(.read_only) = null; 1335 if (parent_id_to_field_to_oid) |map| { 1336 if (try map.getCursor(record_key)) |field_oids_cursor| { 1337 parent_field_oids = try DB.SortedMap(.read_only).init(field_oids_cursor); 1338 } 1339 } 1340 // created only once a field actually comes from the parent 1341 var field_oids: ?DB.SortedMap(.read_write) = null; 1342 1343 // the fields both sides changed differently, space separated 1344 var conflicted_fields: [conflictedFieldsMaxLen(T)]u8 = undefined; 1345 var conflicted_len: usize = 0; 1346 1347 inline for (std.meta.fields(T), 0..) |field, i| { 1348 switch (outcome[i]) { 1349 // the target's value stands, so its oid does too 1350 .kept => {}, 1351 // the parent set this value, so its commit takes the field. a 1352 // parent with no oid took its own value from a merge, leaving 1353 // the field unattributed rather than crediting the target's 1354 // commit for a value it no longer holds. 1355 .parent => { 1356 if (field_oids == null) { 1357 field_oids = try DB.SortedMap(.read_write).init(try id_to_field_to_oid.putCursor(record_key)); 1358 } 1359 const map = field_oids orelse unreachable; 1360 const parent_oid = if (parent_field_oids) |parent_map| try parent_map.getCursor(field.name) else null; 1361 if (parent_oid) |oid_cursor| { 1362 try map.put(field.name, .{ .slot = oid_cursor.slot() }); 1363 } else { 1364 _ = try map.remove(field.name); 1365 } 1366 }, 1367 .conflicted => { 1368 if (conflicted_len > 0) { 1369 conflicted_fields[conflicted_len] = ' '; 1370 conflicted_len += 1; 1371 } 1372 @memcpy(conflicted_fields[conflicted_len..][0..field.name.len], field.name); 1373 conflicted_len += field.name.len; 1374 }, 1375 } 1376 } 1377 1378 if (conflicted_len > 0) { 1379 // each side's whole record is stored by reference, so the view can 1380 // show any of them or take theirs wholesale 1381 _ = try conflicts.remove(&merged_key); 1382 const conflict = try DB.HashMap(.read_write).init(try conflicts.putCursor(&merged_key)); 1383 try conflict.put(hash.hashInt(hash_kind, conflicted_fields_key), .{ .bytes = conflicted_fields[0..conflicted_len] }); 1384 try conflict.put(hash.hashInt(hash_kind, base_record_key), .{ .slot = baseline_slot }); 1385 try conflict.put(hash.hashInt(hash_kind, their_record_key), .{ .slot = parent_slot }); 1386 if (parent_field_oids) |map| { 1387 try conflict.put(hash.hashInt(hash_kind, their_field_to_oid_key), .{ .slot = map.slot() }); 1388 } 1389 continue; 1390 } 1391 1392 // created_order never changes, so the parent keys this record's conflict 1393 // identically, and it carries over when the target has none, unless the 1394 // parent's entry is unchanged from the baseline, meaning the target 1395 // inherited it too and resolving it removed it here. 1396 if (parent_conflicts) |map| { 1397 if (null == try conflicts.getCursor(&merged_key)) { 1398 if (try map.getSlot(&merged_key)) |conflict_slot| { 1399 const baseline_conflict_slot = if (baseline_conflicts) |baseline_map| try baseline_map.getSlot(&merged_key) else null; 1400 const resolved = if (baseline_conflict_slot) |slot| slot.eql(conflict_slot) else false; 1401 if (!resolved) try conflicts.put(&merged_key, .{ .slot = conflict_slot }); 1402 } 1403 } 1404 } 1405 } 1406 } 1407 1408 pub fn currentMoment( 1409 comptime repo_opts: rp.RepoOpts(.xit), 1410 repo: *rp.Repo(.xit, repo_opts), 1411 ) !rp.Repo(.xit, repo_opts).DB.HashMap(.read_only) { 1412 return currentMomentFromDb(repo_opts.hash, &repo.core.db); 1413 } 1414 1415 pub fn currentMomentFromDb( 1416 comptime hash_kind: hash.HashKind, 1417 db: *EventDB(hash_kind), 1418 ) !EventDB(hash_kind).HashMap(.read_only) { 1419 const DB = EventDB(hash_kind); 1420 const history = try DB.ArrayList(.read_only).init(db.rootCursor().readOnly()); 1421 const moment_cursor = try history.getCursor(-1) orelse return error.NotFound; 1422 const moment = try DB.HashMap(.read_only).init(moment_cursor); 1423 return try currentMomentFromRepoMoment(hash_kind, moment); 1424 } 1425 1426 pub fn currentMomentFromRepoMoment( 1427 comptime hash_kind: hash.HashKind, 1428 moment: EventDB(hash_kind).HashMap(.read_only), 1429 ) !EventDB(hash_kind).HashMap(.read_only) { 1430 const DB = EventDB(hash_kind); 1431 const last_object_id_cursor = try moment.getCursor(hash.hashInt(hash_kind, last_object_id_key)) orelse return error.NotFound; 1432 var last_object_id: [hash.byteLen(hash_kind)]u8 = undefined; 1433 _ = try last_object_id_cursor.readBytes(&last_object_id); 1434 const haxy_cursor = try moment.getCursor(hash.hashInt(hash_kind, materialized_key)) orelse return error.NotFound; 1435 const haxy = try DB.ArrayList(.read_only).init(haxy_cursor); 1436 const haxy_moments_cursor = try haxy.getCursor(-1) orelse return error.NotFound; 1437 const haxy_moments = try DB.HashMap(.read_only).init(haxy_moments_cursor); 1438 const haxy_moment_cursor = try haxy_moments.getCursor(hash.bytesToInt(hash_kind, &last_object_id)) orelse return error.NotFound; 1439 return try DB.HashMap(.read_only).init(haxy_moment_cursor); 1440 } 1441 1442 // build a sorted-set key from an order and event id. big-endian order makes 1443 // byte order match numeric order; the event id breaks ties. 1444 pub fn orderKey(order: u64, event_id: *const [event_id_size]u8) [@sizeOf(u64) + event_id_size]u8 { 1445 var key: [@sizeOf(u64) + event_id_size]u8 = undefined; 1446 std.mem.writeInt(u64, key[0..@sizeOf(u64)], order, .big); 1447 @memcpy(key[@sizeOf(u64)..], event_id); 1448 return key; 1449 } 1450 1451 pub fn orderKeyDesc(order: u64, event_id: *const [event_id_size]u8) [@sizeOf(u64) + event_id_size]u8 { 1452 return orderKey(std.math.maxInt(u64) - order, event_id); 1453 } 1454 1455 // add or refresh one entry in the global logical-event index. updates and 1456 // removed records preserve created_order, so the key never moves. 1457 pub fn indexEvent( 1458 comptime DB: type, 1459 comptime hash_kind: hash.HashKind, 1460 haxy_moment: DB.HashMap(.read_write), 1461 event_id: *const [event_id_size]u8, 1462 kind: EventKind, 1463 created_order: u64, 1464 removed: bool, 1465 ) !void { 1466 const kind_map_cursor = try haxy_moment.putCursor(hash.hashInt(hash_kind, event_index_key)); 1467 const kind_map = try DB.HashMap(.read_write).init(kind_map_cursor); 1468 try kind_map.put(hash.hashInt(hash_kind, event_id), .{ .bytes = @tagName(kind) }); 1469 1470 const active_cursor = try haxy_moment.putCursor(hash.hashInt(hash_kind, active_event_id_set_key)); 1471 const active = try DB.SortedSet(.read_write).init(active_cursor); 1472 const removed_cursor = try haxy_moment.putCursor(hash.hashInt(hash_kind, removed_event_id_set_key)); 1473 const removed_set = try DB.SortedSet(.read_write).init(removed_cursor); 1474 const order_key = orderKeyDesc(created_order, event_id); 1475 if (removed) { 1476 _ = try active.remove(&order_key); 1477 try removed_set.put(&order_key); 1478 } else { 1479 _ = try removed_set.remove(&order_key); 1480 try active.put(&order_key); 1481 } 1482 } 1483 1484 pub fn touchThread( 1485 comptime DB: type, 1486 comptime hash_kind: hash.HashKind, 1487 haxy_moment: DB.HashMap(.read_write), 1488 thread_id: *const [event_id_size]u8, 1489 order: u64, 1490 arena: *std.heap.ArenaAllocator, 1491 ) !void { 1492 const kind_map_cursor = try haxy_moment.getCursor(hash.hashInt(hash_kind, event_index_key)) orelse return; 1493 const kind_map = try DB.HashMap(.read_only).init(kind_map_cursor); 1494 const kind = (try readEventKind(hash_kind, kind_map, thread_id)) orelse return; 1495 switch (kind) { 1496 .discuss => try Discussion.touch(DB, hash_kind, haxy_moment, thread_id, order, arena), 1497 else => {}, 1498 } 1499 } 1500 1501 pub fn readEventKind( 1502 comptime hash_kind: hash.HashKind, 1503 kind_map: anytype, 1504 id: *const [event_id_size]u8, 1505 ) !?EventKind { 1506 const cursor = try kind_map.getCursor(hash.hashInt(hash_kind, id)) orelse return null; 1507 var buffer: [64]u8 = undefined; 1508 return std.meta.stringToEnum(EventKind, try cursor.readBytes(&buffer)) orelse error.InvalidEventKind; 1509 } 1510 1511 // the event id embedded in a set entry's order key 1512 pub fn readOrderKeyId(comptime DB: type, kv_pair_cursor: DB.Cursor(.read_only)) ![event_id_size]u8 { 1513 var cursor = kv_pair_cursor; 1514 const kv_pair = try cursor.readKeyValuePair(); 1515 var order_key: [@sizeOf(u64) + event_id_size]u8 = undefined; 1516 _ = try kv_pair.key_cursor.readBytes(&order_key); 1517 return order_key[@sizeOf(u64)..][0..event_id_size].*; 1518 } 1519 1520 // split a pushed "<owner>/<repo>" path into its two components, or null if it 1521 // isn't exactly two non-empty segments. 1522 pub fn parseOwnerRepoPath(path: []const u8) ?struct { owner: []const u8, name: []const u8 } { 1523 var it = std.mem.splitScalar(u8, path, '/'); 1524 const owner = it.next() orelse return null; 1525 const name = it.next() orelse return null; 1526 if (it.next() != null) return null; 1527 if (owner.len == 0 or name.len == 0) return null; 1528 1529 return .{ .owner = owner, .name = name }; 1530 } 1531 1532 pub const CreateRepoOptions = struct { 1533 read_access: Repo.Access = .private, 1534 }; 1535 1536 // resolve a pushed `<owner>/<repo>` to the hex event id that names its on-disk 1537 // directory under the repos dir 1538 pub fn resolveOrCreateRepo( 1539 io: std.Io, 1540 allocator: std.mem.Allocator, 1541 admin_repo_path: []const u8, 1542 owner_name: []const u8, 1543 repo_name: []const u8, 1544 create_options_maybe: ?CreateRepoOptions, 1545 ) !?[event_id_size * 2]u8 { 1546 var repo = rp.Repo(.xit, admin_repo_opts).open(io, allocator, .{ .path = admin_repo_path }) catch |err| switch (err) { 1547 error.RepoNotFound => return null, 1548 else => |e| return e, 1549 }; 1550 defer repo.deinit(io, allocator); 1551 1552 var arena = std.heap.ArenaAllocator.init(allocator); 1553 defer arena.deinit(); 1554 1555 const moment = try currentMoment(admin_repo_opts, &repo); 1556 1557 // an already-registered repo reuses its event id, so a re-push (or a clone) 1558 // lands in the same repo 1559 if (try Repo.readByOwnerAndName(AdminDB, admin_repo_opts.hash, moment, &arena, owner_name, repo_name)) |found| { 1560 return std.fmt.bytesToHex(found.event_id, .lower); 1561 } 1562 1563 const create_options = create_options_maybe orelse return null; 1564 1565 // the new repo is owned by the named user, read from the name index; an 1566 // unknown owner can't own a repo, so the push is rejected 1567 const name_to_user_id_cursor = try moment.getCursor(hash.hashInt(admin_repo_opts.hash, "name->user-id")) orelse return null; 1568 const name_to_user_id = try AdminDB.HashMap(.read_only).init(name_to_user_id_cursor); 1569 const owner_id_cursor = try name_to_user_id.getCursor(hash.hashInt(admin_repo_opts.hash, owner_name)) orelse return null; 1570 var owner_user_id: [event_id_size]u8 = undefined; 1571 _ = try owner_id_cursor.readBytes(&owner_user_id); 1572 1573 const owner = (try User.readById(AdminDB, admin_repo_opts.hash, moment, &arena, &owner_user_id)) orelse unreachable; 1574 1575 var id_bytes: [event_id_size]u8 = undefined; 1576 io.random(&id_bytes); 1577 const event_id_hex = std.fmt.bytesToHex(id_bytes, .lower); 1578 1579 try consume(.admin, .xit, admin_repo_opts, io, allocator, &repo, events_ref, &[_]EventWithId{.{ 1580 .id = event_id_hex, 1581 .timestamp = @intCast(std.Io.Timestamp.now(io, .real).toSeconds()), 1582 .author = .{ .name = owner.event.name, .email = owner.event.email }, 1583 .event = .{ .repo = .{ 1584 .user_id = &owner_user_id, 1585 .name = repo_name, 1586 .description = "", 1587 .read_access = create_options.read_access, 1588 } }, 1589 }}); 1590 1591 return event_id_hex; 1592 } 1593 1594 // 1595 // reading from xitdb 1596 // 1597 1598 pub fn read( 1599 comptime T: type, 1600 comptime DB: type, 1601 comptime hash_kind: hash.HashKind, 1602 arena: *std.heap.ArenaAllocator, 1603 map: DB.HashMap(.read_only), 1604 ) !T { 1605 var event: T = undefined; 1606 1607 switch (@typeInfo(T)) { 1608 .@"struct" => |struct_info| { 1609 inline for (struct_info.fields) |field| { 1610 switch (@typeInfo(field.type)) { 1611 .pointer => |pointer_info| { 1612 if (pointer_info.size == .slice and pointer_info.child == u8 and pointer_info.is_const) { 1613 @field(event, field.name) = try readBytes(DB, hash_kind, arena.allocator(), map, field.name); 1614 } else { 1615 @compileError("unsupported read field type: " ++ @typeName(field.type)); 1616 } 1617 }, 1618 .array => |array_info| { 1619 if (array_info.child == u8) { 1620 const bytes = try readBytes(DB, hash_kind, arena.allocator(), map, field.name); 1621 if (bytes.len != array_info.len) return error.InvalidByteArrayLength; 1622 @memcpy(@field(event, field.name)[0..], bytes); 1623 } else { 1624 @compileError("unsupported read field type: " ++ @typeName(field.type)); 1625 } 1626 }, 1627 .bool => @field(event, field.name) = try readBool(DB, hash_kind, map, field.name), 1628 .int => |int_info| switch (int_info.signedness) { 1629 .unsigned => @field(event, field.name) = @intCast(try readUint(DB, hash_kind, map, field.name)), 1630 .signed => @field(event, field.name) = @intCast(try readInt(DB, hash_kind, map, field.name)), 1631 }, 1632 .@"enum" => { 1633 const bytes = try readBytes(DB, hash_kind, arena.allocator(), map, field.name); 1634 @field(event, field.name) = std.meta.stringToEnum(field.type, bytes) orelse return error.InvalidEnumTag; 1635 }, 1636 .optional => |optional_info| { 1637 // a missing key is null 1638 if (try map.getCursor(hash.hashInt(hash_kind, field.name))) |cursor| { 1639 if (optional_info.child == []const u8) { 1640 @field(event, field.name) = try cursor.readBytesAlloc(arena.allocator(), null); 1641 } else switch (@typeInfo(optional_info.child)) { 1642 .array => |array_info| { 1643 if (array_info.child != u8) @compileError("unsupported read field type: " ++ @typeName(field.type)); 1644 var bytes: optional_info.child = undefined; 1645 const value = try cursor.readBytes(&bytes); 1646 if (value.len != bytes.len) return error.InvalidByteArrayLength; 1647 @field(event, field.name) = bytes; 1648 }, 1649 .@"struct" => @field(event, field.name) = try read( 1650 optional_info.child, 1651 DB, 1652 hash_kind, 1653 arena, 1654 try DB.HashMap(.read_only).init(cursor), 1655 ), 1656 else => @compileError("unsupported read field type: " ++ @typeName(field.type)), 1657 } 1658 } else { 1659 @field(event, field.name) = null; 1660 } 1661 }, 1662 .@"struct" => { 1663 const cursor = try map.getCursor(hash.hashInt(hash_kind, field.name)) orelse return error.NotFound; 1664 @field(event, field.name) = try read( 1665 field.type, 1666 DB, 1667 hash_kind, 1668 arena, 1669 try DB.HashMap(.read_only).init(cursor), 1670 ); 1671 }, 1672 else => @compileError("unsupported read field type: " ++ @typeName(field.type)), 1673 } 1674 } 1675 }, 1676 else => @compileError("read expects a struct"), 1677 } 1678 1679 return event; 1680 } 1681 1682 // whether two events carry the same data 1683 fn fieldsEqual(comptime T: type, a: T, b: T) bool { 1684 switch (@typeInfo(T)) { 1685 .@"struct" => |struct_info| { 1686 inline for (struct_info.fields) |field| { 1687 if (!fieldEqual(field.type, @field(a, field.name), @field(b, field.name))) return false; 1688 } 1689 }, 1690 else => @compileError("fieldsEqual expects a struct"), 1691 } 1692 1693 return true; 1694 } 1695 1696 pub fn fieldEqual(comptime Field: type, a: Field, b: Field) bool { 1697 switch (@typeInfo(Field)) { 1698 .pointer => |pointer_info| { 1699 if (pointer_info.size == .slice and pointer_info.child == u8) { 1700 return std.mem.eql(u8, a, b); 1701 } else { 1702 @compileError("unsupported field type: " ++ @typeName(Field)); 1703 } 1704 }, 1705 .array => |array_info| { 1706 if (array_info.child == u8) { 1707 return std.mem.eql(u8, &a, &b); 1708 } else { 1709 @compileError("unsupported field type: " ++ @typeName(Field)); 1710 } 1711 }, 1712 .bool, .int, .@"enum" => return a == b, 1713 .@"struct" => return fieldsEqual(Field, a, b), 1714 .optional => |optional_info| { 1715 if (a) |a_value| { 1716 if (b) |b_value| return fieldEqual(optional_info.child, a_value, b_value); 1717 return false; 1718 } 1719 return b == null; 1720 }, 1721 else => @compileError("unsupported field type: " ++ @typeName(Field)), 1722 } 1723 } 1724 1725 fn readBytes( 1726 comptime DB: type, 1727 comptime hash_kind: hash.HashKind, 1728 allocator: std.mem.Allocator, 1729 map: DB.HashMap(.read_only), 1730 field_name: []const u8, 1731 ) ![]const u8 { 1732 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 1733 return try cursor.readBytesAlloc(allocator, null); 1734 } 1735 1736 fn readUint( 1737 comptime DB: type, 1738 comptime hash_kind: hash.HashKind, 1739 map: DB.HashMap(.read_only), 1740 field_name: []const u8, 1741 ) !u64 { 1742 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 1743 return try cursor.readUint(); 1744 } 1745 1746 fn readInt( 1747 comptime DB: type, 1748 comptime hash_kind: hash.HashKind, 1749 map: DB.HashMap(.read_only), 1750 field_name: []const u8, 1751 ) !i64 { 1752 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 1753 return try cursor.readInt(); 1754 } 1755 1756 fn readBool( 1757 comptime DB: type, 1758 comptime hash_kind: hash.HashKind, 1759 map: DB.HashMap(.read_only), 1760 field_name: []const u8, 1761 ) !bool { 1762 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 1763 var buffer: [5]u8 = undefined; 1764 const bytes = try cursor.readBytesObject(&buffer); 1765 const format_tag = bytes.format_tag orelse return error.InvalidFormatTag; 1766 if (!std.mem.eql(u8, &format_tag, "bl")) return error.InvalidFormatTag; 1767 if (std.mem.eql(u8, bytes.value, "true")) return true; 1768 if (std.mem.eql(u8, bytes.value, "false")) return false; 1769 return error.InvalidBool; 1770 } 1771 1772 // 1773 // writing to xitdb 1774 // 1775 1776 pub fn upsert( 1777 comptime T: type, 1778 comptime DB: type, 1779 comptime hash_kind: hash.HashKind, 1780 map: DB.HashMap(.read_write), 1781 event: T, 1782 ) !void { 1783 switch (@typeInfo(T)) { 1784 .@"struct" => |struct_info| { 1785 inline for (struct_info.fields) |field| { 1786 try upsertField(DB, hash_kind, map, field.name, field.type, @field(event, field.name)); 1787 } 1788 }, 1789 else => @compileError("upsert expects a struct"), 1790 } 1791 } 1792 1793 fn upsertField( 1794 comptime DB: type, 1795 comptime hash_kind: hash.HashKind, 1796 map: DB.HashMap(.read_write), 1797 comptime field_name: []const u8, 1798 comptime Field: type, 1799 value: Field, 1800 ) !void { 1801 const key = hash.hashInt(hash_kind, field_name); 1802 1803 switch (@typeInfo(Field)) { 1804 .pointer => |pointer_info| { 1805 if (pointer_info.size == .slice and pointer_info.child == u8) { 1806 try upsertBytes(DB, hash_kind, map, key, value); 1807 } else { 1808 @compileError("unsupported upsert field type: " ++ @typeName(Field)); 1809 } 1810 }, 1811 .array => |array_info| { 1812 if (array_info.child == u8) { 1813 try upsertBytes(DB, hash_kind, map, key, &value); 1814 } else { 1815 @compileError("unsupported upsert field type: " ++ @typeName(Field)); 1816 } 1817 }, 1818 .int => |int_info| switch (int_info.signedness) { 1819 .unsigned => { 1820 if (try map.getCursor(key)) |value_cursor| { 1821 if (try value_cursor.readUint() == value) { 1822 return; 1823 } 1824 } 1825 1826 try map.put(key, .{ .uint = value }); 1827 }, 1828 .signed => { 1829 if (try map.getCursor(key)) |value_cursor| { 1830 if (try value_cursor.readInt() == value) { 1831 return; 1832 } 1833 } 1834 1835 try map.put(key, .{ .int = value }); 1836 }, 1837 }, 1838 .bool => { 1839 const bytes = if (value) "true" else "false"; 1840 if (try map.getCursor(key)) |value_cursor| { 1841 var buffer: [5]u8 = undefined; 1842 const existing = try value_cursor.readBytesObject(&buffer); 1843 if (existing.format_tag) |format_tag| { 1844 if (std.mem.eql(u8, &format_tag, "bl") and std.mem.eql(u8, existing.value, bytes)) { 1845 return; 1846 } 1847 } 1848 } 1849 1850 try map.put(key, .{ .bytes_object = .{ .value = bytes, .format_tag = "bl".* } }); 1851 }, 1852 .@"enum" => try upsertBytes(DB, hash_kind, map, key, @tagName(value)), 1853 .optional => |optional_info| { 1854 // a missing key is null 1855 if (value) |child| { 1856 if (optional_info.child == []const u8) { 1857 try upsertBytes(DB, hash_kind, map, key, child); 1858 } else switch (@typeInfo(optional_info.child)) { 1859 .array => |array_info| { 1860 if (array_info.child != u8) @compileError("unsupported upsert field type: " ++ @typeName(Field)); 1861 try upsertBytes(DB, hash_kind, map, key, &child); 1862 }, 1863 .@"struct" => try upsert( 1864 optional_info.child, 1865 DB, 1866 hash_kind, 1867 try DB.HashMap(.read_write).init(try map.putCursor(key)), 1868 child, 1869 ), 1870 else => @compileError("unsupported upsert field type: " ++ @typeName(Field)), 1871 } 1872 } else { 1873 _ = try map.remove(key); 1874 } 1875 }, 1876 .@"struct" => try upsert( 1877 Field, 1878 DB, 1879 hash_kind, 1880 try DB.HashMap(.read_write).init(try map.putCursor(key)), 1881 value, 1882 ), 1883 else => @compileError("unsupported upsert field type: " ++ @typeName(Field)), 1884 } 1885 } 1886 1887 fn upsertBytes( 1888 comptime DB: type, 1889 comptime hash_kind: hash.HashKind, 1890 map: DB.HashMap(.read_write), 1891 key: hash.HashInt(hash_kind), 1892 value: []const u8, 1893 ) !void { 1894 var existing_cursor_maybe = try map.getCursor(key); 1895 if (existing_cursor_maybe) |*existing_cursor| { 1896 if (try bytesEqual(DB, existing_cursor, value)) { 1897 return; 1898 } 1899 } 1900 1901 var value_cursor = try map.putCursor(key); 1902 var write_buffer: [1024]u8 = undefined; 1903 var writer = try value_cursor.writer(&write_buffer); 1904 try writer.interface.writeAll(value); 1905 try writer.finish(); 1906 } 1907 1908 fn bytesEqual( 1909 comptime DB: type, 1910 cursor: *DB.Cursor(.read_only), 1911 value: []const u8, 1912 ) !bool { 1913 var read_buffer: [1024]u8 = undefined; 1914 var reader = try cursor.reader(&read_buffer); 1915 if (reader.size != value.len) { 1916 return false; 1917 } 1918 1919 var chunk_buffer: [1024]u8 = undefined; 1920 var offset: usize = 0; 1921 while (offset < value.len) { 1922 const chunk_len = @min(chunk_buffer.len, value.len - offset); 1923 try reader.interface.readSliceAll(chunk_buffer[0..chunk_len]); 1924 if (!std.mem.eql(u8, chunk_buffer[0..chunk_len], value[offset .. offset + chunk_len])) { 1925 return false; 1926 } 1927 offset += chunk_len; 1928 } 1929 1930 return true; 1931 } 1932 1933 // 1934 // CommitIterator 1935 // 1936 // differs from xit's ObjectIterator in a few ways: 1937 // 1938 // 1. iterates parents first, rather than children first 1939 // 2. `next` only returns the object id, not the full Object 1940 // 3. stores its temporary state in a file on disk to avoid OOMs 1941 // 1942 1943 pub fn CommitIterator(comptime repo_kind: rp.RepoKind, comptime repo_opts: rp.RepoOpts(repo_kind)) type { 1944 return struct { 1945 const DB = EventDB(repo_opts.hash); 1946 const db_name = "haxy-repo-events.db"; 1947 1948 repo_dir: std.Io.Dir, 1949 db_file: std.Io.File, 1950 db: *DB, 1951 // map of each object id to its children 1952 parent_to_children: DB.HashMap(.read_write), 1953 // tracking pending parents is necessary because a child can only 1954 // be returned after all of its unconsumed parents have been returned 1955 child_to_pending_parents: DB.HashMap(.read_write), 1956 // queue of object ids that are ready to be returned from `next` 1957 ready_queue: DB.ArrayList(.read_write), 1958 ready_queue_index: u64, 1959 // the object id at the tip of this branch 1960 newest_object_id: ?[hash.byteLen(repo_opts.hash)]u8, 1961 // the object id furthest back in the history that hasn't been consumed 1962 oldest_object_id: ?[hash.byteLen(repo_opts.hash)]u8, 1963 1964 pub fn init( 1965 state: rp.Repo(repo_kind, repo_opts).State(.read_only), 1966 io: std.Io, 1967 allocator: std.mem.Allocator, 1968 consumed_object_ids: DB.HashMap(.read_only), 1969 ref: rf.Ref, 1970 ) !CommitIterator(repo_kind, repo_opts) { 1971 const db_file = try state.core.repo_dir.createFile(io, db_name, .{ .truncate = true, .lock = .exclusive, .read = true }); 1972 errdefer { 1973 db_file.close(io); 1974 state.core.repo_dir.deleteFile(io, db_name) catch {}; 1975 } 1976 1977 const buffer_ptr = try allocator.create(std.Io.Writer.Allocating); 1978 errdefer allocator.destroy(buffer_ptr); 1979 1980 buffer_ptr.* = std.Io.Writer.Allocating.init(allocator); 1981 errdefer buffer_ptr.deinit(); 1982 1983 const db_ptr = try allocator.create(DB); 1984 errdefer allocator.destroy(db_ptr); 1985 // the db is scratch state that gets deleted after iteration, so 1986 // there is nothing worth fsyncing 1987 db_ptr.* = try DB.init(.{ .io = io, .file = db_file, .buffer = buffer_ptr, .fsync = false }); 1988 1989 const map = try DB.HashMap(.read_write).init(db_ptr.rootCursor()); 1990 1991 const parent_to_children_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "parent->children")); 1992 const parent_to_children = try DB.HashMap(.read_write).init(parent_to_children_cursor); 1993 1994 const child_to_pending_parents_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "child->pending-parents")); 1995 const child_to_pending_parents = try DB.HashMap(.read_write).init(child_to_pending_parents_cursor); 1996 1997 const ready_queue_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "ready-queue")); 1998 const ready_queue = try DB.ArrayList(.read_write).init(ready_queue_cursor); 1999 2000 var self = CommitIterator(repo_kind, repo_opts){ 2001 .repo_dir = state.core.repo_dir, 2002 .db_file = db_file, 2003 .db = db_ptr, 2004 .parent_to_children = parent_to_children, 2005 .child_to_pending_parents = child_to_pending_parents, 2006 .ready_queue = ready_queue, 2007 .ready_queue_index = 0, 2008 .newest_object_id = null, 2009 .oldest_object_id = null, 2010 }; 2011 errdefer self.deinit(io, allocator); 2012 2013 try self.collect(state, io, allocator, consumed_object_ids, ref); 2014 2015 return self; 2016 } 2017 2018 pub fn deinit(self: *CommitIterator(repo_kind, repo_opts), io: std.Io, allocator: std.mem.Allocator) void { 2019 self.db_file.close(io); 2020 self.db.core.memory.buffer.deinit(); 2021 allocator.destroy(self.db.core.memory.buffer); 2022 self.repo_dir.deleteFile(io, db_name) catch {}; 2023 allocator.destroy(self.db); 2024 } 2025 2026 pub fn next(self: *CommitIterator(repo_kind, repo_opts)) !?[hash.byteLen(repo_opts.hash)]u8 { 2027 if (self.ready_queue_index >= try self.ready_queue.count()) return null; 2028 2029 const oid = try readOidFromList(self.ready_queue.readOnly(), self.ready_queue_index); 2030 self.ready_queue_index += 1; 2031 2032 try self.enqueueChildren(&oid); 2033 return oid; 2034 } 2035 2036 fn collect( 2037 self: *CommitIterator(repo_kind, repo_opts), 2038 state: rp.Repo(repo_kind, repo_opts).State(.read_only), 2039 io: std.Io, 2040 allocator: std.mem.Allocator, 2041 consumed_object_ids: DB.HashMap(.read_only), 2042 ref: rf.Ref, 2043 ) !void { 2044 const map = try DB.HashMap(.read_write).init(self.db.rootCursor()); 2045 2046 const walk_queue_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "walk-queue")); 2047 const walk_queue = try DB.ArrayList(.read_write).init(walk_queue_cursor); 2048 var walk_queue_index: u64 = 0; 2049 2050 const seen_object_ids_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "seen-object-ids")); 2051 const seen_object_ids = try DB.HashSet(.read_write).init(seen_object_ids_cursor); 2052 2053 const head_oid_hex = (try rf.readRecur(repo_kind, repo_opts, state, io, .{ .ref = ref })) orelse return error.OidNotFound; 2054 var head_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 2055 _ = try std.fmt.hexToBytes(&head_oid, &head_oid_hex); 2056 try walk_queue.append(.{ .bytes = &head_oid }); 2057 2058 while (walk_queue_index < try walk_queue.count()) { 2059 const oid = try readOidFromList(walk_queue.readOnly(), walk_queue_index); 2060 const oid_int = hash.bytesToInt(repo_opts.hash, &oid); 2061 2062 walk_queue_index += 1; 2063 2064 // if we've seen this object id, skip it 2065 if (null != try seen_object_ids.getCursor(oid_int)) { 2066 continue; 2067 } 2068 2069 try seen_object_ids.put(oid_int, .{ .bytes = &oid }); 2070 2071 var commit_object = try obj.Object(repo_kind, repo_opts).init(state, io, allocator, &std.fmt.bytesToHex(oid, .lower)); 2072 defer commit_object.deinit(); 2073 2074 // if this object id has already been consumed, skip it 2075 if (null != try consumed_object_ids.getCursor(oid_int)) { 2076 continue; 2077 } 2078 2079 if (self.newest_object_id == null) self.newest_object_id = oid; 2080 self.oldest_object_id = oid; 2081 2082 const parent_oids = commit_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound; 2083 var pending_parent_count: u64 = 0; 2084 for (parent_oids) |*parent_oid| { 2085 var parent_oid_bytes: [hash.byteLen(repo_opts.hash)]u8 = undefined; 2086 _ = try std.fmt.hexToBytes(&parent_oid_bytes, parent_oid); 2087 const parent_oid_int = hash.bytesToInt(repo_opts.hash, &parent_oid_bytes); 2088 2089 // if this object id has already been consumed, skip it 2090 if (null != try consumed_object_ids.getCursor(parent_oid_int)) { 2091 continue; 2092 } 2093 2094 // if this object id hasn't already been seen, add it to the walk queue 2095 if (null == try seen_object_ids.getCursor(parent_oid_int)) { 2096 try walk_queue.append(.{ .bytes = &parent_oid_bytes }); 2097 } 2098 2099 const children_cursor = try self.parent_to_children.putCursor(parent_oid_int); 2100 const children = try DB.HashSet(.read_write).init(children_cursor); 2101 try children.put(oid_int, .{ .bytes = &oid }); 2102 pending_parent_count += 1; 2103 } 2104 2105 if (pending_parent_count == 0) { 2106 try self.ready_queue.append(.{ .bytes = &oid }); 2107 } else { 2108 try self.child_to_pending_parents.put(oid_int, .{ .uint = pending_parent_count }); 2109 } 2110 } 2111 } 2112 2113 fn enqueueChildren(self: *CommitIterator(repo_kind, repo_opts), oid: *const [hash.byteLen(repo_opts.hash)]u8) !void { 2114 const oid_int = hash.bytesToInt(repo_opts.hash, oid); 2115 const children_cursor = try self.parent_to_children.getCursor(oid_int) orelse return; 2116 const children = try DB.HashSet(.read_only).init(children_cursor); 2117 var children_iter = try children.iterator(); 2118 2119 while (try children_iter.next()) |*child_cursor| { 2120 const kv_pair = try child_cursor.readKeyValuePair(); 2121 const child_oid_int = kv_pair.hash; 2122 const pending_parent_cursor = try self.child_to_pending_parents.getCursor(child_oid_int) orelse return error.CursorNotFound; 2123 const pending_parent_count = try pending_parent_cursor.readUint(); 2124 2125 if (pending_parent_count <= 1) { 2126 _ = try self.child_to_pending_parents.remove(child_oid_int); 2127 const child_oid = hash.intToBytes(hash.HashInt(repo_opts.hash), child_oid_int); 2128 try self.ready_queue.append(.{ .bytes = &child_oid }); 2129 } else { 2130 try self.child_to_pending_parents.put(child_oid_int, .{ .uint = pending_parent_count - 1 }); 2131 } 2132 } 2133 } 2134 2135 fn readOidFromList(list: DB.ArrayList(.read_only), index: u64) ![hash.byteLen(repo_opts.hash)]u8 { 2136 const oid_cursor = try list.getCursor(index) orelse return error.CursorNotFound; 2137 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 2138 _ = try oid_cursor.readBytes(&oid); 2139 return oid; 2140 } 2141 }; 2142 } 2143