diff --git a/apps/firehose-service/package.json b/apps/firehose-service/package.json index 875171b..5c8a7cd 100644 --- a/apps/firehose-service/package.json +++ b/apps/firehose-service/package.json @@ -4,6 +4,7 @@ "type": "module", "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", "build": "bun run build.ts", "start": "NODE_OPTIONS='--max-old-space-size=2048' tsx src/index.ts", "check": "tsc --noEmit", @@ -20,10 +21,11 @@ "@wispplace/fs-utils": "workspace:*", "@wispplace/lexicons": "workspace:*", "@wispplace/safe-fetch": "workspace:*", + "@wispplace/tiered-storage": "workspace:*", "hono": "^4.10.4", + "ioredis": "^5.9.2", "multiformats": "^13.4.1", - "postgres": "^3.4.5", - "@wispplace/tiered-storage": "workspace:*" + "postgres": "^3.4.5" }, "devDependencies": { "@types/node": "^22.10.5", diff --git a/apps/firehose-service/src/config.ts b/apps/firehose-service/src/config.ts index 7f461fc..b79ec83 100644 --- a/apps/firehose-service/src/config.ts +++ b/apps/firehose-service/src/config.ts @@ -23,6 +23,11 @@ export const config = { // Health check server healthPort: parseInt(process.env.HEALTH_PORT || '3001', 10), + // Redis revalidation queue + redisUrl: process.env.REDIS_URL, + revalidateStream: process.env.WISP_REVALIDATE_STREAM || 'wisp:revalidate', + revalidateGroup: process.env.WISP_REVALIDATE_GROUP || 'firehose-service', + // Mode isBackfill: process.argv.includes('--backfill'), } as const; diff --git a/apps/firehose-service/src/index.ts b/apps/firehose-service/src/index.ts index 852495a..0539e70 100644 --- a/apps/firehose-service/src/index.ts +++ b/apps/firehose-service/src/index.ts @@ -10,9 +10,10 @@ import { Hono } from 'hono'; import { serve } from '@hono/node-server'; import { config } from './config'; import { startFirehose, stopFirehose, getFirehoseHealth } from './lib/firehose'; -import { closeDatabase, listAllSiteCaches, getSiteCache } from './lib/db'; +import { closeDatabase, listAllSiteCaches, listAllSites, getSiteCache } from './lib/db'; import { storage } from './lib/storage'; import { handleSiteCreateOrUpdate, fetchSiteRecord } from './lib/cache-writer'; +import { startRevalidateWorker, stopRevalidateWorker } from './lib/revalidate-worker'; const app = new Hono(); @@ -39,6 +40,7 @@ async function shutdown(signal: string) { console.log(`\n[Service] Received ${signal}, shutting down...`); stopFirehose(); + await stopRevalidateWorker(); await closeDatabase(); console.log('[Service] Shutdown complete'); @@ -53,8 +55,20 @@ process.on('SIGTERM', () => shutdown('SIGTERM')); */ async function runBackfill(): Promise { console.log('[Backfill] Starting backfill mode'); + const startTime = Date.now(); + const forceRewriteHtml = process.env.BACKFILL_FORCE_REWRITE_HTML === 'true'; + + if (forceRewriteHtml) { + console.log('[Backfill] Forcing HTML rewrite for all sites'); + } + + let sites = await listAllSites(); + if (sites.length === 0) { + const cachedSites = await listAllSiteCaches(); + sites = cachedSites.map(site => ({ did: site.did, rkey: site.rkey })); + console.log('[Backfill] Sites table empty; falling back to site_cache entries'); + } - const sites = await listAllSiteCaches(); console.log(`[Backfill] Found ${sites.length} sites in database`); let processed = 0; @@ -72,15 +86,18 @@ async function runBackfill(): Promise { continue; } + const existingCache = await getSiteCache(site.did, site.rkey); // Check if CID matches (already up to date) - if (result.cid === site.record_cid) { + if (!forceRewriteHtml && existingCache && result.cid === existingCache.record_cid) { console.log(`[Backfill] Site already up to date: ${site.did}/${site.rkey}`); skipped++; continue; } // Process the site - await handleSiteCreateOrUpdate(site.did, site.rkey, result.record, result.cid); + await handleSiteCreateOrUpdate(site.did, site.rkey, result.record, result.cid, { + forceRewriteHtml, + }); processed++; console.log(`[Backfill] Progress: ${processed + skipped + failed}/${sites.length}`); @@ -90,7 +107,13 @@ async function runBackfill(): Promise { } } - console.log(`[Backfill] Complete: ${processed} processed, ${skipped} skipped, ${failed} failed`); + const elapsedMs = Date.now() - startTime; + const elapsedSec = Math.round(elapsedMs / 1000); + const elapsedMin = Math.floor(elapsedSec / 60); + const elapsedRemSec = elapsedSec % 60; + const elapsedLabel = elapsedMin > 0 ? `${elapsedMin}m ${elapsedRemSec}s` : `${elapsedSec}s`; + + console.log(`[Backfill] Complete: ${processed} processed, ${skipped} skipped, ${failed} failed (${elapsedLabel} elapsed)`); } // Main entry point @@ -115,6 +138,7 @@ async function main() { } else { // Start firehose startFirehose(); + await startRevalidateWorker(); } } diff --git a/apps/firehose-service/src/lib/cache-writer.ts b/apps/firehose-service/src/lib/cache-writer.ts index bffff00..b15a86c 100644 --- a/apps/firehose-service/src/lib/cache-writer.ts +++ b/apps/firehose-service/src/lib/cache-writer.ts @@ -8,7 +8,7 @@ import type { Record as SubfsRecord } from '@wispplace/lexicons/types/place/wisp import type { Record as WispSettings } from '@wispplace/lexicons/types/place/wisp/settings'; import { safeFetchJson, safeFetchBlob } from '@wispplace/safe-fetch'; import { extractBlobCid, getPdsForDid } from '@wispplace/atproto-utils'; -import { collectFileCidsFromEntries, countFilesInDirectory } from '@wispplace/fs-utils'; +import { collectFileCidsFromEntries, countFilesInDirectory, normalizeFileCids } from '@wispplace/fs-utils'; import { shouldCompressMimeType } from '@wispplace/atproto-utils/compression'; import { MAX_BLOB_SIZE, MAX_FILE_COUNT, MAX_SITE_SIZE } from '@wispplace/constants'; import { writeFile, deleteFile, listFiles } from './storage'; @@ -244,49 +244,59 @@ interface FileInfo { base64?: boolean; } -type FileCidsNormalization = { - value: Record; - source: 'object' | 'array' | 'string-json' | 'string-invalid' | 'null' | 'other'; -}; +function isTextLikeMime(mimeType?: string, path?: string): boolean { + if (mimeType) { + if (mimeType === 'text/html') return true; + if (mimeType === 'text/css') return true; + if (mimeType === 'text/javascript') return true; + if (mimeType === 'application/javascript') return true; + if (mimeType === 'application/json') return true; + if (mimeType === 'application/xml') return true; + if (mimeType === 'image/svg+xml') return true; + } -function normalizeFileCids(value: unknown): FileCidsNormalization { - if (value == null) return { value: {}, source: 'null' }; + if (!path) return false; + const lower = path.toLowerCase(); + return lower.endsWith('.html') || + lower.endsWith('.htm') || + lower.endsWith('.css') || + lower.endsWith('.js') || + lower.endsWith('.json') || + lower.endsWith('.xml') || + lower.endsWith('.svg'); +} - if (typeof value === 'string') { - try { - const parsed = JSON.parse(value) as unknown; - if (Array.isArray(parsed)) { - const normalized = normalizeFileCids(parsed); - return { value: normalized.value, source: 'string-json' }; - } - if (parsed && typeof parsed === 'object') { - return { value: parsed as Record, source: 'string-json' }; - } - } catch { - // fall through to invalid +function looksLikeBase64(content: Uint8Array): boolean { + if (content.length === 0) return false; + let nonWhitespace = 0; + for (const byte of content) { + const char = byte; + if (char === 0x0a || char === 0x0d || char === 0x20 || char === 0x09) { + continue; } - return { value: {}, source: 'string-invalid' }; + nonWhitespace++; + const isBase64Char = + (char >= 0x41 && char <= 0x5a) || // A-Z + (char >= 0x61 && char <= 0x7a) || // a-z + (char >= 0x30 && char <= 0x39) || // 0-9 + char === 0x2b || // + + char === 0x2f || // / + char === 0x3d; // = + if (!isBase64Char) return false; } - if (Array.isArray(value)) { - const result: Record = {}; - for (const item of value) { - if (item && typeof item === 'object' && 'path' in item && 'cid' in item) { - const path = (item as any).path; - const cid = (item as any).cid; - if (typeof path === 'string' && typeof cid === 'string') { - result[path] = cid; - } - } - } - return { value: result, source: 'array' }; - } + // Base64 length should be divisible by 4 (ignoring whitespace) + return nonWhitespace % 4 === 0; +} - if (typeof value === 'object') { - return { value: value as Record, source: 'object' }; +function tryDecodeBase64(content: Uint8Array): Uint8Array | null { + if (!looksLikeBase64(content)) return null; + const base64String = new TextDecoder().decode(content).replace(/\s+/g, ''); + try { + return Buffer.from(base64String, 'base64'); + } catch { + return null; } - - return { value: {}, source: 'other' }; } /** @@ -334,17 +344,24 @@ async function downloadAndWriteBlob( console.log(`[Cache] Downloading ${file.path}`); let content = await safeFetchBlob(blobUrl, { maxSize: MAX_BLOB_SIZE, timeout: 300000 }); + let encoding = file.encoding; // Decode base64 if needed if (file.base64) { const textDecoder = new TextDecoder(); const base64String = textDecoder.decode(content); content = Buffer.from(base64String, 'base64'); + } else if (isTextLikeMime(file.mimeType, file.path)) { + // Heuristic fallback: some records omit base64 flag but content is base64 text + const decoded = tryDecodeBase64(content); + if (decoded) { + console.warn(`[Cache] Decoded base64 fallback for ${file.path} (base64 flag missing)`); + content = decoded; + } } // Decompress if needed and shouldn't stay compressed const shouldStayCompressed = shouldCompressMimeType(file.mimeType); - let encoding = file.encoding; if (encoding === 'gzip' && !shouldStayCompressed && content.length >= 2 && content[0] === 0x1f && content[1] === 0x8b) { @@ -354,6 +371,26 @@ async function downloadAndWriteBlob( } catch (error) { console.error(`[Cache] Failed to decompress ${file.path}, storing gzipped`); } + } else if (encoding === 'gzip' && content.length >= 2 && + !(content[0] === 0x1f && content[1] === 0x8b)) { + // If marked gzip but doesn't look gzipped, attempt base64 decode and retry + const decoded = tryDecodeBase64(content); + if (decoded && decoded.length >= 2 && decoded[0] === 0x1f && decoded[1] === 0x8b) { + console.warn(`[Cache] Decoded base64+gzip fallback for ${file.path}`); + try { + content = gunzipSync(decoded); + encoding = undefined; + } catch (error) { + console.error(`[Cache] Failed to decompress base64+gzip fallback for ${file.path}, storing gzipped`); + content = decoded; + } + } + } + + // If encoding is missing but data looks gzipped for a text-like file, mark it + if (!encoding && isTextLikeMime(file.mimeType, file.path) && content.length >= 2 && + content[0] === 0x1f && content[1] === 0x8b) { + encoding = 'gzip'; } // Build storage key @@ -370,7 +407,16 @@ async function downloadAndWriteBlob( // If HTML, also write rewritten version if (isHtmlFile(file.path)) { const basePath = `/${did}/${rkey}/`; - const htmlString = new TextDecoder().decode(content); + let rewriteSource = content; + if (encoding === 'gzip' && content.length >= 2 && content[0] === 0x1f && content[1] === 0x8b) { + try { + rewriteSource = gunzipSync(content); + } catch (error) { + console.error(`[Cache] Failed to decompress ${file.path} for rewrite, using raw content`); + } + } + + const htmlString = new TextDecoder().decode(rewriteSource); const rewritten = rewriteHtmlPaths(htmlString, basePath, file.path); const rewrittenContent = new TextEncoder().encode(rewritten); @@ -389,9 +435,15 @@ export async function handleSiteCreateOrUpdate( did: string, rkey: string, record: WispFsRecord, - recordCid: string + recordCid: string, + options?: { + forceRewriteHtml?: boolean; + } ): Promise { - console.log(`[Cache] Processing site ${did}/${rkey}, record CID: ${recordCid}`); + const forceRewriteHtml = options?.forceRewriteHtml === true; + console.log(`[Cache] Processing site ${did}/${rkey}, record CID: ${recordCid}`, { + forceRewriteHtml, + }); if (!record.root?.entries) { console.error('[Cache] Invalid record structure'); @@ -445,7 +497,8 @@ export async function handleSiteCreateOrUpdate( // Find new or changed files for (const file of newFiles) { - if (oldFileCids[file.path] !== file.cid) { + const shouldForceRewrite = forceRewriteHtml && isHtmlFile(file.path); + if (oldFileCids[file.path] !== file.cid || shouldForceRewrite) { filesToDownload.push(file); } } @@ -523,9 +576,9 @@ export async function handleSiteDelete(did: string, rkey: string): Promise * Handle settings create/update event */ export async function handleSettingsUpdate(did: string, rkey: string, settings: WispSettings, recordCid: string): Promise { - console.log(`[Cache] Updating settings for ${did}`); + console.log(`[Cache] Updating settings for ${did}/${rkey}`); - await upsertSiteSettingsCache(did, recordCid, { + await upsertSiteSettingsCache(did, rkey, recordCid, { directoryListing: settings.directoryListing, spaMode: settings.spaMode, custom404: settings.custom404, @@ -538,7 +591,7 @@ export async function handleSettingsUpdate(did: string, rkey: string, settings: /** * Handle settings delete event */ -export async function handleSettingsDelete(did: string): Promise { - console.log(`[Cache] Deleting settings for ${did}`); - await deleteSiteSettingsCache(did); +export async function handleSettingsDelete(did: string, rkey: string): Promise { + console.log(`[Cache] Deleting settings for ${did}/${rkey}`); + await deleteSiteSettingsCache(did, rkey); } diff --git a/apps/firehose-service/src/lib/db.ts b/apps/firehose-service/src/lib/db.ts index d1f3f3d..ac09c55 100644 --- a/apps/firehose-service/src/lib/db.ts +++ b/apps/firehose-service/src/lib/db.ts @@ -1,5 +1,5 @@ import postgres from 'postgres'; -import type { SiteCache, SiteSettingsCache } from '@wispplace/database'; +import type { SiteCache, SiteRecord, SiteSettingsCache } from '@wispplace/database'; import { config } from '../config'; const sql = postgres(config.databaseUrl, { @@ -20,11 +20,11 @@ export async function getSiteCache(did: string, rkey: string): Promise { +export async function getSiteSettingsCache(did: string, rkey: string): Promise { const result = await sql` - SELECT did, record_cid, directory_listing, spa_mode, custom_404, index_files, clean_urls, headers, cached_at, updated_at + SELECT did, rkey, record_cid, directory_listing, spa_mode, custom_404, index_files, clean_urls, headers, cached_at, updated_at FROM site_settings_cache - WHERE did = ${did} + WHERE did = ${did} AND rkey = ${rkey} LIMIT 1 `; return result[0] || null; @@ -38,6 +38,14 @@ export async function listAllSiteCaches(): Promise { `; } +export async function listAllSites(): Promise { + return await sql` + SELECT did, rkey, display_name, created_at, updated_at + FROM sites + ORDER BY updated_at DESC + `; +} + // Write functions export async function upsertSiteCache( @@ -72,6 +80,7 @@ export async function deleteSiteCache(did: string, rkey: string): Promise export async function upsertSiteSettingsCache( did: string, + rkey: string, recordCid: string, settings: { directoryListing: boolean; @@ -89,7 +98,7 @@ export async function upsertSiteSettingsCache( const cleanUrls = settings.cleanUrls ?? true; const headersJson = settings.headers ?? []; - console.log(`[DB] upsertSiteSettingsCache starting for ${did}`, { + console.log(`[DB] upsertSiteSettingsCache starting for ${did}/${rkey}`, { directoryListing, spaMode, custom404, @@ -100,9 +109,10 @@ export async function upsertSiteSettingsCache( try { await sql` - INSERT INTO site_settings_cache (did, record_cid, directory_listing, spa_mode, custom_404, index_files, clean_urls, headers, cached_at, updated_at) + INSERT INTO site_settings_cache (did, rkey, record_cid, directory_listing, spa_mode, custom_404, index_files, clean_urls, headers, cached_at, updated_at) VALUES ( ${did}, + ${rkey}, ${recordCid}, ${directoryListing}, ${spaMode}, @@ -113,7 +123,7 @@ export async function upsertSiteSettingsCache( EXTRACT(EPOCH FROM NOW()), EXTRACT(EPOCH FROM NOW()) ) - ON CONFLICT (did) + ON CONFLICT (did, rkey) DO UPDATE SET record_cid = EXCLUDED.record_cid, directory_listing = EXCLUDED.directory_listing, @@ -124,16 +134,16 @@ export async function upsertSiteSettingsCache( headers = EXCLUDED.headers, updated_at = EXTRACT(EPOCH FROM NOW()) `; - console.log(`[DB] upsertSiteSettingsCache completed for ${did}`); + console.log(`[DB] upsertSiteSettingsCache completed for ${did}/${rkey}`); } catch (err) { const error = err instanceof Error ? err : new Error(String(err)); - console.error('[DB] upsertSiteSettingsCache error:', { did, error: error.message, stack: error.stack }); + console.error('[DB] upsertSiteSettingsCache error:', { did, rkey, error: error.message, stack: error.stack }); throw error; } } -export async function deleteSiteSettingsCache(did: string): Promise { - await sql`DELETE FROM site_settings_cache WHERE did = ${did}`; +export async function deleteSiteSettingsCache(did: string, rkey: string): Promise { + await sql`DELETE FROM site_settings_cache WHERE did = ${did} AND rkey = ${rkey}`; } export async function closeDatabase(): Promise { diff --git a/apps/firehose-service/src/lib/firehose.ts b/apps/firehose-service/src/lib/firehose.ts index 24225f1..b83a4e6 100644 --- a/apps/firehose-service/src/lib/firehose.ts +++ b/apps/firehose-service/src/lib/firehose.ts @@ -106,7 +106,7 @@ async function handleEvent(evt: Event | CommitEvt): Promise { processWithConcurrencyLimit(async () => { try { if (commitEvt.event === 'delete') { - await handleSettingsDelete(did); + await handleSettingsDelete(did, rkey); } else if (record) { const cidStr = cid?.toString() || ''; await handleSettingsUpdate(did, rkey, record as WispSettings, cidStr); diff --git a/apps/firehose-service/src/lib/revalidate-worker.ts b/apps/firehose-service/src/lib/revalidate-worker.ts new file mode 100644 index 0000000..37a0c77 --- /dev/null +++ b/apps/firehose-service/src/lib/revalidate-worker.ts @@ -0,0 +1,181 @@ +import Redis from 'ioredis'; +import os from 'os'; +import { config } from '../config'; +import { fetchSiteRecord, handleSiteCreateOrUpdate } from './cache-writer'; + +const consumerName = process.env.WISP_REVALIDATE_CONSUMER || `${os.hostname()}:${process.pid}`; +const batchSize = Number.parseInt(process.env.WISP_REVALIDATE_BATCH_SIZE || '10', 10); +const claimIdleMs = Number.parseInt(process.env.WISP_REVALIDATE_CLAIM_IDLE_MS || '60000', 10); +const blockMs = Number.parseInt(process.env.WISP_REVALIDATE_BLOCK_MS || '5000', 10); + +let redis: Redis | null = null; +let running = false; +let loopPromise: Promise | null = null; + +function parseFields(raw: string[]): Record { + const fields: Record = {}; + for (let i = 0; i < raw.length; i += 2) { + const key = raw[i]; + const value = raw[i + 1]; + if (key) { + fields[key] = value ?? ''; + } + } + return fields; +} + +async function processMessage(id: string, rawFields: string[]): Promise { + if (!redis) return; + + const fields = parseFields(rawFields); + const did = fields.did; + const rkey = fields.rkey; + const reason = fields.reason || 'storage-miss'; + + if (!did || !rkey) { + console.warn('[Revalidate] Missing did/rkey in message', { id, fields }); + await redis.xack(config.revalidateStream, config.revalidateGroup, id); + return; + } + + console.log('[Revalidate] Processing', { did, rkey, reason, id }); + + const record = await fetchSiteRecord(did, rkey); + if (!record) { + console.warn('[Revalidate] Site record not found', { did, rkey }); + await redis.xack(config.revalidateStream, config.revalidateGroup, id); + return; + } + + await handleSiteCreateOrUpdate(did, rkey, record.record, record.cid); + + await redis.xack(config.revalidateStream, config.revalidateGroup, id); +} + +async function processMessages(messages: Array<[string, string[]]>): Promise { + for (const [id, rawFields] of messages) { + try { + await processMessage(id, rawFields); + } catch (err) { + const error = err instanceof Error ? err : new Error(String(err)); + console.error('[Revalidate] Failed to process message', { id, error: error.message, stack: error.stack }); + } + } +} + +async function ensureGroup(): Promise { + if (!redis) return; + try { + await redis.xgroup('CREATE', config.revalidateStream, config.revalidateGroup, '0', 'MKSTREAM'); + } catch (err) { + const error = err instanceof Error ? err : new Error(String(err)); + if (!error.message.includes('BUSYGROUP')) { + throw error; + } + } +} + +async function claimStaleMessages(): Promise { + if (!redis) return; + + let startId = '0-0'; + + while (running) { + const response = (await redis.xautoclaim( + config.revalidateStream, + config.revalidateGroup, + consumerName, + claimIdleMs, + startId, + 'COUNT', + batchSize + )) as unknown as [string, Array<[string, string[]]>]; + + const nextId = response[0]; + const messages = response[1] || []; + + if (messages.length === 0) { + break; + } + + await processMessages(messages); + + if (nextId === startId) { + break; + } + startId = nextId; + } +} + +async function readNewMessages(): Promise { + if (!redis) return; + + const response = await redis.xreadgroup( + 'GROUP', + config.revalidateGroup, + consumerName, + 'BLOCK', + blockMs, + 'COUNT', + batchSize, + 'STREAMS', + config.revalidateStream, + '>' + ); + + if (!response) return; + + for (const [, messages] of response) { + await processMessages(messages as Array<[string, string[]]>); + } +} + +async function runLoop(): Promise { + if (!redis) return; + + await ensureGroup(); + + while (running) { + try { + await claimStaleMessages(); + await readNewMessages(); + } catch (err) { + const error = err instanceof Error ? err : new Error(String(err)); + console.error('[Revalidate] Loop error', { error: error.message, stack: error.stack }); + await new Promise((resolve) => setTimeout(resolve, 1000)); + } + } +} + +export async function startRevalidateWorker(): Promise { + if (!config.redisUrl) { + console.warn('[Revalidate] REDIS_URL not set; revalidate worker disabled'); + return; + } + + if (running) return; + + redis = new Redis(config.redisUrl, { + maxRetriesPerRequest: 2, + enableReadyCheck: true, + }); + + redis.on('error', (err) => { + console.error('[Revalidate] Redis error:', err); + }); + + running = true; + loopPromise = runLoop(); +} + +export async function stopRevalidateWorker(): Promise { + running = false; + await loopPromise; + loopPromise = null; + + if (redis) { + const toClose = redis; + redis = null; + await toClose.quit(); + } +} diff --git a/apps/hosting-service/docker-entrypoint.sh b/apps/hosting-service/docker-entrypoint.sh index 62446cf..c73dcde 100644 --- a/apps/hosting-service/docker-entrypoint.sh +++ b/apps/hosting-service/docker-entrypoint.sh @@ -4,29 +4,17 @@ set -e # Run different modes based on MODE environment variable # Modes: # - server (default): Start the hosting service -# - backfill: Run cache backfill and exit -# - backfill-server: Run cache backfill, then start the server MODE="${MODE:-server}" case "$MODE" in - backfill) - echo "šŸ”„ Running in backfill-only mode..." - exec npm run backfill - ;; - backfill-server) - echo "šŸ”„ Running backfill, then starting server..." - npm run backfill - echo "āœ… Backfill complete, starting server..." - exec npm run start - ;; server) echo "šŸš€ Starting server..." exec npm run start ;; *) echo "āŒ Unknown MODE: $MODE" - echo "Valid modes: server, backfill, backfill-server" + echo "Valid modes: server" exit 1 ;; esac diff --git a/apps/hosting-service/package.json b/apps/hosting-service/package.json index 5b49953..454723b 100644 --- a/apps/hosting-service/package.json +++ b/apps/hosting-service/package.json @@ -6,8 +6,7 @@ "dev": "NODE_OPTIONS='--max-old-space-size=2048' tsx --env-file=.env src/index.ts", "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" + "check": "tsc --noEmit" }, "dependencies": { "@atproto/api": "^0.17.4", @@ -23,11 +22,12 @@ "@wispplace/lexicons": "workspace:*", "@wispplace/observability": "workspace:*", "@wispplace/safe-fetch": "workspace:*", + "@wispplace/tiered-storage": "workspace:*", "hono": "^4.10.4", + "ioredis": "^5.9.2", "mime-types": "^2.1.35", "multiformats": "^13.4.1", - "postgres": "^3.4.5", - "@wispplace/tiered-storage": "workspace:*" + "postgres": "^3.4.5" }, "devDependencies": { "@types/bun": "^1.3.1", diff --git a/apps/hosting-service/src/index.ts b/apps/hosting-service/src/index.ts index 1c4c520..f9cf8e3 100644 --- a/apps/hosting-service/src/index.ts +++ b/apps/hosting-service/src/index.ts @@ -1,10 +1,9 @@ import app from './server'; import { serve } from '@hono/node-server'; -import { FirehoseWorker } from './lib/firehose'; -import { createLogger, initializeGrafanaExporters } from '@wispplace/observability'; +import { initializeGrafanaExporters } from '@wispplace/observability'; import { mkdirSync, existsSync } from 'fs'; -import { backfillCache } from './lib/backfill'; -import { startDomainCacheCleanup, stopDomainCacheCleanup, setCacheOnlyMode, closeDatabase } from './lib/db'; +import { startDomainCacheCleanup, stopDomainCacheCleanup, closeDatabase } from './lib/db'; +import { closeRevalidateQueue } from './lib/revalidate-queue'; import { storage, getStorageConfig } from './lib/storage'; // Initialize Grafana exporters if configured @@ -13,27 +12,8 @@ initializeGrafanaExporters({ serviceVersion: '1.0.0' }); -const logger = createLogger('hosting-service'); - const PORT = process.env.PORT ? parseInt(process.env.PORT) : 3001; const CACHE_DIR = process.env.CACHE_DIR || './cache/sites'; -const BACKFILL_CONCURRENCY = process.env.BACKFILL_CONCURRENCY - ? parseInt(process.env.BACKFILL_CONCURRENCY) - : undefined; // Let backfill.ts default (10) apply - -// Parse CLI arguments -const args = process.argv.slice(2); -const hasBackfillFlag = args.includes('--backfill'); -const backfillOnStartup = hasBackfillFlag || process.env.BACKFILL_ON_STARTUP === 'true'; - -// Cache-only mode: service will only cache files locally, no DB writes -const hasCacheOnlyFlag = args.includes('--cache-only'); -export const CACHE_ONLY_MODE = hasCacheOnlyFlag || process.env.CACHE_ONLY_MODE === 'true'; - -// Configure cache-only mode in database module -if (CACHE_ONLY_MODE) { - setCacheOnlyMode(true); -} // Ensure cache directory exists if (!existsSync(CACHE_DIR)) { @@ -44,13 +24,6 @@ if (!existsSync(CACHE_DIR)) { // Start domain cache cleanup startDomainCacheCleanup(); -// Start firehose worker with observability logger -const firehose = new FirehoseWorker((msg, data) => { - logger.info(msg, data); -}); - -firehose.start(); - // Optional: Bootstrap hot cache from warm tier on startup const BOOTSTRAP_HOT_ON_STARTUP = process.env.BOOTSTRAP_HOT_ON_STARTUP === 'true'; const BOOTSTRAP_HOT_LIMIT = process.env.BOOTSTRAP_HOT_LIMIT ? parseInt(process.env.BOOTSTRAP_HOT_LIMIT) : 100; @@ -66,27 +39,12 @@ if (BOOTSTRAP_HOT_ON_STARTUP) { }); } -// Run backfill if requested -if (backfillOnStartup) { - console.log('šŸ”„ Backfill requested, starting cache backfill...'); - backfillCache({ - skipExisting: true, - concurrency: BACKFILL_CONCURRENCY, - }).then((stats) => { - console.log('āœ… Cache backfill completed'); - }).catch((err) => { - console.error('āŒ Cache backfill error:', err); - }); -} - // Add health check endpoint app.get('/health', async (c) => { - const firehoseHealth = firehose.getHealth(); const storageStats = await storage.getStats(); return c.json({ status: 'ok', - firehose: firehoseHealth, storage: storageStats, }); }); @@ -101,12 +59,10 @@ const server = serve({ const storageConfig = getStorageConfig(); console.log(` -Wisp Hosting Service with Tiered Storage +Wisp Hosting Service (Read-Only) with Tiered Storage Server: http://localhost:${PORT} Health: http://localhost:${PORT}/health -Cache-Only: ${CACHE_ONLY_MODE ? 'ENABLED (no DB writes)' : 'DISABLED'} -Backfill: ${backfillOnStartup ? `ENABLED (concurrency: ${BACKFILL_CONCURRENCY || 10})` : 'DISABLED'} Tiered Storage Configuration: Hot Cache: ${storageConfig.hotCacheSize} (${storageConfig.hotCacheCount} items max) @@ -117,14 +73,14 @@ Tiered Storage Configuration: S3 Prefix: ${storageConfig.s3Prefix} Metadata Bucket: ${storageConfig.metadataBucket} -Firehose: Connecting... +Firehose: DISABLED (read-only) `); // Graceful shutdown process.on('SIGINT', async () => { console.log('\nšŸ›‘ Shutting down...'); - await firehose.stop(); stopDomainCacheCleanup(); + await closeRevalidateQueue(); await closeDatabase(); server.close(); process.exit(0); @@ -132,8 +88,8 @@ process.on('SIGINT', async () => { process.on('SIGTERM', async () => { console.log('\nšŸ›‘ Shutting down...'); - await firehose.stop(); stopDomainCacheCleanup(); + await closeRevalidateQueue(); await closeDatabase(); server.close(); process.exit(0); diff --git a/apps/hosting-service/src/lib/backfill.ts b/apps/hosting-service/src/lib/backfill.ts deleted file mode 100644 index c3a50aa..0000000 --- a/apps/hosting-service/src/lib/backfill.ts +++ /dev/null @@ -1,159 +0,0 @@ -import { getAllSites } from './db'; -import { fetchSiteRecord, getPdsForDid, downloadAndCacheSite, isCached } from './utils'; -import { createLogger } from '@wispplace/observability'; -import { markSiteAsBeingCached, unmarkSiteAsBeingCached } from './cache'; -import { clearRedirectRulesCache } from './site-cache'; - -const logger = createLogger('hosting-service'); - -export interface BackfillOptions { - skipExisting?: boolean; // Skip sites already in cache - concurrency?: number; // Number of sites to cache concurrently - maxSites?: number; // Maximum number of sites to backfill (for testing) -} - -export interface BackfillStats { - total: number; - cached: number; - skipped: number; - failed: number; - duration: number; -} - -/** - * Backfill all sites from the database into the local cache - */ -export async function backfillCache(options: BackfillOptions = {}): Promise { - const { - skipExisting = true, - concurrency = 10, // Increased from 3 to 10 for better parallelization - maxSites, - } = options; - - const startTime = Date.now(); - const stats: BackfillStats = { - total: 0, - cached: 0, - skipped: 0, - failed: 0, - duration: 0, - }; - - logger.info('Starting cache backfill', { skipExisting, concurrency, maxSites }); - console.log(` -╔══════════════════════════════════════════╗ -ā•‘ CACHE BACKFILL STARTING ā•‘ -ā•šā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā• - `); - - try { - // Get all sites from database - let sites = await getAllSites(); - stats.total = sites.length; - - logger.info(`Found ${sites.length} sites in database`); - console.log(`šŸ“Š Found ${sites.length} sites in database`); - - // Limit if specified - if (maxSites && maxSites > 0) { - sites = sites.slice(0, maxSites); - console.log(`āš™ļø Limited to ${maxSites} sites for backfill`); - } - - // Process sites with sliding window concurrency pool - const executing = new Set>(); - let processed = 0; - - for (const site of sites) { - // Create task for this site - const processSite = async () => { - try { - // Check if already cached - if (skipExisting && await isCached(site.did, site.rkey)) { - stats.skipped++; - processed++; - logger.debug(`Skipping already cached site`, { did: site.did, rkey: site.rkey }); - console.log(`ā­ļø [${processed}/${sites.length}] Skipped (cached): ${site.display_name || site.rkey}`); - return; - } - - // Fetch site record - const siteData = await fetchSiteRecord(site.did, site.rkey); - if (!siteData) { - stats.failed++; - processed++; - logger.error('Site record not found during backfill', null, { did: site.did, rkey: site.rkey }); - console.log(`āŒ [${processed}/${sites.length}] Failed (not found): ${site.display_name || site.rkey}`); - return; - } - - // Get PDS endpoint - const pdsEndpoint = await getPdsForDid(site.did); - if (!pdsEndpoint) { - stats.failed++; - processed++; - logger.error('PDS not found during backfill', null, { did: site.did }); - console.log(`āŒ [${processed}/${sites.length}] Failed (no PDS): ${site.display_name || site.rkey}`); - return; - } - - // Mark site as being cached to prevent serving stale content during update - markSiteAsBeingCached(site.did, site.rkey); - - try { - // Download and cache site - console.log(`[Backfill] Caching site from backfill - ${site.did}:${site.rkey}`); - await downloadAndCacheSite(site.did, site.rkey, siteData.record, pdsEndpoint, siteData.cid); - // Clear redirect rules cache since the site was updated - clearRedirectRulesCache(site.did, site.rkey); - stats.cached++; - processed++; - logger.info('Successfully cached site during backfill', { did: site.did, rkey: site.rkey }); - console.log(`āœ… [${processed}/${sites.length}] Cached (backfill): ${site.display_name || site.rkey}`); - } finally { - // Always unmark, even if caching fails - unmarkSiteAsBeingCached(site.did, site.rkey); - } - } catch (err) { - stats.failed++; - processed++; - logger.error('Failed to cache site during backfill', err, { did: site.did, rkey: site.rkey }); - console.log(`āŒ [${processed}/${sites.length}] Failed: ${site.display_name || site.rkey}`); - } - }; - - // Add to executing pool and remove when done - const promise = processSite().finally(() => executing.delete(promise)); - executing.add(promise); - - // When pool is full, wait for at least one to complete - if (executing.size >= concurrency) { - await Promise.race(executing); - } - } - - // Wait for all remaining tasks to complete - await Promise.all(executing); - - stats.duration = Date.now() - startTime; - - console.log(` -╔══════════════════════════════════════════╗ -ā•‘ CACHE BACKFILL COMPLETED ā•‘ -ā•šā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā•ā• - -šŸ“Š Total Sites: ${stats.total} -āœ… Cached: ${stats.cached} -ā­ļø Skipped: ${stats.skipped} -āŒ Failed: ${stats.failed} -ā±ļø Duration: ${(stats.duration / 1000).toFixed(2)}s - `); - - logger.info('Cache backfill completed', stats); - } catch (err) { - logger.error('Cache backfill failed', err); - console.error('āŒ Cache backfill failed:', err); - } - - return stats; -} diff --git a/apps/hosting-service/src/lib/cache.ts b/apps/hosting-service/src/lib/cache.ts index 77f76dd..8f9aaac 100644 --- a/apps/hosting-service/src/lib/cache.ts +++ b/apps/hosting-service/src/lib/cache.ts @@ -2,7 +2,7 @@ * Cache management for wisp-hosting-service * * With tiered storage, most caching is handled transparently. - * This module tracks sites being cached and manages rewritten HTML cache. + * This module provides a generic LRU cache and exposes storage stats. */ import { storage } from './storage'; @@ -125,64 +125,11 @@ export class LRUCache { } } -// Rewritten HTML cache: stores HTML after path rewriting for subdomain routes -export const rewrittenHtmlCache = new LRUCache(50 * 1024 * 1024, 200); // 50MB for rewritten HTML - -// Helper to generate cache keys for rewritten HTML -export function getCacheKey(did: string, rkey: string, filePath: string, suffix?: string): string { - const base = `${did}:${rkey}:${filePath}`; - return suffix ? `${base}:${suffix}` : base; -} - -/** - * Invalidate site cache via tiered storage - * Also invalidates locally cached rewritten HTML - */ -export async function invalidateSiteCache(did: string, rkey: string): Promise { - // Invalidate in tiered storage - const prefix = `${did}/${rkey}/`; - const deleted = await storage.invalidate(prefix); - - // Invalidate rewritten HTML cache for this site - const sitePrefix = `${did}:${rkey}:`; - let htmlCount = 0; - const cacheKeys = Array.from((rewrittenHtmlCache as any).cache?.keys() || []) as string[]; - for (const key of cacheKeys) { - if (key.startsWith(sitePrefix)) { - rewrittenHtmlCache.delete(key); - htmlCount++; - } - } - - console.log(`[Cache] Invalidated site ${did}:${rkey} - ${deleted} files in tiered storage, ${htmlCount} rewritten HTML`); -} - -// Track sites currently being cached (to prevent serving stale cache during updates) -const sitesBeingCached = new Set(); - -export function markSiteAsBeingCached(did: string, rkey: string): void { - const key = `${did}:${rkey}`; - sitesBeingCached.add(key); -} - -export function unmarkSiteAsBeingCached(did: string, rkey: string): void { - const key = `${did}:${rkey}`; - sitesBeingCached.delete(key); -} - -export function isSiteBeingCached(did: string, rkey: string): boolean { - const key = `${did}:${rkey}`; - return sitesBeingCached.has(key); -} - // Get overall cache statistics export async function getCacheStats() { const tieredStats = await storage.getStats(); return { tieredStorage: tieredStats, - rewrittenHtml: rewrittenHtmlCache.getStats(), - rewrittenHtmlHitRate: rewrittenHtmlCache.getHitRate(), - sitesBeingCached: sitesBeingCached.size, }; } diff --git a/apps/hosting-service/src/lib/db.ts b/apps/hosting-service/src/lib/db.ts index 9f1e843..294d775 100644 --- a/apps/hosting-service/src/lib/db.ts +++ b/apps/hosting-service/src/lib/db.ts @@ -2,16 +2,6 @@ import postgres from 'postgres'; import { createHash } from 'crypto'; import type { DomainLookup, CustomDomainLookup, SiteCache, SiteSettingsCache } from '@wispplace/database'; -// Global cache-only mode flag (set by index.ts) -let cacheOnlyMode = false; - -export function setCacheOnlyMode(enabled: boolean) { - cacheOnlyMode = enabled; - if (enabled) { - console.log('[DB] Cache-only mode enabled - database writes will be skipped'); - } -} - const sql = postgres( process.env.DATABASE_URL || 'postgres://postgres:postgres@localhost:5432/wisp', { @@ -126,30 +116,7 @@ export async function getCustomDomainByHash(hash: string): Promise { +export async function getSiteSettingsCache(did: string, rkey: string): Promise { const result = await sql` - SELECT did, record_cid, directory_listing, spa_mode, custom_404, index_files, clean_urls, headers, cached_at, updated_at + SELECT did, rkey, record_cid, directory_listing, spa_mode, custom_404, index_files, clean_urls, headers, cached_at, updated_at FROM site_settings_cache - WHERE did = ${did} + WHERE did = ${did} AND rkey = ${rkey} LIMIT 1 `; return result[0] || null; diff --git a/apps/hosting-service/src/lib/file-serving.ts b/apps/hosting-service/src/lib/file-serving.ts index e3cd3f4..b5b309f 100644 --- a/apps/hosting-service/src/lib/file-serving.ts +++ b/apps/hosting-service/src/lib/file-serving.ts @@ -3,18 +3,21 @@ * Handles file retrieval, caching, redirects, and HTML rewriting */ -import { readFile } from 'fs/promises'; import { lookup } from 'mime-types'; +import { gunzipSync } from 'zlib'; import type { Record as WispSettings } from '@wispplace/lexicons/types/place/wisp/settings'; import { shouldCompressMimeType } from '@wispplace/atproto-utils/compression'; -import { rewrittenHtmlCache, getCacheKey, isSiteBeingCached } from './cache'; -import { getCachedFilePath, getCachedSettings } from './utils'; +import { getCachedSettings } from './utils'; import { loadRedirectRules, matchRedirectRule, parseCookies, parseQueryString } from './redirects'; -import { rewriteHtmlPaths, isHtmlContent } from './html-rewriter'; -import { generate404Page, generateDirectoryListing, siteUpdatingResponse } from './page-generators'; -import { getIndexFiles, applyCustomHeaders, fileExists } from './request-utils'; +import { isHtmlContent } from './html-rewriter'; +import { generate404Page, generateDirectoryListing } from './page-generators'; +import { getIndexFiles, applyCustomHeaders } from './request-utils'; import { getRedirectRulesFromCache, setRedirectRulesInCache } from './site-cache'; import { storage } from './storage'; +import { getSiteCache } from './db'; +import { enqueueRevalidate } from './revalidate-queue'; +import { recordStorageMiss } from './revalidate-metrics'; +import { normalizeFileCids } from '@wispplace/fs-utils'; /** * Helper to retrieve a file with metadata from tiered storage @@ -37,6 +40,114 @@ async function getFileWithMetadata(did: string, rkey: string, filePath: string) return result; } +function buildStorageKey(did: string, rkey: string, filePath: string): string { + const normalized = filePath.startsWith('/') ? filePath.slice(1) : filePath; + return `${did}/${rkey}/${normalized}`; +} + +async function storageExists(did: string, rkey: string, filePath: string): Promise { + const key = buildStorageKey(did, rkey, filePath); + return storage.exists(key); +} + +function buildStorageMissResponse(): Response { + return new Response('Storage temporarily unavailable', { + status: 503, + headers: { + 'Cache-Control': 'no-store', + 'Retry-After': '5', + }, + }); +} + +async function listDirectoryEntries( + did: string, + rkey: string, + requestPath: string +): Promise> { + const prefix = buildStorageKey(did, rkey, requestPath ? `${requestPath}/` : ''); + const entries = new Map(); + + for await (const key of storage.listKeys(prefix)) { + const relative = key.slice(prefix.length); + if (!relative) continue; + if (relative.startsWith('.rewritten/')) continue; + + const [name, ...rest] = relative.split('/'); + if (!name || name === '.metadata.json' || name.endsWith('.meta')) continue; + + const isDirectory = rest.length > 0; + const existing = entries.get(name); + if (existing === undefined || (isDirectory && !existing)) { + entries.set(name, isDirectory); + } + } + + return Array.from(entries.entries()).map(([name, isDirectory]) => ({ name, isDirectory })); +} + +async function getFileForRequest( + did: string, + rkey: string, + filePath: string, + preferRewrittenHtml: boolean +): Promise<{ result: Awaited>; filePath: string } | null> { + const mimeTypeGuess = lookup(filePath) || 'application/octet-stream'; + if (preferRewrittenHtml && isHtmlContent(filePath, mimeTypeGuess)) { + const rewrittenPath = `.rewritten/${filePath}`; + const rewritten = await getFileWithMetadata(did, rkey, rewrittenPath); + if (rewritten) { + return { result: rewritten, filePath }; + } + } + + const result = await getFileWithMetadata(did, rkey, filePath); + if (!result) return null; + return { result, filePath }; +} + +function buildResponseFromStorageResult( + result: Awaited>, + filePath: string, + settings: WispSettings | null, + requestHeaders?: Record +): Response { + const content = Buffer.from(result.data); + const meta = result.metadata.customMetadata as { encoding?: string; mimeType?: string } | undefined; + const mimeType = meta?.mimeType || lookup(filePath) || 'application/octet-stream'; + + const headers: Record = { + 'Content-Type': mimeType, + 'Cache-Control': mimeType.startsWith('text/html') + ? 'public, max-age=300' + : 'public, max-age=31536000, immutable', + 'X-Cache-Tier': result.source, + }; + + if (meta?.encoding === 'gzip') { + const shouldServeCompressed = shouldCompressMimeType(mimeType); + const acceptEncoding = requestHeaders?.['accept-encoding'] ?? ''; + const clientAcceptsGzip = acceptEncoding.includes('gzip'); + const hasGzipMagic = content.length >= 2 && content[0] === 0x1f && content[1] === 0x8b; + + if (!clientAcceptsGzip || !shouldServeCompressed) { + if (hasGzipMagic) { + const decompressed = gunzipSync(content); + applyCustomHeaders(headers, filePath, settings); + return new Response(decompressed, { headers }); + } + console.warn(`File ${filePath} marked as gzipped but lacks magic bytes, serving as-is`); + applyCustomHeaders(headers, filePath, settings); + return new Response(content, { headers }); + } + + headers['Content-Encoding'] = 'gzip'; + } + + applyCustomHeaders(headers, filePath, settings); + return new Response(content, { headers }); +} + /** * Helper to serve files from cache (for custom domains and subdomains) */ @@ -83,12 +194,11 @@ export async function serveFromCache( checkPath += indexFiles[0] || 'index.html'; } - const cachedFile = getCachedFilePath(did, rkey, checkPath); - const fileExistsOnDisk = await fileExists(cachedFile); + const fileExistsInStorage = await storageExists(did, rkey, checkPath); // If file exists and redirect is not forced, serve the file normally - if (fileExistsOnDisk) { - return serveFileInternal(did, rkey, filePath, settings); + if (fileExistsInStorage) { + return serveFileInternal(did, rkey, filePath, settings, headers); } } @@ -97,7 +207,7 @@ export async function serveFromCache( // Rewrite: serve different content but keep URL the same // Remove leading slash for internal path resolution const rewritePath = targetPath.startsWith('/') ? targetPath.slice(1) : targetPath; - return serveFileInternal(did, rkey, rewritePath, settings); + return serveFileInternal(did, rkey, rewritePath, settings, headers); } else if (status === 301 || status === 302) { // External redirect: change the URL return new Response(null, { @@ -110,7 +220,7 @@ export async function serveFromCache( } else if (status === 404) { // Custom 404 page from _redirects (wins over settings.custom404) const custom404Path = targetPath.startsWith('/') ? targetPath.slice(1) : targetPath; - const response = await serveFileInternal(did, rkey, custom404Path, settings); + const response = await serveFileInternal(did, rkey, custom404Path, settings, headers); // Override status to 404 return new Response(response.body, { status: 404, @@ -121,7 +231,7 @@ export async function serveFromCache( } // No redirect matched, serve normally with settings - return serveFileInternal(did, rkey, filePath, settings); + return serveFileInternal(did, rkey, filePath, settings, headers); } /** @@ -131,12 +241,39 @@ export async function serveFileInternal( did: string, rkey: string, filePath: string, - settings: WispSettings | null = null + settings: WispSettings | null = null, + requestHeaders?: Record ): Promise { - // Check if site is currently being cached - if so, return updating response - if (isSiteBeingCached(did, rkey)) { - return siteUpdatingResponse(); - } + let expectedFileCids: Record | null | undefined; + let expectedMissPath: string | null = null; + + const getExpectedFileCids = async (): Promise | null> => { + if (expectedFileCids !== undefined) return expectedFileCids; + const siteCache = await getSiteCache(did, rkey); + if (!siteCache) { + expectedFileCids = null; + return null; + } + expectedFileCids = normalizeFileCids(siteCache.file_cids).value; + return expectedFileCids; + }; + + const markExpectedMiss = async (path: string) => { + if (expectedMissPath) return; + const fileCids = await getExpectedFileCids(); + if (!fileCids) return; + const normalized = path.startsWith('/') ? path.slice(1) : path; + if (fileCids[normalized]) { + expectedMissPath = normalized; + } + }; + + const maybeReturnStorageMiss = async (): Promise => { + if (!expectedMissPath) return null; + recordStorageMiss(expectedMissPath); + await enqueueRevalidate(did, rkey, `storage-miss:${expectedMissPath}`); + return buildStorageMissResponse(); + }; const indexFiles = getIndexFiles(settings); @@ -146,54 +283,30 @@ export async function serveFileInternal( requestPath = requestPath.slice(0, -1); } - // Check if this path is a directory first - const directoryPath = getCachedFilePath(did, rkey, requestPath); - if (await fileExists(directoryPath)) { - const { stat, readdir } = await import('fs/promises'); - try { - const stats = await stat(directoryPath); - if (stats.isDirectory()) { - // It's a directory, try each index file in order - for (const indexFile of indexFiles) { - const indexPath = requestPath ? `${requestPath}/${indexFile}` : indexFile; - const indexFilePath = getCachedFilePath(did, rkey, indexPath); - if (await fileExists(indexFilePath)) { - return serveFileInternal(did, rkey, indexPath, settings); - } - } - // No index file found - check if directory listing is enabled - if (settings?.directoryListing) { - const { stat } = await import('fs/promises'); - const entries = await readdir(directoryPath); - // Filter out .meta files and other hidden files - const visibleEntries = entries.filter(entry => !entry.endsWith('.meta') && entry !== '.metadata.json'); - - // Check which entries are directories - const entriesWithType = await Promise.all( - visibleEntries.map(async (name) => { - try { - const entryPath = `${directoryPath}/${name}`; - const stats = await stat(entryPath); - return { name, isDirectory: stats.isDirectory() }; - } catch { - return { name, isDirectory: false }; - } - }) - ); - - const html = generateDirectoryListing(requestPath, entriesWithType); - return new Response(html, { - headers: { - 'Content-Type': 'text/html; charset=utf-8', - 'Cache-Control': 'public, max-age=300', - }, - }); - } - // Fall through to 404/SPA handling + // Check if this path is a directory first (best-effort via prefix scan) + const directoryEntries = await listDirectoryEntries(did, rkey, requestPath); + if (directoryEntries.length > 0) { + // It's a directory, try each index file in order + for (const indexFile of indexFiles) { + const indexPath = requestPath ? `${requestPath}/${indexFile}` : indexFile; + if (await storageExists(did, rkey, indexPath)) { + return serveFileInternal(did, rkey, indexPath, settings, requestHeaders); } - } catch (err) { - // If stat fails, continue with normal flow + await markExpectedMiss(indexPath); } + // No index file found - check if directory listing is enabled + if (settings?.directoryListing) { + const missResponse = await maybeReturnStorageMiss(); + if (missResponse) return missResponse; + const html = generateDirectoryListing(requestPath, directoryEntries); + return new Response(html, { + headers: { + 'Content-Type': 'text/html; charset=utf-8', + 'Cache-Control': 'public, max-age=300', + }, + }); + } + // Fall through to 404/SPA handling } // Not a directory, try to serve as a file @@ -203,55 +316,9 @@ export async function serveFileInternal( const result = await getFileWithMetadata(did, rkey, fileRequestPath); if (result) { - const content = Buffer.from(result.data); - const meta = result.metadata.customMetadata as { encoding?: string; mimeType?: string } | undefined; - - // Build headers with caching - const headers: Record = { - 'X-Cache-Tier': result.source, - }; - - if (meta?.encoding === 'gzip' && meta.mimeType) { - const shouldServeCompressed = shouldCompressMimeType(meta.mimeType); - - if (!shouldServeCompressed) { - // Verify content is actually gzipped before attempting decompression - const isGzipped = content.length >= 2 && content[0] === 0x1f && content[1] === 0x8b; - if (isGzipped) { - const { gunzipSync } = await import('zlib'); - const decompressed = gunzipSync(content); - headers['Content-Type'] = meta.mimeType; - headers['Cache-Control'] = 'public, max-age=31536000, immutable'; - applyCustomHeaders(headers, fileRequestPath, settings); - return new Response(decompressed, { headers }); - } else { - // Meta says gzipped but content isn't - serve as-is - console.warn(`File ${filePath} has gzip encoding in meta but content lacks gzip magic bytes`); - headers['Content-Type'] = meta.mimeType; - headers['Cache-Control'] = 'public, max-age=31536000, immutable'; - applyCustomHeaders(headers, fileRequestPath, settings); - return new Response(content, { headers }); - } - } - - headers['Content-Type'] = meta.mimeType; - headers['Content-Encoding'] = 'gzip'; - headers['Cache-Control'] = meta.mimeType.startsWith('text/html') - ? 'public, max-age=300' - : 'public, max-age=31536000, immutable'; - applyCustomHeaders(headers, fileRequestPath, settings); - return new Response(content, { headers }); - } - - // Non-compressed files - const mimeType = meta?.mimeType || lookup(fileRequestPath) || 'application/octet-stream'; - headers['Content-Type'] = mimeType; - headers['Cache-Control'] = mimeType.startsWith('text/html') - ? 'public, max-age=300' - : 'public, max-age=31536000, immutable'; - applyCustomHeaders(headers, fileRequestPath, settings); - return new Response(content, { headers }); + return buildResponseFromStorageResult(result, fileRequestPath, settings, requestHeaders); } + await markExpectedMiss(fileRequestPath); // Try index files for directory-like paths if (!fileRequestPath.includes('.')) { @@ -277,106 +344,85 @@ export async function serveFileInternal( applyCustomHeaders(headers, indexPath, settings); return new Response(indexContent, { headers }); } + await markExpectedMiss(indexPath); } } // Try clean URLs: /about -> /about.html if (settings?.cleanUrls && !fileRequestPath.includes('.')) { const htmlPath = `${fileRequestPath}.html`; - const htmlFile = getCachedFilePath(did, rkey, htmlPath); - if (await fileExists(htmlFile)) { - return serveFileInternal(did, rkey, htmlPath, settings); + if (await storageExists(did, rkey, htmlPath)) { + return serveFileInternal(did, rkey, htmlPath, settings, requestHeaders); } + await markExpectedMiss(htmlPath); // Also try /about/index.html for (const indexFileName of indexFiles) { const indexPath = fileRequestPath ? `${fileRequestPath}/${indexFileName}` : indexFileName; - const indexFile = getCachedFilePath(did, rkey, indexPath); - if (await fileExists(indexFile)) { - return serveFileInternal(did, rkey, indexPath, settings); + if (await storageExists(did, rkey, indexPath)) { + return serveFileInternal(did, rkey, indexPath, settings, requestHeaders); } + await markExpectedMiss(indexPath); } } // SPA mode: serve SPA file for all non-existing routes (wins over custom404 but loses to _redirects) if (settings?.spaMode) { const spaFile = settings.spaMode; - const spaFilePath = getCachedFilePath(did, rkey, spaFile); - if (await fileExists(spaFilePath)) { - return serveFileInternal(did, rkey, spaFile, settings); + if (await storageExists(did, rkey, spaFile)) { + return serveFileInternal(did, rkey, spaFile, settings, requestHeaders); } + await markExpectedMiss(spaFile); } // Custom 404: serve custom 404 file if configured (wins conflict battle) if (settings?.custom404) { const custom404File = settings.custom404; - const custom404Path = getCachedFilePath(did, rkey, custom404File); - if (await fileExists(custom404Path)) { - const response: Response = await serveFileInternal(did, rkey, custom404File, settings); + if (await storageExists(did, rkey, custom404File)) { + const response: Response = await serveFileInternal(did, rkey, custom404File, settings, requestHeaders); // Override status to 404 return new Response(response.body, { status: 404, headers: response.headers, }); } + await markExpectedMiss(custom404File); } // Autodetect 404 pages (GitHub Pages: 404.html, Neocities/Nekoweb: not_found.html) const auto404Pages = ['404.html', 'not_found.html']; for (const auto404Page of auto404Pages) { - const auto404Path = getCachedFilePath(did, rkey, auto404Page); - if (await fileExists(auto404Path)) { - const response: Response = await serveFileInternal(did, rkey, auto404Page, settings); + if (await storageExists(did, rkey, auto404Page)) { + const response: Response = await serveFileInternal(did, rkey, auto404Page, settings, requestHeaders); // Override status to 404 return new Response(response.body, { status: 404, headers: response.headers, }); } + await markExpectedMiss(auto404Page); } // Directory listing fallback: if enabled, show root directory listing on 404 if (settings?.directoryListing) { - const rootPath = getCachedFilePath(did, rkey, ''); - if (await fileExists(rootPath)) { - const { stat, readdir } = await import('fs/promises'); - try { - const stats = await stat(rootPath); - if (stats.isDirectory()) { - const entries = await readdir(rootPath); - // Filter out .meta files and metadata - const visibleEntries = entries.filter(entry => - !entry.endsWith('.meta') && entry !== '.metadata.json' - ); - - // Check which entries are directories - const entriesWithType = await Promise.all( - visibleEntries.map(async (name) => { - try { - const entryPath = `${rootPath}/${name}`; - const entryStats = await stat(entryPath); - return { name, isDirectory: entryStats.isDirectory() }; - } catch { - return { name, isDirectory: false }; - } - }) - ); - - const html = generateDirectoryListing('', entriesWithType); - return new Response(html, { - status: 404, - headers: { - 'Content-Type': 'text/html; charset=utf-8', - 'Cache-Control': 'public, max-age=300', - }, - }); - } - } catch (err) { - // If directory listing fails, fall through to 404 - } + const rootEntries = await listDirectoryEntries(did, rkey, ''); + if (rootEntries.length > 0) { + const missResponse = await maybeReturnStorageMiss(); + if (missResponse) return missResponse; + const html = generateDirectoryListing('', rootEntries); + return new Response(html, { + status: 404, + headers: { + 'Content-Type': 'text/html; charset=utf-8', + 'Cache-Control': 'public, max-age=300', + }, + }); } } + const missResponse = await maybeReturnStorageMiss(); + if (missResponse) return missResponse; + // Default styled 404 page const html = generate404Page(); return new Response(html, { @@ -435,12 +481,11 @@ export async function serveFromCacheWithRewrite( checkPath += indexFiles[0] || 'index.html'; } - const cachedFile = getCachedFilePath(did, rkey, checkPath); - const fileExistsOnDisk = await fileExists(cachedFile); + const fileExistsInStorage = await storageExists(did, rkey, checkPath); // If file exists and redirect is not forced, serve the file normally - if (fileExistsOnDisk) { - return serveFileInternalWithRewrite(did, rkey, filePath, basePath, settings); + if (fileExistsInStorage) { + return serveFileInternalWithRewrite(did, rkey, filePath, basePath, settings, headers); } } @@ -448,7 +493,7 @@ export async function serveFromCacheWithRewrite( if (status === 200) { // Rewrite: serve different content but keep URL the same const rewritePath = targetPath.startsWith('/') ? targetPath.slice(1) : targetPath; - return serveFileInternalWithRewrite(did, rkey, rewritePath, basePath, settings); + return serveFileInternalWithRewrite(did, rkey, rewritePath, basePath, settings, headers); } else if (status === 301 || status === 302) { // External redirect: change the URL // For sites.wisp.place, we need to adjust the target path to include the base path @@ -467,7 +512,7 @@ export async function serveFromCacheWithRewrite( } else if (status === 404) { // Custom 404 page from _redirects (wins over settings.custom404) const custom404Path = targetPath.startsWith('/') ? targetPath.slice(1) : targetPath; - const response = await serveFileInternalWithRewrite(did, rkey, custom404Path, basePath, settings); + const response = await serveFileInternalWithRewrite(did, rkey, custom404Path, basePath, settings, headers); // Override status to 404 return new Response(response.body, { status: 404, @@ -478,7 +523,7 @@ export async function serveFromCacheWithRewrite( } // No redirect matched, serve normally with settings - return serveFileInternalWithRewrite(did, rkey, filePath, basePath, settings); + return serveFileInternalWithRewrite(did, rkey, filePath, basePath, settings, headers); } /** @@ -489,12 +534,39 @@ export async function serveFileInternalWithRewrite( rkey: string, filePath: string, basePath: string, - settings: WispSettings | null = null + settings: WispSettings | null = null, + requestHeaders?: Record ): Promise { - // Check if site is currently being cached - if so, return updating response - if (isSiteBeingCached(did, rkey)) { - return siteUpdatingResponse(); - } + let expectedFileCids: Record | null | undefined; + let expectedMissPath: string | null = null; + + const getExpectedFileCids = async (): Promise | null> => { + if (expectedFileCids !== undefined) return expectedFileCids; + const siteCache = await getSiteCache(did, rkey); + if (!siteCache) { + expectedFileCids = null; + return null; + } + expectedFileCids = normalizeFileCids(siteCache.file_cids).value; + return expectedFileCids; + }; + + const markExpectedMiss = async (path: string) => { + if (expectedMissPath) return; + const fileCids = await getExpectedFileCids(); + if (!fileCids) return; + const normalized = path.startsWith('/') ? path.slice(1) : path; + if (fileCids[normalized]) { + expectedMissPath = normalized; + } + }; + + const maybeReturnStorageMiss = async (): Promise => { + if (!expectedMissPath) return null; + recordStorageMiss(expectedMissPath); + await enqueueRevalidate(did, rkey, `storage-miss:${expectedMissPath}`); + return buildStorageMissResponse(); + }; const indexFiles = getIndexFiles(settings); @@ -504,329 +576,128 @@ export async function serveFileInternalWithRewrite( requestPath = requestPath.slice(0, -1); } - // Check if this path is a directory first - const directoryPath = getCachedFilePath(did, rkey, requestPath); - if (await fileExists(directoryPath)) { - const { stat, readdir } = await import('fs/promises'); - try { - const stats = await stat(directoryPath); - if (stats.isDirectory()) { - // It's a directory, try each index file in order - for (const indexFile of indexFiles) { - const indexPath = requestPath ? `${requestPath}/${indexFile}` : indexFile; - const indexFilePath = getCachedFilePath(did, rkey, indexPath); - if (await fileExists(indexFilePath)) { - return serveFileInternalWithRewrite(did, rkey, indexPath, basePath, settings); - } - } - // No index file found - check if directory listing is enabled - if (settings?.directoryListing) { - const { stat } = await import('fs/promises'); - const entries = await readdir(directoryPath); - // Filter out .meta files and other hidden files - const visibleEntries = entries.filter(entry => !entry.endsWith('.meta') && entry !== '.metadata.json'); - - // Check which entries are directories - const entriesWithType = await Promise.all( - visibleEntries.map(async (name) => { - try { - const entryPath = `${directoryPath}/${name}`; - const stats = await stat(entryPath); - return { name, isDirectory: stats.isDirectory() }; - } catch { - return { name, isDirectory: false }; - } - }) - ); - - const html = generateDirectoryListing(requestPath, entriesWithType); - return new Response(html, { - headers: { - 'Content-Type': 'text/html; charset=utf-8', - 'Cache-Control': 'public, max-age=300', - }, - }); - } - // Fall through to 404/SPA handling + // Check if this path is a directory first (best-effort via prefix scan) + const directoryEntries = await listDirectoryEntries(did, rkey, requestPath); + if (directoryEntries.length > 0) { + // It's a directory, try each index file in order + for (const indexFile of indexFiles) { + const indexPath = requestPath ? `${requestPath}/${indexFile}` : indexFile; + if (await storageExists(did, rkey, indexPath)) { + return serveFileInternalWithRewrite(did, rkey, indexPath, basePath, settings, requestHeaders); } - } catch (err) { - // If stat fails, continue with normal flow + await markExpectedMiss(indexPath); + } + // No index file found - check if directory listing is enabled + if (settings?.directoryListing) { + const missResponse = await maybeReturnStorageMiss(); + if (missResponse) return missResponse; + const html = generateDirectoryListing(requestPath, directoryEntries); + return new Response(html, { + headers: { + 'Content-Type': 'text/html; charset=utf-8', + 'Cache-Control': 'public, max-age=300', + }, + }); } + // Fall through to 404/SPA handling } // Not a directory, try to serve as a file const fileRequestPath: string = requestPath || indexFiles[0] || 'index.html'; - // Check for rewritten HTML in cache first (if it's HTML) - const mimeTypeGuess = lookup(fileRequestPath) || 'application/octet-stream'; - if (isHtmlContent(fileRequestPath, mimeTypeGuess)) { - const rewrittenKey = getCacheKey(did, rkey, fileRequestPath, `rewritten:${basePath}`); - const rewrittenContent = rewrittenHtmlCache.get(rewrittenKey); - if (rewrittenContent) { - console.log(`[HTML Rewrite] Serving from rewritten cache: ${rewrittenKey}`); - const headers: Record = { - 'Content-Type': 'text/html; charset=utf-8', - 'Content-Encoding': 'gzip', - 'Cache-Control': 'public, max-age=300', - 'X-Cache-Tier': 'local', // Rewritten HTML is stored locally - }; - applyCustomHeaders(headers, fileRequestPath, settings); - return new Response(rewrittenContent, { headers }); - } - } - - // Retrieve from tiered storage - const result = await getFileWithMetadata(did, rkey, fileRequestPath); - - if (result) { - const content = Buffer.from(result.data); - const meta = result.metadata.customMetadata as { encoding?: string; mimeType?: string } | undefined; - const mimeType = meta?.mimeType || lookup(fileRequestPath) || 'application/octet-stream'; - const isGzipped = meta?.encoding === 'gzip'; - - console.log(`[File Serve] Serving ${fileRequestPath}, mimeType: ${mimeType}, isHTML: ${isHtmlContent(fileRequestPath, mimeType)}, basePath: ${basePath}`); - - // Check if this is HTML content that needs rewriting - if (isHtmlContent(fileRequestPath, mimeType)) { - console.log(`[HTML Rewrite] Processing ${fileRequestPath}, basePath: ${basePath}, mimeType: ${mimeType}, isGzipped: ${isGzipped}`); - let htmlContent: string; - if (isGzipped) { - // Verify content is actually gzipped - const hasGzipMagic = content.length >= 2 && content[0] === 0x1f && content[1] === 0x8b; - if (hasGzipMagic) { - const { gunzipSync } = await import('zlib'); - htmlContent = gunzipSync(content).toString('utf-8'); - } else { - console.warn(`File ${fileRequestPath} marked as gzipped but lacks magic bytes, serving as-is`); - htmlContent = content.toString('utf-8'); - } - } else { - htmlContent = content.toString('utf-8'); - } - // Check for tag which can override paths - const baseTagMatch = htmlContent.match(/]*href=["'][^"']+["'][^>]*>/i); - if (baseTagMatch) { - console.warn(`[HTML Rewrite] WARNING: tag found: ${baseTagMatch[0]} - this may override path rewrites`); - } - - // Find src/href attributes (quoted and unquoted) to debug - const allMatches = htmlContent.match(/(?:src|href)\s*=\s*["']?\/[^"'\s>]+/g); - console.log(`[HTML Rewrite] Found ${allMatches ? allMatches.length : 0} local path attrs`); - if (allMatches && allMatches.length > 0) { - console.log(`[HTML Rewrite] Sample paths: ${allMatches.slice(0, 5).join(', ')}`); - } - - const rewritten = rewriteHtmlPaths(htmlContent, basePath, fileRequestPath); - - const rewrittenMatches = rewritten.match(/(?:src|href)\s*=\s*["']?\/[^"'\s>]+/g); - console.log(`[HTML Rewrite] After rewrite, found ${rewrittenMatches ? rewrittenMatches.length : 0} local paths`); - if (rewrittenMatches && rewrittenMatches.length > 0) { - console.log(`[HTML Rewrite] Sample rewritten: ${rewrittenMatches.slice(0, 5).join(', ')}`); - } - - // Recompress and cache the rewritten HTML - const { gzipSync } = await import('zlib'); - const recompressed = gzipSync(Buffer.from(rewritten, 'utf-8')); - - const rewrittenKey = getCacheKey(did, rkey, fileRequestPath, `rewritten:${basePath}`); - rewrittenHtmlCache.set(rewrittenKey, recompressed, recompressed.length); - - const htmlHeaders: Record = { - 'Content-Type': 'text/html; charset=utf-8', - 'Content-Encoding': 'gzip', - 'Cache-Control': 'public, max-age=300', - 'X-Cache-Tier': result.source, - }; - applyCustomHeaders(htmlHeaders, fileRequestPath, settings); - return new Response(recompressed, { headers: htmlHeaders }); - } - - // Non-HTML files: serve as-is - const headers: Record = { - 'Content-Type': mimeType, - 'Cache-Control': 'public, max-age=31536000, immutable', - 'X-Cache-Tier': result.source, - }; - - if (isGzipped) { - const shouldServeCompressed = shouldCompressMimeType(mimeType); - if (!shouldServeCompressed) { - // Verify content is actually gzipped - const hasGzipMagic = content.length >= 2 && content[0] === 0x1f && content[1] === 0x8b; - if (hasGzipMagic) { - const { gunzipSync } = await import('zlib'); - const decompressed = gunzipSync(content); - applyCustomHeaders(headers, fileRequestPath, settings); - return new Response(decompressed, { headers }); - } else { - console.warn(`File ${fileRequestPath} marked as gzipped but lacks magic bytes, serving as-is`); - applyCustomHeaders(headers, fileRequestPath, settings); - return new Response(content, { headers }); - } - } - headers['Content-Encoding'] = 'gzip'; - } - - applyCustomHeaders(headers, fileRequestPath, settings); - return new Response(content, { headers }); + const fileResult = await getFileForRequest(did, rkey, fileRequestPath, true); + if (fileResult) { + return buildResponseFromStorageResult(fileResult.result, fileRequestPath, settings, requestHeaders); } + await markExpectedMiss(fileRequestPath); // Try index files for directory-like paths if (!fileRequestPath.includes('.')) { for (const indexFileName of indexFiles) { const indexPath = fileRequestPath ? `${fileRequestPath}/${indexFileName}` : indexFileName; - - // Check for rewritten index file in cache - const rewrittenKey = getCacheKey(did, rkey, indexPath, `rewritten:${basePath}`); - const rewrittenContent = rewrittenHtmlCache.get(rewrittenKey); - if (rewrittenContent) { - const headers: Record = { - 'Content-Type': 'text/html; charset=utf-8', - 'Content-Encoding': 'gzip', - 'Cache-Control': 'public, max-age=300', - 'X-Cache-Tier': 'local', // Rewritten HTML is stored locally - }; - applyCustomHeaders(headers, indexPath, settings); - return new Response(rewrittenContent, { headers }); - } - - const indexResult = await getFileWithMetadata(did, rkey, indexPath); - + const indexResult = await getFileForRequest(did, rkey, indexPath, true); if (indexResult) { - const indexContent = Buffer.from(indexResult.data); - const indexMeta = indexResult.metadata.customMetadata as { encoding?: string; mimeType?: string } | undefined; - const isGzipped = indexMeta?.encoding === 'gzip'; - - let htmlContent: string; - if (isGzipped) { - // Verify content is actually gzipped - const hasGzipMagic = indexContent.length >= 2 && indexContent[0] === 0x1f && indexContent[1] === 0x8b; - if (hasGzipMagic) { - const { gunzipSync } = await import('zlib'); - htmlContent = gunzipSync(indexContent).toString('utf-8'); - } else { - console.warn(`Index file marked as gzipped but lacks magic bytes, serving as-is`); - htmlContent = indexContent.toString('utf-8'); - } - } else { - htmlContent = indexContent.toString('utf-8'); - } - const rewritten = rewriteHtmlPaths(htmlContent, basePath, indexPath); - - const { gzipSync } = await import('zlib'); - const recompressed = gzipSync(Buffer.from(rewritten, 'utf-8')); - - rewrittenHtmlCache.set(rewrittenKey, recompressed, recompressed.length); - - const headers: Record = { - 'Content-Type': 'text/html; charset=utf-8', - 'Content-Encoding': 'gzip', - 'Cache-Control': 'public, max-age=300', - 'X-Cache-Tier': indexResult.source, - }; - applyCustomHeaders(headers, indexPath, settings); - return new Response(recompressed, { headers }); + return buildResponseFromStorageResult(indexResult.result, indexPath, settings, requestHeaders); } + await markExpectedMiss(indexPath); } } // Try clean URLs: /about -> /about.html if (settings?.cleanUrls && !fileRequestPath.includes('.')) { const htmlPath = `${fileRequestPath}.html`; - const htmlFile = getCachedFilePath(did, rkey, htmlPath); - if (await fileExists(htmlFile)) { - return serveFileInternalWithRewrite(did, rkey, htmlPath, basePath, settings); + if (await storageExists(did, rkey, htmlPath)) { + return serveFileInternalWithRewrite(did, rkey, htmlPath, basePath, settings, requestHeaders); } + await markExpectedMiss(htmlPath); // Also try /about/index.html for (const indexFileName of indexFiles) { const indexPath = fileRequestPath ? `${fileRequestPath}/${indexFileName}` : indexFileName; - const indexFile = getCachedFilePath(did, rkey, indexPath); - if (await fileExists(indexFile)) { - return serveFileInternalWithRewrite(did, rkey, indexPath, basePath, settings); + if (await storageExists(did, rkey, indexPath)) { + return serveFileInternalWithRewrite(did, rkey, indexPath, basePath, settings, requestHeaders); } + await markExpectedMiss(indexPath); } } // SPA mode: serve SPA file for all non-existing routes if (settings?.spaMode) { const spaFile = settings.spaMode; - const spaFilePath = getCachedFilePath(did, rkey, spaFile); - if (await fileExists(spaFilePath)) { - return serveFileInternalWithRewrite(did, rkey, spaFile, basePath, settings); + if (await storageExists(did, rkey, spaFile)) { + return serveFileInternalWithRewrite(did, rkey, spaFile, basePath, settings, requestHeaders); } + await markExpectedMiss(spaFile); } // Custom 404: serve custom 404 file if configured (wins conflict battle) if (settings?.custom404) { const custom404File = settings.custom404; - const custom404Path = getCachedFilePath(did, rkey, custom404File); - if (await fileExists(custom404Path)) { - const response: Response = await serveFileInternalWithRewrite(did, rkey, custom404File, basePath, settings); + if (await storageExists(did, rkey, custom404File)) { + const response: Response = await serveFileInternalWithRewrite(did, rkey, custom404File, basePath, settings, requestHeaders); // Override status to 404 return new Response(response.body, { status: 404, headers: response.headers, }); } + await markExpectedMiss(custom404File); } // Autodetect 404 pages (GitHub Pages: 404.html, Neocities/Nekoweb: not_found.html) const auto404Pages = ['404.html', 'not_found.html']; for (const auto404Page of auto404Pages) { - const auto404Path = getCachedFilePath(did, rkey, auto404Page); - if (await fileExists(auto404Path)) { - const response: Response = await serveFileInternalWithRewrite(did, rkey, auto404Page, basePath, settings); + if (await storageExists(did, rkey, auto404Page)) { + const response: Response = await serveFileInternalWithRewrite(did, rkey, auto404Page, basePath, settings, requestHeaders); // Override status to 404 return new Response(response.body, { status: 404, headers: response.headers, }); } + await markExpectedMiss(auto404Page); } // Directory listing fallback: if enabled, show root directory listing on 404 if (settings?.directoryListing) { - const rootPath = getCachedFilePath(did, rkey, ''); - if (await fileExists(rootPath)) { - const { stat, readdir } = await import('fs/promises'); - try { - const stats = await stat(rootPath); - if (stats.isDirectory()) { - const entries = await readdir(rootPath); - // Filter out .meta files and metadata - const visibleEntries = entries.filter(entry => - !entry.endsWith('.meta') && entry !== '.metadata.json' - ); - - // Check which entries are directories - const entriesWithType = await Promise.all( - visibleEntries.map(async (name) => { - try { - const entryPath = `${rootPath}/${name}`; - const entryStats = await stat(entryPath); - return { name, isDirectory: entryStats.isDirectory() }; - } catch { - return { name, isDirectory: false }; - } - }) - ); - - const html = generateDirectoryListing('', entriesWithType); - return new Response(html, { - status: 404, - headers: { - 'Content-Type': 'text/html; charset=utf-8', - 'Cache-Control': 'public, max-age=300', - }, - }); - } - } catch (err) { - // If directory listing fails, fall through to 404 - } + const rootEntries = await listDirectoryEntries(did, rkey, ''); + if (rootEntries.length > 0) { + const missResponse = await maybeReturnStorageMiss(); + if (missResponse) return missResponse; + const html = generateDirectoryListing('', rootEntries); + return new Response(html, { + status: 404, + headers: { + 'Content-Type': 'text/html; charset=utf-8', + 'Cache-Control': 'public, max-age=300', + }, + }); } } + const missResponse = await maybeReturnStorageMiss(); + if (missResponse) return missResponse; + // Default styled 404 page const html = generate404Page(); return new Response(html, { diff --git a/apps/hosting-service/src/lib/firehose.ts b/apps/hosting-service/src/lib/firehose.ts deleted file mode 100644 index 0a4dbea..0000000 --- a/apps/hosting-service/src/lib/firehose.ts +++ /dev/null @@ -1,512 +0,0 @@ -import { existsSync } from 'fs' -import { - getPdsForDid, - downloadAndCacheSite, - fetchSiteRecord -} from './utils' -import { upsertSite, tryAcquireLock, releaseLock } from './db' -import { safeFetch } from '@wispplace/safe-fetch' -// import { isRecord, validateRecord } from '@wispplace/lexicons/types/place/wisp/fs' -import { isRecord } from '@wispplace/lexicons/types/place/wisp/fs' -import { Firehose } from '@atproto/sync' -import { IdResolver } from '@atproto/identity' -import { invalidateSiteCache, markSiteAsBeingCached, unmarkSiteAsBeingCached } from './cache' -import { clearRedirectRulesCache } from './site-cache' - -const CACHE_DIR = process.env.CACHE_DIR || './cache/sites' - -export class FirehoseWorker { - private firehose: Firehose | null = null - private idResolver: IdResolver - private isShuttingDown = false - private lastEventTime = Date.now() - private eventCount = 0 - private cacheCleanupInterval: NodeJS.Timeout | null = null - private healthCheckInterval: NodeJS.Timeout | null = null - private processingQueue: Set> = new Set() - private readonly maxConcurrency = parseInt(process.env.FIREHOSE_MAX_CONCURRENCY || '5', 10) - - constructor( - private logger?: (msg: string, data?: Record) => void - ) { - this.idResolver = new IdResolver() - this.startCacheCleanup() - } - - private log(msg: string, data?: Record) { - const log = this.logger || console.log - log(`[FirehoseWorker] ${msg}`, data || {}) - } - - /** - * Queue a task with concurrency limiting - * Waits if max concurrent tasks are already running - */ - private async queueTask(task: () => Promise): Promise { - // Wait if we're at max concurrency - if (this.processingQueue.size >= this.maxConcurrency) { - this.log(`Queue at max capacity (${this.maxConcurrency}), waiting for slot...`, { - queueSize: this.processingQueue.size - }) - await Promise.race(this.processingQueue) - } - - // Execute task and track in queue - const promise = task() - .catch(err => { - // Errors are already logged in the handlers - }) - .finally(() => { - this.processingQueue.delete(promise) - }) - - this.processingQueue.add(promise) - - // Don't await here - we want handleEvent to return quickly - // The task will process in the background with concurrency limiting - } - - private startCacheCleanup() { - // Clear IdResolver cache every hour to prevent unbounded memory growth - // The IdResolver has an internal cache that never expires and can cause heap exhaustion - this.cacheCleanupInterval = setInterval(() => { - if (this.isShuttingDown) return - - this.log('Clearing IdResolver cache to prevent memory leak') - - // Recreate the IdResolver to clear its internal cache - this.idResolver = new IdResolver() - - this.log('IdResolver cache cleared') - }, 60 * 60 * 1000) // Every hour - - // Health check: log if no events received for 30 seconds - this.healthCheckInterval = setInterval(() => { - if (this.isShuttingDown) return - - const timeSinceLastEvent = Date.now() - this.lastEventTime - if (timeSinceLastEvent > 30000 && this.eventCount === 0) { - this.log('Warning: No firehose events received in the last 30 seconds', { - timeSinceLastEvent, - eventsReceived: this.eventCount - }) - } else if (timeSinceLastEvent > 60000) { - this.log('Firehose status check', { - timeSinceLastEvent, - eventsReceived: this.eventCount - }) - } - }, 30000) // Every 30 seconds - } - - start() { - this.log('Starting firehose worker') - this.connect() - } - - async stop() { - this.log('Stopping firehose worker') - this.isShuttingDown = true - - if (this.cacheCleanupInterval) { - clearInterval(this.cacheCleanupInterval) - this.cacheCleanupInterval = null - } - - if (this.healthCheckInterval) { - clearInterval(this.healthCheckInterval) - this.healthCheckInterval = null - } - - if (this.firehose) { - this.firehose.destroy() - this.firehose = null - } - - // Wait for all queued tasks to complete - if (this.processingQueue.size > 0) { - this.log(`Waiting for ${this.processingQueue.size} queued tasks to complete...`) - await Promise.all(this.processingQueue) - this.log('All queued tasks completed') - } - } - - private connect() { - if (this.isShuttingDown) return - - this.log('Connecting to AT Protocol firehose') - - this.firehose = new Firehose({ - idResolver: this.idResolver, - service: 'wss://bsky.network', - filterCollections: ['place.wisp.fs', 'place.wisp.settings'], - handleEvent: async (evt: any) => { - this.lastEventTime = Date.now() - this.eventCount++ - - if (this.eventCount === 1) { - this.log('First firehose event received - connection established', { - eventType: evt.event, - collection: evt.collection - }) - } - - // Watch for write events - if (evt.event === 'create' || evt.event === 'update') { - const record = evt.record - - // If the write is a valid place.wisp.fs record - if ( - evt.collection === 'place.wisp.fs' && - isRecord(record) - // && validateRecord(record).success - ) { - this.log('Received place.wisp.fs event', { - did: evt.did, - event: evt.event, - rkey: evt.rkey - }) - - await this.queueTask(async () => { - try { - await this.handleCreateOrUpdate( - evt.did, - evt.rkey, - record, - evt.cid?.toString() - ) - } catch (err) { - console.error('Full error details:', err); - this.log('Error handling event', { - did: evt.did, - event: evt.event, - rkey: evt.rkey, - error: - err instanceof Error - ? err.message - : String(err) - }) - } - }) - } - // Handle settings changes - else if (evt.collection === 'place.wisp.settings') { - this.log('Received place.wisp.settings event', { - did: evt.did, - event: evt.event, - rkey: evt.rkey - }) - - await this.queueTask(async () => { - try { - await this.handleSettingsChange(evt.did, evt.rkey) - } catch (err) { - this.log('Error handling settings change', { - did: evt.did, - event: evt.event, - rkey: evt.rkey, - error: - err instanceof Error - ? err.message - : String(err) - }) - } - }) - } - } else if ( - evt.event === 'delete' && - evt.collection === 'place.wisp.fs' - ) { - this.log('Received delete event', { - did: evt.did, - rkey: evt.rkey - }) - - await this.queueTask(async () => { - try { - await this.handleDelete(evt.did, evt.rkey) - } catch (err) { - this.log('Error handling delete', { - did: evt.did, - rkey: evt.rkey, - error: - err instanceof Error ? err.message : String(err) - }) - } - }) - } else if ( - evt.event === 'delete' && - evt.collection === 'place.wisp.settings' - ) { - this.log('Received settings delete event', { - did: evt.did, - rkey: evt.rkey - }) - - await this.queueTask(async () => { - try { - await this.handleSettingsChange(evt.did, evt.rkey) - } catch (err) { - this.log('Error handling settings delete', { - did: evt.did, - rkey: evt.rkey, - error: - err instanceof Error ? err.message : String(err) - }) - } - }) - } - }, - onError: (err: any) => { - this.log('Firehose error', { - error: err instanceof Error ? err.message : String(err), - stack: err instanceof Error ? err.stack : undefined, - fullError: err - }) - console.error('Full firehose error:', err) - } - }) - - this.firehose.start().catch((err: unknown) => { - this.log('Fatal firehose error', { - error: err instanceof Error ? err.message : String(err) - }) - console.error('Fatal firehose error:', err) - }) - this.log('Firehose starting') - } - - private async handleCreateOrUpdate( - did: string, - site: string, - record: any, - eventCid?: string - ) { - console.log(`[Firehose] Processing create/update from firehose - ${did}:${site}`) - this.log('Processing create/update', { did, site }) - - // Record is already validated in handleEvent - const fsRecord = record - - const pdsEndpoint = await getPdsForDid(did) - if (!pdsEndpoint) { - this.log('Could not resolve PDS for DID', { did }) - return - } - - this.log('Resolved PDS', { did, pdsEndpoint }) - - // Verify record exists on PDS and fetch its CID - this.log('Verifying record on PDS', { did, site }) - let verifiedCid: string - try { - const result = await fetchSiteRecord(did, site) - - if (!result) { - this.log('Record not found on PDS, skipping cache', { - did, - site - }) - return - } - - verifiedCid = result.cid - - // Verify event CID matches PDS CID (prevent cache poisoning) - if (eventCid && eventCid !== verifiedCid) { - this.log('CID mismatch detected - potential spoofed event', { - did, - site, - eventCid, - verifiedCid - }) - return - } - - this.log('Record verified on PDS', { did, site, cid: verifiedCid }) - } catch (err) { - this.log('Failed to verify record on PDS', { - did, - site, - error: err instanceof Error ? err.message : String(err) - }) - return - } - - // Invalidate in-memory caches before updating - await invalidateSiteCache(did, site) - - // Mark site as being cached to prevent serving stale content during update - markSiteAsBeingCached(did, site) - - try { - // Cache the record with verified CID (uses atomic swap internally) - // All instances cache locally for edge serving - await downloadAndCacheSite( - did, - site, - fsRecord, - pdsEndpoint, - verifiedCid - ) - - // Clear redirect rules cache since the site was updated - clearRedirectRulesCache(did, site) - - // Acquire distributed lock only for database write to prevent duplicate writes - // Note: upsertSite will check cache-only mode internally and skip if needed - const lockKey = `db:upsert:${did}:${site}` - const lockAcquired = await tryAcquireLock(lockKey) - - if (!lockAcquired) { - this.log('Another instance is writing to DB, skipping upsert', { - did, - site - }) - this.log('Successfully processed create/update (cached locally)', { - did, - site - }) - return - } - - try { - // Upsert site to database (only one instance does this) - // In cache-only mode, this will be a no-op - await upsertSite(did, site, fsRecord.site) - this.log( - 'Successfully processed create/update (cached + DB updated)', - { did, site } - ) - } finally { - // Always release lock, even if DB write fails - await releaseLock(lockKey) - } - } finally { - // Always unmark, even if caching fails - unmarkSiteAsBeingCached(did, site) - } - } - - private async handleDelete(did: string, site: string) { - this.log('Processing delete', { did, site }) - - // All instances should delete their local cache (no lock needed) - const pdsEndpoint = await getPdsForDid(did) - if (!pdsEndpoint) { - this.log('Could not resolve PDS for DID', { did }) - return - } - - // Verify record is actually deleted from PDS - try { - const recordUrl = `${pdsEndpoint}/xrpc/com.atproto.repo.getRecord?repo=${encodeURIComponent(did)}&collection=place.wisp.fs&rkey=${encodeURIComponent(site)}` - const recordRes = await safeFetch(recordUrl) - - if (recordRes.ok) { - this.log('Record still exists on PDS, not deleting cache', { - did, - site - }) - return - } - - this.log('Verified record is deleted from PDS', { - did, - site, - status: recordRes.status - }) - } catch (err) { - this.log('Error verifying deletion on PDS', { - did, - site, - error: err instanceof Error ? err.message : String(err) - }) - } - - // Invalidate all caches (tiered storage invalidation is handled by invalidateSiteCache) - await invalidateSiteCache(did, site) - - this.log('Successfully processed delete', { did, site }) - } - - private async handleSettingsChange(did: string, rkey: string) { - this.log('Processing settings change', { did, rkey }) - - // Invalidate in-memory caches (includes metadata which stores settings) - await invalidateSiteCache(did, rkey) - - // Check if site is already cached - const cacheDir = `${CACHE_DIR}/${did}/${rkey}` - const isCached = existsSync(cacheDir) - - if (!isCached) { - this.log('Site not cached yet, checking if fs record exists', { did, rkey }) - - // If site exists on PDS, cache it (which will include the new settings) - try { - const siteRecord = await fetchSiteRecord(did, rkey) - - if (siteRecord) { - this.log('Site record found, triggering full cache with settings', { did, rkey }) - const pdsEndpoint = await getPdsForDid(did) - - if (pdsEndpoint) { - // Mark as being cached - markSiteAsBeingCached(did, rkey) - - try { - await downloadAndCacheSite(did, rkey, siteRecord.record, pdsEndpoint, siteRecord.cid) - this.log('Successfully cached site with new settings', { did, rkey }) - } finally { - unmarkSiteAsBeingCached(did, rkey) - } - } else { - this.log('Could not resolve PDS for DID', { did }) - } - } else { - this.log('No fs record found for site, skipping cache', { did, rkey }) - } - } catch (err) { - this.log('Failed to cache site after settings change', { - did, - rkey, - error: err instanceof Error ? err.message : String(err) - }) - } - - this.log('Successfully processed settings change (new cache)', { did, rkey }) - return - } - - // Site is already cached, just update the settings in metadata - try { - const { fetchSiteSettings, updateCacheMetadataSettings } = await import('./utils') - const settings = await fetchSiteSettings(did, rkey) - await updateCacheMetadataSettings(did, rkey, settings) - this.log('Updated cached settings', { did, rkey, hasSettings: !!settings }) - } catch (err) { - this.log('Failed to update cached settings', { - did, - rkey, - error: err instanceof Error ? err.message : String(err) - }) - } - - this.log('Successfully processed settings change', { did, rkey }) - } - - getHealth() { - const isConnected = this.firehose !== null - const timeSinceLastEvent = Date.now() - this.lastEventTime - - return { - connected: isConnected, - lastEventTime: this.lastEventTime, - timeSinceLastEvent, - queueSize: this.processingQueue.size, - maxConcurrency: this.maxConcurrency, - healthy: isConnected && timeSinceLastEvent < 300000 // 5 minutes - } - } -} diff --git a/apps/hosting-service/src/lib/redirects.ts b/apps/hosting-service/src/lib/redirects.ts index 8e80a29..a11cd52 100644 --- a/apps/hosting-service/src/lib/redirects.ts +++ b/apps/hosting-service/src/lib/redirects.ts @@ -1,6 +1,5 @@ -import { readFile } from 'fs/promises'; -import { existsSync } from 'fs'; import { parseRedirectsFile, type RedirectRule } from '@wispplace/fs-utils'; +import { storage } from './storage'; // Re-export everything from the shared package export { @@ -17,15 +16,11 @@ export { * Load redirect rules from a cached site */ export async function loadRedirectRules(did: string, rkey: string): Promise { - const CACHE_DIR = process.env.CACHE_DIR || './cache/sites'; - const redirectsPath = `${CACHE_DIR}/${did}/${rkey}/_redirects`; - - if (!existsSync(redirectsPath)) { - return []; - } - + const key = `${did}/${rkey}/_redirects`; try { - const content = await readFile(redirectsPath, 'utf-8'); + const data = await storage.get(key); + if (!data) return []; + const content = new TextDecoder().decode(data as Uint8Array); return parseRedirectsFile(content); } catch (err) { console.error('Failed to load _redirects file', err); diff --git a/apps/hosting-service/src/lib/request-utils.ts b/apps/hosting-service/src/lib/request-utils.ts index 6b0bfb1..3de88c1 100644 --- a/apps/hosting-service/src/lib/request-utils.ts +++ b/apps/hosting-service/src/lib/request-utils.ts @@ -3,7 +3,6 @@ */ import type { Record as WispSettings } from '@wispplace/lexicons/types/place/wisp/settings'; -import { access } from 'fs/promises'; /** * Default index file names to check for directory requests @@ -74,15 +73,6 @@ export function isValidRkey(rkey: string): boolean { /** * Async file existence check */ -export async function fileExists(path: string): Promise { - try { - await access(path); - return true; - } catch { - return false; - } -} - /** * Extract and normalize headers from request */ @@ -93,4 +83,3 @@ export function extractHeaders(rawHeaders: Headers): Record { }); return headers; } - diff --git a/apps/hosting-service/src/lib/revalidate-metrics.ts b/apps/hosting-service/src/lib/revalidate-metrics.ts new file mode 100644 index 0000000..64621fc --- /dev/null +++ b/apps/hosting-service/src/lib/revalidate-metrics.ts @@ -0,0 +1,49 @@ +type EnqueueResult = 'enqueued' | 'deduped' | 'disabled' | 'error'; + +interface RevalidateMetrics { + storageMissExpected: number; + revalidateEnqueued: number; + revalidateDeduped: number; + revalidateDisabled: number; + revalidateErrors: number; + lastStorageMissAt: number | null; + lastStorageMissPath: string | null; +} + +const metrics: RevalidateMetrics = { + storageMissExpected: 0, + revalidateEnqueued: 0, + revalidateDeduped: 0, + revalidateDisabled: 0, + revalidateErrors: 0, + lastStorageMissAt: null, + lastStorageMissPath: null, +}; + +export function recordStorageMiss(path: string): void { + metrics.storageMissExpected += 1; + metrics.lastStorageMissAt = Date.now(); + metrics.lastStorageMissPath = path; +} + +export function recordRevalidateResult(result: EnqueueResult): void { + if (result === 'enqueued') { + metrics.revalidateEnqueued += 1; + return; + } + if (result === 'deduped') { + metrics.revalidateDeduped += 1; + return; + } + if (result === 'disabled') { + metrics.revalidateDisabled += 1; + return; + } + if (result === 'error') { + metrics.revalidateErrors += 1; + } +} + +export function getRevalidateMetrics(): RevalidateMetrics { + return { ...metrics }; +} diff --git a/apps/hosting-service/src/lib/revalidate-queue.ts b/apps/hosting-service/src/lib/revalidate-queue.ts new file mode 100644 index 0000000..4030243 --- /dev/null +++ b/apps/hosting-service/src/lib/revalidate-queue.ts @@ -0,0 +1,83 @@ +import Redis from 'ioredis'; +import { recordRevalidateResult } from './revalidate-metrics'; + +const redisUrl = process.env.REDIS_URL; +const streamName = process.env.WISP_REVALIDATE_STREAM || 'wisp:revalidate'; +const dedupeTtlSeconds = Number.parseInt(process.env.WISP_REVALIDATE_DEDUPE_TTL_SECONDS || '60', 10); + +let client: Redis | null = null; +let loggedMissingRedis = false; + +function getRedisClient(): Redis | null { + if (!redisUrl) { + if (!loggedMissingRedis) { + console.warn('[Revalidate] REDIS_URL not set; skipping queue enqueue'); + loggedMissingRedis = true; + } + return null; + } + + if (!client) { + client = new Redis(redisUrl, { + maxRetriesPerRequest: 2, + enableReadyCheck: true, + }); + + client.on('error', (err) => { + console.error('[Revalidate] Redis error:', err); + }); + } + + return client; +} + +export type EnqueueResult = 'enqueued' | 'deduped' | 'disabled' | 'error'; + +export async function enqueueRevalidate( + did: string, + rkey: string, + reason: string +): Promise<{ enqueued: boolean; result: EnqueueResult }> { + const redis = getRedisClient(); + if (!redis) { + recordRevalidateResult('disabled'); + return { enqueued: false, result: 'disabled' }; + } + + try { + const dedupeKey = `revalidate:site:${did}:${rkey}`; + const set = await redis.set(dedupeKey, '1', 'NX', 'EX', dedupeTtlSeconds); + if (!set) { + recordRevalidateResult('deduped'); + return { enqueued: false, result: 'deduped' }; + } + + await redis.xadd( + streamName, + '*', + 'did', + did, + 'rkey', + rkey, + 'reason', + reason, + 'ts', + Date.now().toString() + ); + + recordRevalidateResult('enqueued'); + return { enqueued: true, result: 'enqueued' }; + } catch (err) { + recordRevalidateResult('error'); + console.error('[Revalidate] Failed to enqueue', { did, rkey, reason, error: err }); + return { enqueued: false, result: 'error' }; + } +} + +export async function closeRevalidateQueue(): Promise { + if (client) { + const toClose = client; + client = null; + await toClose.quit(); + } +} diff --git a/apps/hosting-service/src/lib/site-cache.ts b/apps/hosting-service/src/lib/site-cache.ts index 5fe76cf..4bddad8 100644 --- a/apps/hosting-service/src/lib/site-cache.ts +++ b/apps/hosting-service/src/lib/site-cache.ts @@ -1,14 +1,10 @@ /** - * Site caching management utilities + * Redirect rules cache utilities */ -import { createLogger } from '@wispplace/observability'; -import { fetchSiteRecord, getPdsForDid, downloadAndCacheSite, isCached } from './utils'; -import { markSiteAsBeingCached, unmarkSiteAsBeingCached, LRUCache } from './cache'; +import { LRUCache } from './cache'; import type { RedirectRule } from './redirects'; -const logger = createLogger('hosting-service'); - // Cache for redirect rules (per site) - LRU with 1000 site limit // Each entry is relatively small (array of redirect rules), so 1000 sites should be < 10MB const redirectRulesCache = new LRUCache(10 * 1024 * 1024, 1000); // 10MB, 1000 sites @@ -39,47 +35,3 @@ export function setRedirectRulesInCache(did: string, rkey: string, rules: Redire const estimatedSize = rules.length * 100; redirectRulesCache.set(cacheKey, rules, estimatedSize); } - -/** - * Helper to ensure site is cached - * Returns true if site is successfully cached, false otherwise - */ -export async function ensureSiteCached(did: string, rkey: string): Promise { - if (await isCached(did, rkey)) { - console.log(`[Cache Hit] Site already cached - ${did}:${rkey}`); - return true; - } - - // Fetch and cache the site - console.log(`[On-Demand] Caching site on first request - ${did}:${rkey}`); - const siteData = await fetchSiteRecord(did, rkey); - if (!siteData) { - logger.error('Site record not found', null, { did, rkey }); - return false; - } - - const pdsEndpoint = await getPdsForDid(did); - if (!pdsEndpoint) { - logger.error('PDS not found for DID', null, { did }); - return false; - } - - // Mark site as being cached to prevent serving stale content during update - markSiteAsBeingCached(did, rkey); - - try { - await downloadAndCacheSite(did, rkey, siteData.record, pdsEndpoint, siteData.cid); - // Clear redirect rules cache since the site was updated - clearRedirectRulesCache(did, rkey); - logger.info('Site cached successfully (on-demand)', { did, rkey }); - console.log(`[On-Demand] Successfully cached ${did}:${rkey}`); - return true; - } catch (err) { - logger.error('Failed to cache site on-demand', err, { did, rkey }); - return false; - } finally { - // Always unmark, even if caching fails - unmarkSiteAsBeingCached(did, rkey); - } -} - diff --git a/apps/hosting-service/src/lib/storage.ts b/apps/hosting-service/src/lib/storage.ts index 3a012c1..b617cc3 100644 --- a/apps/hosting-service/src/lib/storage.ts +++ b/apps/hosting-service/src/lib/storage.ts @@ -7,7 +7,7 @@ * - Cold (S3/R2): Object storage as source of truth (optional) * * When S3 is not configured, falls back to disk-only mode (warm tier acts as source of truth). - * In cache-only mode (non-master nodes), S3 writes are skipped even if configured. + * Hosting service is read-only: S3 writes are always skipped. */ import { @@ -25,10 +25,6 @@ const HOT_CACHE_COUNT = parseInt(process.env.HOT_CACHE_COUNT || '500', 10); const WARM_CACHE_SIZE = parseInt(process.env.WARM_CACHE_SIZE || '10737418240', 10); // 10GB default const WARM_EVICTION_POLICY = (process.env.WARM_EVICTION_POLICY || 'lru') as 'lru' | 'fifo' | 'size'; -// Cache-only mode: skip S3 writes (non-master nodes) -// This is the same flag used to skip database writes -const CACHE_ONLY_MODE = process.env.CACHE_ONLY_MODE === 'true'; - // S3/Cold tier configuration (optional) const S3_BUCKET = process.env.S3_BUCKET || ''; const S3_METADATA_BUCKET = process.env.S3_METADATA_BUCKET; @@ -55,8 +51,8 @@ const identityDeserialize = async (data: Uint8Array): Promise => { }; /** - * Read-only wrapper for S3 tier in cache-only mode. - * Allows reads from S3 but skips all writes (for non-master nodes). + * Read-only wrapper for S3 tier. + * Allows reads from S3 but skips all writes (hosting-service is read-only). */ class ReadOnlyS3Tier implements StorageTier { private static hasLoggedWriteSkip = false; @@ -92,7 +88,7 @@ class ReadOnlyS3Tier implements StorageTier { return this.tier.getStats(); } - // Write operations - no-op in cache-only mode + // Write operations - no-op in read-only mode async set(key: string, _data: Uint8Array, _metadata: StorageMetadata) { this.logWriteSkip('set', key); } @@ -120,7 +116,7 @@ class ReadOnlyS3Tier implements StorageTier { private logWriteSkip(operation: string, key: string) { // Only log once to avoid spam if (!ReadOnlyS3Tier.hasLoggedWriteSkip) { - console.log(`[Storage] Cache-only mode: skipping S3 writes (operation: ${operation})`); + console.log(`[Storage] Read-only mode: skipping S3 writes (operation: ${operation})`); ReadOnlyS3Tier.hasLoggedWriteSkip = true; } } @@ -157,15 +153,11 @@ function initializeStorage(): TieredStorage { prefix: S3_PREFIX, }); - // In cache-only mode, wrap S3 tier to make it read-only - coldTier = CACHE_ONLY_MODE ? new ReadOnlyS3Tier(s3Tier) : s3Tier; + // Hosting service is read-only: always wrap S3 tier to make it read-only + coldTier = new ReadOnlyS3Tier(s3Tier); warmTier = diskTier; - if (CACHE_ONLY_MODE) { - console.log('[Storage] Cache-only mode: S3 as read-only cold tier (no writes), disk as warm tier'); - } else { - console.log('[Storage] Using S3 as cold tier, disk as warm tier'); - } + console.log('[Storage] Read-only mode: S3 as cold tier (no writes), disk as warm tier'); } else { // Disk-only mode: disk tier acts as source of truth (cold) coldTier = diskTier; @@ -190,9 +182,9 @@ function initializeStorage(): TieredStorage { // Placement rules: determine which tiers each file goes to placementRules: [ - // Metadata is critical: frequently accessed for cache validity checks + // Rewritten HTML: keep hot for fast serving { - pattern: '**/.metadata.json', + pattern: '**/.rewritten/**/*.html', tiers: ['hot', 'warm', 'cold'], }, diff --git a/apps/hosting-service/src/lib/utils.ts b/apps/hosting-service/src/lib/utils.ts index b62ce46..abc46ed 100644 --- a/apps/hosting-service/src/lib/utils.ts +++ b/apps/hosting-service/src/lib/utils.ts @@ -1,118 +1,14 @@ -import type { Record as WispFsRecord, Directory, Entry, File } from '@wispplace/lexicons/types/place/wisp/fs'; +import type { Directory, Entry } from '@wispplace/lexicons/types/place/wisp/fs'; import type { Record as SubfsRecord } from '@wispplace/lexicons/types/place/wisp/subfs'; import type { Record as WispSettings } from '@wispplace/lexicons/types/place/wisp/settings'; -import { existsSync, mkdirSync, readFileSync, rmSync } from 'fs'; -import { writeFile, readFile, rename } from 'fs/promises'; -import { Readable } from 'stream'; -import { safeFetchJson, safeFetchBlob } from '@wispplace/safe-fetch'; -import { CID } from 'multiformats'; +import { safeFetchJson } from '@wispplace/safe-fetch'; import { extractBlobCid, resolveDid, getPdsForDid, didWebToHttps } from '@wispplace/atproto-utils'; -import { sanitizePath, collectFileCidsFromEntries, countFilesInDirectory } from '@wispplace/fs-utils'; -import { shouldCompressMimeType } from '@wispplace/atproto-utils/compression'; -import { MAX_BLOB_SIZE, MAX_FILE_COUNT, MAX_SITE_SIZE } from '@wispplace/constants'; -import { storage } from './storage'; +import { sanitizePath } from '@wispplace/fs-utils'; +import { getSiteSettingsCache } from './db'; // Re-export shared utilities for local usage and tests export { extractBlobCid, sanitizePath, resolveDid, getPdsForDid, didWebToHttps }; -const CACHE_DIR = process.env.CACHE_DIR || './cache/sites'; -const CACHE_TTL = 14 * 24 * 60 * 60 * 1000; // 14 days cache TTL - -interface CacheMetadata { - recordCid: string; - cachedAt: number; - did: string; - rkey: string; - // Map of file path to blob CID for incremental updates - fileCids?: Record; - // Site settings (null = explicitly no settings, undefined = not yet checked) - settings?: WispSettings | null; -} - - -export async function fetchSiteRecord(did: string, rkey: string): Promise<{ record: WispFsRecord; cid: string } | null> { - try { - const pdsEndpoint = await getPdsForDid(did); - if (!pdsEndpoint) { - console.error('[hosting-service] Failed to get PDS endpoint for DID', { did, rkey }); - return null; - } - - const url = `${pdsEndpoint}/xrpc/com.atproto.repo.getRecord?repo=${encodeURIComponent(did)}&collection=place.wisp.fs&rkey=${encodeURIComponent(rkey)}`; - const data = await safeFetchJson(url); - - return { - record: data.value as WispFsRecord, - cid: data.cid || '' - }; - } catch (err) { - const errorCode = (err as any)?.code; - const errorMsg = err instanceof Error ? err.message : String(err); - - // Better error logging to distinguish between network errors and 404s - if (errorMsg.includes('HTTP 404') || errorMsg.includes('Not Found')) { - console.log('[hosting-service] Site record not found', { did, rkey }); - } else if (errorCode && ['ECONNRESET', 'ERR_SSL_TLSV1_ALERT_INTERNAL_ERROR', 'ETIMEDOUT'].includes(errorCode)) { - console.error('[hosting-service] Network/SSL error fetching site record (after retries)', { - did, - rkey, - error: errorMsg, - code: errorCode - }); - } else { - console.error('[hosting-service] Failed to fetch site record', { - did, - rkey, - error: errorMsg, - code: errorCode - }); - } - - return null; - } -} - -export async function fetchSiteSettings(did: string, rkey: string): Promise { - try { - const pdsEndpoint = await getPdsForDid(did); - if (!pdsEndpoint) return null; - - const url = `${pdsEndpoint}/xrpc/com.atproto.repo.getRecord?repo=${encodeURIComponent(did)}&collection=place.wisp.settings&rkey=${encodeURIComponent(rkey)}`; - const data = await safeFetchJson(url); - - return data.value as WispSettings; - } catch (err) { - // Settings are optional, so return null if not found - return null; - } -} - -/** - * Calculate total size of all blobs in a directory tree from manifest metadata - */ -function calculateTotalBlobSize(directory: Directory): number { - let totalSize = 0; - - function sumBlobSizes(entries: Entry[]) { - for (const entry of entries) { - const node = entry.node; - - if ('type' in node && node.type === 'directory' && 'entries' in node) { - // Recursively sum subdirectories - sumBlobSizes(node.entries); - } else if ('type' in node && node.type === 'file' && 'blob' in node) { - // Add blob size from manifest - const fileNode = node as File; - const blobSize = (fileNode.blob as any)?.size || 0; - totalSize += blobSize; - } - } - } - - sumBlobSizes(directory.entries); - return totalSize; -} - /** * Extract all subfs URIs from a directory tree with their mount paths */ @@ -298,336 +194,17 @@ export async function expandSubfsNodes( } -export async function downloadAndCacheSite(did: string, rkey: string, record: WispFsRecord, pdsEndpoint: string, recordCid: string): Promise { - console.log('Caching site', did, rkey); - - if (!record.root) { - console.error('Record missing root directory:', JSON.stringify(record, null, 2)); - throw new Error('Invalid record structure: missing root directory'); - } - - if (!record.root.entries || !Array.isArray(record.root.entries)) { - console.error('Record root missing entries array:', JSON.stringify(record.root, null, 2)); - throw new Error('Invalid record structure: root missing entries array'); - } - - // Expand subfs nodes before caching - const expandedRoot = await expandSubfsNodes(record.root, pdsEndpoint); - - // Verify all subfs nodes were expanded - const remainingSubfs = extractSubfsUris(expandedRoot); - if (remainingSubfs.length > 0) { - console.warn(`[Cache] Warning: ${remainingSubfs.length} subfs nodes remain unexpanded after expansion`, remainingSubfs); - } - - // Validate file count limit - const fileCount = countFilesInDirectory(expandedRoot); - if (fileCount > MAX_FILE_COUNT) { - throw new Error(`Site exceeds file count limit: ${fileCount} files (max ${MAX_FILE_COUNT})`); - } - console.log(`[Cache] File count validation passed: ${fileCount} files (limit: ${MAX_FILE_COUNT})`); - - // Validate total size from blob metadata - const totalBlobSize = calculateTotalBlobSize(expandedRoot); - if (totalBlobSize > MAX_SITE_SIZE) { - throw new Error(`Site exceeds size limit: ${(totalBlobSize / 1024 / 1024).toFixed(2)}MB (max ${(MAX_SITE_SIZE / 1024 / 1024).toFixed(0)}MB)`); - } - console.log(`[Cache] Size validation passed: ${(totalBlobSize / 1024 / 1024).toFixed(2)}MB (limit: ${(MAX_SITE_SIZE / 1024 / 1024).toFixed(0)}MB)`); - - // Get existing cache metadata to check for incremental updates - const existingMetadata = await getCacheMetadata(did, rkey); - const existingFileCids = existingMetadata?.fileCids || {}; - - // Collect file CIDs from the new record (using expanded root) - const newFileCids: Record = {}; - collectFileCidsFromEntries(expandedRoot.entries, '', newFileCids); - - // Fetch site settings (optional) - const settings = await fetchSiteSettings(did, rkey); - - // Determine if this is an incremental update or full cache - const isIncremental = Object.keys(existingFileCids).length > 0; - const updateType = isIncremental ? 'incremental update' : 'full cache'; - console.log(`[Cache] Starting ${updateType} for ${did}:${rkey}`); - - // Download files directly to tiered storage (with incremental logic) - await cacheFiles(did, rkey, expandedRoot.entries, pdsEndpoint, '', existingFileCids); - await saveCacheMetadata(did, rkey, recordCid, newFileCids, settings); - - console.log(`[Cache] Successfully cached site ${did}:${rkey} (${updateType})`); -} - - -async function cacheFiles( - did: string, - site: string, - entries: Entry[], - pdsEndpoint: string, - pathPrefix: string, - existingFileCids: Record = {} -): Promise { - // Collect file download tasks (skip unchanged files) - const downloadTasks: Array<() => Promise> = []; - let skippedCount = 0; - - function collectFileTasks( - entries: Entry[], - currentPathPrefix: string - ) { - for (const entry of entries) { - const currentPath = currentPathPrefix ? `${currentPathPrefix}/${entry.name}` : entry.name; - const node = entry.node; - - if ('type' in node && node.type === 'directory' && 'entries' in node) { - collectFileTasks(node.entries, currentPath); - } else if ('type' in node && node.type === 'file' && 'blob' in node) { - const fileNode = node as File; - const cid = extractBlobCid(fileNode.blob); - - // Check if file is unchanged (same CID as existing cache) - if (cid && existingFileCids[currentPath] === cid) { - // File unchanged - skip download (already in tiered storage) - skippedCount++; - } else { - // File new or changed - download it - downloadTasks.push(() => cacheFileBlob( - did, - site, - currentPath, - fileNode.blob, - pdsEndpoint, - fileNode.encoding, - fileNode.mimeType, - fileNode.base64 - )); - } - } - } - } - - collectFileTasks(entries, pathPrefix); - - console.log(`[Incremental Update] Files to copy: ${skippedCount}, Files to download: ${downloadTasks.length}`); - - // Download new/changed files concurrently - const downloadLimit = 20; - let successCount = 0; - let failureCount = 0; - - for (let i = 0; i < downloadTasks.length; i += downloadLimit) { - const batch = downloadTasks.slice(i, i + downloadLimit); - const results = await Promise.allSettled(batch.map(task => task())); - - // Count successes and failures - results.forEach((result, index) => { - if (result.status === 'fulfilled') { - successCount++; - } else { - failureCount++; - console.error(`[Cache] Failed to download file (continuing with others):`, result.reason); - } - }); - - if (downloadTasks.length > downloadLimit) { - console.log(`[Cache Progress] Downloaded ${Math.min(i + downloadLimit, downloadTasks.length)}/${downloadTasks.length} files (${failureCount} failed)`); - } - } - - if (failureCount > 0) { - console.warn(`[Cache] Completed with ${successCount} successful and ${failureCount} failed file downloads`); - } -} - -async function cacheFileBlob( - did: string, - site: string, - filePath: string, - blobRef: any, - pdsEndpoint: string, - encoding?: 'gzip', - mimeType?: string, - base64?: boolean -): Promise { - const cid = extractBlobCid(blobRef); - if (!cid) { - console.error('Could not extract CID from blob', blobRef); - return; - } - - const blobUrl = `${pdsEndpoint}/xrpc/com.atproto.sync.getBlob?did=${encodeURIComponent(did)}&cid=${encodeURIComponent(cid)}`; - - console.log(`[Cache] Fetching blob for file: ${filePath}, CID: ${cid}`); - - let content = await safeFetchBlob(blobUrl, { maxSize: MAX_BLOB_SIZE, timeout: 300000 }); - - // If content is base64-encoded, decode it back to raw binary (gzipped or not) - if (base64) { - // Decode base64 directly from raw bytes - no string conversion - // The blob contains base64-encoded text as raw bytes, decode it in-place - const textDecoder = new TextDecoder(); - const base64String = textDecoder.decode(content); - content = Buffer.from(base64String, 'base64'); - } - - // Use the shared function to determine if this should remain compressed - const shouldStayCompressed = shouldCompressMimeType(mimeType); - - // Decompress files that shouldn't be stored compressed - if (encoding === 'gzip' && !shouldStayCompressed && content.length >= 2 && - content[0] === 0x1f && content[1] === 0x8b) { - try { - const { gunzipSync } = await import('zlib'); - const decompressed = gunzipSync(content); - content = decompressed; - // Clear the encoding flag since we're storing decompressed - encoding = undefined; - } catch (error) { - console.error(`Failed to decompress ${filePath}, storing original gzipped content:`, error); - } - } - - // Write to tiered storage with metadata - const stream = Readable.from([content]); - const key = `${did}/${site}/${filePath}`; - - // Build metadata object, only including defined values - const customMetadata: Record = {}; - if (encoding) customMetadata.encoding = encoding; - if (mimeType) customMetadata.mimeType = mimeType; - - await storage.setStream(key, stream, { - size: content.length, - skipTiers: ['hot'], // Don't put in memory on ingest, only on access - metadata: customMetadata, - }); - - // Log completion with tier info - const tierInfo = 'to warm/cold tiers'; - if (encoding === 'gzip' && mimeType) { - console.log(`[Cache] Stored ${filePath} ${tierInfo} (${content.length} bytes, gzipped, ${mimeType})`); - } else { - console.log(`[Cache] Stored ${filePath} ${tierInfo} (${content.length} bytes)`); - } -} - - -export function getCachedFilePath(did: string, site: string, filePath: string): string { - const sanitizedPath = sanitizePath(filePath); - return `${CACHE_DIR}/${did}/${site}/${sanitizedPath}`; -} - -/** - * Check if a site exists in any tier of the cache (without checking metadata) - * This is a quick existence check - for actual retrieval, use storage.get() - */ -export async function isCached(did: string, site: string): Promise { - // Check if any file exists for this site by checking for the index.html - // If index.html exists, the site is cached - const indexKey = `${did}/${site}/index.html`; - return await storage.exists(indexKey); -} - -async function saveCacheMetadata(did: string, rkey: string, recordCid: string, fileCids?: Record, settings?: WispSettings | null): Promise { - const metadata: CacheMetadata = { - recordCid, - cachedAt: Date.now(), - did, - rkey, - fileCids, - settings: settings || undefined - }; - - // Store through tiered storage for persistence to S3/cold tier - const metadataKey = `${did}/${rkey}/.metadata.json`; - const metadataBytes = new TextEncoder().encode(JSON.stringify(metadata, null, 2)); - await storage.set(metadataKey, metadataBytes); -} - -async function getCacheMetadata(did: string, rkey: string): Promise { - try { - // Retrieve metadata from tiered storage - const metadataKey = `${did}/${rkey}/.metadata.json`; - const data = await storage.get(metadataKey); - - if (!data) return null; - - // Deserialize from Uint8Array to JSON (storage uses identity serialization) - const jsonString = new TextDecoder().decode(data as Uint8Array); - return JSON.parse(jsonString) as CacheMetadata; - } catch (err) { - console.error('Failed to read cache metadata', err); - return null; - } -} - export async function getCachedSettings(did: string, rkey: string): Promise { - const metadata = await getCacheMetadata(did, rkey); - - // If metadata has settings (including explicit null for "no settings"), return them - if (metadata && 'settings' in metadata) { - return metadata.settings ?? null; - } - - // If metadata exists but has never checked for settings, try to fetch from PDS and update cache - if (metadata) { - console.log('[Cache] Metadata missing settings, fetching from PDS', { did, rkey }); - try { - const settings = await fetchSiteSettings(did, rkey); - // Update cache with settings (or null if none found) - // This caches the "no settings" state to avoid repeated PDS fetches - await updateCacheMetadataSettings(did, rkey, settings); - console.log('[Cache] Updated metadata with fetched settings', { did, rkey, hasSettings: !!settings }); - return settings; - } catch (err) { - console.error('[Cache] Failed to fetch/update settings', { did, rkey, err }); - } - } - - return null; -} - -export async function updateCacheMetadataSettings(did: string, rkey: string, settings: WispSettings | null): Promise { - try { - // Read existing metadata from tiered storage - const metadata = await getCacheMetadata(did, rkey); - - if (!metadata) { - console.warn('Metadata does not exist, cannot update settings', { did, rkey }); - return; - } - - // Update settings field - // Store null explicitly to cache "no settings" state and avoid repeated fetches - metadata.settings = settings ?? null; - - // Write back through tiered storage - // Convert to Uint8Array since storage is typed for binary data - const metadataKey = `${did}/${rkey}/.metadata.json`; - const metadataBytes = new TextEncoder().encode(JSON.stringify(metadata, null, 2)); - await storage.set(metadataKey, metadataBytes); - console.log('Updated metadata settings', { did, rkey, hasSettings: !!settings }); - } catch (err) { - console.error('Failed to update metadata settings', err); - throw err; - } -} - -export async function isCacheValid(did: string, rkey: string, currentRecordCid?: string): Promise { - const metadata = await getCacheMetadata(did, rkey); - if (!metadata) return false; - - // Check if cache has expired (14 days TTL) - const cacheAge = Date.now() - metadata.cachedAt; - if (cacheAge > CACHE_TTL) { - console.log('[Cache] Cache expired for', did, rkey); - return false; - } - - // If current CID is provided, verify it matches - if (currentRecordCid && metadata.recordCid !== currentRecordCid) { - console.log('[Cache] CID mismatch for', did, rkey, 'cached:', metadata.recordCid, 'current:', currentRecordCid); - return false; - } - - return true; + const cached = await getSiteSettingsCache(did, rkey); + if (!cached) return null; + + return { + $type: 'place.wisp.settings', + directoryListing: cached.directory_listing, + spaMode: cached.spa_mode ?? undefined, + custom404: cached.custom_404 ?? undefined, + indexFiles: cached.index_files ?? undefined, + cleanUrls: cached.clean_urls, + headers: cached.headers ?? undefined, + }; } diff --git a/apps/hosting-service/src/server.ts b/apps/hosting-service/src/server.ts index 88843dd..cc58308 100644 --- a/apps/hosting-service/src/server.ts +++ b/apps/hosting-service/src/server.ts @@ -10,11 +10,9 @@ import { resolveDid } from './lib/utils'; import { logCollector, errorTracker, metricsCollector } from '@wispplace/observability'; import { observabilityMiddleware, observabilityErrorHandler } from '@wispplace/observability/middleware/hono'; import { sanitizePath } from '@wispplace/fs-utils'; -import { isSiteBeingCached } from './lib/cache'; import { isValidRkey, extractHeaders } from './lib/request-utils'; -import { siteUpdatingResponse } from './lib/page-generators'; -import { ensureSiteCached } from './lib/site-cache'; import { serveFromCache, serveFromCacheWithRewrite } from './lib/file-serving'; +import { getRevalidateMetrics } from './lib/revalidate-metrics'; const BASE_HOST = process.env.BASE_HOST || 'wisp.place'; @@ -40,11 +38,11 @@ app.onError(observabilityErrorHandler('hosting-service')); app.get('/*', async (c) => { const url = new URL(c.req.url); const hostname = c.req.header('host') || ''; + const hostnameWithoutPort = hostname.split(':')[0]; const rawPath = url.pathname.replace(/^\//, ''); const path = sanitizePath(rawPath); // Check if this is sites.wisp.place subdomain (strip port for comparison) - const hostnameWithoutPort = hostname.split(':')[0]; if (hostnameWithoutPort === `sites.${BASE_HOST}`) { // Sanitize the path FIRST to prevent path traversal const sanitizedFullPath = sanitizePath(rawPath); @@ -82,17 +80,6 @@ app.get('/*', async (c) => { console.log(`[Server] sites.wisp.place request: identifier=${identifier}, site=${site}, filePath=${filePath}`); - // Check if site is currently being cached - return updating response early - if (isSiteBeingCached(did, site)) { - return siteUpdatingResponse(); - } - - // Ensure site is cached - const cached = await ensureSiteCached(did, site); - if (!cached) { - return c.text('Site not found', 404); - } - // Serve with HTML path rewriting to handle absolute paths const basePath = `/${identifier}/${site}/`; console.log(`[Server] Serving with basePath: ${basePath}`); @@ -128,23 +115,13 @@ app.get('/*', async (c) => { return c.text('Invalid site configuration', 500); } - // Check if site is currently being cached - return updating response early - if (isSiteBeingCached(customDomain.did, rkey)) { - return siteUpdatingResponse(); - } - - const cached = await ensureSiteCached(customDomain.did, rkey); - if (!cached) { - return c.text('Site not found', 404); - } - const headers = extractHeaders(c.req.raw.headers); return serveFromCache(customDomain.did, rkey, path, c.req.url, headers); } // Route 2: Registered subdomains - /*.wisp.place/* - if (hostname.endsWith(`.${BASE_HOST}`)) { - const domainInfo = await getWispDomain(hostname); + if (hostnameWithoutPort.endsWith(`.${BASE_HOST}`)) { + const domainInfo = await getWispDomain(hostnameWithoutPort); if (!domainInfo) { return c.text('Subdomain not registered', 404); } @@ -158,22 +135,12 @@ app.get('/*', async (c) => { return c.text('Invalid site configuration', 500); } - // Check if site is currently being cached - return updating response early - if (isSiteBeingCached(domainInfo.did, rkey)) { - return siteUpdatingResponse(); - } - - const cached = await ensureSiteCached(domainInfo.did, rkey); - if (!cached) { - return c.text('Site not found', 404); - } - const headers = extractHeaders(c.req.raw.headers); return serveFromCache(domainInfo.did, rkey, path, c.req.url, headers); } // Route 1: Custom domains - /* - const customDomain = await getCustomDomain(hostname); + const customDomain = await getCustomDomain(hostnameWithoutPort); if (!customDomain) { return c.text('Custom domain not found or not verified', 404); } @@ -187,16 +154,6 @@ app.get('/*', async (c) => { return c.text('Invalid site configuration', 500); } - // Check if site is currently being cached - return updating response early - if (isSiteBeingCached(customDomain.did, rkey)) { - return siteUpdatingResponse(); - } - - const cached = await ensureSiteCached(customDomain.did, rkey); - if (!cached) { - return c.text('Site not found', 404); - } - const headers = extractHeaders(c.req.raw.headers); return serveFromCache(customDomain.did, rkey, path, c.req.url, headers); }); @@ -225,7 +182,7 @@ app.get('/__internal__/observability/metrics', (c) => { const query = c.req.query(); const timeWindow = query.timeWindow ? parseInt(query.timeWindow as string) : 3600000; const stats = metricsCollector.getStats('hosting-service', timeWindow); - return c.json({ stats, timeWindow }); + return c.json({ stats, revalidate: getRevalidateMetrics(), timeWindow }); }); app.get('/__internal__/observability/cache', async (c) => { diff --git a/apps/main-app/src/lib/db.ts b/apps/main-app/src/lib/db.ts index 5ba1f2f..d3437e3 100644 --- a/apps/main-app/src/lib/db.ts +++ b/apps/main-app/src/lib/db.ts @@ -138,7 +138,8 @@ await db` // Site settings cache table - cached place.wisp.settings records await db` CREATE TABLE IF NOT EXISTS site_settings_cache ( - did TEXT PRIMARY KEY, + did TEXT NOT NULL, + rkey TEXT NOT NULL, record_cid TEXT NOT NULL, directory_listing BOOLEAN NOT NULL DEFAULT false, spa_mode TEXT, @@ -147,7 +148,8 @@ await db` clean_urls BOOLEAN NOT NULL DEFAULT true, headers JSONB, cached_at BIGINT DEFAULT EXTRACT(EPOCH FROM NOW()), - updated_at BIGINT DEFAULT EXTRACT(EPOCH FROM NOW()) + updated_at BIGINT DEFAULT EXTRACT(EPOCH FROM NOW()), + PRIMARY KEY (did, rkey) ) `; diff --git a/bun.lock b/bun.lock index 44a0981..1dc999a 100644 --- a/bun.lock +++ b/bun.lock @@ -31,6 +31,7 @@ "@wispplace/safe-fetch": "workspace:*", "@wispplace/tiered-storage": "workspace:*", "hono": "^4.10.4", + "ioredis": "^5.9.2", "multiformats": "^13.4.1", "postgres": "^3.4.5", }, @@ -59,6 +60,7 @@ "@wispplace/safe-fetch": "workspace:*", "@wispplace/tiered-storage": "workspace:*", "hono": "^4.10.4", + "ioredis": "^5.9.2", "mime-types": "^2.1.35", "multiformats": "^13.4.1", "postgres": "^3.4.5", @@ -578,6 +580,8 @@ "@humanwhocodes/retry": ["@humanwhocodes/retry@0.4.3", "", {}, "sha512-bV0Tgo9K4hfPCek+aMAn81RppFKv2ySDQeMoSZuvTASywNTnVJCArCZE2FWqpvIatKu7VMRLWlR1EazvVhDyhQ=="], + "@ioredis/commands": ["@ioredis/commands@1.5.0", "", {}, "sha512-eUgLqrMf8nJkZxT24JvVRrQya1vZkQh8BBeYNwGDqa5I0VUi8ACx7uFvAaLxintokpTenkK6DASvo/bvNbBGow=="], + "@ipld/dag-cbor": ["@ipld/dag-cbor@7.0.3", "", { "dependencies": { "cborg": "^1.6.0", "multiformats": "^9.5.4" } }, "sha512-1VVh2huHsuohdXC1bGJNE8WR72slZ9XE2T3wbBBq31dm7ZBatmKLLxrB+XAqafxfRFjv08RZmj/W/ZqaM13AuA=="], "@jridgewell/gen-mapping": ["@jridgewell/gen-mapping@0.3.13", "", { "dependencies": { "@jridgewell/sourcemap-codec": "^1.5.0", "@jridgewell/trace-mapping": "^0.3.24" } }, "sha512-2kkt/7niJ6MgEPxF0bYdQ6etZaA+fQvDcLKckhy1yIQOzaoKjBBjSj63/aLVjYE3qhRt5dvM+uUyfCg6UKCBbA=="], @@ -1126,6 +1130,8 @@ "clsx": ["clsx@2.1.1", "", {}, "sha512-eYm0QWBtUrBWZWG0d386OGAw16Z995PiOVo2B7bjWSbHedGl5e0ZWaq65kOGgUSNesEIDkB9ISbTg/JK9dhCZA=="], + "cluster-key-slot": ["cluster-key-slot@1.1.2", "", {}, "sha512-RMr0FhtfXemyinomL4hrWcYJxmX6deFdCxpJzhDttxgO1+bcCnkk+9drydLVDmAMG7NE6aN/fl4F7ucU/90gAA=="], + "code-block-writer": ["code-block-writer@13.0.3", "", {}, "sha512-Oofo0pq3IKnsFtuHqSF7TqBfr71aeyZDVJ0HpmqB7FBM2qEigL0iPONSCZSO9pE9dZTAxANe5XHG9Uy0YMv8cg=="], "color-convert": ["color-convert@2.0.1", "", { "dependencies": { "color-name": "~1.1.4" } }, "sha512-RRECPsj7iu/xb5oKYcsFHSppFNnsj/52OVTRKb4zP5onXwVF3zVmmToNcOfGC+CRDpfK/U584fMg38ZHCaElKQ=="], @@ -1160,6 +1166,8 @@ "define-lazy-prop": ["define-lazy-prop@3.0.0", "", {}, "sha512-N+MeXYoqr3pOgn8xfyRPREN7gHakLYjhsHhWGT3fWAiL4IkAt0iDw14QiiEm2bE30c5XX5q0FtAA3CK5f9/BUg=="], + "denque": ["denque@2.1.0", "", {}, "sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw=="], + "depd": ["depd@2.0.0", "", {}, "sha512-g7nH6P6dyDioJogAAGprGpCtVImJhpPk/roCzdb3fIh61/s/nPsfR6onyMwkCAR/OlC3yBC0lESvUoQEAssIrw=="], "destroy": ["destroy@1.2.0", "", {}, "sha512-2sJGJTaXIIaR1w4iJSNoN0hnMY7Gpc/n8D4qSCJw8QqFWXf7cuAgnEHxBpweaVcPevC2l3KpjYCx3NypQQgaJg=="], @@ -1314,6 +1322,8 @@ "inherits": ["inherits@2.0.4", "", {}, "sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ=="], + "ioredis": ["ioredis@5.9.2", "", { "dependencies": { "@ioredis/commands": "1.5.0", "cluster-key-slot": "^1.1.0", "debug": "^4.3.4", "denque": "^2.1.0", "lodash.defaults": "^4.2.0", "lodash.isarguments": "^3.1.0", "redis-errors": "^1.2.0", "redis-parser": "^3.0.0", "standard-as-callback": "^2.1.0" } }, "sha512-tAAg/72/VxOUW7RQSX1pIxJVucYKcjFjfvj60L57jrZpYCHC3XN0WCQ3sNYL4Gmvv+7GPvTAjc+KSdeNuE8oWQ=="], + "ipaddr.js": ["ipaddr.js@1.9.1", "", {}, "sha512-0KI/607xoxSToH7GjN1FfSbLoU0+btTicjsQSWQlh/hZykN8KpmMf7uYwPW3R+akZ6R/w18ZlXSHBYXiYUPO3g=="], "iron-session": ["iron-session@8.0.4", "", { "dependencies": { "cookie": "^0.7.2", "iron-webcrypto": "^1.2.1", "uncrypto": "^0.1.3" } }, "sha512-9ivNnaKOd08osD0lJ3i6If23GFS2LsxyMU8Gf/uBUEgm8/8CC1hrrCHFDpMo3IFbpBgwoo/eairRsaD3c5itxA=="], @@ -1384,6 +1394,10 @@ "lodash.camelcase": ["lodash.camelcase@4.3.0", "", {}, "sha512-TwuEnCnxbc3rAvhf/LbG7tJUDzhqXyFnv3dtzLOPgCG/hODL7WFnsbwktkD7yUV0RrreP/l1PALq/YSg6VvjlA=="], + "lodash.defaults": ["lodash.defaults@4.2.0", "", {}, "sha512-qjxPLHd3r5DnsdGacqOMU6pb/avJzdh9tFX2ymgoZE27BmjXrNy/y4LoaiTeAb+O3gL8AfpJGtqfX/ae2leYYQ=="], + + "lodash.isarguments": ["lodash.isarguments@3.1.0", "", {}, "sha512-chi4NHZlZqZD18a0imDHnZPrDeBbTtVN7GXMwuGdRH9qotxAjYs3aVLKc7zNOG9eddR5Ksd8rvFEBc9SsggPpg=="], + "lodash.merge": ["lodash.merge@4.6.2", "", {}, "sha512-0KpjqXRVvrYyCsX1swR/XTK0va6VQkQM6MNo7PqW77ByjAhoARA8EfrP1N4+KlKj8YS0ZUCtRT/YUuhyYDujIQ=="], "long": ["long@5.3.2", "", {}, "sha512-mNAgZ1GmyNhD7AuqnTG3/VQ26o760+ZYBPKjPvugO8+nLbYfX6TVpJPseBvopbdY+qpZ/lKUnmEc1LeZYS3QAA=="], @@ -1536,6 +1550,10 @@ "real-require": ["real-require@0.2.0", "", {}, "sha512-57frrGM/OCTLqLOAh0mhVA9VBMHd+9U7Zb2THMGdBUoZVOtGbJzjxsYGDJ3A9AYYCP4hn6y1TVbaOfzWtm5GFg=="], + "redis-errors": ["redis-errors@1.2.0", "", {}, "sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w=="], + + "redis-parser": ["redis-parser@3.0.0", "", { "dependencies": { "redis-errors": "^1.0.0" } }, "sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A=="], + "require-directory": ["require-directory@2.1.1", "", {}, "sha512-fGxEI7+wsG9xrvdjsrlmL22OMTTiHRwAMroiEeMgq8gzoLC/PQr7RsRDSTLUg/bZAZtF+TVIkHc6/4RIKrui+Q=="], "require-in-the-middle": ["require-in-the-middle@7.5.2", "", { "dependencies": { "debug": "^4.3.5", "module-details-from-path": "^1.0.3", "resolve": "^1.22.8" } }, "sha512-gAZ+kLqBdHarXB64XpAe2VCjB7rIRv+mU8tfRWziHRJ5umKsIHN2tLLv6EtMw7WCdP19S0ERVMldNvxYCHnhSQ=="], @@ -1594,6 +1612,8 @@ "stackback": ["stackback@0.0.2", "", {}, "sha512-1XMJE5fQo1jGH6Y/7ebnwPOBEkIEnT4QF32d5R1+VXdXveM0IBMJt8zfaxX1P3QhVwrYe+576+jkANtSS2mBbw=="], + "standard-as-callback": ["standard-as-callback@2.1.0", "", {}, "sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A=="], + "statuses": ["statuses@2.0.2", "", {}, "sha512-DvEy55V3DB7uknRo+4iOGT5fP1slR8wQohVdknigZPMpMstaKJQWhwiYBACJE3Ul2pTnATihhBYnRhZQHGBiRw=="], "std-env": ["std-env@3.10.0", "", {}, "sha512-5GS12FdOZNliM5mAOxFRg7Ir0pWz8MdpYm6AY6VPkGpbA7ZzmbzNcBJQ0GPvvyWgcY7QAhCgf9Uy89I03faLkg=="], diff --git a/packages/@wispplace/database/src/types.ts b/packages/@wispplace/database/src/types.ts index f6a76ae..123bb85 100644 --- a/packages/@wispplace/database/src/types.ts +++ b/packages/@wispplace/database/src/types.ts @@ -73,6 +73,7 @@ export interface SiteCache { */ export interface SiteSettingsCache { did: string; + rkey: string; record_cid: string; directory_listing: boolean; spa_mode: string | null; diff --git a/packages/@wispplace/fs-utils/src/file-cids.ts b/packages/@wispplace/fs-utils/src/file-cids.ts new file mode 100644 index 0000000..05ae2d2 --- /dev/null +++ b/packages/@wispplace/fs-utils/src/file-cids.ts @@ -0,0 +1,62 @@ +export type FileCidsNormalizationSource = + | 'object' + | 'array' + | 'string-json' + | 'string-invalid' + | 'null' + | 'other'; + +export type FileCidsNormalization = { + value: Record; + source: FileCidsNormalizationSource; +}; + +export function normalizeFileCids(value: unknown): FileCidsNormalization { + if (value == null) { + return { value: {}, source: 'null' }; + } + + if (typeof value === 'string') { + try { + const parsed = JSON.parse(value) as unknown; + if (Array.isArray(parsed)) { + const normalized = normalizeFileCids(parsed); + return { value: normalized.value, source: 'string-json' }; + } + if (parsed && typeof parsed === 'object') { + return { value: parsed as Record, source: 'string-json' }; + } + } catch { + // fall through to invalid + } + return { value: {}, source: 'string-invalid' }; + } + + if (Array.isArray(value)) { + const result: Record = {}; + for (const item of value) { + if (Array.isArray(item) && item.length >= 2) { + const [path, cid] = item; + if (typeof path === 'string' && typeof cid === 'string') { + result[path] = cid; + } + continue; + } + + if (item && typeof item === 'object' && 'path' in item && 'cid' in item) { + const path = (item as any).path; + const cid = (item as any).cid; + if (typeof path === 'string' && typeof cid === 'string') { + result[path] = cid; + } + } + } + return { value: result, source: 'array' }; + } + + if (typeof value === 'object') { + return { value: value as Record, source: 'object' }; + } + + return { value: {}, source: 'other' }; +} diff --git a/packages/@wispplace/fs-utils/src/index.ts b/packages/@wispplace/fs-utils/src/index.ts index 4c14a64..53b6189 100644 --- a/packages/@wispplace/fs-utils/src/index.ts +++ b/packages/@wispplace/fs-utils/src/index.ts @@ -5,6 +5,10 @@ export { sanitizePath, normalizePath } from './path'; export type { UploadedFile, FileUploadResult, ProcessedDirectory, ProcessUploadedFilesOptions, UpdateFileBlobsOptions } from './tree'; export { processUploadedFiles, updateFileBlobs, countFilesInDirectory, collectFileCidsFromEntries } from './tree'; +// File CID normalization +export type { FileCidsNormalization, FileCidsNormalizationSource } from './file-cids'; +export { normalizeFileCids } from './file-cids'; + // Manifest creation export { createManifest } from './manifest';