import { gzip } from 'node:zlib'
import type { Agent } from '@atproto/api'
import { TID } from '@atproto/common-web'
import {
	computeCID,
	extractBlobMap,
	extractSubfsUris,
	isTextMimeType,
	parseSubfsSubject,
	shouldCompressFile,
} from '@wispplace/atproto-utils'
import { GZIP_COMPRESSION_LEVEL, MAX_FILE_COUNT, MAX_FILE_SIZE } from '@wispplace/constants'
import {
	countFilesInDirectory,
	createManifest,
	estimateDirectorySize,
	type FileUploadResult,
	findLargeDirectories,
	normalizeSitePath,
	processUploadedFiles,
	replaceDirectoryWithSubfs,
	splitDirectoryIntoChunks,
	toSubfsDirectory,
	type UploadedFile,
	updateFileBlobs,
} from '@wispplace/fs-utils'
import type { Directory, Entry } from '@wispplace/lexicons/types/place/wisp/fs'
import { createLogger } from '@wispplace/observability'
import { createIgnoreMatcher, parseWispignore, shouldIgnore } from './ignore-patterns'
import { completeUploadJob, failUploadJob, getUploadJob, updateJobProgress } from './upload-jobs'

const logger = createLogger('main-app')

const MAX_SKIPPED_FILES_REPORTED = 50
const MAX_WISPIGNORE_BYTES = 64 * 1024
const MAX_WISPIGNORE_PATTERNS = 1_000
const MAX_BUFFERED_GZIP_INPUT_BYTES = 16 * 1024 * 1024
const MAX_OWNED_SUBFS_RECORDS = 100
const SUBFS_CONCURRENCY = 4
const MAX_PROGRESS_FILE_NAME_LENGTH = 160
const MAX_MANIFEST_SIZE = 140 * 1024
const MAX_SUBFS_SIZE = 75 * 1024
const SUBFS_CHUNK_TARGET_SIZE = MAX_SUBFS_SIZE - 2 * 1024
const INLINE_FILE_COUNT_TARGET = 200
const ROOT_FILE_BATCH_SIZE = 100
const MAX_MANIFEST_SPLIT_ATTEMPTS = 100
const RESERVED_SEGMENTS = new Set(['__proto__', 'constructor', 'prototype'])

export const INVALID_UPLOAD_MESSAGE = 'Invalid upload'
export const UPLOAD_TOO_LARGE_MESSAGE = 'Upload too large'
export const UPLOAD_BUSY_MESSAGE = 'Upload capacity reached'
export const UPLOAD_UNAVAILABLE_MESSAGE = 'Uploads are temporarily unavailable'
export const UPLOAD_CONFLICT_MESSAGE = 'Upload conflict. Please retry'
export const UPLOAD_FAILED_MESSAGE = 'Upload failed'

export class PublicUploadError extends Error {
	constructor(
		public readonly status: 400 | 409 | 413 | 429 | 500 | 503,
		message: string,
	) {
		super(message)
		this.name = 'PublicUploadError'
	}
}

export interface ValidatedPublicUploadFile {
	file: File
	path: string
	mimeType: string
	size: number
}

interface RawPublicUploadFile extends ValidatedPublicUploadFile {
	rawPath: string
	fromWebkitRelativePath: boolean
}

interface OwnedSubfsSubject {
	uri: string
	rkey: string
	path: string
}

interface ExistingUploadState {
	rootCid: string | null
	blobMap: Map<string, { blobRef: any; cid: string }>
	ownedSubfs: OwnedSubfsSubject[]
}

interface UploadedBlob {
	result: FileUploadResult
	filePath: string
	reused: boolean
}

interface ManifestBuildState {
	directory: Directory
	totalFileCount: number
	inlineFileCount: number
}

interface ManifestCommit {
	record: Awaited<ReturnType<Agent['com']['atproto']['repo']['putRecord']>>
	referencedSubfs: ReadonlySet<string>
}

type IgnoreMatcher = ReturnType<typeof createIgnoreMatcher>
type PutSubfs = (record: Record<string, unknown>) => Promise<string>

function invalidUpload(): never {
	throw new PublicUploadError(400, INVALID_UPLOAD_MESSAGE)
}

function uploadTooLarge(): never {
	throw new PublicUploadError(413, UPLOAD_TOO_LARGE_MESSAGE)
}

function throwIfAborted(signal?: AbortSignal): void {
	if (signal?.aborted) throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
}

function throwIfUploadCancelled(jobId: string, signal?: AbortSignal): void {
	throwIfAborted(signal)
	const job = getUploadJob(jobId)
	if (!job || job.status === 'failed' || job.status === 'completed') {
		throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
	}
}

function safeFileLabel(path: string): string {
	if (path.length <= MAX_PROGRESS_FILE_NAME_LENGTH) return path
	return `${path.slice(0, 120)}…${path.slice(-24)}`
}

