+27

apps/firehose-service/.env.example

··· 1 1 + # Database 2 2 + DATABASE_URL=postgres://postgres:postgres@localhost:5432/wisp 3 3 + 4 4 + # Firehose 5 5 + FIREHOSE_SERVICE=wss://bsky.network 6 6 + FIREHOSE_MAX_CONCURRENCY=5 7 7 + 8 8 + # Redis (cache invalidation + revalidation queue) 9 9 + REDIS_URL=redis://localhost:6379 10 10 + 11 11 + # S3 Storage (leave empty for local disk fallback) 12 12 + S3_BUCKET= 13 13 + S3_METADATA_BUCKET= 14 14 + S3_REGION=auto 15 15 + S3_ENDPOINT= 16 16 + S3_PREFIX=sites/ 17 17 + S3_FORCE_PATH_STYLE=true 18 18 + 19 19 + # AWS Credentials (required if using S3) 20 20 + AWS_ACCESS_KEY_ID= 21 21 + AWS_SECRET_ACCESS_KEY= 22 22 + 23 23 + # Health check server 24 24 + HEALTH_PORT=3001 25 25 + 26 26 + # For local disk fallback (when S3_BUCKET is empty) 27 27 + CACHE_DIR=./cache/sites

+2

apps/firehose-service/src/index.ts

··· 14 14 import { storage } from './lib/storage'; 15 15 import { handleSiteCreateOrUpdate, fetchSiteRecord } from './lib/cache-writer'; 16 16 import { startRevalidateWorker, stopRevalidateWorker } from './lib/revalidate-worker'; 17 17 + import { closeCacheInvalidationPublisher } from './lib/cache-invalidation'; 17 18 18 19 const app = new Hono(); 19 20 ··· 41 42 42 43 stopFirehose(); 43 44 await stopRevalidateWorker(); 45 45 + await closeCacheInvalidationPublisher(); 44 46 await closeDatabase(); 45 47 46 48 console.log('[Service] Shutdown complete');

+62

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

··· 1 1 + /** 2 2 + * Cache invalidation publisher 3 3 + * 4 4 + * Publishes invalidation messages to Redis pub/sub so the hosting-service 5 5 + * can clear its local caches (tiered storage, redirect rules) when a site 6 6 + * is updated or deleted via the firehose. 7 7 + */ 8 8 + 9 9 + import Redis from 'ioredis'; 10 10 + import { config } from '../config'; 11 11 + 12 12 + const CHANNEL = 'wisp:cache-invalidate'; 13 13 + 14 14 + let publisher: Redis | null = null; 15 15 + let loggedMissingRedis = false; 16 16 + 17 17 + function getPublisher(): Redis | null { 18 18 + if (!config.redisUrl) { 19 19 + if (!loggedMissingRedis) { 20 20 + console.warn('[CacheInvalidation] REDIS_URL not set; cache invalidation publishing disabled'); 21 21 + loggedMissingRedis = true; 22 22 + } 23 23 + return null; 24 24 + } 25 25 + 26 26 + if (!publisher) { 27 27 + publisher = new Redis(config.redisUrl, { 28 28 + maxRetriesPerRequest: 2, 29 29 + enableReadyCheck: true, 30 30 + }); 31 31 + 32 32 + publisher.on('error', (err) => { 33 33 + console.error('[CacheInvalidation] Redis error:', err); 34 34 + }); 35 35 + } 36 36 + 37 37 + return publisher; 38 38 + } 39 39 + 40 40 + export async function publishCacheInvalidation( 41 41 + did: string, 42 42 + rkey: string, 43 43 + action: 'update' | 'delete' | 'settings' 44 44 + ): Promise<void> { 45 45 + const redis = getPublisher(); 46 46 + if (!redis) return; 47 47 + 48 48 + try { 49 49 + const message = JSON.stringify({ did, rkey, action }); 50 50 + await redis.publish(CHANNEL, message); 51 51 + } catch (err) { 52 52 + console.error('[CacheInvalidation] Failed to publish:', err); 53 53 + } 54 54 + } 55 55 + 56 56 + export async function closeCacheInvalidationPublisher(): Promise<void> { 57 57 + if (publisher) { 58 58 + const toClose = publisher; 59 59 + publisher = null; 60 60 + await toClose.quit(); 61 61 + } 62 62 + }

