import {
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 write
const MAX_CONCURRENT_DOWNLOADS = 3
export const MAX_PDS_RECORD_RESPONSE_BYTES = 1024 * 1024
export const PDS_RECORD_TIMEOUT_MS = 10_000
export const PDS_BLOB_TIMEOUT_MS = 300_000
const WINDOWS_RESERVED_NAME = /^(con|prn|aux|nul|com[1-9]|lpt[1-9])(?:\..*)?$/i
const 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
}
}