🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 fc7fa606.. ╭fc7fa606..╮ see-em: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 files │diff │ refs discuss issues new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────────╯ user
╭ changed ─────────╮ │src/test/event.zig│ ╰──────────────────╯ ╭────────────────────────────────────────────────────────────────────────╮ │31 var repo = try Repo.init(io, allocator, .{ .path = work_path });│ │32 defer repo.deinit(io, allocator); │ │33 │ │34 + // │ │35 // define test events │ │36 + // │ │37 │ │38 const AddIssueData = struct { │ │39 title: []const u8, │ ╰────────────────────────────────────────────────────────────────────────╯ ╭──────────────────────────────────────╮ │41 tags: []const []const u8, │ │42 }; │ │43 │ │42 - const id_size: usize = 20; │ │44 + const id_size: usize = 32; │ │45 const AddIssueEvent = struct {│ │44 - id: [2 * id_size]u8, │ │46 + id: [id_size * 2]u8, │ │47 kind: []const u8, │ │48 data: AddIssueData, │ │49 }; │ ╰──────────────────────────────────────╯ ╭──────────────────────────────────────────────────────────────────────────╮ │76 }, │ │77 }; │ │78 │ │79 + // │ │80 // insert issues as commits in the repo │ │81 + // │ │82 │ │83 var prng = std.Random.DefaultPrng.init(std.testing.random_seed); │ │84 var json: std.Io.Writer.Allocating = .init(std.testing.allocator);│ ╰──────────────────────────────────────────────────────────────────────────╯ ╭───────────────────────────────────────────────────────────╮ │87 for (issues_data) |issue_data| { │ │88 json.clearRetainingCapacity(); │ │89 │ │86 - var entropy: [id_size]u8 = undefined; │ │87 - prng.random().bytes(&entropy); │ │90 + var id_bytes: [id_size]u8 = undefined; │ │91 + prng.random().bytes(&id_bytes); │ │92 │ │93 const event = AddIssueEvent{ │ │90 - .id = std.fmt.bytesToHex(entropy, .lower), │ │94 + .id = std.fmt.bytesToHex(id_bytes, .lower),│ │95 .kind = "add-issue", │ │96 .data = issue_data, │ │97 }; │ ╰───────────────────────────────────────────────────────────╯ ╭────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │ 98 │ │ 99 try std.json.Stringify.value(event, .{}, &json.writer); │ │100 + │ │101 + // commit the event into a special branch │ │102 _ = try repo.commitAtRef(io, allocator, .{ .message = json.written() }, null, .{ .kind = .head, .name = "haxy/meta" });│ │103 } │ │104 │ ╰────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭───────────────────────────────────────────────────────────╮ │105 + // │ │106 // read and parse all of the events from the repo │ │107 + // │ │108 │ │109 var events: std.ArrayList(AddIssueEvent) = .empty;│ │110 defer events.deinit(allocator); │ ╰───────────────────────────────────────────────────────────╯ ╭──────────────────────────────────────────────────────────────────────────╮ │119 var commit_iter = try repo.log(io, allocator, &.{ref_haxy_meta});│ │120 defer commit_iter.deinit(); │ │121 │ │122 + // read the message from each commit │ │123 while (try commit_iter.next()) |commit_object| { │ │124 defer commit_object.deinit(); │ │125 │ ╰──────────────────────────────────────────────────────────────────────────╯ ╭──────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │129 try event_strs.append(allocator, message); │ │130 } │ │131 │ │132 + // parse commit messages as JSON into struct instances. │ │133 + // add events in reverse order so the earliest event is first. │ │134 for (0..event_strs.items.len) |i| { │ │135 const event_str = event_strs.items[event_strs.items.len - i - 1]; │ │136 const event = try std.json.parseFromSliceLeaky(AddIssueEvent, arena.allocator(), event_str, .{});│ ╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │137 try events.append(allocator, event); │ │138 } │ │139 │ │140 + // │ │141 // process events into the database │ │142 + // │ │143 + │ │144 + const Ctx = struct { │ │145 + events: []AddIssueEvent, │ │146 + │ │147 + pub fn run(ctx: @This(), cursor: *Repo.DB.Cursor(.read_write)) !void { │ │148 + const moment = try Repo.DB.HashMap(.read_write).init(cursor.*); │ │149 │ │131 - for (events.items) |event| { │ │132 - const Ctx = struct { │ │133 - event: AddIssueEvent, │ │150 + // the map with all of haxy's state and materialized views │ │151 + const haxy_cursor = try moment.putCursor(hash.hashInt(repo_opts.hash, "haxy")); │ │152 + const haxy = try Repo.DB.HashMap(.read_write).init(haxy_cursor); │ │153 + │ │154 + // try reading the last event id that was processed in the db │ │155 + var last_event_id_maybe: ?[id_size]u8 = null; │ │156 + if (try haxy.getCursor(hash.hashInt(repo_opts.hash, "last-event-id"))) |last_event_id_cursor| { │ │157 + var last_event_id_buffer: [id_size]u8 = undefined; │ │158 + _ = try last_event_id_cursor.readBytes(&last_event_id_buffer); │ │159 + last_event_id_maybe = last_event_id_buffer; │ │160 + } │ │161 + │ │162 + // for each event we want to process... │ │163 + for (ctx.events) |event| { │ │164 + // get the id of the current event as bytes │ │165 + var current_event_id_buffer: [id_size]u8 = undefined; │ │166 + _ = try std.fmt.hexToBytes(&current_event_id_buffer, &event.id); │ │167 + │ │168 + // if this event has already been processed, skip it │ │169 + if (try haxy.getCursor(hash.hashInt(repo_opts.hash, "event-id->views"))) |event_id_to_views_cursor| { │ │170 + const event_id_to_views = try Repo.DB.HashMap(.read_only).init(event_id_to_views_cursor); │ │171 + if (null != try event_id_to_views.getCursor(hash.hashInt(repo_opts.hash, &current_event_id_buffer))) { │ │172 + continue; │ │173 + } │ │174 + } │ │175 + │ │176 + // map with the views as they appeared when each event was processed. │ │177 + // we can use this to see (and revert) the views to any previous state. │ │178 + const event_id_to_views_cursor = try haxy.putCursor(hash.hashInt(repo_opts.hash, "event-id->views")); │ │179 + const event_id_to_views = try Repo.DB.HashMap(.read_write).init(event_id_to_views_cursor); │ │180 + │ │181 + // create a new views map for the current event we are processing │ │182 + var views_cursor = try event_id_to_views.putCursor(hash.hashInt(repo_opts.hash, &current_event_id_buffer)); │ │183 + │ │184 + // if there was a previous event, set the views map to have the same value as it. │ │185 + // this efficiently "clones" the map so we make further modifications based on it. │ │186 + if (last_event_id_maybe) |*last_event_id| { │ │187 + if (try event_id_to_views.getCursor(hash.hashInt(repo_opts.hash, last_event_id))) |last_view_cursor| { │ │188 + try views_cursor.write(.{ .slot = last_view_cursor.slot() }); │ │189 + } │ │190 + } │ │191 + │ │192 + // process the event into the views map │ │193 + { │ │194 + const views = try Repo.DB.HashMap(.read_write).init(views_cursor); │ │195 │ │135 - pub fn run(ctx: @This(), cursor: *Repo.DB.Cursor(.read_write)) !void { │ │136 - const moment = try Repo.DB.HashMap(.read_write).init(cursor.*); │ │196 + if (std.mem.eql(u8, "add-issue", event.kind)) { │ │197 + const issues_cursor = try views.putCursor(hash.hashInt(repo_opts.hash, "issues")); │ │198 + const issues = try Repo.DB.ArrayList(.read_write).init(issues_cursor); │ │199 │ │138 - const haxy_views_cursor = try moment.putCursor(hash.hashInt(repo_opts.hash, "haxy-views")); │ │139 - const haxy_views = try Repo.DB.HashMap(.read_write).init(haxy_views_cursor); │ │200 + const issue_cursor = try issues.appendCursor(); │ │201 + const issue = try Repo.DB.HashMap(.read_write).init(issue_cursor); │ │202 │ │141 - const issues_cursor = try haxy_views.putCursor(hash.hashInt(repo_opts.hash, "issues")); │ │142 - const issues = try Repo.DB.ArrayList(.read_write).init(issues_cursor); │ │203 + try issue.put(hash.hashInt(repo_opts.hash, "id"), .{ .bytes = &event.id }); │ │204 + try issue.put(hash.hashInt(repo_opts.hash, "title"), .{ .bytes = event.data.title }); │ │205 + try issue.put(hash.hashInt(repo_opts.hash, "description"), .{ .bytes = event.data.description }); │ │206 + } else { │ │207 + return error.InvalidEventKind; │ │208 + } │ │209 + } │ │210 │ │144 - const issue_cursor = try issues.appendCursor(); │ │145 - const issue = try Repo.DB.HashMap(.read_write).init(issue_cursor); │ │211 + // the current event id is now the last one │ │212 + last_event_id_maybe = current_event_id_buffer; │ │213 │ │147 - try issue.put(hash.hashInt(repo_opts.hash, "id"), .{ .bytes = &ctx.event.id }); │ │148 - try issue.put(hash.hashInt(repo_opts.hash, "title"), .{ .bytes = ctx.event.data.title }); │ │149 - try issue.put(hash.hashInt(repo_opts.hash, "description"), .{ .bytes = ctx.event.data.description }); │ │214 + // prevent any of the data created above from being mutated by future iterations of this loop │ │215 + try cursor.db.freeze(); │ │216 } │ │151 - }; │ │217 │ │153 - try repo.core.db_file.lock(io, .exclusive); │ │154 - defer repo.core.db_file.unlock(io); │ │218 + if (last_event_id_maybe) |*last_event_id| { │ │219 + try haxy.put(hash.hashInt(repo_opts.hash, "last-event-id"), .{ .bytes = last_event_id }); │ │220 + } │ │221 + } │ │222 + }; │ │223 + │ │224 + try repo.core.db_file.lock(io, .exclusive); │ │225 + defer repo.core.db_file.unlock(io); │ │226 + │ │227 + // create a new transaction in the database that runs the above-defined Ctx function │ │228 + const history = try Repo.DB.ArrayList(.read_write).init(repo.core.db.rootCursor()); │ │229 + try history.appendContext(.{ .slot = try history.getSlot(-1) }, Ctx{ .events = events.items }); │ │230 + │ │231 + // read the moment we just created │ │232 + const moment_cursor = try history.getCursor(-1) orelse return error.NotFound; │ │233 + const moment = try Repo.DB.HashMap(.read_only).init(moment_cursor); │ │234 + │ │235 + const haxy_cursor = try moment.getCursor(hash.hashInt(repo_opts.hash, "haxy")) orelse return error.NotFound; │ │236 + const haxy = try Repo.DB.HashMap(.read_only).init(haxy_cursor); │ │237 + │ │238 + const event_id_to_views_cursor = try haxy.getCursor(hash.hashInt(repo_opts.hash, "event-id->views")) orelse return error.NotFound;│ │239 + const event_id_to_views = try Repo.DB.HashMap(.read_only).init(event_id_to_views_cursor); │ │240 │ │156 - const history = try Repo.DB.ArrayList(.read_write).init(repo.core.db.rootCursor()); │ │157 - try history.appendContext( │ │158 - .{ .slot = try history.getSlot(-1) }, │ │159 - Ctx{ .event = event }, │ │160 - ); │ │241 + // make sure all events have been processed │ │242 + var count: usize = 0; │ │243 + var event_id_to_views_iter = try event_id_to_views.iterator(); │ │244 + while (try event_id_to_views_iter.next()) |kv_pair_cursor| { │ │245 + _ = try kv_pair_cursor.readKeyValuePair(); │ │246 + count += 1; │ │247 } │ │248 + try std.testing.expectEqual(issues_data.len, count); │ │249 } │ │250 │ ╰───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