hosting service writes on cache miss, firehose service properly notifies hosting service on new updates #6

apps/firehose-service/.env.example

This file has not been changed.

apps/firehose-service/src/index.ts

This file has not been changed.

+6

apps/firehose-service/src/lib/cache-invalidation.ts

··· 24 } 25 26 if (!publisher) { 0 27 publisher = new Redis(config.redisUrl, { 28 maxRetriesPerRequest: 2, 29 enableReadyCheck: true, ··· 32 publisher.on('error', (err) => { 33 console.error('[CacheInvalidation] Redis error:', err); 34 }); 0 0 0 0 35 } 36 37 return publisher; ··· 47 48 try { 49 const message = JSON.stringify({ did, rkey, action }); 0 50 await redis.publish(CHANNEL, message); 51 } catch (err) { 52 console.error('[CacheInvalidation] Failed to publish:', err);

··· 24 } 25 26 if (!publisher) { 27 + console.log(`[CacheInvalidation] Connecting to Redis for publishing: ${config.redisUrl}`); 28 publisher = new Redis(config.redisUrl, { 29 maxRetriesPerRequest: 2, 30 enableReadyCheck: true, ··· 33 publisher.on('error', (err) => { 34 console.error('[CacheInvalidation] Redis error:', err); 35 }); 36 + 37 + publisher.on('ready', () => { 38 + console.log('[CacheInvalidation] Redis publisher connected'); 39 + }); 40 } 41 42 return publisher; ··· 52 53 try { 54 const message = JSON.stringify({ did, rkey, action }); 55 + console.log(`[CacheInvalidation] Publishing ${action} for ${did}/${rkey} to ${CHANNEL}`); 56 await redis.publish(CHANNEL, message); 57 } catch (err) { 58 console.error('[CacheInvalidation] Failed to publish:', err);

+5 -1

apps/firehose-service/src/lib/cache-writer.ts

··· 439 recordCid: string, 440 options?: { 441 forceRewriteHtml?: boolean; 0 442 } 443 ): Promise<void> { 444 const forceRewriteHtml = options?.forceRewriteHtml === true; ··· 551 } 552 553 // Notify hosting-service to invalidate its local caches 554 - await publishCacheInvalidation(did, rkey, 'update'); 0 0 0 555 556 console.log(`[Cache] Successfully cached site ${did}/${rkey}`); 557 }

··· 439 recordCid: string, 440 options?: { 441 forceRewriteHtml?: boolean; 442 + skipInvalidation?: boolean; 443 } 444 ): Promise<void> { 445 const forceRewriteHtml = options?.forceRewriteHtml === true; ··· 552 } 553 554 // Notify hosting-service to invalidate its local caches 555 + // (skip for revalidate/backfill since hosting-service already has the files locally) 556 + if (!options?.skipInvalidation) { 557 + await publishCacheInvalidation(did, rkey, 'update'); 558 + } 559 560 console.log(`[Cache] Successfully cached site ${did}/${rkey}`); 561 }

+8 -8

apps/firehose-service/src/lib/db.ts

··· 54 recordCid: string, 55 fileCids: Record<string, string> 56 ): Promise<void> { 57 - const fileCidsJson = JSON.stringify(fileCids ?? {}); 58 console.log(`[DB] upsertSiteCache starting for ${did}/${rkey}`); 59 try { 60 await sql` 61 INSERT INTO site_cache (did, rkey, record_cid, file_cids, cached_at, updated_at) 62 - VALUES (${did}, ${rkey}, ${recordCid}, ${fileCidsJson}::jsonb, EXTRACT(EPOCH FROM NOW()), EXTRACT(EPOCH FROM NOW())) 63 ON CONFLICT (did, rkey) 64 DO UPDATE SET 65 record_cid = EXCLUDED.record_cid, ··· 94 const directoryListing = settings.directoryListing ?? false; 95 const spaMode = settings.spaMode ?? null; 96 const custom404 = settings.custom404 ?? null; 97 - const indexFilesJson = JSON.stringify(settings.indexFiles ?? []); 98 const cleanUrls = settings.cleanUrls ?? true; 99 - const headersJson = JSON.stringify(settings.headers ?? []); 100 0 0 0 101 console.log(`[DB] upsertSiteSettingsCache starting for ${did}/${rkey}`, { 102 directoryListing, 103 spaMode, 104 custom404, 105 - indexFiles: indexFilesJson, 106 cleanUrls, 107 - headers: headersJson, 108 }); 109 110 try { ··· 117 ${directoryListing}, 118 ${spaMode}, 119 ${custom404}, 120 - ${indexFilesJson}::jsonb, 121 ${cleanUrls}, 122 - ${headersJson}::jsonb, 123 EXTRACT(EPOCH FROM NOW()), 124 EXTRACT(EPOCH FROM NOW()) 125 )

··· 54 recordCid: string, 55 fileCids: Record<string, string> 56 ): Promise<void> { 0 57 console.log(`[DB] upsertSiteCache starting for ${did}/${rkey}`); 58 try { 59 await sql` 60 INSERT INTO site_cache (did, rkey, record_cid, file_cids, cached_at, updated_at) 61 + VALUES (${did}, ${rkey}, ${recordCid}, ${sql.json(fileCids ?? {})}, EXTRACT(EPOCH FROM NOW()), EXTRACT(EPOCH FROM NOW())) 62 ON CONFLICT (did, rkey) 63 DO UPDATE SET 64 record_cid = EXCLUDED.record_cid, ··· 93 const directoryListing = settings.directoryListing ?? false; 94 const spaMode = settings.spaMode ?? null; 95 const custom404 = settings.custom404 ?? null; 0 96 const cleanUrls = settings.cleanUrls ?? true; 0 97 98 + const indexFiles = settings.indexFiles ?? []; 99 + const headers = settings.headers ?? []; 100 + 101 console.log(`[DB] upsertSiteSettingsCache starting for ${did}/${rkey}`, { 102 directoryListing, 103 spaMode, 104 custom404, 105 + indexFiles, 106 cleanUrls, 107 + headers, 108 }); 109 110 try { ··· 117 ${directoryListing}, 118 ${spaMode}, 119 ${custom404}, 120 + ${sql.json(indexFiles)}, 121 ${cleanUrls}, 122 + ${sql.json(headers)}, 123 EXTRACT(EPOCH FROM NOW()), 124 EXTRACT(EPOCH FROM NOW()) 125 )

+11 -3

apps/firehose-service/src/lib/revalidate-worker.ts

··· 38 return; 39 } 40 41 - console.log('[Revalidate] Processing', { did, rkey, reason, id }); 42 43 const record = await fetchSiteRecord(did, rkey); 44 if (!record) { 45 - console.warn('[Revalidate] Site record not found', { did, rkey }); 46 await redis.xack(config.revalidateStream, config.revalidateGroup, id); 47 return; 48 } 49 50 - await handleSiteCreateOrUpdate(did, rkey, record.record, record.cid); 0 0 51 0 52 await redis.xack(config.revalidateStream, config.revalidateGroup, id); 53 } 54 ··· 155 156 if (running) return; 157 0 158 redis = new Redis(config.redisUrl, { 159 maxRetriesPerRequest: 2, 160 enableReadyCheck: true, ··· 162 163 redis.on('error', (err) => { 164 console.error('[Revalidate] Redis error:', err); 0 0 0 0 165 }); 166 167 running = true;

··· 38 return; 39 } 40 41 + console.log(`[Revalidate] Received message ${id}: ${did}/${rkey} (${reason})`); 42 43 const record = await fetchSiteRecord(did, rkey); 44 if (!record) { 45 + console.warn(`[Revalidate] Site record not found on PDS: ${did}/${rkey}`); 46 await redis.xack(config.revalidateStream, config.revalidateGroup, id); 47 return; 48 } 49 50 + await handleSiteCreateOrUpdate(did, rkey, record.record, record.cid, { 51 + skipInvalidation: true, 52 + }); 53 54 + console.log(`[Revalidate] Completed ${id}: ${did}/${rkey}`); 55 await redis.xack(config.revalidateStream, config.revalidateGroup, id); 56 } 57 ··· 158 159 if (running) return; 160 161 + console.log(`[Revalidate] Connecting to Redis: ${config.redisUrl}`); 162 redis = new Redis(config.redisUrl, { 163 maxRetriesPerRequest: 2, 164 enableReadyCheck: true, ··· 166 167 redis.on('error', (err) => { 168 console.error('[Revalidate] Redis error:', err); 169 + }); 170 + 171 + redis.on('ready', () => { 172 + console.log(`[Revalidate] Redis connected, stream: ${config.revalidateStream}, group: ${config.revalidateGroup}`); 173 }); 174 175 running = true;

+2 -1

apps/hosting-service/.env.example

··· 3 4 # Server 5 PORT=3001 6 - BASE_HOST=wisp.place 0 7 8 # Redis (cache invalidation + revalidation queue) 9 REDIS_URL=redis://localhost:6379

··· 3 4 # Server 5 PORT=3001 6 + # Base domain (e.g., "localhost" for sites.localhost, "wisp.place" for sites.wisp.place) 7 + BASE_HOST=localhost 8 9 # Redis (cache invalidation + revalidation queue) 10 REDIS_URL=redis://localhost:6379

apps/hosting-service/src/index.ts

This file has not been changed.

+5

apps/hosting-service/src/lib/cache-invalidation.ts

··· 21 return; 22 } 23 0 24 subscriber = new Redis(redisUrl, { 25 maxRetriesPerRequest: 2, 26 enableReadyCheck: true, ··· 28 29 subscriber.on('error', (err) => { 30 console.error('[CacheInvalidation] Redis error:', err); 0 0 0 0 31 }); 32 33 subscriber.subscribe(CHANNEL, (err) => {

··· 21 return; 22 } 23 24 + console.log(`[CacheInvalidation] Connecting to Redis for subscribing: ${redisUrl}`); 25 subscriber = new Redis(redisUrl, { 26 maxRetriesPerRequest: 2, 27 enableReadyCheck: true, ··· 29 30 subscriber.on('error', (err) => { 31 console.error('[CacheInvalidation] Redis error:', err); 32 + }); 33 + 34 + subscriber.on('ready', () => { 35 + console.log('[CacheInvalidation] Redis subscriber connected'); 36 }); 37 38 subscriber.subscribe(CHANNEL, (err) => {

+1 -2

apps/hosting-service/src/lib/db.ts

··· 128 recordCid: string, 129 fileCids: Record<string, string> 130 ): Promise<void> { 131 - const fileCidsJson = JSON.stringify(fileCids ?? {}); 132 try { 133 await sql` 134 INSERT INTO site_cache (did, rkey, record_cid, file_cids, cached_at, updated_at) 135 - VALUES (${did}, ${rkey}, ${recordCid}, ${fileCidsJson}::jsonb, EXTRACT(EPOCH FROM NOW()), EXTRACT(EPOCH FROM NOW())) 136 ON CONFLICT (did, rkey) 137 DO UPDATE SET 138 record_cid = EXCLUDED.record_cid,

··· 128 recordCid: string, 129 fileCids: Record<string, string> 130 ): Promise<void> { 0 131 try { 132 await sql` 133 INSERT INTO site_cache (did, rkey, record_cid, file_cids, cached_at, updated_at) 134 + VALUES (${did}, ${rkey}, ${recordCid}, ${sql.json(fileCids ?? {})}, EXTRACT(EPOCH FROM NOW()), EXTRACT(EPOCH FROM NOW())) 135 ON CONFLICT (did, rkey) 136 DO UPDATE SET 137 record_cid = EXCLUDED.record_cid,

+22 -4

apps/hosting-service/src/lib/file-serving.ts

··· 158 */ 159 async function ensureSiteCached(did: string, rkey: string): Promise<void> { 160 const existing = await getSiteCache(did, rkey); 161 - if (existing) return; // Site is known, proceed normally 0 0 0 0 0 0 0 0 0 0 0 162 163 - // Site is completely unknown — try on-demand fetch 164 - console.log(`[FileServing] Site ${did}/${rkey} not in DB, attempting on-demand cache`); 165 - await fetchAndCacheSite(did, rkey); 0 0 0 0 0 0 0 166 } 167 168 /**

··· 158 */ 159 async function ensureSiteCached(did: string, rkey: string): Promise<void> { 160 const existing = await getSiteCache(did, rkey); 161 + if (existing) { 162 + // Site is in DB — check if any files actually exist in storage 163 + const prefix = `${did}/${rkey}/`; 164 + const hasFiles = await storage.exists(prefix.slice(0, -1)) || 165 + await checkAnyFileExists(did, rkey, existing.file_cids); 166 + if (hasFiles) { 167 + return; 168 + } 169 + console.log(`[FileServing] Site ${did}/${rkey} in DB but no files in storage, re-fetching`); 170 + } else { 171 + console.log(`[FileServing] Site ${did}/${rkey} not in DB, attempting on-demand cache`); 172 + } 173 174 + const success = await fetchAndCacheSite(did, rkey); 175 + console.log(`[FileServing] On-demand cache for ${did}/${rkey}: ${success ? 'success' : 'failed'}`); 176 + } 177 + 178 + async function checkAnyFileExists(did: string, rkey: string, fileCids: unknown): Promise<boolean> { 179 + if (!fileCids || typeof fileCids !== 'object') return false; 180 + const cids = fileCids as Record<string, string>; 181 + const firstFile = Object.keys(cids)[0]; 182 + if (!firstFile) return false; 183 + return storage.exists(`${did}/${rkey}/${firstFile}`); 184 } 185 186 /**

apps/hosting-service/src/lib/on-demand-cache.ts

This file has not been changed.

+6

apps/hosting-service/src/lib/revalidate-queue.ts

··· 18 } 19 20 if (!client) { 0 21 client = new Redis(redisUrl, { 22 maxRetriesPerRequest: 2, 23 enableReadyCheck: true, ··· 26 client.on('error', (err) => { 27 console.error('[Revalidate] Redis error:', err); 28 }); 0 0 0 0 29 } 30 31 return client; ··· 65 Date.now().toString() 66 ); 67 0 68 recordRevalidateResult('enqueued'); 69 return { enqueued: true, result: 'enqueued' }; 70 } catch (err) {

··· 18 } 19 20 if (!client) { 21 + console.log(`[Revalidate] Connecting to Redis: ${redisUrl}`); 22 client = new Redis(redisUrl, { 23 maxRetriesPerRequest: 2, 24 enableReadyCheck: true, ··· 27 client.on('error', (err) => { 28 console.error('[Revalidate] Redis error:', err); 29 }); 30 + 31 + client.on('ready', () => { 32 + console.log(`[Revalidate] Redis connected, stream: ${streamName}`); 33 + }); 34 } 35 36 return client; ··· 70 Date.now().toString() 71 ); 72 73 + console.log(`[Revalidate] Enqueued ${did}/${rkey} (${reason}) to ${streamName}`); 74 recordRevalidateResult('enqueued'); 75 return { enqueued: true, result: 'enqueued' }; 76 } catch (err) {

+9 -1

apps/hosting-service/src/server.ts

··· 14 import { serveFromCache, serveFromCacheWithRewrite } from './lib/file-serving'; 15 import { getRevalidateMetrics } from './lib/revalidate-metrics'; 16 17 - const BASE_HOST = process.env.BASE_HOST || 'wisp.place'; 0 18 19 const app = new Hono(); 20 ··· 42 const rawPath = url.pathname.replace(/^\//, ''); 43 const path = sanitizePath(rawPath); 44 0 0 45 // Check if this is sites.wisp.place subdomain (strip port for comparison) 46 if (hostnameWithoutPort === `sites.${BASE_HOST}`) { 47 ··· 76 const did = await resolveDid(identifier); 77 if (!did) { 78 return c.text('Invalid identifier', 400); 0 0 0 0 0 79 } 80 81 console.log(`[Server] sites.wisp.place request: identifier=${identifier}, site=${site}, filePath=${filePath}`);

··· 14 import { serveFromCache, serveFromCacheWithRewrite } from './lib/file-serving'; 15 import { getRevalidateMetrics } from './lib/revalidate-metrics'; 16 17 + const BASE_HOST_ENV = process.env.BASE_HOST || 'wisp.place'; 18 + const BASE_HOST = BASE_HOST_ENV.split(':')[0] || BASE_HOST_ENV; 19 20 const app = new Hono(); 21 ··· 43 const rawPath = url.pathname.replace(/^\//, ''); 44 const path = sanitizePath(rawPath); 45 46 + console.log(`[Server] Request: host=${hostname} hostnameWithoutPort=${hostnameWithoutPort} path=${path} BASE_HOST=${BASE_HOST}`); 47 + 48 // Check if this is sites.wisp.place subdomain (strip port for comparison) 49 if (hostnameWithoutPort === `sites.${BASE_HOST}`) { 50 ··· 79 const did = await resolveDid(identifier); 80 if (!did) { 81 return c.text('Invalid identifier', 400); 82 + } 83 + 84 + // Redirect to trailing slash when accessing site root so relative paths resolve correctly 85 + if (!filePath && !url.pathname.endsWith('/')) { 86 + return c.redirect(`${url.pathname}/${url.search}`, 301); 87 } 88 89 console.log(`[Server] sites.wisp.place request: identifier=${identifier}, site=${site}, filePath=${filePath}`);

docker-compose.yml

This file has not been changed.

+36 -6

apps/hosting-service/src/lib/storage.ts

··· 59 60 constructor(private tier: StorageTier) {} 61 62 - // Read operations - pass through to underlying tier 63 async get(key: string) { 64 - return this.tier.get(key); 0 0 0 0 0 65 } 66 67 async getWithMetadata(key: string) { 68 - return this.tier.getWithMetadata?.(key) ?? null; 0 0 0 0 0 69 } 70 71 async getStream(key: string) { 72 - return this.tier.getStream?.(key) ?? null; 0 0 0 0 0 73 } 74 75 async exists(key: string) { 76 - return this.tier.exists(key); 0 0 0 0 0 77 } 78 79 async getMetadata(key: string) { 80 - return this.tier.getMetadata(key); 0 0 0 0 0 81 } 82 83 async *listKeys(prefix?: string) { ··· 111 112 async clear() { 113 this.logWriteSkip('clear', 'all keys'); 0 0 0 0 0 114 } 115 116 private logWriteSkip(operation: string, key: string) {

··· 59 60 constructor(private tier: StorageTier) {} 61 62 + // Read operations - pass through to underlying tier, catch errors as cache misses 63 async get(key: string) { 64 + try { 65 + return await this.tier.get(key); 66 + } catch (err) { 67 + this.logReadError('get', key, err); 68 + return null; 69 + } 70 } 71 72 async getWithMetadata(key: string) { 73 + try { 74 + return await this.tier.getWithMetadata?.(key) ?? null; 75 + } catch (err) { 76 + this.logReadError('getWithMetadata', key, err); 77 + return null; 78 + } 79 } 80 81 async getStream(key: string) { 82 + try { 83 + return await this.tier.getStream?.(key) ?? null; 84 + } catch (err) { 85 + this.logReadError('getStream', key, err); 86 + return null; 87 + } 88 } 89 90 async exists(key: string) { 91 + try { 92 + return await this.tier.exists(key); 93 + } catch (err) { 94 + this.logReadError('exists', key, err); 95 + return false; 96 + } 97 } 98 99 async getMetadata(key: string) { 100 + try { 101 + return await this.tier.getMetadata(key); 102 + } catch (err) { 103 + this.logReadError('getMetadata', key, err); 104 + return null; 105 + } 106 } 107 108 async *listKeys(prefix?: string) { ··· 136 137 async clear() { 138 this.logWriteSkip('clear', 'all keys'); 139 + } 140 + 141 + private logReadError(operation: string, key: string, err: unknown) { 142 + const msg = err instanceof Error ? err.message : String(err); 143 + console.warn(`[Storage] S3 read error (${operation}) for ${key}: ${msg}`); 144 } 145 146 private logWriteSkip(operation: string, key: string) {