Honor advertised repo and sync parameters · zat.dev/zds@3b9b11f

+17

bench/README.md

··· 215 215 | official Bluesky PDS | `bench/official-pds-records.bench.ts` copied into `packages/pds` | best match for actor-store record CID/get/list/write paths | 216 216 | Pegasus | `pegasus/bench/bench_repository.ml`, `mist/bench/bench_mst.ml` | best match for repository block/MST internals; not a full HTTP/PDS route comparison yet | 217 217 218 + ## stage checklist 219 + 220 + For each protocol-parity or storage-behavior stage, do this before calling the 221 + stage complete: 222 + 223 + - Re-read the relevant atproto spec or lexicon, then cross-check the official 224 + PDS implementation and Tranquil. 225 + - Decide whether any new parser, codec, verifier, resolver, repository 226 + primitive, OAuth helper, or JSON-path helper belongs in `zat`; keep it local 227 + unless the reusable boundary is clear. 228 + - Add or update a focused ZDS benchmark for the changed surface. 229 + - Run the comparable Tranquil benchmark when Tranquil has the same seam. 230 + - Run the comparable official PDS probe when the official PDS has the same 231 + seam. 232 + - Record any mismatch as either fixed, intentionally local with rationale, or a 233 + concrete follow-up issue. Do not leave “close enough” behavior unnamed. 234 + 218 235 Run Tranquil's metastore benchmark: 219 236 220 237 ```sh

+5

docs/operations.md

