chadtmiller.com/pds.js
An atproto PDS in JavaScript, for Cloudflare Workers and Node
0
0
// @pdsjs/sites - server-side app installs
//
// Fulfills core's SiteInstallerPort: the account page's install sheet and
// install button run through this. The reads mirror the CLI's; the writes
// go straight to the ports instead of over XRPC, and every copied blob is
// hashed here, so integrity does not rest on a remote server's answer.
import {
cborDecode,
cidToString,
createBlobCid,
createTid,
findBlobRefs,
toDisplayValue,
} from '@pdsjs/core/repo';
import { isValidSiteName } from './hostname.js';
import { planInstall, settingsSchema } from './install.js';
import { QUERY_COLLECTION, SITE_COLLECTION } from './lexicons.js';
const INSTALL_COLLECTION = 'dev.pdsjs.app.install';
/** Upper bound on one installed file, matching the deploy CLI's default. */
const BLOB_LIMIT = 5 * 1024 * 1024;
/**
* @typedef {Object} SiteInstallerContext
* @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 {(cid: string, uri: string, time?: number) => Promise<void>} linkBlob
* @property {(record: Record<string, unknown>) => Promise<{cid: string, bytes: Uint8Array}>} encodeRecord - DAG-CBOR encode plus CID, injected from core's repo helpers by the platform
* @property {typeof fetch} [fetchImpl]
*/
/**
* Resolve a manifest at-uri to the developer's DID, PDS, and record key.
* @param {string} uri
* @param {typeof fetch} fetchImpl
*/
async function resolveManifestUri(uri, fetchImpl) {
const match = uri.match(
/^at:\/\/([^/]+)\/dev\.pdsjs\.app\.manifest\/([^/]+)$/,
);
if (!match) {
throw new Error(
'Give the app as at://<did-or-handle>/dev.pdsjs.app.manifest/<rkey>',
);
}
let did = match[1];
if (!did.startsWith('did:')) {
const resolved = await fetchImpl(
`https://public.api.bsky.app/xrpc/com.atproto.identity.resolveHandle?handle=${encodeURIComponent(did)}`,
);
if (!resolved.ok) throw new Error(`Could not resolve handle ${did}.`);
did = (await resolved.json()).did;
}
const doc = await fetchImpl(`https://plc.directory/${did}`);
if (!doc.ok) throw new Error(`Could not resolve ${did}.`);
const services = (await doc.json()).service || [];
const pds = services.find(
(/** @type {{id?: string, type?: string}} */ s) =>
s.id === '#atproto_pds' || s.type === 'AtprotoPersonalDataServer',
)?.serviceEndpoint;
if (!pds) throw new Error(`${did} names no PDS in its document.`);
return { did, pds: pds.replace(/\/$/, ''), rkey: match[2] };
}
/**
* Build the installer.
* @param {SiteInstallerContext} ctx
* @returns {import('./ports.js').SiteInstallerPort}
*/
export function createSiteInstaller(ctx) {
const { actorStorage, blobs, getDid, createCommit, linkBlob, encodeRecord } =
ctx;
for (const [name, value] of Object.entries({
actorStorage,
blobs,
getDid,
createCommit,
linkBlob,
encodeRecord,
})) {
if (!value) throw new Error(`createSiteInstaller requires ${name}`);
}
const fetchImpl = ctx.fetchImpl || fetch;
/** @param {string} pds @param {string} path */
const remote = async (pds, path) => {
const response = await fetchImpl(`${pds}/xrpc/${path}`);
if (!response.ok) {
throw new Error(
`The developer's PDS answered ${response.status} for ${path.split('?')[0]}.`,
);
}
return response;
};
/** @param {string} uri */
async function fetchManifestAndSite(uri) {
const dev = await resolveManifestUri(uri, fetchImpl);
const manifest = await (
await remote(
dev.pds,
`com.atproto.repo.getRecord?repo=${encodeURIComponent(dev.did)}&collection=dev.pdsjs.app.manifest&rkey=${encodeURIComponent(dev.rkey)}`,
)
).json();
const siteUri = manifest.value?.site?.uri;
if (typeof siteUri !== 'string') {
throw new Error('The manifest names no site.');
}
const siteRkey = siteUri.split('/').pop() || '';
const site = await (
await remote(
dev.pds,
`com.atproto.repo.getRecord?repo=${encodeURIComponent(dev.did)}&collection=${SITE_COLLECTION}&rkey=${encodeURIComponent(siteRkey)}`,
)
).json();
return { dev, manifest, site };
}
/**
* The current install record pinning this manifest, if one exists.
* @param {string} _did @param {string} manifestUri
*/
async function existingInstall(_did, manifestUri) {
const page = await actorStorage.listRecords(INSTALL_COLLECTION, null, 100);
for (const record of page.records) {
const value =
/** @type {{manifest?: {uri?: string}, config?: Record<string, unknown>, site?: string}} */ (
toDisplayValue(
cborDecode(
record.value instanceof Uint8Array
? record.value
: new Uint8Array(record.value),
),
)
);
if (value.manifest?.uri === manifestUri) {
return {
uri: record.uri,
config: value.config,
site: typeof value.site === 'string' ? value.site : undefined,
};
}
}
return null;
}
return {
async preview(uri) {
const { dev, manifest, site } = await fetchManifestAndSite(uri);
const files = Array.isArray(site.value?.files) ? site.value.files : [];
let totalSize = 0;
for (const file of files) {
if (typeof file?.blob?.size === 'number') totalSize += file.blob.size;
}
let handle = dev.did;
try {
const described = await (
await remote(
dev.pds,
`com.atproto.repo.describeRepo?repo=${encodeURIComponent(dev.did)}`,
)
).json();
if (typeof described.handle === 'string') handle = described.handle;
} catch {}
const iconCid = manifest.value?.icon?.ref?.$link;
return {
uri: manifest.uri,
cid: manifest.cid,
name:
typeof manifest.value?.name === 'string' ? manifest.value.name : '',
description:
typeof manifest.value?.description === 'string'
? manifest.value.description
: '',
author: { did: dev.did, handle },
external: manifest.value?.external === true,
scopes: Array.isArray(manifest.value?.scopes)
? manifest.value.scopes
: [],
settings: settingsSchema(manifest.value?.settings),
queries: Array.isArray(manifest.value?.queries)
? manifest.value.queries.length
: 0,
files: { count: files.length, totalSize },
defaultSite:
typeof manifest.value?.suggestedSite === 'string' &&
isValidSiteName(manifest.value.suggestedSite)
? manifest.value.suggestedSite
: site.uri.split('/').pop() || '',
iconUrl: iconCid
? `${dev.pds}/xrpc/com.atproto.sync.getBlob?did=${encodeURIComponent(dev.did)}&cid=${iconCid}`
: null,
pinDrifted: manifest.value?.site?.cid !== site.cid,
};
},
async install(uri, options = {}) {
const did = await getDid();
if (!did) throw new Error('This PDS is not initialized.');
const { dev, manifest, site } = await fetchManifestAndSite(uri);
if (
Array.isArray(manifest.value?.scopes) &&
manifest.value.scopes.length
) {
throw new Error(
'This app requests scopes, which installs cannot grant yet.',
);
}
const existing = await existingInstall(did, manifest.uri);
// An update stays where the user put it: the existing install's name
// outranks the manifest's suggestion.
const plan = planInstall({
manifest,
site,
localName: options.site || existing?.site,
overrides: options.config,
existingConfig: existing?.config,
});
// Copy every blob first, hashing each one here: the site record only
// lands once all the bytes it names verifiably exist.
const toCopy = [...plan.copies];
if (plan.icon) {
toCopy.push({
path: 'icon',
cid: plan.icon.cid,
contentType: plan.icon.contentType,
});
}
for (const copy of toCopy) {
const bytes = new Uint8Array(
await (
await remote(
dev.pds,
`com.atproto.sync.getBlob?did=${encodeURIComponent(dev.did)}&cid=${copy.cid}`,
)
).arrayBuffer(),
);
if (bytes.length > BLOB_LIMIT) {
throw new Error(`${copy.path} is over the ${BLOB_LIMIT} byte limit.`);
}
const hashed = cidToString(await createBlobCid(bytes));
if (hashed !== copy.cid) {
throw new Error(
`${copy.path} hashed to ${hashed}, not the ${copy.cid} the record names. Not installing.`,
);
}
await actorStorage.putBlob(copy.cid, copy.contentType, bytes.length);
await blobs.put(did, copy.cid, bytes, copy.contentType);
}
/** @type {Array<{action: string, uri: string, cid: string}>} */
const ops = [];
/** @param {string} collection @param {string} rkey @param {Record<string, unknown>} value @param {string[]} blobCids */
const write = async (collection, rkey, value, blobCids) => {
const recordUri = `at://${did}/${collection}/${rkey}`;
const { cid, bytes } = await encodeRecord(value);
const had = await actorStorage.getRecord(recordUri);
if (blobCids.length) {
await actorStorage.unlinkBlobsFromRecord(recordUri);
}
await actorStorage.putBlock(cid, bytes);
await actorStorage.putRecord(recordUri, cid, collection, rkey, bytes);
for (const blobCid of blobCids) {
await linkBlob(blobCid, recordUri, Date.now());
}
ops.push({ action: had ? 'update' : 'create', uri: recordUri, cid });
};
await write(
SITE_COLLECTION,
plan.localName,
plan.siteRecord,
plan.copies.map((copy) => copy.cid),
);
for (const def of plan.queryDefs) {
await write(QUERY_COLLECTION, def.rkey, def.record, []);
}
const installRkey = existing
? existing.uri.split('/').pop() || createTid()
: createTid();
await write(
INSTALL_COLLECTION,
installRkey,
plan.installRecord,
plan.icon ? [plan.icon.cid] : [],
);
await createCommit(did, ops);
return { site: plan.localName, name: manifest.value?.name || '' };
},
async rename(installUri, newName) {
const did = await getDid();
if (!did) throw new Error('This PDS is not initialized.');
if (!isValidSiteName(newName)) {
throw new Error(
`"${newName}" cannot name a site: names are lowercase DNS labels.`,
);
}
const stored = await actorStorage.getRecord(installUri);
if (!stored) throw new Error('No such install.');
const installValue = /** @type {Record<string, unknown>} */ (
toDisplayValue(
cborDecode(
stored.value instanceof Uint8Array
? stored.value
: new Uint8Array(stored.value),
),
)
);
const oldName = installValue.site;
if (typeof oldName !== 'string') {
throw new Error('This install names no site.');
}
if (oldName === newName) return { site: newName };
const oldUri = `at://${did}/${SITE_COLLECTION}/${oldName}`;
const newUri = `at://${did}/${SITE_COLLECTION}/${newName}`;
if (await actorStorage.getRecord(newUri)) {
throw new Error(`A site named "${newName}" already exists.`);
}
const siteStored = await actorStorage.getRecord(oldUri);
if (!siteStored) throw new Error('The installed site record is gone.');
const siteBytes =
siteStored.value instanceof Uint8Array
? siteStored.value
: new Uint8Array(siteStored.value);
// The record moves byte for byte: same CID under the new key, so the
// manifest pin and the update check are untouched by a rename.
const siteValue = cborDecode(siteBytes);
await actorStorage.putBlock(siteStored.cid, siteBytes);
await actorStorage.putRecord(
newUri,
siteStored.cid,
SITE_COLLECTION,
newName,
siteBytes,
);
for (const ref of findBlobRefs(siteValue)) {
await linkBlob(ref, newUri, Date.now());
}
await actorStorage.unlinkBlobsFromRecord(oldUri);
await actorStorage.deleteRecord(oldUri);
const updatedInstall = { ...installValue, site: newName };
const { cid, bytes } = await encodeRecord(updatedInstall);
await actorStorage.putBlock(cid, bytes);
await actorStorage.putRecord(
installUri,
cid,
INSTALL_COLLECTION,
installUri.split('/').pop() || '',
bytes,
);
await createCommit(did, [
{ action: 'create', uri: newUri, cid: siteStored.cid },
{ action: 'delete', uri: oldUri },
{ action: 'update', uri: installUri, cid },
]);
return { site: newName };
},
async checkUpdate(uri, pinnedCid) {
const dev = await resolveManifestUri(uri, fetchImpl);
const manifest = await (
await remote(
dev.pds,
`com.atproto.repo.getRecord?repo=${encodeURIComponent(dev.did)}&collection=dev.pdsjs.app.manifest&rkey=${encodeURIComponent(dev.rkey)}`,
)
).json();
return {
currentCid: manifest.cid,
changed: manifest.cid !== pinnedCid,
};
},
};
}