atproto pds in zig
pds.zat.dev
pds
atproto
Track permissioned record blob refs
+85
+85
src/storage/store.zig
+85
src/storage/store.zig
···
367
367
cid: []const u8,
368
368
value_json: []const u8,
369
369
validation_status: []const u8,
370
+
blob_cids: []const []const u8,
370
371
};
371
372
372
373
pub const SpaceWriteOp = union(enum) {
···
3046
3047
const cbor_value = try jsonToDagCbor(allocator, value);
3047
3048
const record_bytes = try zat.cbor.encodeAlloc(allocator, cbor_value);
3048
3049
const record_cid = try zat.cbor.Cid.forDagCbor(allocator, record_bytes);
3050
+
var blob_cids: std.ArrayList([]const u8) = .empty;
3051
+
try collectBlobCids(allocator, cbor_value, &blob_cids);
3049
3052
return .{
3050
3053
.cid = try cidText(allocator, record_cid.raw),
3051
3054
.value_json = try stringifyValue(allocator, value),
3052
3055
.validation_status = validationStatusForRecord(collection, .known),
3056
+
.blob_cids = try blob_cids.toOwnedSlice(allocator),
3053
3057
};
3054
3058
}
3055
3059
···
3593
3597
if (try getSpaceRecordLocked(allocator, space, repo_did, write.collection, write.rkey) != null) return Error.InvalidRefreshSession;
3594
3598
try addRecordElement(allocator, &set_hash, write.collection, write.rkey, write.prepared.cid);
3595
3599
try upsertSpaceRecordLocked(space, repo_did, write.collection, write.rkey, write.prepared, rev);
3600
+
try syncSpaceRecordBlobRefsLocked(space, repo_did, write.collection, write.rkey, write.prepared.blob_cids);
3596
3601
try insertRecordOplogLocked(space, repo_did, rev, idx, "create", write.collection, write.rkey, write.prepared.cid, null);
3597
3602
try results.append(allocator, .{ .create = (try getSpaceRecordLocked(allocator, space, repo_did, write.collection, write.rkey)) orelse return Error.MissingRecord });
3598
3603
},
···
3601
3606
if (existing) |old| try removeRecordElement(allocator, &set_hash, old.collection, old.rkey, old.cid);
3602
3607
try addRecordElement(allocator, &set_hash, write.collection, write.rkey, write.prepared.cid);
3603
3608
try upsertSpaceRecordLocked(space, repo_did, write.collection, write.rkey, write.prepared, rev);
3609
+
try syncSpaceRecordBlobRefsLocked(space, repo_did, write.collection, write.rkey, write.prepared.blob_cids);
3604
3610
const action = if (existing == null) "create" else "update";
3605
3611
try insertRecordOplogLocked(space, repo_did, rev, idx, action, write.collection, write.rkey, write.prepared.cid, if (existing) |old| old.cid else null);
3606
3612
try results.append(allocator, .{ .update = (try getSpaceRecordLocked(allocator, space, repo_did, write.collection, write.rkey)) orelse return Error.MissingRecord });
···
3610
3616
try removeRecordElement(allocator, &set_hash, existing.collection, existing.rkey, existing.cid);
3611
3617
try addRecordElement(allocator, &set_hash, write.collection, write.rkey, write.prepared.cid);
3612
3618
try upsertSpaceRecordLocked(space, repo_did, write.collection, write.rkey, write.prepared, rev);
3619
+
try syncSpaceRecordBlobRefsLocked(space, repo_did, write.collection, write.rkey, write.prepared.blob_cids);
3613
3620
try insertRecordOplogLocked(space, repo_did, rev, idx, "update", write.collection, write.rkey, write.prepared.cid, existing.cid);
3614
3621
try results.append(allocator, .{ .update = (try getSpaceRecordLocked(allocator, space, repo_did, write.collection, write.rkey)) orelse return Error.MissingRecord });
3615
3622
},
3616
3623
.delete => |write| {
3617
3624
const existing = (try getSpaceRecordLocked(allocator, space, repo_did, write.collection, write.rkey)) orelse return Error.MissingRecord;
3618
3625
try removeRecordElement(allocator, &set_hash, existing.collection, existing.rkey, existing.cid);
3626
+
try deleteSpaceRecordBlobRefsLocked(space, repo_did, write.collection, write.rkey);
3619
3627
try conn.exec(
3620
3628
\\DELETE FROM permissioned_space_records
3621
3629
\\WHERE space = ? AND repo_did = ? AND collection = ? AND rkey = ?
···
3728
3736
, .{ space, repo_did, collection, rkey, prepared.cid, prepared.value_json, prepared.validation_status, rev });
3729
3737
}
3730
3738
3739
+
fn syncSpaceRecordBlobRefsLocked(
3740
+
space: []const u8,
3741
+
repo_did: []const u8,
3742
+
collection: []const u8,
3743
+
rkey: []const u8,
3744
+
blob_cids: []const []const u8,
3745
+
) !void {
3746
+
try deleteSpaceRecordBlobRefsLocked(space, repo_did, collection, rkey);
3747
+
for (blob_cids) |blob_cid| {
3748
+
try conn.exec(
3749
+
\\INSERT INTO permissioned_space_record_blobs (space, repo_did, collection, rkey, blob_cid)
3750
+
\\VALUES (?, ?, ?, ?, ?)
3751
+
\\ON CONFLICT(space, repo_did, collection, rkey, blob_cid) DO UPDATE SET
3752
+
\\ blob_cid = excluded.blob_cid
3753
+
, .{ space, repo_did, collection, rkey, blob_cid });
3754
+
}
3755
+
}
3756
+
3757
+
fn deleteSpaceRecordBlobRefsLocked(
3758
+
space: []const u8,
3759
+
repo_did: []const u8,
3760
+
collection: []const u8,
3761
+
rkey: []const u8,
3762
+
) !void {
3763
+
try conn.exec(
3764
+
\\DELETE FROM permissioned_space_record_blobs
3765
+
\\WHERE space = ? AND repo_did = ? AND collection = ? AND rkey = ?
3766
+
, .{ space, repo_did, collection, rkey });
3767
+
}
3768
+
3731
3769
fn insertRecordOplogLocked(
3732
3770
space: []const u8,
3733
3771
repo_did: []const u8,
···
5805
5843
\\)
5806
5844
,
5807
5845
"CREATE INDEX IF NOT EXISTS permissioned_space_records_list_idx ON permissioned_space_records (space, repo_did, collection, rkey)",
5846
+
\\CREATE TABLE IF NOT EXISTS permissioned_space_record_blobs (
5847
+
\\ space TEXT NOT NULL,
5848
+
\\ repo_did TEXT NOT NULL,
5849
+
\\ collection TEXT NOT NULL,
5850
+
\\ rkey TEXT NOT NULL,
5851
+
\\ blob_cid TEXT NOT NULL,
5852
+
\\ PRIMARY KEY (space, repo_did, collection, rkey, blob_cid),
5853
+
\\ FOREIGN KEY (space, repo_did, collection, rkey)
5854
+
\\ REFERENCES permissioned_space_records(space, repo_did, collection, rkey)
5855
+
\\ ON DELETE CASCADE
5856
+
\\)
5857
+
,
5858
+
"CREATE INDEX IF NOT EXISTS permissioned_space_record_blobs_blob_idx ON permissioned_space_record_blobs (repo_did, blob_cid)",
5808
5859
\\CREATE TABLE IF NOT EXISTS permissioned_space_repos (
5809
5860
\\ space TEXT NOT NULL REFERENCES permissioned_spaces(uri) ON DELETE CASCADE,
5810
5861
\\ repo_did TEXT NOT NULL,
···
6250
6301
try std.testing.expectEqualStrings("track-one", stored.rkey);
6251
6302
try std.testing.expectEqualStrings("unknown", stored.validation_status);
6252
6303
try std.testing.expect(get(account.did, "fm.plyr.track", "track-one") == null);
6304
+
try expectSpaceBlobRefCount(space.uri, account.did, "fm.plyr.track", "track-one", "bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku", 1);
6305
+
try std.testing.expect(getPublicBlob(allocator, account.did, "bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku") == null);
6253
6306
6254
6307
const found = (try getSpaceRecord(allocator, space.uri, account.did, "fm.plyr.track", "track-one")).?;
6255
6308
try std.testing.expectEqualStrings(stored.cid, found.cid);
···
6259
6312
try std.testing.expectEqual(@as(usize, 1), listed.len);
6260
6313
try std.testing.expectEqualStrings("fm.plyr.track", listed[0].collection);
6261
6314
try std.testing.expectEqualStrings("track-one", listed[0].rkey);
6315
+
6316
+
const updated_parsed = try std.json.parseFromSlice(
6317
+
std.json.Value,
6318
+
allocator,
6319
+
"{\"$type\":\"fm.plyr.track\",\"title\":\"secret track\",\"audioBlob\":{\"$type\":\"blob\",\"ref\":{\"$link\":\"bafkreibm6jgipb22ignhvbmf6avdkvuevprpyx6y6722fsxk7wxiofq4wu\"},\"mimeType\":\"audio/mpeg\",\"size\":12}}",
6320
+
.{},
6321
+
);
6322
+
defer updated_parsed.deinit();
6323
+
const updated_prepared = try prepareRecordValue(allocator, "fm.plyr.track", "track-one", updated_parsed.value);
6324
+
_ = try putSpaceRecord(allocator, space.uri, account.did, "fm.plyr.track", "track-one", updated_prepared);
6325
+
try expectSpaceBlobRefCount(space.uri, account.did, "fm.plyr.track", "track-one", "bafkreihdwdcefgh4dqkjv67uzcmw7ojee6xedzdetojuzjevtenxquvyku", 0);
6326
+
try expectSpaceBlobRefCount(space.uri, account.did, "fm.plyr.track", "track-one", "bafkreibm6jgipb22ignhvbmf6avdkvuevprpyx6y6722fsxk7wxiofq4wu", 1);
6327
+
6328
+
try deleteSpaceRecord(allocator, space.uri, account.did, "fm.plyr.track", "track-one");
6329
+
try expectSpaceBlobRefCount(space.uri, account.did, "fm.plyr.track", "track-one", "bafkreibm6jgipb22ignhvbmf6avdkvuevprpyx6y6722fsxk7wxiofq4wu", 0);
6330
+
}
6331
+
6332
+
fn expectSpaceBlobRefCount(
6333
+
space: []const u8,
6334
+
repo_did: []const u8,
6335
+
collection: []const u8,
6336
+
rkey: []const u8,
6337
+
blob_cid: []const u8,
6338
+
expected: i64,
6339
+
) !void {
6340
+
const row = try conn.row(
6341
+
\\SELECT COUNT(*)
6342
+
\\FROM permissioned_space_record_blobs
6343
+
\\WHERE space = ? AND repo_did = ? AND collection = ? AND rkey = ? AND blob_cid = ?
6344
+
, .{ space, repo_did, collection, rkey, blob_cid });
6345
+
defer row.?.deinit();
6346
+
try std.testing.expectEqual(expected, row.?.int(0));
6262
6347
}
6263
6348
6264
6349
test "permissioned spaces keep authority-local state per space URI" {