Something went wrong. Try again.
Monorepo for wisp.place. A static site hosting service built on top of the AT Protocol.
Something went wrong. Try again.
18 kB · 551 lines
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552import { copyFileSync, existsSync, lstatSync, mkdirSync, mkdtempSync, realpathSync, renameSync, rmSync, statSync, writeFileSync,} from 'node:fs'import { dirname, isAbsolute, join, relative, resolve, sep } from 'node:path'import { promisify } from 'node:util'import { gunzip } from 'node:zlib'import { expandSubfs, extractBlobCid, getPdsForDid, readBoundedIdentityJson, resolveDid, type SubfsSubject, unsafeRawIdentityGet,} from '@wispplace/atproto-utils'import { MAX_BLOB_SIZE, MAX_FILE_COUNT, MAX_SITE_SIZE_SUPPORTER } from '@wispplace/constants'import { normalizeSitePath } from '@wispplace/fs-utils'import { parseLexiconJson } from '@wispplace/lexicons/public-json'import type { Entry, Record as FsRecord } from '@wispplace/lexicons/types/place/wisp/fs'import { validateRecord as validateFsRecord } from '@wispplace/lexicons/types/place/wisp/fs'import { loadMetadata, type SiteMetadata, saveMetadata } from '../lib/metadata.ts'import { createSpinner, pc } from '../lib/progress.ts'
// full blobs remain resident through base64/gzip decoding and the atomic writeconst MAX_CONCURRENT_DOWNLOADS = 3export const MAX_PDS_RECORD_RESPONSE_BYTES = 1024 * 1024export const PDS_RECORD_TIMEOUT_MS = 10_000export const PDS_BLOB_TIMEOUT_MS = 300_000const WINDOWS_RESERVED_NAME = /^(con|prn|aux|nul|com[1-9]|lpt[1-9])(?:\..*)?$/iconst gunzipAsync = promisify(gunzip)
/** Decompress one pulled site blob without accepting unbounded gzip output. */export async function decompressPulledGzip(content: Uint8Array, maxOutputBytes = MAX_BLOB_SIZE): Promise<Buffer> { if (!Number.isSafeInteger(maxOutputBytes) || maxOutputBytes < 0 || maxOutputBytes > MAX_BLOB_SIZE) { throw new RangeError('Pulled gzip output limit must be within the blob limit') } try { return await gunzipAsync(content, { maxOutputLength: maxOutputBytes }) } catch { throw new Error('Could not safely decompress gzip blob') }}
/** Buffer a remote site blob only until the shared per-file byte limit. */export async function readPulledBlob(response: Response, maxBytes = MAX_BLOB_SIZE): Promise<Buffer> { if (!Number.isSafeInteger(maxBytes) || maxBytes < 0 || maxBytes > MAX_BLOB_SIZE) { throw new RangeError('Pulled blob limit must be within the blob limit') } const reader = response.body?.getReader() if (!reader) return Buffer.alloc(0)
const chunks: Uint8Array[] = [] let totalBytes = 0 try { while (true) { const { done, value } = await reader.read() if (done) break totalBytes += value.byteLength if (totalBytes > maxBytes) throw new Error(`Downloaded blob exceeds the ${maxBytes}-byte limit`) chunks.push(value) } } catch (error) { await reader.cancel(error).catch(() => undefined) throw error } finally { reader.releaseLock() } return Buffer.concat(chunks, totalBytes)}
export function validatePulledGzipHeader(content: Uint8Array): void { if (content.length < 2 || content[0] !== 0x1f || content[1] !== 0x8b) { throw new Error('Blob is marked gzip but is not a gzip stream') }}
export function createLogicalSiteBudget(maxBytes = MAX_SITE_SIZE_SUPPORTER) { if (!Number.isSafeInteger(maxBytes) || maxBytes < 0 || maxBytes > MAX_SITE_SIZE_SUPPORTER) { throw new RangeError('Logical site size limit is invalid') } let totalSize = 0 return { reserve(bytes: number) { if (!Number.isSafeInteger(bytes) || bytes < 0 || totalSize > maxBytes - bytes) { throw new Error(`Pulled site exceeds the ${maxBytes}-byte logical size limit`) } totalSize += bytes }, get totalSize() { return totalSize }, }}
function requireSiteFilePath(path: string): string { const normalized = normalizeSitePath(path) if ( !normalized || normalized !== path || normalized .split('/') .some((segment) => segment.includes(':') || /[. ]$/.test(segment) || WINDOWS_RESERVED_NAME.test(segment)) ) { throw new Error('Site contains an invalid file path') } return normalized}
function isWithinRoot(root: string, target: string): boolean { const fromRoot = relative(root, target) return fromRoot !== '' && fromRoot !== '..' && !fromRoot.startsWith(`..${sep}`) && !isAbsolute(fromRoot)}
function lstatOrNull(path: string): ReturnType<typeof lstatSync> | null { try { return lstatSync(path) } catch { return null }}
/** Resolve a validated site path without following untrusted child symlinks. */export function resolvePullFilePath(root: string, sitePath: string, createParents = false): string { const relativePath = requireSiteFilePath(sitePath) const rootInfo = lstatSync(root) if (rootInfo.isSymbolicLink() || !rootInfo.isDirectory()) { throw new Error('Pull root must be a directory, not a symbolic link') }
const rootPath = realpathSync(root) const parts = relativePath.split('/') let directory = rootPath for (const segment of parts.slice(0, -1)) { directory = join(directory, segment) let info = lstatOrNull(directory) if (!info) { if (!createParents) continue mkdirSync(directory) info = lstatOrNull(directory) } if (!info || info.isSymbolicLink() || !info.isDirectory()) { throw new Error('Pull path contains an unsafe directory') } }
const target = resolve(rootPath, ...parts) if (!isWithinRoot(rootPath, target)) { throw new Error('Resolved pull path escapes its root') } if (lstatOrNull(target)?.isSymbolicLink()) { throw new Error('Pull path resolves to a symbolic link') } return target}
export interface PullOptions { site: string path: string}
const SUBFS_EXPANSION_LIMITS = { maxConcurrentFetches: 4, maxDepth: 10, maxEntries: MAX_FILE_COUNT * 4, maxFiles: MAX_FILE_COUNT, maxRecords: 100,} as const
interface GetRecordResponse { value: unknown cid?: string}
function isGetRecordResponse(value: unknown): value is GetRecordResponse { if (typeof value !== 'object' || value === null || !('value' in value)) return false return !('cid' in value) || value.cid === undefined || typeof value.cid === 'string'}
async function fetchRecord( pdsEndpoint: string, did: string, collection: string, rkey: string,): Promise<GetRecordResponse> { const query = new URLSearchParams({ repo: did, collection, rkey }) const controller = new AbortController() const timeout = setTimeout(() => controller.abort(), PDS_RECORD_TIMEOUT_MS) let res: Response try { res = await fetch(`${pdsEndpoint}/xrpc/com.atproto.repo.getRecord?${query.toString()}`, { signal: controller.signal, }) if (!res.ok) { await res.body?.cancel().catch(() => undefined) throw new Error(`Failed to fetch record: ${res.status}`) } const data: unknown = await readBoundedIdentityJson(res, MAX_PDS_RECORD_RESPONSE_BYTES, controller.signal) if (!isGetRecordResponse(data)) throw new Error('PDS returned an invalid record response') return data } finally { clearTimeout(timeout) }}
type PdsEndpointResolver = (did: string) => Promise<string>
function createPdsEndpointResolver(rootDid: string, rootPdsEndpoint: string): PdsEndpointResolver { const endpoints = new Map<string, Promise<string>>([[rootDid, Promise.resolve(rootPdsEndpoint)]])
return (sourceDid) => { const existing = endpoints.get(sourceDid) if (existing) return existing
const pending = (async () => { const endpoint = await getPdsForDid(sourceDid, unsafeRawIdentityGet, { allowLoopback: true }) if (!endpoint) throw new Error('Could not resolve a source PDS endpoint') return endpoint })() endpoints.set(sourceDid, pending) return pending }}
async function fetchSubfsRecord(subject: SubfsSubject, resolvePdsEndpoint: PdsEndpointResolver): Promise<unknown> { const pdsEndpoint = await resolvePdsEndpoint(subject.repo) const response = await fetchRecord(pdsEndpoint, subject.repo, subject.collection, subject.rkey) return response.value}
interface FileToDownload { path: string cid: string ownerDid: string encoding?: 'gzip' mimeType?: string base64?: boolean}
function collectFiles( entries: Entry[], ownerDidByFilePath: ReadonlyMap<string, string>, pathPrefix: string, existingCids: Record<string, string>,): { toDownload: FileToDownload[]; unchangedFiles: FileToDownload[] } { const toDownload: FileToDownload[] = [] const seenPaths = new Set<string>() const unchangedFiles: FileToDownload[] = []
function entryPath(parent: string, name: string): string { const segment = requireSiteFilePath(name) if (segment.includes('/')) throw new Error('Site entry names must be single path segments') const path = parent ? `${parent}/${segment}` : segment if (seenPaths.has(path)) throw new Error('Site contains duplicate file paths') seenPaths.add(path) return path }
function collect(entries: Entry[], currentPath: string) { for (const entry of entries) { const fullPath = entryPath(currentPath, entry.name) const node = entry.node
if ('type' in node && node.type === 'directory' && 'entries' in node) { collect(node.entries, fullPath) } else if ('type' in node && node.type === 'file' && 'blob' in node) { const cid = extractBlobCid(node.blob) if (!cid) continue
const ownerDid = ownerDidByFilePath.get(fullPath) if (!ownerDid) throw new Error('Expanded file is missing its source repository') const file = { path: fullPath, cid, ownerDid, encoding: node.encoding, mimeType: node.mimeType, base64: node.base64, } satisfies FileToDownload if (existingCids[fullPath] === cid) unchangedFiles.push(file) else toDownload.push(file) } } }
collect(entries, pathPrefix) return { toDownload, unchangedFiles }}
function withDeadline( parent: AbortSignal | undefined, timeoutMs: number,): { signal: AbortSignal; dispose: () => void } { const controller = new AbortController() const timeout = setTimeout(() => controller.abort(), timeoutMs) const abort = () => controller.abort(parent?.reason) if (parent?.aborted) controller.abort(parent.reason) parent?.addEventListener('abort', abort, { once: true }) return { signal: controller.signal, dispose: () => { clearTimeout(timeout) parent?.removeEventListener('abort', abort) }, }}
async function downloadBlob(pdsEndpoint: string, file: FileToDownload, signal?: AbortSignal): Promise<Buffer> { const url = `${pdsEndpoint}/xrpc/com.atproto.sync.getBlob?did=${encodeURIComponent(file.ownerDid)}&cid=${encodeURIComponent(file.cid)}` const deadline = withDeadline(signal, PDS_BLOB_TIMEOUT_MS) try { const res = await fetch(url, { signal: deadline.signal })
if (!res.ok) { await res.body?.cancel().catch(() => undefined) throw new Error(`Failed to download blob ${file.cid}: ${res.status}`) }
let content: Buffer = await readPulledBlob(res)
// Decode base64 if needed if (file.base64) { const base64String = content.toString('utf-8') content = Buffer.from(base64String, 'base64') }
// The CLI writes decoded site files, so malformed or oversized gzip must // fail the pull rather than becoming an opaque canonical-looking output. if (file.encoding === 'gzip') { validatePulledGzipHeader(content) content = await decompressPulledGzip(content) }
return content } finally { deadline.dispose() }}
export async function pull(identifier: string, options: PullOptions): Promise<void> { const { site, path: outputPath } = options
console.log(pc.cyan(`\nPulling ${pc.bold(site)} from ${identifier}\n`))
// 1. Resolve DID const spinner = createSpinner('Resolving identity...').start() const did = await resolveDid(identifier, unsafeRawIdentityGet)
if (!did) { spinner.fail('Failed to resolve identity') throw new Error(`Could not resolve: ${identifier}`) }
spinner.succeed(`Resolved to ${did}`)
// 2. Get PDS endpoint const pdsSpinner = createSpinner('Getting PDS endpoint...').start() const pdsEndpoint = await getPdsForDid(did, unsafeRawIdentityGet, { allowLoopback: true })
if (!pdsEndpoint) { pdsSpinner.fail('Failed to get PDS endpoint') throw new Error(`Could not get PDS for: ${did}`) }
pdsSpinner.succeed('Got PDS endpoint') const resolveSourcePdsEndpoint = createPdsEndpointResolver(did, pdsEndpoint)
// 3. Fetch and validate the site record const recordSpinner = createSpinner('Fetching site record...').start() let recordData: GetRecordResponse try { recordData = await fetchRecord(pdsEndpoint, did, 'place.wisp.fs', site) } catch { recordSpinner.fail('Site not found') throw new Error(`Site not found: ${site}`) }
let record: FsRecord try { record = parseLexiconJson<FsRecord>(recordData.value) if (!validateFsRecord(record).success) throw new Error('Invalid site record') } catch { recordSpinner.fail('Site record is invalid') throw new Error('Site record is invalid') } const recordCid = recordData.cid || '' recordSpinner.succeed('Fetched site record')
// 4. Expand SubFS nodes. The shared helper validates children, preserves // owner DID provenance, and fails closed on partial expansion. const expandSpinner = createSpinner('Expanding subfs nodes...').start() let expandedRoot: FsRecord['root'] let ownerDidByFilePath: ReadonlyMap<string, string> try { const expanded = await expandSubfs(record.root, { rootOwnerDid: did, fetchSubfsRecord: (subject) => fetchSubfsRecord(subject, resolveSourcePdsEndpoint), limits: SUBFS_EXPANSION_LIMITS, }) expandedRoot = expanded.root ownerDidByFilePath = expanded.ownerDidByFilePath } catch { expandSpinner.fail('Could not expand SubFS nodes') throw new Error('Could not expand SubFS nodes') } expandSpinner.succeed('Expanded SubFS nodes')
// 5. Load existing metadata for incremental updates. const existingMetadata = loadMetadata(outputPath) const existingCids = existingMetadata?.fileCids || {} for (const filePath of Object.keys(existingCids)) { requireSiteFilePath(filePath) } if (existsSync(outputPath)) { const outputInfo = lstatSync(outputPath) if (outputInfo.isSymbolicLink() || !outputInfo.isDirectory()) { throw new Error('Output path must be a directory, not a symbolic link') } }
// 6. Collect files to download. const collected = collectFiles(expandedRoot.entries, ownerDidByFilePath, '', existingCids) const unchangedFiles = collected.unchangedFiles.filter((file) => { try { return statSync(resolvePullFilePath(outputPath, file.path)).isFile() } catch { return false } }) const toDownload = [ ...collected.toDownload, ...collected.unchangedFiles.filter((file) => !unchangedFiles.includes(file)), ] const unchangedPaths = unchangedFiles.map((file) => file.path) const toSkip = unchangedFiles.length
console.log(pc.dim(`Files to download: ${toDownload.length}, unchanged: ${toSkip}`))
const logicalBudget = createLogicalSiteBudget() if (toDownload.length === 0 && toSkip > 0) { for (const filePath of unchangedPaths) { const srcPath = resolvePullFilePath(outputPath, filePath) if (existsSync(srcPath)) logicalBudget.reserve(statSync(srcPath).size) } console.log(pc.green('\n✓ Site is already up to date\n')) return }
// 7. Create a non-symlink temp directory. mkdirSync(dirname(outputPath), { recursive: true }) const tempDir = mkdtempSync(`${outputPath}.tmp-`) const tempInfo = lstatSync(tempDir) if (tempInfo.isSymbolicLink() || !tempInfo.isDirectory()) { throw new Error('Temporary pull path must be a directory, not a symbolic link') }
// 8. Download files. const downloadSpinner = createSpinner(`Downloading ${toDownload.length} files...`).start() const newFileCids: Record<string, string> = Object.fromEntries( unchangedFiles.map((file) => [file.path, file.cid]), ) as Record<string, string> let downloaded = 0
try { for (let i = 0; i < toDownload.length; i += MAX_CONCURRENT_DOWNLOADS) { const batch = toDownload.slice(i, i + MAX_CONCURRENT_DOWNLOADS)
const batchController = new AbortController() const results = await Promise.allSettled( batch.map(async (file) => { try { const sourcePdsEndpoint = await resolveSourcePdsEndpoint(file.ownerDid) const content = await downloadBlob(sourcePdsEndpoint, file, batchController.signal) logicalBudget.reserve(content.byteLength) const filePath = resolvePullFilePath(tempDir, file.path, true) writeFileSync(filePath, content)
newFileCids[file.path] = file.cid downloaded++ downloadSpinner.text = `Downloading files: ${downloaded}/${toDownload.length}` } catch (error) { batchController.abort(error) throw error } }), ) const failure = results.find((result): result is PromiseRejectedResult => result.status === 'rejected') if (failure) throw failure.reason }
downloadSpinner.succeed(`Downloaded ${downloaded} files`)
// 9. Copy unchanged files from existing directory if (toSkip > 0 && existsSync(outputPath)) { const copySpinner = createSpinner(`Copying ${toSkip} unchanged files...`).start()
for (const filePath of unchangedPaths) { const srcPath = resolvePullFilePath(outputPath, filePath) const destPath = resolvePullFilePath(tempDir, filePath, true)
if (existsSync(srcPath)) { logicalBudget.reserve(statSync(srcPath).size) copyFileSync(srcPath, destPath) } }
copySpinner.succeed(`Copied ${toSkip} unchanged files`) }
// 10. Atomic replace if (existsSync(outputPath)) { const backupPath = `${outputPath}.backup-${Date.now()}` renameSync(outputPath, backupPath) try { renameSync(tempDir, outputPath) } catch (error) { try { renameSync(backupPath, outputPath) } catch { // preserve the original replacement failure; the backup remains recoverable } throw error } rmSync(backupPath, { recursive: true, force: true }) } else { renameSync(tempDir, outputPath) }
// 11. Save metadata const metadata: SiteMetadata = { recordCid, fileCids: newFileCids, lastSync: Date.now(), } saveMetadata(outputPath, metadata)
console.log(pc.green(`\n✓ Pulled ${site} to ${outputPath}\n`)) } catch (err) { // Cleanup temp dir on error rmSync(tempDir, { recursive: true, force: true }) throw err }}