function hasAsciiControlCharacter(value: string): boolean {
	for (const character of value) {
		const codePoint = character.codePointAt(0) ?? 0
		if (codePoint < 32 || codePoint === 127) return true
	}
	return false
}

function getSafeMimeType(file: File): string {
	const mimeType = typeof file.type === 'string' ? file.type : ''
	if (mimeType.length === 0 || mimeType.length > 255 || hasAsciiControlCharacter(mimeType)) {
		return 'application/octet-stream'
	}
	return mimeType
}

function getUploadPathSource(file: File): { path: string; fromWebkitRelativePath: boolean } | null {
	const webkitRelativePath = (file as File & { webkitRelativePath?: unknown }).webkitRelativePath
	if (typeof webkitRelativePath === 'string' && webkitRelativePath.length > 0) {
		return { path: webkitRelativePath, fromWebkitRelativePath: true }
	}
	if (typeof file.name !== 'string') return null
	return { path: file.name, fromWebkitRelativePath: false }
}

function asFile(candidate: unknown): File | null {
	if (!candidate || typeof candidate !== 'object') return null
	const file = candidate as File
	if (typeof file.arrayBuffer !== 'function' || typeof file.name !== 'string' || typeof file.size !== 'number')
		return null
	return file
}

function validateRawUploadFile(candidate: unknown): RawPublicUploadFile {
	const file = asFile(candidate)
	if (!file) invalidUpload()
	const source = getUploadPathSource(file)
	if (!source) invalidUpload()
	const normalizedPath = normalizeSitePath(source.path)
	if (!normalizedPath || normalizedPath !== source.path) invalidUpload()
	if (!Number.isSafeInteger(file.size) || file.size < 0) invalidUpload()
	return {
		file,
		rawPath: normalizedPath,
		path: normalizedPath,
		fromWebkitRelativePath: source.fromWebkitRelativePath,
		mimeType: getSafeMimeType(file),
		size: file.size,
	}
}

function sharedWebkitDirectoryRoot(files: readonly RawPublicUploadFile[]): string | null {
	if (files.length === 0 || files.some((file) => !file.fromWebkitRelativePath || !file.rawPath.includes('/'))) {
		return null
	}
	const root = files[0]!.rawPath.slice(0, files[0]!.rawPath.indexOf('/'))
	return files.every((file) => file.rawPath.startsWith(`${root}/`)) ? root : null
}

function isReservedInternalSegment(segment: string): boolean {
	const normalized = segment.normalize('NFKC').toLowerCase()
	return normalized.startsWith('__subfs_') || RESERVED_SEGMENTS.has(normalized)
}

function hasReservedInternalPath(path: string): boolean {
	return path.split('/').some(isReservedInternalSegment)
}

function canonicalPath(rawFile: RawPublicUploadFile, sharedRoot: string | null): string {
	const path = sharedRoot ? rawFile.rawPath.slice(sharedRoot.length + 1) : rawFile.rawPath
	const normalizedPath = normalizeSitePath(path)
	if (!normalizedPath || normalizedPath !== path || hasReservedInternalPath(normalizedPath)) invalidUpload()
	return normalizedPath
}

function assertNoPathCollisions(paths: readonly string[]): void {
	const sortedPaths = [...paths].sort()
	for (let index = 1; index < sortedPaths.length; index++) {
		if (
			sortedPaths[index] === sortedPaths[index - 1] ||
			sortedPaths[index]!.startsWith(`${sortedPaths[index - 1]!}/`)
		) {
			invalidUpload()
		}
	}
}

/** Validate multipart metadata before any file body is read. */
export function validatePublicUploadFiles(
	files: readonly unknown[],
	maxLogicalBytes: number,
): ValidatedPublicUploadFile[] {
	if (!Number.isSafeInteger(maxLogicalBytes) || maxLogicalBytes < 0) invalidUpload()
	// Reject a hostile multipart count before touching even file metadata. Bun
	// has already parsed the multipart envelope, but no File body is read here.
	if (files.length > MAX_FILE_COUNT) uploadTooLarge()

	let totalSize = 0
	const rawFiles: RawPublicUploadFile[] = []
	for (const candidate of files) {
		const file = validateRawUploadFile(candidate)
		if (file.size > MAX_FILE_SIZE || file.size > maxLogicalBytes - totalSize) uploadTooLarge()
		totalSize += file.size
		rawFiles.push(file)
	}

	const sharedRoot = sharedWebkitDirectoryRoot(rawFiles)
	const validatedFiles = rawFiles.map((file) => ({ ...file, path: canonicalPath(file, sharedRoot) }))
	assertNoPathCollisions(validatedFiles.map((file) => file.path))
	return validatedFiles
}

function isWispignorePath(path: string): boolean {
	const parts = path.split('/')
	return parts[parts.length - 1] === '.wispignore'
}

function rootWispignoreFile(files: readonly ValidatedPublicUploadFile[]): ValidatedPublicUploadFile | undefined {
	return files.find((file) => file.path === '.wispignore')
}

