#!/usr/bin/env bun
/**
 * Read-only warm-cache source identity audit.
 *
 * Inputs are local DiskStorageTier cache directories or sanitized JSONL
 * inventories containing only: node, key, sourceCid, and sourceDid.
 * The script does not invoke SSH, use environment endpoint overrides, or accept credentials.
 */
import { opendir, readFile, writeFile } from 'node:fs/promises'
import { basename, join, relative } from 'node:path'
import { parseArgs } from 'node:util'
import { expandSubfs, getPdsForDid, type SubfsSubject } from '@wispplace/atproto-utils'
import { collectFileCidsFromEntries } from '@wispplace/fs-utils'
import { parseLexiconJson } from '@wispplace/lexicons/public-json'
import type { Record as WispFsRecord } from '@wispplace/lexicons/types/place/wisp/fs'
import { validateRecord as validateFsRecord } from '@wispplace/lexicons/types/place/wisp/fs'
import { safeFetch, safeFetchJson } from '@wispplace/safe-fetch'

const DEFAULT_MAX_SITES = 50
const DEFAULT_MAX_OBJECTS = 2_000
const DEFAULT_CONCURRENCY = 8
const MAX_PDS_RECORD_BYTES = 1024 * 1024
const MAX_METADATA_BYTES = 64 * 1024
const MAX_ISSUE_DETAILS = 500

// Do not let Bun's automatic .env loading redirect this standalone audit to a
// local identity service. The audit always uses the public identity defaults.
for (const name of ['WISP_PLC_DIRECTORY_URL', 'WISP_HANDLE_RESOLVER_URL', 'WISP_ALLOW_LOCALHOST_FETCH']) {
	delete process.env[name]
}

interface WarmEntry {
	node: string
	key: string
	sourceCid?: string
	sourceDid?: string
}

interface SiteRef {
	did: string
	rkey: string
	id: string
}

interface CanonicalSite extends SiteRef {
	recordCid: string
	fileCids: Readonly<Record<string, string>>
	ownerDidByPath: ReadonlyMap<string, string>
}

type FindingKind =
	| 'match'
	| 'missing_source_identity'
	| 'source_cid_mismatch'
	| 'source_did_mismatch'
	| 'source_cid_and_did_mismatch'
	| 'orphaned_warm_path'
	| 'record_absent'
	| 'record_error'

interface Finding {
	kind: FindingKind
	node: string
	key: string
	expectedCid?: string
	observedCid?: string
	expectedDid?: string
	observedDid?: string
	errorCode?: string
}

class AuditError extends Error {
	constructor(readonly code: string) {
		super(code)
		this.name = 'AuditError'
	}
}

function help(): never {
	console.log(`Usage:
  bun --no-env-file apps/firehose-service/scripts/audit-warm-source-cids.ts [options]

Input options (at least one is required):
  --inventory <path>    Sanitized JSONL inventory; repeatable
  --cache-dir <path>    Local DiskStorageTier cache directory; repeatable
  --node <label>        Node label for one --cache-dir input (default: local)

Audit options:
  --max-sites <n>       Deterministically sample at most this many sites (default: ${DEFAULT_MAX_SITES})
  --max-objects <n>     Audit at most this many warm objects (default: ${DEFAULT_MAX_OBJECTS})
  --concurrency <n>     Concurrent PDS reads (default: ${DEFAULT_CONCURRENCY})
  --site <did/rkey>     Audit a specific site; repeatable
  --seed <text>         Deterministic sample seed (default: wisp-warm-cid-audit-v1)
  --report <path>       JSON report path (default: /tmp/wisp-warm-cid-audit-<timestamp>.json)
  --all                 Audit every discovered site and inventory entry
  --fail-on-findings    Exit 2 for mismatches, orphans, missing identity, or incomplete PDS reads
  --help                Show this help

Inventory JSONL schema:
  {"node":"edge-a","key":"did:plc:.../rkey/index.html","sourceCid":"bafk...","sourceDid":"did:plc:..."}

The script is read-only. It does not invoke SSH or accept credentials. It clears
identity endpoint overrides and does not include cache bytes, checksums, environment
values, or connection strings in reports. Use --no-env-file as shown above.`)
	process.exit(0)
}