··· 176 176 just docker-publish-release vX.Y.Z 177 177 ``` 178 178 179 + For protocol-parity or storage-behavior releases, also complete the 180 + [benchmark stage checklist](../bench/README.md#stage-checklist): revisit the 181 + Zat boundary, run matching Tranquil and official PDS probes where comparable, 182 + and document any remaining mismatch as an explicit decision or follow-up. 183 + 179 184 ## migration 180 185 181 186 ZDS supports browser-based account migration flows such as PDS Moover:

+8 -5

docs/permissioned-data.md

··· 1 1 # permissioned data 2 2 3 - Permissioned data support is experimental and operator gated. ZDS only exposes 4 - `com.atproto.space.*` when explicitly enabled: 3 + Permissioned data support is experimental and operator gated. ZDS documents the 4 + `com.atproto.space.*` surface as experimental, and only enables the handlers 5 + when explicitly configured: 5 6 6 7 ```sh 7 8 ZDS_PERMISSIONED_DATA=true 8 9 ``` 9 10 10 - When the flag is off, permissioned-data routes stay dark and should not create 11 - or mutate permissioned-space rows. When the flag is on, the API reference marks 12 - the space routes with `x-zds-experimental: true`. 11 + When the flag is off, permissioned-data routes return a structured 12 + `MethodNotImplemented` response and do not create or mutate permissioned-space 13 + rows. The API reference marks the space routes with 14 + `x-zds-experimental: true` so clients can treat the namespace as experimental 15 + even when an operator has not enabled it. 13 16 14 17 ## references 15 18

+115 -14

src/atproto/repo.zig

··· 8 8 const zat = @import("zat"); 9 9 10 10 const http = std.http; 11 + const max_repo_write_body_len = 5 * 1024 * 1024; 11 12 const max_apply_writes_body_len = 5 * 1024 * 1024; 12 13 13 14 pub fn createRecord(request: *http_api.Request) !void { ··· 17 18 18 19 const auth_ctx = requireAccount(request, allocator) catch return; 19 20 const account = auth_ctx.account; 20 - var body_buf: [65536]u8 = undefined; 21 - const body = try http_api.readBody(request, &body_buf); 21 + const body = http_api.readBodyAlloc(request, allocator, max_repo_write_body_len) catch |err| switch (err) { 22 + error.BodyTooLarge => return http_api.xrpcError(request, .payload_too_large, "InvalidRequest", "record write body is too large"), 23 + else => return err, 24 + }; 22 25 const parsed = try http_api.parseJsonBody(request, allocator, body); 23 26 try requireRepoMatches(request, account, parsed.value); 27 + const options = parseWriteOptions(request, parsed.value) catch return; 24 28 const collection = zat.json.getString(parsed.value, "collection") orelse { 25 29 return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing collection"); 26 30 }; ··· 30 34 return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing record"); 31 35 }; 32 36 33 - const result = store.applyWrites(allocator, account, &.{.{ .create = .{ 37 + const result = store.applyWritesWithOptions(allocator, account, &.{.{ .create = .{ 34 38 .collection = collection, 35 39 .rkey = rkey, 36 40 .value = record_value, 37 - } }}) catch |err| switch (err) { 41 + } }}, options) catch |err| switch (err) { 38 42 error.InvalidCollection => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid collection"), 39 43 error.InvalidRecordKey => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"), 40 44 error.InvalidRecordType => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid record type"), 45 + error.InvalidSwap => return http_api.xrpcError(request, .bad_request, "InvalidSwap", "swapCommit did not match current commit"), 46 + error.ValidationRequired => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Record type is not known to this PDS"), 41 47 error.InvalidDagCbor => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "record is outside the atproto data model"), 42 48 else => return err, 43 49 }; ··· 52 58 53 59 const auth_ctx = requireAccount(request, allocator) catch return; 54 60 const account = auth_ctx.account; 55 - var body_buf: [65536]u8 = undefined; 56 - const body = try http_api.readBody(request, &body_buf); 61 + const body = http_api.readBodyAlloc(request, allocator, max_repo_write_body_len) catch |err| switch (err) { 62 + error.BodyTooLarge => return http_api.xrpcError(request, .payload_too_large, "InvalidRequest", "record write body is too large"), 63 + else => return err, 64 + }; 57 65 const parsed = try http_api.parseJsonBody(request, allocator, body); 58 66 try requireRepoMatches(request, account, parsed.value); 67 + const options = parseWriteOptions(request, parsed.value) catch return; 68 + const swap = parseRecordSwap(request, parsed.value) catch return; 59 69 const collection = zat.json.getString(parsed.value, "collection") orelse { 60 70 return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing collection"); 61 71 }; ··· 67 77 return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing record"); 68 78 }; 69 79 70 - const result = store.applyWrites(allocator, account, &.{.{ .update = .{ 80 + const result = store.applyWritesWithOptions(allocator, account, &.{.{ .update = .{ 71 81 .collection = collection, 72 82 .rkey = rkey, 73 83 .value = record_value, 74 - } }}) catch |err| switch (err) { 84 + .swap = swap, 85 + } }}, options) catch |err| switch (err) { 75 86 error.InvalidCollection => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid collection"), 76 87 error.InvalidRecordKey => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"), 77 88 error.InvalidRecordType => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid record type"), 89 + error.InvalidSwap => return http_api.xrpcError(request, .bad_request, "InvalidSwap", "swapCommit or swapRecord did not match current repo state"), 90 + error.ValidationRequired => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Record type is not known to this PDS"), 78 91 error.InvalidDagCbor => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "record is outside the atproto data model"), 79 92 else => return err, 80 93 }; ··· 134 147 const record = store.get(account.did, collection, rkey) orelse { 135 148 return http_api.xrpcError(request, .not_found, "RecordNotFound", "Record not found"); 136 149 }; 150 + var cid_buf: [256]u8 = undefined; 151 + if (http_api.queryParam(request.url.raw, "cid", &cid_buf)) |expected_cid| { 152 + if (!std.mem.eql(u8, record.cid, expected_cid)) { 153 + return http_api.xrpcError(request, .not_found, "RecordNotFound", "Record not found"); 154 + } 155 + } 137 156 const body = try store.writeRecordJson(allocator, record); 138 157 return http_api.json(request, .ok, body); 139 158 } ··· 154 173 const collection = http_api.queryParam(request.url.raw, "collection", &collection_buf) orelse { 155 174 return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing collection"); 156 175 }; 157 - const body = try store.writeListJson(allocator, account.did, collection, @min(http_api.queryLimit(request.url.raw, 50), 100)); 176 + var cursor_buf: [512]u8 = undefined; 177 + const cursor = http_api.queryParam(request.url.raw, "cursor", &cursor_buf); 178 + const reverse = queryBool(request.url.raw, "reverse", false); 179 + const body = try store.writeListJson(allocator, account.did, collection, cursor, reverse, @min(http_api.queryLimit(request.url.raw, 50), 100)); 158 180 return http_api.json(request, .ok, body); 159 181 } 160 182 ··· 165 187 166 188 const auth_ctx = requireAccount(request, allocator) catch return; 167 189 const account = auth_ctx.account; 168 - var body_buf: [4096]u8 = undefined; 169 - const body = try http_api.readBody(request, &body_buf); 190 + const body = http_api.readBodyAlloc(request, allocator, max_repo_write_body_len) catch |err| switch (err) { 191 + error.BodyTooLarge => return http_api.xrpcError(request, .payload_too_large, "InvalidRequest", "record write body is too large"), 192 + else => return err, 193 + }; 170 194 const parsed = try http_api.parseJsonBody(request, allocator, body); 171 195 try requireRepoMatches(request, account, parsed.value); 196 + const options = parseWriteOptions(request, parsed.value) catch return; 197 + const swap = parseRecordSwap(request, parsed.value) catch return; 172 198 const collection = zat.json.getString(parsed.value, "collection") orelse { 173 199 return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing collection"); 174 200 }; 201 + try requireRepoScope(request, auth_ctx.oauth_scope, .delete, collection); 175 202 const rkey = zat.json.getString(parsed.value, "rkey") orelse { 176 203 return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing rkey"); 177 204 }; 178 - _ = store.delete(allocator, account, collection, rkey) catch |err| switch (err) { 205 + _ = store.applyWritesWithOptions(allocator, account, &.{.{ .delete = .{ 206 + .collection = collection, 207 + .rkey = rkey, 208 + .swap = swap, 209 + } }}, options) catch |err| switch (err) { 179 210 error.InvalidCollection => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid collection"), 180 211 error.InvalidRecordKey => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"), 212 + error.InvalidSwap => return http_api.xrpcError(request, .bad_request, "InvalidSwap", "swapCommit or swapRecord did not match current repo state"), 181 213 else => return err, 182 214 }; 183 215 sync.notifyCrawlers(false); ··· 197 229 }; 198 230 const parsed = try http_api.parseJsonBody(request, allocator, body); 199 231 try requireRepoMatches(request, account, parsed.value); 232 + const options = parseWriteOptions(request, parsed.value) catch return; 200 233 const writes = switch (parsed.value) { 201 234 .object => |object| object.get("writes") orelse return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing writes"), 202 235 else => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Expected object"), ··· 236 269 .collection = collection, 237 270 .rkey = rkey, 238 271 .value = record_value, 272 + .swap = parseRecordSwap(request, write) catch return, 239 273 } }); 240 274 } else if (std.mem.eql(u8, type_name, "com.atproto.repo.applyWrites#delete")) { 241 275 try requireRepoScope(request, auth_ctx.oauth_scope, .delete, collection); ··· 245 279 try staged.append(allocator, .{ .delete = .{ 246 280 .collection = collection, 247 281 .rkey = rkey, 282 + .swap = parseRecordSwap(request, write) catch return, 248 283 } }); 249 284 } else { 250 285 return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Unknown write operation type"); 251 286 } 252 287 } 253 - const result = store.applyWrites(allocator, account, staged.items) catch |err| switch (err) { 288 + const result = store.applyWritesWithOptions(allocator, account, staged.items, options) catch |err| switch (err) { 254 289 error.InvalidCollection => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid collection"), 255 290 error.InvalidRecordKey => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"), 256 291 error.InvalidRecordType => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Invalid record type"), 292 + error.InvalidSwap => return http_api.xrpcError(request, .bad_request, "InvalidSwap", "swapCommit or swapRecord did not match current repo state"), 293 + error.ValidationRequired => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Record type is not known to this PDS"), 257 294 error.InvalidDagCbor => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "record is outside the atproto data model"), 258 295 else => return err, 259 296 }; ··· 516 553 517 554 const auth_ctx = requireAccount(request, allocator) catch return; 518 555 const account = auth_ctx.account; 519 - const body = try store.writeMissingBlobsJson(allocator, account.did, @min(http_api.queryLimit(request.url.raw, 500), 1000)); 556 + var cursor_buf: [256]u8 = undefined; 557 + const body = try store.writeMissingBlobsJson( 558 + allocator, 559 + account.did, 560 + http_api.queryParam(request.url.raw, "cursor", &cursor_buf), 561 + @min(http_api.queryLimit(request.url.raw, 500), 1000), 562 + ); 520 563 return http_api.json(request, .ok, body); 521 564 } 522 565 ··· 553 596 .object => |object| object.get("record"), 554 597 else => null, 555 598 }; 599 + } 600 + 601 + fn parseWriteOptions(request: *http_api.Request, value: std.json.Value) !store.WriteOptions { 602 + const object = switch (value) { 603 + .object => |object| object, 604 + else => { 605 + try http_api.xrpcError(request, .bad_request, "InvalidRequest", "Expected object"); 606 + return error.HandledResponse; 607 + }, 608 + }; 609 + 610 + var options: store.WriteOptions = .{}; 611 + if (object.get("validate")) |validate| { 612 + options.validate = switch (validate) { 613 + .bool => |enabled| if (enabled) .require else .skip, 614 + else => { 615 + try http_api.xrpcError(request, .bad_request, "InvalidRequest", "validate must be a boolean"); 616 + return error.HandledResponse; 617 + }, 618 + }; 619 + } 620 + if (object.get("swapCommit")) |swap_commit| { 621 + options.swap_commit = switch (swap_commit) { 622 + .string => |cid| cid, 623 + else => { 624 + try http_api.xrpcError(request, .bad_request, "InvalidRequest", "swapCommit must be a CID string"); 625 + return error.HandledResponse; 626 + }, 627 + }; 628 + } 629 + return options; 630 + } 631 + 632 + fn parseRecordSwap(request: *http_api.Request, value: std.json.Value) !store.RecordSwap { 633 + const object = switch (value) { 634 + .object => |object| object, 635 + else => { 636 + try http_api.xrpcError(request, .bad_request, "InvalidRequest", "Expected object"); 637 + return error.HandledResponse; 638 + }, 639 + }; 640 + const swap_record = object.get("swapRecord") orelse return .none; 641 + return switch (swap_record) { 642 + .null => .missing, 643 + .string => |cid| .{ .cid = cid }, 644 + else => { 645 + try http_api.xrpcError(request, .bad_request, "InvalidRequest", "swapRecord must be a CID string or null"); 646 + return error.HandledResponse; 647 + }, 648 + }; 649 + } 650 + 651 + fn queryBool(target: []const u8, name: []const u8, default: bool) bool { 652 + var buf: [16]u8 = undefined; 653 + const raw = http_api.queryParam(target, name, &buf) orelse return default; 654 + if (std.ascii.eqlIgnoreCase(raw, "true") or std.mem.eql(u8, raw, "1")) return true; 655 + if (std.ascii.eqlIgnoreCase(raw, "false") or std.mem.eql(u8, raw, "0")) return false; 656 + return default; 556 657 } 557 658 558 659 fn writeRecordRef(request: *http_api.Request, allocator: std.mem.Allocator, record: store.Record, commit: store.CommitInfo) !void {

+17 -3

src/atproto/sync.zig

··· 203 203 const did = http_api.queryParam(request.url.raw, "did", &did_buf) orelse { 204 204 return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing did"); 205 205 }; 206 - const body = store.writeRepoCar(allocator, did) catch { 206 + var since_buf: [256]u8 = undefined; 207 + const since = http_api.queryParam(request.url.raw, "since", &since_buf); 208 + const body = store.writeRepoCarSince(allocator, did, since) catch { 207 209 return http_api.xrpcError(request, .not_found, "RepoNotFound", "Repo not found"); 208 210 }; 209 211 const headers = [_]http.Header{ ··· 224 226 const did = http_api.queryParam(request.url.raw, "did", &did_buf) orelse { 225 227 return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Missing did"); 226 228 }; 227 - const body = try store.writeBlobListJson(allocator, did, @min(http_api.queryLimit(request.url.raw, 500), 1000)); 229 + var since_buf: [256]u8 = undefined; 230 + var cursor_buf: [256]u8 = undefined; 231 + const body = try store.writeBlobListJson( 232 + allocator, 233 + did, 234 + http_api.queryParam(request.url.raw, "since", &since_buf), 235 + http_api.queryParam(request.url.raw, "cursor", &cursor_buf), 236 + @min(http_api.queryLimit(request.url.raw, 500), 1000), 237 + ); 228 238 return http_api.json(request, .ok, body); 229 239 } 230 240 ··· 249 259 defer arena.deinit(); 250 260 const allocator = arena.allocator(); 251 261 252 - const body = try store.writeRepoListJson(allocator, http_api.queryLimit(request.url.raw, 500)); 262 + var cursor_buf: [512]u8 = undefined; 263 + const body = store.writeRepoListJson(allocator, http_api.queryParam(request.url.raw, "cursor", &cursor_buf), http_api.queryLimit(request.url.raw, 500)) catch |err| switch (err) { 264 + error.InvalidCursor => return http_api.xrpcError(request, .bad_request, "InvalidRequest", "Malformed cursor"), 265 + else => return err, 266 + }; 253 267 return http_api.json(request, .ok, body); 254 268 } 255 269

+424 -81

src/storage/store.zig

··· 12 12 13 13 pub const Error = error{ 14 14 InvalidCollection, 15 + InvalidCursor, 15 16 InvalidRecordKey, 16 17 InvalidRecordType, 18 + InvalidSwap, 19 + ValidationRequired, 17 20 InvalidReservedSigningKey, 18 21 MissingRecord, 19 22 MissingReservedSigningKey, ··· 359 362 collection: []const u8, 360 363 rkey: []const u8, 361 364 value: std.json.Value, 365 + swap: RecordSwap = .none, 362 366 }, 363 367 delete: struct { 364 368 collection: []const u8, 365 369 rkey: []const u8, 370 + swap: RecordSwap = .none, 366 371 }, 372 + }; 373 + 374 + pub const RecordSwap = union(enum) { 375 + none, 376 + missing, 377 + cid: []const u8, 378 + }; 379 + 380 + pub const ValidationMode = enum { 381 + known, 382 + require, 383 + skip, 384 + }; 385 + 386 + pub const WriteOptions = struct { 387 + swap_commit: ?[]const u8 = null, 388 + validate: ValidationMode = .known, 367 389 }; 368 390 369 391 const BlobRef = struct { ··· 1683 1705 } 1684 1706 1685 1707 pub fn applyWrites(allocator: std.mem.Allocator, account: auth.Account, ops: []const WriteOp) !WriteResult { 1686 - return applyWritesMeasured(allocator, account, ops, null); 1708 + return applyWritesWithOptions(allocator, account, ops, .{}); 1709 + } 1710 + 1711 + pub fn applyWritesWithOptions(allocator: std.mem.Allocator, account: auth.Account, ops: []const WriteOp, options: WriteOptions) !WriteResult { 1712 + return applyWritesMeasured(allocator, account, ops, options, null); 1687 1713 } 1688 1714 1689 1715 pub fn applyWritesProfiled(allocator: std.mem.Allocator, account: auth.Account, ops: []const WriteOp, profile: *WriteProfile) !WriteResult { 1690 1716 profile.* = .{}; 1691 - return applyWritesMeasured(allocator, account, ops, profile); 1717 + return applyWritesMeasured(allocator, account, ops, .{}, profile); 1692 1718 } 1693 1719 1694 - fn applyWritesMeasured(allocator: std.mem.Allocator, account: auth.Account, ops: []const WriteOp, profile: ?*WriteProfile) !WriteResult { 1720 + fn applyWritesMeasured(allocator: std.mem.Allocator, account: auth.Account, ops: []const WriteOp, options: WriteOptions, profile: ?*WriteProfile) !WriteResult { 1695 1721 const total_start = monotonicNs(); 1696 1722 if (ops.len == 0) return Error.MissingRecord; 1697 1723 ··· 1700 1726 .create => |create_op| { 1701 1727 if (zat.Nsid.parse(create_op.collection) == null) return Error.InvalidCollection; 1702 1728 if (create_op.rkey) |rkey| if (zat.Rkey.parse(rkey) == null) return Error.InvalidRecordKey; 1703 - try validateRecordForWrite(create_op.collection, create_op.rkey, create_op.value); 1729 + try validateRecordForWrite(create_op.collection, create_op.rkey, create_op.value, options.validate); 1704 1730 }, 1705 1731 .update => |update_op| { 1706 1732 if (zat.Nsid.parse(update_op.collection) == null) return Error.InvalidCollection; 1707 1733 if (zat.Rkey.parse(update_op.rkey) == null) return Error.InvalidRecordKey; 1708 - try validateRecordForWrite(update_op.collection, update_op.rkey, update_op.value); 1734 + try validateRecordForWrite(update_op.collection, update_op.rkey, update_op.value, options.validate); 1709 1735 }, 1710 1736 .delete => |delete_op| { 1711 1737 if (zat.Nsid.parse(delete_op.collection) == null) return Error.InvalidCollection; ··· 1727 1753 defer db_mutex.unlock(store_io); 1728 1754 const current = try latestCommitRawLocked(allocator, account.did); 1729 1755 const resolved_ops = try resolveWriteOpsLocked(allocator, ops); 1756 + try requireWriteSwapsLocked(allocator, account.did, current, options, resolved_ops); 1730 1757 break :blk .{ 1731 1758 .current = current, 1732 1759 .rev = try revForSeq(allocator, seq, if (current) |root| root.rev else null), ··· 1758 1785 for (resolved_ops) |op| switch (op) { 1759 1786 .create => |create_op| { 1760 1787 const rkey = create_op.rkey orelse return Error.InvalidRecordKey; 1761 - const record = try stageRecordWrite(allocator, &tree, account, create_op.collection, rkey, create_op.value, rev, seq, &record_blocks, &blob_refs); 1788 + const record = try stageRecordWrite(allocator, &tree, account, create_op.collection, rkey, create_op.value, options.validate, rev, seq, &record_blocks, &blob_refs); 1762 1789 try records.append(allocator, record); 1763 1790 }, 1764 1791 .update => |update_op| { 1765 - const record = try stageRecordWrite(allocator, &tree, account, update_op.collection, update_op.rkey, update_op.value, rev, seq, &record_blocks, &blob_refs); 1792 + const record = try stageRecordWrite(allocator, &tree, account, update_op.collection, update_op.rkey, update_op.value, options.validate, rev, seq, &record_blocks, &blob_refs); 1766 1793 try records.append(allocator, record); 1767 1794 }, 1768 1795 .delete => |delete_op| { ··· 1785 1812 errdefer conn.rollback(); 1786 1813 for (record_blocks.items) |block| { 1787 1814 try conn.exec( 1788 - \\INSERT INTO repo_blocks (did, cid, data) 1789 - \\VALUES (?, ?, ?) 1790 - \\ON CONFLICT(did, cid) DO UPDATE SET data = excluded.data 1791 - , .{ account.did, block.cid, zqlite.blob(block.data) }); 1815 + \\INSERT INTO repo_blocks (did, cid, data, repo_rev) 1816 + \\VALUES (?, ?, ?, ?) 1817 + \\ON CONFLICT(did, cid) DO UPDATE SET 1818 + \\ data = excluded.data, 1819 + \\ repo_rev = COALESCE(repo_blocks.repo_rev, excluded.repo_rev) 1820 + , .{ account.did, block.cid, zqlite.blob(block.data), rev }); 1792 1821 } 1793 1822 for (mst_blocks.items) |block| { 1794 1823 try conn.exec( 1795 - \\INSERT INTO repo_blocks (did, cid, data) 1796 - \\VALUES (?, ?, ?) 1797 - \\ON CONFLICT(did, cid) DO UPDATE SET data = excluded.data 1798 - , .{ account.did, try cidText(allocator, block.cid_raw), zqlite.blob(block.data) }); 1824 + \\INSERT INTO repo_blocks (did, cid, data, repo_rev) 1825 + \\VALUES (?, ?, ?, ?) 1826 + \\ON CONFLICT(did, cid) DO UPDATE SET 1827 + \\ data = excluded.data, 1828 + \\ repo_rev = COALESCE(repo_blocks.repo_rev, excluded.repo_rev) 1829 + , .{ account.did, try cidText(allocator, block.cid_raw), zqlite.blob(block.data), rev }); 1799 1830 } 1800 1831 try conn.exec( 1801 - \\INSERT INTO repo_blocks (did, cid, data) 1802 - \\VALUES (?, ?, ?) 1803 - \\ON CONFLICT(did, cid) DO UPDATE SET data = excluded.data 1804 - , .{ account.did, commit.cid, zqlite.blob(commit.data) }); 1832 + \\INSERT INTO repo_blocks (did, cid, data, repo_rev) 1833 + \\VALUES (?, ?, ?, ?) 1834 + \\ON CONFLICT(did, cid) DO UPDATE SET 1835 + \\ data = excluded.data, 1836 + \\ repo_rev = COALESCE(repo_blocks.repo_rev, excluded.repo_rev) 1837 + , .{ account.did, commit.cid, zqlite.blob(commit.data), rev }); 1805 1838 1806 1839 for (resolved_ops) |op| switch (op) { 1807 1840 .delete => |delete_op| { ··· 2314 2347 }; 2315 2348 } 2316 2349 2317 - pub fn writeBlobListJson(allocator: std.mem.Allocator, did: []const u8, limit: usize) ![]const u8 { 2350 + pub fn writeBlobListJson(allocator: std.mem.Allocator, did: []const u8, since: ?[]const u8, cursor: ?[]const u8, limit: usize) ![]const u8 { 2318 2351 db_mutex.lockUncancelable(store_io); 2319 2352 defer db_mutex.unlock(store_io); 2320 2353 try requireInitialized(); 2321 2354 2322 - var rows = try conn.rows( 2323 - \\SELECT DISTINCT b.cid 2324 - \\FROM blobs b 2325 - \\JOIN expected_blobs eb ON eb.blob_cid = b.cid 2326 - \\JOIN records r ON r.uri = eb.record_uri AND r.did = b.did 2327 - \\WHERE b.did = ? 2328 - \\ORDER BY b.cid ASC 2329 - \\LIMIT ? 2330 - , .{ did, @as(i64, @intCast(if (limit == 0) 500 else limit)) }); 2355 + const page_limit = @as(i64, @intCast(if (limit == 0) 500 else limit)); 2356 + var rows = if (since) |since_rev| blk: { 2357 + if (cursor) |after_cid| { 2358 + break :blk try conn.rows( 2359 + \\SELECT DISTINCT b.cid 2360 + \\FROM blobs b 2361 + \\JOIN expected_blobs eb ON eb.blob_cid = b.cid 2362 + \\JOIN records r ON r.uri = eb.record_uri AND r.did = b.did 2363 + \\WHERE b.did = ? AND r.rev > ? AND b.cid > ? 2364 + \\ORDER BY b.cid ASC 2365 + \\LIMIT ? 2366 + , .{ did, since_rev, after_cid, page_limit }); 2367 + } 2368 + break :blk try conn.rows( 2369 + \\SELECT DISTINCT b.cid 2370 + \\FROM blobs b 2371 + \\JOIN expected_blobs eb ON eb.blob_cid = b.cid 2372 + \\JOIN records r ON r.uri = eb.record_uri AND r.did = b.did 2373 + \\WHERE b.did = ? AND r.rev > ? 2374 + \\ORDER BY b.cid ASC 2375 + \\LIMIT ? 2376 + , .{ did, since_rev, page_limit }); 2377 + } else blk: { 2378 + if (cursor) |after_cid| { 2379 + break :blk try conn.rows( 2380 + \\SELECT DISTINCT b.cid 2381 + \\FROM blobs b 2382 + \\JOIN expected_blobs eb ON eb.blob_cid = b.cid 2383 + \\JOIN records r ON r.uri = eb.record_uri AND r.did = b.did 2384 + \\WHERE b.did = ? AND b.cid > ? 2385 + \\ORDER BY b.cid ASC 2386 + \\LIMIT ? 2387 + , .{ did, after_cid, page_limit }); 2388 + } 2389 + break :blk try conn.rows( 2390 + \\SELECT DISTINCT b.cid 2391 + \\FROM blobs b 2392 + \\JOIN expected_blobs eb ON eb.blob_cid = b.cid 2393 + \\JOIN records r ON r.uri = eb.record_uri AND r.did = b.did 2394 + \\WHERE b.did = ? 2395 + \\ORDER BY b.cid ASC 2396 + \\LIMIT ? 2397 + , .{ did, page_limit }); 2398 + }; 2331 2399 defer rows.deinit(); 2332 2400 2333 2401 var out: std.Io.Writer.Allocating = .init(allocator); ··· 2339 2407 if (!first) try out.writer.writeByte(','); 2340 2408 first = false; 2341 2409 const cid = row.text(0); 2342 - last_cid = cid; 2410 + last_cid = try allocator.dupe(u8, cid); 2343 2411 try out.writer.print("{f}", .{std.json.fmt(cid, .{})}); 2344 2412 } 2345 2413 if (rows.err) |err| return err; ··· 2349 2417 return out.toOwnedSlice(); 2350 2418 } 2351 2419 2352 - pub fn writeMissingBlobsJson(allocator: std.mem.Allocator, did: []const u8, limit: usize) ![]const u8 { 2420 + pub fn writeMissingBlobsJson(allocator: std.mem.Allocator, did: []const u8, cursor: ?[]const u8, limit: usize) ![]const u8 { 2353 2421 db_mutex.lockUncancelable(store_io); 2354 2422 defer db_mutex.unlock(store_io); 2355 2423 try requireInitialized(); 2356 2424 2357 - var rows = try conn.rows( 2358 - \\SELECT rb.blob_cid, rb.record_uri 2359 - \\FROM expected_blobs rb 2360 - \\JOIN records r ON r.uri = rb.record_uri 2361 - \\LEFT JOIN blobs b ON b.cid = rb.blob_cid AND b.did = r.did 2362 - \\WHERE r.did = ? AND b.cid IS NULL 2363 - \\ORDER BY rb.blob_cid ASC 2364 - \\LIMIT ? 2365 - , .{ did, @as(i64, @intCast(if (limit == 0) 500 else limit)) }); 2425 + const page_limit = @as(i64, @intCast(if (limit == 0) 500 else limit)); 2426 + var rows = if (cursor) |after_cid| 2427 + try conn.rows( 2428 + \\SELECT rb.blob_cid, rb.record_uri 2429 + \\FROM expected_blobs rb 2430 + \\JOIN records r ON r.uri = rb.record_uri 2431 + \\LEFT JOIN blobs b ON b.cid = rb.blob_cid AND b.did = r.did 2432 + \\WHERE r.did = ? AND b.cid IS NULL AND rb.blob_cid > ? 2433 + \\ORDER BY rb.blob_cid ASC 2434 + \\LIMIT ? 2435 + , .{ did, after_cid, page_limit }) 2436 + else 2437 + try conn.rows( 2438 + \\SELECT rb.blob_cid, rb.record_uri 2439 + \\FROM expected_blobs rb 2440 + \\JOIN records r ON r.uri = rb.record_uri 2441 + \\LEFT JOIN blobs b ON b.cid = rb.blob_cid AND b.did = r.did 2442 + \\WHERE r.did = ? AND b.cid IS NULL 2443 + \\ORDER BY rb.blob_cid ASC 2444 + \\LIMIT ? 2445 + , .{ did, page_limit }); 2366 2446 defer rows.deinit(); 2367 2447 2368 2448 var out: std.Io.Writer.Allocating = .init(allocator); ··· 2374 2454 if (!first) try out.writer.writeByte(','); 2375 2455 first = false; 2376 2456 const cid = row.text(0); 2377 - last_cid = cid; 2457 + last_cid = try allocator.dupe(u8, cid); 2378 2458 try out.writer.print( 2379 2459 "{{\"cid\":{f},\"recordUri\":{f}}}", 2380 2460 .{ std.json.fmt(cid, .{}), std.json.fmt(row.text(1), .{}) }, ··· 2453 2533 2454 2534 for (blocks) |block| { 2455 2535 try conn.exec( 2456 - \\INSERT INTO repo_blocks (did, cid, data) 2457 - \\VALUES (?, ?, ?) 2458 - \\ON CONFLICT(did, cid) DO UPDATE SET data = excluded.data 2459 - , .{ account.did, block.cid, zqlite.blob(block.data) }); 2536 + \\INSERT INTO repo_blocks (did, cid, data, repo_rev) 2537 + \\VALUES (?, ?, ?, ?) 2538 + \\ON CONFLICT(did, cid) DO UPDATE SET 2539 + \\ data = excluded.data, 2540 + \\ repo_rev = COALESCE(repo_blocks.repo_rev, excluded.repo_rev) 2541 + , .{ account.did, block.cid, zqlite.blob(block.data), rev }); 2460 2542 } 2461 2543 for (records) |record| { 2462 2544 const uri = try std.fmt.allocPrint(std.heap.page_allocator, "at://{s}/{s}/{s}", .{ account.did, record.collection, record.rkey }); ··· 2494 2576 } 2495 2577 2496 2578 pub fn writeRepoCar(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { 2579 + return writeRepoCarSince(allocator, did, null); 2580 + } 2581 + 2582 + pub fn writeRepoCarSince(allocator: std.mem.Allocator, did: []const u8, since: ?[]const u8) ![]const u8 { 2497 2583 db_mutex.lockUncancelable(store_io); 2498 2584 defer db_mutex.unlock(store_io); 2499 2585 try requireInitialized(); ··· 2502 2588 const root_raw = try zat.multibase.base32lower.decode(allocator, root.cid[1..]); 2503 2589 const car_root = zat.cbor.Cid{ .raw = root_raw }; 2504 2590 2505 - var rows = try conn.rows( 2506 - \\SELECT cid, data 2507 - \\FROM repo_blocks 2508 - \\WHERE did = ? 2509 - \\ORDER BY cid ASC 2510 - , .{did}); 2591 + var rows = if (since) |since_rev| 2592 + try conn.rows( 2593 + \\SELECT cid, data 2594 + \\FROM repo_blocks 2595 + \\WHERE did = ? AND (repo_rev IS NULL OR repo_rev > ?) 2596 + \\ORDER BY repo_rev DESC, cid DESC 2597 + , .{ did, since_rev }) 2598 + else 2599 + try conn.rows( 2600 + \\SELECT cid, data 2601 + \\FROM repo_blocks 2602 + \\WHERE did = ? 2603 + \\ORDER BY cid ASC 2604 + , .{did}); 2511 2605 defer rows.deinit(); 2512 2606 2513 2607 var blocks: std.ArrayList(zat.car.Block) = .empty; ··· 2528 2622 return zat.car.writeAlloc(allocator, c); 2529 2623 } 2530 2624 2531 - pub fn writeRepoListJson(allocator: std.mem.Allocator, limit: usize) ![]const u8 { 2625 + pub fn writeRepoListJson(allocator: std.mem.Allocator, cursor: ?[]const u8, limit: usize) ![]const u8 { 2532 2626 db_mutex.lockUncancelable(store_io); 2533 2627 defer db_mutex.unlock(store_io); 2534 2628 try requireInitialized(); 2535 2629 2536 2630 const actual_limit = if (limit == 0) 500 else @min(limit, 1000); 2537 - var rows = try conn.rows( 2538 - \\SELECT a.did, c.cid, c.rev, a.activated_at, a.deactivated_at 2539 - \\FROM accounts a 2540 - \\JOIN ( 2541 - \\ SELECT did, MAX(seq) AS seq 2542 - \\ FROM commits 2543 - \\ GROUP BY did 2544 - \\) latest ON latest.did = a.did 2545 - \\JOIN commits c ON c.did = latest.did AND c.seq = latest.seq 2546 - \\ORDER BY a.did ASC 2547 - \\LIMIT ? 2548 - , .{@as(i64, @intCast(actual_limit))}); 2631 + const parsed_cursor = try parseRepoListCursor(cursor); 2632 + var rows = if (parsed_cursor) |after| 2633 + try conn.rows( 2634 + \\SELECT a.did, c.cid, c.rev, a.activated_at, a.deactivated_at, a.created_at 2635 + \\FROM accounts a 2636 + \\JOIN ( 2637 + \\ SELECT did, MAX(seq) AS seq 2638 + \\ FROM commits 2639 + \\ GROUP BY did 2640 + \\) latest ON latest.did = a.did 2641 + \\JOIN commits c ON c.did = latest.did AND c.seq = latest.seq 2642 + \\WHERE a.created_at > ? OR (a.created_at = ? AND a.did > ?) 2643 + \\ORDER BY a.created_at ASC, a.did ASC 2644 + \\LIMIT ? 2645 + , .{ after.created_at, after.created_at, after.did, @as(i64, @intCast(actual_limit)) }) 2646 + else 2647 + try conn.rows( 2648 + \\SELECT a.did, c.cid, c.rev, a.activated_at, a.deactivated_at, a.created_at 2649 + \\FROM accounts a 2650 + \\JOIN ( 2651 + \\ SELECT did, MAX(seq) AS seq 2652 + \\ FROM commits 2653 + \\ GROUP BY did 2654 + \\) latest ON latest.did = a.did 2655 + \\JOIN commits c ON c.did = latest.did AND c.seq = latest.seq 2656 + \\ORDER BY a.created_at ASC, a.did ASC 2657 + \\LIMIT ? 2658 + , .{@as(i64, @intCast(actual_limit))}); 2549 2659 defer rows.deinit(); 2550 2660 2551 2661 var out: std.Io.Writer.Allocating = .init(allocator); ··· 2554 2664 try json.beginObject(); 2555 2665 try json.objectField("repos"); 2556 2666 try json.beginArray(); 2667 + var last_created_at: ?i64 = null; 2668 + var last_did: ?[]const u8 = null; 2557 2669 while (rows.next()) |row| { 2558 2670 const did = row.text(0); 2671 + last_did = try allocator.dupe(u8, did); 2672 + last_created_at = row.int(5); 2559 2673 const active = row.nullableInt(3) != null and row.nullableInt(4) == null; 2560 2674 try json.beginObject(); 2561 2675 try json.objectField("did"); ··· 2574 2688 } 2575 2689 if (rows.err) |err| return err; 2576 2690 try json.endArray(); 2691 + if (last_created_at) |created_at| { 2692 + try json.objectField("cursor"); 2693 + try json.write(try std.fmt.allocPrint(allocator, "{d}:{s}", .{ created_at, last_did.? })); 2694 + } 2577 2695 try json.endObject(); 2578 2696 return out.toOwnedSlice(); 2579 2697 } 2580 2698 2699 + const RepoListCursor = struct { 2700 + created_at: i64, 2701 + did: []const u8, 2702 + }; 2703 + 2704 + fn parseRepoListCursor(cursor: ?[]const u8) Error!?RepoListCursor { 2705 + const raw = cursor orelse return null; 2706 + const sep = std.mem.indexOfScalar(u8, raw, ':') orelse return Error.InvalidCursor; 2707 + if (sep == 0 or sep + 1 >= raw.len) return Error.InvalidCursor; 2708 + return .{ 2709 + .created_at = std.fmt.parseInt(i64, raw[0..sep], 10) catch return Error.InvalidCursor, 2710 + .did = raw[sep + 1 ..], 2711 + }; 2712 + } 2713 + 2581 2714 pub fn writeLatestCommitJson(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { 2582 2715 db_mutex.lockUncancelable(store_io); 2583 2716 defer db_mutex.unlock(store_io); ··· 2668 2801 allocator: std.mem.Allocator, 2669 2802 did: []const u8, 2670 2803 collection: []const u8, 2804 + cursor: ?[]const u8, 2805 + reverse: bool, 2671 2806 limit: usize, 2672 2807 ) ![]const u8 { 2673 2808 db_mutex.lockUncancelable(store_io); 2674 2809 defer db_mutex.unlock(store_io); 2675 2810 try requireInitialized(); 2676 2811 2677 - var rows = try conn.rows( 2678 - \\SELECT r.did, r.collection, r.rkey, r.cid, rb.data, r.rev, r.seq 2679 - \\FROM records r 2680 - \\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid 2681 - \\WHERE r.did = ? AND r.collection = ? 2682 - \\ORDER BY r.seq DESC 2683 - \\LIMIT ? 2684 - , .{ did, collection, @as(i64, @intCast(if (limit == 0) 100 else limit)) }); 2812 + const page_limit = @as(i64, @intCast(if (limit == 0) 100 else limit)); 2813 + var rows = if (cursor) |rkey_cursor| blk: { 2814 + if (reverse) { 2815 + break :blk try conn.rows( 2816 + \\SELECT r.did, r.collection, r.rkey, r.cid, rb.data, r.rev, r.seq 2817 + \\FROM records r 2818 + \\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid 2819 + \\WHERE r.did = ? AND r.collection = ? AND r.rkey > ? 2820 + \\ORDER BY r.rkey ASC 2821 + \\LIMIT ? 2822 + , .{ did, collection, rkey_cursor, page_limit }); 2823 + } 2824 + break :blk try conn.rows( 2825 + \\SELECT r.did, r.collection, r.rkey, r.cid, rb.data, r.rev, r.seq 2826 + \\FROM records r 2827 + \\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid 2828 + \\WHERE r.did = ? AND r.collection = ? AND r.rkey < ? 2829 + \\ORDER BY r.rkey DESC 2830 + \\LIMIT ? 2831 + , .{ did, collection, rkey_cursor, page_limit }); 2832 + } else blk: { 2833 + if (reverse) { 2834 + break :blk try conn.rows( 2835 + \\SELECT r.did, r.collection, r.rkey, r.cid, rb.data, r.rev, r.seq 2836 + \\FROM records r 2837 + \\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid 2838 + \\WHERE r.did = ? AND r.collection = ? 2839 + \\ORDER BY r.rkey ASC 2840 + \\LIMIT ? 2841 + , .{ did, collection, page_limit }); 2842 + } 2843 + break :blk try conn.rows( 2844 + \\SELECT r.did, r.collection, r.rkey, r.cid, rb.data, r.rev, r.seq 2845 + \\FROM records r 2846 + \\JOIN repo_blocks rb ON rb.did = r.did AND rb.cid = r.cid 2847 + \\WHERE r.did = ? AND r.collection = ? 2848 + \\ORDER BY r.rkey DESC 2849 + \\LIMIT ? 2850 + , .{ did, collection, page_limit }); 2851 + }; 2685 2852 defer rows.deinit(); 2686 2853 2687 2854 var out: std.Io.Writer.Allocating = .init(allocator); 2688 2855 defer out.deinit(); 2689 2856 try out.writer.writeAll("{\"records\":["); 2690 2857 var first = true; 2858 + var last_rkey: ?[]const u8 = null; 2691 2859 while (rows.next()) |row| { 2692 2860 const record = try recordFromRow(row, std.heap.page_allocator); 2693 2861 if (!first) try out.writer.writeByte(','); 2694 2862 first = false; 2863 + last_rkey = try allocator.dupe(u8, row.text(2)); 2695 2864 const uri = try record.uri(allocator); 2696 2865 defer allocator.free(uri); 2697 2866 try out.writer.print( ··· 2700 2869 ); 2701 2870 } 2702 2871 if (rows.err) |err| return err; 2703 - try out.writer.writeAll("]}"); 2872 + try out.writer.writeByte(']'); 2873 + if (last_rkey) |rkey| { 2874 + try out.writer.print(",\"cursor\":{f}", .{std.json.fmt(rkey, .{})}); 2875 + } 2876 + try out.writer.writeByte('}'); 2704 2877 return out.toOwnedSlice(); 2705 2878 } 2706 2879 ··· 2716 2889 ) !PreparedRecord { 2717 2890 if (zat.Nsid.parse(collection) == null) return Error.InvalidCollection; 2718 2891 if (zat.Rkey.parse(rkey) == null) return Error.InvalidRecordKey; 2719 - try validateRecordForWrite(collection, rkey, value); 2892 + try validateRecordForWrite(collection, rkey, value, .known); 2720 2893 2721 2894 const cbor_value = try jsonToDagCbor(allocator, value); 2722 2895 const record_bytes = try zat.cbor.encodeAlloc(allocator, cbor_value); ··· 2724 2897 return .{ 2725 2898 .cid = try cidText(allocator, record_cid.raw), 2726 2899 .value_json = try stringifyValue(allocator, value), 2727 - .validation_status = validationStatusForRecord(collection), 2900 + .validation_status = validationStatusForRecord(collection, .known), 2728 2901 }; 2729 2902 } 2730 2903 ··· 3438 3611 try migrateBlobTable(); 3439 3612 try migrateAppPreferencesTable(); 3440 3613 inline for (post_schema_statements) |sql| try conn.execNoArgs(sql); 3614 + conn.execNoArgs("ALTER TABLE repo_blocks ADD COLUMN repo_rev TEXT") catch {}; 3615 + try migrateRepoBlockRevs(); 3441 3616 try migrateRecordsTable(); 3442 3617 try validateRecordBlocksPresent(); 3443 3618 try migrateSeqEventsSync11(); ··· 3613 3788 try conn.execNoArgs("PRAGMA foreign_keys = ON"); 3614 3789 try conn.execNoArgs("CREATE INDEX IF NOT EXISTS records_collection_idx ON records (did, collection, seq DESC)"); 3615 3790 try conn.execNoArgs("CREATE INDEX IF NOT EXISTS records_cid_idx ON records (cid)"); 3791 + } 3792 + 3793 + fn migrateRepoBlockRevs() !void { 3794 + try conn.execNoArgs( 3795 + \\UPDATE repo_blocks 3796 + \\SET repo_rev = ( 3797 + \\ SELECT r.rev 3798 + \\ FROM records r 3799 + \\ WHERE r.did = repo_blocks.did AND r.cid = repo_blocks.cid 3800 + \\ LIMIT 1 3801 + \\) 3802 + \\WHERE repo_rev IS NULL 3803 + \\ AND EXISTS ( 3804 + \\ SELECT 1 FROM records r 3805 + \\ WHERE r.did = repo_blocks.did AND r.cid = repo_blocks.cid 3806 + \\ ) 3807 + ); 3808 + try conn.execNoArgs( 3809 + \\UPDATE repo_blocks 3810 + \\SET repo_rev = ( 3811 + \\ SELECT c.rev 3812 + \\ FROM commits c 3813 + \\ WHERE c.did = repo_blocks.did AND c.cid = repo_blocks.cid 3814 + \\ LIMIT 1 3815 + \\) 3816 + \\WHERE repo_rev IS NULL 3817 + \\ AND EXISTS ( 3818 + \\ SELECT 1 FROM commits c 3819 + \\ WHERE c.did = repo_blocks.did AND c.cid = repo_blocks.cid 3820 + \\ ) 3821 + ); 3822 + try conn.execNoArgs("CREATE INDEX IF NOT EXISTS repo_blocks_rev_idx ON repo_blocks (did, repo_rev DESC, cid DESC)"); 3616 3823 } 3617 3824 3618 3825 fn recordsTableHasValueJsonColumn() !bool { ··· 3896 4103 return resolved; 3897 4104 } 3898 4105 4106 + fn requireWriteSwapsLocked(allocator: std.mem.Allocator, did: []const u8, current: ?CurrentCommit, options: WriteOptions, ops: []const WriteOp) !void { 4107 + if (options.swap_commit) |expected| { 4108 + if (current == null or !std.mem.eql(u8, current.?.commit_cid_text, expected)) return Error.InvalidSwap; 4109 + } 4110 + 4111 + for (ops) |op| switch (op) { 4112 + .create => {}, 4113 + .update => |update_op| try requireRecordSwapLocked(allocator, did, update_op.collection, update_op.rkey, update_op.swap), 4114 + .delete => |delete_op| try requireRecordSwapLocked(allocator, did, delete_op.collection, delete_op.rkey, delete_op.swap), 4115 + }; 4116 + } 4117 + 4118 + fn requireRecordSwapLocked(allocator: std.mem.Allocator, did: []const u8, collection: []const u8, rkey: []const u8, swap: RecordSwap) !void { 4119 + switch (swap) { 4120 + .none => {}, 4121 + .missing => { 4122 + if (try currentRecordCidLocked(allocator, did, collection, rkey) != null) return Error.InvalidSwap; 4123 + }, 4124 + .cid => |expected| { 4125 + const actual = try currentRecordCidLocked(allocator, did, collection, rkey) orelse return Error.InvalidSwap; 4126 + if (!std.mem.eql(u8, actual, expected)) return Error.InvalidSwap; 4127 + }, 4128 + } 4129 + } 4130 + 3899 4131 fn previousRecordCidsLocked(allocator: std.mem.Allocator, did: []const u8, ops: []const WriteOp) ![]?[]const u8 { 3900 4132 var cids = try allocator.alloc(?[]const u8, ops.len); 3901 4133 for (ops, 0..) |op, idx| { ··· 3920 4152 return try allocator.dupe(u8, row.?.text(0)); 3921 4153 } 3922 4154 3923 - fn validateRecordForWrite(collection: []const u8, rkey: ?[]const u8, value: std.json.Value) Error!void { 4155 + fn validateRecordForWrite(collection: []const u8, rkey: ?[]const u8, value: std.json.Value, mode: ValidationMode) Error!void { 3924 4156 const object = switch (value) { 3925 4157 .object => |object| object, 3926 4158 else => return Error.InvalidRecordType, ··· 3930 4162 else => return Error.InvalidRecordType, 3931 4163 }; 3932 4164 if (!std.mem.eql(u8, record_type, collection)) return Error.InvalidRecordType; 4165 + if (mode == .require and !knownRecordType(record_type)) return Error.ValidationRequired; 3933 4166 3934 4167 if (rkey) |key| try validateKnownRecordKey(record_type, key); 3935 4168 } ··· 3947 4180 3948 4181 const KnownRecordKeyRule = enum { tid, literal_self }; 3949 4182 3950 - fn validationStatusForRecord(record_type: []const u8) []const u8 { 3951 - return if (knownRecordType(record_type)) "valid" else "unknown"; 4183 + fn validationStatusForRecord(record_type: []const u8, mode: ValidationMode) []const u8 { 4184 + return if (mode != .skip and knownRecordType(record_type)) "valid" else "unknown"; 3952 4185 } 3953 4186 3954 4187 fn knownRecordType(record_type: []const u8) bool { ··· 4012 4245 collection: []const u8, 4013 4246 rkey: []const u8, 4014 4247 value: std.json.Value, 4248 + validation_mode: ValidationMode, 4015 4249 rev: []const u8, 4016 4250 seq: u64, 4017 4251 blocks: *std.ArrayList(ImportedBlock), ··· 4042 4276 .rkey = try allocator.dupe(u8, rkey), 4043 4277 .cid = record_cid_text, 4044 4278 .value_json = try stringifyValue(allocator, value), 4045 - .validation_status = validationStatusForRecord(collection), 4279 + .validation_status = validationStatusForRecord(collection, validation_mode), 4046 4280 .rev = try allocator.dupe(u8, rev), 4047 4281 .seq = seq, 4048 4282 }; ··· 4307 4541 .rkey = try allocator.dupe(u8, row.text(2)), 4308 4542 .cid = try allocator.dupe(u8, row.text(3)), 4309 4543 .value_json = try recordJsonFromBlock(allocator, record_bytes), 4310 - .validation_status = validationStatusForRecord(row.text(1)), 4544 + .validation_status = validationStatusForRecord(row.text(1), .known), 4311 4545 .rev = try allocator.dupe(u8, row.text(5)), 4312 4546 .seq = @intCast(row.int(6)), 4313 4547 }; ··· 5198 5432 \\ did TEXT NOT NULL REFERENCES accounts(did) ON DELETE CASCADE, 5199 5433 \\ cid TEXT NOT NULL, 5200 5434 \\ data BLOB NOT NULL, 5435 + \\ repo_rev TEXT, 5201 5436 \\ PRIMARY KEY (did, cid) 5202 5437 \\) 5203 5438 , 5439 + "CREATE INDEX IF NOT EXISTS repo_blocks_rev_idx ON repo_blocks (did, repo_rev DESC, cid DESC)", 5204 5440 }; 5205 5441 5206 5442 test "persists records in sqlite" { ··· 5238 5474 try std.testing.expectEqualStrings(record.cid, fetched.cid); 5239 5475 try std.testing.expect(std.mem.indexOf(u8, fetched.value_json, "hello") != null); 5240 5476 5241 - const listed = try writeListJson(allocator, account.did, "app.bsky.feed.post", 10); 5477 + const listed = try writeListJson(allocator, account.did, "app.bsky.feed.post", null, false, 10); 5242 5478 try std.testing.expect(std.mem.indexOf(u8, listed, "hello") != null); 5243 5479 5244 5480 const repo_car = try writeRepoCar(allocator, account.did); ··· 5256 5492 .{ account.did, record.cid }, 5257 5493 ); 5258 5494 try std.testing.expect(get(account.did, "app.bsky.feed.post", "3ztest") == null); 5495 + } 5496 + 5497 + test "repo writes enforce swap and explicit validation preconditions" { 5498 + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); 5499 + defer arena.deinit(); 5500 + const allocator = arena.allocator(); 5501 + 5502 + try init(std.Options.debug_io, ":memory:"); 5503 + defer close(); 5504 + 5505 + const account = try createAccount( 5506 + allocator, 5507 + "swaps.test", 5508 + "swaps@test.com", 5509 + "password", 5510 + "did:plc:swapscheck", 5511 + true, 5512 + ); 5513 + 5514 + const first = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"app.bsky.feed.post\",\"text\":\"before\"}", .{}); 5515 + defer first.deinit(); 5516 + const created_result = try applyWritesWithOptions(allocator, account, &.{.{ .create = .{ 5517 + .collection = "app.bsky.feed.post", 5518 + .rkey = "3zswap", 5519 + .value = first.value, 5520 + } }}, .{}); 5521 + const first_cid = created_result.records[0].cid; 5522 + 5523 + const second = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"app.bsky.feed.post\",\"text\":\"after\"}", .{}); 5524 + defer second.deinit(); 5525 + try std.testing.expectError(Error.InvalidSwap, applyWritesWithOptions(allocator, account, &.{.{ .update = .{ 5526 + .collection = "app.bsky.feed.post", 5527 + .rkey = "3zswap", 5528 + .value = second.value, 5529 + } }}, .{ .swap_commit = "bafkreiwrongcommit" })); 5530 + try std.testing.expectError(Error.InvalidSwap, applyWritesWithOptions(allocator, account, &.{.{ .update = .{ 5531 + .collection = "app.bsky.feed.post", 5532 + .rkey = "3zswap", 5533 + .value = second.value, 5534 + .swap = .{ .cid = "bafkreiwrongrecord" }, 5535 + } }}, .{})); 5536 + 5537 + const updated_result = try applyWritesWithOptions(allocator, account, &.{.{ .update = .{ 5538 + .collection = "app.bsky.feed.post", 5539 + .rkey = "3zswap", 5540 + .value = second.value, 5541 + .swap = .{ .cid = first_cid }, 5542 + } }}, .{ .swap_commit = created_result.commit.cid }); 5543 + try std.testing.expectEqualStrings("after", (try std.json.parseFromSlice(std.json.Value, allocator, updated_result.records[0].value_json, .{})).value.object.get("text").?.string); 5544 + 5545 + const custom = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"example.test.record\",\"text\":\"custom\"}", .{}); 5546 + defer custom.deinit(); 5547 + try std.testing.expectError(Error.ValidationRequired, applyWritesWithOptions(allocator, account, &.{.{ .create = .{ 5548 + .collection = "example.test.record", 5549 + .rkey = "custom", 5550 + .value = custom.value, 5551 + } }}, .{ .validate = .require })); 5552 + 5553 + const custom_result = try applyWritesWithOptions(allocator, account, &.{.{ .create = .{ 5554 + .collection = "example.test.record", 5555 + .rkey = "custom", 5556 + .value = custom.value, 5557 + } }}, .{ .validate = .skip }); 5558 + try std.testing.expectEqualStrings("unknown", custom_result.records[0].validation_status); 5559 + } 5560 + 5561 + test "getRepo since filters repo blocks by revision" { 5562 + var arena = std.heap.ArenaAllocator.init(std.testing.allocator); 5563 + defer arena.deinit(); 5564 + const allocator = arena.allocator(); 5565 + 5566 + try init(std.Options.debug_io, ":memory:"); 5567 + defer close(); 5568 + 5569 + const account = try createAccount( 5570 + allocator, 5571 + "since.test", 5572 + "since@test.com", 5573 + "password", 5574 + "did:plc:sincecheck", 5575 + true, 5576 + ); 5577 + 5578 + const first = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"app.bsky.feed.post\",\"text\":\"first\"}", .{}); 5579 + defer first.deinit(); 5580 + const first_result = try applyWritesWithOptions(allocator, account, &.{.{ .create = .{ 5581 + .collection = "app.bsky.feed.post", 5582 + .rkey = "3zsincea", 5583 + .value = first.value, 5584 + } }}, .{}); 5585 + 5586 + const second = try std.json.parseFromSlice(std.json.Value, allocator, "{\"$type\":\"app.bsky.feed.post\",\"text\":\"second\"}", .{}); 5587 + defer second.deinit(); 5588 + const second_result = try applyWritesWithOptions(allocator, account, &.{.{ .create = .{ 5589 + .collection = "app.bsky.feed.post", 5590 + .rkey = "3zsinceb", 5591 + .value = second.value, 5592 + } }}, .{}); 5593 + 5594 + const full_car = try zat.car.read(allocator, try writeRepoCar(allocator, account.did)); 5595 + const diff_car = try zat.car.read(allocator, try writeRepoCarSince(allocator, account.did, first_result.commit.rev)); 5596 + const empty_diff_car = try zat.car.read(allocator, try writeRepoCarSince(allocator, account.did, second_result.commit.rev)); 5597 + 5598 + try std.testing.expect(diff_car.blocks.len < full_car.blocks.len); 5599 + try std.testing.expect(diff_car.blocks.len > 0); 5600 + try std.testing.expectEqual(@as(usize, 0), empty_diff_car.blocks.len); 5601 + try std.testing.expectEqualStrings(second_result.commit.cid, try cidText(allocator, empty_diff_car.roots[0].raw)); 5259 5602 } 5260 5603 5261 5604 test "permissioned spaces store self-owned records outside public repo" {

Configure Feed

Configure Feed