╭─────────────╮ │scroll to top│ ╰─────────────╯ ╭ notify.zig ───────────────────────────────────────────────────╮ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ╰───────────────────────────────────────────────────────────────╯
🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 ╭d8748222..╮ d8748222.. haxy: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 │files │ commits refs discuss issues patches new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────────╯ (685) 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
68 history.appendContext(.{ .slot = try history.getSlot(-1) }, ctx) catch |err| switch (err) { 69 error.CancelTransaction => {}, 70 else => |e| return e, 71 }; 72 } 73 74 // a stored notif's time, or null when the user has none for the id 75 fn readTime(arena: *std.heap.ArenaAllocator, records: UserDB.HashMap(.read_only), id: *const [evt.event_id_size]u8) !?u64 { 76 const record_cursor = try records.getCursor(hash.hashInt(hash_kind, id)) orelse return null; 77 return (try evt.read(struct { time: u64 }, UserDB, hash_kind, arena, try UserDB.HashMap(.read_only).init(record_cursor))).time; 78 } 79 80 // drop a notif from the map and both sets 81 fn removeNotif(moment: UserDB.HashMap(.read_write), records: UserDB.HashMap(.read_write), id: *const [evt.event_id_size]u8, time: u64) !void { 82 const order_key = evt.orderKeyDesc(time, id); 83 _ = try records.remove(hash.hashInt(hash_kind, id)); 84 _ = try (try UserDB.SortedSet(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, recent_notif_id_set_key)))).remove(&order_key); 85 _ = try (try UserDB.SortedSet(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, unread_notif_id_set_key)))).remove(&order_key); 86 } 87 88 // upsert one user's batch as unread and newest, pruning past the cap 89 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 { 90 var arena = std.heap.ArenaAllocator.init(allocator); 91 defer arena.deinit(); 92 93 const Ctx = struct { 94 arena: *std.heap.ArenaAllocator, 95 notifs: []const Notif, 96 time: u64, 97 98 pub fn run(ctx: @This(), cursor: *UserDB.Cursor(.read_write)) !void { 99 const moment = try UserDB.HashMap(.read_write).init(cursor.*); 100 const records = try UserDB.HashMap(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, notif_map_key))); 101 const recent = try UserDB.SortedSet(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, recent_notif_id_set_key))); 102 const unread = try UserDB.SortedSet(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, unread_notif_id_set_key))); 103 104 for (ctx.notifs) |notif| { 105 const id = idOf(&notif.repo_id, &notif.thread_id); 106 if (try readTime(ctx.arena, records.readOnly(), &id)) |old_time| { 107 const old_key = evt.orderKeyDesc(old_time, &id); 108 _ = try recent.remove(&old_key); 109 _ = try unread.remove(&old_key); 110 } 111 var stored = notif; 112 stored.time = ctx.time; 113 try evt.upsert(Notif, UserDB, hash_kind, try UserDB.HashMap(.read_write).init(try records.putCursor(hash.hashInt(hash_kind, &id))), stored); 114 const order_key = evt.orderKeyDesc(ctx.time, &id); 115 try recent.put(&order_key); 116 try unread.put(&order_key); 117 } 118 119 // the oldest go past the cap 120 var count = try recent.count(); 121 while (count > max_notifs) : (count -= 1) { 122 const oldest = (try recent.getIndexKeyValuePair(@intCast(count - 1))) orelse unreachable; 123 var order_key: [@sizeOf(u64) + evt.event_id_size]u8 = undefined; 124 _ = try oldest.key_cursor.readBytes(&order_key); 125 const id = order_key[@sizeOf(u64)..].*; 126 const time = std.math.maxInt(u64) - std.mem.readInt(u64, order_key[0..@sizeOf(u64)], .big); 127 try removeNotif(moment, records, &id, time); 128 } 129 } 130 }; 131 132 try transact(io, allocator, users_dir, user_id, Ctx{ 133 .arena = &arena, 134 .notifs = notifs, 135 .time = @intCast(std.Io.Timestamp.now(io, .real).toSeconds()), 136 }); 137 } 138 139 // mark a notif read, unless it was bumped past `last_child_id` since the user saw it 140 pub fn markRead( 141 io: std.Io, 142 allocator: std.mem.Allocator, 143 users_dir: []const u8, 144 user_id: *const [evt.event_id_size]u8, 145 notif_id: *const [evt.event_id_size]u8, 146 last_child_id: *const [evt.event_id_size]u8, 147 ) !void { 148 var arena = std.heap.ArenaAllocator.init(allocator); 149 defer arena.deinit(); 150 151 const Ctx = struct { 152 arena: *std.heap.ArenaAllocator, 153 notif_id: *const [evt.event_id_size]u8, 154 last_child_id: *const [evt.event_id_size]u8, 155 156 pub fn run(ctx: @This(), cursor: *UserDB.Cursor(.read_write)) !void { 157 const moment = try UserDB.HashMap(.read_write).init(cursor.*); 158 const records_cursor = try moment.readOnly().getCursor(hash.hashInt(hash_kind, notif_map_key)) orelse return error.CancelTransaction; 159 const record_cursor = try (try UserDB.HashMap(.read_only).init(records_cursor)).getCursor(hash.hashInt(hash_kind, ctx.notif_id)) orelse return error.CancelTransaction; 160 const notif = try evt.read(Notif, UserDB, hash_kind, ctx.arena, try UserDB.HashMap(.read_only).init(record_cursor)); 161 if (!std.mem.eql(u8, &notif.last_child_id, ctx.last_child_id)) return error.CancelTransaction; 162 163 const unread = try UserDB.SortedSet(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, unread_notif_id_set_key))); 164 if (!try unread.remove(&evt.orderKeyDesc(notif.time, ctx.notif_id))) return error.CancelTransaction; 165 } 166 }; 167 168 try transact(io, allocator, users_dir, user_id, Ctx{ .arena = &arena, .notif_id = notif_id, .last_child_id = last_child_id }); 169 } 170 171 // remove a notif. the thread's next bump recreates it. 172 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 { 173 var arena = std.heap.ArenaAllocator.init(allocator); 174 defer arena.deinit(); 175 176 const Ctx = struct { 177 arena: *std.heap.ArenaAllocator, 178 notif_id: *const [evt.event_id_size]u8, 179 180 pub fn run(ctx: @This(), cursor: *UserDB.Cursor(.read_write)) !void { 181 const moment = try UserDB.HashMap(.read_write).init(cursor.*); 182 const records = try UserDB.HashMap(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, notif_map_key))); 183 const time = (try readTime(ctx.arena, records.readOnly(), ctx.notif_id)) orelse return error.CancelTransaction; 184 try removeNotif(moment, records, ctx.notif_id, time); 185 } 186 }; 187 188 try transact(io, allocator, users_dir, user_id, Ctx{ .arena = &arena, .notif_id = notif_id }); 189 } 190 191 // up to `count` notifs from `start`, newest first. `moment` is the user repo's latest. 192 pub fn readRecent(moment: UserDB.HashMap(.read_only), arena: *std.heap.ArenaAllocator, start: u64, count: usize) ![]const Item { 193 var items: std.ArrayList(Item) = .empty; 194 const recent_cursor = try moment.getCursor(hash.hashInt(hash_kind, recent_notif_id_set_key)) orelse return items.items; 195 const recent = try UserDB.SortedSet(.read_only).init(recent_cursor); 196 197 var iter = try recent.iteratorFromIndex(start); 198 while (items.items.len < count) { 199 const kv_cursor = (try iter.next()) orelse break; 200 const id = try evt.readOrderKeyId(UserDB, kv_cursor); 201 try items.append(arena.allocator(), (try readItem(moment, arena, &id)) orelse return error.NotFound); 202 } 203 return items.items; 204 } 205 206 // the notif `id` names with its read state, or null when the user has none. `moment` is the user repo's latest. 207 pub fn readItem(moment: UserDB.HashMap(.read_only), arena: *std.heap.ArenaAllocator, id: *const [evt.event_id_size]u8) !?Item { 208 const records_cursor = try moment.getCursor(hash.hashInt(hash_kind, notif_map_key)) orelse return null; 209 const record_cursor = try (try UserDB.HashMap(.read_only).init(records_cursor)).getCursor(hash.hashInt(hash_kind, id)) orelse return null; 210 const notif = try evt.read(Notif, UserDB, hash_kind, arena, try UserDB.HashMap(.read_only).init(record_cursor)); 211 const unread_cursor = try moment.getCursor(hash.hashInt(hash_kind, unread_notif_id_set_key)) orelse return .{ .notif = notif, .unread = false }; 212 const unread = try (try UserDB.SortedSet(.read_only).init(unread_cursor)).contains(&evt.orderKeyDesc(notif.time, id)); 213 return .{ .notif = notif, .unread = unread }; 214 } 215 216 // a user repo's latest moment, or null before its first transaction 217 pub fn latestMoment(user_repo: *rp.Repo(.xit, evt.user_repo_opts)) !?UserDB.HashMap(.read_only) { 218 return user_repo.core.latestMoment() catch |err| switch (err) { 219 error.DatabaseIsEmpty => null, 220 else => |e| e, 221 }; 222 } 223 224 // how many notifs are unread. `moment` is the user repo's latest. 225 pub fn unreadCount(moment: UserDB.HashMap(.read_only)) !u64 { 226 const unread_cursor = try moment.getCursor(hash.hashInt(hash_kind, unread_notif_id_set_key)) orelse return 0; 227 return try (try UserDB.SortedSet(.read_only).init(unread_cursor)).count(); 228 } 229 230 // always notify `user_id` of the thread `thread_id` in the repo at `location` 231 pub fn subscribe( 232 io: std.Io, 233 allocator: std.mem.Allocator, 234 users_dir: []const u8, 235 user_id: *const [evt.event_id_size]u8, 236 location: evt.RepoLocation, 237 thread_id: *const [evt.event_id_size]u8, 238 author: evt.CommitAuthor, 239 ) !void { 240 try writeMode(io, allocator, users_dir, user_id, &location, thread_id, .subscribe, author); 241 indexSubscriber(io, allocator, users_dir, location, thread_id, user_id, true); 242 } 243 244 // never notify `user_id` of the thread `thread_id` in the repo at `location`, dropping its notif 245 pub fn mute( 246 io: std.Io, 247 allocator: std.mem.Allocator, 248 users_dir: []const u8, 249 user_id: *const [evt.event_id_size]u8, 250 location: evt.RepoLocation, 251 thread_id: *const [evt.event_id_size]u8, 252 author: evt.CommitAuthor, 253 ) !void { 254 try writeMode(io, allocator, users_dir, user_id, &location, thread_id, .mute, author); 255 try dismiss(io, allocator, users_dir, user_id, &idOf(&location.repo_id, thread_id)); 256 indexSubscriber(io, allocator, users_dir, location, thread_id, user_id, false); 257 } 258 259 // write the user's notify event for a thread 260 fn writeMode( 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: *const evt.RepoLocation, 266 thread_id: *const [evt.event_id_size]u8, 267 mode: evt.Notify.Mode, 268 author: evt.CommitAuthor, 269 ) !void { 270 var user_repo = (try evt.openUserRepo(io, allocator, users_dir, user_id)) orelse return error.NotFound; 271 defer user_repo.deinit(io, allocator); 272 try evt.consume(.{ .server = .{ .users_dir = users_dir } }, .user, .xit, evt.user_repo_opts, io, allocator, &user_repo, evt.events_ref, &[_]evt.EventWithId{.{ 273 .id = std.fmt.bytesToHex(evt.Notify.idOf(&location.repo_id, thread_id), .lower), 274 .timestamp = @intCast(std.Io.Timestamp.now(io, .real).toSeconds()), 275 .author = author, 276 .event = .{ .notify = .{ 277 .owner_id = std.fmt.bytesToHex(location.owner_id, .lower), 278 .repo_id = std.fmt.bytesToHex(location.repo_id, .lower), 279 .thread_id = std.fmt.bytesToHex(thread_id.*, .lower), 280 .mode = mode, 281 } }, 282 }}); 283 } 284 285 // add `user_id` to the thread's subscribers in its repo owner's index, or remove 286 // it, logging rather than returning a failure 287 fn indexSubscriber( 288 io: std.Io, 289 allocator: std.mem.Allocator, 290 users_dir: []const u8, 291 location: evt.RepoLocation, 292 thread_id: *const [evt.event_id_size]u8, 293 user_id: *const [evt.event_id_size]u8, 294 subscribed: bool, 295 ) void { 296 const Ctx = struct { 297 notif_id: [evt.event_id_size]u8, 298 user_id: *const [evt.event_id_size]u8, 299 subscribed: bool, 300 301 pub fn run(ctx: @This(), cursor: *UserDB.Cursor(.read_write)) !void { 302 const moment = try UserDB.HashMap(.read_write).init(cursor.*); 303 const sets = try UserDB.HashMap(.read_write).init(try moment.putCursor(hash.hashInt(hash_kind, subscriber_map_key))); 304 const subscribers = try UserDB.SortedSet(.read_write).init(try sets.putCursor(hash.hashInt(hash_kind, &ctx.notif_id))); 305 if (ctx.subscribed) { 306 try subscribers.put(ctx.user_id); 307 } else if (!try subscribers.remove(ctx.user_id)) return error.CancelTransaction; 308 } 309 }; 310 311 transact(io, allocator, users_dir, &location.owner_id, Ctx{ 312 .notif_id = idOf(&location.repo_id, thread_id), 313 .user_id = user_id, 314 .subscribed = subscribed, 315 }) catch |err| { 316 std.log.warn("failed to index thread subscriber: {s}", .{@errorName(err)}); 317 }; 318 } 319 320 // subscribe the users `events` show taking part in their threads, the threads' 321 // authors, the commenters and the assignees, unless they hold a record on the 322 // thread. logs rather than returning a failure. 323 pub fn subscribeParticipants(io: std.Io, allocator: std.mem.Allocator, server: evt.Host.Server, repo_path: []const u8, events: []const evt.EventWithId) void { 324 const admin_repo_path = server.admin_repo_path orelse return; 325 const location = evt.parseRepoPath(server.users_dir, repo_path) orelse return; 326 subscribeEach(io, allocator, server.users_dir, admin_repo_path, location, events) catch |err| { 327 std.log.warn("failed to subscribe thread participants: {s}", .{@errorName(err)}); 328 }; 329 } 330 331 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 { 332 var admin_repo = try rp.Repo(.xit, evt.admin_repo_opts).open(io, allocator, .{ .path = admin_repo_path }); 333 defer admin_repo.deinit(io, allocator); 334 const admin_moment = try evt.currentMoment(evt.admin_repo_opts, &admin_repo); 335 var arena = std.heap.ArenaAllocator.init(allocator); 336 defer arena.deinit(); 337 338 for (events) |event| { 339 const thread_hex: [evt.event_id_size * 2]u8, const email: []const u8 = switch (event.event) { 340 inline .discuss, .issue, .patch => |thread| if (thread != null) .{ event.id, event.author.email } else continue, 341 .comment => |comment_maybe| if (comment_maybe) |comment| .{ comment.thread_id, event.author.email } else continue, 342 .assign => |assign_maybe| if (assign_maybe) |assign| .{ assign.thread_id, assign.email } else continue, 343 else => continue, 344 }; 345 const user_id = (try evt.User.readIdByEmail(evt.AdminDB, evt.admin_repo_opts.hash, admin_moment, email)) orelse continue; 346 const thread_id = try evt.parseEventId(&thread_hex); 347 // a mute or an earlier subscribe stands 348 { 349 var user_repo = (try evt.openUserRepo(io, allocator, users_dir, &user_id)) orelse continue; 350 defer user_repo.deinit(io, allocator); 351 if (try evt.userMoment(&user_repo)) |moment| { 352 if (try evt.Notify.readMode(UserDB, hash_kind, moment, &arena, &location.repo_id, &thread_id) != null) continue; 353 } 354 } 355 // the write's own author, which for an assignment is the assigner 356 try subscribe(io, allocator, users_dir, &user_id, location, &thread_id, event.author); 357 } 358 } 359 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); 366 } 367 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 { 370 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); 379 } 380 return user_ids.items; 381 } 382 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); 387 } 388 389 // 390 // delivery worker 391 // 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 } 404 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, 419 io: std.Io, 420 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 } 451 } 452 453 // a round's notifs, coalesced per user and thread 454 const Inboxes = struct { 455 arena: *std.heap.ArenaAllocator, 456 admin_moment: evt.AdminDB.HashMap(.read_only), 457 owner_moment: UserDB.HashMap(.read_only), 458 repo: evt.Repo.Record, 459 repo_id: *const [evt.event_id_size]u8, 460 // whether each user is active and can read the repo 461 readers: std.AutoHashMapUnmanaged([evt.event_id_size]u8, bool) = .empty, 462 by_user: std.AutoArrayHashMapUnmanaged([evt.event_id_size]u8, std.ArrayList(Notif)) = .empty, 463 464 fn add(self: *Inboxes, user_id: [evt.event_id_size]u8, notif: Notif) !void { 465 if (!try self.canRead(user_id)) return; 466 const entry = try self.by_user.getOrPut(self.arena.allocator(), user_id); 467 if (!entry.found_existing) entry.value_ptr.* = .empty; 468 // events arrive newest first, so a thread keeps its latest child 469 for (entry.value_ptr.items) |existing| { 470 if (std.mem.eql(u8, &existing.thread_id, &notif.thread_id)) return; 471 } 472 try entry.value_ptr.append(self.arena.allocator(), notif); 473 } 474 475 fn canRead(self: *Inboxes, user_id: [evt.event_id_size]u8) !bool { 476 if (self.readers.get(user_id)) |can_read| return can_read; 477 const can_read = if (try evt.User.readById(evt.AdminDB, evt.admin_repo_opts.hash, self.admin_moment, self.arena, &user_id)) |user| 478 !user.removed and try evt.Repo.roleOf(UserDB, hash_kind, self.owner_moment, self.arena, self.repo, self.repo_id, user_id) != null 479 else 480 false; 481 try self.readers.put(self.arena.allocator(), user_id, can_read); 482 return can_read; 483 } 484 }; 485 486 // deliver the comments and assignments created since the repo's cursor, then advance it 487 fn notifyRepoEvents( 488 comptime repo_opts: rp.RepoOpts(.xit), 489 io: std.Io, 490 allocator: std.mem.Allocator, 491 users_dir: []const u8, 492 admin_repo_path: []const u8, 493 location: evt.RepoLocation, 494 repo: *rp.Repo(.xit, repo_opts), 495 ) !void { 496 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(); 503 504 var owner_repo = (try evt.openUserRepo(io, allocator, users_dir, &location.owner_id)) orelse return; 505 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; 513 514 var arena = std.heap.ArenaAllocator.init(allocator); 515 defer arena.deinit(); 516 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; 523 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); 526 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 }; 537 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); 540 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); 554 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 }; 567 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; 577 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 }; 587 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 } 596 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 }; 604 } 605 if (failed) return error.DeliveryFailed; 606 } 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 } 617 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 }); 631 } 632