chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/drive - writing a stored file or folder
//
// Fulfills core's DriveWriterPort. A file's bytes arrive as a stream and leave
// as blobs without the whole file ever being held: one chunk is in memory at a
// time. Records are written straight to the ports, the same way the sites
// installer writes, rather than back over XRPC.
import {
blobCidFromDigest,
cborDecode,
cborEncodeDagCbor,
cidToString,
createBlobCid,
createCid,
createTid,
} from '@pdsjs/core/repo';
import { createSha256 } from '@pdsjs/core/sha256';
import {
DRIVE_FILE_COLLECTION,
DRIVE_FOLDER_COLLECTION,
isValidName,
MAX_FOLDER_DEPTH,
normalizeFolderKey,
} from './lexicon.js';
/** Type carried by the parts of a file that did not fit in one blob. */
const PART_MIME_TYPE = 'application/octet-stream';
/** Type a file carries when the caller names none. */
const DEFAULT_MIME_TYPE = 'application/octet-stream';
/** How many chunks one file may take, when the caller sets no other bound. */
const DEFAULT_MAX_CHUNKS = 200;
/**
* How many bytes a client sends per request of a multi-request upload. A
* hint, not a limit the server enforces: it sits under the smallest
* per-request body cap a platform in front of the PDS imposes (100 MB on
* the lowest Cloudflare plan).
*/
export const UPLOAD_PART_SIZE = 64 * 1024 * 1024;
/** How long an upload session waits between requests before it is swept. */
const UPLOAD_SESSION_TTL = 60 * 60 * 1000;
/** Upper bound on folder records read while checking a parent. */
const FOLDER_LIMIT = 2000;
/** Writes per space commit, under the cap the space writer enforces. */
const SPACE_WRITE_BATCH = 100;
/**
* @typedef {Object} DriveWriterContext
* @property {import('@pdsjs/core/ports').ActorStoragePort} actorStorage
* @property {import('@pdsjs/core/ports').BlobPort} blobs
* @property {() => Promise<string|null>} getDid
* @property {(did: string, ops: Array<{action: string, uri: string, cid?: string}>) => Promise<Object>} createCommit
* @property {number} chunkSize - Largest blob this server accepts, so no part exceeds it
* @property {number} [maxFileSize] - Largest file accepted, in bytes
* @property {import('@pdsjs/core/ports').HasherFactory} [hasher] - SHA-256 factory for the single-blob path; absent, the plain JavaScript one
* @property {(space: string, writes: Array<{action: 'create'|'update'|'put'|'delete', collection: string, rkey: string, record?: Object}>) => Promise<unknown>} [spaceWrite] - commits into a permissioned space, injected by the platform from the spaces package. Absent, a drive can only be written in the public repo.
* @property {import('./ports.js').DriveSpaceReader} [spaceReader] - reads space records back, so a folder can be checked before something is written into it
*/
/**
* A space record's URI, which namespaces by the DID that wrote it because
* every member of a space keeps their own copy. The blob link table is keyed
* on this exact string, so it has to match what the space write route builds.
* @param {string} space
* @param {string} did
* @param {string} collection
* @param {string} rkey
* @returns {string}
*/
function spaceRecordUri(space, did, collection, rkey) {
return `${space}/${did}/${collection}/${rkey}`;
}
/**
* Join a list of chunks into one array.
* @param {Uint8Array[]} chunks
* @param {number} size
* @returns {Uint8Array}
*/
function join(chunks, size) {
const joined = new Uint8Array(size);
let offset = 0;
for (const chunk of chunks) {
joined.set(chunk, offset);
offset += chunk.length;
}
return joined;
}
/**
* @param {DriveWriterContext} ctx
* @returns {import('./ports.js').DriveWriterPort}
*/
export function createDriveWriter(ctx) {
const {
actorStorage,
blobs,
getDid,
createCommit,
chunkSize,
hasher = createSha256,
spaceWrite,
spaceReader,
} = ctx;
if (!actorStorage) throw new Error('createDriveWriter requires actorStorage');
if (!blobs) throw new Error('createDriveWriter requires blobs');
if (!getDid) throw new Error('createDriveWriter requires getDid');
if (!createCommit) throw new Error('createDriveWriter requires createCommit');
if (!chunkSize) throw new Error('createDriveWriter requires chunkSize');
const maxFileSize = ctx.maxFileSize ?? chunkSize * DEFAULT_MAX_CHUNKS;
/**
* Every folder record, keyed by record key.
* @param {string|null|undefined} space
* @returns {Promise<Map<string, {name: string, parent: string, value: Record<string, unknown>}>>}
*/
async function readFolders(space) {
/** @type {Map<string, {name: string, parent: string, value: Record<string, unknown>}>} */
const folders = new Map();
/** @type {Array<{rkey: string, value: Record<string, unknown>}>} */
const rows = [];
if (space) {
if (!spaceReader) return folders;
const listing = await spaceReader.listRecords(
space,
DRIVE_FOLDER_COLLECTION,
FOLDER_LIMIT,
);
for (const record of listing.records) {
rows.push({
rkey: record.rkey,
value: /** @type {Record<string, unknown>} */ (record.value),
});
}
} else {
/** @type {string|null} */
let cursor = null;
while (rows.length < FOLDER_LIMIT) {
const page = await actorStorage.listRecords(
DRIVE_FOLDER_COLLECTION,
cursor,
100,
);
for (const record of page.records) {
rows.push({
rkey: record.uri.split('/').pop() || '',
value: /** @type {Record<string, unknown>} */ (
cborDecode(
record.value instanceof Uint8Array
? record.value
: new Uint8Array(record.value),
)
),
});
}
if (!page.cursor || page.records.length === 0) break;
cursor = page.cursor;
}
}
for (const row of rows) {
if (typeof row.value.name !== 'string') continue;
folders.set(row.rkey, {
name: row.value.name,
parent: typeof row.value.parent === 'string' ? row.value.parent : '',
value: row.value,
});
}
return folders;
}
/**
* Commit one record, into the repo or into a space.
* @param {string} did
* @param {string|null|undefined} space
* @param {string} collection
* @param {string} rkey
* @param {Record<string, unknown>} record
* @param {string[]} [blobCids] - parts to link, for a record that holds any
* @returns {Promise<{uri: string, cid: string}>}
*/
async function writeRecord(did, space, collection, rkey, record, blobCids) {
if (space && spaceWrite) {
const uri = spaceRecordUri(space, did, collection, rkey);
// The space commit advances the set hash. It does not index blobs:
// only the space write route does that, so the parts are linked here
// instead. Without the link the reap takes them, and worse, an
// unreferenced blob is served by the public getBlob endpoint.
const commit = /** @type {{results?: Array<{cid: string|null}>}} */ (
await spaceWrite(space, [{ action: 'put', collection, rkey, record }])
);
await actorStorage.unlinkBlobsFromRecord(uri);
for (const cid of blobCids ?? []) {
await actorStorage.linkBlobToRecord(cid, uri, Date.now());
}
return { uri, cid: commit?.results?.[0]?.cid ?? '' };
}
const uri = `at://${did}/${collection}/${rkey}`;
const encoded = cborEncodeDagCbor(record);
const cid = cidToString(await createCid(encoded));
await actorStorage.putBlock(cid, encoded);
await actorStorage.putRecord(uri, cid, collection, rkey, encoded);
// Linked before the commit, so a reap between the two cannot take a part
// this record already holds. An update replaces the old links.
await actorStorage.unlinkBlobsFromRecord(uri);
for (const cid of blobCids ?? []) {
await actorStorage.linkBlobToRecord(cid, uri, Date.now());
}
await createCommit(did, [{ action: 'create', uri, cid }]);
return { uri, cid };
}
/**
* Write several records as one commit. A cascade touches every file under a
* folder, and one commit each would be as many commits as there are files.
* @param {string} did
* @param {string|null|undefined} space
* @param {Array<{collection: string, rkey: string, record: Record<string, unknown>}>} writes
* @returns {Promise<void>}
*/
async function writeMany(did, space, writes) {
if (writes.length === 0) return;
if (space && spaceWrite) {
// The space writer caps one commit, so a long cascade goes in batches.
for (let at = 0; at < writes.length; at += SPACE_WRITE_BATCH) {
await spaceWrite(
space,
writes.slice(at, at + SPACE_WRITE_BATCH).map((write) => ({
action: /** @type {'put'} */ ('put'),
collection: write.collection,
rkey: write.rkey,
record: write.record,
})),
);
}
return;
}
/** @type {Array<{action: string, uri: string, cid: string}>} */
const ops = [];
for (const write of writes) {
const uri = `at://${did}/${write.collection}/${write.rkey}`;
const encoded = cborEncodeDagCbor(write.record);
const cid = cidToString(await createCid(encoded));
await actorStorage.putBlock(cid, encoded);
await actorStorage.putRecord(
uri,
cid,
write.collection,
write.rkey,
encoded,
);
ops.push({ action: 'create', uri, cid });
}
await createCommit(did, ops);
}
/**
* Remove one record, from the repo or from a space.
* @param {string} did
* @param {string|null|undefined} space
* @param {string} collection
* @param {string} rkey
* @returns {Promise<boolean>}
*/
async function removeRecord(did, space, collection, rkey) {
if (space) {
if (!spaceWrite) return false;
try {
await spaceWrite(space, [{ action: 'delete', collection, rkey }]);
} catch {
// The space write path reports a missing record by throwing; the
// caller only needs to know it was not there.
return false;
}
// Linked here on the way in, so unlinked here on the way out.
await actorStorage.unlinkBlobsFromRecord(
spaceRecordUri(space, did, collection, rkey),
);
return true;
}
const uri = `at://${did}/${collection}/${rkey}`;
if (!(await actorStorage.getRecord(uri))) return false;
await actorStorage.unlinkBlobsFromRecord(uri);
await actorStorage.deleteRecord(uri);
await createCommit(did, [{ action: 'delete', uri }]);
return true;
}
/**
* Every file record, as {rkey, value}.
* @param {string|null|undefined} space
* @returns {Promise<Array<{rkey: string, value: Record<string, unknown>}>>}
*/
async function readFiles(space) {
/** @type {Array<{rkey: string, value: Record<string, unknown>}>} */
const rows = [];
if (space) {
if (!spaceReader) return rows;
const listing = await spaceReader.listRecords(
space,
DRIVE_FILE_COLLECTION,
FOLDER_LIMIT,
);
for (const record of listing.records) {
rows.push({
rkey: record.rkey,
value: /** @type {Record<string, unknown>} */ (record.value),
});
}
return rows;
}
/** @type {string|null} */
let cursor = null;
while (rows.length < FOLDER_LIMIT) {
const page = await actorStorage.listRecords(
DRIVE_FILE_COLLECTION,
cursor,
100,
);
for (const record of page.records) {
rows.push({
rkey: record.uri.split('/').pop() || '',
value: /** @type {Record<string, unknown>} */ (
cborDecode(
record.value instanceof Uint8Array
? record.value
: new Uint8Array(record.value),
)
),
});
}
if (!page.cursor || page.records.length === 0) break;
cursor = page.cursor;
}
return rows;
}
/**
* The folder and everything under it, at any depth. A folder goes to the
* trash with what it holds, the way every drive does it, so nothing is left
* loose at the top level with its folder gone.
* @param {string} rkey
* @param {Map<string, {parent: string}>} folders
* @param {Array<{rkey: string, value: Record<string, unknown>}>} files
* @returns {{folders: string[], files: string[]}}
*/
function subtreeOf(rkey, folders, files) {
const within = new Set([rkey]);
// A parent always precedes its children in no particular order, so this
// repeats until nothing new joins rather than assuming a walk order.
let grew = true;
while (grew) {
grew = false;
for (const [key, folder] of folders) {
if (!within.has(key) && within.has(folder.parent)) {
within.add(key);
grew = true;
}
}
}
return {
folders: [...within],
files: files
.filter((file) => within.has(String(file.value.folder ?? '')))
.map((file) => file.rkey),
};
}
/**
* Reject a parent that does not exist, or a move that would put a folder
* inside itself.
* @param {Map<string, {parent: string}>} folders
* @param {string} parent
* @param {string} [moving] - the folder being moved
* @returns {string}
*/
function checkParent(folders, parent, moving) {
if (!parent) return '';
if (!folders.has(parent)) throw new Error('No such folder.');
let at = parent;
for (let step = 0; at && step <= MAX_FOLDER_DEPTH; step += 1) {
if (moving && at === moving) {
throw new Error('A folder cannot be moved inside itself.');
}
at = folders.get(at)?.parent ?? '';
}
return parent;
}
/**
* Store one chunk as a blob and describe it as a record part.
* @param {string} did
* @param {Uint8Array} data
* @param {string} mimeType
*/
async function putPart(did, data, mimeType) {
const digest = await createBlobCid(data);
const cid = cidToString(digest);
await actorStorage.putBlob(cid, mimeType, data.length);
await blobs.put(did, cid, data, mimeType);
// `{$link}` is the shape every write path in this repo stores, because
// createRecord encodes the JSON body as it arrives. findBlobRefs finds
// this shape and not a CID instance, so the space write path can only
// link a part written this way.
return {
cid,
part: {
$type: 'blob',
ref: { $link: cid },
mimeType,
size: data.length,
},
};
}
/**
* @typedef {Object} StoredBytes
* @property {Array<{$type: string, ref: {$link: string}, mimeType: string, size: number}>} parts
* @property {string[]} partCids
* @property {number} total
*/
/**
* @typedef {Object} UploadSession
* @property {string} did
* @property {string} folderKey
* @property {string|null} space
* @property {string} name
* @property {string} type
* @property {import('@pdsjs/core/ports').BlobWriter} writer
* @property {import('@pdsjs/core/ports').Hasher} hash
* @property {number} total
* @property {number} touched
* @property {boolean} busy
*/
/**
* Multi-request uploads in flight, keyed by session id. In memory: a
* platform restart between parts fails the upload and the client starts
* it over, which content addressing makes safe.
* @type {Map<string, UploadSession>}
*/
const uploads = new Map();
/**
* The named session, ready for its next request.
* @param {string} id
* @returns {UploadSession}
*/
function takeUpload(id) {
const session = uploads.get(id);
if (!session) throw new Error('No such upload.');
if (session.busy) throw new Error('That upload has a request in flight.');
return session;
}
/**
* Forget one session and abandon its staged bytes.
* @param {string} id
*/
async function dropUpload(id) {
const session = uploads.get(id);
if (!session) return;
uploads.delete(id);
try {
await session.writer.abort();
} catch {
// staged bytes an adapter cannot reach expire on their own
}
}
/** Abandon every session that has waited past the TTL. */
async function sweepUploads() {
const cutoff = Date.now() - UPLOAD_SESSION_TTL;
for (const [id, session] of uploads) {
if (session.touched < cutoff && !session.busy) await dropUpload(id);
}
}
/**
* The checks every file write shares, before any bytes move.
* @param {string} name
* @param {string|null|undefined} folder
* @param {string|null|undefined} space
* @returns {Promise<{did: string, folderKey: string}>}
*/
async function checkFileTarget(name, folder, space) {
const did = await getDid();
if (!did) throw new Error('This server has no account yet.');
if (space && !spaceWrite) {
throw new Error('This server cannot write to a space.');
}
if (!isValidName(name)) throw new Error('That file name is not valid.');
const folderKey = normalizeFolderKey(folder);
if (folderKey === null) throw new Error('No such folder.');
if (folderKey) {
const folders = await readFolders(space);
if (!folders.has(folderKey)) throw new Error('No such folder.');
}
return { did, folderKey };
}
/**
* Write the file record over bytes already in blob storage.
* @param {string} did
* @param {string|null|undefined} space
* @param {string} name
* @param {string} folderKey
* @param {string} type
* @param {StoredBytes} stored
* @returns {Promise<import('./ports.js').DriveFileRow>}
*/
async function commitFileRecord(did, space, name, folderKey, type, stored) {
const { parts, partCids, total } = stored;
const rkey = createTid();
const now = new Date().toISOString();
const record = {
$type: DRIVE_FILE_COLLECTION,
name,
...(folderKey ? { folder: folderKey } : {}),
parts,
mimeType: type,
createdAt: now,
};
const written = await writeRecord(
did,
space,
DRIVE_FILE_COLLECTION,
rkey,
record,
partCids,
);
return {
rkey,
uri: written.uri,
cid: written.cid,
name,
folder: folderKey,
mimeType: type,
size: total,
partCount: parts.length,
createdAt: now,
updatedAt: null,
};
}
/**
* Store the stream as one blob of the file's own type, so its CID is
* something any record in the repo can reference. One chunk and one part
* buffer are the most this holds; the blob port stages the rest.
* @param {string} did
* @param {ReadableStream} body
* @param {string} type
* @returns {Promise<StoredBytes>}
*/
async function storeWhole(did, body, type) {
const writer = await /** @type {NonNullable<typeof blobs.createWriter>} */ (
blobs.createWriter
)(did);
const hash = hasher();
const reader = body.getReader();
let total = 0;
try {
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk =
value instanceof Uint8Array ? value : new Uint8Array(value);
total += chunk.length;
if (total > maxFileSize) {
await reader.cancel();
throw new Error(
`That file is larger than this server accepts (${maxFileSize} bytes).`,
);
}
await hash.update(chunk);
await writer.write(chunk);
}
const cid = cidToString(blobCidFromDigest(await hash.digest()));
await actorStorage.putBlob(cid, type, total);
await writer.commit(cid, type);
return {
parts: [
{ $type: 'blob', ref: { $link: cid }, mimeType: type, size: total },
],
partCids: [cid],
total,
};
} catch (e) {
try {
await writer.abort();
} catch {
// the original error is the one to surface
}
throw e;
}
}
/**
* Store the stream as chunk-sized blobs, for a port that only takes a
* whole buffer per blob. A file past one chunk splits into opaque parts.
* @param {string} did
* @param {ReadableStream} body
* @param {string} type
* @returns {Promise<StoredBytes>}
*/
async function storeChunked(did, body, type) {
const reader = body.getReader();
/** @type {Array<{$type: string, ref: {$link: string}, mimeType: string, size: number}>} */
const parts = [];
/** @type {string[]} */
const partCids = [];
/** @type {Uint8Array[]} */
let pending = [];
let pendingSize = 0;
let total = 0;
/** @param {string} partType */
const flush = async (partType) => {
const { cid, part } = await putPart(
did,
join(pending, pendingSize),
partType,
);
parts.push(part);
partCids.push(cid);
pending = [];
pendingSize = 0;
};
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk = value instanceof Uint8Array ? value : new Uint8Array(value);
total += chunk.length;
if (total > maxFileSize) {
throw new Error(
`That file is larger than this server accepts (${maxFileSize} bytes).`,
);
}
pending.push(chunk);
pendingSize += chunk.length;
// Hold a full chunk rather than writing it: a file that ends here is
// one part and keeps its own type, and only a second chunk makes it a
// split file whose parts are opaque.
while (pendingSize > chunkSize) {
const joined = join(pending, pendingSize);
pending = [joined.subarray(chunkSize)];
pendingSize = pending[0].length;
const { cid, part } = await putPart(
did,
joined.subarray(0, chunkSize),
PART_MIME_TYPE,
);
parts.push(part);
partCids.push(cid);
}
}
// The remainder, which for a file within one chunk is the whole of it.
// An empty file still gets a part, since the record requires one.
await flush(parts.length === 0 ? type : PART_MIME_TYPE);
return { parts, partCids, total };
}
/**
* The item and, for a folder, everything under it. Null when there is no
* such record.
* @param {string} rkey
* @param {string|null|undefined} space
* @returns {Promise<{folders: string[], files: string[]}|null>}
*/
async function gather(rkey, space) {
const folders = await readFolders(space);
const files = await readFiles(space);
if (folders.has(rkey)) return subtreeOf(rkey, folders, files);
if (files.some((file) => file.rkey === rkey)) {
return { folders: [], files: [rkey] };
}
return null;
}
/**
* Put an item in the trash, or take it out, by stamping every record it
* covers. The parts stay linked either way: a trashed file keeps its bytes,
* which is what makes taking it back out possible at all.
* @param {string} rkey
* @param {string|null|undefined} space
* @param {string|null} at - ISO time to trash at, null to restore
* @returns {Promise<number>} how many records the change covered
*/
async function stamp(rkey, space, at) {
const did = await getDid();
if (!did) return 0;
if (space && !spaceWrite) return 0;
const found = await gather(rkey, space);
if (!found) return 0;
const folders = await readFolders(space);
const files = await readFiles(space);
/** @type {Array<{collection: string, rkey: string, record: Record<string, unknown>}>} */
const writes = [];
/** @param {string} collection @param {string} key @param {Record<string, unknown>} value */
const change = (collection, key, value) => {
/** @type {Record<string, unknown>} */
const record = { ...value, $type: collection };
if (at) record.deletedAt = at;
else delete record.deletedAt;
writes.push({ collection, rkey: key, record });
};
for (const key of found.folders) {
const folder = folders.get(key);
if (folder) change(DRIVE_FOLDER_COLLECTION, key, folder.value);
}
for (const key of found.files) {
const file = files.find((row) => row.rkey === key);
if (file) change(DRIVE_FILE_COLLECTION, key, file.value);
}
await writeMany(did, space, writes);
return writes.length;
}
/**
* @param {string} rkey
* @param {string|null|undefined} space
* @param {string} collection - what the caller believes it is deleting
* @returns {Promise<boolean>}
*/
async function trash(rkey, space, collection) {
const folders = await readFolders(space);
const isFolder = folders.has(rkey);
if (collection === DRIVE_FOLDER_COLLECTION && !isFolder) return false;
if (collection === DRIVE_FILE_COLLECTION && isFolder) return false;
return (await stamp(rkey, space, new Date().toISOString())) > 0;
}
return {
async createFolder({ name, parent, space }) {
const did = await getDid();
if (!did) throw new Error('This server has no account yet.');
if (space && !spaceWrite) {
throw new Error('This server cannot write to a space.');
}
if (!isValidName(name)) throw new Error('That folder name is not valid.');
const parentKey = normalizeFolderKey(parent);
if (parentKey === null) throw new Error('No such folder.');
const folders = await readFolders(space);
checkParent(folders, parentKey);
for (const folder of folders.values()) {
if (folder.parent === parentKey && folder.name === name) {
throw new Error('A folder of that name is already here.');
}
}
const rkey = createTid();
const now = new Date().toISOString();
const record = {
$type: DRIVE_FOLDER_COLLECTION,
name,
...(parentKey ? { parent: parentKey } : {}),
createdAt: now,
};
const written = await writeRecord(
did,
space,
DRIVE_FOLDER_COLLECTION,
rkey,
record,
);
return {
rkey,
uri: written.uri,
cid: written.cid,
name,
parent: parentKey,
createdAt: now,
};
},
async renameFolder({ rkey, name, parent, space }) {
const did = await getDid();
if (!did) throw new Error('This server has no account yet.');
if (space && !spaceWrite) {
throw new Error('This server cannot write to a space.');
}
const folders = await readFolders(space);
const folder = folders.get(rkey);
if (!folder) throw new Error('No such folder.');
if (name !== undefined && !isValidName(name)) {
throw new Error('That folder name is not valid.');
}
let parentKey = folder.parent;
if (parent !== undefined) {
const requested = normalizeFolderKey(parent);
if (requested === null) throw new Error('No such folder.');
parentKey = checkParent(folders, requested, rkey);
}
const now = new Date().toISOString();
/** @type {Record<string, unknown>} */
const record = {
...folder.value,
$type: DRIVE_FOLDER_COLLECTION,
name: name ?? folder.name,
updatedAt: now,
};
if (parentKey) record.parent = parentKey;
else delete record.parent;
const written = await writeRecord(
did,
space,
DRIVE_FOLDER_COLLECTION,
rkey,
record,
);
return {
rkey,
uri: written.uri,
cid: written.cid,
name: /** @type {string} */ (record.name),
parent: parentKey,
createdAt:
typeof folder.value.createdAt === 'string'
? folder.value.createdAt
: now,
};
},
async deleteFolder(rkey, space) {
return trash(rkey, space, DRIVE_FOLDER_COLLECTION);
},
async restore(rkey, space) {
return stamp(rkey, space, null);
},
async purge(rkey, space) {
const did = await getDid();
if (!did) return 0;
const found = await gather(rkey, space);
if (!found) return 0;
// Unlinked here, so the orphan sweep is what frees the bytes. The
// record goes with it: nothing is left naming a blob that is gone.
for (const key of found.files) {
await removeRecord(did, space, DRIVE_FILE_COLLECTION, key);
}
for (const key of found.folders) {
await removeRecord(did, space, DRIVE_FOLDER_COLLECTION, key);
}
return found.files.length + found.folders.length;
},
async createFile({ name, folder, mimeType, body, space }) {
const { did, folderKey } = await checkFileTarget(name, folder, space);
if (!body) throw new Error('No file was sent.');
const type = mimeType || DEFAULT_MIME_TYPE;
const stored = blobs.createWriter
? await storeWhole(did, body, type)
: await storeChunked(did, body, type);
return commitFileRecord(did, space, name, folderKey, type, stored);
},
// The session surface only exists over a blob port that streams, so its
// absence is what tells the account API to say no.
...(blobs.createWriter
? {
uploadPartSize: UPLOAD_PART_SIZE,
/**
* @param {{name: string, folder: string|null, mimeType: string, space?: string|null}} file
*/
async createUpload({ name, folder, mimeType, space }) {
const { did, folderKey } = await checkFileTarget(
name,
folder,
space,
);
await sweepUploads();
const id = crypto.randomUUID();
uploads.set(id, {
did,
folderKey,
space: space || null,
name,
type: mimeType || DEFAULT_MIME_TYPE,
writer:
await /** @type {NonNullable<typeof blobs.createWriter>} */ (
blobs.createWriter
)(did),
hash: hasher(),
total: 0,
touched: Date.now(),
busy: false,
});
return { id };
},
/**
* @param {string} id
* @param {ReadableStream<Uint8Array>} body
*/
async appendUpload(id, body) {
const session = takeUpload(id);
session.busy = true;
try {
const reader = body.getReader();
while (true) {
const { done, value } = await reader.read();
if (done) break;
const chunk =
value instanceof Uint8Array ? value : new Uint8Array(value);
session.total += chunk.length;
if (session.total > maxFileSize) {
await reader.cancel();
throw new Error(
`That file is larger than this server accepts (${maxFileSize} bytes).`,
);
}
await session.hash.update(chunk);
await session.writer.write(chunk);
}
session.busy = false;
session.touched = Date.now();
return { received: session.total };
} catch (e) {
// A part that fails partway has already fed the hash, so the
// session cannot go on. The client starts the upload over.
await dropUpload(id);
throw e;
}
},
/** @param {string} id */
async finishUpload(id) {
const session = takeUpload(id);
session.busy = true;
uploads.delete(id);
const { did, space, name, folderKey, type, writer, hash, total } =
session;
try {
const cid = cidToString(blobCidFromDigest(await hash.digest()));
await actorStorage.putBlob(cid, type, total);
await writer.commit(cid, type);
const stored = {
parts: [
{
$type: 'blob',
ref: { $link: cid },
mimeType: type,
size: total,
},
],
partCids: [cid],
total,
};
return await commitFileRecord(
did,
space,
name,
folderKey,
type,
stored,
);
} catch (e) {
try {
await writer.abort();
} catch {
// the original error is the one to surface
}
throw e;
}
},
/** @param {string} id */
async abortUpload(id) {
await dropUpload(id);
},
}
: {}),
async deleteFile(rkey, space) {
return trash(rkey, space, DRIVE_FILE_COLLECTION);
},
};
}