From 8df049ce4e8a1a308c60e8e79dc9285642ffa5c8 Mon Sep 17 00:00:00 2001 From: zzstoatzz Date: Mon, 13 Jul 2026 00:46:28 -0500 Subject: [PATCH] match encoded filters and archive HTTP semantics --- src/internal/filter.zig | 36 ++++++++++++++++++++++++++- src/internal/server.zig | 26 ++++++++++++++++--- src/internal/xrpcapi.zig | 54 ++++++++++++++++++++++++++++++++++------ 3 files changed, 104 insertions(+), 12 deletions(-) diff --git a/src/internal/filter.zig b/src/internal/filter.zig index 488547d..fd792ac 100644 --- a/src/internal/filter.zig +++ b/src/internal/filter.zig @@ -18,6 +18,7 @@ pub const ParseError = error{ InvalidCollection, TooManyCollections, InvalidDid, + InvalidEncoding, TooManyDids, OutOfMemory, }; @@ -48,7 +49,7 @@ pub const Filter = struct { if (pair.len == 0) continue; const eq = std.mem.indexOfScalar(u8, pair, '=') orelse continue; const key = pair[0..eq]; - const value = pair[eq + 1 ..]; + const value = try percentDecodeAlloc(f.arena.allocator(), pair[eq + 1 ..]); if (std.mem.eql(u8, key, "wantedCollections")) { try f.addCollection(value); } else if (std.mem.eql(u8, key, "wantedDids")) { @@ -118,6 +119,32 @@ test "match-all when no params" { try testing.expect(f.wants(.identity, "did:plc:x", "")); } +fn percentDecodeAlloc(allocator: Allocator, input: []const u8) ParseError![]const u8 { + const out = allocator.alloc(u8, input.len) catch return error.OutOfMemory; + var read: usize = 0; + var write: usize = 0; + while (read < input.len) { + switch (input[read]) { + '%' => { + if (read + 2 >= input.len) return error.InvalidEncoding; + out[write] = std.fmt.parseInt(u8, input[read + 1 .. read + 3], 16) catch + return error.InvalidEncoding; + read += 3; + }, + '+' => { + out[write] = ' '; + read += 1; + }, + else => |c| { + out[write] = c; + read += 1; + }, + } + write += 1; + } + return out[0..write]; +} + test "collection exact and prefix matching" { var f = try Filter.parseQuery(testing.allocator, "wantedCollections=app.bsky.feed.post&wantedCollections=sh.tangled.*"); defer f.deinit(); @@ -146,6 +173,13 @@ test "invalid collection and did rejected" { try testing.expectError(error.InvalidDid, f.addDid("bob")); } +test "query values are percent decoded" { + var f = try Filter.parseQuery(testing.allocator, "wantedDids=did%3Aplc%3Aabc&wantedCollections=app.bsky.feed.post"); + defer f.deinit(); + try testing.expect(f.wants(.commit, "did:plc:abc", "app.bsky.feed.post")); + try testing.expect(!f.wants(.commit, "did:plc:other", "app.bsky.feed.post")); +} + test "did filtering applies to all kinds" { var f = try Filter.parseQuery(testing.allocator, "wantedDids=did:plc:aaa"); defer f.deinit(); diff --git a/src/internal/server.zig b/src/internal/server.zig index a936511..7b8f8af 100644 --- a/src/internal/server.zig +++ b/src/internal/server.zig @@ -399,7 +399,16 @@ pub const Handler = struct { if (hub.xrpc) |api| { const query = if (std.mem.indexOfScalar(u8, url, '?')) |i| url[i + 1 ..] else ""; var responder: XrpcResponder = .{ .conn = conn }; - if (api.handle(&responder, method, path, query, body, headers.get("range"))) return; + if (api.handle( + &responder, + method, + path, + query, + body, + headers.get("range"), + headers.get("if-none-match"), + headers.get("if-range"), + )) return; } if (!std.mem.eql(u8, method, "GET")) return respond(conn, "405 Method Not Allowed", "text/plain", "method not allowed\n"); if (std.mem.eql(u8, path, "/healthz")) return respond(conn, "200 OK", "text/plain", "ok\n"); @@ -444,7 +453,7 @@ const XrpcResponder = struct { var buf: [320]u8 = undefined; const header = std.fmt.bufPrint( &buf, - "HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nETag: \"{s}\"\r\nCache-Control: no-cache\r\nConnection: close\r\nServer: stream\r\n\r\n", + "HTTP/1.1 200 OK\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\nConnection: close\r\nServer: stream\r\n\r\n", .{ body.len, etag_hex }, ) catch return; self.conn.writeFramed(header) catch return; @@ -455,7 +464,7 @@ const XrpcResponder = struct { var buf: [400]u8 = undefined; const header = std.fmt.bufPrint( &buf, - "HTTP/1.1 206 Partial Content\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nContent-Range: bytes {d}-{d}/{d}\r\nETag: \"{s}\"\r\nConnection: close\r\nServer: stream\r\n\r\n", + "HTTP/1.1 206 Partial Content\r\nContent-Type: application/octet-stream\r\nContent-Length: {d}\r\nContent-Range: bytes {d}-{d}/{d}\r\nAccept-Ranges: bytes\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\nConnection: close\r\nServer: stream\r\n\r\n", .{ body.len, start, end, total, etag_hex }, ) catch return; self.conn.writeFramed(header) catch return; @@ -472,6 +481,16 @@ const XrpcResponder = struct { self.conn.writeFramed(header) catch return; } + pub fn notModified(self: *XrpcResponder, etag_hex: []const u8) void { + var buf: [256]u8 = undefined; + const header = std.fmt.bufPrint( + &buf, + "HTTP/1.1 304 Not Modified\r\nETag: \"{s}\"\r\nCache-Control: public, no-cache\r\nConnection: close\r\nServer: stream\r\n\r\n", + .{etag_hex}, + ) catch return; + self.conn.writeFramed(header) catch return; + } + pub fn err(self: *XrpcResponder, status: u16, name: []const u8, message: []const u8) void { var buf: [512]u8 = undefined; const body = std.fmt.bufPrint(&buf, "{{\"error\":\"{s}\",\"message\":\"{s}\"}}", .{ name, message }) catch return; @@ -510,6 +529,7 @@ fn addAll( reason.* = switch (err) { error.InvalidCollection => "invalid options: invalid collection", error.InvalidDid => "invalid options: invalid DID", + error.InvalidEncoding => "invalid options: invalid encoding", error.TooManyCollections => "too many wanted collections", error.TooManyDids => "too many wanted DIDs", error.OutOfMemory => "internal error", diff --git a/src/internal/xrpcapi.zig b/src/internal/xrpcapi.zig index 9f4aa0d..f813a23 100644 --- a/src/internal/xrpcapi.zig +++ b/src/internal/xrpcapi.zig @@ -25,7 +25,7 @@ pub const Api = struct { /// route an /xrpc/network.bsky.jetstream.* request. returns false when /// the path isn't ours (caller 404s). - pub fn handle(self: *Api, respond: anytype, http_method: []const u8, path: []const u8, query: []const u8, body: []const u8, range_header: ?[]const u8) bool { + pub fn handle(self: *Api, respond: anytype, http_method: []const u8, path: []const u8, query: []const u8, body: []const u8, range_header: ?[]const u8, if_none_match: ?[]const u8, if_range: ?[]const u8) bool { const prefix = "/xrpc/network.bsky.jetstream."; if (!std.mem.startsWith(u8, path, prefix)) return false; const method = path[prefix.len..]; @@ -45,9 +45,9 @@ pub const Api = struct { if (std.mem.eql(u8, method, "listSegments")) { self.listSegments(respond, query); } else if (std.mem.eql(u8, method, "getSegment")) { - self.getSegment(respond, query, range_header); + self.getSegment(respond, query, range_header, if_none_match, if_range); } else if (std.mem.eql(u8, method, "getBlock")) { - self.getBlock(respond, query); + self.getBlock(respond, query, range_header, if_none_match, if_range); } else { respond.err(400, "MethodNotImplemented", method); } @@ -127,7 +127,7 @@ pub const Api = struct { respond.json(200, out.written()); } - fn getSegment(self: *Api, respond: anytype, query: []const u8, range_header: ?[]const u8) void { + fn getSegment(self: *Api, respond: anytype, query: []const u8, range_header: ?[]const u8, if_none_match: ?[]const u8, if_range: ?[]const u8) void { const name = queryParam(query, "name") orelse return respond.err(400, "InvalidRequest", "missing name"); if (archive_mod.parseSegmentIndex(name) == null) @@ -136,10 +136,16 @@ pub const Api = struct { return respond.err(400, "SegmentNotFound", name); var hex_buf: [16]u8 = undefined; const etag = checksumHex(&hex_buf, meta.header.checksum); + if (if_none_match) |candidate| { + if (etagMatches(candidate, etag)) return respond.notModified(etag); + } // single-range requests (bytes=a-b / a- / -n): sealed files are // immutable so ranges need no If-Range dance beyond the strong ETag - if (range_header) |raw| { + if (range_header) |raw| range: { + if (if_range) |candidate| { + if (!etagMatches(candidate, etag)) break :range; + } const r = parseRange(raw, meta.size) orelse return respond.rangeNotSatisfiable(meta.size); var file = self.archive.dir.openFile(self.io, name, .{}) catch @@ -161,7 +167,7 @@ pub const Api = struct { respond.raw(200, bytes, etag); } - fn getBlock(self: *Api, respond: anytype, query: []const u8) void { + fn getBlock(self: *Api, respond: anytype, query: []const u8, range_header: ?[]const u8, if_none_match: ?[]const u8, if_range: ?[]const u8) void { const name = queryParam(query, "segment") orelse return respond.err(400, "InvalidRequest", "missing segment"); const block_idx = blk: { @@ -196,8 +202,22 @@ pub const Api = struct { defer self.allocator.free(frame); _ = file.readPositionalAll(self.io, frame, entry.offset + 8) catch return respond.err(400, "BlockNotFound", name); - var hex_buf: [16]u8 = undefined; - respond.raw(200, frame, checksumHex(&hex_buf, header.checksum)); + var checksum_buf: [16]u8 = undefined; + var etag_buf: [32]u8 = undefined; + const etag = std.fmt.bufPrint(&etag_buf, "{s}:{d}", .{ checksumHex(&checksum_buf, header.checksum), block_idx }) catch unreachable; + if (if_none_match) |candidate| { + if (etagMatches(candidate, etag)) return respond.notModified(etag); + } + if (range_header) |raw| range: { + if (if_range) |candidate| { + if (!etagMatches(candidate, etag)) break :range; + } + const r = parseRange(raw, frame.len) orelse + return respond.rangeNotSatisfiable(frame.len); + respond.partial(frame[@intCast(r.start)..@intCast(r.end + 1)], etag, r.start, r.end, frame.len); + return; + } + respond.raw(200, frame, etag); } const SealedMeta = struct { header: segment.Header, size: u64 }; @@ -242,6 +262,17 @@ fn parseRange(raw: []const u8, size: u64) ?ByteRange { return .{ .start = start, .end = end }; } +fn etagMatches(raw: []const u8, etag: []const u8) bool { + var it = std.mem.splitScalar(u8, raw, ','); + while (it.next()) |part_raw| { + const part = std.mem.trim(u8, part_raw, " \t"); + if (std.mem.eql(u8, part, "*")) return true; + if (part.len == etag.len + 2 and part[0] == '"' and part[part.len - 1] == '"' and + std.mem.eql(u8, part[1 .. part.len - 1], etag)) return true; + } + return false; +} + test "parseRange forms" { try std.testing.expectEqual(ByteRange{ .start = 0, .end = 99 }, parseRange("bytes=0-99", 1000).?); try std.testing.expectEqual(ByteRange{ .start = 500, .end = 999 }, parseRange("bytes=500-", 1000).?); @@ -251,6 +282,13 @@ test "parseRange forms" { try std.testing.expect(parseRange("bytes=0-1,5-9", 1000) == null); } +test "strong etag matching" { + try std.testing.expect(etagMatches("\"abc:0\"", "abc:0")); + try std.testing.expect(etagMatches("\"other\", \"abc:0\"", "abc:0")); + try std.testing.expect(etagMatches("*", "abc:0")); + try std.testing.expect(!etagMatches("W/\"abc:0\"", "abc:0")); +} + fn queryParam(query: []const u8, name: []const u8) ?[]const u8 { var it = std.mem.splitScalar(u8, query, '&'); while (it.next()) |pair| { -- 2.54.0