╭─────────────╮ │scroll to top│ ╰─────────────╯ ╭ notify.zig ───────────────────────────────────────────────────╮ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ╰───────────────────────────────────────────────────────────────╯
🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 ╭03aedad0..╮ 03aedad0.. haxy: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 │files │ commits refs discuss issues patches new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────────╯ (701) user
╭────╮ ╭ clone ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─╮ │http│ ssh │ │ ╰────╯ ╰ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─╯
.. embed/ event/ test/ ui/ command.zig diff3.zig event.zig find.zig fork.zig lib.zig main.zig main_wasm.zig markdown.zig ╭─────────────────────────────────╮ │notify.zig │ ╰─────────────────────────────────╯ patch.zig progress.zig push.zig search.zig search_commit.zig search_thread.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
445 server: evt.Host.Server, 446 comptime repo_opts: rp.RepoOpts(.xit), 447 repo: *rp.Repo(.xit, repo_opts), 448 moment_index: u64, 449 pusher_id: ?*const [evt.event_id_size]u8, 450 progress_ctx_maybe: ?repo_opts.ProgressCtx, 451 ) void { 452 const admin_repo_path = server.admin_repo_path orelse return; 453 const location = evt.parseRepoPath(server.users_dir, repo.core.work_path) orelse return; 454 deliverEach(repo_opts, io, allocator, server.users_dir, admin_repo_path, location, repo, moment_index, pusher_id, progress_ctx_maybe) catch |err| { 455 std.log.warn("failed to deliver notifications: {s}", .{@errorName(err)}); 456 }; 457 } 458 459 // a transaction's notifs, coalesced per user and thread 460 const Inboxes = struct { 461 arena: *std.heap.ArenaAllocator, 462 admin_moment: evt.AdminDB.HashMap(.read_only), 463 owner_moment: UserDB.HashMap(.read_only), 464 repo: evt.Repo.Record, 465 repo_id: *const [evt.event_id_size]u8, 466 // whether each user is active and can read the repo 467 readers: std.AutoHashMapUnmanaged([evt.event_id_size]u8, bool) = .empty, 468 by_user: std.AutoArrayHashMapUnmanaged([evt.event_id_size]u8, std.ArrayList(Notif)) = .empty, 469 470 fn add(self: *Inboxes, user_id: [evt.event_id_size]u8, notif: Notif) !void { 471 if (!try self.canRead(user_id)) return; 472 const entry = try self.by_user.getOrPut(self.arena.allocator(), user_id); 473 if (!entry.found_existing) entry.value_ptr.* = .empty; 474 // events arrive newest first, so a thread keeps its latest child 475 for (entry.value_ptr.items) |existing| { 476 if (std.mem.eql(u8, &existing.thread_id, &notif.thread_id)) return; 477 } 478 try entry.value_ptr.append(self.arena.allocator(), notif); 479 } 480 481 fn canRead(self: *Inboxes, user_id: [evt.event_id_size]u8) !bool { 482 if (self.readers.get(user_id)) |can_read| return can_read; 483 const can_read = if (try evt.User.readById(evt.AdminDB, evt.admin_repo_opts.hash, self.admin_moment, self.arena, &user_id)) |user| 484 !user.removed and try evt.Repo.roleOf(UserDB, hash_kind, self.owner_moment, self.arena, self.repo, self.repo_id, user_id) != null 485 else 486 false; 487 try self.readers.put(self.arena.allocator(), user_id, can_read); 488 return can_read; 489 } 490 }; 491 492 fn deliverEach( 493 comptime repo_opts: rp.RepoOpts(.xit), 494 io: std.Io, 495 allocator: std.mem.Allocator, 496 users_dir: []const u8, 497 admin_repo_path: []const u8, 498 location: evt.RepoLocation, 499 repo: *rp.Repo(.xit, repo_opts), 500 moment_index: u64, 501 pusher_id: ?*const [evt.event_id_size]u8, 502 progress_ctx_maybe: ?repo_opts.ProgressCtx, 503 ) !void { 504 const DB = evt.EventDB(repo_opts.hash); 505 const haxy_moment = try evt.currentMoment(repo_opts, repo); 506 // the transaction's events, newest first 507 const ids = (try evt.momentEventIds(DB, repo_opts.hash, haxy_moment, moment_index)) orelse return; 508 509 var owner_repo = (try evt.openUserRepo(io, allocator, users_dir, &location.owner_id)) orelse return; 510 defer owner_repo.deinit(io, allocator); 511 const owner_moment = (try evt.userMoment(&owner_repo)) orelse return; 512 // the owner's plain data, which holds the tier index 513 const owner_latest = try owner_repo.core.latestMoment(); 514 515 var arena = std.heap.ArenaAllocator.init(allocator); 516 defer arena.deinit(); 517 518 const repo_record = (try evt.Repo.readById(UserDB, hash_kind, owner_moment, &arena, &location.repo_id)) orelse return; 519 if (repo_record.removed) return; 520 521 var admin_repo = try rp.Repo(.xit, evt.admin_repo_opts).open(io, allocator, .{ .path = admin_repo_path }); 522 defer admin_repo.deinit(io, allocator); 523 524 const admin_moment = try evt.currentMoment(evt.admin_repo_opts, &admin_repo); 525 const owner = (try evt.User.readById(evt.AdminDB, evt.admin_repo_opts.hash, admin_moment, &arena, &location.owner_id)) orelse return; 526 527 var inboxes: Inboxes = .{ 528 .arena = &arena, 529 .admin_moment = admin_moment, 530 .owner_moment = owner_moment, 531 .repo = repo_record, 532 .repo_id = &location.repo_id, 533 }; 534 535 const kinds_cursor = try haxy_moment.getCursor(hash.hashInt(repo_opts.hash, evt.event_index_key)) orelse return error.CursorNotFound; 536 const kinds = try DB.HashMap(.read_only).init(kinds_cursor); 537 538 var iter = try ids.iteratorFromIndex(0); 539 while (try iter.next()) |kv_cursor| { 540 var cursor = kv_cursor; 541 const kv_pair = try cursor.readKeyValuePair(); 542 var order_key: [@sizeOf(u64) + evt.event_id_size]u8 = undefined; 543 _ = try kv_pair.key_cursor.readBytes(&order_key); 544 const id = order_key[@sizeOf(u64)..].*; 545 const order = std.math.maxInt(u64) - std.mem.readInt(u64, order_key[0..@sizeOf(u64)], .big); 546 547 // edits keep the created order, so only creations notify 548 const Created = struct { thread_id: [evt.event_id_size * 2]u8, actor_email: ?[]const u8 }; 549 const created: Created = switch ((try evt.readEventKind(repo_opts.hash, kinds, &id)) orelse continue) { 550 inline .comment, .assign => |kind| blk: { 551 const Fields = struct { removed: bool, author_email: ?[]const u8, created_order: u64, event: struct { thread_id: [evt.event_id_size * 2]u8 } }; 552 const T = if (kind == .comment) evt.Comment else evt.Assignment; 553 const record = (try evt.readRecordSubset(T, Fields, DB, repo_opts.hash, haxy_moment, &arena, &id)) orelse continue; 554 if (record.removed or record.created_order != order) continue; 555 break :blk .{ .thread_id = record.event.thread_id, .actor_email = record.author_email }; 556 }, 557 else => continue, 558 }; 559 560 const thread_id = try evt.parseEventId(&created.thread_id); 561 const Titled = struct { event: struct { title: []const u8 } }; 562 const thread_kind = (try evt.readEventKind(repo_opts.hash, kinds, &thread_id)) orelse continue; 563 const thread = (switch (thread_kind) { 564 .issue => try evt.readRecordSubset(evt.Issue, Titled, DB, repo_opts.hash, haxy_moment, &arena, &thread_id), 565 .patch => try evt.readRecordSubset(evt.Patch, Titled, DB, repo_opts.hash, haxy_moment, &arena, &thread_id), 566 .discuss => try evt.readRecordSubset(evt.Discussion, Titled, DB, repo_opts.hash, haxy_moment, &arena, &thread_id), 567 else => null, 568 }) orelse continue; 569 570 const notif: Notif = .{ 571 .repo_id = location.repo_id, 572 .owner_name = owner.event.name, 573 .repo_name = repo_record.event.name, 574 .thread_id = thread_id, 575 .thread_kind = thread_kind, 576 .title = thread.event.title, 577 .last_child_id = id, 578 }; 579 580 // a user hears of no comment or assignment they made or pushed 581 const actor_id = if (created.actor_email) |email| try evt.User.readIdByEmail(evt.AdminDB, evt.admin_repo_opts.hash, admin_moment, email) else null; 582 for (try readNotified(owner_latest, &arena, &idOf(&location.repo_id, &thread_id))) |user_id| { 583 if (actor_id) |actor| if (std.mem.eql(u8, &actor, &user_id)) continue; 584 if (pusher_id) |pusher| if (std.mem.eql(u8, pusher, &user_id)) continue; 585 try inboxes.add(user_id, notif); 586 } 587 } 588 589 if (inboxes.by_user.count() == 0) return; 590 progress.report(repo_opts, io, progress_ctx_maybe, .{ .text = "Delivering notifications" }); 591 progress.report(repo_opts, io, progress_ctx_maybe, .{ .start = .{ .kind = .writing_patch, .estimated_total_items = inboxes.by_user.count() } }); 592 for (inboxes.by_user.keys(), inboxes.by_user.values()) |*user_id, notifs| { 593 deliver(io, allocator, users_dir, user_id, notifs.items) catch |err| { 594 std.log.warn("failed to deliver notifications: {s}", .{@errorName(err)}); 595 }; 596 progress.report(repo_opts, io, progress_ctx_maybe, .{ .complete_one = .writing_patch }); 597 } 598 progress.report(repo_opts, io, progress_ctx_maybe, .{ .end = .writing_patch }); 599 } 600