+5 -5

src/atproto/repo.zig

Reviewed

··· 40 40 error.InvalidDagCbor => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "record is outside the atproto data model"), 41 41 else => return err, 42 42 }; 43 43 - sync.requestConfiguredCrawls(allocator); 43 43 + sync.notifyCrawlers(false); 44 44 return writeRecordRef(request, allocator, result.records[0], result.commit); 45 45 } 46 46 ··· 77 77 error.InvalidDagCbor => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "record is outside the atproto data model"), 78 78 else => return err, 79 79 }; 80 80 - sync.requestConfiguredCrawls(allocator); 80 80 + sync.notifyCrawlers(false); 81 81 return writeRecordRef(request, allocator, result.records[0], result.commit); 82 82 } 83 83 ··· 179 179 error.InvalidRecordKey => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"), 180 180 else => return err, 181 181 }; 182 182 - sync.requestConfiguredCrawls(allocator); 182 182 + sync.notifyCrawlers(false); 183 183 return http_api.json(request, .ok, "{}"); 184 184 } 185 185 ··· 289 289 "{{\"commit\":{{\"cid\":{f},\"rev\":{f}}},\"results\":{s}}}", 290 290 .{ std.json.fmt(result.commit.cid, .{}), std.json.fmt(result.commit.rev, .{}), out.written() }, 291 291 ); 292 292 - sync.requestConfiguredCrawls(allocator); 292 292 + sync.notifyCrawlers(false); 293 293 return http_api.json(request, .ok, body_out); 294 294 } 295 295 ··· 338 338 imported_records.items, 339 339 imported_blocks.items, 340 340 ); 341 341 - sync.requestConfiguredCrawls(allocator); 341 341 + if (store.isAccountActive(account.did)) sync.notifyCrawlers(true); 342 342 return http_api.json(request, .ok, "{}"); 343 343 } 344 344

+26 -21

src/atproto/server.zig

Reviewed

··· 747 747 refresh: ?[]const u8, 748 748 ) ![]const u8 { 749 749 const did_doc = try sessionDidDocJson(allocator, account); 750 750 - const status = if (active) "active" else "deactivated"; 751 751 - const token_fields = if (access) |access_jwt| 752 752 - try std.fmt.allocPrint(allocator, "\"accessJwt\":{f},\"refreshJwt\":{f},", .{ std.json.fmt(access_jwt, .{}), std.json.fmt(refresh.?, .{}) }) 753 753 - else 754 754 - ""; 755 755 - return std.fmt.allocPrint( 756 756 - allocator, 757 757 - "{{{s}\"did\":{f},\"didDoc\":{s},\"handle\":{f},\"email\":{f},\"emailConfirmed\":{},\"active\":{},\"status\":{f}}}", 758 758 - .{ 759 759 - token_fields, 760 760 - std.json.fmt(account.did, .{}), 761 761 - did_doc, 762 762 - std.json.fmt(account.handle, .{}), 763 763 - std.json.fmt(email, .{}), 764 764 - email_confirmed, 765 765 - active, 766 766 - std.json.fmt(status, .{}), 767 767 - }, 768 768 - ); 750 750 + const parsed_did_doc = try std.json.parseFromSlice(std.json.Value, allocator, did_doc, .{}); 751 751 + 752 752 + const Response = struct { 753 753 + accessJwt: ?[]const u8 = null, 754 754 + refreshJwt: ?[]const u8 = null, 755 755 + did: []const u8, 756 756 + didDoc: std.json.Value, 757 757 + handle: []const u8, 758 758 + email: []const u8, 759 759 + emailConfirmed: bool, 760 760 + active: bool, 761 761 + status: ?[]const u8 = null, 762 762 + }; 763 763 + return std.json.Stringify.valueAlloc(allocator, Response{ 764 764 + .accessJwt = access, 765 765 + .refreshJwt = refresh, 766 766 + .did = account.did, 767 767 + .didDoc = parsed_did_doc.value, 768 768 + .handle = account.handle, 769 769 + .email = email, 770 770 + .emailConfirmed = email_confirmed, 771 771 + .active = active, 772 772 + .status = if (active) null else "deactivated", 773 773 + }, .{ .emit_null_optional_fields = false }); 769 774 } 770 775 771 776 fn sessionDidDocJson(allocator: std.mem.Allocator, account: auth.Account) ![]const u8 { ··· 794 799 try store.sequenceAccountEvent(allocator, account.did, true); 795 800 try store.sequenceIdentityEvent(allocator, account.did, account.handle); 796 801 try store.sequenceSyncEvent(allocator, account.did); 797 797 - sync.requestConfiguredCrawls(allocator); 802 802 + sync.notifyCrawlers(true); 798 803 return http_api.json(request, .ok, "{}"); 799 804 } 800 805 ··· 805 810 const account = requireAccount(request, allocator) catch return; 806 811 try store.setAccountActive(account.did, false); 807 812 try store.sequenceAccountEvent(allocator, account.did, false); 808 808 - sync.requestConfiguredCrawls(allocator); 813 813 + sync.notifyCrawlers(true); 809 814 return http_api.json(request, .ok, "{}"); 810 815 } 811 816

