chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* The git appview's HTTP surface, on node.
*
* One process serves everything the forge needs: the smart HTTP endpoint
* and raw files under /git/, the dev.pdsjs.git.* XRPC methods, the blob
* and identity routes the browser app reads from its own origin, and the
* browser app's static build.
*/
import { createReadStream, existsSync, statSync } from 'node:fs';
import http from 'node:http';
import { extname, join, normalize } from 'node:path';
import { Readable } from 'node:stream';
import { requestContext } from './context.js';
import { createGitAppViewHost } from './host.js';
/** File extensions the static surface answers with a fixed content type. */
/** @type {Record<string, string>} */
const MIME = {
'.html': 'text/html; charset=utf-8',
'.js': 'text/javascript; charset=utf-8',
'.css': 'text/css; charset=utf-8',
'.json': 'application/json',
'.svg': 'image/svg+xml',
'.png': 'image/png',
'.ico': 'image/x-icon',
'.woff2': 'font/woff2',
'.txt': 'text/plain; charset=utf-8',
'.map': 'application/json',
};
/**
* @param {import('node:http').IncomingMessage} req
* @param {string} origin
* @returns {Promise<Request>}
*/
async function toWebRequest(req, origin) {
const url = `${origin}${req.url ?? '/'}`;
const headers = new Headers();
for (const [name, value] of Object.entries(req.headers)) {
if (typeof value === 'string') headers.set(name, value);
else if (Array.isArray(value)) headers.set(name, value.join(', '));
}
// Undici sets its own Host and length; the originals describe another
// hop's connection.
headers.delete('host');
headers.delete('content-length');
if (req.method === 'GET' || req.method === 'HEAD') {
return new Request(url, { method: req.method, headers });
}
// Node's web stream type and the DOM's disagree on generic details; the
// bytes are what this needs.
const bytes = new Uint8Array(
await new Response(/** @type {any} */ (Readable.toWeb(req))).arrayBuffer(),
);
return new Request(url, {
method: req.method,
headers,
body: /** @type {BodyInit} */ (bytes),
});
}
/**
* @param {import('node:http').ServerResponse} res
* @param {Response} response
*/
async function reply(res, response) {
/** @type {Record<string, string>} */
const headers = {};
response.headers.forEach((value, name) => {
headers[name] = value;
});
res.writeHead(response.status, headers);
if (!response.body) {
res.end();
return;
}
const reader = response.body.getReader();
try {
for (;;) {
const { done, value } = await reader.read();
if (done) break;
if (!res.write(value)) {
await new Promise((resolve) => res.once('drain', resolve));
}
}
res.end();
} catch {
res.destroy();
} finally {
reader.releaseLock();
}
}
/**
* @param {string} root
* @param {string} pathname
* @returns {Promise<Response|null>}
*/
async function serveStatic(root, pathname) {
const relative = normalize(decodeURIComponent(pathname)).replace(
/^(\.\.[/\\])+/,
'',
);
let file = join(root, relative);
if (!file.startsWith(root)) return null;
if (!existsSync(file) || statSync(file).isDirectory()) {
// The app routes itself; anything that names no file gets the shell.
const index = join(root, 'index.html');
if (!existsSync(index)) return null;
file = index;
}
const type = MIME[extname(file)] ?? 'application/octet-stream';
const stream = Readable.toWeb(createReadStream(file));
return new Response(/** @type {any} */ (stream), {
headers: { 'content-type': type },
});
}
/**
* @param {string} header - a Range header, `bytes=start-end`
* @returns {{start: number, end: number}|undefined}
*/
function parseRange(header) {
const match = /^bytes=(\d+)-(\d*)$/.exec(header.trim());
if (!match) return undefined;
const start = Number(match[1]);
const end = match[2] ? Number(match[2]) : Number.MAX_SAFE_INTEGER;
return Number.isFinite(start) && start <= end ? { start, end } : undefined;
}
/**
* @typedef {Object} GitHostServerOptions
* @property {import('./remote.js').RemoteAccount} account
* @property {import('@pdsjs/appview/store').AppViewStore} store
* @property {import('./auth.js').ForgeAuth} [auth] - the forge owner's
* login; absent, the owner procedures answer 401
* @property {string} [hostname]
* @property {string} [plcUrl]
* @property {number} [blobUploadLimit]
* @property {(request: Request) => Promise<string|null>} [sessionDid]
* @property {string} [uiDir] - the browser app's build, served at /
* @property {string} [origin] - the origin requests are read as carrying;
* derived from the Host header when absent
* @property {(message: string) => void} [onNotice]
*/
/**
* @param {GitHostServerOptions} options
* @returns {{server: http.Server, listen: (port: number) => Promise<void>, close: () => Promise<void>}}
*/
export function createGitHostServer(options) {
const { account, store, auth } = options;
const { extension } = createGitAppViewHost({
account,
store,
hostname: options.hostname,
plcUrl: options.plcUrl,
blobUploadLimit: options.blobUploadLimit,
sessionDid: auth
? (/** @type {Request} */ request) => auth.sessionDid(request)
: options.sessionDid,
});
const routes = extension.routes ?? {};
const prefixes = extension.prefixes ?? [];
const handle = extension.handle ?? null;
const onNotice = options.onNotice ?? (() => {});
/**
* @param {Request} request
* @param {URL} url
* @returns {Promise<Response>}
*/
async function dispatch(request, url) {
// Smart HTTP and raw files, the extension's own prefix surface.
if (handle && prefixes.some((prefix) => url.pathname.startsWith(prefix))) {
const response = await handle(request, url);
if (response) return response;
}
// The dev.pdsjs.git.* methods. The bearer-token half of the route auth
// is the PDS's to verify; this host serves the anonymous reads and the
// same-origin owner session.
if (url.pathname === '/xrpc/community.lexicon.service.describe') {
// Derived from the route table, the same way the PDS derives its own,
// so the answer cannot drift from what is served.
const methods = Object.keys(routes)
.map((path) => path.slice('/xrpc/'.length))
.filter((nsid) => nsid.includes('.'))
.sort()
.map((nsid) => ({
$type: 'community.lexicon.service.describe#nsid',
value: nsid,
}));
return Response.json({ roles: ['appview'], methods });
}
const route = routes[url.pathname];
if (route) {
const method = route.method ?? 'GET';
if (
request.method !== method &&
!(request.method === 'HEAD' && method === 'GET')
) {
return Response.json(
{ error: 'InvalidRequest', message: 'Wrong method' },
{ status: 405 },
);
}
return route.handler(request, url, null);
}
const did = await account.getDid();
// The forge owner's login round trip, and the metadata document the
// client_id names.
if (auth) {
if (url.pathname === '/oauth/client-metadata.json') {
return Response.json(auth.metadataDocument());
}
if (url.pathname === '/login') {
try {
const handle =
url.searchParams.get('handle') ?? (await account.getHandle());
const { authorizeUrl } = await auth.beginLogin(handle);
return Response.redirect(authorizeUrl, 302);
} catch (err) {
return new Response(
`login failed: ${err instanceof Error ? err.message : err}\n`,
{ status: 400, headers: { 'content-type': 'text/plain' } },
);
}
}
if (url.pathname === '/oauth/callback') {
try {
const login = await auth.completeLogin(url);
return new Response(`Logged in as ${login.did}\n`, {
status: 200,
headers: {
'content-type': 'text/plain; charset=utf-8',
'set-cookie': login.cookie,
},
});
} catch (err) {
return new Response(
`login failed: ${err instanceof Error ? err.message : err}\n`,
{ status: 400, headers: { 'content-type': 'text/plain' } },
);
}
}
if (url.pathname === '/logout') {
const cookie = auth.logout(request);
return new Response('Logged out.\n', {
status: 200,
headers: {
'content-type': 'text/plain; charset=utf-8',
'set-cookie': cookie,
},
});
}
}
// The identity the browser app discovers from its own origin.
if (url.pathname === '/.well-known/atproto-did') {
return new Response(`${did}\n`, {
headers: { 'content-type': 'text/plain; charset=utf-8' },
});
}
// Blobs by cid, from the PDS, immutable. The browser app's in-page pack
// reader reads through here when the PDS's own CORS answer will not do.
if (url.pathname.startsWith('/.blobs/')) {
const cid = url.pathname.slice('/.blobs/'.length);
const rangeHeader = request.headers.get('range');
const found = await account.getBlobStream(
did,
cid,
rangeHeader ? parseRange(rangeHeader) : undefined,
);
if (!found) {
return new Response('blob not found\n', { status: 404 });
}
/** @type {Record<string, string>} */
const headers = {
'content-type': found.mimeType,
'accept-ranges': 'bytes',
etag: `"${cid}"`,
};
if (rangeHeader) {
headers['content-range'] =
`bytes ${rangeHeader.slice('bytes='.length)}`;
}
return new Response(found.body, { status: 206, headers });
}
// The browser app.
if (options.uiDir) {
const page = await serveStatic(options.uiDir, url.pathname);
if (page) return page;
}
return new Response('not found\n', { status: 404 });
}
const server = http.createServer((req, res) => {
void (async () => {
const origin =
options.origin ?? `http://${req.headers.host ?? 'localhost'}`;
try {
const request = await toWebRequest(req, origin);
const url = new URL(request.url);
// A write is made with the session this request carries: the owner's
// OAuth token, resolved here and visible to the storage layer for
// this request alone.
const token = auth ? await auth.tokenFor(request) : null;
const response = await requestContext.run({ token }, () =>
dispatch(request, url),
);
await reply(res, response);
} catch (err) {
onNotice(`request failed: ${err instanceof Error ? err.message : err}`);
if (!res.headersSent) {
res.writeHead(500, { 'content-type': 'text/plain' });
}
res.end('git host error\n');
}
})();
});
return {
server,
listen: (/** @type {number} */ port) =>
new Promise((resolve, reject) => {
server.once('error', reject);
server.listen(port, () => resolve());
}),
close: () =>
new Promise((resolve) => {
server.close(() => resolve());
}),
};
}