chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
/**
* One account's PDS, reached over the wire.
*
* The git appview holds no storage of its own. Every record and blob it
* reads comes from the account's PDS through the public XRPC surface, and
* every write it makes on an account's behalf goes back through the same
* surface with that account's credentials. This module is the storage
* adapter that makes the ordinary ExtensionHost calls land on the wire.
*
* The read surface is anonymous, because a PDS's records and blobs are
* public unless a space holds them. The write surface needs a token, which
* arrives with the request that caused the write; `getToken` reads it from
* the dispatching context.
*/
import { cborDecode, cborEncodeDagCbor } from '@pdsjs/core/repo';
/**
* What carries a request; the global fetch unless one is injected.
* @typedef {(input: string|URL|Request, init?: RequestInit) => Promise<Response>} FetchLike
*/
/**
* @param {string} uri - at://did/collection/rkey
* @returns {[string, string, string]|null}
*/
function splitUri(uri) {
const parts = uri.replace(/^at:\/\//, '').split('/');
if (parts.length !== 3) return null;
return /** @type {[string, string, string]} */ (parts);
}
/**
* @param {Object} options
* @param {string} options.service - the PDS base URL
* @param {string} options.account - the hosted account's handle or DID
* @param {FetchLike} [options.fetch]
* @param {() => Promise<string|null>} [options.getToken] - the caller's
* access token, for the writes a request causes
* @returns {RemoteAccount}
*/
export function createRemoteAccount(options) {
const service = options.service.replace(/\/+$/, '');
const send =
options.fetch ?? ((input, init) => globalThis.fetch(input, init));
const getToken = options.getToken ?? (async () => null);
/** Resolved once from describeRepo: the DID and the handle. */
let identity = null;
const readIdentity = async () => {
identity ??= (async () => {
const params = new URLSearchParams({ repo: options.account });
const res = await send(
`${service}/xrpc/com.atproto.repo.describeRepo?${params}`,
);
if (!res.ok) {
throw new Error(
`the account ${options.account} does not answer at ${service}`,
);
}
const body = /** @type {{did: string, handle: string}} */ (
await res.json()
);
return { did: body.did, handle: body.handle };
})();
return identity;
};
/**
* The server's own XRPC reads, re-addressed to the PDS. This is the fetch
* the smart HTTP handler and the browser read through: a Request naming
* any origin comes back aimed at the PDS, headers and body intact.
* @param {Request} request
* @returns {Promise<Response>}
*/
const xrpc = async (request) => {
const url = new URL(request.url);
return send(new Request(`${service}${url.pathname}${url.search}`, request));
};
return {
service,
xrpc,
getDid: async () => (await readIdentity()).did,
getHandle: async () => (await readIdentity()).handle,
/**
* One blob from the PDS, whole.
* @param {string} did
* @param {string} cid
* @returns {Promise<{data: Uint8Array, mimeType: string}|null>}
*/
getBlob: async (did, cid) => {
const params = new URLSearchParams({ did, cid });
const res = await send(
`${service}/xrpc/com.atproto.sync.getBlob?${params}`,
);
if (!res.ok) return null;
return {
data: new Uint8Array(await res.arrayBuffer()),
mimeType: res.headers.get('content-type') ?? 'application/octet-stream',
};
},
/**
* One byte range of a blob, as the lazy readers ask for it. The end is
* inclusive, matching the port's `getStream`.
* @param {string} did
* @param {string} cid
* @param {{start: number, end: number}} [range]
* @returns {Promise<{body: ReadableStream<Uint8Array>, mimeType: string}|null>}
*/
getBlobStream: async (did, cid, range) => {
const params = new URLSearchParams({ did, cid });
const headers = range
? { range: `bytes=${range.start}-${range.end}` }
: undefined;
const res = await send(
`${service}/xrpc/com.atproto.sync.getBlob?${params}`,
{ headers },
);
if (!res.ok || !res.body) return null;
return {
body: res.body,
mimeType: res.headers.get('content-type') ?? 'application/octet-stream',
};
},
/**
* @param {string} collection
* @param {string|null} cursor
* @param {number} limit
* @returns {Promise<{records: Array<{uri: string, cid: string, value: Uint8Array}>, cursor: string|null}>}
*/
listRecords: async (collection, cursor, limit) => {
const did = (await readIdentity()).did;
const params = new URLSearchParams({
repo: did,
collection,
limit: String(limit),
});
if (cursor) params.set('cursor', cursor);
const res = await send(
`${service}/xrpc/com.atproto.repo.listRecords?${params}`,
);
if (!res.ok) return { records: [], cursor: null };
const body = /** @type {{records?: any[], cursor?: string}} */ (
await res.json()
);
return {
records: (body.records ?? []).map((row) => ({
uri: String(row.uri),
cid: String(row.cid),
value: cborEncodeDagCbor(row.value),
})),
cursor: typeof body.cursor === 'string' ? body.cursor : null,
};
},
/**
* @param {string} uri
* @returns {Promise<{cid: string, value: Uint8Array}|null>}
*/
getRecord: async (uri) => {
const parts = splitUri(uri);
if (!parts) return null;
const params = new URLSearchParams({
repo: parts[0],
collection: parts[1],
rkey: parts[2],
});
const res = await send(
`${service}/xrpc/com.atproto.repo.getRecord?${params}`,
);
if (!res.ok) return null;
const body = /** @type {{cid: string, value: unknown}} */ (
await res.json()
);
return { cid: body.cid, value: cborEncodeDagCbor(body.value) };
},
/**
* A record write, carried by the caller's credentials. The PDS computes
* the commit; `swapRecord` holds the previous version's cid, so a write
* that raced another is refused instead of landing blind.
* @param {string} uri
* @param {string} collection
* @param {string} rkey
* @param {Uint8Array} value - the record, CBOR-encoded
* @param {string|null} [swapRecord] - the previous version's cid
*/
putRecord: async (uri, collection, rkey, value, swapRecord) => {
const parts = splitUri(uri);
if (!parts) throw new Error(`not a record uri: ${uri}`);
const token = await getToken();
const res = await send(`${service}/xrpc/com.atproto.repo.putRecord`, {
method: 'POST',
headers: {
'content-type': 'application/json',
...(token ? { authorization: `Bearer ${token}` } : {}),
},
body: JSON.stringify({
repo: parts[0],
collection,
rkey,
record: cborDecode(value),
...(swapRecord ? { swapRecord } : {}),
}),
});
if (!res.ok) {
throw new Error(
`record write refused (${res.status}): ${await res.text()}`,
);
}
},
/**
* One blob uploaded with the caller's credentials. The PDS computes the
* content address; the local `cid` argument is what the caller derived
* and the server confirms.
* @param {Uint8Array} data
* @param {string} mimeType
*/
uploadBlob: async (data, mimeType) => {
const token = await getToken();
const form = new FormData();
form.append(
'file',
new Blob([/** @type {BlobPart} */ (data)], { type: mimeType }),
);
const res = await send(`${service}/xrpc/com.atproto.repo.uploadBlob`, {
method: 'POST',
headers: token ? { authorization: `Bearer ${token}` } : {},
body: form,
});
if (!res.ok) {
throw new Error(
`blob upload refused (${res.status}): ${await res.text()}`,
);
}
},
};
}
/**
* @typedef {Object} RemoteAccount
* @property {string} service
* @property {(request: Request) => Promise<Response>} xrpc - the PDS's own
* XRPC surface, re-addressed; the smart HTTP handler and the browser read
* through this
* @property {() => Promise<string>} getDid
* @property {() => Promise<string>} getHandle
* @property {(did: string, cid: string) => Promise<{data: Uint8Array, mimeType: string}|null>} getBlob
* @property {(did: string, cid: string, range?: {start: number, end: number}) => Promise<{body: ReadableStream<Uint8Array>, mimeType: string}|null>} getBlobStream
* @property {(collection: string, cursor: string|null, limit: number) => Promise<{records: Array<{uri: string, cid: string, value: Uint8Array}>, cursor: string|null}>} listRecords
* @property {(uri: string) => Promise<{cid: string, value: Uint8Array}|null>} getRecord
* @property {(uri: string, collection: string, rkey: string, value: Uint8Array, swapRecord?: string|null) => Promise<void>} putRecord
* @property {(data: Uint8Array, mimeType: string) => Promise<void>} uploadBlob
*/