import fs from 'node:fs'; import path from 'node:path'; import { DEFAULT_CCTV_SOURCE_FILE, DEFAULT_CCTV_MAX_SOURCES, CCTV_SOURCE_CACHE_MS, } from './constants.js'; import { normalizeSourceItem } from './normalize.js'; import { loadAustinSourcesFromOpenData, loadCaltransSourcesFromOpenData, loadTflSourcesFromOpenData, } from './sources.js'; /** * Load CCTV sources from a local JSON file (CCTV_SOURCES_FILE env or default). * * @returns {Array} Array of raw source objects, or [] on error. */ function loadSourcesFromFile(sourceRoot) { const sourceFile = process.env.CCTV_SOURCES_FILE || DEFAULT_CCTV_SOURCE_FILE; const resolved = path.isAbsolute(sourceFile) ? sourceFile : path.resolve(sourceRoot, sourceFile); try { if (!fs.existsSync(resolved)) return []; const raw = fs.readFileSync(resolved, 'utf8'); const parsed = JSON.parse(raw); return Array.isArray(parsed) ? parsed : []; } catch (error) { console.warn( '[CCTV] failed to read source file:', resolved, error?.message || error, ); return []; } } /** * Load CCTV sources from the CCTV_SOURCES_JSON env variable (inline JSON). * * @returns {Array} Array of raw source objects, or [] if unset/invalid. */ function loadSourcesFromEnv() { const raw = process.env.CCTV_SOURCES_JSON; if (!raw) return []; try { const parsed = JSON.parse(raw); return Array.isArray(parsed) ? parsed : []; } catch { return []; } } /** Create an independent catalog rooted in the consuming application. */ export function createCctvCatalog({ sourceRoot = process.cwd() } = {}) { /** @type {Array} Cached merged + normalized CCTV source list. */ let _cctvSourceCache = []; /** @type {number} Epoch-ms when the source cache was last refreshed. */ let _cctvSourceCacheAt = 0; /** @type {Promise>|null} In-flight refresh, shared by concurrent * callers so a post-TTL burst launches ONE refetch, not one per request. */ let _cctvSourceInflight = null; /** * Assemble and cache the merged CCTV source list. * * Merges sources from three origins (Austin Open Data, local file, * env variable), deduplicates by ID, applies the global max cap, and * caches for CCTV_SOURCE_CACHE_MS. * * @returns {Promise>} Deduplicated, capped source list. */ async function getCctvSources() { const now = Date.now(); if ( _cctvSourceCache.length && now - _cctvSourceCacheAt <= CCTV_SOURCE_CACHE_MS ) { return _cctvSourceCache; } // Single-flight: a burst of requests arriving past the TTL shares ONE refresh // instead of each launching the full multi-provider refetch. The `.finally` // clears the ref so the next post-TTL cycle starts fresh. if (_cctvSourceInflight) return _cctvSourceInflight; _cctvSourceInflight = refreshCctvSources().finally(() => { _cctvSourceInflight = null; }); return _cctvSourceInflight; } /** * Assemble and cache the merged CCTV source list from file/env + live packs. * Always resolves (loaders self-catch to []); on a fully-empty refresh with a * good prior catalog it serves stale rather than blanking the CCTV layer. * * @returns {Promise>} Deduplicated, capped source list. */ async function refreshCctvSources() { const fromFile = loadSourcesFromFile(sourceRoot); const fromEnv = loadSourcesFromEnv(); const forceAustin = String(process.env.CCTV_FORCE_AUSTIN || '').trim() === '1'; const preferAustin = String(process.env.CCTV_PREFER_AUSTIN || '1').trim() !== '0'; // Live open-data packs (Austin + Caltrans + TfL) load unless a file/env pack // is configured and live packs aren't forced — same gate that governed the // Austin-only fetch, now governing all three. Each pack fails independently. const needsLiveSources = forceAustin || (fromFile.length + fromEnv.length === 0 && preferAustin); const tflEnabled = String(process.env.CCTV_TFL_ENABLED || '1').trim() !== '0'; let fromAustin = []; let fromCaltrans = []; let fromTfl = []; if (needsLiveSources) { const [austinResult, caltransResult, tflResult] = await Promise.allSettled([ loadAustinSourcesFromOpenData(), loadCaltransSourcesFromOpenData(), tflEnabled ? loadTflSourcesFromOpenData() : Promise.resolve([]), ]); fromAustin = austinResult.status === 'fulfilled' ? austinResult.value : []; fromCaltrans = caltransResult.status === 'fulfilled' ? caltransResult.value : []; fromTfl = tflResult.status === 'fulfilled' ? tflResult.value : []; } // Live sources first so file/env overrides win on duplicate IDs (Map last-write). const merged = [ ...fromAustin, ...fromCaltrans, ...fromTfl, ...fromFile, ...fromEnv, ]; // Deduplicate by camera ID (last-write wins because of Map.set) const byId = new Map(); for (const item of merged) { if (!item || typeof item !== 'object') continue; const normalized = normalizeSourceItem(item); if (!normalized.id) continue; byId.set(normalized.id, normalized); } const mergedSources = Array.from(byId.values()); const maxRaw = Number( process.env.CCTV_MAX_SOURCES || DEFAULT_CCTV_MAX_SOURCES, ); const maxCount = Number.isFinite(maxRaw) ? Math.max(8, Math.min(1200, Math.floor(maxRaw))) : DEFAULT_CCTV_MAX_SOURCES; if (mergedSources.length > maxCount) { console.warn( `[CCTV] source catalog ${mergedSources.length} exceeds cap ${maxCount}; keeping the first ${maxCount} (raise CCTV_MAX_SOURCES or lower a per-pack cap to change which).`, ); } const capped = mergedSources.length > maxCount ? mergedSources.slice(0, maxCount) : mergedSources; if (capped.length > 0 || _cctvSourceCache.length === 0) { _cctvSourceCache = capped; } else { // Every source came back empty (all live packs timed out / upstream outage) // but a good catalog is already cached — serve it stale rather than blanking // every CCTV route. Advancing the timestamp waits one TTL before retrying, // which (with single-flight) bounds load on a persistently-down upstream. console.warn( `[CCTV] source refresh returned empty; serving ${_cctvSourceCache.length} stale cameras`, ); } _cctvSourceCacheAt = Date.now(); return _cctvSourceCache; } return getCctvSources; }