╭─────────────╮ │scroll to top│ ╰─────────────╯ ╭ event.zig ────────────────────────────────────────────────────╮ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ╰───────────────────────────────────────────────────────────────╯
🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 ╭bc93d7b1..╮ bc93d7b1.. see-em: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 │files │ commits refs discuss issues new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────────╯ (111) user
╭────╮ ╭ clone ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ╮ │http│ ssh │ │ ╰────╯ ╰ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ╯
.. embed/ event/ test/ ui/ command.zig ╭─────────────────────────────────╮ │event.zig │ ╰─────────────────────────────────╯ lib.zig main.zig main_wasm.zig serve.zig serve_ssh.zig serve_ssh_protocol.zig test.zig testnet.zig try.zig ui.zig web.zig
375 } 376 377 fn readBool( 378 comptime DB: type, 379 comptime hash_kind: hash.HashKind, 380 map: DB.HashMap(.read_only), 381 field_name: []const u8, 382 ) !bool { 383 const cursor = try map.getCursor(hash.hashInt(hash_kind, field_name)) orelse return error.NotFound; 384 var buffer: [5]u8 = undefined; 385 const bytes = try cursor.readBytesObject(&buffer); 386 const format_tag = bytes.format_tag orelse return error.InvalidFormatTag; 387 if (!std.mem.eql(u8, &format_tag, "bl")) return error.InvalidFormatTag; 388 if (std.mem.eql(u8, bytes.value, "true")) return true; 389 if (std.mem.eql(u8, bytes.value, "false")) return false; 390 return error.InvalidBool; 391 } 392 393 // 394 // writing to xitdb 395 // 396 397 pub fn upsert( 398 comptime T: type, 399 comptime DB: type, 400 comptime hash_kind: hash.HashKind, 401 map: DB.HashMap(.read_write), 402 event: T, 403 ) !void { 404 switch (@typeInfo(T)) { 405 .@"struct" => |struct_info| { 406 inline for (struct_info.fields) |field| { 407 try upsertField(DB, hash_kind, map, field.name, field.type, @field(event, field.name)); 408 } 409 }, 410 else => @compileError("upsert expects a struct"), 411 } 412 } 413 414 fn mergeChangedMapEntries( 415 comptime DB: type, 416 target: DB.HashMap(.read_write), 417 parent: DB.HashMap(.read_only), 418 baseline: DB.HashMap(.read_only), 419 comptime is_top_level: bool, 420 ) !void { 421 var parent_iter = try parent.iterator(); 422 while (try parent_iter.next()) |kv_pair_cursor| { 423 const kv_pair = try kv_pair_cursor.readKeyValuePair(); 424 425 const baseline_value_cursor = try baseline.getCursor(kv_pair.hash) orelse { 426 if ((try target.getCursor(kv_pair.hash)) != null) { 427 return error.MergeConflict; 428 } 429 430 try target.put(kv_pair.hash, .{ .slot = kv_pair.value_cursor.slot() }); 431 continue; 432 }; 433 434 const tag = kv_pair.value_cursor.slot().tag; 435 436 if (tag != baseline_value_cursor.slot().tag) { 437 return error.UnexpectedTag; 438 } 439 440 if (kv_pair.value_cursor.slot().value == baseline_value_cursor.slot().value) { 441 continue; 442 } 443 444 if (tag == .hash_map) { 445 const target_existing_cursor = try target.getCursor(kv_pair.hash) orelse return error.MergeConflict; 446 if (target_existing_cursor.slot().tag != .hash_map) { 447 return error.MergeConflict; 448 } 449 450 const target_child_cursor = try target.putCursor(kv_pair.hash); 451 const target_child = try DB.HashMap(.read_write).init(target_child_cursor); 452 const parent_child = try DB.HashMap(.read_only).init(kv_pair.value_cursor); 453 const baseline_child = try DB.HashMap(.read_only).init(baseline_value_cursor); 454 try mergeChangedMapEntries(DB, target_child, parent_child, baseline_child, false); 455 } else if (is_top_level) { 456 // the moment-index is a top-level key in the haxy moment 457 // and is not a map. we don't want to do any merge of this. 458 continue; 459 } else { 460 const target_value_cursor = try target.getCursor(kv_pair.hash) orelse return error.MergeConflict; 461 if (target_value_cursor.slot().value != baseline_value_cursor.slot().value) { 462 return error.MergeConflict; 463 } 464 465 try target.put(kv_pair.hash, .{ .slot = kv_pair.value_cursor.slot() }); 466 } 467 } 468 } 469 470 fn upsertField( 471 comptime DB: type, 472 comptime hash_kind: hash.HashKind, 473 map: DB.HashMap(.read_write), 474 comptime field_name: []const u8, 475 comptime Field: type, 476 value: Field, 477 ) !void { 478 const key = hash.hashInt(hash_kind, field_name); 479 480 switch (@typeInfo(Field)) { 481 .pointer => |pointer_info| { 482 if (pointer_info.size == .slice and pointer_info.child == u8) { 483 try upsertBytes(DB, hash_kind, map, key, value); 484 } else { 485 @compileError("unsupported upsert field type: " ++ @typeName(Field)); 486 } 487 }, 488 .array => |array_info| { 489 if (array_info.child == u8) { 490 try upsertBytes(DB, hash_kind, map, key, &value); 491 } else { 492 @compileError("unsupported upsert field type: " ++ @typeName(Field)); 493 } 494 }, 495 .int => |int_info| switch (int_info.signedness) { 496 .unsigned => { 497 if (try map.getCursor(key)) |value_cursor| { 498 if (try value_cursor.readUint() == value) { 499 return; 500 } 501 } 502 503 try map.put(key, .{ .uint = value }); 504 }, 505 .signed => { 506 if (try map.getCursor(key)) |value_cursor| { 507 if (try value_cursor.readInt() == value) { 508 return; 509 } 510 } 511 512 try map.put(key, .{ .int = value }); 513 }, 514 }, 515 .bool => { 516 const bytes = if (value) "true" else "false"; 517 if (try map.getCursor(key)) |value_cursor| { 518 var buffer: [5]u8 = undefined; 519 const existing = try value_cursor.readBytesObject(&buffer); 520 if (existing.format_tag) |format_tag| { 521 if (std.mem.eql(u8, &format_tag, "bl") and std.mem.eql(u8, existing.value, bytes)) { 522 return; 523 } 524 } 525 } 526 527 try map.put(key, .{ .bytes_object = .{ .value = bytes, .format_tag = "bl".* } }); 528 }, 529 else => @compileError("unsupported upsert field type: " ++ @typeName(Field)), 530 } 531 } 532 533 fn upsertBytes( 534 comptime DB: type, 535 comptime hash_kind: hash.HashKind, 536 map: DB.HashMap(.read_write), 537 key: hash.HashInt(hash_kind), 538 value: []const u8, 539 ) !void { 540 var existing_cursor_maybe = try map.getCursor(key); 541 if (existing_cursor_maybe) |*existing_cursor| { 542 if (try bytesEqual(DB, existing_cursor, value)) { 543 return; 544 } 545 } 546 547 var value_cursor = try map.putCursor(key); 548 var write_buffer: [1024]u8 = undefined; 549 var writer = try value_cursor.writer(&write_buffer); 550 try writer.interface.writeAll(value); 551 try writer.finish(); 552 } 553 554 fn bytesEqual( 555 comptime DB: type, 556 cursor: *DB.Cursor(.read_only), 557 value: []const u8, 558 ) !bool { 559 var read_buffer: [1024]u8 = undefined; 560 var reader = try cursor.reader(&read_buffer); 561 if (reader.size != value.len) { 562 return false; 563 } 564 565 var chunk_buffer: [1024]u8 = undefined; 566 var offset: usize = 0; 567 while (offset < value.len) { 568 const chunk_len = @min(chunk_buffer.len, value.len - offset); 569 try reader.interface.readSliceAll(chunk_buffer[0..chunk_len]); 570 if (!std.mem.eql(u8, chunk_buffer[0..chunk_len], value[offset .. offset + chunk_len])) { 571 return false; 572 } 573 offset += chunk_len; 574 } 575 576 return true; 577 } 578 579 // 580 // CommitIterator 581 // 582 // differs from xit's ObjectIterator in a few ways: 583 // 584 // 1. iterates parents first, rather than children first 585 // 2. `next` only returns the object id, not the full Object 586 // 3. stores its temporary state in a file on disk to avoid OOMs 587 // 588 589 pub fn CommitIterator(comptime repo_opts: rp.RepoOpts(.xit)) type { 590 return struct { 591 const DB = rp.Repo(.xit, repo_opts).DB; 592 const db_name = "haxy-repo-events.db"; 593 594 repo_dir: std.Io.Dir, 595 db_file: std.Io.File, 596 db: *DB, 597 // map of each object id to its children 598 parent_to_children: DB.HashMap(.read_write), 599 // tracking pending parents is necessary because a child can only 600 // be returned after all of its unconsumed parents have been returned 601 child_to_pending_parents: DB.HashMap(.read_write), 602 // queue of object ids that are ready to be returned from `next` 603 ready_queue: DB.ArrayList(.read_write), 604 ready_queue_index: u64, 605 // the object id at the tip of this branch 606 newest_object_id: ?[hash.byteLen(repo_opts.hash)]u8, 607 // the object id furthest back in the history that hasn't been consumed 608 oldest_object_id: ?[hash.byteLen(repo_opts.hash)]u8, 609 610 pub fn init( 611 state: rp.Repo(.xit, repo_opts).State(.read_only), 612 io: std.Io, 613 allocator: std.mem.Allocator, 614 consumed_object_ids: DB.HashMap(.read_only), 615 ref: rf.Ref, 616 ) !CommitIterator(repo_opts) { 617 const db_file = try state.core.repo_dir.createFile(io, db_name, .{ .truncate = true, .lock = .exclusive, .read = true }); 618 errdefer { 619 db_file.close(io); 620 state.core.repo_dir.deleteFile(io, db_name) catch {}; 621 } 622 623 const buffer_ptr = try allocator.create(std.Io.Writer.Allocating); 624 errdefer allocator.destroy(buffer_ptr); 625 626 buffer_ptr.* = std.Io.Writer.Allocating.init(allocator); 627 errdefer buffer_ptr.deinit(); 628 629 const db_ptr = try allocator.create(DB); 630 errdefer allocator.destroy(db_ptr); 631 db_ptr.* = try DB.init(.{ .io = io, .file = db_file, .buffer = buffer_ptr }); 632 633 const map = try DB.HashMap(.read_write).init(db_ptr.rootCursor()); 634 635 const parent_to_children_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "parent->children")); 636 const parent_to_children = try DB.HashMap(.read_write).init(parent_to_children_cursor); 637 638 const child_to_pending_parents_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "child->pending-parents")); 639 const child_to_pending_parents = try DB.HashMap(.read_write).init(child_to_pending_parents_cursor); 640 641 const ready_queue_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "ready-queue")); 642 const ready_queue = try DB.ArrayList(.read_write).init(ready_queue_cursor); 643 644 var self = CommitIterator(repo_opts){ 645 .repo_dir = state.core.repo_dir, 646 .db_file = db_file, 647 .db = db_ptr, 648 .parent_to_children = parent_to_children, 649 .child_to_pending_parents = child_to_pending_parents, 650 .ready_queue = ready_queue, 651 .ready_queue_index = 0, 652 .newest_object_id = null, 653 .oldest_object_id = null, 654 }; 655 errdefer self.deinit(io, allocator); 656 657 try self.collect(state, io, allocator, consumed_object_ids, ref); 658 659 return self; 660 } 661 662 pub fn deinit(self: *CommitIterator(repo_opts), io: std.Io, allocator: std.mem.Allocator) void { 663 self.db_file.close(io); 664 self.db.core.memory.buffer.deinit(); 665 allocator.destroy(self.db.core.memory.buffer); 666 self.repo_dir.deleteFile(io, db_name) catch {}; 667 allocator.destroy(self.db); 668 } 669 670 pub fn next(self: *CommitIterator(repo_opts)) !?[hash.byteLen(repo_opts.hash)]u8 { 671 if (self.ready_queue_index >= try self.ready_queue.count()) return null; 672 673 const oid = try readOidFromList(self.ready_queue.readOnly(), self.ready_queue_index); 674 self.ready_queue_index += 1; 675 676 try self.enqueueChildren(&oid); 677 return oid; 678 } 679 680 fn collect( 681 self: *CommitIterator(repo_opts), 682 state: rp.Repo(.xit, repo_opts).State(.read_only), 683 io: std.Io, 684 allocator: std.mem.Allocator, 685 consumed_object_ids: DB.HashMap(.read_only), 686 ref: rf.Ref, 687 ) !void { 688 const map = try DB.HashMap(.read_write).init(self.db.rootCursor()); 689 690 const walk_queue_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "walk-queue")); 691 const walk_queue = try DB.ArrayList(.read_write).init(walk_queue_cursor); 692 var walk_queue_index: u64 = 0; 693 694 const seen_object_ids_cursor = try map.putCursor(hash.hashInt(repo_opts.hash, "seen-object-ids")); 695 const seen_object_ids = try DB.HashSet(.read_write).init(seen_object_ids_cursor); 696 697 const head_oid_hex = (try rf.readRecur(.xit, repo_opts, state, io, .{ .ref = ref })) orelse return error.OidNotFound; 698 var head_oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 699 _ = try std.fmt.hexToBytes(&head_oid, &head_oid_hex); 700 try walk_queue.append(.{ .bytes = &head_oid }); 701 702 while (walk_queue_index < try walk_queue.count()) { 703 const oid = try readOidFromList(walk_queue.readOnly(), walk_queue_index); 704 const oid_int = hash.bytesToInt(repo_opts.hash, &oid); 705 706 walk_queue_index += 1; 707 708 // if we've seen this object id, skip it 709 if (null != try seen_object_ids.getCursor(oid_int)) { 710 continue; 711 } 712 713 try seen_object_ids.put(oid_int, .{ .bytes = &oid }); 714 715 var commit_object = try obj.Object(.xit, repo_opts, .full).init(state, io, allocator, &std.fmt.bytesToHex(oid, .lower)); 716 defer commit_object.deinit(); 717 718 // if this object id has already been consumed, skip it 719 if (null != try consumed_object_ids.getCursor(oid_int)) { 720 continue; 721 } 722 723 if (self.newest_object_id == null) self.newest_object_id = oid; 724 self.oldest_object_id = oid; 725 726 const parent_oids = commit_object.content.commit.metadata.parent_oids orelse return error.ParentOidsNotFound; 727 var pending_parent_count: u64 = 0; 728 for (parent_oids) |*parent_oid| { 729 var parent_oid_bytes: [hash.byteLen(repo_opts.hash)]u8 = undefined; 730 _ = try std.fmt.hexToBytes(&parent_oid_bytes, parent_oid); 731 const parent_oid_int = hash.bytesToInt(repo_opts.hash, &parent_oid_bytes); 732 733 // if this object id has already been consumed, skip it 734 if (null != try consumed_object_ids.getCursor(parent_oid_int)) { 735 continue; 736 } 737 738 // if this object id hasn't already been seen, add it to the walk queue 739 if (null == try seen_object_ids.getCursor(parent_oid_int)) { 740 try walk_queue.append(.{ .bytes = &parent_oid_bytes }); 741 } 742 743 const children_cursor = try self.parent_to_children.putCursor(parent_oid_int); 744 const children = try DB.HashSet(.read_write).init(children_cursor); 745 try children.put(oid_int, .{ .bytes = &oid }); 746 pending_parent_count += 1; 747 } 748 749 if (pending_parent_count == 0) { 750 try self.ready_queue.append(.{ .bytes = &oid }); 751 } else { 752 try self.child_to_pending_parents.put(oid_int, .{ .uint = pending_parent_count }); 753 } 754 } 755 } 756 757 fn enqueueChildren(self: *CommitIterator(repo_opts), oid: *const [hash.byteLen(repo_opts.hash)]u8) !void { 758 const oid_int = hash.bytesToInt(repo_opts.hash, oid); 759 const children_cursor = try self.parent_to_children.getCursor(oid_int) orelse return; 760 const children = try DB.HashSet(.read_only).init(children_cursor); 761 var children_iter = try children.iterator(); 762 763 while (try children_iter.next()) |*child_cursor| { 764 const kv_pair = try child_cursor.readKeyValuePair(); 765 const child_oid_int = kv_pair.hash; 766 const pending_parent_cursor = try self.child_to_pending_parents.getCursor(child_oid_int) orelse return error.CursorNotFound; 767 const pending_parent_count = try pending_parent_cursor.readUint(); 768 769 if (pending_parent_count <= 1) { 770 _ = try self.child_to_pending_parents.remove(child_oid_int); 771 const child_oid = hash.intToBytes(hash.HashInt(repo_opts.hash), child_oid_int); 772 try self.ready_queue.append(.{ .bytes = &child_oid }); 773 } else { 774 try self.child_to_pending_parents.put(child_oid_int, .{ .uint = pending_parent_count - 1 }); 775 } 776 } 777 } 778 779 fn readOidFromList(list: DB.ArrayList(.read_only), index: u64) ![hash.byteLen(repo_opts.hash)]u8 { 780 const oid_cursor = try list.getCursor(index) orelse return error.CursorNotFound; 781 var oid: [hash.byteLen(repo_opts.hash)]u8 = undefined; 782 _ = try oid_cursor.readBytes(&oid); 783 return oid; 784 } 785 }; 786 } 787