+100 -6

src/atproto/sync.zig

Reviewed

··· 10 10 const http = std.http; 11 11 12 12 const max_subscribe_repos_connections = 32; 13 13 + const notify_threshold_ms = 20 * 60 * 1000; 14 14 + 13 15 var subscribe_repos_connections: usize = 0; 16 16 + var crawler_last_notified_ms: std.atomic.Value(i64) = .init(0); 17 17 + var crawler_notify_in_flight: std.atomic.Value(bool) = .init(false); 14 18 15 19 pub const SubscribeReposClient = struct { 16 20 state: *StreamState, ··· 29 33 self.state.retain(); 30 34 const thread = std.Thread.spawn(.{}, streamEvents, .{self.state}) catch |err| { 31 35 self.state.release(); 36 36 + log.err("sync subscribeRepos spawn failed cursor={d} err={s}\n", .{ self.state.cursor, @errorName(err) }); 32 37 return err; 33 38 }; 34 39 thread.detach(); ··· 80 85 { 81 86 var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); 82 87 defer arena.deinit(); 83 83 - const events = store.listSeqEvents(arena.allocator(), state.cursor, 100) catch return; 88 88 + const events = store.listSeqEvents(arena.allocator(), state.cursor, 100) catch |err| { 89 89 + log.err("sync subscribeRepos list failed cursor={d} err={s}\n", .{ state.cursor, @errorName(err) }); 90 90 + return; 91 91 + }; 84 92 for (events) |event| { 85 93 if (state.isClosed()) return; 86 86 - state.conn.writeBin(event.frame) catch return; 94 94 + writeFirehoseFrame(state.conn, event.frame) catch |err| { 95 95 + switch (err) { 96 96 + error.BrokenPipe, error.ConnectionResetByPeer => return, 97 97 + else => log.err("sync subscribeRepos write failed cursor={d} seq={d} bytes={d} err={s}\n", .{ state.cursor, event.seq, event.frame.len, @errorName(err) }), 98 98 + } 99 99 + return; 100 100 + }; 87 101 state.cursor = event.seq; 88 102 sent = true; 89 103 } ··· 98 112 } 99 113 } 100 114 } 115 115 + 116 116 + fn writeFirehoseFrame(conn: *httpz.websocket.Conn, payload: []const u8) !void { 117 117 + var header: [10]u8 = undefined; 118 118 + const header_len = websocketBinaryHeader(&header, payload.len); 119 119 + 120 120 + conn.lock.lockUncancelable(conn.io); 121 121 + defer conn.lock.unlock(conn.io); 122 122 + 123 123 + const socket = conn.stream.socket.handle; 124 124 + try writeSocketAll(socket, header[0..header_len]); 125 125 + try writeSocketAll(socket, payload); 126 126 + } 127 127 + 128 128 + fn websocketBinaryHeader(header: *[10]u8, payload_len: usize) usize { 129 129 + header[0] = 0x82; 130 130 + if (payload_len < 126) { 131 131 + header[1] = @intCast(payload_len); 132 132 + return 2; 133 133 + } 134 134 + if (payload_len <= std.math.maxInt(u16)) { 135 135 + header[1] = 126; 136 136 + std.mem.writeInt(u16, header[2..4], @intCast(payload_len), .big); 137 137 + return 4; 138 138 + } 139 139 + header[1] = 127; 140 140 + std.mem.writeInt(u64, header[2..10], @intCast(payload_len), .big); 141 141 + return 10; 142 142 + } 143 143 + 144 144 + fn writeSocketAll(socket: std.posix.socket_t, data: []const u8) !void { 145 145 + var remaining = data; 146 146 + while (remaining.len > 0) { 147 147 + const written = std.c.write(socket, remaining.ptr, remaining.len); 148 148 + if (written > 0) { 149 149 + remaining = remaining[@intCast(written)..]; 150 150 + continue; 151 151 + } 152 152 + if (written == 0) return error.WriteZero; 153 153 + switch (std.posix.errno(written)) { 154 154 + .AGAIN => { 155 155 + std.Thread.yield() catch {}; 156 156 + continue; 157 157 + }, 158 158 + .INTR => continue, 159 159 + .PIPE => return error.BrokenPipe, 160 160 + .CONNRESET => return error.ConnectionResetByPeer, 161 161 + else => return error.WriteFailed, 162 162 + } 163 163 + } 164 164 + } 101 165 }; 102 166 103 167 pub fn getBlob(request: *http_api.Request) !void { ··· 116 180 const blob = store.getBlob(allocator, did, cid) orelse { 117 181 return http_api.xrpcError(request, .not_found, "BlobNotFound", "Blob not found"); 118 182 }; 119 119 - const content_length = try std.fmt.allocPrint(allocator, "{d}", .{blob.data.len}); 120 183 const disposition = try std.fmt.allocPrint(allocator, "attachment; filename=\"{s}\"", .{cid}); 121 184 const headers = [_]http.Header{ 122 185 .{ .name = "content-type", .value = blob.mime_type }, 123 123 - .{ .name = "content-length", .value = content_length }, 124 186 .{ .name = "cache-control", .value = "public, max-age=31536000, immutable" }, 125 187 .{ .name = "x-content-type-options", .value = "nosniff" }, 126 188 .{ .name = "content-disposition", .value = disposition }, ··· 129 191 .{ .name = "access-control-allow-private-network", .value = "true" }, 130 192 .{ .name = "connection", .value = "close" }, 131 193 }; 132 132 - try http_api.respond(request, .ok, if (request.method == .HEAD) "" else blob.data, &headers); 194 194 + try http_api.respondNowClose(request, .ok, if (request.method == .HEAD) "" else blob.data, &headers); 133 195 } 134 196 135 197 pub fn getRepo(request: *http_api.Request) !void { ··· 240 302 return http_api.json(request, .ok, "{}"); 241 303 } 242 304 243 243 - pub fn requestConfiguredCrawls(allocator: std.mem.Allocator) void { 305 305 + pub fn notifyCrawlers(force: bool) void { 306 306 + const now = nowMillis(); 307 307 + if (!force) { 308 308 + const last = crawler_last_notified_ms.load(.acquire); 309 309 + if (last != 0 and now - last < notify_threshold_ms) return; 310 310 + } 311 311 + 312 312 + if (crawler_notify_in_flight.cmpxchgStrong(false, true, .acq_rel, .acquire) != null) return; 313 313 + crawler_last_notified_ms.store(now, .release); 314 314 + const thread = std.Thread.spawn(.{}, notifyCrawlersThread, .{}) catch |err| { 315 315 + crawler_notify_in_flight.store(false, .release); 316 316 + log.err("sync request_crawl spawn failed err={s}\n", .{@errorName(err)}); 317 317 + return; 318 318 + }; 319 319 + thread.detach(); 320 320 + } 321 321 + 322 322 + fn nowMillis() i64 { 323 323 + var ts: std.posix.timespec = undefined; 324 324 + return switch (std.posix.errno(std.posix.system.clock_gettime(.REALTIME, &ts))) { 325 325 + .SUCCESS => ts.sec * 1000 + @divTrunc(ts.nsec, 1_000_000), 326 326 + else => 0, 327 327 + }; 328 328 + } 329 329 + 330 330 + fn notifyCrawlersThread() void { 331 331 + defer crawler_notify_in_flight.store(false, .release); 332 332 + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); 333 333 + defer arena.deinit(); 334 334 + requestConfiguredCrawls(arena.allocator()); 335 335 + } 336 336 + 337 337 + fn requestConfiguredCrawls(allocator: std.mem.Allocator) void { 244 338 const host = publicHostname(config.publicUrl()) orelse { 245 339 log.err("sync request_crawl skipped invalid_public_url url={s}\n", .{config.publicUrl()}); 246 340 return;

+39 -13

src/http/api.zig

Reviewed

··· 168 168 169 169 pub fn corsPreflight(request: *Request) !void { 170 170 const res = response(); 171 171 - if (headerValue(request, "access-control-request-headers")) |requested| { 171 171 + const allow_headers = if (headerValue(request, "access-control-request-headers")) |requested| blk: { 172 172 const trimmed = std.mem.trim(u8, requested, " \t"); 173 173 - if (trimmed.len > 0) try addHeader(res, "access-control-allow-headers", trimmed); 174 174 - } 175 175 - try setHeaders(res, &cors_headers); 173 173 + break :blk if (trimmed.len > 0) trimmed else default_cors_allow_headers; 174 174 + } else default_cors_allow_headers; 175 175 + 176 176 + const headers = [_]http.Header{ 177 177 + .{ .name = "access-control-allow-origin", .value = "*" }, 178 178 + .{ .name = "access-control-allow-methods", .value = "GET, HEAD, POST, OPTIONS" }, 179 179 + .{ .name = "vary", .value = "Access-Control-Request-Headers" }, 180 180 + .{ .name = "access-control-allow-headers", .value = allow_headers }, 181 181 + .{ .name = "access-control-allow-private-network", .value = "true" }, 182 182 + .{ .name = "access-control-max-age", .value = "600" }, 183 183 + .{ .name = "connection", .value = "close" }, 184 184 + }; 185 185 + try setHeaders(res, &headers); 176 186 res.setStatus(.no_content); 177 187 } 178 188 ··· 248 258 .{ .name = "connection", .value = "close" }, 249 259 }; 250 260 251 251 - const cors_headers = [_]http.Header{ 252 252 - .{ .name = "access-control-allow-origin", .value = "*" }, 253 253 - .{ .name = "access-control-allow-methods", .value = "GET, HEAD, POST, OPTIONS" }, 254 254 - .{ .name = "vary", .value = "Access-Control-Request-Headers" }, 255 255 - .{ .name = "access-control-allow-headers", .value = "atproto-accept-labelers, atproto-proxy, authorization, content-type, dpop, x-bsky-topics" }, 256 256 - .{ .name = "access-control-allow-private-network", .value = "true" }, 257 257 - .{ .name = "access-control-max-age", .value = "600" }, 258 258 - .{ .name = "connection", .value = "close" }, 259 259 - }; 261 261 + const default_cors_allow_headers = "atproto-accept-labelers, atproto-proxy, authorization, content-type, dpop, x-bsky-topics"; 260 262 261 263 pub fn respond(request: *Request, status: http.Status, body: []const u8, headers: []const http.Header) !void { 262 264 _ = request; ··· 266 268 res.body = try res.arena.dupe(u8, body); 267 269 } 268 270 271 271 + pub fn respondNowClose(request: *Request, status: http.Status, body: []const u8, headers: []const http.Header) !void { 272 272 + const res = response(); 273 273 + var out: std.Io.Writer.Allocating = .init(res.arena); 274 274 + defer out.deinit(); 275 275 + 276 276 + try out.writer.print("HTTP/1.1 {d} \r\n", .{@intFromEnum(status)}); 277 277 + for (headers) |header| { 278 278 + try out.writer.print("{s}: {s}\r\n", .{ header.name, header.value }); 279 279 + } 280 280 + try out.writer.print("content-length: {d}\r\n\r\n", .{body.len}); 281 281 + 282 282 + request.conn.handover = .close; 283 283 + try request.conn.writeAll(out.written()); 284 284 + if (body.len > 0) try request.conn.writeAll(body); 285 285 + res.written = true; 286 286 + } 287 287 + 269 288 fn setHeaders(res: *Response, headers: []const http.Header) !void { 270 289 for (headers) |header| try addHeader(res, header.name, header.value); 271 290 } 272 291 273 292 fn addHeader(res: *Response, name: []const u8, value: []const u8) !void { 293 293 + for (res.headers.keys[0..res.headers.len], 0..) |existing, i| { 294 294 + if (std.ascii.eqlIgnoreCase(existing, name)) { 295 295 + res.headers.keys[i] = try res.arena.dupe(u8, name); 296 296 + res.headers.values[i] = try res.arena.dupe(u8, value); 297 297 + return; 298 298 + } 299 299 + } 274 300 try res.headerOpts(name, value, .{ .dupe_name = true, .dupe_value = true }); 275 301 } 276 302

+5 -1

src/main.zig

Reviewed

··· 5 5 6 6 var app_threaded_io: Io.Threaded = undefined; 7 7 pub const std_options_debug_threaded_io: ?*Io.Threaded = &app_threaded_io; 8 8 + const io_worker_limit = 16; 8 9 9 10 pub fn main(init: std.process.Init) !void { 10 11 const allocator = std.heap.smp_allocator; 11 11 - app_threaded_io = Io.Threaded.init(allocator, .{}); 12 12 + app_threaded_io = Io.Threaded.init(allocator, .{ 13 13 + .concurrent_limit = .limited(io_worker_limit), 14 14 + }); 12 15 const io = app_threaded_io.io(); 13 16 14 17 const options = zds.internal.cli.parse(init) catch |err| switch (err) { ··· 46 49 zds.core.log.info("bootstrap invite code: {s}\n", .{code}); 47 50 } 48 51 } 52 52 + zds.atproto.sync.notifyCrawlers(true); 49 53 try zds.http.server.listen(io, .{ .host = options.host, .port = options.port }); 50 54 }

