#!/usr/bin/env bun
/**
* Read-only audit of S3 cache object source identity metadata against current
* place.wisp.fs records on the records' authoritative PDSes.
*
* This script never writes to S3, a PDS, Redis, or Postgres.
*/
import { writeFile } from 'node:fs/promises'
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'
import { S3StorageTier } from '@wispplace/tiered-storage'
const DEFAULT_MAX_SITES = 50
const DEFAULT_MAX_OBJECTS = 2_000
const DEFAULT_CONCURRENCY = 8
const MAX_PDS_RECORD_BYTES = 1024 * 1024
const MAX_ISSUE_DETAILS = 500
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_cached_path'
| 'metadata_missing'
| 'metadata_read_error'
interface Finding {
kind: FindingKind
key: string
expectedCid?: string
observedCid?: string
expectedDid?: string
observedDid?: string
error?: string
}
interface SiteResult extends SiteRef {
status: 'audited' | 'record_absent' | 'record_error'
recordCid?: string
canonicalFiles?: number
selectedObjects: number
counts: Record<string, number>
findings: Finding[]
error?: string
}
function help(): never {
console.log(`Usage:
bun --env-file=apps/firehose-service/.env apps/firehose-service/scripts/audit-s3-source-cids.ts [options]
Options:
--max-sites <n> Deterministically sample at most this many sites (default: ${DEFAULT_MAX_SITES})
--max-objects <n> Audit at most this many objects across sampled sites (default: ${DEFAULT_MAX_OBJECTS})
--concurrency <n> Concurrent PDS and S3 metadata reads (default: ${DEFAULT_CONCURRENCY})
--site <did/rkey> Audit a specific site; repeatable
--prefix <prefix> Limit the initial S3 key listing
--seed <text> Deterministic sample seed (default: wisp-s3-cid-audit-v1)
--report <path> JSON report path (default: /tmp/wisp-s3-cid-audit-<timestamp>.json)
--all Audit every discovered site and object
--fail-on-findings Exit 2 for CID/DID/path findings or missing identity metadata
--help Show this help
The audit is read-only. Rewritten HTML keys are checked against the original
file path because their sourceCid identifies the immutable source blob.`)
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 Error(`${name} must be a positive integer`)
return value
}
function requiredEnvironment(name: string): string {
const value = process.env[name]
if (!value) throw new Error(`Missing required environment variable: ${name}`)
return value
}
function parseBooleanEnvironment(value: string | undefined, fallback: boolean): boolean {
if (value === undefined) return fallback
return ['1', 'true', 'yes'].includes(value.toLowerCase())
}
function parseSiteId(value: string): SiteRef {
const slash = value.indexOf('/')
if (slash <= 0 || slash === value.length - 1) throw new Error(`Invalid --site value: ${value}`)
const did = value.slice(0, slash)
const rkey = value.slice(slash + 1)
if (!did.startsWith('did:') || rkey.includes('/')) throw new Error(`Invalid --site value: ${value}`)
return { did, rkey, id: `${did}/${rkey}` }
}
function parseObjectKey(key: string): (SiteRef & { objectPath: string; 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}`, objectPath, canonicalPath }
}
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 delta = stableHash(`${seed}:${identity(left)}`) - stableHash(`${seed}:${identity(right)}`)
return delta || identity(left).localeCompare(identity(right))
})
.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 safeError(error: unknown): string {
if (!(error instanceof Error)) return 'UNKNOWN_ERROR'
if (/^[A-Z][A-Z0-9_]{2,127}$/.test(error.message)) return error.message
return /^[A-Za-z][A-Za-z0-9_]{0,63}$/.test(error.name) ? error.name : 'UNKNOWN_ERROR'
}
const { values } = parseArgs({
options: {
all: { type: 'boolean', default: false },
concurrency: { type: 'string' },
'fail-on-findings': { type: 'boolean', default: false },
help: { type: 'boolean', short: 'h', default: false },
'max-objects': { type: 'string' },
'max-sites': { type: 'string' },
prefix: { type: 'string' },
report: { type: 'string' },
seed: { type: 'string', default: 'wisp-s3-cid-audit-v1' },
site: { type: 'string', multiple: true, default: [] },
},
strict: true,
allowPositionals: false,
})
if (values.help) help()
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-s3-cid-audit-${new Date().toISOString().replaceAll(/[:.]/g, '-')}.json`
const tier = new S3StorageTier({
bucket: requiredEnvironment('S3_BUCKET'),
region: process.env.S3_REGION || 'us-east-1',
endpoint: process.env.S3_ENDPOINT,
prefix: process.env.S3_PREFIX,
forcePathStyle: parseBooleanEnvironment(process.env.S3_FORCE_PATH_STYLE, true),
credentials:
process.env.AWS_ACCESS_KEY_ID && process.env.AWS_SECRET_ACCESS_KEY
? {
accessKeyId: process.env.AWS_ACCESS_KEY_ID,
secretAccessKey: process.env.AWS_SECRET_ACCESS_KEY,
}
: undefined,
})
console.error('Listing S3 object keys (read-only)...')
const keysBySite = new Map<string, string[]>()
let listedObjects = 0
let malformedKeys = 0
for await (const key of tier.listKeys(values.prefix)) {
listedObjects++
const parsed = parseObjectKey(key)
if (!parsed) {
malformedKeys++
continue
}
const existing = keysBySite.get(parsed.id)
if (existing) existing.push(key)
else keysBySite.set(parsed.id, [key])
}
const discoveredSites = [...keysBySite.keys()].sort().map(parseSiteId)
let selectedSites: SiteRef[]
if (requestedSites.length > 0) {
selectedSites = [...new Map(requestedSites.map((site) => [site.id, site])).values()]
for (const site of selectedSites) {
if (!keysBySite.has(site.id)) keysBySite.set(site.id, [])
}
} else {
selectedSites = deterministicSample(discoveredSites, maxSites, (site) => site.id, values.seed!)
}
// Round-robin deterministic object sampling preserves coverage across selected sites.
const perSiteCandidates = new Map<string, string[]>()
for (const site of selectedSites) {
perSiteCandidates.set(
site.id,
deterministicSample(keysBySite.get(site.id) ?? [], Number.MAX_SAFE_INTEGER, (key) => key, values.seed!),
)
}
const selectedKeysBySite = new Map<string, string[]>()
let selectedObjects = 0
for (let offset = 0; selectedObjects < maxObjects; offset++) {
let added = false
for (const site of selectedSites) {
const key = perSiteCandidates.get(site.id)?.[offset]
if (!key || selectedObjects >= maxObjects) continue
const existing = selectedKeysBySite.get(site.id)
if (existing) existing.push(key)
else selectedKeysBySite.set(site.id, [key])
selectedObjects++
added = true
}
if (!added) break
}
selectedSites = selectedSites.filter(
(site) => requestedSites.length > 0 || (selectedKeysBySite.get(site.id)?.length ?? 0) > 0,
)
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 Error('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 errorBody = (await response.json()) as { error?: unknown }
if (errorBody.error === 'RecordNotFound') return null
} else {
await response.body?.cancel()
}
throw new Error(`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 Error('PDS_RECORD_MISSING_CID')
const record = parseLexiconJson<WispFsRecord>(body.value)
if (!validateFsRecord(record).success || !record.root?.entries) throw new Error('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 }
}
function increment(counts: Record<string, number>, kind: string): void {
counts[kind] = (counts[kind] ?? 0) + 1
}
async function auditSite(site: SiteRef): Promise<SiteResult> {
const keys = selectedKeysBySite.get(site.id) ?? []
let canonical: CanonicalSite | null
try {
canonical = await fetchCanonicalSite(site)
} catch (error) {
return {
...site,
status: 'record_error',
selectedObjects: keys.length,
counts: {},
findings: [],
error: safeError(error),
}
}
if (!canonical) {
return {
...site,
status: 'record_absent',
selectedObjects: keys.length,
counts: { record_absent: 1, orphaned_cached_path: keys.length },
findings: keys.slice(0, MAX_ISSUE_DETAILS).map((key) => ({ kind: 'orphaned_cached_path', key })),
}
}
const counts: Record<string, number> = {}
const findings = await mapConcurrent(keys, concurrency, async (key): Promise<Finding> => {
const parsed = parseObjectKey(key)!
const expectedCid = canonical.fileCids[parsed.canonicalPath]
const expectedDid = canonical.ownerDidByPath.get(parsed.canonicalPath)
if (!expectedCid || !expectedDid) {
increment(counts, 'orphaned_cached_path')
return { kind: 'orphaned_cached_path', key }
}
try {
const metadata = await tier.getMetadata(key)
if (!metadata) {
increment(counts, 'metadata_missing')
return { kind: 'metadata_missing', key, expectedCid, expectedDid }
}
const observedCid = metadata.customMetadata?.sourceCid
const observedDid = metadata.customMetadata?.sourceDid
if (!observedCid || !observedDid) {
increment(counts, 'missing_source_identity')
return { kind: 'missing_source_identity', key, expectedCid, observedCid, expectedDid, observedDid }
}
const cidMatches = observedCid === expectedCid
const didMatches = observedDid === expectedDid
if (cidMatches && didMatches) {
increment(counts, 'match')
return { kind: 'match', key }
}
const kind =
!cidMatches && !didMatches
? 'source_cid_and_did_mismatch'
: cidMatches
? 'source_did_mismatch'
: 'source_cid_mismatch'
increment(counts, kind)
return { kind, key, expectedCid, observedCid, expectedDid, observedDid }
} catch (error) {
increment(counts, 'metadata_read_error')
return { kind: 'metadata_read_error', key, expectedCid, expectedDid, error: safeError(error) }
}
})
return {
...site,
status: 'audited',
recordCid: canonical.recordCid,
canonicalFiles: Object.keys(canonical.fileCids).length,
selectedObjects: keys.length,
counts,
findings: findings.filter((finding) => finding.kind !== 'match').slice(0, MAX_ISSUE_DETAILS),
}
}
console.error(
`Auditing ${selectedObjects} objects across ${selectedSites.length} sites with concurrency ${concurrency} (read-only)...`,
)
const siteResults = await mapConcurrent(selectedSites, Math.min(concurrency, 4), auditSite)
const totals: Record<string, number> = {}
for (const site of siteResults) {
increment(totals, `site_${site.status}`)
for (const [kind, count] of Object.entries(site.counts)) totals[kind] = (totals[kind] ?? 0) + count
}
const report = {
schemaVersion: 1,
generatedAt: new Date().toISOString(),
readOnly: true,
security: {
sshInvoked: false,
credentialValuesReported: false,
reportIncludesObjectBytes: false,
reportIncludesConnectionStrings: false,
reportMode: '0600',
},
scope: {
all: values.all,
prefix: values.prefix ?? null,
seed: values.seed,
requestedSites: requestedSites.map((site) => site.id),
listedObjects,
discoveredSites: discoveredSites.length,
malformedKeys,
selectedSites: selectedSites.length,
selectedObjects,
maxSites: values.all ? null : maxSites,
maxObjects: values.all ? null : maxObjects,
concurrency,
},
totals,
sites: siteResults,
}
await writeFile(reportPath, `${JSON.stringify(report, null, 2)}\n`, { mode: 0o600 })
console.log(JSON.stringify({ reportPath, scope: report.scope, totals }, null, 2))
const findingKinds = [
'missing_source_identity',
'source_cid_mismatch',
'source_did_mismatch',
'source_cid_and_did_mismatch',
'orphaned_cached_path',
'metadata_missing',
'metadata_read_error',
]
if (
values['fail-on-findings'] &&
((totals.site_record_error ?? 0) > 0 || findingKinds.some((kind) => (totals[kind] ?? 0) > 0))
) {
process.exitCode = 2
}