async function readCustomIgnorePatterns(files: readonly ValidatedPublicUploadFile[]): Promise<string[]> {
	const ignoreFile = rootWispignoreFile(files)
	if (!ignoreFile) return []
	if (ignoreFile.size > MAX_WISPIGNORE_BYTES) invalidUpload()
	try {
		const patterns = parseWispignore(await ignoreFile.file.text())
		if (patterns.length > MAX_WISPIGNORE_PATTERNS) invalidUpload()
		return patterns
	} catch {
		invalidUpload()
	}
}

function createIgnoreMatchers(patterns: string[]): { enforced: IgnoreMatcher; requested: IgnoreMatcher } {
	try {
		return { enforced: createIgnoreMatcher(), requested: createIgnoreMatcher(patterns) }
	} catch {
		invalidUpload()
	}
}

function ignoredFileReason(
	file: ValidatedPublicUploadFile,
	enforced: IgnoreMatcher,
	requested: IgnoreMatcher,
): string | null {
	if (isWispignorePath(file.path)) return 'matched ignore pattern'
	if (shouldIgnore(enforced, file.path)) return 'matched ignore pattern'
	if (shouldIgnore(requested, file.path)) return 'matched ignore pattern'
	return null
}

function addSkippedFile(skippedFiles: Array<{ name: string; reason: string }>, path: string, reason: string): void {
	if (skippedFiles.length < MAX_SKIPPED_FILES_REPORTED) {
		skippedFiles.push({ name: safeFileLabel(path), reason })
	}
}

export async function selectPublicUploadFiles(files: readonly ValidatedPublicUploadFile[]): Promise<{
	files: ValidatedPublicUploadFile[]
	skippedFiles: Array<{ name: string; reason: string }>
}> {
	const patterns = await readCustomIgnorePatterns(files)
	const matchers = createIgnoreMatchers(patterns)
	const selectedFiles: ValidatedPublicUploadFile[] = []
	const skippedFiles: Array<{ name: string; reason: string }> = []

	for (const file of files) {
		const reason = ignoredFileReason(file, matchers.enforced, matchers.requested)
		if (reason) addSkippedFile(skippedFiles, file.path, reason)
		else selectedFiles.push(file)
	}
	return { files: selectedFiles, skippedFiles }
}

function gzipAsync(content: Buffer): Promise<Buffer> {
	return new Promise((resolve, reject) => {
		gzip(content, { level: GZIP_COMPRESSION_LEVEL }, (error, compressed) => {
			if (error) reject(error)
			else resolve(compressed)
		})
	})
}

function shouldGzip(file: ValidatedPublicUploadFile): boolean {
	// zlib needs the input and output live at once. Keep compression below this
	// bound rather than briefly doubling a 200 MiB upload buffer.
	return shouldCompressFile(file.mimeType, file.path) && file.size <= MAX_BUFFERED_GZIP_INPUT_BYTES
}

async function readFileForUpload(file: ValidatedPublicUploadFile, signal?: AbortSignal): Promise<UploadedFile> {
	throwIfAborted(signal)
	let content = Buffer.from(await file.file.arrayBuffer())
	throwIfAborted(signal)
	if (content.length !== file.size) throw new PublicUploadError(400, INVALID_UPLOAD_MESSAGE)
	if (!shouldGzip(file)) return createUploadedFile(file, content, false)

	const compressed = await gzipAsync(content)
	content = Buffer.alloc(0)
	throwIfAborted(signal)
	return createUploadedFile(file, compressed, true)
}

function createUploadedFile(file: ValidatedPublicUploadFile, content: Buffer, compressed: boolean): UploadedFile {
	return {
		name: file.path,
		content,
		mimeType: file.mimeType,
		size: content.length,
		compressed,
		originalMimeType: file.mimeType,
	}
}

function errorStatus(error: unknown, fallback = 502): number {
	const status = typeof error === 'object' && error !== null ? Number((error as { status?: unknown }).status) : NaN
	return Number.isInteger(status) && status >= 400 && status <= 599 ? status : fallback
}

function swapConflict(error: unknown): boolean {
	const code = typeof error === 'object' && error !== null ? String((error as { error?: unknown }).error ?? '') : ''
	return errorStatus(error) === 409 || code === 'InvalidSwap' || code === 'SwapConflict'
}

function asPipelineError(error: unknown, rootSwapAttempted = false): PublicUploadError {
	if (error instanceof PublicUploadError) return error
	if (rootSwapAttempted && swapConflict(error)) return new PublicUploadError(409, UPLOAD_CONFLICT_MESSAGE)
	const status = errorStatus(error)
	return status === 413 || status === 419
		? new PublicUploadError(413, UPLOAD_TOO_LARGE_MESSAGE)
		: new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
}

async function runBounded<T>(items: readonly T[], limit: number, operation: (item: T) => Promise<void>): Promise<void> {
	let nextIndex = 0
	const workerCount = Math.min(Math.max(1, limit), items.length)
	const workers = Array.from({ length: workerCount }, async () => {
		while (true) {
			const index = nextIndex++
			if (index >= items.length) return
			await operation(items[index]!)
		}
	})
	await Promise.all(workers)
}

