🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 9cd00aac.. ╭9cd00aac..╮ see-em: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 files │diff │ refs discuss issues new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────────╯ user
╭──────────╮ │← previous│ ╰──────────╯ ╭ changed ─────╮ │src/notify.zig│ ╰──────────────╯ ╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │360 -// the thread `notif_id` names' subscriber set, or null before its first subscriber. │ │361 -// `moment` is its repo owner's user repo's latest. │ │362 -fn subscriberSet(moment: UserDB.HashMap(.read_only), notif_id: *const [evt.event_id_size]u8) !?UserDB.SortedSet(.read_only) { │ │363 - const sets_cursor = try moment.getCursor(hash.hashInt(hash_kind, subscriber_map_key)) orelse return null; │ │364 - const subscribers_cursor = try (try UserDB.HashMap(.read_only).init(sets_cursor)).getCursor(hash.hashInt(hash_kind, notif_id)) orelse return null; │ │365 - return try UserDB.SortedSet(.read_only).init(subscribers_cursor); │ │387 +// the thread `notif_id` names' set under a tier's `map_key`, or null before its │ │388 +// first user. `moment` is its repo owner's user repo's latest. │ │389 +fn tierSet(moment: UserDB.HashMap(.read_only), notif_id: *const [evt.event_id_size]u8, map_key: []const u8) !?UserDB.SortedSet(.read_only) { │ │390 + const sets_cursor = try moment.getCursor(hash.hashInt(hash_kind, map_key)) orelse return null; │ │391 + const users_cursor = try (try UserDB.HashMap(.read_only).init(sets_cursor)).getCursor(hash.hashInt(hash_kind, notif_id)) orelse return null; │ │392 + return try UserDB.SortedSet(.read_only).init(users_cursor); │ │393 } │ │394 │ │368 -// the users subscribed to the thread `notif_id` names. `moment` is its repo owner's user repo's latest. │ │369 -pub fn readSubscribers(moment: UserDB.HashMap(.read_only), arena: *std.heap.ArenaAllocator, notif_id: *const [evt.event_id_size]u8) ![]const [evt.event_id_size]u8 {│ │395 +// the users in the tiers of the thread `notif_id` names. `moment` is its repo owner's user repo's latest. │ │396 +pub fn readNotified(moment: UserDB.HashMap(.read_only), arena: *std.heap.ArenaAllocator, notif_id: *const [evt.event_id_size]u8) ![]const [evt.event_id_size]u8 { │ │397 var user_ids: std.ArrayList([evt.event_id_size]u8) = .empty; │ │371 - const subscribers = (try subscriberSet(moment, notif_id)) orelse return user_ids.items; │ │372 - var iter = try subscribers.iteratorFromIndex(0); │ │373 - while (try iter.next()) |kv_cursor| { │ │374 - var cursor = kv_cursor; │ │375 - const kv_pair = try cursor.readKeyValuePair(); │ │376 - var user_id: [evt.event_id_size]u8 = undefined; │ │377 - _ = try kv_pair.key_cursor.readBytes(&user_id); │ │378 - try user_ids.append(arena.allocator(), user_id); │ │398 + inline for (comptime std.enums.values(evt.Notify.Mode)) |mode| { │ │399 + if (comptime tierOf(mode)) |tier| { │ │400 + if (try tierSet(moment, notif_id, tier.map_key)) |users| { │ │401 + var iter = try users.iteratorFromIndex(0); │ │402 + while (try iter.next()) |kv_cursor| { │ │403 + var cursor = kv_cursor; │ │404 + const kv_pair = try cursor.readKeyValuePair(); │ │405 + var user_id: [evt.event_id_size]u8 = undefined; │ │406 + _ = try kv_pair.key_cursor.readBytes(&user_id); │ │407 + try user_ids.append(arena.allocator(), user_id); │ │408 + } │ │409 + } │ │410 + } │ │411 } │ │412 return user_ids.items; │ │413 } │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │414 │ │383 -// whether `user_id` is subscribed to the thread `notif_id` names. `moment` is its repo owner's user repo's latest. │ │384 -pub fn isSubscribed(moment: UserDB.HashMap(.read_only), notif_id: *const [evt.event_id_size]u8, user_id: *const [evt.event_id_size]u8) !bool { │ │385 - const subscribers = (try subscriberSet(moment, notif_id)) orelse return false; │ │386 - return try subscribers.contains(user_id); │ │415 +// whether a tier of the thread `notif_id` names holds `user_id`. `moment` is its repo owner's user repo's latest. │ │416 +pub fn isNotified(moment: UserDB.HashMap(.read_only), notif_id: *const [evt.event_id_size]u8, user_id: *const [evt.event_id_size]u8) !bool { │ │417 + inline for (comptime std.enums.values(evt.Notify.Mode)) |mode| { │ │418 + if (comptime tierOf(mode)) |tier| { │ │419 + if (try tierSet(moment, notif_id, tier.map_key)) |users| { │ │420 + if (try users.contains(user_id)) return true; │ │421 + } │ │422 + } │ │423 + } │ │424 + return false; │ │425 } │ │426 │ │427 +// whether `mode`'s tier on the thread `notif_id` names is at its cap. `moment` is its repo owner's user repo's latest. │ │428 +pub fn tierFull(moment: UserDB.HashMap(.read_only), notif_id: *const [evt.event_id_size]u8, comptime mode: evt.Notify.Mode) !bool { │ │429 + const tier = comptime tierOf(mode) orelse @compileError("a mute has no tier"); │ │430 + const cap = tier.cap orelse return false; │ │431 + const users = (try tierSet(moment, notif_id, tier.map_key)) orelse return false; │ │432 + return try users.count() >= cap; │ │433 +} │ │434 + │ │435 // │ │390 -// delivery worker │ │436 +// delivery │ │437 // │ │392 - │ │393 -// repos with events the worker hasn't looked at, deduped │ │394 -pub const DirtyRepos = struct { │ │395 - allocator: std.mem.Allocator, │ │396 - mutex: std.Io.Mutex = .init, │ │397 - // signaled when the set goes non-empty │ │398 - condition: std.Io.Condition = .init, │ │399 - set: std.AutoArrayHashMapUnmanaged(evt.RepoLocation, void) = .empty, │ │400 - │ │401 - pub fn deinit(self: *DirtyRepos) void { │ │402 - self.set.deinit(self.allocator); │ │403 - } │ │438 │ │405 - pub fn mark(self: *DirtyRepos, io: std.Io, location: evt.RepoLocation) void { │ │406 - self.mutex.lockUncancelable(io); │ │407 - defer self.mutex.unlock(io); │ │408 - self.set.put(self.allocator, location, {}) catch |err| { │ │409 - std.log.warn("failed to queue repo notifications: {s}", .{@errorName(err)}); │ │410 - return; │ │411 - }; │ │412 - self.condition.signal(io); │ │413 - } │ │414 -}; │ │415 - │ │416 -// deliver notifs for each repo marked dirty, until canceled │ │417 -pub fn run( │ │418 - dirty_repos: *DirtyRepos, │ │439 +// deliver the notifs of the comments and assignments the repo transaction at │ │440 +// `moment_index` created, logging rather than returning a failure. │ │441 +// `pusher_id` is the user who pushed it, if a push did. │ │442 +pub fn deliverTransaction( │ │443 io: std.Io, │ │444 allocator: std.mem.Allocator, │ │421 - users_dir: []const u8, │ │422 - admin_repo_path: []const u8, │ │423 -) std.Io.Cancelable!void { │ │424 - while (true) { │ │425 - var batch = blk: { │ │426 - try dirty_repos.mutex.lock(io); │ │427 - defer dirty_repos.mutex.unlock(io); │ │428 - while (dirty_repos.set.count() == 0) try dirty_repos.condition.wait(io, &dirty_repos.mutex); │ │429 - const batch = dirty_repos.set; │ │430 - dirty_repos.set = .empty; │ │431 - break :blk batch; │ │432 - }; │ │433 - defer batch.deinit(dirty_repos.allocator); │ │434 - │ │435 - for (batch.keys()) |location| { │ │436 - notifyRepo(io, allocator, users_dir, admin_repo_path, location) catch |err| { │ │437 - std.log.warn("failed to deliver notifications: {s}", .{@errorName(err)}); │ │438 - }; │ │439 - } │ │440 - } │ │441 -} │ │442 - │ │443 -pub fn notifyRepo(io: std.Io, allocator: std.mem.Allocator, users_dir: []const u8, admin_repo_path: []const u8, location: evt.RepoLocation) !void { │ │444 - const path = try evt.repoPath(allocator, users_dir, &location.owner_id, &location.repo_id); │ │445 - defer allocator.free(path); │ │446 - var any_repo = try rp.AnyRepo(.xit, .{}).open(io, allocator, .{ .path = path, .require_repo_root = true }); │ │447 - defer any_repo.deinit(io, allocator); │ │448 - switch (any_repo) { │ │449 - inline else => |*repo| try notifyRepoEvents(repo.self_repo_opts, io, allocator, users_dir, admin_repo_path, location, repo), │ │450 - } │ │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 │ │453 -// a round's notifs, coalesced per user and thread │ │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), │ ╰────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭─────────────────────────────────────────────────────────────────────────────────────────────╮ │489 } │ │490 }; │ │491 │ │486 -// deliver the comments and assignments created since the repo's cursor, then advance it│ │487 -fn notifyRepoEvents( │ │492 +fn deliverEach( │ │493 comptime repo_opts: rp.RepoOpts(.xit), │ │494 io: std.Io, │ │495 allocator: std.mem.Allocator, │ ╰─────────────────────────────────────────────────────────────────────────────────────────────╯ ╭───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │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); │ │497 - const haxy_moment = evt.currentMoment(repo_opts, repo) catch |err| switch (err) { │ │498 - error.NotFound => return, │ │499 - else => |e| return e, │ │500 - }; │ │501 - const index_cursor = try haxy_moment.getCursor(hash.hashInt(repo_opts.hash, evt.moment_index_key)) orelse return error.CursorNotFound;│ │502 - const current = try index_cursor.readUint(); │ │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); │ ╰───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭───────────────────────────────────────────────────────────────────────────────────────────────────────╮ │506 - │ │507 - // a missing cursor starts from the repo's first transaction, and a rewound history resets it │ │508 - const first = if (try readLastNotified(&owner_repo, &location.repo_id)) |last| blk: { │ │509 - if (current < last) return writeLastNotified(io, &owner_repo, &location.repo_id, current);│ │510 - if (current == last) return; │ │511 - break :blk last + 1; │ │512 - } else 0; │ │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 │ │517 - deliver: { │ │518 - const owner_moment = (try evt.userMoment(&owner_repo)) orelse break :deliver; │ │519 - // the owner's plain data, which holds the subscriber index │ │520 - const owner_latest = try owner_repo.core.latestMoment(); │ │521 - const repo_record = (try evt.Repo.readById(UserDB, hash_kind, owner_moment, &arena, &location.repo_id)) orelse break :deliver; │ │522 - if (repo_record.removed) break :deliver; │ │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 │ │524 - var admin_repo = try rp.Repo(.xit, evt.admin_repo_opts).open(io, allocator, .{ .path = admin_repo_path }); │ │525 - defer admin_repo.deinit(io, allocator); │ │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 │ │527 - const admin_moment = try evt.currentMoment(evt.admin_repo_opts, &admin_repo); │ │528 - const owner = (try evt.User.readById(evt.AdminDB, evt.admin_repo_opts.hash, admin_moment, &arena, &location.owner_id)) orelse break :deliver; │ │529 - │ │530 - var inboxes: Inboxes = .{ │ │531 - .arena = &arena, │ │532 - .admin_moment = admin_moment, │ │533 - .owner_moment = owner_moment, │ │534 - .repo = repo_record, │ │535 - .repo_id = &location.repo_id, │ │536 - }; │ │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 │ │538 - const kinds_cursor = try haxy_moment.getCursor(hash.hashInt(repo_opts.hash, evt.event_index_key)) orelse return error.CursorNotFound; │ │539 - const kinds = try DB.HashMap(.read_only).init(kinds_cursor); │ │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 │ │541 - // the events each transaction since the cursor consumed, newest first │ │542 - var index = current + 1; │ │543 - while (index > first) { │ │544 - index -= 1; │ │545 - const ids = (try evt.momentEventIds(DB, repo_opts.hash, haxy_moment, index)) orelse continue; │ │546 - var iter = try ids.iteratorFromIndex(0); │ │547 - while (try iter.next()) |kv_cursor| { │ │548 - var cursor = kv_cursor; │ │549 - const kv_pair = try cursor.readKeyValuePair(); │ │550 - var order_key: [@sizeOf(u64) + evt.event_id_size]u8 = undefined; │ │551 - _ = try kv_pair.key_cursor.readBytes(&order_key); │ │552 - const id = order_key[@sizeOf(u64)..].*; │ │553 - const order = std.math.maxInt(u64) - std.mem.readInt(u64, order_key[0..@sizeOf(u64)], .big); │ │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 │ │555 - // edits keep the created order, so only creations notify │ │556 - const Created = struct { thread_id: [evt.event_id_size * 2]u8, actor_email: ?[]const u8 }; │ │557 - const created: Created = switch ((try evt.readEventKind(repo_opts.hash, kinds, &id)) orelse continue) { │ │558 - inline .comment, .assign => |kind| blk: { │ │559 - const Fields = struct { removed: bool, author_email: ?[]const u8, created_order: u64, event: struct { thread_id: [evt.event_id_size * 2]u8 } };│ │560 - const T = if (kind == .comment) evt.Comment else evt.Assignment; │ │561 - const record = (try evt.readRecordSubset(T, Fields, DB, repo_opts.hash, haxy_moment, &arena, &id)) orelse continue; │ │562 - if (record.removed or record.created_order != order) continue; │ │563 - break :blk .{ .thread_id = record.event.thread_id, .actor_email = record.author_email }; │ │564 - }, │ │565 - else => continue, │ │566 - }; │ │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 │ │568 - const thread_id = try evt.parseEventId(&created.thread_id); │ │569 - const Titled = struct { event: struct { title: []const u8 } }; │ │570 - const thread_kind = (try evt.readEventKind(repo_opts.hash, kinds, &thread_id)) orelse continue; │ │571 - const thread = (switch (thread_kind) { │ │572 - .issue => try evt.readRecordSubset(evt.Issue, Titled, DB, repo_opts.hash, haxy_moment, &arena, &thread_id), │ │573 - .patch => try evt.readRecordSubset(evt.Patch, Titled, DB, repo_opts.hash, haxy_moment, &arena, &thread_id), │ │574 - .discuss => try evt.readRecordSubset(evt.Discussion, Titled, DB, repo_opts.hash, haxy_moment, &arena, &thread_id), │ │575 - else => null, │ │576 - }) orelse continue; │ │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 │ │578 - const notif: Notif = .{ │ │579 - .repo_id = location.repo_id, │ │580 - .owner_name = owner.event.name, │ │581 - .repo_name = repo_record.event.name, │ │582 - .thread_id = thread_id, │ │583 - .thread_kind = thread_kind, │ │584 - .title = thread.event.title, │ │585 - .last_child_id = id, │ │586 - }; │ │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 │ │588 - // subscribers hear of every comment and assignment but their own │ │589 - const actor_id = if (created.actor_email) |email| try evt.User.readIdByEmail(evt.AdminDB, evt.admin_repo_opts.hash, admin_moment, email) else null; │ │590 - for (try readSubscribers(owner_latest, &arena, &idOf(&location.repo_id, &thread_id))) |user_id| { │ │591 - if (actor_id) |actor| if (std.mem.eql(u8, &actor, &user_id)) continue; │ │592 - try inboxes.add(user_id, notif); │ │593 - } │ │594 - } │ │595 - } │ │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 │ │597 - // a failure leaves the cursor, so the repo's next round redelivers │ │598 - var failed = false; │ │599 - for (inboxes.by_user.keys(), inboxes.by_user.values()) |*user_id, notifs| { │ │600 - deliver(io, allocator, users_dir, user_id, notifs.items) catch |err| { │ │601 - std.log.warn("failed to deliver notifications: {s}", .{@errorName(err)}); │ │602 - failed = true; │ │603 - }; │ │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 } │ │605 - if (failed) return error.DeliveryFailed; │ │587 } │ │607 - │ │608 - try writeLastNotified(io, &owner_repo, &location.repo_id, current); │ │609 -} │ │610 - │ │611 -fn readLastNotified(owner_repo: *rp.Repo(.xit, evt.user_repo_opts), repo_id: *const [evt.event_id_size]u8) !?u64 { │ │612 - const map_cursor = try (try owner_repo.core.latestMoment()).getCursor(hash.hashInt(hash_kind, last_notified_map_key)) orelse return null; │ │613 - const map = try UserDB.HashMap(.read_only).init(map_cursor); │ │614 - const cursor = try map.getCursor(hash.hashInt(hash_kind, repo_id)) orelse return null; │ │615 - return try cursor.readUint(); │ │616 -} │ │588 │ │618 -fn writeLastNotified(io: std.Io, owner_repo: *rp.Repo(.xit, evt.user_repo_opts), repo_id: *const [evt.event_id_size]u8, moment_index: u64) !void { │ │619 - const Ctx = struct { │ │620 - repo_id: *const [evt.event_id_size]u8, │ │621 - moment_index: u64, │ │622 - │ │623 - pub fn run(ctx: @This(), cursor: *UserDB.Cursor(.read_write)) !void { │ │624 - const moment = try UserDB.HashMap(.read_write).init(cursor.*); │ │625 - const map = try UserDB.HashMap(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, last_notified_map_key))); │ │626 - try map.put(hash.hashInt(hash_kind, ctx.repo_id), .{ .uint = ctx.moment_index }); │ │627 - } │ │628 - }; │ │629 - │ │630 - try transactIn(io, owner_repo, Ctx{ .repo_id = repo_id, .moment_index = moment_index }); │ │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 │ ╰────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭ changed ────╮ │src/patch.zig│ ╰─────────────╯ ╭─────────────────────────────────────────────────────╮ │11 const fork = @import("fork.zig"); │ │12 const serve_common = @import("serve_common.zig");│ │13 const progress = @import("./progress.zig"); │ │14 +const notify = @import("notify.zig"); │ │15 │ │16 // the action a patch merge records │ │17 pub const merge_undo_action = "haxy/merge"; │ ╰─────────────────────────────────────────────────────╯ ╭────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ 95 const history = try DB.ArrayList(.read_write).init(repo.core.db.rootCursor()); │ │ 96 try history.appendContext(.{ .slot = try history.getSlot(-1) }, Ctx{ .repo = repo, .io = io, .arena = &arena, .id = id, .patch = patch, .expected_patch = expected_patch, .author = author, .checks = &checks, .first_parent = first_parent });│ │ 97 } │ │ 97 - if (host == .server) evt.bumpRepoActivityAt(io, allocator, host.server, repo.core.work_path); │ │ 98 + if (host == .server) { │ │ 99 + notify.subscribeParticipants(io, allocator, host.server, repo.core.work_path, &.{.{ .id = id, .author = author, .event = .{ .patch = patch } }}); │ │100 + evt.bumpRepoActivityAt(io, allocator, host.server, repo.core.work_path); │ │101 + } │ │102 } │ │103 } │ │104 │ ╰────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │132 { │ │133 const parent = if (try rf.readRecur(.xit, repo_opts, state.readOnly(), io, .{ .ref = evt.events_ref }) == null) first_parent else null; │ │134 try evt.commitEvents(.xit, repo_opts, state, io, arena.child_allocator, evt.events_ref, &events, parent); │ │131 - if (!try evt.consumeInTransaction(.repo, .xit, repo_opts, state, &state.core.db, moment, io, arena.child_allocator, evt.events_ref)) return error.CancelTransaction;│ │135 + _ = try evt.consumeInTransaction(.repo, .xit, repo_opts, state, &state.core.db, moment, io, arena.child_allocator, evt.events_ref); │ │136 } │ │137 │ │138 // save mergeability before finishing the patch transaction │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭─────────────────────────────────────────────────────────────────────────────────╮ │944 if (existing == null) patch.status = .open; │ │945 │ │946 // write the published patch and its forker to the target repository│ │947 + const events = &[_]evt.EventWithId{.{ │ │948 + .id = input.id, │ │949 + .timestamp = input.timestamp, │ │950 + .author = input.author, │ │951 + .event = .{ .patch = patch }, │ │952 + }}; │ │953 { │ │954 var checks: MergeCheckInputs(repo_opts) = .{}; │ │955 defer checks.deinit(io, allocator); │ ╰─────────────────────────────────────────────────────────────────────────────────╯ ╭──────╮ │next →│ ╰──────╯