╭─────────────╮ │scroll to top│ ╰─────────────╯ ╭ notify.zig ───────────────────────────────────────────────────╮ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ╰───────────────────────────────────────────────────────────────╯
🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 ╭master╮ master haxy: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 │files │ commits refs discuss issues patches new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────╯ (716) user
╭ find file ─ ─ ─ ─ ─ ─ ─ ─ ─ ─╮ ╭────╮ ╭ 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
66 // run `ctx` in a transaction on the user's repo 67 fn transact(io: std.Io, allocator: std.mem.Allocator, users_dir: []const u8, user_id: *const [evt.event_id_size]u8, ctx: anytype) !void { 68 var user_repo = (try evt.openUserRepo(io, allocator, users_dir, user_id)) orelse return error.NotFound; 69 defer user_repo.deinit(io, allocator); 70 71 try user_repo.core.db_file.lock(io, .exclusive); 72 defer user_repo.core.db_file.unlock(io); 73 74 const history = try UserDB.ArrayList(.read_write).init(user_repo.core.db.rootCursor()); 75 history.appendContext(.{ .slot = try history.getSlot(-1) }, ctx) catch |err| switch (err) { 76 error.CancelTransaction => {}, 77 else => |e| return e, 78 }; 79 } 80 81 // a stored notif's time, or null when the user has none for the id 82 fn readTime(arena: *std.heap.ArenaAllocator, records: UserDB.HashMap(.read_only), id: *const [evt.event_id_size]u8) !?u64 { 83 const record_cursor = try records.getCursor(hash.hashInt(hash_kind, id)) orelse return null; 84 return (try evt.read(struct { time: u64 }, UserDB, hash_kind, arena, try UserDB.HashMap(.read_only).init(record_cursor))).time; 85 } 86 87 // drop a notif from the map and both sets 88 fn removeNotif(moment: UserDB.HashMap(.read_write), records: UserDB.HashMap(.read_write), id: *const [evt.event_id_size]u8, time: u64) !void { 89 const order_key = evt.orderKeyDesc(time, id); 90 _ = try records.remove(hash.hashInt(hash_kind, id)); 91 _ = try (try UserDB.SortedSet(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, recent_notif_id_set_key)))).remove(&order_key); 92 _ = try (try UserDB.SortedSet(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, unread_notif_id_set_key)))).remove(&order_key); 93 } 94 95 // upsert one user's batch as unread and newest, pruning past the cap 96 pub fn deliver(io: std.Io, allocator: std.mem.Allocator, users_dir: []const u8, user_id: *const [evt.event_id_size]u8, notifs: []const Notif) !void { 97 var arena = std.heap.ArenaAllocator.init(allocator); 98 defer arena.deinit(); 99 100 const Ctx = struct { 101 arena: *std.heap.ArenaAllocator, 102 notifs: []const Notif, 103 time: u64, 104 105 pub fn run(ctx: @This(), cursor: *UserDB.Cursor(.read_write)) !void { 106 const moment = try UserDB.HashMap(.read_write).init(cursor.*); 107 const records = try UserDB.HashMap(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, notif_map_key))); 108 const recent = try UserDB.SortedSet(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, recent_notif_id_set_key))); 109 const unread = try UserDB.SortedSet(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, unread_notif_id_set_key))); 110 111 for (ctx.notifs) |notif| { 112 const id = idOf(&notif.repo_id, &notif.thread_id); 113 if (try readTime(ctx.arena, records.readOnly(), &id)) |old_time| { 114 const old_key = evt.orderKeyDesc(old_time, &id); 115 _ = try recent.remove(&old_key); 116 _ = try unread.remove(&old_key); 117 } 118 var stored = notif; 119 stored.time = ctx.time; 120 try evt.upsert(Notif, UserDB, hash_kind, try UserDB.HashMap(.read_write).init(try records.putCursor(hash.hashInt(hash_kind, &id))), stored); 121 const order_key = evt.orderKeyDesc(ctx.time, &id); 122 try recent.put(&order_key); 123 try unread.put(&order_key); 124 } 125 126 // the oldest go past the cap 127 var count = try recent.count(); 128 while (count > max_notifs) : (count -= 1) { 129 const oldest = (try recent.getIndexKeyValuePair(@intCast(count - 1))) orelse unreachable; 130 var order_key: [@sizeOf(u64) + evt.event_id_size]u8 = undefined; 131 _ = try oldest.key_cursor.readBytes(&order_key); 132 const id = order_key[@sizeOf(u64)..].*; 133 const time = std.math.maxInt(u64) - std.mem.readInt(u64, order_key[0..@sizeOf(u64)], .big); 134 try removeNotif(moment, records, &id, time); 135 } 136 } 137 }; 138 139 try transact(io, allocator, users_dir, user_id, Ctx{ 140 .arena = &arena, 141 .notifs = notifs, 142 .time = @intCast(std.Io.Timestamp.now(io, .real).toSeconds()), 143 }); 144 } 145 146 // mark a notif read, unless it was bumped past `last_child_id` since the user saw it 147 pub fn markRead( 148 io: std.Io, 149 allocator: std.mem.Allocator, 150 users_dir: []const u8, 151 user_id: *const [evt.event_id_size]u8, 152 notif_id: *const [evt.event_id_size]u8, 153 last_child_id: *const [evt.event_id_size]u8, 154 ) !void { 155 var arena = std.heap.ArenaAllocator.init(allocator); 156 defer arena.deinit(); 157 158 const Ctx = struct { 159 arena: *std.heap.ArenaAllocator, 160 notif_id: *const [evt.event_id_size]u8, 161 last_child_id: *const [evt.event_id_size]u8, 162 163 pub fn run(ctx: @This(), cursor: *UserDB.Cursor(.read_write)) !void { 164 const moment = try UserDB.HashMap(.read_write).init(cursor.*); 165 const records_cursor = try moment.readOnly().getCursor(hash.hashInt(hash_kind, notif_map_key)) orelse return error.CancelTransaction; 166 const record_cursor = try (try UserDB.HashMap(.read_only).init(records_cursor)).getCursor(hash.hashInt(hash_kind, ctx.notif_id)) orelse return error.CancelTransaction; 167 const notif = try evt.read(Notif, UserDB, hash_kind, ctx.arena, try UserDB.HashMap(.read_only).init(record_cursor)); 168 if (!std.mem.eql(u8, &notif.last_child_id, ctx.last_child_id)) return error.CancelTransaction; 169 170 const unread = try UserDB.SortedSet(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, unread_notif_id_set_key))); 171 if (!try unread.remove(&evt.orderKeyDesc(notif.time, ctx.notif_id))) return error.CancelTransaction; 172 } 173 }; 174 175 try transact(io, allocator, users_dir, user_id, Ctx{ .arena = &arena, .notif_id = notif_id, .last_child_id = last_child_id }); 176 } 177 178 // remove a notif. the thread's next bump recreates it. 179 pub fn dismiss(io: std.Io, allocator: std.mem.Allocator, users_dir: []const u8, user_id: *const [evt.event_id_size]u8, notif_id: *const [evt.event_id_size]u8) !void { 180 var arena = std.heap.ArenaAllocator.init(allocator); 181 defer arena.deinit(); 182 183 const Ctx = struct { 184 arena: *std.heap.ArenaAllocator, 185 notif_id: *const [evt.event_id_size]u8, 186 187 pub fn run(ctx: @This(), cursor: *UserDB.Cursor(.read_write)) !void { 188 const moment = try UserDB.HashMap(.read_write).init(cursor.*); 189 const records = try UserDB.HashMap(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, notif_map_key))); 190 const time = (try readTime(ctx.arena, records.readOnly(), ctx.notif_id)) orelse return error.CancelTransaction; 191 try removeNotif(moment, records, ctx.notif_id, time); 192 } 193 }; 194 195 try transact(io, allocator, users_dir, user_id, Ctx{ .arena = &arena, .notif_id = notif_id }); 196 } 197 198 // up to `count` notifs from `start`, newest first. `moment` is the user repo's latest. 199 pub fn readRecent(moment: UserDB.HashMap(.read_only), arena: *std.heap.ArenaAllocator, start: u64, count: usize) ![]const Item { 200 var items: std.ArrayList(Item) = .empty; 201 const recent_cursor = try moment.getCursor(hash.hashInt(hash_kind, recent_notif_id_set_key)) orelse return items.items; 202 const recent = try UserDB.SortedSet(.read_only).init(recent_cursor); 203 204 var iter = try recent.iteratorFromIndex(start); 205 while (items.items.len < count) { 206 const kv_cursor = (try iter.next()) orelse break; 207 const id = try evt.readOrderKeyId(UserDB, kv_cursor); 208 try items.append(arena.allocator(), (try readItem(moment, arena, &id)) orelse return error.NotFound); 209 } 210 return items.items; 211 } 212 213 // the notif `id` names with its read state, or null when the user has none. `moment` is the user repo's latest. 214 pub fn readItem(moment: UserDB.HashMap(.read_only), arena: *std.heap.ArenaAllocator, id: *const [evt.event_id_size]u8) !?Item { 215 const records_cursor = try moment.getCursor(hash.hashInt(hash_kind, notif_map_key)) orelse return null; 216 const record_cursor = try (try UserDB.HashMap(.read_only).init(records_cursor)).getCursor(hash.hashInt(hash_kind, id)) orelse return null; 217 const notif = try evt.read(Notif, UserDB, hash_kind, arena, try UserDB.HashMap(.read_only).init(record_cursor)); 218 const unread_cursor = try moment.getCursor(hash.hashInt(hash_kind, unread_notif_id_set_key)) orelse return .{ .notif = notif, .unread = false }; 219 const unread = try (try UserDB.SortedSet(.read_only).init(unread_cursor)).contains(&evt.orderKeyDesc(notif.time, id)); 220 return .{ .notif = notif, .unread = unread }; 221 } 222 223 // a user repo's latest moment, or null before its first transaction 224 pub fn latestMoment(user_repo: *rp.Repo(.xit, evt.user_repo_opts)) !?UserDB.HashMap(.read_only) { 225 return user_repo.core.latestMoment() catch |err| switch (err) { 226 error.DatabaseIsEmpty => null, 227 else => |e| e, 228 }; 229 } 230 231 // how many notifs are unread. `moment` is the user repo's latest. 232 pub fn unreadCount(moment: UserDB.HashMap(.read_only)) !u64 { 233 const unread_cursor = try moment.getCursor(hash.hashInt(hash_kind, unread_notif_id_set_key)) orelse return 0; 234 return try (try UserDB.SortedSet(.read_only).init(unread_cursor)).count(); 235 } 236 237 // notify `user_id` of the thread `thread_id` in the repo at `location` by 238 // choice, unless a tier already holds them or the subscribed tier is full 239 pub fn subscribe( 240 io: std.Io, 241 allocator: std.mem.Allocator, 242 users_dir: []const u8, 243 user_id: *const [evt.event_id_size]u8, 244 location: evt.RepoLocation, 245 thread_id: *const [evt.event_id_size]u8, 246 author: evt.CommitAuthor, 247 ) !void { 248 { 249 var owner_repo = (try evt.openUserRepo(io, allocator, users_dir, &location.owner_id)) orelse return error.NotFound; 250 defer owner_repo.deinit(io, allocator); 251 const moment = try owner_repo.core.latestMoment(); 252 const notif_id = idOf(&location.repo_id, thread_id); 253 if (try isNotified(moment, &notif_id, user_id)) return; 254 if (try tierFull(moment, &notif_id, .subscribed)) return; 255 } 256 try writeMode(io, allocator, users_dir, user_id, &location, thread_id, .subscribed, author); 257 } 258 259 // never notify `user_id` of the thread `thread_id` in the repo at `location`, dropping its notif 260 pub fn mute( 261 io: std.Io, 262 allocator: std.mem.Allocator, 263 users_dir: []const u8, 264 user_id: *const [evt.event_id_size]u8, 265 location: evt.RepoLocation, 266 thread_id: *const [evt.event_id_size]u8, 267 author: evt.CommitAuthor, 268 ) !void { 269 try writeMode(io, allocator, users_dir, user_id, &location, thread_id, .muted, author); 270 try dismiss(io, allocator, users_dir, user_id, &idOf(&location.repo_id, thread_id)); 271 } 272 273 // write the user's notify event for a thread, then move them to its tier in the owner's index 274 pub fn writeMode( 275 io: std.Io, 276 allocator: std.mem.Allocator, 277 users_dir: []const u8, 278 user_id: *const [evt.event_id_size]u8, 279 location: *const evt.RepoLocation, 280 thread_id: *const [evt.event_id_size]u8, 281 mode: evt.Notify.Mode, 282 author: evt.CommitAuthor, 283 ) !void { 284 { 285 var user_repo = (try evt.openUserRepo(io, allocator, users_dir, user_id)) orelse return error.NotFound; 286 defer user_repo.deinit(io, allocator); 287 try evt.consume(.{ .server = .{ .users_dir = users_dir } }, .user, .xit, evt.user_repo_opts, io, allocator, &user_repo, evt.events_ref, &[_]evt.EventWithId{.{ 288 .id = std.fmt.bytesToHex(evt.Notify.idOf(&location.repo_id, thread_id), .lower), 289 .timestamp = @intCast(std.Io.Timestamp.now(io, .real).toSeconds()), 290 .author = author, 291 .event = .{ .notify = .{ 292 .owner_id = std.fmt.bytesToHex(location.owner_id, .lower), 293 .repo_id = std.fmt.bytesToHex(location.repo_id, .lower), 294 .thread_id = std.fmt.bytesToHex(thread_id.*, .lower), 295 .mode = mode, 296 } }, 297 }}); 298 } 299 indexMode(io, allocator, users_dir, location, thread_id, user_id, mode); 300 } 301 302 // move `user_id` to `mode`'s tier on the thread in its repo owner's index, or 303 // out of every tier for a mute, logging rather than returning a failure 304 fn indexMode( 305 io: std.Io, 306 allocator: std.mem.Allocator, 307 users_dir: []const u8, 308 location: *const evt.RepoLocation, 309 thread_id: *const [evt.event_id_size]u8, 310 user_id: *const [evt.event_id_size]u8, 311 mode: evt.Notify.Mode, 312 ) void { 313 const Ctx = struct { 314 notif_id: [evt.event_id_size]u8, 315 user_id: *const [evt.event_id_size]u8, 316 mode: evt.Notify.Mode, 317 318 pub fn run(ctx: @This(), cursor: *UserDB.Cursor(.read_write)) !void { 319 const moment = try UserDB.HashMap(.read_write).init(cursor.*); 320 inline for (comptime std.enums.values(evt.Notify.Mode)) |tier_mode| { 321 if (comptime tierOf(tier_mode)) |tier| { 322 const sets = try UserDB.HashMap(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, tier.map_key))); 323 const users = try UserDB.SortedSet(.read_write).init(try sets.putCursor(hash.hashInt(hash_kind, &ctx.notif_id))); 324 if (tier_mode == ctx.mode) try users.put(ctx.user_id) else _ = try users.remove(ctx.user_id); 325 } 326 } 327 } 328 }; 329 330 transact(io, allocator, users_dir, &location.owner_id, Ctx{ 331 .notif_id = idOf(&location.repo_id, thread_id), 332 .user_id = user_id, 333 .mode = mode, 334 }) catch |err| { 335 std.log.warn("failed to index thread notify mode: {s}", .{@errorName(err)}); 336 }; 337 } 338 339 // put the users `events` show taking part in their threads in a tier, the 340 // threads' authors and the assignees as involved and the commenters as 341 // commented. logs rather than returning a failure. 342 pub fn subscribeParticipants(io: std.Io, allocator: std.mem.Allocator, server: evt.Host.Server, repo_path: []const u8, events: []const evt.EventWithId) void { 343 const admin_repo_path = server.admin_repo_path orelse return; 344 const location = evt.parseRepoPath(server.users_dir, repo_path) orelse return; 345 subscribeEach(io, allocator, server.users_dir, admin_repo_path, location, events) catch |err| { 346 std.log.warn("failed to subscribe thread participants: {s}", .{@errorName(err)}); 347 }; 348 } 349 350 fn subscribeEach(io: std.Io, allocator: std.mem.Allocator, users_dir: []const u8, admin_repo_path: []const u8, location: evt.RepoLocation, events: []const evt.EventWithId) !void { 351 var admin_repo = try rp.Repo(.xit, evt.admin_repo_opts).open(io, allocator, .{ .path = admin_repo_path }); 352 defer admin_repo.deinit(io, allocator); 353 const admin_moment = try evt.currentMoment(evt.admin_repo_opts, &admin_repo); 354 var arena = std.heap.ArenaAllocator.init(allocator); 355 defer arena.deinit(); 356 357 for (events) |event| { 358 const thread_hex: [evt.event_id_size * 2]u8, const email: []const u8, const mode: evt.Notify.Mode = switch (event.event) { 359 inline .discuss, .issue, .patch => |thread| if (thread != null) .{ event.id, event.author.email, .involved } else continue, 360 .comment => |comment_maybe| if (comment_maybe) |comment| .{ comment.thread_id, event.author.email, .commented } else continue, 361 .assign => |assign_maybe| if (assign_maybe) |assign| .{ assign.thread_id, assign.email, .involved } else continue, 362 else => continue, 363 }; 364 const user_id = (try evt.User.readIdByEmail(evt.AdminDB, evt.admin_repo_opts.hash, admin_moment, email)) orelse continue; 365 const thread_id = try evt.parseEventId(&thread_hex); 366 // a mute stands, and a record only ever moves up to involved 367 { 368 var user_repo = (try evt.openUserRepo(io, allocator, users_dir, &user_id)) orelse continue; 369 defer user_repo.deinit(io, allocator); 370 if (try evt.userMoment(&user_repo)) |moment| { 371 if (try evt.Notify.readMode(UserDB, hash_kind, moment, &arena, &location.repo_id, &thread_id)) |existing| { 372 if (mode != .involved or existing == .involved or existing == .muted) continue; 373 } 374 } 375 } 376 // a full commented tier takes nobody new, so a later comment checks again 377 if (mode == .commented) { 378 var owner_repo = (try evt.openUserRepo(io, allocator, users_dir, &location.owner_id)) orelse return error.NotFound; 379 defer owner_repo.deinit(io, allocator); 380 if (try tierFull(try owner_repo.core.latestMoment(), &idOf(&location.repo_id, &thread_id), .commented)) continue; 381 } 382 // the write's own author, which for an assignment is the assigner 383 try writeMode(io, allocator, users_dir, &user_id, &location, &thread_id, mode, event.author); 384 } 385 } 386 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 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; 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 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 // 436 // delivery 437 // 438 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, 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