export function collectOwnedSubfsSubjects(directory: Directory, did: string): OwnedSubfsSubject[] {
	const seen = new Set<string>()
	const subjects: OwnedSubfsSubject[] = []
	for (const candidate of extractSubfsUris(directory)) {
		if (subjects.length >= MAX_OWNED_SUBFS_RECORDS) break
		if (candidate.uri.length > 4096) continue
		const subject = parseOwnedSubfsSubject(candidate.uri, did)
		if (!subject || seen.has(subject.uri)) continue
		seen.add(subject.uri)
		subjects.push({ ...subject, path: candidate.path })
	}
	return subjects
}

function parseOwnedSubfsSubject(uri: string, did: string): Omit<OwnedSubfsSubject, 'path'> | null {
	try {
		const subject = parseSubfsSubject(uri)
		if (subject.repo !== did || subject.collection !== 'place.wisp.subfs') return null
		return { uri: subject.uri, rkey: subject.rkey }
	} catch {
		return null
	}
}

function emptyExistingState(rootCid: string | null): ExistingUploadState {
	return { rootCid, blobMap: new Map(), ownedSubfs: [] }
}

function manifestRoot(value: unknown): Directory | null {
	if (!value || typeof value !== 'object' || !('root' in value)) return null
	return (value as { root: Directory }).root
}

function isRecordNotFound(error: unknown): boolean {
	const code = typeof error === 'object' && error !== null ? String((error as { error?: unknown }).error ?? '') : ''
	return code === 'RecordNotFound' || errorStatus(error) === 404
}

function existingRecordCid(record: unknown): string {
	const cid =
		typeof record === 'object' && record !== null && 'data' in record
			? (record as { data?: { cid?: unknown } }).data?.cid
			: undefined
	if (typeof cid !== 'string' || cid.length === 0 || cid.length > 256) {
		throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
	}
	return cid
}

export async function loadExistingUploadState(
	agent: Agent,
	did: string,
	siteName: string,
): Promise<ExistingUploadState> {
	try {
		const record = await agent.com.atproto.repo.getRecord({ repo: did, collection: 'place.wisp.fs', rkey: siteName })
		return await existingStateFromRecord(agent, did, record.data.value, existingRecordCid(record))
	} catch (error) {
		if (isRecordNotFound(error)) return emptyExistingState(null)
		logger.warn('Unable to inspect existing manifest', { errorKind: 'existing_manifest_fetch_failed' })
		throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
	}
}

async function existingStateFromRecord(
	agent: Agent,
	did: string,
	value: unknown,
	rootCid: string,
): Promise<ExistingUploadState> {
	const root = manifestRoot(value)
	if (!root) return emptyExistingState(rootCid)
	const state: ExistingUploadState = {
		rootCid,
		blobMap: extractBlobMap(root),
		ownedSubfs: collectOwnedSubfsSubjects(root, did),
	}
	await mergeOwnedSubfsBlobMaps(agent, did, state)
	return state
}

async function mergeOwnedSubfsBlobMaps(agent: Agent, did: string, state: ExistingUploadState): Promise<void> {
	await runBounded(state.ownedSubfs, SUBFS_CONCURRENCY, async (subject) => {
		const root = await fetchOwnedSubfsRoot(agent, did, subject.rkey)
		if (!root) return
		extractBlobMap(root, subject.path).forEach((blob, path) => {
			state.blobMap.set(path, blob)
		})
	})
}

async function fetchOwnedSubfsRoot(agent: Agent, did: string, rkey: string): Promise<Directory | null> {
	try {
		const record = await agent.com.atproto.repo.getRecord({ repo: did, collection: 'place.wisp.subfs', rkey })
		return manifestRoot(record.data.value)
	} catch {
		logger.warn('Failed to fetch an owned SubFS record for blob reuse', { errorKind: 'subfs_reuse_fetch_failed' })
		return null
	}
}

function createMetadataDirectory(files: readonly ValidatedPublicUploadFile[]): {
	directory: Directory
	fileCount: number
} {
	const metadataFiles = files.map((file) => createUploadedFile(file, Buffer.alloc(0), false))
	return processUploadedFiles(metadataFiles, { skipNormalization: true })
}

function uploadMimeType(file: UploadedFile): string {
	return file.compressed || file.mimeType.startsWith('text/html') ? 'application/octet-stream' : file.mimeType
}

function reuseResult(file: UploadedFile, existing: { blobRef: any; cid: string }): FileUploadResult {
	return {
		hash: existing.cid,
		blobRef: existing.blobRef,
		...(file.compressed && { encoding: 'gzip' as const, mimeType: file.originalMimeType || file.mimeType }),
		base64: !!file.base64Encoded,
	}
}

