Something went wrong. Try again.
An AT Protocol Personal Data Server written in JavaScript pdsjs.dev
pds atproto
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261/** * 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 */