+13

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

··· 15 15 import { getSiteCache, upsertSiteCache, deleteSiteCache, upsertSiteSettingsCache, deleteSiteSettingsCache } from './db'; 16 16 import { rewriteHtmlPaths, isHtmlFile } from './html-rewriter'; 17 17 import { gunzipSync } from 'zlib'; 18 18 + import { publishCacheInvalidation } from './cache-invalidation'; 18 19 19 20 /** 20 21 * Fetch a site record from the PDS ··· 549 550 await handleSettingsUpdate(did, rkey, settingsRecord.record, settingsRecord.cid); 550 551 } 551 552 553 553 + // Notify hosting-service to invalidate its local caches 554 554 + await publishCacheInvalidation(did, rkey, 'update'); 555 555 + 552 556 console.log(`[Cache] Successfully cached site ${did}/${rkey}`); 553 557 } 554 558 ··· 568 572 569 573 // Delete from DB 570 574 await deleteSiteCache(did, rkey); 575 575 + 576 576 + // Notify hosting-service to invalidate its local caches 577 577 + await publishCacheInvalidation(did, rkey, 'delete'); 571 578 572 579 console.log(`[Cache] Deleted site ${did}/${rkey} (${keys.length} files)`); 573 580 } ··· 586 593 cleanUrls: settings.cleanUrls, 587 594 headers: settings.headers, 588 595 }); 596 596 + 597 597 + // Notify hosting-service to invalidate its local caches (redirect rules depend on settings) 598 598 + await publishCacheInvalidation(did, rkey, 'settings'); 589 599 } 590 600 591 601 /** ··· 594 604 export async function handleSettingsDelete(did: string, rkey: string): Promise<void> { 595 605 console.log(`[Cache] Deleting settings for ${did}/${rkey}`); 596 606 await deleteSiteSettingsCache(did, rkey); 607 607 + 608 608 + // Notify hosting-service to invalidate its local caches 609 609 + await publishCacheInvalidation(did, rkey, 'settings'); 597 610 }

+3 -3

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 = fileCids ?? {}; 57 57 + const fileCidsJson = JSON.stringify(fileCids ?? {}); 58 58 console.log(`[DB] upsertSiteCache starting for ${did}/${rkey}`); 59 59 try { 60 60 await sql` ··· 94 94 const directoryListing = settings.directoryListing ?? false; 95 95 const spaMode = settings.spaMode ?? null; 96 96 const custom404 = settings.custom404 ?? null; 97 97 - const indexFilesJson = settings.indexFiles ?? []; 97 97 + const indexFilesJson = JSON.stringify(settings.indexFiles ?? []); 98 98 const cleanUrls = settings.cleanUrls ?? true; 99 99 - const headersJson = settings.headers ?? []; 99 99 + const headersJson = JSON.stringify(settings.headers ?? []); 100 100 101 101 console.log(`[DB] upsertSiteSettingsCache starting for ${did}/${rkey}`, { 102 102 directoryListing,

+4 -4

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

··· 114 114 'GROUP', 115 115 config.revalidateGroup, 116 116 consumerName, 117 117 + 'COUNT', 118 118 + batchSize, 117 119 'BLOCK', 118 120 blockMs, 119 119 - 'COUNT', 120 120 - batchSize, 121 121 'STREAMS', 122 122 config.revalidateStream, 123 123 '>' 124 124 - ); 124 124 + ) as [string, Array<[string, string[]]>][] | null; 125 125 126 126 if (!response) return; 127 127 128 128 for (const [, messages] of response) { 129 129 - await processMessages(messages as Array<[string, string[]]>); 129 129 + await processMessages(messages); 130 130 } 131 131 } 132 132

+18

apps/hosting-service/.env.example

··· 4 4 # Server 5 5 PORT=3001 6 6 BASE_HOST=wisp.place 7 7 + 8 8 + # Redis (cache invalidation + revalidation queue) 9 9 + REDIS_URL=redis://localhost:6379 10 10 + 11 11 + # S3 Storage (leave empty for local disk fallback) 12 12 + S3_BUCKET= 13 13 + S3_METADATA_BUCKET= 14 14 + S3_REGION=auto 15 15 + S3_ENDPOINT= 16 16 + S3_PREFIX=sites/ 17 17 + S3_FORCE_PATH_STYLE=true 18 18 + 19 19 + # AWS Credentials (required if using S3) 20 20 + AWS_ACCESS_KEY_ID= 21 21 + AWS_SECRET_ACCESS_KEY= 22 22 + 23 23 + # For local disk fallback (when S3_BUCKET is empty) 24 24 + CACHE_DIR=./cache/sites

+6

apps/hosting-service/src/index.ts

··· 4 4 import { mkdirSync, existsSync } from 'fs'; 5 5 import { startDomainCacheCleanup, stopDomainCacheCleanup, closeDatabase } from './lib/db'; 6 6 import { closeRevalidateQueue } from './lib/revalidate-queue'; 7 7 + import { startCacheInvalidationSubscriber, stopCacheInvalidationSubscriber } from './lib/cache-invalidation'; 7 8 import { storage, getStorageConfig } from './lib/storage'; 8 9 9 10 // Initialize Grafana exporters if configured ··· 23 24 24 25 // Start domain cache cleanup 25 26 startDomainCacheCleanup(); 27 27 + 28 28 + // Start cache invalidation subscriber (listens for firehose-service updates via Redis pub/sub) 29 29 + startCacheInvalidationSubscriber(); 26 30 27 31 // Optional: Bootstrap hot cache from warm tier on startup 28 32 const BOOTSTRAP_HOT_ON_STARTUP = process.env.BOOTSTRAP_HOT_ON_STARTUP === 'true'; ··· 80 84 process.on('SIGINT', async () => { 81 85 console.log('\n🛑 Shutting down...'); 82 86 stopDomainCacheCleanup(); 87 87 + await stopCacheInvalidationSubscriber(); 83 88 await closeRevalidateQueue(); 84 89 await closeDatabase(); 85 90 server.close(); ··· 89 94 process.on('SIGTERM', async () => { 90 95 console.log('\n🛑 Shutting down...'); 91 96 stopDomainCacheCleanup(); 97 97 + await stopCacheInvalidationSubscriber(); 92 98 await closeRevalidateQueue(); 93 99 await closeDatabase(); 94 100 server.close();

+75

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

··· 1 1 + /** 2 2 + * Cache invalidation subscriber 3 3 + * 4 4 + * Listens to Redis pub/sub for cache invalidation messages from the firehose-service. 5 5 + * When a site is updated/deleted, clears the hosting-service's local caches 6 6 + * (tiered storage hot+warm tiers, redirect rules) so stale data isn't served. 7 7 + */ 8 8 + 9 9 + import Redis from 'ioredis'; 10 10 + import { storage } from './storage'; 11 11 + import { clearRedirectRulesCache } from './site-cache'; 12 12 + 13 13 + const CHANNEL = 'wisp:cache-invalidate'; 14 14 + 15 15 + let subscriber: Redis | null = null; 16 16 + 17 17 + export function startCacheInvalidationSubscriber(): void { 18 18 + const redisUrl = process.env.REDIS_URL; 19 19 + if (!redisUrl) { 20 20 + console.warn('[CacheInvalidation] REDIS_URL not set; cache invalidation disabled'); 21 21 + return; 22 22 + } 23 23 + 24 24 + subscriber = new Redis(redisUrl, { 25 25 + maxRetriesPerRequest: 2, 26 26 + enableReadyCheck: true, 27 27 + }); 28 28 + 29 29 + subscriber.on('error', (err) => { 30 30 + console.error('[CacheInvalidation] Redis error:', err); 31 31 + }); 32 32 + 33 33 + subscriber.subscribe(CHANNEL, (err) => { 34 34 + if (err) { 35 35 + console.error('[CacheInvalidation] Failed to subscribe:', err); 36 36 + } else { 37 37 + console.log('[CacheInvalidation] Subscribed to', CHANNEL); 38 38 + } 39 39 + }); 40 40 + 41 41 + subscriber.on('message', async (_channel: string, message: string) => { 42 42 + try { 43 43 + const { did, rkey, action } = JSON.parse(message) as { 44 44 + did: string; 45 45 + rkey: string; 46 46 + action: 'update' | 'delete' | 'settings'; 47 47 + }; 48 48 + 49 49 + if (!did || !rkey) { 50 50 + console.warn('[CacheInvalidation] Invalid message:', message); 51 51 + return; 52 52 + } 53 53 + 54 54 + console.log(`[CacheInvalidation] Invalidating ${did}/${rkey} (${action})`); 55 55 + 56 56 + // Clear tiered storage (hot + warm) for this site 57 57 + const prefix = `${did}/${rkey}/`; 58 58 + const deleted = await storage.invalidate(prefix); 59 59 + console.log(`[CacheInvalidation] Cleared ${deleted} keys from tiered storage for ${did}/${rkey}`); 60 60 + 61 61 + // Clear redirect rules cache 62 62 + clearRedirectRulesCache(did, rkey); 63 63 + } catch (err) { 64 64 + console.error('[CacheInvalidation] Error processing message:', err); 65 65 + } 66 66 + }); 67 67 + } 68 68 + 69 69 + export async function stopCacheInvalidationSubscriber(): Promise<void> { 70 70 + if (subscriber) { 71 71 + const toClose = subscriber; 72 72 + subscriber = null; 73 73 + await toClose.quit(); 74 74 + } 75 75 + }

+27

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

··· 119 119 console.log('[DB] Read-only mode: skipping upsertSite', { did, rkey, displayName }); 120 120 } 121 121 122 122 + /** 123 123 + * Upsert site cache entry (used by on-demand caching when a site is completely missing) 124 124 + */ 125 125 + export async function upsertSiteCache( 126 126 + did: string, 127 127 + rkey: string, 128 128 + recordCid: string, 129 129 + fileCids: Record<string, string> 130 130 + ): Promise<void> { 131 131 + const fileCidsJson = JSON.stringify(fileCids ?? {}); 132 132 + try { 133 133 + await sql` 134 134 + 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())) 136 136 + ON CONFLICT (did, rkey) 137 137 + DO UPDATE SET 138 138 + record_cid = EXCLUDED.record_cid, 139 139 + file_cids = EXCLUDED.file_cids, 140 140 + updated_at = EXTRACT(EPOCH FROM NOW()) 141 141 + `; 142 142 + } catch (err) { 143 143 + const error = err instanceof Error ? err : new Error(String(err)); 144 144 + console.error('[DB] upsertSiteCache error:', { did, rkey, error: error.message }); 145 145 + throw error; 146 146 + } 147 147 + } 148 148 + 122 149 export interface SiteRecord { 123 150 did: string; 124 151 rkey: string;

+27 -2

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

··· 18 18 import { enqueueRevalidate } from './revalidate-queue'; 19 19 import { recordStorageMiss } from './revalidate-metrics'; 20 20 import { normalizeFileCids } from '@wispplace/fs-utils'; 21 21 + import { fetchAndCacheSite } from './on-demand-cache'; 22 22 + import type { StorageResult } from '@wispplace/tiered-storage'; 23 23 + 24 24 + type FileStorageResult = StorageResult<Uint8Array>; 21 25 22 26 /** 23 27 * Helper to retrieve a file with metadata from tiered storage ··· 91 95 rkey: string, 92 96 filePath: string, 93 97 preferRewrittenHtml: boolean 94 94 - ): Promise<{ result: Awaited<ReturnType<typeof storage.getWithMetadata>>; filePath: string } | null> { 98 98 + ): Promise<{ result: FileStorageResult; filePath: string } | null> { 95 99 const mimeTypeGuess = lookup(filePath) || 'application/octet-stream'; 96 100 if (preferRewrittenHtml && isHtmlContent(filePath, mimeTypeGuess)) { 97 101 const rewrittenPath = `.rewritten/${filePath}`; ··· 107 111 } 108 112 109 113 function buildResponseFromStorageResult( 110 110 - result: Awaited<ReturnType<typeof storage.getWithMetadata>>, 114 114 + result: FileStorageResult, 111 115 filePath: string, 112 116 settings: WispSettings | null, 113 117 requestHeaders?: Record<string, string> ··· 149 153 } 150 154 151 155 /** 156 156 + * Ensure a site is cached locally. If the site has no DB entry (completely unknown), 157 157 + * attempt to fetch and cache it on-demand from the PDS. 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 162 162 + 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); 166 166 + } 167 167 + 168 168 + /** 152 169 * Helper to serve files from cache (for custom domains and subdomains) 153 170 */ 154 171 export async function serveFromCache( ··· 158 175 fullUrl?: string, 159 176 headers?: Record<string, string> 160 177 ): Promise<Response> { 178 178 + // Check if this site is completely unknown (not in DB, no files in storage) 179 179 + // If so, attempt to fetch and cache it on-demand from the PDS 180 180 + await ensureSiteCached(did, rkey); 181 181 + 161 182 // Load settings for this site 162 183 const settings = await getCachedSettings(did, rkey); 163 184 const indexFiles = getIndexFiles(settings); ··· 445 466 fullUrl?: string, 446 467 headers?: Record<string, string> 447 468 ): Promise<Response> { 469 469 + // Check if this site is completely unknown (not in DB, no files in storage) 470 470 + // If so, attempt to fetch and cache it on-demand from the PDS 471 471 + await ensureSiteCached(did, rkey); 472 472 + 448 473 // Load settings for this site 449 474 const settings = await getCachedSettings(did, rkey); 450 475 const indexFiles = getIndexFiles(settings);

+259

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

··· 1 1 + /** 2 2 + * On-demand site caching for the hosting service 3 3 + * 4 4 + * When a request hits a site that is completely missing (no DB entry, no files), 5 5 + * this module fetches the site record from the PDS, downloads all blobs, 6 6 + * writes them to local storage (hot + warm tiers), and updates the DB. 7 7 + * 8 8 + * This gives immediate serving capability. A revalidate is also enqueued 9 9 + * so the firehose-service backfills S3 (cold tier). 10 10 + */ 11 11 + 12 12 + import type { Record as WispFsRecord, Directory, Entry, File } from '@wispplace/lexicons/types/place/wisp/fs'; 13 13 + import { safeFetchJson, safeFetchBlob } from '@wispplace/safe-fetch'; 14 14 + import { extractBlobCid, getPdsForDid } from '@wispplace/atproto-utils'; 15 15 + import { shouldCompressMimeType } from '@wispplace/atproto-utils/compression'; 16 16 + import { collectFileCidsFromEntries, countFilesInDirectory } from '@wispplace/fs-utils'; 17 17 + import { MAX_BLOB_SIZE, MAX_FILE_COUNT, MAX_SITE_SIZE } from '@wispplace/constants'; 18 18 + import { expandSubfsNodes } from './utils'; 19 19 + import { storage } from './storage'; 20 20 + import { upsertSiteCache, tryAcquireLock, releaseLock } from './db'; 21 21 + import { enqueueRevalidate } from './revalidate-queue'; 22 22 + import { gunzipSync } from 'zlib'; 23 23 + 24 24 + // Track in-flight fetches to avoid duplicate work 25 25 + const inFlightFetches = new Map<string, Promise<boolean>>(); 26 26 + 27 27 + interface FileInfo { 28 28 + path: string; 29 29 + cid: string; 30 30 + blob: any; 31 31 + encoding?: 'gzip'; 32 32 + mimeType?: string; 33 33 + base64?: boolean; 34 34 + } 35 35 + 36 36 + /** 37 37 + * Attempt to fetch and cache a completely missing site on-demand. 38 38 + * Returns true if the site was successfully cached, false otherwise. 39 39 + * 40 40 + * Uses a distributed lock (pg advisory lock) to prevent multiple 41 41 + * hosting-service instances from fetching the same site simultaneously. 42 42 + */ 43 43 + export async function fetchAndCacheSite(did: string, rkey: string): Promise<boolean> { 44 44 + const key = `${did}:${rkey}`; 45 45 + 46 46 + // Check if there's already an in-flight fetch for this site 47 47 + const existing = inFlightFetches.get(key); 48 48 + if (existing) { 49 49 + return existing; 50 50 + } 51 51 + 52 52 + const fetchPromise = doFetchAndCache(did, rkey); 53 53 + inFlightFetches.set(key, fetchPromise); 54 54 + 55 55 + try { 56 56 + return await fetchPromise; 57 57 + } finally { 58 58 + inFlightFetches.delete(key); 59 59 + } 60 60 + } 61 61 + 62 62 + async function doFetchAndCache(did: string, rkey: string): Promise<boolean> { 63 63 + const lockKey = `on-demand-cache:${did}:${rkey}`; 64 64 + 65 65 + // Try to acquire a distributed lock 66 66 + const acquired = await tryAcquireLock(lockKey); 67 67 + if (!acquired) { 68 68 + console.log(`[OnDemandCache] Lock not acquired for ${did}/${rkey}, another instance is handling it`); 69 69 + return false; 70 70 + } 71 71 + 72 72 + try { 73 73 + console.log(`[OnDemandCache] Fetching missing site ${did}/${rkey}`); 74 74 + 75 75 + // Fetch site record from PDS 76 76 + const pdsEndpoint = await getPdsForDid(did); 77 77 + if (!pdsEndpoint) { 78 78 + console.error(`[OnDemandCache] Could not resolve PDS for ${did}`); 79 79 + return false; 80 80 + } 81 81 + 82 82 + const recordUrl = `${pdsEndpoint}/xrpc/com.atproto.repo.getRecord?repo=${encodeURIComponent(did)}&collection=place.wisp.fs&rkey=${encodeURIComponent(rkey)}`; 83 83 + 84 84 + let data: any; 85 85 + try { 86 86 + data = await safeFetchJson(recordUrl); 87 87 + } catch (err) { 88 88 + const msg = err instanceof Error ? err.message : String(err); 89 89 + if (msg.includes('HTTP 404') || msg.includes('Not Found')) { 90 90 + console.log(`[OnDemandCache] Site record not found on PDS: ${did}/${rkey}`); 91 91 + } else { 92 92 + console.error(`[OnDemandCache] Failed to fetch site record: ${did}/${rkey}`, msg); 93 93 + } 94 94 + return false; 95 95 + } 96 96 + 97 97 + const record = data.value as WispFsRecord; 98 98 + const recordCid = data.cid || ''; 99 99 + 100 100 + if (!record?.root?.entries) { 101 101 + console.error(`[OnDemandCache] Invalid record structure for ${did}/${rkey}`); 102 102 + return false; 103 103 + } 104 104 + 105 105 + // Expand subfs nodes 106 106 + const expandedRoot = await expandSubfsNodes(record.root, pdsEndpoint); 107 107 + 108 108 + // Validate limits 109 109 + const fileCount = countFilesInDirectory(expandedRoot); 110 110 + if (fileCount > MAX_FILE_COUNT) { 111 111 + console.error(`[OnDemandCache] Site exceeds file limit: ${fileCount} > ${MAX_FILE_COUNT}`); 112 112 + return false; 113 113 + } 114 114 + 115 115 + // Collect files 116 116 + const files = collectFileInfo(expandedRoot.entries); 117 117 + 118 118 + // Collect file CIDs for DB 119 119 + const fileCids: Record<string, string> = {}; 120 120 + collectFileCidsFromEntries(expandedRoot.entries, '', fileCids); 121 121 + 122 122 + // Download and write all files to local storage (hot + warm tiers) 123 123 + const CONCURRENCY = 10; 124 124 + let downloaded = 0; 125 125 + let failed = 0; 126 126 + 127 127 + for (let i = 0; i < files.length; i += CONCURRENCY) { 128 128 + const batch = files.slice(i, i + CONCURRENCY); 129 129 + const results = await Promise.allSettled( 130 130 + batch.map(file => downloadAndWriteBlob(did, rkey, file, pdsEndpoint)) 131 131 + ); 132 132 + 133 133 + for (const result of results) { 134 134 + if (result.status === 'fulfilled') { 135 135 + downloaded++; 136 136 + } else { 137 137 + failed++; 138 138 + console.error(`[OnDemandCache] Failed to download blob:`, result.reason); 139 139 + } 140 140 + } 141 141 + } 142 142 + 143 143 + console.log(`[OnDemandCache] Downloaded ${downloaded} files (${failed} failed) for ${did}/${rkey}`); 144 144 + 145 145 + // Update DB with file CIDs so future storage misses can be detected 146 146 + await upsertSiteCache(did, rkey, recordCid, fileCids); 147 147 + 148 148 + // Enqueue revalidate so firehose-service backfills S3 (cold tier) 149 149 + await enqueueRevalidate(did, rkey, `on-demand-cache`); 150 150 + 151 151 + console.log(`[OnDemandCache] Successfully cached site ${did}/${rkey}`); 152 152 + return downloaded > 0; 153 153 + } catch (err) { 154 154 + console.error(`[OnDemandCache] Error caching site ${did}/${rkey}:`, err); 155 155 + return false; 156 156 + } finally { 157 157 + await releaseLock(lockKey); 158 158 + } 159 159 + } 160 160 + 161 161 + function collectFileInfo(entries: Entry[], pathPrefix: string = ''): FileInfo[] { 162 162 + const files: FileInfo[] = []; 163 163 + 164 164 + for (const entry of entries) { 165 165 + const currentPath = pathPrefix ? `${pathPrefix}/${entry.name}` : entry.name; 166 166 + const node = entry.node; 167 167 + 168 168 + if ('type' in node && node.type === 'directory' && 'entries' in node) { 169 169 + files.push(...collectFileInfo(node.entries, currentPath)); 170 170 + } else if ('type' in node && node.type === 'file' && 'blob' in node) { 171 171 + const fileNode = node as File; 172 172 + const cid = extractBlobCid(fileNode.blob); 173 173 + if (cid) { 174 174 + files.push({ 175 175 + path: currentPath, 176 176 + cid, 177 177 + blob: fileNode.blob, 178 178 + encoding: fileNode.encoding, 179 179 + mimeType: fileNode.mimeType, 180 180 + base64: fileNode.base64, 181 181 + }); 182 182 + } 183 183 + } 184 184 + } 185 185 + 186 186 + return files; 187 187 + } 188 188 + 189 189 + async function downloadAndWriteBlob( 190 190 + did: string, 191 191 + rkey: string, 192 192 + file: FileInfo, 193 193 + pdsEndpoint: string 194 194 + ): Promise<void> { 195 195 + const blobUrl = `${pdsEndpoint}/xrpc/com.atproto.sync.getBlob?did=${encodeURIComponent(did)}&cid=${encodeURIComponent(file.cid)}`; 196 196 + 197 197 + let content = await safeFetchBlob(blobUrl, { maxSize: MAX_BLOB_SIZE, timeout: 300000 }); 198 198 + let encoding = file.encoding; 199 199 + 200 200 + // Decode base64 if flagged 201 201 + if (file.base64) { 202 202 + const base64String = new TextDecoder().decode(content); 203 203 + content = Buffer.from(base64String, 'base64'); 204 204 + } 205 205 + 206 206 + // Decompress if needed and shouldn't stay compressed 207 207 + const shouldStayCompressed = shouldCompressMimeType(file.mimeType); 208 208 + 209 209 + if (encoding === 'gzip' && !shouldStayCompressed && content.length >= 2 && 210 210 + content[0] === 0x1f && content[1] === 0x8b) { 211 211 + try { 212 212 + content = gunzipSync(content); 213 213 + encoding = undefined; 214 214 + } catch { 215 215 + // Keep gzipped if decompression fails 216 216 + } 217 217 + } 218 218 + 219 219 + // If encoding is missing but data looks gzipped for a text-like file, mark it 220 220 + if (!encoding && isTextLikeMime(file.mimeType, file.path) && content.length >= 2 && 221 221 + content[0] === 0x1f && content[1] === 0x8b) { 222 222 + encoding = 'gzip'; 223 223 + } 224 224 + 225 225 + // Build storage key and metadata 226 226 + const key = `${did}/${rkey}/${file.path}`; 227 227 + const metadata: Record<string, string> = {}; 228 228 + if (encoding) metadata.encoding = encoding; 229 229 + if (file.mimeType) metadata.mimeType = file.mimeType; 230 230 + 231 231 + // Write to hot + warm tiers only (cold/S3 is read-only in hosting-service, 232 232 + // firehose-service will backfill via revalidate) 233 233 + await storage.set(key as any, content as any, { 234 234 + metadata, 235 235 + skipTiers: [], 236 236 + }); 237 237 + } 238 238 + 239 239 + function isTextLikeMime(mimeType?: string, path?: string): boolean { 240 240 + if (mimeType) { 241 241 + if (mimeType === 'text/html') return true; 242 242 + if (mimeType === 'text/css') return true; 243 243 + if (mimeType === 'text/javascript') return true; 244 244 + if (mimeType === 'application/javascript') return true; 245 245 + if (mimeType === 'application/json') return true; 246 246 + if (mimeType === 'application/xml') return true; 247 247 + if (mimeType === 'image/svg+xml') return true; 248 248 + } 249 249 + 250 250 + if (!path) return false; 251 251 + const lower = path.toLowerCase(); 252 252 + return lower.endsWith('.html') || 253 253 + lower.endsWith('.htm') || 254 254 + lower.endsWith('.css') || 255 255 + lower.endsWith('.js') || 256 256 + lower.endsWith('.json') || 257 257 + lower.endsWith('.xml') || 258 258 + lower.endsWith('.svg'); 259 259 + }

+1 -1

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

··· 46 46 47 47 try { 48 48 const dedupeKey = `revalidate:site:${did}:${rkey}`; 49 49 - const set = await redis.set(dedupeKey, '1', 'NX', 'EX', dedupeTtlSeconds); 49 49 + const set = await redis.set(dedupeKey, '1', 'EX', dedupeTtlSeconds, 'NX'); 50 50 if (!set) { 51 51 recordRevalidateResult('deduped'); 52 52 return { enqueued: false, result: 'deduped' };

+1 -1

apps/hosting-service/src/server.ts

··· 38 38 app.get('/*', async (c) => { 39 39 const url = new URL(c.req.url); 40 40 const hostname = c.req.header('host') || ''; 41 41 - const hostnameWithoutPort = hostname.split(':')[0]; 41 41 + const hostnameWithoutPort = hostname.split(':')[0] || ''; 42 42 const rawPath = url.pathname.replace(/^\//, ''); 43 43 const path = sanitizePath(rawPath); 44 44

+15

docker-compose.yml

··· 17 17 timeout: 5s 18 18 retries: 5 19 19 20 20 + redis: 21 21 + image: redis:7-alpine 22 22 + container_name: wisp-redis 23 23 + restart: unless-stopped 24 24 + ports: 25 25 + - "6379:6379" 26 26 + volumes: 27 27 + - redis_data:/data 28 28 + healthcheck: 29 29 + test: ["CMD", "redis-cli", "ping"] 30 30 + interval: 5s 31 31 + timeout: 5s 32 32 + retries: 5 33 33 + 20 34 volumes: 21 35 postgres_data: 36 36 + redis_data: