chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/drive - the account page's read-only view of stored files
//
// Fulfills core's DriveBrowserPort. A listing reads records alone: sizes come
// from the blob refs, so it never touches blob storage. Only readFile and
// readRange join parts.
//
// A drive can live in the public repo or in a permissioned space. The space
// side reads through the injected reader rather than the public endpoints,
// the same way @pdsjs/git browses a private repository.
import { cborDecode, cidToString } from '@pdsjs/core/repo';
import {
DRIVE_FILE_COLLECTION,
DRIVE_FOLDER_COLLECTION,
folderAncestry,
folderPath,
} from './lexicon.js';
/** Upper bound on records read in one listing. */
const RECORD_LIMIT = 2000;
/** Page size for the record listing behind that bound. */
const PAGE_SIZE = 100;
/** Type a file carries when the record names none. */
const DEFAULT_MIME_TYPE = 'application/octet-stream';
/**
* @typedef {Object} BlobRef
* @property {{$link: string}|Uint8Array|string} [ref]
* @property {string} [mimeType]
* @property {number} [size]
*/
/**
* @typedef {Object} DriveFileRecord
* @property {string} [name]
* @property {string} [folder]
* @property {BlobRef[]} [parts]
* @property {string} [mimeType]
* @property {string} [createdAt]
* @property {string} [updatedAt]
* @property {string} [deletedAt]
*/
/**
* The CID string of a blob ref. A ref written over XRPC decodes to `{$link}`;
* one that came through CBOR decodes to the raw CID bytes. Both shapes reach
* this package, the same way findBlobRefs handles both.
* @param {BlobRef} part
* @returns {string}
*/
function partCid(part) {
const ref = part?.ref;
if (typeof ref === 'string') return ref;
if (ref instanceof Uint8Array) return cidToString(ref);
if (ref && typeof ref === 'object' && typeof ref.$link === 'string') {
return ref.$link;
}
return '';
}
/**
* @param {DriveFileRecord} value
* @returns {number}
*/
function totalSize(value) {
let size = 0;
for (const part of value.parts ?? []) {
if (typeof part?.size === 'number') size += part.size;
}
return size;
}
/**
* @param {Object} ctx
* @param {import('@pdsjs/core/ports').ActorStoragePort} ctx.actorStorage
* @param {Pick<import('@pdsjs/core/ports').BlobPort, 'get'|'getStream'>} ctx.blobs
* @param {() => Promise<string|null>} ctx.getDid - the hosted account's DID
* @param {import('./ports.js').DriveSpaceReader} [ctx.spaceReader] - direct storage access for drives held in a space; absent, only the public repo is browsable
* @returns {import('./ports.js').DriveBrowserPort}
*/
export function createDriveBrowser(ctx) {
const { actorStorage, blobs, getDid, spaceReader } = ctx;
if (!actorStorage)
throw new Error('createDriveBrowser requires actorStorage');
if (!blobs) throw new Error('createDriveBrowser requires blobs');
if (!getDid) throw new Error('createDriveBrowser requires getDid');
// listRecords and getRecord answer with different shapes; only the encoded
// value is common to both.
/**
* @param {{value: Uint8Array|ArrayBuffer}} record
* @returns {Record<string, unknown>}
*/
const decode = (record) =>
/** @type {Record<string, unknown>} */ (
cborDecode(
record.value instanceof Uint8Array
? record.value
: new Uint8Array(record.value),
)
);
/**
* Every record of one collection, as {rkey, cid, value}.
* @param {string} collection
* @param {string|null|undefined} space
* @returns {Promise<Array<{rkey: string, cid: string, value: Record<string, unknown>}>>}
*/
async function readAll(collection, space) {
if (space) {
if (!spaceReader) return [];
const listing = await spaceReader.listRecords(
space,
collection,
RECORD_LIMIT,
);
return listing.records.map((record) => ({
rkey: record.rkey,
cid: record.cid,
value: /** @type {Record<string, unknown>} */ (record.value),
}));
}
const rows = [];
/** @type {string|null} */
let cursor = null;
while (rows.length < RECORD_LIMIT) {
const page = await actorStorage.listRecords(
collection,
cursor,
PAGE_SIZE,
);
for (const record of page.records) {
rows.push({
rkey: record.uri.split('/').pop() || '',
cid: record.cid,
value: decode(record),
});
}
if (!page.cursor || page.records.length === 0) break;
cursor = page.cursor;
}
return rows;
}
/**
* The folder tree, keyed by record key.
* @param {string|null|undefined} space
* @returns {Promise<Map<string, {name: string, parent: string, cid: string, createdAt: string|null}>>}
*/
async function readFolders(space) {
/** @type {Map<string, {name: string, parent: string, cid: string, createdAt: string|null}>} */
const folders = new Map();
for (const row of await readAll(DRIVE_FOLDER_COLLECTION, space)) {
const name = row.value.name;
if (typeof name !== 'string' || name === '') continue;
// A trashed folder is out of the tree, so nothing under it lists and a
// file inside it does not surface at the top level.
if (typeof row.value.deletedAt === 'string') continue;
folders.set(row.rkey, {
name,
parent: typeof row.value.parent === 'string' ? row.value.parent : '',
cid: row.cid,
createdAt:
typeof row.value.createdAt === 'string' ? row.value.createdAt : null,
});
}
return folders;
}
/**
* @param {string} did
* @param {string|null|undefined} space
* @param {string} collection
* @param {string} rkey
*/
const uriFor = (did, space, collection, rkey) =>
space
? `${space}/${did}/${collection}/${rkey}`
: `at://${did}/${collection}/${rkey}`;
/**
* Every stored file, with its folder resolved.
* @param {string} did
* @param {string|null|undefined} space
* @param {Map<string, {name: string, parent: string}>} folders
* @returns {Promise<import('./ports.js').DriveFileRow[]>}
*/
async function readFiles(did, space, folders) {
const rows = [];
for (const row of await readAll(DRIVE_FILE_COLLECTION, space)) {
const value = /** @type {DriveFileRecord} */ (row.value);
if (typeof value.deletedAt === 'string') continue;
// A file naming a folder that is not there lists at the top level
// rather than disappearing. That is a broken record, not a deletion:
// a folder goes to the trash with everything under it.
const named = typeof value.folder === 'string' ? value.folder : '';
const folder = folders.has(named) ? named : '';
rows.push({
rkey: row.rkey,
uri: uriFor(did, space, DRIVE_FILE_COLLECTION, row.rkey),
cid: row.cid,
name: typeof value.name === 'string' ? value.name : row.rkey,
folder,
mimeType:
typeof value.mimeType === 'string'
? value.mimeType
: DEFAULT_MIME_TYPE,
size: totalSize(value),
partCount: (value.parts ?? []).length,
blobCid:
(value.parts ?? []).length === 1
? partCid((value.parts ?? [])[0]) || null
: null,
createdAt: typeof value.createdAt === 'string' ? value.createdAt : null,
updatedAt: typeof value.updatedAt === 'string' ? value.updatedAt : null,
});
}
rows.sort((left, right) => left.name.localeCompare(right.name));
return rows;
}
/**
* Read one file's record, from the repo or the space.
* @param {string} did
* @param {string} rkey
* @param {string|null|undefined} space
* @returns {Promise<DriveFileRecord|null>}
*/
async function readFileRecord(did, rkey, space) {
if (space) {
if (!spaceReader) return null;
const record = await spaceReader.getRecord(space, rkey);
return record ? /** @type {DriveFileRecord} */ (record.value) : null;
}
const record = await actorStorage.getRecord(
`at://${did}/${DRIVE_FILE_COLLECTION}/${rkey}`,
);
return record ? /** @type {DriveFileRecord} */ (decode(record)) : null;
}
/**
* One part's bytes, from blob storage or the space reader.
* @param {string} did
* @param {string} cid
* @param {string|null|undefined} space
* @returns {Promise<Uint8Array|null>}
*/
async function readPart(did, cid, space) {
// A space file's blobs are private, so they are read through the space
// reader rather than the public blob endpoint.
if (space) return spaceReader ? await spaceReader.getBlob(cid) : null;
const blob = await blobs.get(did, cid);
if (!blob) return null;
return blob.data instanceof Uint8Array
? blob.data
: new Uint8Array(blob.data);
}
/**
* @param {ReadableStream<Uint8Array>} stream
* @param {number} size
* @returns {Promise<Uint8Array>}
*/
async function drain(stream, size) {
const out = new Uint8Array(size);
const reader = stream.getReader();
let at = 0;
while (true) {
const { done, value } = await reader.read();
if (done) break;
out.set(value, at);
at += value.length;
}
return out.subarray(0, at);
}
/**
* The slice of one part a range touches, inclusive on both ends. When blob
* storage can stream a range, only the slice crosses the wire; a
* single-blob file then costs the range, not the file. Otherwise the whole
* part comes back and is cut here.
* @param {string} did
* @param {string} cid
* @param {string|null|undefined} space
* @param {number} from
* @param {number} to
* @returns {Promise<Uint8Array|null>}
*/
async function readPartRange(did, cid, space, from, to) {
if (!space && blobs.getStream) {
const found = await blobs.getStream(did, cid, { start: from, end: to });
if (!found) return null;
return drain(found.body, to - from + 1);
}
const data = await readPart(did, cid, space);
return data ? data.subarray(from, to + 1) : null;
}
return {
async listFolder(requested, space) {
if (space && !spaceReader) return null;
const did = await getDid();
if (!did) return null;
const rkey = typeof requested === 'string' ? requested : '';
const folders = await readFolders(space);
if (rkey && !folders.has(rkey)) return null;
const ancestry = folderAncestry(rkey, folders);
if (ancestry === null) return null;
const files = await readFiles(did, space, folders);
// What each folder holds, counting everything beneath it. One pass up
// each file's parent chain totals every ancestor at once.
/** @type {Map<string, {files: number, size: number}>} */
const beneath = new Map();
for (const file of files) {
let at = file.folder;
const seen = new Set();
while (at && !seen.has(at)) {
seen.add(at);
const tally = beneath.get(at) ?? { files: 0, size: 0 };
tally.files += 1;
tally.size += file.size;
beneath.set(at, tally);
at = folders.get(at)?.parent ?? '';
}
}
const children = [...folders.entries()]
.filter(([, folder]) => folder.parent === rkey)
.map(([key, folder]) => ({
rkey: key,
name: folder.name,
files: beneath.get(key)?.files ?? 0,
size: beneath.get(key)?.size ?? 0,
createdAt: folder.createdAt,
}))
.sort((left, right) => left.name.localeCompare(right.name));
return {
folder: rkey,
path: folderPath(rkey, folders) ?? '',
ancestry,
space: space || null,
folders: children,
files: files.filter((file) => file.folder === rkey),
totalFiles: files.length,
totalSize: files.reduce((sum, file) => sum + file.size, 0),
};
},
async listTrash(space) {
if (space && !spaceReader) return null;
const did = await getDid();
if (!did) return null;
/** @type {import('./ports.js').DriveTrashRow[]} */
const rows = [];
for (const row of await readAll(DRIVE_FOLDER_COLLECTION, space)) {
const at = row.value.deletedAt;
if (typeof at !== 'string') continue;
rows.push({
rkey: row.rkey,
uri: uriFor(did, space, DRIVE_FOLDER_COLLECTION, row.rkey),
name: typeof row.value.name === 'string' ? row.value.name : row.rkey,
isFolder: true,
mimeType: null,
size: 0,
deletedAt: at,
});
}
for (const row of await readAll(DRIVE_FILE_COLLECTION, space)) {
const value = /** @type {DriveFileRecord} */ (row.value);
if (typeof value.deletedAt !== 'string') continue;
rows.push({
rkey: row.rkey,
uri: uriFor(did, space, DRIVE_FILE_COLLECTION, row.rkey),
name: typeof value.name === 'string' ? value.name : row.rkey,
isFolder: false,
mimeType:
typeof value.mimeType === 'string'
? value.mimeType
: DEFAULT_MIME_TYPE,
size: totalSize(value),
deletedAt: value.deletedAt,
});
}
// Newest first: the thing just deleted by mistake is the one being
// looked for. A whole folder goes at one timestamp, so the name settles
// the order within it rather than leaving it to chance.
rows.sort(
(left, right) =>
right.deletedAt.localeCompare(left.deletedAt) ||
left.name.localeCompare(right.name),
);
return {
space: space || null,
items: rows,
totalSize: rows.reduce((sum, row) => sum + row.size, 0),
};
},
async readFile(rkey, space) {
const did = await getDid();
if (!did) return null;
const value = await readFileRecord(did, rkey, space);
if (!value) return null;
const parts = value.parts ?? [];
if (parts.length === 0) return null;
/** @type {Uint8Array[]} */
const chunks = [];
let size = 0;
for (const part of parts) {
const cid = partCid(part);
if (!cid) return null;
const data = await readPart(did, cid, space);
// One missing part makes the file unreadable. Serving the rest would
// hand back bytes that are not the file.
if (!data) return null;
chunks.push(data);
size += data.length;
}
const bytes = new Uint8Array(size);
let offset = 0;
for (const chunk of chunks) {
bytes.set(chunk, offset);
offset += chunk.length;
}
return {
name: typeof value.name === 'string' ? value.name : rkey,
mimeType:
typeof value.mimeType === 'string'
? value.mimeType
: DEFAULT_MIME_TYPE,
size,
bytes,
};
},
async statFile(rkey, space) {
const did = await getDid();
if (!did) return null;
const value = await readFileRecord(did, rkey, space);
if (!value || (value.parts ?? []).length === 0) return null;
return {
name: typeof value.name === 'string' ? value.name : rkey,
mimeType:
typeof value.mimeType === 'string'
? value.mimeType
: DEFAULT_MIME_TYPE,
size: totalSize(value),
};
},
async readRange(rkey, start, end, space) {
const did = await getDid();
if (!did) return null;
const value = await readFileRecord(did, rkey, space);
if (!value) return null;
const parts = value.parts ?? [];
if (parts.length === 0) return null;
const size = totalSize(value);
if (start < 0 || start >= size || end < start) return null;
const last = Math.min(end, size - 1);
const bytes = new Uint8Array(last - start + 1);
let written = 0;
let at = 0;
for (const part of parts) {
const partSize = typeof part?.size === 'number' ? part.size : 0;
const partEnd = at + partSize - 1;
// Only the parts the range touches are read, and of each, only the
// slice the range wants.
if (partEnd >= start && at <= last) {
const cid = partCid(part);
if (!cid) return null;
const from = Math.max(0, start - at);
const to = Math.min(partSize - 1, last - at);
const data = await readPartRange(did, cid, space, from, to);
if (!data) return null;
bytes.set(data, written);
written += data.length;
}
at += partSize;
if (at > last) break;
}
return {
name: typeof value.name === 'string' ? value.name : rkey,
mimeType:
typeof value.mimeType === 'string'
? value.mimeType
: DEFAULT_MIME_TYPE,
start,
end: last,
total: size,
bytes: bytes.subarray(0, written),
};
},
};
}