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

+5 -1

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

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

+8 -8

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

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

+11 -3

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

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

+2 -1

apps/hosting-service/.env.example

··· 3 3 4 4 # Server 5 5 PORT=3001 6 6 - BASE_HOST=wisp.place 6 6 + # Base domain (e.g., "localhost" for sites.localhost, "wisp.place" for sites.wisp.place) 7 7 + BASE_HOST=localhost 7 8 8 9 # Redis (cache invalidation + revalidation queue) 9 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 21 return; 22 22 } 23 23 24 24 + console.log(`[CacheInvalidation] Connecting to Redis for subscribing: ${redisUrl}`); 24 25 subscriber = new Redis(redisUrl, { 25 26 maxRetriesPerRequest: 2, 26 27 enableReadyCheck: true, ··· 28 29 29 30 subscriber.on('error', (err) => { 30 31 console.error('[CacheInvalidation] Redis error:', err); 32 32 + }); 33 33 + 34 34 + subscriber.on('ready', () => { 35 35 + console.log('[CacheInvalidation] Redis subscriber connected'); 31 36 }); 32 37 33 38 subscriber.subscribe(CHANNEL, (err) => {

+1 -2

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

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

+22 -4

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

··· 158 158 */ 159 159 async function ensureSiteCached(did: string, rkey: string): Promise<void> { 160 160 const existing = await getSiteCache(did, rkey); 161 161 - if (existing) return; // Site is known, proceed normally 161 161 + if (existing) { 162 162 + // Site is in DB — check if any files actually exist in storage 163 163 + const prefix = `${did}/${rkey}/`; 164 164 + const hasFiles = await storage.exists(prefix.slice(0, -1)) || 165 165 + await checkAnyFileExists(did, rkey, existing.file_cids); 166 166 + if (hasFiles) { 167 167 + return; 168 168 + } 169 169 + console.log(`[FileServing] Site ${did}/${rkey} in DB but no files in storage, re-fetching`); 170 170 + } else { 171 171 + console.log(`[FileServing] Site ${did}/${rkey} not in DB, attempting on-demand cache`); 172 172 + } 162 173 163 163 - // Site is completely unknown — try on-demand fetch 164 164 - console.log(`[FileServing] Site ${did}/${rkey} not in DB, attempting on-demand cache`); 165 165 - await fetchAndCacheSite(did, rkey); 174 174 + const success = await fetchAndCacheSite(did, rkey); 175 175 + console.log(`[FileServing] On-demand cache for ${did}/${rkey}: ${success ? 'success' : 'failed'}`); 176 176 + } 177 177 + 178 178 + async function checkAnyFileExists(did: string, rkey: string, fileCids: unknown): Promise<boolean> { 179 179 + if (!fileCids || typeof fileCids !== 'object') return false; 180 180 + const cids = fileCids as Record<string, string>; 181 181 + const firstFile = Object.keys(cids)[0]; 182 182 + if (!firstFile) return false; 183 183 + return storage.exists(`${did}/${rkey}/${firstFile}`); 166 184 } 167 185 168 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 18 } 19 19 20 20 if (!client) { 21 21 + console.log(`[Revalidate] Connecting to Redis: ${redisUrl}`); 21 22 client = new Redis(redisUrl, { 22 23 maxRetriesPerRequest: 2, 23 24 enableReadyCheck: true, ··· 26 27 client.on('error', (err) => { 27 28 console.error('[Revalidate] Redis error:', err); 28 29 }); 30 30 + 31 31 + client.on('ready', () => { 32 32 + console.log(`[Revalidate] Redis connected, stream: ${streamName}`); 33 33 + }); 29 34 } 30 35 31 36 return client; ··· 65 70 Date.now().toString() 66 71 ); 67 72 73 73 + console.log(`[Revalidate] Enqueued ${did}/${rkey} (${reason}) to ${streamName}`); 68 74 recordRevalidateResult('enqueued'); 69 75 return { enqueued: true, result: 'enqueued' }; 70 76 } catch (err) {

+9 -1

apps/hosting-service/src/server.ts

··· 14 14 import { serveFromCache, serveFromCacheWithRewrite } from './lib/file-serving'; 15 15 import { getRevalidateMetrics } from './lib/revalidate-metrics'; 16 16 17 17 - const BASE_HOST = process.env.BASE_HOST || 'wisp.place'; 17 17 + const BASE_HOST_ENV = process.env.BASE_HOST || 'wisp.place'; 18 18 + const BASE_HOST = BASE_HOST_ENV.split(':')[0] || BASE_HOST_ENV; 18 19 19 20 const app = new Hono(); 20 21 ··· 42 43 const rawPath = url.pathname.replace(/^\//, ''); 43 44 const path = sanitizePath(rawPath); 44 45 46 46 + console.log(`[Server] Request: host=${hostname} hostnameWithoutPort=${hostnameWithoutPort} path=${path} BASE_HOST=${BASE_HOST}`); 47 47 + 45 48 // Check if this is sites.wisp.place subdomain (strip port for comparison) 46 49 if (hostnameWithoutPort === `sites.${BASE_HOST}`) { 47 50 ··· 76 79 const did = await resolveDid(identifier); 77 80 if (!did) { 78 81 return c.text('Invalid identifier', 400); 82 82 + } 83 83 + 84 84 + // Redirect to trailing slash when accessing site root so relative paths resolve correctly 85 85 + if (!filePath && !url.pathname.endsWith('/')) { 86 86 + return c.redirect(`${url.pathname}/${url.search}`, 301); 79 87 } 80 88 81 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 59 60 60 constructor(private tier: StorageTier) {} 61 61 62 62 - // Read operations - pass through to underlying tier 62 62 + // Read operations - pass through to underlying tier, catch errors as cache misses 63 63 async get(key: string) { 64 64 - return this.tier.get(key); 64 64 + try { 65 65 + return await this.tier.get(key); 66 66 + } catch (err) { 67 67 + this.logReadError('get', key, err); 68 68 + return null; 69 69 + } 65 70 } 66 71 67 72 async getWithMetadata(key: string) { 68 68 - return this.tier.getWithMetadata?.(key) ?? null; 73 73 + try { 74 74 + return await this.tier.getWithMetadata?.(key) ?? null; 75 75 + } catch (err) { 76 76 + this.logReadError('getWithMetadata', key, err); 77 77 + return null; 78 78 + } 69 79 } 70 80 71 81 async getStream(key: string) { 72 72 - return this.tier.getStream?.(key) ?? null; 82 82 + try { 83 83 + return await this.tier.getStream?.(key) ?? null; 84 84 + } catch (err) { 85 85 + this.logReadError('getStream', key, err); 86 86 + return null; 87 87 + } 73 88 } 74 89 75 90 async exists(key: string) { 76 76 - return this.tier.exists(key); 91 91 + try { 92 92 + return await this.tier.exists(key); 93 93 + } catch (err) { 94 94 + this.logReadError('exists', key, err); 95 95 + return false; 96 96 + } 77 97 } 78 98 79 99 async getMetadata(key: string) { 80 80 - return this.tier.getMetadata(key); 100 100 + try { 101 101 + return await this.tier.getMetadata(key); 102 102 + } catch (err) { 103 103 + this.logReadError('getMetadata', key, err); 104 104 + return null; 105 105 + } 81 106 } 82 107 83 108 async *listKeys(prefix?: string) { ··· 111 136 112 137 async clear() { 113 138 this.logWriteSkip('clear', 'all keys'); 139 139 + } 140 140 + 141 141 + private logReadError(operation: string, key: string, err: unknown) { 142 142 + const msg = err instanceof Error ? err.message : String(err); 143 143 + console.warn(`[Storage] S3 read error (${operation}) for ${key}: ${msg}`); 114 144 } 115 145 116 146 private logWriteSkip(operation: string, key: string) {