nekomimi.pet
Monorepo for wisp.place. A static site hosting service built on top of the AT Protocol.
wisp.place
4.6 kB
150 lines
1/**
2 * Firehose Service - Ingests AT Protocol firehose events and caches sites to S3
3 *
4 * Modes:
5 * - Normal: Watch firehose for place.wisp.fs events
6 * - Backfill: Process existing sites from database
7 */
8
9import { Hono } from 'hono';
10import { serve } from '@hono/node-server';
11import { config } from './config';
12import { startFirehose, stopFirehose, getFirehoseHealth } from './lib/firehose';
13import { closeDatabase, listAllSiteCaches, listAllSites, getSiteCache } from './lib/db';
14import { storage } from './lib/storage';
15import { handleSiteCreateOrUpdate, fetchSiteRecord } from './lib/cache-writer';
16import { startRevalidateWorker, stopRevalidateWorker } from './lib/revalidate-worker';
17import { closeCacheInvalidationPublisher } from './lib/cache-invalidation';
18
19const app = new Hono();
20
21// Health endpoint
22app.get('/health', async (c) => {
23 const firehoseHealth = getFirehoseHealth();
24 const storageStats = await storage.getStats();
25
26 return c.json({
27 status: firehoseHealth.healthy ? 'healthy' : 'degraded',
28 mode: config.isBackfill ? 'backfill' : 'firehose',
29 firehose: firehoseHealth,
30 storage: storageStats,
31 });
32});
33
34// Graceful shutdown
35let isShuttingDown = false;
36
37async function shutdown(signal: string) {
38 if (isShuttingDown) return;
39 isShuttingDown = true;
40
41 console.log(`\n[Service] Received ${signal}, shutting down...`);
42
43 stopFirehose();
44 await stopRevalidateWorker();
45 await closeCacheInvalidationPublisher();
46 await closeDatabase();
47
48 console.log('[Service] Shutdown complete');
49 process.exit(0);
50}
51
52process.on('SIGINT', () => shutdown('SIGINT'));
53process.on('SIGTERM', () => shutdown('SIGTERM'));
54
55/**
56 * Backfill mode - process existing sites from database
57 */
58async function runBackfill(): Promise<void> {
59 console.log('[Backfill] Starting backfill mode');
60 const startTime = Date.now();
61 const forceRewriteHtml = process.env.BACKFILL_FORCE_REWRITE_HTML === 'true';
62
63 if (forceRewriteHtml) {
64 console.log('[Backfill] Forcing HTML rewrite for all sites');
65 }
66
67 let sites = await listAllSites();
68 if (sites.length === 0) {
69 const cachedSites = await listAllSiteCaches();
70 sites = cachedSites.map(site => ({ did: site.did, rkey: site.rkey }));
71 console.log('[Backfill] Sites table empty; falling back to site_cache entries');
72 }
73
74 console.log(`[Backfill] Found ${sites.length} sites in database`);
75
76 let processed = 0;
77 let skipped = 0;
78 let failed = 0;
79
80 for (const site of sites) {
81 try {
82 // Fetch current record from PDS
83 const result = await fetchSiteRecord(site.did, site.rkey);
84
85 if (!result) {
86 console.log(`[Backfill] Site not found on PDS: ${site.did}/${site.rkey}`);
87 skipped++;
88 continue;
89 }
90
91 const existingCache = await getSiteCache(site.did, site.rkey);
92 // Check if CID matches (already up to date)
93 if (!forceRewriteHtml && existingCache && result.cid === existingCache.record_cid) {
94 console.log(`[Backfill] Site already up to date: ${site.did}/${site.rkey}`);
95 skipped++;
96 continue;
97 }
98
99 // Process the site
100 await handleSiteCreateOrUpdate(site.did, site.rkey, result.record, result.cid, {
101 forceRewriteHtml,
102 });
103 processed++;
104
105 console.log(`[Backfill] Progress: ${processed + skipped + failed}/${sites.length}`);
106 } catch (err) {
107 console.error(`[Backfill] Failed to process ${site.did}/${site.rkey}:`, err);
108 failed++;
109 }
110 }
111
112 const elapsedMs = Date.now() - startTime;
113 const elapsedSec = Math.round(elapsedMs / 1000);
114 const elapsedMin = Math.floor(elapsedSec / 60);
115 const elapsedRemSec = elapsedSec % 60;
116 const elapsedLabel = elapsedMin > 0 ? `${elapsedMin}m ${elapsedRemSec}s` : `${elapsedSec}s`;
117
118 console.log(`[Backfill] Complete: ${processed} processed, ${skipped} skipped, ${failed} failed (${elapsedLabel} elapsed)`);
119}
120
121// Main entry point
122async function main() {
123 console.log('[Service] Starting firehose-service');
124 console.log(`[Service] Mode: ${config.isBackfill ? 'backfill' : 'firehose'}`);
125 console.log(`[Service] S3 Bucket: ${config.s3Bucket || '(disk fallback)'}`);
126
127 // Start health server
128 const server = serve({
129 fetch: app.fetch,
130 port: config.healthPort,
131 });
132
133 console.log(`[Service] Health endpoint: http://localhost:${config.healthPort}/health`);
134
135 if (config.isBackfill) {
136 // Run backfill and exit
137 await runBackfill();
138 await closeDatabase();
139 process.exit(0);
140 } else {
141 // Start firehose
142 startFirehose();
143 await startRevalidateWorker();
144 }
145}
146
147main().catch((err) => {
148 console.error('[Service] Fatal error:', err);
149 process.exit(1);
150});