From 4e22ee5d892e9878de3b423fb1a34c3feb1ddf77 Mon Sep 17 00:00:00 2001 From: "nekomimi.pet" Date: Thu, 26 Feb 2026 02:18:54 -0500 Subject: [PATCH] backfill fills db rpoperly --- apps/firehose-service/package.json | 4 +- apps/firehose-service/src/config.ts | 3 +- apps/firehose-service/src/index.ts | 78 +++++++++++++++++-- apps/firehose-service/src/lib/cache-writer.ts | 56 +++++++++++++ apps/firehose-service/src/lib/db.ts | 73 +++++++++++++++++ packages/@wispplace/constants/src/index.ts | 4 +- 6 files changed, 208 insertions(+), 10 deletions(-) diff --git a/apps/firehose-service/package.json b/apps/firehose-service/package.json index a4f695f..6177dff 100644 --- a/apps/firehose-service/package.json +++ b/apps/firehose-service/package.json @@ -5,10 +5,12 @@ "scripts": { "dev": "NODE_OPTIONS='--max-old-space-size=2048' tsx --env-file=.env src/index.ts", "dev:backfill": "NODE_OPTIONS='--max-old-space-size=2048' tsx --env-file=.env src/index.ts --backfill", + "dev:db-fill": "NODE_OPTIONS='--max-old-space-size=2048' tsx --env-file=.env src/index.ts --db-fill-only", "build": "bun run build.ts", "start": "NODE_OPTIONS='--max-old-space-size=2048' tsx src/index.ts", "check": "tsc --noEmit", - "backfill": "NODE_OPTIONS='--max-old-space-size=2048' tsx src/index.ts --backfill" + "backfill": "NODE_OPTIONS='--max-old-space-size=2048' tsx src/index.ts --backfill", + "db-fill": "NODE_OPTIONS='--max-old-space-size=2048' tsx src/index.ts --db-fill-only" }, "dependencies": { "@atproto/api": "^0.17.4", diff --git a/apps/firehose-service/src/config.ts b/apps/firehose-service/src/config.ts index ff4abfe..1176aa1 100644 --- a/apps/firehose-service/src/config.ts +++ b/apps/firehose-service/src/config.ts @@ -28,5 +28,6 @@ export const config = { revalidateGroup: process.env.WISP_REVALIDATE_GROUP || 'firehose-service', // Mode - isBackfill: process.argv.includes('--backfill'), + isDbFillOnly: process.argv.includes('--db-fill-only'), + isBackfill: process.argv.includes('--backfill') || process.argv.includes('--db-fill-only'), } as const; diff --git a/apps/firehose-service/src/index.ts b/apps/firehose-service/src/index.ts index 9e9fbca..5bac43b 100644 --- a/apps/firehose-service/src/index.ts +++ b/apps/firehose-service/src/index.ts @@ -4,15 +4,16 @@ * Modes: * - Normal: Watch firehose for place.wisp.fs events * - Backfill: Process existing sites from database + * - DB Fill Only: Collect DIDs and backfill sites table (skip S3 writes) */ import { Hono } from 'hono'; import { serve } from '@hono/node-server'; import { config } from './config'; import { startFirehose, stopFirehose, getFirehoseHealth } from './lib/firehose'; -import { closeDatabase, listAllSiteCaches, listAllSites, getSiteCache } from './lib/db'; +import { closeDatabase, listAllKnownDids, listAllSiteCaches, listAllSites, getSiteCache, upsertSite } from './lib/db'; import { storage } from './lib/storage'; -import { handleSiteCreateOrUpdate, fetchSiteRecord } from './lib/cache-writer'; +import { handleSiteCreateOrUpdate, fetchSiteRecord, listSiteRecordsForDid } from './lib/cache-writer'; import { startRevalidateWorker, stopRevalidateWorker } from './lib/revalidate-worker'; import { closeCacheInvalidationPublisher } from './lib/cache-invalidation'; import { initializeGrafanaExporters, createLogger, logCollector, errorTracker, metricsCollector } from '@wispplace/observability'; @@ -41,7 +42,7 @@ app.get('/health', async (c) => { return c.json({ status: firehoseHealth.healthy ? 'healthy' : 'degraded', - mode: config.isBackfill ? 'backfill' : 'firehose', + mode: config.isDbFillOnly ? 'db-fill-only' : (config.isBackfill ? 'backfill' : 'firehose'), firehose: firehoseHealth, storage: storageStats, }); @@ -69,16 +70,80 @@ process.on('SIGINT', () => shutdown('SIGINT')); process.on('SIGTERM', () => shutdown('SIGTERM')); /** - * Backfill mode - process existing sites from database + * Backfill phase 1+2: + * - Collect all known DIDs from DB + * - Backfill each DID's place.wisp.fs records into the sites table + */ +async function backfillSitesTableFromKnownDids(): Promise { + logger.info('Phase 1/3: Collecting known DIDs'); + const dids = await listAllKnownDids(); + logger.info(`Collected ${dids.length} known DIDs`); + + if (dids.length === 0) { + logger.warn('No known DIDs found; skipping sites table backfill'); + return; + } + + logger.info('Phase 2/3: Backfilling place.wisp.fs records into sites table'); + + let didsProcessed = 0; + let didsFailed = 0; + let sitesSynced = 0; + let sitesFailed = 0; + + for (const did of dids) { + try { + const records = await listSiteRecordsForDid(did); + for (const row of records) { + try { + const siteName = typeof row.record.site === 'string' && row.record.site.length > 0 + ? row.record.site + : row.rkey; + await upsertSite(did, row.rkey, siteName); + sitesSynced++; + } catch (err) { + logger.error(`[Backfill:sites] Failed to upsert site ${did}/${row.rkey}`, err); + sitesFailed++; + } + } + didsProcessed++; + logger.info(`[Backfill:sites] Progress ${didsProcessed + didsFailed}/${dids.length} DIDs`); + } catch (err) { + logger.error(`[Backfill:sites] Failed to list records for DID ${did}`, err); + didsFailed++; + } + } + + logger.info( + `Phase 2/3 complete: ${didsProcessed} DIDs processed, ${didsFailed} DIDs failed, ${sitesSynced} sites synced, ${sitesFailed} sites failed` + ); +} + +/** + * Backfill phase 3: + * - process sites from database and backfill blobs into S3 */ async function runBackfill(): Promise { logger.info('Starting backfill mode'); const startTime = Date.now(); const forceRewriteHtml = process.env.BACKFILL_FORCE_REWRITE_HTML === 'true'; + const forceDownload = process.env.BACKFILL_FORCE_DOWNLOAD === 'true'; if (forceRewriteHtml) { logger.info('Forcing HTML rewrite for all sites'); } + if (forceDownload) { + logger.info('Forcing full file download/write for all backfilled sites'); + } + + await backfillSitesTableFromKnownDids(); + + if (config.isDbFillOnly) { + logger.info('DB fill only mode complete; skipping phase 3/3 S3 backfill'); + return; + } + + logger.info('Phase 3/3: Backfilling site blobs into S3'); let sites = await listAllSites(); if (sites.length === 0) { @@ -106,7 +171,7 @@ async function runBackfill(): Promise { const existingCache = await getSiteCache(site.did, site.rkey); // Check if CID matches (already up to date) - if (!forceRewriteHtml && existingCache && result.cid === existingCache.record_cid) { + if (!forceRewriteHtml && !forceDownload && existingCache && result.cid === existingCache.record_cid) { logger.info(`Site already up to date: ${site.did}/${site.rkey}`); skipped++; continue; @@ -115,6 +180,7 @@ async function runBackfill(): Promise { // Process the site await handleSiteCreateOrUpdate(site.did, site.rkey, result.record, result.cid, { forceRewriteHtml, + forceDownload, }); processed++; @@ -164,7 +230,7 @@ app.get('/__internal__/observability/metrics', (c) => { // Main entry point async function main() { logger.info('Starting firehose-service'); - logger.info(`Mode: ${config.isBackfill ? 'backfill' : 'firehose'}`); + logger.info(`Mode: ${config.isDbFillOnly ? 'db-fill-only' : (config.isBackfill ? 'backfill' : 'firehose')}`); logger.info(`S3 Bucket: ${config.s3Bucket || '(disk fallback)'}`); // Start health server diff --git a/apps/firehose-service/src/lib/cache-writer.ts b/apps/firehose-service/src/lib/cache-writer.ts index c4dc2cb..dc4fa1e 100644 --- a/apps/firehose-service/src/lib/cache-writer.ts +++ b/apps/firehose-service/src/lib/cache-writer.ts @@ -49,6 +49,62 @@ export async function fetchSiteRecord(did: string, rkey: string): Promise<{ reco } } +/** + * List all place.wisp.fs records for a DID. + */ +export async function listSiteRecordsForDid( + did: string +): Promise> { + const pdsEndpoint = await getPdsForDid(did); + if (!pdsEndpoint) { + logger.error('Failed to get PDS endpoint for DID (listRecords)', undefined, { did }); + return []; + } + + const records: Array<{ rkey: string; record: WispFsRecord; cid: string }> = []; + let cursor: string | undefined; + + while (true) { + const params = new URLSearchParams({ + repo: did, + collection: 'place.wisp.fs', + limit: '100', + }); + if (cursor) params.set('cursor', cursor); + + const url = `${pdsEndpoint}/xrpc/com.atproto.repo.listRecords?${params.toString()}`; + const data = await safeFetchJson(url) as { + records?: Array<{ uri?: string; value?: unknown; cid?: string }>; + cursor?: string; + }; + + const pageRecords = Array.isArray(data.records) ? data.records : []; + for (const row of pageRecords) { + const uri = row.uri; + const record = row.value as WispFsRecord | undefined; + if (!uri || !record || record.$type !== 'place.wisp.fs') continue; + + const uriParts = uri.split('/'); + const rkey = uriParts[uriParts.length - 1]; + if (!rkey) continue; + + records.push({ + rkey, + record, + cid: row.cid || '', + }); + } + + const nextCursor = typeof data.cursor === 'string' && data.cursor.length > 0 + ? data.cursor + : undefined; + if (!nextCursor) break; + cursor = nextCursor; + } + + return records; +} + /** * Fetch a settings record from the PDS */ diff --git a/apps/firehose-service/src/lib/db.ts b/apps/firehose-service/src/lib/db.ts index ea7556f..f43b554 100644 --- a/apps/firehose-service/src/lib/db.ts +++ b/apps/firehose-service/src/lib/db.ts @@ -49,6 +49,79 @@ export async function listAllSites(): Promise { `; } +/** + * List all known DIDs from all DID-bearing tables. + * Missing tables are skipped to keep bootstrapping resilient. + */ +export async function listAllKnownDids(): Promise { + const sources: Array<{ name: string; fetch: () => Promise> }> = [ + { + name: 'sites', + fetch: () => sql>` + SELECT DISTINCT did + FROM sites + WHERE did IS NOT NULL AND did <> '' + `, + }, + { + name: 'site_cache', + fetch: () => sql>` + SELECT DISTINCT did + FROM site_cache + WHERE did IS NOT NULL AND did <> '' + `, + }, + { + name: 'site_settings_cache', + fetch: () => sql>` + SELECT DISTINCT did + FROM site_settings_cache + WHERE did IS NOT NULL AND did <> '' + `, + }, + { + name: 'domains', + fetch: () => sql>` + SELECT DISTINCT did + FROM domains + WHERE did IS NOT NULL AND did <> '' + `, + }, + { + name: 'custom_domains', + fetch: () => sql>` + SELECT DISTINCT did + FROM custom_domains + WHERE did IS NOT NULL AND did <> '' + `, + }, + { + name: 'supporter', + fetch: () => sql>` + SELECT DISTINCT did + FROM supporter + WHERE did IS NOT NULL AND did <> '' + `, + }, + ]; + const dids = new Set(); + + for (const source of sources) { + try { + const rows = await source.fetch(); + for (const row of rows) { + if (typeof row.did === 'string' && row.did.length > 0) { + dids.add(row.did); + } + } + } catch { + logger.warn(`[DB] Skipping DID source table ${source.name}`); + } + } + + return [...dids].sort(); +} + // Write functions export async function upsertSiteCache( diff --git a/packages/@wispplace/constants/src/index.ts b/packages/@wispplace/constants/src/index.ts index f7cac6a..99939ba 100644 --- a/packages/@wispplace/constants/src/index.ts +++ b/packages/@wispplace/constants/src/index.ts @@ -82,9 +82,9 @@ export const DEFAULT_IGNORE_PATTERNS: string[] = [ // CLI Binary URLs export const CLI_BINARY_BASE_URL = "https://sites.wisp.place/nekomimi.pet/wisp-cli-binaries"; export const CLI_BINARIES = { - "darwin-universal": `${CLI_BINARY_BASE_URL}/wisp-cli-darwin-universal`, + "darwin-universal": `${CLI_BINARY_BASE_URL}/wisp-cli-universal-darwin`, "darwin-arm64": `${CLI_BINARY_BASE_URL}/wisp-cli-aarch64-darwin`, - "darwin-x86_64": `${CLI_BINARY_BASE_URL}/wisp-cli-darwin-x86_64`, + "darwin-x86_64": `${CLI_BINARY_BASE_URL}/wisp-cli-x86_64-darwin`, "linux-arm64": `${CLI_BINARY_BASE_URL}/wisp-cli-aarch64-linux`, "linux-x86_64": `${CLI_BINARY_BASE_URL}/wisp-cli-x86_64-linux`, } as const; -- 2.54.0