function shouldRetryBlobUpload(status: number, attempt: number): boolean {
	return attempt < 4 && [408, 409, 429].includes(status)
}

function waitForBlobRetry(attempt: number, signal?: AbortSignal): Promise<void> {
	return new Promise((resolve, reject) => {
		const cleanup = () => signal?.removeEventListener('abort', abort)
		const timer = setTimeout(
			() => {
				cleanup()
				resolve()
			},
			Math.min(2_000 * 2 ** attempt, 16_000),
		)
		const abort = () => {
			clearTimeout(timer)
			cleanup()
			reject(new PublicUploadError(500, UPLOAD_FAILED_MESSAGE))
		}
		if (signal?.aborted) return abort()
		signal?.addEventListener('abort', abort, { once: true })
	})
}

async function uploadBlobWithRetry(
	agent: Agent,
	content: Buffer,
	mimeType: string,
	path: string,
	signal?: AbortSignal,
): Promise<FileUploadResult> {
	for (let attempt = 0; attempt < 5; attempt++) {
		throwIfAborted(signal)
		try {
			const response = await agent.com.atproto.repo.uploadBlob(content, { encoding: mimeType })
			return { hash: response.data.blob.ref.toString(), blobRef: response.data.blob, base64: false }
		} catch (error) {
			const status = errorStatus(error)
			if (status === 413 || status === 419) throw new PublicUploadError(413, UPLOAD_TOO_LARGE_MESSAGE)
			if (shouldRetryBlobUpload(status, attempt)) {
				await waitForBlobRetry(attempt, signal)
				continue
			}
			logger.warn('PDS blob upload failed', { errorKind: 'blob_upload_failed', status, path: safeFileLabel(path) })
			throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
		}
	}
	throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
}

async function uploadOneFile(
	jobId: string,
	input: ValidatedPublicUploadFile,
	index: number,
	total: number,
	existingBlobMap: ExistingUploadState['blobMap'],
	agent: Agent,
	signal?: AbortSignal,
): Promise<UploadedBlob> {
	throwIfUploadCancelled(jobId, signal)
	const prepared = await readFileForUpload(input, signal)
	throwIfUploadCancelled(jobId, signal)
	updateJobProgress(jobId, {
		filesProcessed: index + 1,
		currentFile: `${index + 1}/${total}: ${safeFileLabel(input.path)}`,
	})
	try {
		return await uploadPreparedFile(jobId, prepared, input.path, index, total, existingBlobMap, agent, signal)
	} finally {
		prepared.content = Buffer.alloc(0)
	}
}

async function uploadPreparedFile(
	jobId: string,
	file: UploadedFile,
	path: string,
	index: number,
	total: number,
	existingBlobMap: ExistingUploadState['blobMap'],
	agent: Agent,
	signal?: AbortSignal,
): Promise<UploadedBlob> {
	throwIfUploadCancelled(jobId, signal)
	const existing = existingBlobMap.get(path)
	if (existing?.cid === computeCID(file.content)) return markReusedFile(jobId, file, path, index, total, existing)
	return await putPreparedFile(jobId, file, path, index, total, agent, signal)
}

function markReusedFile(
	jobId: string,
	file: UploadedFile,
	path: string,
	index: number,
	total: number,
	existing: { blobRef: any; cid: string },
): UploadedBlob {
	const filesReused = (getUploadJob(jobId)?.progress.filesReused ?? 0) + 1
	updateJobProgress(jobId, { filesReused, currentFile: `${index + 1}/${total}: ${safeFileLabel(path)} (reused)` })
	return { result: reuseResult(file, existing), filePath: path, reused: true }
}

async function putPreparedFile(
	jobId: string,
	file: UploadedFile,
	path: string,
	index: number,
	total: number,
	agent: Agent,
	signal?: AbortSignal,
): Promise<UploadedBlob> {
	throwIfUploadCancelled(jobId, signal)
	const result = await uploadBlobWithRetry(agent, file.content, uploadMimeType(file), path, signal)
	throwIfUploadCancelled(jobId, signal)
	if (file.compressed) {
		result.encoding = 'gzip'
		result.mimeType = file.originalMimeType || file.mimeType
	}
	const filesUploaded = (getUploadJob(jobId)?.progress.filesUploaded ?? 0) + 1
	updateJobProgress(jobId, { filesUploaded, currentFile: `${index + 1}/${total}: ${safeFileLabel(path)} (uploaded)` })
	return { result, filePath: path, reused: false }
}

async function uploadAllFiles(
	jobId: string,
	files: readonly ValidatedPublicUploadFile[],
	existingState: ExistingUploadState,
	agent: Agent,
	signal?: AbortSignal,
): Promise<UploadedBlob[]> {
	const uploadedBlobs: UploadedBlob[] = []
	updateJobProgress(jobId, { phase: 'compressing', totalFiles: files.length })
	for (const [index, file] of files.entries()) {
		uploadedBlobs.push(await uploadOneFile(jobId, file, index, files.length, existingState.blobMap, agent, signal))
	}
	return uploadedBlobs
}

