chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* The /account/api/drive routes: folder listings, downloads, uploads in one
* request or in parts, the trash, folders, drives kept in spaces and the
* change feed, each behind the owner's account session.
*
* This module must stay free of node-only imports; it runs on Cloudflare
* Workers.
*/
import { parseByteRange } from '@pdsjs/core/http';
import { cborDecode } from '@pdsjs/core/repo';
import { DRIVE_SPACE_TYPE } from './lexicon.js';
// How many events a drive change query walks back before it tells the caller
// to enumerate instead. A client away longer than this is cheaper to refresh
// wholesale than to replay.
const DRIVE_CHANGE_SCAN = 1000;
// A space key is the name its owner gave the drive, so it has to read as one.
const DRIVE_SPACE_KEY = /^[a-z0-9][a-z0-9-]{0,62}$/;
/**
* The drives among an account's spaces. Every drive space shares one type, so
* the key is what names each: the owner chose it when the drive was made.
* @param {Array<{uri: string, spaceType: string, deletedAt?: string|null}>} rows
* @returns {Array<{uri: string, type: string, key: string, name: string}>}
*/
function driveSpaces(rows) {
return rows
.filter((row) => row.spaceType === DRIVE_SPACE_TYPE && !row.deletedAt)
.map((row) => {
const key = row.uri.split('/').pop() ?? '';
return { uri: row.uri, type: row.spaceType, key, name: key };
});
}
/**
* An attachment header naming the download. The quoted form carries ASCII
* alone, so a name with any other character rides the RFC 5987 field beside
* it, which every current browser prefers.
* @param {string} name
* @returns {string}
*/
function contentDisposition(name) {
const ascii = name.replace(/[^\x20-\x7e]/g, '_').replace(/["\\]/g, '_');
return `attachment; filename="${ascii}"; filename*=UTF-8''${encodeURIComponent(name)}`;
}
/**
* The part of the space admin port a drive needs.
* @typedef {Pick<import('@pdsjs/core/ports').SpaceAdminPort, 'createSpace'>} DriveSpaceAdmin
*/
/**
* @param {Object} ctx
* @param {import('@pdsjs/core/ports').ExtensionHost} ctx.host
* @param {import('./ports.js').DriveBrowserPort} ctx.browser
* @param {import('./ports.js').DriveWriterPort|null} ctx.writer - Absent, the drive lists and downloads but cannot upload.
* @param {import('@pdsjs/core/ports').SpaceBrowserPort|null} ctx.spaceBrowser - Lists the drives held in spaces. Absent, the page offers none.
* @param {DriveSpaceAdmin|null} ctx.spaceAdmin - Makes a drive space. Absent, that request refuses.
* @returns {Record<string, import('@pdsjs/core/ports').ExtensionRoute>}
*/
export function createDriveAccountRoutes({
host,
browser,
writer,
spaceBrowser,
spaceAdmin,
}) {
const { accountApi, actorStorage, getDid, reapOrphanedBlobs } = host;
/**
* GET /account/api/drive/files?folder= - One folder's listing: the files
* directly in it, and the names of the folders directly under it.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveFiles(request, url) {
return accountApi(request, url, async () => {
const listing = await browser.listFolder(
url.searchParams.get('folder'),
url.searchParams.get('space'),
);
if (!listing) throw new Error('That folder name is not valid.');
// The spaces this drive can also live in, so the page can offer them.
const rows = spaceBrowser ? await spaceBrowser.listSpaces() : [];
return {
enabled: true,
...listing,
spaces: driveSpaces(rows),
canMakeSpace: Boolean(spaceAdmin && writer),
// Present when the writer takes a file across several requests; the
// page slices anything larger than one part.
uploadPartSize:
writer?.createUpload && writer.uploadPartSize
? writer.uploadPartSize
: null,
};
});
}
/**
* GET /account/api/drive/file?rkey= - One stored file, as its own bytes.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveFile(request, url) {
return accountApi(request, url, async () => {
const rkey = url.searchParams.get('rkey');
if (!rkey) throw new Error('rkey parameter required');
const space = url.searchParams.get('space');
const stat = await browser.statFile(rkey, space);
if (!stat) throw new Error('No such file.');
/** @type {Record<string, string>} */
const headers = {
'Content-Type': stat.mimeType,
'Content-Disposition': contentDisposition(stat.name),
// A file split across blobs is still one byte range to a client, so
// the whole of it need never be held to serve part of it.
'Accept-Ranges': 'bytes',
'Cache-Control': 'private, no-store',
'X-Content-Type-Options': 'nosniff',
'Content-Security-Policy': "default-src 'none'; sandbox",
};
const requested = parseByteRange(request.headers.get('range'), stat.size);
if (requested === 'unsatisfiable') {
return new Response(null, {
status: 416,
headers: { ...headers, 'Content-Range': `bytes */${stat.size}` },
});
}
if (requested) {
const part = await browser.readRange(
rkey,
requested.start,
requested.end,
space,
);
if (!part) throw new Error('No such file.');
return new Response(/** @type {BodyInit} */ (part.bytes), {
status: 206,
headers: {
...headers,
'Content-Length': String(part.bytes.length),
'Content-Range': `bytes ${part.start}-${part.end}/${part.total}`,
},
});
}
const file = await browser.readFile(rkey, space);
if (!file) throw new Error('No such file.');
return new Response(/** @type {BodyInit} */ (file.bytes), {
headers: { ...headers, 'Content-Length': String(file.size) },
});
});
}
/**
* POST /account/api/drive/upload?name=&folder= - Store one file. The body is
* the file itself, streamed through to blob storage rather than buffered.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveUpload(request, url) {
return accountApi(
request,
url,
async () => {
if (!writer) throw new Error('Drive uploads are not enabled.');
return writer.createFile({
name: url.searchParams.get('name') || '',
folder: url.searchParams.get('folder'),
mimeType:
request.headers.get('content-type') || 'application/octet-stream',
body: request.body,
space: url.searchParams.get('space'),
});
},
{ write: true, raw: true },
);
}
/**
* POST /account/api/drive/upload/start - Open a multi-request upload, for
* a file larger than one request body may carry.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveUploadStart(request, url) {
return accountApi(
request,
url,
async (_did, body) => {
if (!writer?.createUpload) {
throw new Error('This server cannot take a file in parts.');
}
return writer.createUpload({
name: typeof body?.name === 'string' ? body.name : '',
folder: typeof body?.folder === 'string' ? body.folder : null,
mimeType:
typeof body?.mimeType === 'string'
? body.mimeType
: 'application/octet-stream',
space: typeof body?.space === 'string' ? body.space : null,
});
},
{ write: true },
);
}
/**
* POST /account/api/drive/upload/part?id= - The next run of bytes, whole
* and in order. A part that fails abandons the whole upload.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveUploadPart(request, url) {
return accountApi(
request,
url,
async () => {
if (!writer?.appendUpload) {
throw new Error('This server cannot take a file in parts.');
}
if (!request.body) throw new Error('No bytes were sent.');
return writer.appendUpload(
url.searchParams.get('id') || '',
request.body,
);
},
{ write: true, raw: true },
);
}
/**
* POST /account/api/drive/upload/finish - File the blob under its hash and
* write the record.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveUploadFinish(request, url) {
return accountApi(
request,
url,
async (_did, body) => {
if (!writer?.finishUpload) {
throw new Error('This server cannot take a file in parts.');
}
return writer.finishUpload(typeof body?.id === 'string' ? body.id : '');
},
{ write: true },
);
}
/**
* POST /account/api/drive/upload/abort - Abandon an upload and its staged
* bytes.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveUploadAbort(request, url) {
return accountApi(
request,
url,
async (_did, body) => {
if (!writer?.abortUpload) {
throw new Error('This server cannot take a file in parts.');
}
await writer.abortUpload(typeof body?.id === 'string' ? body.id : '');
return { aborted: true };
},
{ write: true },
);
}
/**
* GET /account/api/drive/trash - What the trash holds. A trashed file keeps
* its parts linked, so it still takes up storage until it is emptied.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveTrash(request, url) {
return accountApi(request, url, async () => {
const listing = await browser.listTrash(url.searchParams.get('space'));
if (!listing) throw new Error('No such drive.');
return { enabled: true, ...listing };
});
}
/**
* POST /account/api/drive/restore - Take an item back out of the trash,
* with everything a folder took in with it.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveRestore(request, url) {
return accountApi(
request,
url,
async (_did, body) => {
if (!writer) throw new Error('Drive is not enabled.');
const rkey = typeof body?.rkey === 'string' ? body.rkey : '';
if (!rkey) throw new Error('Name the item to restore.');
const restored = await writer.restore(
rkey,
typeof body?.space === 'string' ? body.space : null,
);
if (!restored) throw new Error('No such item.');
return { restored };
},
{ write: true },
);
}
/**
* POST /account/api/drive/purge - Remove a trashed item for good. Its parts
* are unlinked here, so the orphan sweep frees the bytes.
* @param {Request} request @param {URL} url
*/
async function handleApiDrivePurge(request, url) {
return accountApi(
request,
url,
async (did, body) => {
if (!writer) throw new Error('Drive is not enabled.');
const rkey = typeof body?.rkey === 'string' ? body.rkey : '';
if (!rkey) throw new Error('Name the item to remove.');
const purged = await writer.purge(
rkey,
typeof body?.space === 'string' ? body.space : null,
);
if (!purged) throw new Error('No such item.');
await reapOrphanedBlobs(did, Date.now() + 1000);
return { purged };
},
{ write: true },
);
}
/**
* POST /account/api/drive/space - Make a drive that lives in a permissioned
* space, so its files are private and shared with whoever joins it.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveSpaceCreate(request, url) {
return accountApi(
request,
url,
async (_did, body) => {
if (!writer) throw new Error('Drive is not enabled.');
if (!spaceAdmin) throw new Error('Spaces are not enabled.');
const name = typeof body?.name === 'string' ? body.name.trim() : '';
if (!DRIVE_SPACE_KEY.test(name)) {
throw new Error(
'Name a drive with lowercase letters, numbers and dashes.',
);
}
const { uri } = await spaceAdmin.createSpace({
type: DRIVE_SPACE_TYPE,
skey: name,
});
return { uri, type: DRIVE_SPACE_TYPE, key: name, name };
},
{ write: true },
);
}
/**
* POST /account/api/drive/folder - Make a folder. It is a record of its
* own, so a folder with nothing in it stays.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveFolderCreate(request, url) {
return accountApi(
request,
url,
async (_did, body) => {
if (!writer) throw new Error('Drive is not enabled.');
return writer.createFolder({
name: typeof body?.name === 'string' ? body.name : '',
parent: typeof body?.parent === 'string' ? body.parent : null,
space: typeof body?.space === 'string' ? body.space : null,
});
},
{ write: true },
);
}
/**
* POST /account/api/drive/folder/rename - Rename or move one folder. Every
* file keeps its own record key, so nothing else is rewritten.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveFolderRename(request, url) {
return accountApi(
request,
url,
async (_did, body) => {
if (!writer) throw new Error('Drive is not enabled.');
const rkey = typeof body?.rkey === 'string' ? body.rkey : '';
if (!rkey) throw new Error('Name the folder to rename.');
return writer.renameFolder({
rkey,
...(typeof body?.name === 'string' ? { name: body.name } : {}),
...(body?.parent === null || typeof body?.parent === 'string'
? { parent: body.parent }
: {}),
space: typeof body?.space === 'string' ? body.space : null,
});
},
{ write: true },
);
}
/**
* POST /account/api/drive/folder/delete - Remove one empty folder.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveFolderDelete(request, url) {
return accountApi(
request,
url,
async (_did, body) => {
if (!writer) throw new Error('Drive is not enabled.');
const rkey = typeof body?.rkey === 'string' ? body.rkey : '';
if (!rkey) throw new Error('Name the folder to delete.');
const removed = await writer.deleteFolder(
rkey,
typeof body?.space === 'string' ? body.space : null,
);
if (!removed) throw new Error('No such folder.');
return {};
},
{ write: true },
);
}
/**
* GET /account/api/drive/changes?since= - What changed in the drive since a
* given anchor, so a client enumerating the drive can catch up rather than
* list it again. The anchor is this repo's event sequence.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveChanges(request, url) {
return accountApi(request, url, async () => {
// A space keeps its own commit log, which this sequence does not cover.
// A client watching a space re-enumerates instead.
if (url.searchParams.get('space')) {
return {
enabled: true,
anchor: null,
reenumerate: true,
changed: [],
deleted: [],
};
}
const sinceParam = url.searchParams.get('since');
const since = sinceParam === null ? null : Number(sinceParam);
if (since !== null && !Number.isFinite(since)) {
throw new Error('That anchor is not valid.');
}
if (!actorStorage.getEventsBefore) {
return {
enabled: true,
anchor: null,
reenumerate: true,
changed: [],
deleted: [],
};
}
const anchor = await actorStorage.getLatestSeq();
// With no anchor the caller has nothing yet, so it enumerates rather
// than replaying the whole log.
if (since === null) {
return {
enabled: true,
anchor,
reenumerate: true,
changed: [],
deleted: [],
};
}
/** @type {Set<string>} */
const touched = new Set();
/** @type {Set<string>} */
const deleted = new Set();
let before = anchor + 1;
let reenumerate = false;
// Walk back to the anchor. The log is capped so a caller that has been
// away longer than DRIVE_CHANGE_SCAN events is told to enumerate.
let scanned = 0;
while (before > since + 1) {
const rows = await actorStorage.getEventsBefore(before, 100);
if (rows.length === 0) break;
for (const row of rows) {
if (row.seq <= since) continue;
let decoded;
try {
decoded = cborDecode(row.evt);
} catch {
continue;
}
for (const op of Array.isArray(decoded?.ops) ? decoded.ops : []) {
const path = String(op.path);
const slash = path.indexOf('/');
const collection = slash >= 0 ? path.slice(0, slash) : path;
if (
collection !== 'dev.pdsjs.drive.file' &&
collection !== 'dev.pdsjs.drive.folder'
) {
continue;
}
const rkey = slash >= 0 ? path.slice(slash + 1) : '';
if (op.action === 'delete') deleted.add(`${collection}/${rkey}`);
else touched.add(`${collection}/${rkey}`);
}
}
before = Math.min(...rows.map((row) => row.seq));
scanned += rows.length;
if (scanned >= DRIVE_CHANGE_SCAN) {
reenumerate = true;
break;
}
}
// A record put in the trash still exists, so the log calls it a change.
// To anything enumerating the drive it is gone, and saying otherwise
// leaves a client showing a file no listing will ever return again.
const changed = [];
for (const key of touched) {
if (deleted.has(key)) continue;
const slash = key.indexOf('/');
const uri = `at://${await getDid()}/${key.slice(0, slash)}/${key.slice(slash + 1)}`;
const record = await actorStorage.getRecord(uri);
let trashed = false;
if (record) {
try {
const value = /** @type {{deletedAt?: unknown}} */ (
cborDecode(
record.value instanceof Uint8Array
? record.value
: new Uint8Array(record.value),
)
);
trashed = typeof value.deletedAt === 'string';
} catch {
// An unreadable record is reported as a change, which makes the
// client ask for it rather than assume anything.
}
}
if (trashed) deleted.add(key);
else changed.push(key);
}
return {
enabled: true,
anchor,
reenumerate,
changed,
deleted: [...deleted],
};
});
}
/**
* POST /account/api/drive/delete - Remove one stored file.
* @param {Request} request @param {URL} url
*/
async function handleApiDriveDelete(request, url) {
return accountApi(
request,
url,
async (did, body) => {
if (!writer) throw new Error('Drive is not enabled.');
const rkey = typeof body?.rkey === 'string' ? body.rkey : '';
if (!rkey) throw new Error('Name the file to delete.');
// The trash keeps the parts linked, so the sweep leaves the bytes
// alone: taking a file back out has to give back its contents.
const removed = await writer.deleteFile(
rkey,
typeof body?.space === 'string' ? body.space : null,
);
if (!removed) throw new Error('No such file.');
return {};
},
{ write: true },
);
}
return {
'/account/api/drive/files': {
method: 'GET',
handler: handleApiDriveFiles,
},
'/account/api/drive/file': {
method: 'GET',
handler: handleApiDriveFile,
},
'/account/api/drive/upload': {
method: 'POST',
handler: handleApiDriveUpload,
},
'/account/api/drive/upload/start': {
method: 'POST',
handler: handleApiDriveUploadStart,
},
'/account/api/drive/upload/part': {
method: 'POST',
handler: handleApiDriveUploadPart,
},
'/account/api/drive/upload/finish': {
method: 'POST',
handler: handleApiDriveUploadFinish,
},
'/account/api/drive/upload/abort': {
method: 'POST',
handler: handleApiDriveUploadAbort,
},
'/account/api/drive/changes': {
method: 'GET',
handler: handleApiDriveChanges,
},
'/account/api/drive/space': {
method: 'POST',
handler: handleApiDriveSpaceCreate,
},
'/account/api/drive/trash': {
method: 'GET',
handler: handleApiDriveTrash,
},
'/account/api/drive/restore': {
method: 'POST',
handler: handleApiDriveRestore,
},
'/account/api/drive/purge': {
method: 'POST',
handler: handleApiDrivePurge,
},
'/account/api/drive/folder': {
method: 'POST',
handler: handleApiDriveFolderCreate,
},
'/account/api/drive/folder/rename': {
method: 'POST',
handler: handleApiDriveFolderRename,
},
'/account/api/drive/folder/delete': {
method: 'POST',
handler: handleApiDriveFolderDelete,
},
'/account/api/drive/delete': {
method: 'POST',
handler: handleApiDriveDelete,
},
};
}