function parsePositiveInteger(raw: string | undefined, name: string, fallback: number): number {
	if (raw === undefined) return fallback
	const value = Number(raw)
	if (!Number.isSafeInteger(value) || value < 1) throw new AuditError(`${name.slice(2).toUpperCase()}_INVALID`)
	return value
}

function parseSiteId(value: string): SiteRef {
	const slash = value.indexOf('/')
	if (slash <= 0 || slash === value.length - 1) throw new AuditError('SITE_INVALID')
	const did = value.slice(0, slash)
	const rkey = value.slice(slash + 1)
	if (!did.startsWith('did:') || rkey.includes('/')) throw new AuditError('SITE_INVALID')
	return { did, rkey, id: `${did}/${rkey}` }
}

function parseObjectKey(key: string): (SiteRef & { canonicalPath: string }) | null {
	const firstSlash = key.indexOf('/')
	const secondSlash = key.indexOf('/', firstSlash + 1)
	if (firstSlash <= 0 || secondSlash <= firstSlash + 1 || secondSlash === key.length - 1) return null
	const did = key.slice(0, firstSlash)
	const rkey = key.slice(firstSlash + 1, secondSlash)
	const objectPath = key.slice(secondSlash + 1)
	if (!did.startsWith('did:')) return null
	const canonicalPath = objectPath.startsWith('.rewritten/') ? objectPath.slice('.rewritten/'.length) : objectPath
	if (!canonicalPath) return null
	return { did, rkey, id: `${did}/${rkey}`, canonicalPath }
}

function boundedString(value: unknown, maximum: number): string | undefined {
	return typeof value === 'string' && value.length > 0 && value.length <= maximum ? value : undefined
}

function parseInventoryRow(raw: unknown): WarmEntry | null {
	if (typeof raw !== 'object' || raw === null || Array.isArray(raw)) return null
	const row = raw as Record<string, unknown>
	const node = boundedString(row.node, 128)
	const key = boundedString(row.key, 8192)
	if (!node || !key) return null
	const sourceCid = boundedString(row.sourceCid, 512)
	const sourceDid = boundedString(row.sourceDid, 2048)
	return { node, key, ...(sourceCid ? { sourceCid } : {}), ...(sourceDid ? { sourceDid } : {}) }
}

async function loadInventory(path: string): Promise<{ entries: WarmEntry[]; invalidRows: number }> {
	const text = await readFile(path, 'utf8')
	const entries: WarmEntry[] = []
	let invalidRows = 0
	for (const line of text.split('\n')) {
		if (!line.trim()) continue
		try {
			const entry = parseInventoryRow(JSON.parse(line))
			if (entry) entries.push(entry)
			else invalidRows++
		} catch {
			invalidRows++
		}
	}
	return { entries, invalidRows }
}

async function* metadataFiles(root: string, directory = root): AsyncGenerator<string> {
	const handle = await opendir(directory)
	for await (const entry of handle) {
		const path = join(directory, entry.name)
		if (entry.isDirectory()) yield* metadataFiles(root, path)
		else if (entry.isFile() && entry.name.endsWith('.meta')) yield path
	}
}

async function loadCacheDirectory(root: string, node: string): Promise<{ entries: WarmEntry[]; invalidRows: number }> {
	const entries: WarmEntry[] = []
	let invalidRows = 0
	for await (const path of metadataFiles(root)) {
		try {
			const bytes = await readFile(path)
			if (bytes.byteLength > MAX_METADATA_BYTES) throw new AuditError('METADATA_OVERSIZED')
			const metadata = JSON.parse(bytes.toString('utf8')) as Record<string, unknown>
			const fallbackKey = relative(root, path).slice(0, -'.meta'.length)
			const key = boundedString(metadata.key, 8192) ?? fallbackKey
			const custom =
				typeof metadata.customMetadata === 'object' && metadata.customMetadata !== null
					? (metadata.customMetadata as Record<string, unknown>)
					: {}
			const parsed = parseInventoryRow({
				node,
				key,
				sourceCid: custom.sourceCid,
				sourceDid: custom.sourceDid,
			})
			if (parsed) entries.push(parsed)
			else invalidRows++
		} catch {
			invalidRows++
		}
	}
	return { entries, invalidRows }
}