function updatedDirectoryForBlobs(
	directory: Directory,
	blobs: readonly UploadedBlob[],
	expectedFileCount: number,
): Directory {
	const results = blobs.map((blob) => blob.result)
	const paths = blobs.map((blob) => blob.filePath)
	const updatedDirectory = updateFileBlobs(directory, results, paths, '', undefined, { skipNormalization: true })
	if (countFilesInDirectory(updatedDirectory) !== expectedFileCount)
		throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
	return updatedDirectory
}

function manifestForState(siteName: string, state: ManifestBuildState) {
	return createManifest(siteName, state.directory, state.totalFileCount)
}

function manifestNeedsSplit(siteName: string, state: ManifestBuildState): boolean {
	return (
		state.inlineFileCount > INLINE_FILE_COUNT_TARGET ||
		JSON.stringify(manifestForState(siteName, state)).length > MAX_MANIFEST_SIZE
	)
}

function assertManifestFits(siteName: string, state: ManifestBuildState): void {
	if (manifestNeedsSplit(siteName, state)) throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
}

function largestDirectory(directory: Directory) {
	return findLargeDirectories(directory).sort((left, right) => right.size - left.size)[0] ?? null
}

function directoryContainsSubfs(directory: Directory): boolean {
	return directory.entries.some((entry) => nodeContainsSubfs(entry.node))
}

function nodeContainsSubfs(node: Entry['node']): boolean {
	if (!('type' in node)) return false
	if (node.type === 'subfs') return true
	return node.type === 'directory' && directoryContainsSubfs(node as Directory)
}

function makeFlatSubfsEntry(name: string, subject: string): Entry {
	return {
		name,
		node: { $type: 'place.wisp.fs#subfs', type: 'subfs', subject, flat: true },
	}
}

function replaceDirectoryEntries(directory: Directory, targetPath: string, entries: Entry[]): Directory {
	return replaceDirectoryEntriesAtPath(directory, targetPath.split('/'), entries)
}

function replaceDirectoryEntriesAtPath(directory: Directory, path: readonly string[], entries: Entry[]): Directory {
	if (path.length === 0) return { ...directory, entries }
	const [head, ...tail] = path
	return {
		...directory,
		entries: directory.entries.map((entry) => replaceMatchingDirectoryEntry(entry, head!, tail, entries)),
	}
}

function replaceMatchingDirectoryEntry(entry: Entry, head: string, tail: readonly string[], entries: Entry[]): Entry {
	if (entry.name !== head || !('type' in entry.node) || entry.node.type !== 'directory') return entry
	return {
		...entry,
		node: {
			...replaceDirectoryEntriesAtPath(entry.node as Directory, tail, entries),
			$type: 'place.wisp.fs#directory',
		},
	}
}

function assertChunksFit(chunks: readonly Directory[]): void {
	if (chunks.length < 2 || chunks.some((chunk) => estimateDirectorySize(chunk) > MAX_SUBFS_SIZE)) {
		throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
	}
}

async function splitLargeDirectory(
	state: ManifestBuildState,
	target: NonNullable<ReturnType<typeof largestDirectory>>,
	putSubfs: PutSubfs,
	splitId: number,
): Promise<ManifestBuildState> {
	if (directoryContainsSubfs(target.directory)) throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
	const chunks = splitDirectoryIntoChunks(target.directory, SUBFS_CHUNK_TARGET_SIZE)
	assertChunksFit(chunks)
	const entries: Entry[] = []
	for (const [index, chunk] of chunks.entries()) {
		const subject = await putSubfs({
			$type: 'place.wisp.subfs',
			root: toSubfsDirectory(chunk),
			fileCount: countFilesInDirectory(chunk),
			createdAt: new Date().toISOString(),
		})
		entries.push(makeFlatSubfsEntry(`__subfs_${splitId}_${index}`, subject))
	}
	return {
		...state,
		directory: replaceDirectoryEntries(state.directory, target.path, entries),
		inlineFileCount: state.inlineFileCount - target.fileCount,
	}
}

async function splitWholeDirectory(
	state: ManifestBuildState,
	target: NonNullable<ReturnType<typeof largestDirectory>>,
	putSubfs: PutSubfs,
): Promise<ManifestBuildState> {
	if (directoryContainsSubfs(target.directory)) throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
	const subject = await putSubfs({
		$type: 'place.wisp.subfs',
		root: toSubfsDirectory(target.directory),
		fileCount: target.fileCount,
		createdAt: new Date().toISOString(),
	})
	return {
		...state,
		directory: replaceDirectoryWithSubfs(state.directory, target.path, subject),
		inlineFileCount: state.inlineFileCount - target.fileCount,
	}
}

async function splitDirectoryState(
	state: ManifestBuildState,
	target: NonNullable<ReturnType<typeof largestDirectory>>,
	putSubfs: PutSubfs,
	splitId: number,
): Promise<ManifestBuildState> {
	return target.size > MAX_SUBFS_SIZE
		? await splitLargeDirectory(state, target, putSubfs, splitId)
		: await splitWholeDirectory(state, target, putSubfs)
}

