+2

src/root.zig

View file Reviewed

··· 2 2 pub const auth = @import("auth.zig"); 3 3 pub const repo = @import("repo.zig"); 4 4 pub const server = @import("server.zig"); 5 5 + pub const store = @import("store.zig"); 5 6 pub const syntax = @import("syntax.zig"); 6 7 pub const xrpc = @import("xrpc.zig"); 7 8 pub const zat = @import("zat"); ··· 10 11 _ = atid; 11 12 _ = repo; 12 13 _ = server; 14 14 + _ = store; 13 15 _ = syntax; 14 16 _ = xrpc; 15 17 }

+189

src/server.zig

View file Reviewed

··· 1 1 const std = @import("std"); 2 2 const auth = @import("auth.zig"); 3 3 + const store = @import("store.zig"); 3 4 const xrpc = @import("xrpc.zig"); 4 5 const zat = @import("zat"); 5 6 ··· 19 20 create_session, 20 21 get_session, 21 22 get_service_auth, 23 23 + repo_create_record, 24 24 + repo_put_record, 25 25 + repo_get_record, 26 26 + repo_list_records, 27 27 + repo_delete_record, 22 28 not_found, 23 29 }; 24 30 ··· 40 46 if (method == .GET and std.mem.eql(u8, path, "/xrpc/com.atproto.server.getServiceAuth")) { 41 47 return .get_service_auth; 42 48 } 49 49 + if (method == .POST and std.mem.eql(u8, path, "/xrpc/com.atproto.repo.createRecord")) { 50 50 + return .repo_create_record; 51 51 + } 52 52 + if (method == .POST and std.mem.eql(u8, path, "/xrpc/com.atproto.repo.putRecord")) { 53 53 + return .repo_put_record; 54 54 + } 55 55 + if (method == .GET and std.mem.eql(u8, path, "/xrpc/com.atproto.repo.getRecord")) { 56 56 + return .repo_get_record; 57 57 + } 58 58 + if (method == .GET and std.mem.eql(u8, path, "/xrpc/com.atproto.repo.listRecords")) { 59 59 + return .repo_list_records; 60 60 + } 61 61 + if (method == .POST and std.mem.eql(u8, path, "/xrpc/com.atproto.repo.deleteRecord")) { 62 62 + return .repo_delete_record; 63 63 + } 43 64 return .not_found; 44 65 } 45 66 ··· 96 117 .create_session => try createSession(request), 97 118 .get_session => try getSession(request), 98 119 .get_service_auth => try getServiceAuth(request), 120 120 + .repo_create_record => try repoCreateRecord(request), 121 121 + .repo_put_record => try repoPutRecord(request), 122 122 + .repo_get_record => try repoGetRecord(request), 123 123 + .repo_list_records => try repoListRecords(request), 124 124 + .repo_delete_record => try repoDeleteRecord(request), 99 125 .not_found => try xrpcError(request, .not_found, "MethodNotImplemented", "Method not implemented"), 100 126 } 101 127 } ··· 185 211 return json(request, .ok, body_out); 186 212 } 187 213 214 214 + fn repoCreateRecord(request: *http.Server.Request) !void { 215 215 + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); 216 216 + defer arena.deinit(); 217 217 + const allocator = arena.allocator(); 218 218 + 219 219 + const account = requireBearerAccount(request, allocator) catch |err| switch (err) { 220 220 + error.AuthRequired => return xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"), 221 221 + error.InvalidToken => return xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"), 222 222 + }; 223 223 + 224 224 + var body_buf: [65536]u8 = undefined; 225 225 + const body = try readBody(request, &body_buf); 226 226 + const parsed = try parseJsonBody(request, allocator, body); 227 227 + const collection = zat.json.getString(parsed.value, "collection") orelse { 228 228 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing collection"); 229 229 + }; 230 230 + const rkey = zat.json.getString(parsed.value, "rkey"); 231 231 + const record_value = recordBodyValue(parsed.value) orelse { 232 232 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing record"); 233 233 + }; 234 234 + 235 235 + const record = store.create(allocator, account, collection, rkey, record_value) catch |err| switch (err) { 236 236 + error.InvalidCollection => return xrpcError(request, .bad_request, "InvalidRequest", "Invalid collection"), 237 237 + error.InvalidRecordKey => return xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"), 238 238 + else => return err, 239 239 + }; 240 240 + return writeRecordRef(request, allocator, record); 241 241 + } 242 242 + 243 243 + fn repoPutRecord(request: *http.Server.Request) !void { 244 244 + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); 245 245 + defer arena.deinit(); 246 246 + const allocator = arena.allocator(); 247 247 + 248 248 + const account = requireBearerAccount(request, allocator) catch |err| switch (err) { 249 249 + error.AuthRequired => return xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"), 250 250 + error.InvalidToken => return xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"), 251 251 + }; 252 252 + 253 253 + var body_buf: [65536]u8 = undefined; 254 254 + const body = try readBody(request, &body_buf); 255 255 + const parsed = try parseJsonBody(request, allocator, body); 256 256 + const collection = zat.json.getString(parsed.value, "collection") orelse { 257 257 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing collection"); 258 258 + }; 259 259 + const rkey = zat.json.getString(parsed.value, "rkey") orelse { 260 260 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing rkey"); 261 261 + }; 262 262 + const record_value = recordBodyValue(parsed.value) orelse { 263 263 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing record"); 264 264 + }; 265 265 + 266 266 + const record = store.put(account, collection, rkey, record_value) catch |err| switch (err) { 267 267 + error.InvalidCollection => return xrpcError(request, .bad_request, "InvalidRequest", "Invalid collection"), 268 268 + error.InvalidRecordKey => return xrpcError(request, .bad_request, "InvalidRequest", "Invalid rkey"), 269 269 + else => return err, 270 270 + }; 271 271 + return writeRecordRef(request, allocator, record); 272 272 + } 273 273 + 274 274 + fn repoGetRecord(request: *http.Server.Request) !void { 275 275 + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); 276 276 + defer arena.deinit(); 277 277 + const allocator = arena.allocator(); 278 278 + 279 279 + var repo_buf: [256]u8 = undefined; 280 280 + const repo = queryParam(request.head.target, "repo", &repo_buf) orelse { 281 281 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing repo"); 282 282 + }; 283 283 + const account = store.resolveRepo(repo) orelse { 284 284 + return xrpcError(request, .not_found, "RepoNotFound", "Repo not found"); 285 285 + }; 286 286 + var collection_buf: [256]u8 = undefined; 287 287 + const collection = queryParam(request.head.target, "collection", &collection_buf) orelse { 288 288 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing collection"); 289 289 + }; 290 290 + var rkey_buf: [512]u8 = undefined; 291 291 + const rkey = queryParam(request.head.target, "rkey", &rkey_buf) orelse { 292 292 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing rkey"); 293 293 + }; 294 294 + 295 295 + const record = store.get(account.did, collection, rkey) orelse { 296 296 + return xrpcError(request, .not_found, "RecordNotFound", "Record not found"); 297 297 + }; 298 298 + const body = try store.writeRecordJson(allocator, record); 299 299 + return json(request, .ok, body); 300 300 + } 301 301 + 302 302 + fn repoListRecords(request: *http.Server.Request) !void { 303 303 + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); 304 304 + defer arena.deinit(); 305 305 + const allocator = arena.allocator(); 306 306 + 307 307 + var repo_buf: [256]u8 = undefined; 308 308 + const repo = queryParam(request.head.target, "repo", &repo_buf) orelse { 309 309 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing repo"); 310 310 + }; 311 311 + const account = store.resolveRepo(repo) orelse { 312 312 + return xrpcError(request, .not_found, "RepoNotFound", "Repo not found"); 313 313 + }; 314 314 + var collection_buf: [256]u8 = undefined; 315 315 + const collection = queryParam(request.head.target, "collection", &collection_buf) orelse { 316 316 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing collection"); 317 317 + }; 318 318 + var limit_buf: [16]u8 = undefined; 319 319 + const limit = if (queryParam(request.head.target, "limit", &limit_buf)) |raw| 320 320 + std.fmt.parseInt(usize, raw, 10) catch 50 321 321 + else 322 322 + 50; 323 323 + 324 324 + const body = try store.writeListJson(allocator, account.did, collection, @min(limit, 100)); 325 325 + return json(request, .ok, body); 326 326 + } 327 327 + 328 328 + fn repoDeleteRecord(request: *http.Server.Request) !void { 329 329 + var arena = std.heap.ArenaAllocator.init(std.heap.page_allocator); 330 330 + defer arena.deinit(); 331 331 + const allocator = arena.allocator(); 332 332 + 333 333 + const account = requireBearerAccount(request, allocator) catch |err| switch (err) { 334 334 + error.AuthRequired => return xrpcError(request, .unauthorized, "AuthenticationRequired", "Authentication required"), 335 335 + error.InvalidToken => return xrpcError(request, .unauthorized, "InvalidToken", "Invalid token"), 336 336 + }; 337 337 + 338 338 + var body_buf: [4096]u8 = undefined; 339 339 + const body = try readBody(request, &body_buf); 340 340 + const parsed = try parseJsonBody(request, allocator, body); 341 341 + const collection = zat.json.getString(parsed.value, "collection") orelse { 342 342 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing collection"); 343 343 + }; 344 344 + const rkey = zat.json.getString(parsed.value, "rkey") orelse { 345 345 + return xrpcError(request, .bad_request, "InvalidRequest", "Missing rkey"); 346 346 + }; 347 347 + store.delete(account.did, collection, rkey); 348 348 + return json(request, .ok, "{}"); 349 349 + } 350 350 + 188 351 fn requireBearerAccount(request: *const http.Server.Request, allocator: std.mem.Allocator) !auth.Account { 189 352 const auth_header = headerValue(request, "authorization") orelse { 190 353 return error.AuthRequired; ··· 195 358 196 359 const token = std.mem.trim(u8, auth_header["bearer ".len..], " \t"); 197 360 return auth.accountFromDevJwt(allocator, token) orelse error.InvalidToken; 361 361 + } 362 362 + 363 363 + fn parseJsonBody(request: *http.Server.Request, allocator: std.mem.Allocator, body: []const u8) !std.json.Parsed(std.json.Value) { 364 364 + return std.json.parseFromSlice(std.json.Value, allocator, body, .{}) catch { 365 365 + try xrpcError(request, .bad_request, "InvalidRequest", "Expected JSON body"); 366 366 + return error.InvalidJson; 367 367 + }; 368 368 + } 369 369 + 370 370 + fn recordBodyValue(value: std.json.Value) ?std.json.Value { 371 371 + return switch (value) { 372 372 + .object => |object| object.get("record"), 373 373 + else => null, 374 374 + }; 375 375 + } 376 376 + 377 377 + fn writeRecordRef(request: *http.Server.Request, allocator: std.mem.Allocator, record: store.Record) !void { 378 378 + const uri = try record.uri(allocator); 379 379 + const body_out = try std.fmt.allocPrint( 380 380 + allocator, 381 381 + "{{\"uri\":{f},\"cid\":{f}}}", 382 382 + .{ std.json.fmt(uri, .{}), std.json.fmt(record.cid, .{}) }, 383 383 + ); 384 384 + return json(request, .ok, body_out); 198 385 } 199 386 200 387 fn readBody(request: *http.Server.Request, buf: []u8) ![]const u8 { ··· 305 492 try std.testing.expectEqual(Route.create_session, route(.POST, "/xrpc/com.atproto.server.createSession")); 306 493 try std.testing.expectEqual(Route.get_session, route(.GET, "/xrpc/com.atproto.server.getSession")); 307 494 try std.testing.expectEqual(Route.get_service_auth, route(.GET, "/xrpc/com.atproto.server.getServiceAuth?aud=did%3Aplc%3Aservice")); 495 495 + try std.testing.expectEqual(Route.repo_list_records, route(.GET, "/xrpc/com.atproto.repo.listRecords?repo=alice.test&collection=app.bsky.feed.post")); 496 496 + try std.testing.expectEqual(Route.repo_create_record, route(.POST, "/xrpc/com.atproto.repo.createRecord")); 308 497 } 309 498 310 499 test "does not route wrong methods" {

+201

src/store.zig

View file Reviewed

··· 1 1 + const std = @import("std"); 2 2 + const auth = @import("auth.zig"); 3 3 + const zat = @import("zat"); 4 4 + 5 5 + const max_records = 4096; 6 6 + 7 7 + pub const Error = error{ 8 8 + InvalidCollection, 9 9 + InvalidRecordKey, 10 10 + MissingRecord, 11 11 + StoreFull, 12 12 + }; 13 13 + 14 14 + pub const Record = struct { 15 15 + active: bool = false, 16 16 + did: []const u8 = "", 17 17 + collection: []const u8 = "", 18 18 + rkey: []const u8 = "", 19 19 + cid: []const u8 = "", 20 20 + value_json: []const u8 = "", 21 21 + seq: u64 = 0, 22 22 + 23 23 + pub fn uri(self: Record, allocator: std.mem.Allocator) ![]const u8 { 24 24 + return std.fmt.allocPrint(allocator, "at://{s}/{s}/{s}", .{ self.did, self.collection, self.rkey }); 25 25 + } 26 26 + }; 27 27 + 28 28 + var records: [max_records]Record = [_]Record{.{}} ** max_records; 29 29 + var next_seq: u64 = 1; 30 30 + 31 31 + pub fn resolveRepo(repo: []const u8) ?auth.Account { 32 32 + return auth.findAccount(repo); 33 33 + } 34 34 + 35 35 + pub fn create( 36 36 + allocator: std.mem.Allocator, 37 37 + account: auth.Account, 38 38 + collection: []const u8, 39 39 + maybe_rkey: ?[]const u8, 40 40 + value: std.json.Value, 41 41 + ) !Record { 42 42 + const rkey = maybe_rkey orelse try nextRkey(allocator); 43 43 + return put(account, collection, rkey, value); 44 44 + } 45 45 + 46 46 + pub fn put( 47 47 + account: auth.Account, 48 48 + collection: []const u8, 49 49 + rkey: []const u8, 50 50 + value: std.json.Value, 51 51 + ) !Record { 52 52 + if (zat.Nsid.parse(collection) == null) return Error.InvalidCollection; 53 53 + if (zat.Rkey.parse(rkey) == null) return Error.InvalidRecordKey; 54 54 + 55 55 + const value_json = try stringifyValue(std.heap.page_allocator, value); 56 56 + const cid = try std.fmt.allocPrint(std.heap.page_allocator, "bafyreidzds{d}", .{next_seq}); 57 57 + 58 58 + if (findIndex(account.did, collection, rkey)) |idx| { 59 59 + records[idx] = .{ 60 60 + .active = true, 61 61 + .did = records[idx].did, 62 62 + .collection = records[idx].collection, 63 63 + .rkey = records[idx].rkey, 64 64 + .cid = cid, 65 65 + .value_json = value_json, 66 66 + .seq = next_seq, 67 67 + }; 68 68 + next_seq += 1; 69 69 + return records[idx]; 70 70 + } 71 71 + 72 72 + const idx = freeIndex() orelse return Error.StoreFull; 73 73 + records[idx] = .{ 74 74 + .active = true, 75 75 + .did = try std.heap.page_allocator.dupe(u8, account.did), 76 76 + .collection = try std.heap.page_allocator.dupe(u8, collection), 77 77 + .rkey = try std.heap.page_allocator.dupe(u8, rkey), 78 78 + .cid = cid, 79 79 + .value_json = value_json, 80 80 + .seq = next_seq, 81 81 + }; 82 82 + next_seq += 1; 83 83 + return records[idx]; 84 84 + } 85 85 + 86 86 + pub fn delete(did: []const u8, collection: []const u8, rkey: []const u8) void { 87 87 + if (findIndex(did, collection, rkey)) |idx| { 88 88 + records[idx].active = false; 89 89 + } 90 90 + } 91 91 + 92 92 + pub fn get(did: []const u8, collection: []const u8, rkey: []const u8) ?Record { 93 93 + if (findIndex(did, collection, rkey)) |idx| return records[idx]; 94 94 + return null; 95 95 + } 96 96 + 97 97 + pub fn writeRecordJson(allocator: std.mem.Allocator, record: Record) ![]const u8 { 98 98 + var out: std.Io.Writer.Allocating = .init(allocator); 99 99 + defer out.deinit(); 100 100 + const uri = try record.uri(allocator); 101 101 + defer allocator.free(uri); 102 102 + 103 103 + try out.writer.print( 104 104 + "{{\"uri\":{f},\"cid\":{f},\"value\":{s}}}", 105 105 + .{ std.json.fmt(uri, .{}), std.json.fmt(record.cid, .{}), record.value_json }, 106 106 + ); 107 107 + return out.toOwnedSlice(); 108 108 + } 109 109 + 110 110 + pub fn writeListJson( 111 111 + allocator: std.mem.Allocator, 112 112 + did: []const u8, 113 113 + collection: []const u8, 114 114 + limit: usize, 115 115 + ) ![]const u8 { 116 116 + var out: std.Io.Writer.Allocating = .init(allocator); 117 117 + defer out.deinit(); 118 118 + 119 119 + try out.writer.writeAll("{\"records\":["); 120 120 + var remaining = if (limit == 0) max_records else limit; 121 121 + var first = true; 122 122 + var before: ?u64 = null; 123 123 + while (remaining > 0) { 124 124 + const maybe_record = newestBefore(did, collection, before); 125 125 + const record = maybe_record orelse break; 126 126 + if (!first) try out.writer.writeByte(','); 127 127 + first = false; 128 128 + 129 129 + const uri = try record.uri(allocator); 130 130 + defer allocator.free(uri); 131 131 + try out.writer.print( 132 132 + "{{\"uri\":{f},\"cid\":{f},\"value\":{s}}}", 133 133 + .{ std.json.fmt(uri, .{}), std.json.fmt(record.cid, .{}), record.value_json }, 134 134 + ); 135 135 + before = record.seq; 136 136 + remaining -= 1; 137 137 + } 138 138 + try out.writer.writeAll("]}"); 139 139 + return out.toOwnedSlice(); 140 140 + } 141 141 + 142 142 + fn stringifyValue(allocator: std.mem.Allocator, value: std.json.Value) ![]const u8 { 143 143 + var out: std.Io.Writer.Allocating = .init(allocator); 144 144 + defer out.deinit(); 145 145 + try out.writer.print("{f}", .{std.json.fmt(value, .{})}); 146 146 + return out.toOwnedSlice(); 147 147 + } 148 148 + 149 149 + fn nextRkey(allocator: std.mem.Allocator) ![]const u8 { 150 150 + return std.fmt.allocPrint(allocator, "3zds{d}", .{next_seq}); 151 151 + } 152 152 + 153 153 + fn findIndex(did: []const u8, collection: []const u8, rkey: []const u8) ?usize { 154 154 + for (records, 0..) |record, idx| { 155 155 + if (!record.active) continue; 156 156 + if (std.mem.eql(u8, record.did, did) and 157 157 + std.mem.eql(u8, record.collection, collection) and 158 158 + std.mem.eql(u8, record.rkey, rkey)) 159 159 + { 160 160 + return idx; 161 161 + } 162 162 + } 163 163 + return null; 164 164 + } 165 165 + 166 166 + fn freeIndex() ?usize { 167 167 + for (records, 0..) |record, idx| { 168 168 + if (!record.active) return idx; 169 169 + } 170 170 + return null; 171 171 + } 172 172 + 173 173 + fn newestBefore(did: []const u8, collection: []const u8, before: ?u64) ?Record { 174 174 + var best: ?Record = null; 175 175 + for (records) |record| { 176 176 + if (!record.active) continue; 177 177 + if (!std.mem.eql(u8, record.did, did)) continue; 178 178 + if (!std.mem.eql(u8, record.collection, collection)) continue; 179 179 + if (before) |seq| { 180 180 + if (record.seq >= seq) continue; 181 181 + } 182 182 + if (best == null or record.seq > best.?.seq) best = record; 183 183 + } 184 184 + return best; 185 185 + } 186 186 + 187 187 + test "stores and lists records newest first" { 188 188 + const alice = auth.findAccount("alice.test").?; 189 189 + const one = try std.json.parseFromSlice(std.json.Value, std.testing.allocator, "{\"text\":\"one\"}", .{}); 190 190 + defer one.deinit(); 191 191 + const two = try std.json.parseFromSlice(std.json.Value, std.testing.allocator, "{\"text\":\"two\"}", .{}); 192 192 + defer two.deinit(); 193 193 + 194 194 + _ = try create(std.testing.allocator, alice, "app.bsky.feed.post", "a", one.value); 195 195 + _ = try create(std.testing.allocator, alice, "app.bsky.feed.post", "b", two.value); 196 196 + const body = try writeListJson(std.testing.allocator, alice.did, "app.bsky.feed.post", 10); 197 197 + defer std.testing.allocator.free(body); 198 198 + 199 199 + try std.testing.expect(std.mem.indexOf(u8, body, "\"text\":\"two\"").? < 200 200 + std.mem.indexOf(u8, body, "\"text\":\"one\"").?); 201 201 + }