function stableHash(value: string): number {
	let hash = 0x811c9dc5
	for (let index = 0; index < value.length; index++) {
		hash ^= value.charCodeAt(index)
		hash = Math.imul(hash, 0x01000193)
	}
	return hash >>> 0
}

function deterministicSample<T>(
	values: readonly T[],
	limit: number,
	identity: (value: T) => string,
	seed: string,
): T[] {
	if (values.length <= limit) return [...values]
	return [...values]
		.sort((left, right) => {
			const leftId = identity(left)
			const rightId = identity(right)
			return stableHash(`${seed}:${leftId}`) - stableHash(`${seed}:${rightId}`) || leftId.localeCompare(rightId)
		})
		.slice(0, limit)
}

async function mapConcurrent<T, R>(
	values: readonly T[],
	concurrency: number,
	mapper: (value: T) => Promise<R>,
): Promise<R[]> {
	const results = new Array<R>(values.length)
	let nextIndex = 0
	async function worker(): Promise<void> {
		while (true) {
			const index = nextIndex++
			if (index >= values.length) return
			results[index] = await mapper(values[index]!)
		}
	}
	await Promise.all(Array.from({ length: Math.min(concurrency, values.length) }, () => worker()))
	return results
}

function errorCode(error: unknown): string {
	if (error instanceof AuditError) return error.code
	if (error instanceof Error && error.name === 'TimeoutError') return 'TIMEOUT'
	return 'FETCH_FAILED'
}

const { values } = parseArgs({
	options: {
		all: { type: 'boolean', default: false },
		'cache-dir': { type: 'string', multiple: true, default: [] },
		concurrency: { type: 'string' },
		'fail-on-findings': { type: 'boolean', default: false },
		help: { type: 'boolean', short: 'h', default: false },
		inventory: { type: 'string', multiple: true, default: [] },
		'max-objects': { type: 'string' },
		'max-sites': { type: 'string' },
		node: { type: 'string' },
		report: { type: 'string' },
		seed: { type: 'string', default: 'wisp-warm-cid-audit-v1' },
		site: { type: 'string', multiple: true, default: [] },
	},
	strict: true,
	allowPositionals: false,
})
if (values.help) help()
if ((values.inventory?.length ?? 0) === 0 && (values['cache-dir']?.length ?? 0) === 0)
	throw new AuditError('INPUT_REQUIRED')
if (values.node && (values['cache-dir']?.length ?? 0) !== 1) throw new AuditError('NODE_REQUIRES_ONE_CACHE_DIR')

const concurrency = parsePositiveInteger(values.concurrency, '--concurrency', DEFAULT_CONCURRENCY)
const maxSites = values.all
	? Number.MAX_SAFE_INTEGER
	: parsePositiveInteger(values['max-sites'], '--max-sites', DEFAULT_MAX_SITES)
const maxObjects = values.all
	? Number.MAX_SAFE_INTEGER
	: parsePositiveInteger(values['max-objects'], '--max-objects', DEFAULT_MAX_OBJECTS)
const requestedSites = (values.site ?? []).map(parseSiteId)
const reportPath = values.report ?? `/tmp/wisp-warm-cid-audit-${new Date().toISOString().replaceAll(/[:.]/g, '-')}.json`

const loaded: WarmEntry[] = []
let invalidInventoryRows = 0
for (const path of values.inventory ?? []) {
	const result = await loadInventory(path)
	loaded.push(...result.entries)
	invalidInventoryRows += result.invalidRows
}
for (const [index, root] of (values['cache-dir'] ?? []).entries()) {
	const node = values.node ?? `${basename(root) || 'local'}-${index + 1}`
	const result = await loadCacheDirectory(root, node)
	loaded.push(...result.entries)
	invalidInventoryRows += result.invalidRows
}