function rootFileEntries(directory: Directory): Entry[] {
	return directory.entries.filter((entry) => 'type' in entry.node && entry.node.type === 'file')
}

async function splitRootFileState(
	state: ManifestBuildState,
	putSubfs: PutSubfs,
	splitId: number,
): Promise<ManifestBuildState> {
	const rootFiles = rootFileEntries(state.directory)
	if (rootFiles.length === 0) throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
	const batch = rootFiles.slice(0, ROOT_FILE_BATCH_SIZE)
	const subject = await putSubfs({
		$type: 'place.wisp.subfs',
		root: toSubfsDirectory({ $type: 'place.wisp.fs#directory', type: 'directory', entries: batch }),
		fileCount: batch.length,
		createdAt: new Date().toISOString(),
	})
	const remaining = state.directory.entries.filter((entry) => !batch.includes(entry))
	return {
		...state,
		directory: { ...state.directory, entries: [...remaining, makeFlatSubfsEntry(`__subfs_${splitId}_0`, subject)] },
		inlineFileCount: state.inlineFileCount - batch.length,
	}
}

async function splitManifestOnce(
	state: ManifestBuildState,
	putSubfs: PutSubfs,
	splitId: number,
): Promise<ManifestBuildState> {
	const target = largestDirectory(state.directory)
	return target
		? await splitDirectoryState(state, target, putSubfs, splitId)
		: await splitRootFileState(state, putSubfs, splitId)
}

async function planManifest(
	siteName: string,
	directory: Directory,
	totalFileCount: number,
	putSubfs: PutSubfs,
): Promise<ManifestBuildState> {
	let state: ManifestBuildState = { directory, totalFileCount, inlineFileCount: totalFileCount }
	for (let attempt = 0; attempt < MAX_MANIFEST_SPLIT_ATTEMPTS && manifestNeedsSplit(siteName, state); attempt++) {
		state = await splitManifestOnce(state, putSubfs, attempt + 1)
	}
	assertManifestFits(siteName, state)
	return state
}

async function generatedSubfsAreUnreferenced(
	agent: Agent,
	did: string,
	siteName: string,
	generatedUris: ReadonlySet<string>,
): Promise<boolean> {
	try {
		const record = await agent.com.atproto.repo.getRecord({ repo: did, collection: 'place.wisp.fs', rkey: siteName })
		const root = manifestRoot(record.data.value)
		if (!root) return false
		return !extractSubfsUris(root).some((reference) => generatedUris.has(reference.uri))
	} catch (error) {
		const code = typeof error === 'object' && error !== null ? String((error as { error?: unknown }).error ?? '') : ''
		return code === 'RecordNotFound'
	}
}

async function cleanupGeneratedSubfs(
	agent: Agent,
	did: string,
	siteName: string,
	rkeys: ReadonlySet<string>,
): Promise<void> {
	const generatedUris = new Set(Array.from(rkeys, (rkey) => `at://${did}/place.wisp.subfs/${rkey}`))
	if (!(await generatedSubfsAreUnreferenced(agent, did, siteName, generatedUris))) return
	await runBounded(Array.from(rkeys), SUBFS_CONCURRENCY, async (rkey) => {
		try {
			await agent.com.atproto.repo.deleteRecord({ repo: did, collection: 'place.wisp.subfs', rkey })
		} catch {
			logger.warn('Failed to clean up a newly-created SubFS record', { errorKind: 'subfs_cleanup_failed' })
		}
	})
}

export async function commitPublicUploadManifest(
	agent: Agent,
	did: string,
	siteName: string,
	directory: Directory,
	totalFileCount: number,
	jobId?: string,
	signal?: AbortSignal,
	expectedRootCid: string | null = null,
): Promise<ManifestCommit> {
	throwIfAborted(signal)
	const createdRkeys = new Set<string>()
	const referencedSubfs = new Set<string>()
	const putSubfs: PutSubfs = async (record) => {
		throwIfAborted(signal)
		const rkey = TID.nextStr()
		createdRkeys.add(rkey)
		await agent.com.atproto.repo.putRecord({ repo: did, collection: 'place.wisp.subfs', rkey, record })
		throwIfAborted(signal)
		const uri = `at://${did}/place.wisp.subfs/${rkey}`
		referencedSubfs.add(uri)
		return uri
	}

	let rootSwapAttempted = false
	try {
		if (jobId) updateJobProgress(jobId, { phase: 'creating_manifest' })
		const state = await planManifest(siteName, directory, totalFileCount, putSubfs)
		throwIfAborted(signal)
		if (jobId) updateJobProgress(jobId, { phase: 'finalizing' })
		rootSwapAttempted = true
		const record = await agent.com.atproto.repo.putRecord({
			repo: did,
			collection: 'place.wisp.fs',
			rkey: siteName,
			record: manifestForState(siteName, state),
			swapRecord: expectedRootCid,
		})
		return { record, referencedSubfs }
	} catch (error) {
		// Shutdown cancellation must not start extra PDS cleanup calls. Any
		// pre-root records are unreferenced and may be safely collected later.
		if (!signal?.aborted) await cleanupGeneratedSubfs(agent, did, siteName, createdRkeys)
		throw asPipelineError(error, rootSwapAttempted)
	}
}

