chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
import { cborDecode } from '@pdsjs/core/repo';
import { describe, expect, it } from 'vitest';
import { createDriveBrowser } from '../src/browser.js';
import { createDriveWriter } from '../src/writer.js';
const DID = 'did:plc:drivewriter';
const CHUNK = 64;
/**
* A pair of writer and browser over one set of in-memory ports. With
* `streaming` the blob port offers createWriter and getStream, so the writer
* takes its single-blob path and the browser its ranged reads.
* @param {{chunkSize?: number, maxFileSize?: number, streaming?: boolean}} [options]
*/
function harness({ chunkSize = CHUNK, maxFileSize, streaming = false } = {}) {
/** @type {Map<string, {uri: string, cid: string, value: Uint8Array}>} */
const records = new Map();
/** @type {Map<string, {data: Uint8Array, mimeType: string}>} */
const blobData = new Map();
/** @type {Array<{cid: string, mimeType: string, size: number}>} */
const blobMeta = [];
/** @type {Array<{cid: string, uri: string}>} */
const links = [];
/** @type {Array<Array<{action: string, uri: string}>>} */
const commits = [];
const actorStorage = /** @type {any} */ ({
getDid: async () => DID,
getRecord: async (/** @type {string} */ uri) => records.get(uri) ?? null,
listRecords: async (
/** @type {string} */ collection,
/** @type {string|null} */ cursor,
/** @type {number} */ limit,
) => {
const matching = [...records.values()].filter((row) =>
row.uri.includes(`/${collection}/`),
);
const start = cursor ? Number(cursor) : 0;
const page = matching.slice(start, start + limit);
const next = start + page.length;
return {
records: page,
cursor: next < matching.length ? String(next) : null,
};
},
putBlock: async () => {},
putRecord: async (
/** @type {string} */ uri,
/** @type {string} */ cid,
/** @type {string} */ _collection,
/** @type {string} */ _rkey,
/** @type {Uint8Array} */ value,
) => {
records.set(uri, { uri, cid, value });
},
deleteRecord: async (/** @type {string} */ uri) => {
records.delete(uri);
},
putBlob: async (
/** @type {string} */ cid,
/** @type {string} */ mimeType,
/** @type {number} */ size,
) => {
blobMeta.push({ cid, mimeType, size });
},
linkBlobToRecord: async (
/** @type {string} */ cid,
/** @type {string} */ uri,
) => {
links.push({ cid, uri });
},
unlinkBlobsFromRecord: async () => {},
});
/** @type {{aborted: boolean, ranges: Array<{start: number, end: number}>}} */
const blobLog = { aborted: false, ranges: [] };
const blobs = /** @type {any} */ ({
put: async (
/** @type {string} */ _did,
/** @type {string} */ cid,
/** @type {Uint8Array} */ data,
/** @type {string} */ mimeType,
) => {
blobData.set(cid, { data, mimeType });
},
get: async (/** @type {string} */ _did, /** @type {string} */ cid) =>
blobData.get(cid) ?? null,
...(streaming
? {
createWriter: async (/** @type {string} */ _did) => {
/** @type {Uint8Array[]} */
const staged = [];
return {
write: async (/** @type {Uint8Array} */ chunk) => {
staged.push(chunk);
},
commit: async (
/** @type {string} */ cid,
/** @type {string} */ mimeType,
) => {
const size = staged.reduce((sum, c) => sum + c.length, 0);
const data = new Uint8Array(size);
let at = 0;
for (const c of staged) {
data.set(c, at);
at += c.length;
}
blobData.set(cid, { data, mimeType });
},
abort: async () => {
blobLog.aborted = true;
},
};
},
getStream: async (
/** @type {string} */ _did,
/** @type {string} */ cid,
/** @type {{start: number, end: number}|undefined} */ range,
) => {
const held = blobData.get(cid);
if (!held) return null;
if (range) blobLog.ranges.push(range);
const data = range
? held.data.subarray(range.start, range.end + 1)
: held.data;
return {
body: new Response(/** @type {any} */ (data)).body,
mimeType: held.mimeType,
};
},
}
: {}),
});
const ports = { actorStorage, blobs, getDid: async () => DID };
return {
writer: createDriveWriter({
...ports,
createCommit: async (
/** @type {string} */ _did,
/** @type {any[]} */ ops,
) => {
commits.push(ops);
return {};
},
chunkSize,
maxFileSize,
}),
browser: createDriveBrowser(ports),
records,
blobData,
blobMeta,
blobLog,
links,
commits,
};
}
/** @param {Uint8Array} data */
function streamOf(data) {
return new ReadableStream({
start(controller) {
// Several small pushes, so the writer's buffering is what assembles the
// chunks rather than the stream handing it one whole array.
for (let at = 0; at < data.length; at += 7) {
controller.enqueue(data.subarray(at, at + 7));
}
controller.close();
},
});
}
/** @param {number} size */
function bytesOfLength(size) {
const data = new Uint8Array(size);
for (let i = 0; i < size; i += 1) data[i] = i % 256;
return data;
}
const textOf = (/** @type {string} */ text) => new TextEncoder().encode(text);
describe('createDriveWriter', () => {
it('throws on missing ctx entries', () => {
expect(() => createDriveWriter(/** @type {any} */ ({}))).toThrow(
/actorStorage/,
);
});
});
describe('folders', () => {
it('makes a folder that holds nothing and stays', async () => {
const { writer, browser } = harness();
const folder = await writer.createFolder({ name: 'photos', parent: null });
expect(folder.name).toBe('photos');
expect(folder.parent).toBe('');
const root = await browser.listFolder('');
expect(root?.folders.map((entry) => entry.name)).toEqual(['photos']);
expect((await browser.listFolder(folder.rkey))?.files).toEqual([]);
});
it('nests a folder under another', async () => {
const { writer, browser } = harness();
const photos = await writer.createFolder({ name: 'photos', parent: null });
const year = await writer.createFolder({
name: '2026',
parent: photos.rkey,
});
const listing = await browser.listFolder(year.rkey);
expect(listing?.path).toBe('photos/2026');
});
it('renames a folder without touching a single file', async () => {
const { writer, browser, records } = harness();
const folder = await writer.createFolder({ name: 'photos', parent: null });
const file = await writer.createFile({
name: 'shot.jpg',
folder: folder.rkey,
mimeType: 'image/jpeg',
body: streamOf(textOf('x')),
});
const before = records.get(file.uri)?.cid;
await writer.renameFolder({ rkey: folder.rkey, name: 'pictures' });
// The file's own record is untouched, which is the whole reason a folder
// is named by key rather than by path.
expect(records.get(file.uri)?.cid).toBe(before);
const listing = await browser.listFolder(folder.rkey);
expect(listing?.path).toBe('pictures');
expect(listing?.files.map((entry) => entry.name)).toEqual(['shot.jpg']);
});
it('moves a folder to another parent', async () => {
const { writer, browser } = harness();
const a = await writer.createFolder({ name: 'a', parent: null });
const b = await writer.createFolder({ name: 'b', parent: null });
await writer.renameFolder({ rkey: b.rkey, parent: a.rkey });
expect((await browser.listFolder(b.rkey))?.path).toBe('a/b');
});
it('refuses to move a folder inside itself', async () => {
const { writer } = harness();
const a = await writer.createFolder({ name: 'a', parent: null });
const b = await writer.createFolder({ name: 'b', parent: a.rkey });
await expect(
writer.renameFolder({ rkey: a.rkey, parent: b.rkey }),
).rejects.toThrow(/inside itself/);
});
it('refuses two folders of the same name in one place', async () => {
const { writer } = harness();
await writer.createFolder({ name: 'photos', parent: null });
await expect(
writer.createFolder({ name: 'photos', parent: null }),
).rejects.toThrow(/already here/);
});
it('refuses a parent that does not exist', async () => {
const { writer } = harness();
await expect(
writer.createFolder({ name: 'photos', parent: 'nosuch' }),
).rejects.toThrow(/No such folder/);
});
it('takes a folder to the trash with what is under it', async () => {
const { writer, browser } = harness();
const parent = await writer.createFolder({ name: 'a', parent: null });
await writer.createFolder({ name: 'b', parent: parent.rkey });
expect(await writer.deleteFolder(parent.rkey)).toBe(true);
expect((await browser.listFolder(''))?.folders).toEqual([]);
expect(await writer.deleteFolder('nosuch')).toBe(false);
});
});
describe('files', () => {
it('stores a small file as one part keeping its own type', async () => {
const { writer, blobMeta, commits, links } = harness();
const data = textOf('a short note');
const row = await writer.createFile({
name: 'note.txt',
folder: null,
mimeType: 'text/plain',
body: streamOf(data),
});
expect(row.partCount).toBe(1);
expect(row.size).toBe(data.length);
expect(row.mimeType).toBe('text/plain');
// A file within one blob keeps its type, so the blob browser and image
// previews still see what it is.
expect(blobMeta).toEqual([
{ cid: expect.any(String), mimeType: 'text/plain', size: data.length },
]);
expect(links).toHaveLength(1);
expect(commits).toEqual([
[{ action: 'create', uri: row.uri, cid: row.cid }],
]);
});
it('splits a file over the chunk size and reads back byte for byte', async () => {
const { writer, browser, blobMeta } = harness();
const data = bytesOfLength(CHUNK * 3 + 5);
const row = await writer.createFile({
name: 'big.bin',
folder: null,
mimeType: 'application/zip',
body: streamOf(data),
});
expect(row.partCount).toBe(4);
expect(row.size).toBe(data.length);
// No part may exceed what the server accepts as one blob.
for (const meta of blobMeta) expect(meta.size).toBeLessThanOrEqual(CHUNK);
// A split file's parts are opaque; the whole-file type lives on the record.
expect(new Set(blobMeta.map((m) => m.mimeType))).toEqual(
new Set(['application/octet-stream']),
);
expect(row.mimeType).toBe('application/zip');
const read = await browser.readFile(row.rkey);
expect(read?.bytes).toEqual(data);
});
it('stores a file of exactly one chunk as a single part', async () => {
const { writer } = harness();
const row = await writer.createFile({
name: 'exact.bin',
folder: null,
mimeType: 'application/octet-stream',
body: streamOf(bytesOfLength(CHUNK)),
});
expect(row.partCount).toBe(1);
});
it('stores an empty file as one empty part', async () => {
const { writer, browser } = harness();
const row = await writer.createFile({
name: 'empty.txt',
folder: null,
mimeType: 'text/plain',
body: streamOf(new Uint8Array(0)),
});
// The record requires at least one part, so an empty file gets one.
expect(row.partCount).toBe(1);
expect(row.size).toBe(0);
expect((await browser.readFile(row.rkey))?.size).toBe(0);
});
it('lands the file in the folder it names', async () => {
const { writer, browser } = harness();
const folder = await writer.createFolder({ name: 'photos', parent: null });
await writer.createFile({
name: 'shot.jpg',
folder: folder.rkey,
mimeType: 'image/jpeg',
body: streamOf(new Uint8Array([1, 2, 3])),
});
const listing = await browser.listFolder(folder.rkey);
expect(listing?.files.map((file) => file.name)).toEqual(['shot.jpg']);
});
it('writes a record the lexicon describes', async () => {
const { writer, records } = harness();
const folder = await writer.createFolder({ name: 'docs', parent: null });
const row = await writer.createFile({
name: 'note.txt',
folder: folder.rkey,
mimeType: 'text/plain',
body: streamOf(new Uint8Array([1])),
});
const stored = /** @type {any} */ (
cborDecode(/** @type {Uint8Array} */ (records.get(row.uri)?.value))
);
expect(stored.$type).toBe('dev.pdsjs.drive.file');
expect(stored.name).toBe('note.txt');
expect(stored.folder).toBe(folder.rkey);
expect(stored.parts).toHaveLength(1);
});
it('leaves the folder field off a top-level file', async () => {
const { writer, records } = harness();
const row = await writer.createFile({
name: 'note.txt',
folder: null,
mimeType: 'text/plain',
body: streamOf(new Uint8Array([1])),
});
const stored = /** @type {any} */ (
cborDecode(/** @type {Uint8Array} */ (records.get(row.uri)?.value))
);
expect('folder' in stored).toBe(false);
});
it('refuses a name or a folder that is not valid', async () => {
const { writer } = harness();
const send = (
/** @type {string} */ name,
/** @type {string|null} */ folder,
) =>
writer.createFile({
name,
folder,
mimeType: 'text/plain',
body: streamOf(new Uint8Array([1])),
});
await expect(send('a/b.txt', null)).rejects.toThrow(/file name/);
await expect(send('..', null)).rejects.toThrow(/file name/);
await expect(send('ok.txt', 'nosuchfolder')).rejects.toThrow(
/No such folder/,
);
});
it('refuses a body larger than the cap', async () => {
const { writer } = harness({ maxFileSize: CHUNK * 2 });
await expect(
writer.createFile({
name: 'huge.bin',
folder: null,
mimeType: 'application/octet-stream',
body: streamOf(bytesOfLength(CHUNK * 3)),
}),
).rejects.toThrow(/larger than this server accepts/);
});
it('refuses a request with no body', async () => {
const { writer } = harness();
await expect(
writer.createFile({
name: 'nothing.txt',
folder: null,
mimeType: 'text/plain',
body: null,
}),
).rejects.toThrow(/No file was sent/);
});
it('takes a file to the trash, and reports one that is not there', async () => {
const { writer, browser } = harness();
const row = await writer.createFile({
name: 'gone.txt',
folder: null,
mimeType: 'text/plain',
body: streamOf(textOf('x')),
});
expect(await writer.deleteFile(row.rkey)).toBe(true);
expect((await browser.listFolder(''))?.files).toEqual([]);
expect(await writer.deleteFile('nosuch')).toBe(false);
});
});
describe('single-blob files', () => {
it('stores a file past the chunk size as one blob of its own type', async () => {
const { writer, records, blobData, blobMeta, links } = harness({
streaming: true,
});
const data = bytesOfLength(CHUNK * 3 + 5);
const row = await writer.createFile({
name: 'clip.mp4',
folder: null,
mimeType: 'video/mp4',
body: streamOf(data),
});
expect(row.partCount).toBe(1);
expect(row.size).toBe(data.length);
const record = /** @type {any} */ (
cborDecode(/** @type {Uint8Array} */ (records.get(row.uri)?.value))
);
expect(record.parts).toHaveLength(1);
expect(record.parts[0].mimeType).toBe('video/mp4');
expect(record.parts[0].size).toBe(data.length);
const cid = record.parts[0].ref.$link;
expect(blobData.get(cid)?.data).toEqual(data);
expect(blobData.get(cid)?.mimeType).toBe('video/mp4');
expect(blobMeta).toEqual([
{ cid, mimeType: 'video/mp4', size: data.length },
]);
expect(links).toEqual([{ cid, uri: row.uri }]);
});
it('gives the same cid the chunked path gives the same bytes', async () => {
const streamed = harness({ streaming: true });
const buffered = harness();
const data = bytesOfLength(CHUNK - 1);
const one = await streamed.writer.createFile({
name: 'a.bin',
folder: null,
mimeType: 'application/octet-stream',
body: streamOf(data),
});
const two = await buffered.writer.createFile({
name: 'a.bin',
folder: null,
mimeType: 'application/octet-stream',
body: streamOf(data),
});
const cidOf = (/** @type {any} */ h, /** @type {any} */ row) =>
/** @type {any} */ (
cborDecode(/** @type {Uint8Array} */ (h.records.get(row.uri)?.value))
).parts[0].ref.$link;
expect(cidOf(streamed, one)).toBe(cidOf(buffered, two));
});
it('holds the file size limit and abandons the staged bytes', async () => {
const { writer, records, blobData, blobLog } = harness({
streaming: true,
maxFileSize: 10,
});
await expect(
writer.createFile({
name: 'big.bin',
folder: null,
mimeType: 'application/octet-stream',
body: streamOf(bytesOfLength(24)),
}),
).rejects.toThrow(/larger than this server accepts/);
expect(blobLog.aborted).toBe(true);
expect(blobData.size).toBe(0);
expect(records.size).toBe(0);
});
it('reads a range as a slice rather than the whole blob', async () => {
const { writer, browser, blobLog } = harness({ streaming: true });
const data = bytesOfLength(CHUNK * 2);
const row = await writer.createFile({
name: 'clip.mp4',
folder: null,
mimeType: 'video/mp4',
body: streamOf(data),
});
const range = await browser.readRange(row.rkey, 10, 19);
expect(range?.bytes).toEqual(data.subarray(10, 20));
expect(range?.total).toBe(data.length);
expect(blobLog.ranges).toEqual([{ start: 10, end: 19 }]);
});
it('an empty file still stores one blob', async () => {
const { writer, records } = harness({ streaming: true });
const row = await writer.createFile({
name: 'empty.txt',
folder: null,
mimeType: 'text/plain',
body: streamOf(new Uint8Array(0)),
});
const record = /** @type {any} */ (
cborDecode(/** @type {Uint8Array} */ (records.get(row.uri)?.value))
);
expect(record.parts).toHaveLength(1);
expect(record.parts[0].size).toBe(0);
});
});
describe('upload sessions', () => {
it('lands a file sent in parts as one blob of its own type', async () => {
const { writer, records, blobData } = harness({ streaming: true });
const data = bytesOfLength(CHUNK * 3 + 9);
const { id } = await /** @type {any} */ (writer).createUpload({
name: 'clip.mp4',
folder: null,
mimeType: 'video/mp4',
});
for (let at = 0; at < data.length; at += 50) {
const { received } = await /** @type {any} */ (writer).appendUpload(
id,
streamOf(data.subarray(at, at + 50)),
);
expect(received).toBe(Math.min(at + 50, data.length));
}
const row = await /** @type {any} */ (writer).finishUpload(id);
expect(row.partCount).toBe(1);
expect(row.size).toBe(data.length);
const record = /** @type {any} */ (
cborDecode(/** @type {Uint8Array} */ (records.get(row.uri)?.value))
);
const cid = record.parts[0].ref.$link;
expect(blobData.get(cid)?.data).toEqual(data);
expect(blobData.get(cid)?.mimeType).toBe('video/mp4');
// The same bytes in one request give the same blob CID.
const single = await writer.createFile({
name: 'again.mp4',
folder: null,
mimeType: 'video/mp4',
body: streamOf(data),
});
const other = /** @type {any} */ (
cborDecode(/** @type {Uint8Array} */ (records.get(single.uri)?.value))
);
expect(other.parts[0].ref.$link).toBe(cid);
});
it('refuses parts for an unknown or finished upload', async () => {
const { writer } = harness({ streaming: true });
await expect(
/** @type {any} */ (writer).appendUpload(
'nosuch',
streamOf(bytesOfLength(1)),
),
).rejects.toThrow(/No such upload/);
const { id } = await /** @type {any} */ (writer).createUpload({
name: 'a.bin',
folder: null,
mimeType: 'application/octet-stream',
});
await /** @type {any} */ (writer).appendUpload(
id,
streamOf(bytesOfLength(4)),
);
await /** @type {any} */ (writer).finishUpload(id);
await expect(
/** @type {any} */ (writer).appendUpload(id, streamOf(bytesOfLength(1))),
).rejects.toThrow(/No such upload/);
});
it('holds the size limit and abandons the session', async () => {
const { writer, blobData, blobLog } = harness({
streaming: true,
maxFileSize: 10,
});
const { id } = await /** @type {any} */ (writer).createUpload({
name: 'big.bin',
folder: null,
mimeType: 'application/octet-stream',
});
await expect(
/** @type {any} */ (writer).appendUpload(id, streamOf(bytesOfLength(24))),
).rejects.toThrow(/larger than this server accepts/);
expect(blobLog.aborted).toBe(true);
expect(blobData.size).toBe(0);
await expect(
/** @type {any} */ (writer).appendUpload(id, streamOf(bytesOfLength(1))),
).rejects.toThrow(/No such upload/);
});
it('abort abandons the staged bytes', async () => {
const { writer, blobData, blobLog } = harness({ streaming: true });
const { id } = await /** @type {any} */ (writer).createUpload({
name: 'gone.bin',
folder: null,
mimeType: 'application/octet-stream',
});
await /** @type {any} */ (writer).appendUpload(
id,
streamOf(bytesOfLength(8)),
);
await /** @type {any} */ (writer).abortUpload(id);
expect(blobLog.aborted).toBe(true);
expect(blobData.size).toBe(0);
await expect(/** @type {any} */ (writer).finishUpload(id)).rejects.toThrow(
/No such upload/,
);
});
it('is absent when the blob port cannot stream', async () => {
const { writer } = harness();
expect(writer.createUpload).toBeUndefined();
expect(writer.uploadPartSize).toBeUndefined();
});
it('checks the name before opening a session', async () => {
const { writer } = harness({ streaming: true });
await expect(
/** @type {any} */ (writer).createUpload({
name: '../escape',
folder: null,
mimeType: 'application/octet-stream',
}),
).rejects.toThrow(/name is not valid/);
});
});