const entries = [...new Map(loaded.map((entry) => [`${entry.node}\0${entry.key}`, entry])).values()]
const siteById = new Map<string, SiteRef>()
let malformedKeys = 0
for (const entry of entries) {
	const parsed = parseObjectKey(entry.key)
	if (parsed) siteById.set(parsed.id, parsed)
	else malformedKeys++
}
const discoveredSites = [...siteById.values()].sort((left, right) => left.id.localeCompare(right.id))
const selectedSites =
	requestedSites.length > 0
		? [...new Map(requestedSites.map((site) => [site.id, site])).values()]
		: deterministicSample(discoveredSites, maxSites, (site) => site.id, values.seed!)
const selectedSiteIds = new Set(selectedSites.map((site) => site.id))
const eligibleEntries = entries.filter((entry) => {
	const parsed = parseObjectKey(entry.key)
	return parsed !== null && selectedSiteIds.has(parsed.id)
})
const selectedEntries = deterministicSample(
	eligibleEntries,
	maxObjects,
	(entry) => `${entry.node}:${entry.key}`,
	values.seed!,
)

const identityFetch: Parameters<typeof getPdsForDid>[1] = (url, options) =>
	safeFetch(url, { signal: options?.signal, byteBudget: options?.byteBudget })
const pdsEndpoints = new Map<string, string>()
async function resolvePdsEndpoint(did: string, signal: AbortSignal): Promise<string> {
	const existing = pdsEndpoints.get(did)
	if (existing) return existing
	const endpoint = await getPdsForDid(did, identityFetch, undefined, { signal })
	if (!endpoint) throw new AuditError('PDS_UNRESOLVED')
	pdsEndpoints.set(did, endpoint)
	return endpoint
}

async function fetchCanonicalSite(site: SiteRef): Promise<CanonicalSite | null> {
	const signal = AbortSignal.timeout(30_000)
	const rootPds = await resolvePdsEndpoint(site.did, signal)
	const query = new URLSearchParams({ repo: site.did, collection: 'place.wisp.fs', rkey: site.rkey })
	const response = await safeFetch(`${rootPds}/xrpc/com.atproto.repo.getRecord?${query.toString()}`, {
		signal,
		maxSize: MAX_PDS_RECORD_BYTES,
	})
	if (response.status === 404) {
		await response.body?.cancel()
		return null
	}
	if (!response.ok) {
		if (response.status === 400) {
			const body = (await response.json()) as { error?: unknown }
			if (body.error === 'RecordNotFound') return null
		} else await response.body?.cancel()
		throw new AuditError(`PDS_RECORD_HTTP_${response.status}`)
	}
	const body = (await response.json()) as { cid?: unknown; value?: unknown }
	if (typeof body.cid !== 'string' || body.cid.length === 0) throw new AuditError('PDS_RECORD_MISSING_CID')
	const record = parseLexiconJson<WispFsRecord>(body.value)
	if (!validateFsRecord(record).success || !record.root?.entries) throw new AuditError('PDS_RECORD_INVALID')
	const expanded = await expandSubfs(record.root, {
		rootOwnerDid: site.did,
		signal,
		limits: {
			maxConcurrentFetches: Math.min(4, concurrency),
			maxDepth: 10,
			maxEntries: 40_000,
			maxFiles: 10_000,
			maxRecords: 100,
			maxRawJsonBytes: 10 * 1024 * 1024,
		},
		fetchSubfsRecord: async (subject: SubfsSubject) => {
			const endpoint = await resolvePdsEndpoint(subject.repo, signal)
			const subjectQuery = new URLSearchParams({
				repo: subject.repo,
				collection: subject.collection,
				rkey: subject.rkey,
			})
			const data = await safeFetchJson<{ value?: unknown }>(
				`${endpoint}/xrpc/com.atproto.repo.getRecord?${subjectQuery.toString()}`,
				{ signal, maxSize: MAX_PDS_RECORD_BYTES },
			)
			return data.value
		},
	})
	const fileCids: Record<string, string> = {}
	collectFileCidsFromEntries(expanded.root.entries, '', fileCids)
	return { ...site, recordCid: body.cid, fileCids, ownerDidByPath: expanded.ownerDidByFilePath }
}