function fallbackEligible(file: ValidatedPublicUploadFile): boolean {
	return shouldGzip(file) && isTextMimeType(file.mimeType)
}

async function base64FallbackBlob(
	agent: Agent,
	input: ValidatedPublicUploadFile,
	signal?: AbortSignal,
): Promise<FileUploadResult> {
	const prepared = await readFileForUpload(input, signal)
	try {
		if (!prepared.compressed) throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
		const content = Buffer.from(prepared.content.toString('base64'), 'ascii')
		const result = await uploadBlobWithRetry(agent, content, 'application/octet-stream', input.path, signal)
		result.encoding = 'gzip'
		result.mimeType = input.mimeType
		result.base64 = true
		return result
	} finally {
		prepared.content = Buffer.alloc(0)
	}
}

async function base64FallbackBlobs(
	agent: Agent,
	files: readonly ValidatedPublicUploadFile[],
	blobs: readonly UploadedBlob[],
	signal?: AbortSignal,
): Promise<UploadedBlob[]> {
	const fallback = blobs.map((blob) => ({ ...blob, result: { ...blob.result } }))
	for (const file of files) {
		throwIfAborted(signal)
		if (!fallbackEligible(file)) continue
		const index = fallback.findIndex((blob) => blob.filePath === file.path)
		if (index === -1) throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
		fallback[index] = { ...fallback[index]!, result: await base64FallbackBlob(agent, file, signal) }
	}
	return fallback
}

async function commitWithBase64Fallback(
	agent: Agent,
	did: string,
	siteName: string,
	directory: Directory,
	files: readonly ValidatedPublicUploadFile[],
	blobs: readonly UploadedBlob[],
	jobId: string,
	signal: AbortSignal | undefined,
	expectedRootCid: string | null,
): Promise<ManifestCommit> {
	try {
		return await commitPublicUploadManifest(
			agent,
			did,
			siteName,
			directory,
			blobs.length,
			jobId,
			signal,
			expectedRootCid,
		)
	} catch (error) {
		if (errorStatus(error) !== 500) throw error
		const fallback = await base64FallbackBlobs(agent, files, blobs, signal)
		const fallbackDirectory = updatedDirectoryForBlobs(directory, fallback, files.length)
		return await commitPublicUploadManifest(
			agent,
			did,
			siteName,
			fallbackDirectory,
			fallback.length,
			jobId,
			signal,
			expectedRootCid,
		)
	}
}

function uploadFailure(error: unknown): { message: string; status: number } {
	if (!(error instanceof PublicUploadError)) return { message: UPLOAD_FAILED_MESSAGE, status: 500 }
	if (error.status === 413) return { message: UPLOAD_TOO_LARGE_MESSAGE, status: 413 }
	if (error.status === 409) return { message: UPLOAD_CONFLICT_MESSAGE, status: 409 }
	return { message: UPLOAD_FAILED_MESSAGE, status: error.status }
}

export async function processUploadInBackground(
	jobId: string,
	agent: Agent,
	did: string,
	siteName: string,
	files: ValidatedPublicUploadFile[],
	skippedFiles: Array<{ name: string; reason: string }>,
	signal?: AbortSignal,
): Promise<void> {
	try {
		throwIfUploadCancelled(jobId, signal)
		const existingState = await loadExistingUploadState(agent, did, siteName)
		throwIfUploadCancelled(jobId, signal)
		const metadata = createMetadataDirectory(files)
		if (metadata.fileCount !== files.length) throw new PublicUploadError(500, UPLOAD_FAILED_MESSAGE)
		const blobs = await uploadAllFiles(jobId, files, existingState, agent, signal)
		throwIfUploadCancelled(jobId, signal)
		const directory = updatedDirectoryForBlobs(metadata.directory, blobs, metadata.fileCount)
		const committed = await commitWithBase64Fallback(
			agent,
			did,
			siteName,
			directory,
			files,
			blobs,
			jobId,
			signal,
			existingState.rootCid,
		)
		throwIfUploadCancelled(jobId, signal)
		completeUploadJob(jobId, {
			success: true,
			uri: committed.record.data.uri,
			cid: committed.record.data.cid,
			fileCount: metadata.fileCount,
			siteName,
			skippedFiles,
			uploadedCount: files.length,
			hasFailures: false,
		})
	} catch (error) {
		logger.error('Public upload job failed', { errorKind: 'upload_failed' })
		const failure = uploadFailure(error)
		failUploadJob(jobId, failure.message, failure.status)
	}
}