Something went wrong. Try again.
A Zig library for Bluesky activity and feed analytics, built on Zat.
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334const std = @import("std");const zat = @import("zat");const Owned = @import("owned.zig").Owned;const rate = @import("rate.zig");
pub const Profile = struct { did: []const u8, handle: []const u8, displayName: []const u8 = "" };pub const Session = struct { did: []const u8, handle: []const u8, accessJwt: []const u8, refreshJwt: []const u8 };pub const Param = zat.XrpcClient.QueryParam;
pub const Client = struct { io: std.Io, allocator: std.mem.Allocator, rpc: zat.XrpcClient, session: ?Session = null, session_owner: ?std.json.Parsed(Session) = null, fault_owner: ?std.json.Parsed(Fault) = null, appview: ?[]const u8 = null, governor: rate.Governor = .{}, shared_governor: ?*rate.Shared = null, last_response: ?Observation = null, last_fault: ?Fault = null,
pub const Fault = struct { @"error": []const u8, message: ?[]const u8 = null };
pub const Observation = struct { host: []const u8, appview: ?[]const u8, authenticated: bool, method: []const u8, status: u16, rate_limit: rate.Headers, observed_at: i64, };
fn observe(self: *Client, method: []const u8, status: std.http.Status, headers: rate.Headers, retry: u8) void { const now = std.Io.Clock.real.now(self.io).toSeconds(); self.governor.observe(headers, status == .too_many_requests, retry, std.Io.Clock.awake.now(self.io).toMilliseconds(), now); if (self.shared_governor) |shared| shared.observe(self.io, headers, status == .too_many_requests, retry); self.last_response = .{ .host = self.rpc.host, .appview = if (std.mem.startsWith(u8, method, "app.bsky.")) self.appview else null, .authenticated = self.session != null, .method = method, .status = @backingInt(status), .rate_limit = headers, .observed_at = now }; }
fn wait(self: *Client) !void { if (self.shared_governor) |shared| return shared.wait(self.io); return self.governor.wait(self.io); }
pub fn init(io: std.Io, allocator: std.mem.Allocator, pds: []const u8) Client { return .{ .io = io, .allocator = allocator, .rpc = zat.XrpcClient.initWithUserAgent(io, allocator, pds, "tend/0.1") }; }
pub fn deinit(self: *Client) void { if (self.session_owner) |owner| owner.deinit(); if (self.fault_owner) |owner| owner.deinit(); self.rpc.deinit(); }
pub fn login(self: *Client, expected_did: []const u8, password: []const u8) !void { if (!std.mem.startsWith(u8, self.rpc.host, "https://")) return error.InsecurePds; const body = try std.json.Stringify.valueAlloc(self.allocator, .{ .identifier = expected_did, .password = password }, .{}); defer self.allocator.free(body); try self.wait(); var response = try self.rpc.procedure(zat.Nsid.parse("com.atproto.server.createSession").?, body); self.observe("com.atproto.server.createSession", response.status, response.rate_limit, 0); defer response.deinit(); if (!response.ok()) return error.LoginFailed; const parsed = try std.json.parseFromSlice(Session, self.allocator, response.body, .{ .ignore_unknown_fields = true, .allocate = .alloc_always }); errdefer parsed.deinit(); const session = parsed.value; if (!std.mem.eql(u8, expected_did, session.did)) return error.IdentityMismatch; if (self.session_owner) |owner| owner.deinit(); self.session_owner = parsed; self.session = session; }
fn refresh(self: *Client) !void { const old = self.session orelse return error.NotAuthenticated; self.rpc.setAuth(old.refreshJwt); defer self.rpc.access_token = null; try self.wait(); var response = try self.rpc.procedure(zat.Nsid.parse("com.atproto.server.refreshSession").?, null); self.observe("com.atproto.server.refreshSession", response.status, response.rate_limit, 0); defer response.deinit(); if (!response.ok()) return error.RefreshFailed; const parsed = try std.json.parseFromSlice(Session, self.allocator, response.body, .{ .ignore_unknown_fields = true, .allocate = .alloc_always }); errdefer parsed.deinit(); const session = parsed.value; if (!std.mem.eql(u8, old.did, session.did)) return error.IdentityMismatch; if (self.session_owner) |owner| owner.deinit(); self.session_owner = parsed; self.session = session; }
pub fn query(self: *Client, comptime T: type, method: []const u8, params: []const Param) !std.json.Parsed(T) { if (self.fault_owner) |owner| owner.deinit(); self.fault_owner = null; self.last_fault = null; if (zat.Nsid.parse(method) == null) return error.InvalidMethod; const url = try queryUrl(self.allocator, self.rpc.host, method, params); defer self.allocator.free(url); const bsky = std.mem.startsWith(u8, method, "app.bsky."); if (bsky and self.session != null) try validateAppview(self.appview orelse return error.MissingAppview); var refreshed = false; var throttles: u8 = 0; while (true) { try self.wait(); const authorization = if (self.session) |s| try std.fmt.allocPrint(self.allocator, "Bearer {s}", .{s.accessJwt}) else null; defer if (authorization) |header| self.allocator.free(header); var response = try self.rpc.transport.fetch(.{ .url = url, .authorization = authorization, .extra_headers = if (bsky and self.session != null) &.{.{ .name = "atproto-proxy", .value = self.appview.? }} else null, .deadline_ns = 30 * std.time.ns_per_s, }); defer response.deinit(self.allocator); self.observe(method, response.status, response.rate_limit, throttles); if (response.status == .unauthorized and self.session != null and !refreshed) { const fault = std.json.parseFromSlice(struct { @"error": []const u8 }, self.allocator, response.body, .{ .ignore_unknown_fields = true }) catch return error.Unauthorized; defer fault.deinit(); if (!std.mem.eql(u8, fault.value.@"error", "ExpiredToken")) return error.Unauthorized; refreshed = true; try self.refresh(); continue; } if (response.status == .too_many_requests) { if (throttles >= 2) return error.RateLimited; throttles += 1; continue; } if (response.status != .ok) { self.fault_owner = std.json.parseFromSlice(Fault, self.allocator, response.body, .{ .ignore_unknown_fields = true, .allocate = .alloc_always }) catch null; self.last_fault = if (self.fault_owner) |owner| owner.value else null; return error.RequestFailed; } return std.json.parseFromSlice(T, self.allocator, response.body, .{ .ignore_unknown_fields = true, .allocate = .alloc_always }); } }
pub fn follows(self: *Client) !Owned([]const Profile) { return self.profiles("app.bsky.graph.getFollows", "follows", true); }
pub fn download(self: *Client, method: []const u8, params: []const Param, accept: []const u8, sink: *std.Io.Writer, max_bytes: usize) !u64 { if (self.session != null) return error.AuthenticatedArchiveClient; if (zat.Nsid.parse(method) == null) return error.InvalidMethod; const url = try queryUrl(self.allocator, self.rpc.host, method, params); defer self.allocator.free(url); var throttles: u8 = 0; while (true) { try self.wait(); var response = try self.rpc.transport.open(.{ .url = url, .accept = accept, .accept_encoding = "gzip", .deadline_ns = 60 * std.time.ns_per_s }); defer response.deinit(); self.observe(method, response.status, response.rate_limit, throttles); if (response.status == .too_many_requests) { if (throttles >= 2) return error.RateLimited; throttles += 1; continue; } if (response.status != .ok) return error.RequestFailed; return response.stream(sink, .{ .max_bytes = max_bytes }); } }
pub fn mutes(self: *Client) !Owned([]const Profile) { return self.profiles("app.bsky.graph.getMutes", "mutes", false); }
fn profiles(self: *Client, comptime method: []const u8, comptime field: []const u8, actor: bool) !Owned([]const Profile) { const session = self.session orelse return error.NotAuthenticated; var arena = std.heap.ArenaAllocator.init(self.allocator); errdefer arena.deinit(); const a = arena.allocator(); const actor_did = try a.dupe(u8, session.did); var all: std.ArrayList(Profile) = .empty; var cursor: ?[]const u8 = null; var seen = std.StringHashMap(void).init(a); for (0..1000) |_| { var params: std.ArrayList(Param) = .empty; try params.append(a, .{ .name = "limit", .value = "100" }); if (actor) try params.append(a, .{ .name = "actor", .value = actor_did }); if (cursor) |value| try params.append(a, .{ .name = "cursor", .value = value }); const Page = if (comptime std.mem.eql(u8, field, "follows")) struct { follows: []Profile, cursor: ?[]const u8 = null } else struct { mutes: []Profile, cursor: ?[]const u8 = null }; const page = try self.query(Page, method, params.items); defer page.deinit(); for (@field(page.value, field)) |account| try all.append(a, try copyProfile(a, account)); cursor = if (page.value.cursor) |next| try a.dupe(u8, next) else null; if (cursor) |next| { if (seen.contains(next)) return error.RepeatedCursor; try seen.put(next, {}); } else return .{ .value = try all.toOwnedSlice(a), .arena = arena }; } return error.PageLimitExceeded; }};
pub const Identity = struct { did: []const u8, pds: []const u8 };pub fn resolve(io: std.Io, allocator: std.mem.Allocator, identifier: []const u8) !Owned(Identity) { var arena = std.heap.ArenaAllocator.init(allocator); errdefer arena.deinit(); const a = arena.allocator(); var handles = zat.HandleResolver.init(io, a); defer handles.deinit(); const did = if (zat.Did.parse(identifier) != null) try a.dupe(u8, identifier) else try handles.resolve(zat.Handle.parse(identifier) orelse return error.InvalidHandle); var identities = zat.DidResolver.init(io, a); defer identities.deinit(); var doc = try identities.resolve(zat.Did.parse(did) orelse return error.InvalidDid); defer doc.deinit(); return .{ .arena = arena, .value = .{ .did = did, .pds = try a.dupe(u8, doc.pdsEndpoint() orelse return error.NoPdsEndpoint) } };}
pub fn copyProfile(a: std.mem.Allocator, profile: Profile) !Profile { const did = try a.dupe(u8, profile.did); errdefer a.free(did); const handle = try a.dupe(u8, profile.handle); errdefer a.free(handle); return .{ .did = did, .handle = handle, .displayName = try a.dupe(u8, profile.displayName) };}
fn queryUrl(allocator: std.mem.Allocator, host: []const u8, method: []const u8, params: []const Param) ![]const u8 { var url: std.ArrayList(u8) = .empty; errdefer url.deinit(allocator); try url.print(allocator, "{s}/xrpc/{s}", .{ host, method }); for (params, 0..) |param, i| { try url.print(allocator, "{c}{s}=", .{ @as(u8, if (i == 0) '?' else '&'), param.name }); for (param.value) |c| { if (std.ascii.isAlphanumeric(c) or c == '-' or c == '_' or c == '.' or c == '~') try url.append(allocator, c) else try url.print(allocator, "%{X:0>2}", .{c}); } } return url.toOwnedSlice(allocator);}
test "opaque cursors cannot inject query parameters" { const url = try queryUrl(std.testing.allocator, "https://pds.example", "app.bsky.graph.getMutes", &.{.{ .name = "cursor", .value = "a&actor=other?x#" }}); defer std.testing.allocator.free(url); try std.testing.expectEqualStrings("https://pds.example/xrpc/app.bsky.graph.getMutes?cursor=a%26actor%3Dother%3Fx%23", url);}
pub fn validateAppview(service: []const u8) !void { const split = std.mem.indexOfScalar(u8, service, '#') orelse return error.InvalidAppview; if (zat.Did.parse(service[0..split]) == null or split + 1 == service.len) return error.InvalidAppview; for (service[split + 1 ..]) |char| { if (!std.ascii.isAlphanumeric(char) and char != '_' and char != '-' and char != '.') return error.InvalidAppview; }}
test "appview must identify a service, not an HTTP endpoint" { try validateAppview("did:web:api.bsky.app#bsky_appview"); try validateAppview("did:web:alternative.example#microblog"); try std.testing.expectError(error.InvalidAppview, validateAppview("https://public.api.bsky.app")); try std.testing.expectError(error.InvalidAppview, validateAppview("did:web:api.bsky.app")); try std.testing.expectError(error.InvalidAppview, validateAppview("did:web:api.bsky.app#"));}
test "download retries a real 429 before writing successful response bytes" { const origin_module = @import("test_origin.zig"); const io = std.testing.io; var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const a = arena.allocator(); const replies: []const []const u8 = &.{ "HTTP/1.1 429 Too Many Requests\r\nRetry-After: 0\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", try origin_module.response(a, "archive"), }; var origin = try origin_module.Origin.init(io, replies); defer origin.deinit(); var serving = try io.concurrent(origin_module.Origin.serve, .{&origin}); defer _ = serving.cancel(io) catch {}; var client = Client.init(io, a, try origin.url(a)); defer client.deinit(); var sink: std.Io.Writer.Allocating = .init(a); const length = try client.download("com.atproto.sync.getRepo", &.{}, "application/octet-stream", &sink.writer, 100); try serving.await(io); try std.testing.expectEqual(@as(u64, 7), length); try std.testing.expectEqualStrings("archive", sink.written()); try std.testing.expectEqual(@as(usize, 2), origin.requests);}
test "a long-lived client releases request storage and results outlive later requests" { const origin_module = @import("test_origin.zig"); const io = std.testing.io; var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const a = arena.allocator(); const replies: []const []const u8 = &.{ try origin_module.response(a, "{\"name\":\"first\"}"), "HTTP/1.1 400 Bad Request\r\nContent-Length: 15\r\nConnection: close\r\n\r\n{\"error\":\"Bad\"}", try origin_module.response(a, "{\"name\":\"third\"}"), }; var origin = try origin_module.Origin.init(io, replies); defer origin.deinit(); var serving = try io.concurrent(origin_module.Origin.serve, .{&origin}); defer _ = serving.cancel(io) catch {}; var rpc = Client.init(io, std.testing.allocator, try origin.url(a)); defer rpc.deinit(); const T = struct { name: []const u8 }; const first = try rpc.query(T, "app.bsky.actor.getProfile", &.{}); defer first.deinit(); try std.testing.expectError(error.RequestFailed, rpc.query(T, "app.bsky.actor.getProfile", &.{})); try std.testing.expectEqualStrings("Bad", rpc.last_fault.?.@"error"); const third = try rpc.query(T, "app.bsky.actor.getProfile", &.{}); defer third.deinit(); try std.testing.expect(rpc.last_fault == null); try std.testing.expectEqualStrings("first", first.value.name); try std.testing.expectEqualStrings("third", third.value.name); try serving.await(io);}
test "relationship pages own their profiles after the client is closed" { const origin_module = @import("test_origin.zig"); const io = std.testing.io; var arena = std.heap.ArenaAllocator.init(std.testing.allocator); defer arena.deinit(); const a = arena.allocator(); const replies: []const []const u8 = &.{ try origin_module.response(a, "{\"follows\":[{\"did\":\"did:plc:one\",\"handle\":\"one.test\"}],\"cursor\":\"next\"}"), try origin_module.response(a, "{\"follows\":[{\"did\":\"did:plc:two\",\"handle\":\"two.test\"}]}"), }; var origin = try origin_module.Origin.init(io, replies); defer origin.deinit(); var serving = try io.concurrent(origin_module.Origin.serve, .{&origin}); defer _ = serving.cancel(io) catch {}; var result = block: { var rpc = Client.init(io, std.testing.allocator, try origin.url(a)); defer rpc.deinit(); rpc.session = .{ .did = "did:plc:viewer", .handle = "viewer.test", .accessJwt = "test-only", .refreshJwt = "test-only" }; rpc.appview = "did:web:appview.test#bsky_appview"; break :block try rpc.follows(); }; defer result.deinit(); try std.testing.expectEqual(@as(usize, 2), result.value.len); try std.testing.expectEqualStrings("one.test", result.value[0].handle); try std.testing.expectEqualStrings("two.test", result.value[1].handle); try serving.await(io);}