console.error(
	`Auditing ${selectedEntries.length} warm objects across ${selectedSites.length} sites and ${new Set(selectedEntries.map((entry) => entry.node)).size} nodes (read-only)...`,
)
const canonicalResults = await mapConcurrent(selectedSites, Math.min(concurrency, 4), async (site) => {
	try {
		return { site, canonical: await fetchCanonicalSite(site) }
	} catch (error) {
		return { site, canonical: undefined, errorCode: errorCode(error) }
	}
})
const canonicalBySite = new Map(canonicalResults.map((result) => [result.site.id, result]))

const findings: Finding[] = []
const totals: Record<string, number> = {}
const byNode: Record<string, Record<string, number>> = {}
function increment(node: string, kind: FindingKind): void {
	totals[kind] = (totals[kind] ?? 0) + 1
	byNode[node] ??= {}
	byNode[node]![kind] = (byNode[node]![kind] ?? 0) + 1
}
function addFinding(finding: Finding): void {
	increment(finding.node, finding.kind)
	if (finding.kind !== 'match' && findings.length < MAX_ISSUE_DETAILS) findings.push(finding)
}

for (const entry of selectedEntries) {
	const parsed = parseObjectKey(entry.key)!
	const result = canonicalBySite.get(parsed.id)!
	if (result.errorCode) {
		addFinding({ kind: 'record_error', node: entry.node, key: entry.key, errorCode: result.errorCode })
		continue
	}
	if (!result.canonical) {
		addFinding({ kind: 'record_absent', node: entry.node, key: entry.key })
		continue
	}
	const expectedCid = result.canonical.fileCids[parsed.canonicalPath]
	const expectedDid = result.canonical.ownerDidByPath.get(parsed.canonicalPath)
	if (!expectedCid || !expectedDid) {
		addFinding({ kind: 'orphaned_warm_path', node: entry.node, key: entry.key })
		continue
	}
	if (!entry.sourceCid || !entry.sourceDid) {
		addFinding({
			kind: 'missing_source_identity',
			node: entry.node,
			key: entry.key,
			expectedCid,
			observedCid: entry.sourceCid,
			expectedDid,
			observedDid: entry.sourceDid,
		})
		continue
	}
	const cidMatches = entry.sourceCid === expectedCid
	const didMatches = entry.sourceDid === expectedDid
	if (cidMatches && didMatches) {
		addFinding({ kind: 'match', node: entry.node, key: entry.key })
		continue
	}
	const kind =
		!cidMatches && !didMatches
			? 'source_cid_and_did_mismatch'
			: cidMatches
				? 'source_did_mismatch'
				: 'source_cid_mismatch'
	addFinding({
		kind,
		node: entry.node,
		key: entry.key,
		expectedCid,
		observedCid: entry.sourceCid,
		expectedDid,
		observedDid: entry.sourceDid,
	})
}

const report = {
	schemaVersion: 1,
	generatedAt: new Date().toISOString(),
	readOnly: true,
	security: {
		credentialInputs: false,
		sshInvoked: false,
		reportIncludesCacheBytes: false,
		reportIncludesChecksums: false,
		reportMode: '0600',
	},
	scope: {
		all: values.all,
		seed: values.seed,
		inventoryFiles: values.inventory?.length ?? 0,
		cacheDirectories: values['cache-dir']?.length ?? 0,
		invalidInventoryRows,
		loadedObjects: entries.length,
		discoveredSites: discoveredSites.length,
		malformedKeys,
		selectedSites: selectedSites.length,
		selectedObjects: selectedEntries.length,
		selectedNodes: [...new Set(selectedEntries.map((entry) => entry.node))].sort(),
		maxSites: values.all ? null : maxSites,
		maxObjects: values.all ? null : maxObjects,
		concurrency,
	},
	totals,
	byNode,
	findings,
}
await writeFile(reportPath, `${JSON.stringify(report, null, 2)}\n`, { mode: 0o600 })
console.log(JSON.stringify({ reportPath, scope: report.scope, totals, byNode }, null, 2))

if (values['fail-on-findings'] && Object.entries(totals).some(([kind, count]) => kind !== 'match' && count > 0)) {
	process.exitCode = 2
}