+101 -33

src/storage/store.zig

Reviewed

··· 2116 2116 \\INSERT INTO commits (seq, did, cid, rev, prev) 2117 2117 \\VALUES (?, ?, ?, ?, NULL) 2118 2118 , .{ @as(i64, @intCast(seq)), account.did, commit_cid, rev }); 2119 2119 - const event_frame = try importedCommitEventFrame(allocator, seq, account.did, commit_cid, rev, blocks); 2120 2120 - try conn.exec( 2121 2121 - \\INSERT INTO seq_events (seq, did, commit_cid, evt) 2122 2122 - \\VALUES (?, ?, ?, ?) 2123 2123 - \\ON CONFLICT(seq) DO UPDATE SET 2124 2124 - \\ did = excluded.did, 2125 2125 - \\ commit_cid = excluded.commit_cid, 2126 2126 - \\ evt = excluded.evt 2127 2127 - , .{ @as(i64, @intCast(seq)), account.did, commit_cid, zqlite.blob(event_frame) }); 2119 2119 + const publish_event = try accountActiveLocked(account.did); 2120 2120 + if (publish_event) { 2121 2121 + const event_frame = try importedCommitEventFrame(allocator, seq, account.did, commit_cid, rev, blocks); 2122 2122 + try conn.exec( 2123 2123 + \\INSERT INTO seq_events (seq, did, commit_cid, evt) 2124 2124 + \\VALUES (?, ?, ?, ?) 2125 2125 + \\ON CONFLICT(seq) DO UPDATE SET 2126 2126 + \\ did = excluded.did, 2127 2127 + \\ commit_cid = excluded.commit_cid, 2128 2128 + \\ evt = excluded.evt 2129 2129 + , .{ @as(i64, @intCast(seq)), account.did, commit_cid, zqlite.blob(event_frame) }); 2130 2130 + } 2128 2131 try conn.commit(); 2129 2129 - eventlog.publish(seq); 2132 2132 + if (publish_event) eventlog.publish(seq); 2130 2133 } 2131 2134 2132 2135 pub fn writeRepoCar(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { ··· 2186 2189 2187 2190 var out: std.Io.Writer.Allocating = .init(allocator); 2188 2191 defer out.deinit(); 2189 2189 - try out.writer.writeAll("{\"repos\":["); 2190 2190 - var first = true; 2192 2192 + var json: std.json.Stringify = .{ .writer = &out.writer }; 2193 2193 + try json.beginObject(); 2194 2194 + try json.objectField("repos"); 2195 2195 + try json.beginArray(); 2191 2196 while (rows.next()) |row| { 2192 2192 - if (!first) try out.writer.writeByte(','); 2193 2193 - first = false; 2194 2197 const did = row.text(0); 2195 2198 const active = row.nullableInt(3) != null and row.nullableInt(4) == null; 2196 2196 - const status = if (active) "active" else "deactivated"; 2197 2197 - try out.writer.print( 2198 2198 - "{{\"did\":{f},\"head\":{f},\"rev\":{f},\"active\":{},\"status\":{f}}}", 2199 2199 - .{ 2200 2200 - std.json.fmt(did, .{}), 2201 2201 - std.json.fmt(row.text(1), .{}), 2202 2202 - std.json.fmt(row.text(2), .{}), 2203 2203 - active, 2204 2204 - std.json.fmt(status, .{}), 2205 2205 - }, 2206 2206 - ); 2199 2199 + try json.beginObject(); 2200 2200 + try json.objectField("did"); 2201 2201 + try json.write(did); 2202 2202 + try json.objectField("head"); 2203 2203 + try json.write(row.text(1)); 2204 2204 + try json.objectField("rev"); 2205 2205 + try json.write(row.text(2)); 2206 2206 + try json.objectField("active"); 2207 2207 + try json.write(active); 2208 2208 + if (!active) { 2209 2209 + try json.objectField("status"); 2210 2210 + try json.write("deactivated"); 2211 2211 + } 2212 2212 + try json.endObject(); 2207 2213 } 2208 2214 if (rows.err) |err| return err; 2209 2209 - try out.writer.writeByte(']'); 2210 2210 - try out.writer.writeByte('}'); 2215 2215 + try json.endArray(); 2216 2216 + try json.endObject(); 2211 2217 return out.toOwnedSlice(); 2212 2218 } 2213 2219 ··· 2266 2272 try requireInitialized(); 2267 2273 const root = try latestRootLocked(allocator, did); 2268 2274 const active = try accountActiveLocked(did); 2269 2269 - const status = if (active) "active" else "deactivated"; 2270 2270 - return std.fmt.allocPrint( 2271 2271 - allocator, 2272 2272 - "{{\"did\":{f},\"active\":{},\"status\":{f},\"rev\":{f}}}", 2273 2273 - .{ std.json.fmt(did, .{}), active, std.json.fmt(status, .{}), std.json.fmt(root.rev, .{}) }, 2274 2274 - ); 2275 2275 + var out: std.Io.Writer.Allocating = .init(allocator); 2276 2276 + defer out.deinit(); 2277 2277 + var json: std.json.Stringify = .{ .writer = &out.writer }; 2278 2278 + try json.beginObject(); 2279 2279 + try json.objectField("did"); 2280 2280 + try json.write(did); 2281 2281 + try json.objectField("active"); 2282 2282 + try json.write(active); 2283 2283 + if (!active) { 2284 2284 + try json.objectField("status"); 2285 2285 + try json.write("deactivated"); 2286 2286 + } 2287 2287 + try json.objectField("rev"); 2288 2288 + try json.write(root.rev); 2289 2289 + try json.endObject(); 2290 2290 + return out.toOwnedSlice(); 2275 2291 } 2276 2292 2277 2293 pub fn writeRecordJson(allocator: std.mem.Allocator, record: Record) ![]const u8 { ··· 3913 3929 .{ account.did, record.cid }, 3914 3930 ); 3915 3931 try std.testing.expect(get(account.did, "app.bsky.feed.post", "3ztest") == null); 3932 3932 + } 3933 3933 + 3934 3934 + test "inactive repo import does not publish firehose event" { 3935 3935 + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); 3936 3936 + defer arena.deinit(); 3937 3937 + const allocator = arena.allocator(); 3938 3938 + 3939 3939 + try init(std.Options.debug_io, ":memory:"); 3940 3940 + defer close(); 3941 3941 + 3942 3942 + const account = try createAccount( 3943 3943 + allocator, 3944 3944 + "import-inactive.test", 3945 3945 + "import-inactive@test.com", 3946 3946 + "password", 3947 3947 + "did:plc:importinactive", 3948 3948 + true, 3949 3949 + ); 3950 3950 + const parsed = try std.json.parseFromSlice(std.json.Value, allocator, "{\"text\":\"hello\"}", .{}); 3951 3951 + defer parsed.deinit(); 3952 3952 + 3953 3953 + const record = try create(allocator, account, "app.bsky.feed.post", "3ztest", parsed.value); 3954 3954 + const repo_car = try writeRepoCar(allocator, account.did); 3955 3955 + const loaded = try zat.loadCommitFromCAR(allocator, repo_car); 3956 3956 + 3957 3957 + var imported_blocks: std.ArrayList(ImportedBlock) = .empty; 3958 3958 + for (loaded.repo_car.blocks) |block| { 3959 3959 + try imported_blocks.append(allocator, .{ 3960 3960 + .cid = try cidText(allocator, block.cid_raw), 3961 3961 + .data = block.data, 3962 3962 + }); 3963 3963 + } 3964 3964 + const imported_records = [_]ImportedRecord{.{ 3965 3965 + .collection = record.collection, 3966 3966 + .rkey = record.rkey, 3967 3967 + .cid = record.cid, 3968 3968 + .blob_cids = &.{}, 3969 3969 + }}; 3970 3970 + 3971 3971 + try setAccountActive(account.did, false); 3972 3972 + const before = try scalarCountLocked("SELECT COUNT(*) FROM seq_events WHERE did = ?", account.did); 3973 3973 + try importRepo( 3974 3974 + allocator, 3975 3975 + account, 3976 3976 + try cidText(allocator, loaded.commit_cid), 3977 3977 + loaded.commit.rev, 3978 3978 + &imported_records, 3979 3979 + imported_blocks.items, 3980 3980 + ); 3981 3981 + const after = try scalarCountLocked("SELECT COUNT(*) FROM seq_events WHERE did = ?", account.did); 3982 3982 + try std.testing.expectEqual(before, after); 3983 3983 + try std.testing.expectEqualStrings((try latestRootLocked(allocator, account.did)).cid, try cidText(allocator, loaded.commit_cid)); 3916 3984 } 3917 3985 3918 3986 test "put updates record index while content comes from repo blocks" {

+1 -1

tools/smoke.sh

Reviewed

··· 58 58 token=$(printf '%s' "$session" | sed -n 's/.*"accessJwt":"\([^"]*\)".*/\1/p') 59 59 test -n "$token" 60 60 printf '%s' "$session" | grep -q '"didDoc":' 61 61 - printf '%s' "$session" | grep -q '"status":"active"' 61 61 + ! printf '%s' "$session" | grep -q '"status":' 62 62 printf '%s' "$session" | grep -q '"serviceEndpoint":"http://127.0.0.1:' 63 63 64 64 describe=$(curl -fsS "$base/xrpc/com.atproto.server.describeServer")