atproto pds in zig pds.zat.dev
pds atproto
Something went wrong. Try again.
Zig
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218const std = @import("std");const Io = std.Io;
var blob_io: Io = undefined;var root_path: []const u8 = "dev/blobs";var initialized = false;
pub fn init(io: Io, path: []const u8) void { blob_io = io; root_path = trimTrailingSlash(path); initialized = true;}
pub fn put(allocator: std.mem.Allocator, io: Io, did: []const u8, cid: []const u8, data: []const u8) !void { const temp_path = try stage(allocator, io, did, cid, data); errdefer discard(temp_path); try promote(allocator, temp_path, did, cid);}
pub fn stage(allocator: std.mem.Allocator, io: Io, did: []const u8, cid: []const u8, data: []const u8) ![]const u8 { try requireInitialized(); const dir_path = try actorDirPath(allocator, did); try mkdirPath(allocator, dir_path); const path = try blobPath(allocator, did, cid); return writeTempFileC(allocator, io, path, data);}
pub const StagedUpload = struct { temp_path: []const u8, digest: [std.crypto.hash.sha2.Sha256.digest_length]u8, size: usize,};
pub fn stageUpload( allocator: std.mem.Allocator, io: Io, did: []const u8, reader: *std.Io.Reader, expected_size: usize, max_size: usize,) !StagedUpload { if (expected_size > max_size) return error.BlobTooLarge; try requireInitialized(); const dir_path = try actorDirPath(allocator, did); try mkdirPath(allocator, dir_path); const temp_path = try uniqueTempPath(allocator, io, dir_path); const temp_path_z = try allocator.dupeSentinel(u8, temp_path, 0); errdefer Io.Dir.cwd().deleteFile(io, temp_path) catch {};
const file = std.c.fopen(temp_path_z.ptr, "wb") orelse return error.OpenBlobFailed; var closed = false; defer { if (!closed) _ = std.c.fclose(file); }
const Sha256 = std.crypto.hash.sha2.Sha256; var hasher = Sha256.init(.{}); var size: usize = 0; var buf: [64 * 1024]u8 = undefined; while (size < expected_size) { const read_len = @min(buf.len, expected_size - size); reader.readSliceAll(buf[0..read_len]) catch return error.IncompleteBlob; const chunk = buf[0..read_len]; if (std.c.fwrite(chunk.ptr, 1, chunk.len, file) != chunk.len) return error.WriteBlobFailed; hasher.update(chunk); size += chunk.len; } if (std.c.fclose(file) != 0) return error.WriteBlobFailed; closed = true;
var digest: [Sha256.digest_length]u8 = undefined; hasher.final(&digest); return .{ .temp_path = temp_path, .digest = digest, .size = size };}
pub fn promote(allocator: std.mem.Allocator, temp_path: []const u8, did: []const u8, cid: []const u8) !void { try requireInitialized(); const path = try blobPath(allocator, did, cid); try Io.Dir.cwd().rename(temp_path, Io.Dir.cwd(), path, blob_io);}
pub fn discard(temp_path: []const u8) void { Io.Dir.cwd().deleteFile(blob_io, temp_path) catch {};}
pub fn get(allocator: std.mem.Allocator, did: []const u8, cid: []const u8, limit: usize) ![]u8 { try requireInitialized(); const path = try blobPath(allocator, did, cid); return Io.Dir.cwd().readFileAlloc(blob_io, path, allocator, .limited(limit));}
pub fn delete(allocator: std.mem.Allocator, did: []const u8, cid: []const u8) !void { try requireInitialized(); const path = try blobPath(allocator, did, cid); Io.Dir.cwd().deleteFile(blob_io, path) catch |err| switch (err) { error.FileNotFound => {}, else => return err, };}
fn actorDirPath(allocator: std.mem.Allocator, did: []const u8) ![]const u8 { return std.fs.path.join(allocator, &.{ root_path, did });}
fn blobPath(allocator: std.mem.Allocator, did: []const u8, cid: []const u8) ![]const u8 { return std.fs.path.join(allocator, &.{ root_path, did, cid });}
fn requireInitialized() !void { if (!initialized) return error.BlobstoreNotInitialized;}
fn trimTrailingSlash(value: []const u8) []const u8 { if (value.len > 1 and value[value.len - 1] == '/') return value[0 .. value.len - 1]; return value;}
fn mkdirPath(allocator: std.mem.Allocator, path: []const u8) !void { var index: usize = 0; while (index < path.len) : (index += 1) { if (path[index] != std.fs.path.sep) continue; if (index == 0) continue; try mkdirOne(allocator, path[0..index]); } try mkdirOne(allocator, path);}
fn mkdirOne(allocator: std.mem.Allocator, path: []const u8) !void { const path_z = try allocator.dupeSentinel(u8, path, 0); if (std.c.mkdir(path_z.ptr, 0o755) == 0) return; const err: std.posix.E = @fromBackingInt(@intCast(std.posix.system._errno().*)); if (err == .EXIST) return; return error.CreateDirFailed;}
fn writeTempFileC(allocator: std.mem.Allocator, io: Io, path: []const u8, data: []const u8) ![]const u8 { var random_bytes: [8]u8 = undefined; io.random(&random_bytes); const nonce = std.fmt.bytesToHex(random_bytes, .lower); const temp_path = try std.fmt.allocPrint(allocator, "{s}.tmp-{s}", .{ path, &nonce }); const temp_path_z = try allocator.dupeSentinel(u8, temp_path, 0); errdefer Io.Dir.cwd().deleteFile(io, temp_path) catch {};
const file = std.c.fopen(temp_path_z.ptr, "wb") orelse return error.OpenBlobFailed; var closed = false; defer { if (!closed) _ = std.c.fclose(file); } if (data.len > 0 and std.c.fwrite(data.ptr, 1, data.len, file) != data.len) { return error.WriteBlobFailed; } if (std.c.fclose(file) != 0) return error.WriteBlobFailed; closed = true; return temp_path;}
fn uniqueTempPath(allocator: std.mem.Allocator, io: Io, dir_path: []const u8) ![]const u8 { var random_bytes: [8]u8 = undefined; io.random(&random_bytes); const nonce = std.fmt.bytesToHex(random_bytes, .lower); return std.fmt.allocPrint(allocator, "{s}/.upload-{s}.tmp", .{ dir_path, &nonce });}
test "disk blobstore writes and reads account blob bytes" { const allocator = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(std.testing.io, &path_buf); const path = path_buf[0..path_len]; init(std.Options.debug_io, path);
var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); const a = arena.allocator(); try put(a, std.Options.debug_io, "did:plc:test", "bafytest", "hello blobstore"); const data = try get(a, "did:plc:test", "bafytest", 1024); try std.testing.expectEqualStrings("hello blobstore", data);}
test "disk blobstore replaces existing bytes atomically" { const allocator = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(std.testing.io, &path_buf); const path = path_buf[0..path_len]; init(std.Options.debug_io, path);
var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); const a = arena.allocator(); try put(a, std.Options.debug_io, "did:plc:test", "bafytest", "first"); try put(a, std.Options.debug_io, "did:plc:test", "bafytest", "second"); const data = try get(a, "did:plc:test", "bafytest", 1024); try std.testing.expectEqualStrings("second", data);}
test "streamed blob upload enforces its size limit" { const allocator = std.testing.allocator; var tmp = std.testing.tmpDir(.{}); defer tmp.cleanup();
var path_buf: [std.fs.max_path_bytes]u8 = undefined; const path_len = try tmp.dir.realPath(std.testing.io, &path_buf); init(std.Options.debug_io, path_buf[0..path_len]);
var arena = std.heap.ArenaAllocator.init(allocator); defer arena.deinit(); var reader: std.Io.Reader = .fixed("too large"); try std.testing.expectError( error.BlobTooLarge, stageUpload(arena.allocator(), std.Options.debug_io, "did:plc:test", &reader, 9, 3), );}