wip: hydrant backfill discovery and domain cache invalidation on claim
55 unmodified lines565758596055 unmodified lines**/*.bun
server.env.deve2e-artifacts/44 unmodified lines4546474849505152535444 unmodified lines process.argv.includes('--db-fill-only') || process.env.BACKFILL === 'true' || process.env.DB_FILL_ONLY === 'true', backfillConcurrency: parseInt(process.env.BACKFILL_CONCURRENCY || '5', 10),
// Optional hydrant indexer used as the backfill discovery source. When set, // backfill enumerates place.wisp.fs repos from hydrant instead of only // rediscovering DIDs already present in our own database. hydrantUrl: process.env.HYDRANT_URL,} as const8 unmodified lines9101112131415161760 unmodified lines78798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129868788130131132133134135136137202 unmodified linesExpand all8 unmodified lines
import { serve } from '@hono/node-server'import { createLogger, initializeGrafanaExporters } from '@wispplace/observability'import { observabilityErrorHandler, observabilityMiddleware } from '@wispplace/observability/middleware/hono'import { safeFetchJson } from '@wispplace/safe-fetch'import { Hono } from 'hono'import { config } from './config'import { closeCacheInvalidationPublisher } from './lib/cache-invalidation'import { fetchSiteRecord, handleSiteCreateOrUpdate, listSiteRecordsForDid } from './lib/cache-writer'60 unmodified linesprocess.on('SIGINT', () => shutdown('SIGINT'))process.on('SIGTERM', () => shutdown('SIGTERM'))
/** * Enumerate DIDs from a hydrant indexer. * * Unlike listAllKnownDids(), which can only rediscover DIDs we already have a * row for, hydrant tracks every repo in the network that publishes a * place.wisp.* record, so this also surfaces sites we have never seen. */async function listDidsFromHydrant(hydrantUrl: string): Promise<string[]> { const base = hydrantUrl.replace(/\/$/, '') const pageSize = 1000 const dids: string[] = [] // hydrant's JSON response is a bare array with no cursor field; pages are // walked by passing the last DID of the previous page (exclusive). let cursor: string | undefined
while (true) { const params = new URLSearchParams({ limit: String(pageSize) }) if (cursor) params.set('cursor', cursor)
const rows = (await safeFetchJson(`${base}/repos?${params.toString()}`, { headers: { Accept: 'application/json' }, })) as Array<{ did?: string }>
if (!Array.isArray(rows) || rows.length === 0) break
for (const row of rows) { if (row.did) dids.push(row.did) }
const lastDid = rows[rows.length - 1]?.did if (!lastDid || rows.length < pageSize) break cursor = lastDid }
return [...new Set(dids)]}
/** * Backfill phase 1+2: * - Collect all known DIDs from DB * - Discover each DID's place.wisp.fs records directly from the PDS */async function collectSitesFromKnownDids(): Promise<Array<{ did: string; rkey: string }>> { let dids: string[] if (config.hydrantUrl) { logger.info(`Phase 1/3: Collecting DIDs from hydrant (${config.hydrantUrl})`) dids = await listDidsFromHydrant(config.hydrantUrl) logger.info(`Collected ${dids.length} DIDs from hydrant`) } else { logger.info('Phase 1/3: Collecting known DIDs') const dids = await listAllKnownDids() logger.info(`Collected ${dids.length} known DIDs`) logger.info('Phase 1/3: Collecting known DIDs') dids = await listAllKnownDids() logger.info(`Collected ${dids.length} known DIDs`) }
if (dids.length === 0) { logger.warn('No known DIDs found; skipping site discovery') return []202 unmodified linesExpand all371 unmodified linesExpand all37237337437537637737837938020 unmodified lines40140240340440540640740840924 unmodified lines43443543643743843944044144212 unmodified lines45545645745845946046146246326 unmodified lines4904914924934944954964974988 unmodified lines507508509510511512513514515295 unmodified linesExpand all811812813814815816817818819820821822823824825826827236 unmodified linesExpand all371 unmodified linesExpand all
if (existing.registered && existing.did === did) { if (input.siteRkey !== undefined) { await updateWispDomainSite(domain, input.siteRkey) await publishDomainCacheInvalidation(domain, 'wisp') }
return json({ domain,20 unmodified lines
if (input.siteRkey !== undefined) { await updateWispDomainSite(domain, input.siteRkey) } await publishDomainCacheInvalidation(domain, 'wisp')
return json({ domain, kind: 'wisp',24 unmodified lines
if (existing && existing.did === did) { if (input.siteRkey !== undefined) { await updateCustomDomainRkey(existing.id, input.siteRkey) await publishDomainCacheInvalidation(domain, 'custom', existing.id) }
const status = existing.verified ? 'verified' : 'pendingVerification'
12 unmodified lines await claimCustomDomain(did, domain, challengeId, input.siteRkey ?? null) } catch { alreadyClaimed('domain is already claimed') } await publishDomainCacheInvalidation(domain, 'custom', challengeId)
return json({ domain, kind: 'custom',26 unmodified lines }
if (existing.type === 'wisp') { await updateWispDomainSite(domain, siteRkey) await publishDomainCacheInvalidation(domain, 'wisp')
return json({ domain, kind: 'wisp',8 unmodified lines notFound('domain not found') }
await updateCustomDomainRkey(custom.id as string, siteRkey) await publishDomainCacheInvalidation(domain, 'custom', custom.id as string)
return json({ domain, kind: 'custom',295 unmodified linesExpand all }
if (existing.type === 'wisp') { await deleteWispDomain(domain) await publishDomainCacheInvalidation(domain, 'wisp') } else { const custom = await getCustomDomainInfo(domain) if (!custom || custom.did !== did) { notFound() }
await deleteCustomDomain(custom.id as string) await publishDomainCacheInvalidation(domain, 'custom', custom.id as string) }
return json({ domain,236 unmodified linesExpand all