scripts/ais-relay.cjs
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env node
/**
* AIS WebSocket Relay Server
* Proxies aisstream.io data to browsers via WebSocket
*
* Deploy on Railway with:
* AISSTREAM_API_KEY=your_key
*
* Local: node scripts/ais-relay.cjs
*
* @notification-source: domain (ais)
* Every publishNotificationEvent() call in this file builds payload.title
* from structured AIS/vessel/port domain fields (MMSI, vessel name, ETA,
* port code, etc.). Events are NOT RSS-origin and MUST NOT set
* payload.description. Enforced by tests/notification-relay-payload-audit.test.mjs.
*/
const http = require('http');
const https = require('https');
const zlib = require('zlib');
const path = require('path');
const { readFileSync } = require('fs');
const { execFile } = require('child_process');
const crypto = require('crypto');
const v8 = require('v8');
const { WebSocketServer, WebSocket } = require('ws');
const { parseProxyConfig, resolveProxyString, resolveProxyStringForAttempt } = require('./_proxy-utils.cjs');
const { parseWidgetAgentResponse } = require('./_widget-response-parser.cjs');
const { WIDGET_DATA_CATALOG, buildWidgetDataUrl } = require('./_widget-data-policy.cjs');
const {
cooldownKeyForAccount,
OPENSKY_LEGACY_COOLDOWN_KEY,
OPENSKY_MAX_COOLDOWN_MS,
OPENSKY_SHARED_FALLBACK_COOLDOWN_MS,
OPENSKY_MAX_DEADLINE_SET_LUA,
accountFingerprint: openSkyAccountFingerprint,
clampCooldownMs,
ttlSecondsForCooldown,
inspectCooldownRecord,
buildCooldownRecord,
maxDeadlineSetCommand,
legacyCooldownCompatibilityEnabled,
} = require('./_opensky-account-cooldown.cjs');
const OPENSKY_ACCOUNT_FINGERPRINT = openSkyAccountFingerprint(process.env.OPENSKY_CLIENT_ID);
const OPENSKY_COOLDOWN_KEY = cooldownKeyForAccount(OPENSKY_ACCOUNT_FINGERPRINT);
const {
OPENROUTER_FREE_BACKUP_MODEL,
OPENROUTER_FREE_PRIMARY_MODEL,
OPENROUTER_PROVIDER_ROUTING,
} = require('./lib/llm-model-policy.cjs');
const xNewsAccounts = require('./lib/x-news-accounts.cjs');
const { SAUDI_CIVIL_DEFENSE, MAX_POST_CHARS, publishSaudiCivilDefenseAlerts } = require('./lib/saudi-civil-defense-alerts.cjs');
const { publishChokepointDisruptionAlerts } = require('./lib/chokepoint-disruption-alerts.cjs');
// Loaded defensively: the gazetteer validates its data files at require time,
// and a bad regenerated data file must disable only this feed, not the relay.
let middleEastStrikes = null;
try {
middleEastStrikes = require('./lib/middle-east-strikes.cjs');
} catch (e) {
console.error('[MiddleEastStrikes] Module failed to load — strike feed disabled:', e?.message || e);
}
const { createPollGenerationGuard } = require('./lib/poll-generation-guard.cjs');
const { createXPollCycle, xPollSlot } = require('./lib/x-poll-cycle.cjs');
const { createGuardedLookup, SSRF_BLOCKED_ERROR_CODE } = require('./lib/notification-webhook-ssrf.cjs');
// API fan-out of rss_alert / market_alert queue messages to the Shipping v2
// webhook worker (news.breaking, market.move in shared/event-triggers.json).
const {
API_FANOUT_EVENT_TYPES,
API_FANOUT_LIST_CAP,
API_FANOUT_LIST_KEY,
API_FANOUT_LIST_TTL_SECONDS,
API_FANOUT_TEE_SCRIPT,
occurrenceEventIdFor,
} = require('./lib/occurrence-events.cjs');
const {
createXPostBudget,
X_SHARED_COVERAGE_MODEL,
xPostBudgetServiceStatus,
} = require('./lib/x-post-budget.cjs');
const { buildClassifyCandidateMap, isStaleDigestReplay } = require('./lib/digest-stale-gate.cjs');
const {
buildClassifyDigestUrl,
classifyDigestTransport,
buildClassifyDigestHeaders,
formatClassifyDigestFetchFailure,
shouldWriteClassifySeedMeta,
CLASSIFY_DIGEST_RETRY_DELAYS_MS,
classifyDigestRetryDecision,
} = require('./lib/classify-digest-request.cjs');
const {
DIGEST_WARM_ACTIVATION_KEY,
DIGEST_WARM_HEARTBEAT_KEY,
DIGEST_WARM_HEARTBEAT_TTL_SECONDS,
DIGEST_WARM_INTERVAL_MS,
fetchDigestWarmCell,
runDigestWarmCycle,
shouldPublishDigestWarmHeartbeat,
summarizeDigestWarmCycle,
} = require('./lib/news-digest-warm.cjs');
const {
YahooQuoteSummaryClient,
buildSectorSeedMeta,
buildSectorValuationCoverage,
buildSectorValuationPublication,
collectSectorValuations,
} = require('./_yahoo-sector-valuations.cjs');
const { countryNameToIso2 } = require('./shared/country-name-to-iso2.cjs');
const {
buildDedupMaterial,
classifySetNxResult,
recordDedupOutcome,
} = require('./shared/notification-dedup.cjs');
const {
AVIATION_MIN_SERVED_COVERAGE,
RSS_MIN_SERVED_COVERAGE,
classifyUpstreamOutcome,
nextBackoffMs,
summarizeServedCoverage,
} = require('./_ingestion-coverage.cjs');
const { maintainClosedMarketEquityKeys: maintainClosedMarketEquityKeysWithDeps } = require('./shared/closed-market-equity-maintenance.cjs');
const { recordPizzintHistory } = require('./shared/pizzint-history.cjs');
const { pizzintVenuePoint } = require('./shared/pizzint-location.cjs');
const { getUsEquitySession, isMultiMarketEquityTradingDay } = require('./shared/market-hours.cjs');
const { mergeLastGoodQuotes, planYahooRefresh, resolveMergedQuotesAsOf } = require('./shared/market-quote-refresh.cjs');
const { PREVIOUS_SESSION_CHART_QUERY, awaitsSessionOpen, previousSessionClose } = require('./shared/yahoo-closed-session.cjs');
const { SPARKLINE_HISTORY_QUERY, SPARKLINE_HISTORY_RETRY_SECONDS, buildQuoteSparkline } = require('./shared/quote-sparkline.cjs');
const { crudeOilQuoteProvenance } = require('./shared/oil-quote-provenance.cjs');
// ESM module loaded via require(esm) (Node >= 22.12; relay image is node:24).
// Same implementation the RPC handler scores with — see the module header for
// why it lives in shared/ rather than beside the other scoring helpers.
const { detectTrafficAnomaly, summarizePortwatchHistory } = require('../shared/chokepoint-traffic-anomaly.js');
const { CHOKEPOINT_THREAT_LEVELS } = require('../shared/chokepoint-threat-levels.js');
const { classifyVesselType } = require('../shared/ais-vessel-type.js');
const { readTransitWindow } = require('../shared/chokepoint-transit-window.js');
const { createDurableTransitTracker } = require('../shared/chokepoint-transit-store.js');
const { CHOKEPOINT_TRANSIT_GATES } = require('../shared/chokepoint-transit-gates.js');
const { CORRIDOR_RISK_NAME_MAP, deriveCorridorRiskLevel } = require('../shared/corridor-risk.js');
// AIS upstream reconnect policy: failure classification (transport | auth |
// rate-limit), the throttle ceiling escalation, and the silence verdict. Pure and
// environment-free so tests/ais-watchdog.test.mjs exercises the whole policy
// offline (see that module's header for why this lives outside the relay).
const {
aisSilenceVerdict,
classifyAisFailure,
createAisReconnectPolicy,
} = require('../shared/ais-watchdog.js');
const chinaCountryStockIndexHelpersPromise = import('./_country-stock-index.mjs');
// Terminal handler attached AT DECLARATION. This promise is created at module
// load but not awaited until seedWeatherAlerts() runs, so without a .catch()
// here a rejection is an UNHANDLED rejection: under Node's default
// --unhandled-rejections=throw the relay exits 1 at container start, taking AIS,
// market and RSS with it, rather than degrading one seed. seedWeatherAlerts()
// turns the null into a loud, monitor-visible failure (see below).
const weatherAlertSelectPromise = import('./_weather-alert-select.mjs').catch((e) => {
console.error('[Weather] location helper failed to load:', e?.message || e);
return null;
});
const parseProxyUrl = parseProxyConfig;
const httpsKeepAliveAgent = new https.Agent({ keepAlive: true, maxSockets: 6, timeout: 60_000 });
function requireShared(name) {
const candidates = [path.join(__dirname, '..', 'shared', name), path.join(__dirname, 'shared', name)];
for (const p of candidates) { try { return require(p); } catch {} }
throw new Error(`Cannot find shared/${name}`);
}
const RSS_ALLOWED_DOMAINS = new Set(requireShared('rss-allowed-domains.cjs'));
// /rss is public and the allowlisted hosts' DNS belongs to 428 third parties.
// Vet every answer and connect only to a vetted address, so a dangling or
// rebound feed hostname cannot read Railway-internal endpoints through us.
const rssGuardedLookup = createGuardedLookup();
const RSS_ALLOWED_PROTOCOLS = new Set(['http:', 'https:']);
const RSS_MAX_BODY_BYTES = 5 * 1024 * 1024;
const { COMPARE_AND_DELETE_SCRIPT } = requireShared('compare-and-delete-script.cjs');
// Log effective heap limit at startup (verifies NODE_OPTIONS=--max-old-space-size is active)
const _heapStats = v8.getHeapStatistics();
console.log(`[Relay] Heap limit: ${(_heapStats.heap_size_limit / 1024 / 1024).toFixed(0)}MB`);
const RELAY_TEST_MODE = process.env.RELAY_TEST_MODE === 'true';
const AISSTREAM_URL = RELAY_TEST_MODE && process.env.AISSTREAM_URL
? process.env.AISSTREAM_URL
: 'wss://stream.aisstream.io/v0/stream';
const API_KEY = process.env.AISSTREAM_API_KEY || process.env.VITE_AISSTREAM_API_KEY;
const PORT = process.env.PORT || 3004;
// Unset = every interface, which Railway and the Docker network need. A native
// install with no container boundary sets RELAY_BIND_HOST=127.0.0.1.
const RELAY_BIND_HOST = (process.env.RELAY_BIND_HOST || '').trim() || undefined;
// Self-requests must reach the bound address; `localhost` may resolve to the
// other address family than a specific RELAY_BIND_HOST.
const RELAY_SELF_HOST = !RELAY_BIND_HOST || RELAY_BIND_HOST === '0.0.0.0' || RELAY_BIND_HOST === '::'
? 'localhost'
: (RELAY_BIND_HOST.includes(':') ? `[${RELAY_BIND_HOST}]` : RELAY_BIND_HOST);
// Actual bound port, resolved at listen time. Self-requests must use this:
// with PORT=0 (ephemeral bind, used by tests) the env value is not the port
// the server listens on, and `http://localhost:0` fails outright.
let relayBoundPort = PORT;
if (!API_KEY) {
console.warn('[Relay] AIS disabled: AISSTREAM_API_KEY is not set (other relay and seed services remain available)');
}
const MAX_WS_CLIENTS = 10; // Cap WS clients — app uses HTTP snapshots, not WS
const UPSTREAM_QUEUE_HIGH_WATER = Math.max(500, Number(process.env.AIS_UPSTREAM_QUEUE_HIGH_WATER || 4000));
const UPSTREAM_QUEUE_LOW_WATER = Math.max(
100,
Math.min(UPSTREAM_QUEUE_HIGH_WATER - 1, Number(process.env.AIS_UPSTREAM_QUEUE_LOW_WATER || 1000))
);
const UPSTREAM_QUEUE_HARD_CAP = Math.max(
UPSTREAM_QUEUE_HIGH_WATER + 1,
Number(process.env.AIS_UPSTREAM_QUEUE_HARD_CAP || 8000)
);
const UPSTREAM_DRAIN_BATCH = Math.max(1, Number(process.env.AIS_UPSTREAM_DRAIN_BATCH || 250));
const UPSTREAM_DRAIN_BUDGET_MS = Math.max(2, Number(process.env.AIS_UPSTREAM_DRAIN_BUDGET_MS || 20));
function safeInt(envVal, fallback, min) {
if (envVal == null || envVal === '') return fallback;
const n = Number(envVal);
return Number.isFinite(n) ? Math.max(min, Math.floor(n)) : fallback;
}
// Durations are compared on a MONOTONIC clock. A wall-clock step (NTP correction,
// suspend/resume) can move Date.now() backwards, which makes a
// `Date.now() - lastPositionAt` delta negative and suppresses staleness detection
// for an unbounded time — the one failure that hides itself. Wall time is still
// recorded, but only ever to show an operator an age or a timestamp.
const monoNow = () => performance.now();
// Retry-After is either delta-seconds or an HTTP-date. Only those two documented
// forms are accepted and the result is capped at a day: a malformed or hostile
// header must not be able to park the feed for an arbitrary time, so anything
// unparseable returns null and the ladder decides instead.
function parseRetryAfterHeaderMs(raw) {
if (raw == null) return null;
const text = String(raw).trim();
const DAY_MS = 24 * 60 * 60 * 1000;
if (/^\d+$/.test(text)) {
const seconds = Number(text);
return seconds > 0 ? Math.min(seconds * 1000, DAY_MS) : null;
}
const at = Date.parse(text);
if (!Number.isFinite(at)) return null;
const deltaMs = at - Date.now();
return deltaMs > 0 ? Math.min(deltaMs, DAY_MS) : null;
}
const MAX_VESSELS = safeInt(process.env.AIS_MAX_VESSELS, 20000, 1000);
const MAX_VESSEL_HISTORY = safeInt(process.env.AIS_MAX_VESSEL_HISTORY, 20000, 1000);
const MAX_DENSITY_CELLS = 5000;
const MEMORY_CLEANUP_THRESHOLD_GB = (() => {
const n = Number(process.env.RELAY_MEMORY_CLEANUP_GB);
return Number.isFinite(n) && n > 0 ? n : 2.0;
})();
const RELAY_SHARED_SECRET = process.env.RELAY_SHARED_SECRET || '';
const RELAY_AUTH_HEADER = (process.env.RELAY_AUTH_HEADER || 'x-relay-key').toLowerCase();
// Auth bypass: new canonical name + legacy alias. The new name's verbose
// wording is intentional — it should be hard to set in production by accident.
// The legacy `ALLOW_UNAUTHENTICATED_RELAY` is still accepted but emits a
// deprecation warning so existing self-hosted operators are not broken.
const _AUTH_BYPASS_NEW = process.env.I_UNDERSTAND_THIS_DISABLES_AUTH === 'true';
const _AUTH_BYPASS_OLD = process.env.ALLOW_UNAUTHENTICATED_RELAY === 'true';
const ALLOW_UNAUTHENTICATED_RELAY = _AUTH_BYPASS_NEW || _AUTH_BYPASS_OLD;
if (_AUTH_BYPASS_OLD && !_AUTH_BYPASS_NEW) {
console.warn('[DEPRECATED] ALLOW_UNAUTHENTICATED_RELAY=true is deprecated. Use I_UNDERSTAND_THIS_DISABLES_AUTH=true instead.');
}
const RELAY_RATE_LIMIT_WINDOW_MS = Math.max(1000, Number(process.env.RELAY_RATE_LIMIT_WINDOW_MS || 60000));
const RELAY_RATE_LIMIT_MAX = Number.isFinite(Number(process.env.RELAY_RATE_LIMIT_MAX))
? Number(process.env.RELAY_RATE_LIMIT_MAX) : 1200;
const RELAY_OPENSKY_RATE_LIMIT_MAX = Number.isFinite(Number(process.env.RELAY_OPENSKY_RATE_LIMIT_MAX))
? Number(process.env.RELAY_OPENSKY_RATE_LIMIT_MAX) : 600;
const RELAY_RSS_RATE_LIMIT_MAX = Number.isFinite(Number(process.env.RELAY_RSS_RATE_LIMIT_MAX))
? Number(process.env.RELAY_RSS_RATE_LIMIT_MAX) : 300;
const RELAY_GOOGLE_FLIGHTS_RATE_LIMIT_MAX = Number.isFinite(Number(process.env.RELAY_GOOGLE_FLIGHTS_RATE_LIMIT_MAX))
? Number(process.env.RELAY_GOOGLE_FLIGHTS_RATE_LIMIT_MAX) : 60;
const RELAY_LOG_THROTTLE_MS = Math.max(1000, Number(process.env.RELAY_LOG_THROTTLE_MS || 10000));
const ALLOW_VERCEL_PREVIEW_ORIGINS = process.env.ALLOW_VERCEL_PREVIEW_ORIGINS === 'true';
// OpenSky route selection — ONE selected route, never a cascade.
//
// OpenSky's rate limit is per ACCOUNT (4,000 credits/day, keyed on OPENSKY_CLIENT_ID —
// see scripts/_opensky-account-cooldown.cjs and
// docs/solutions/integration-issues/opensky-bbox-area-billing-flat-top-tier.md), NOT per
// exit IP. A residential exit therefore buys no extra quota, and an automatic
// proxy-on-failure fallback would spend a SECOND account credit retrying the one error
// class a different IP cannot fix (429). That is strictly worse on both axes: more paid
// proxy bytes AND faster quota exhaustion. So the route is selected once, the way
// _gdelt-fetch.mjs selects its own.
//
// `direct` is the default because it is free — OpenSky was 82.5% of the residential
// proxy bill (24.86 GB / $136.75 of a ~$165 quarter) for no quota benefit. `proxy` is
// the operator escape hatch for a genuine IP-level rejection, which is counted
// separately as openskyRouteRejection (403/451) so the health surface can tell
// "flip the route" apart from "wait for credits".
const OPENSKY_ROUTE_DEFAULT = 'direct';
const OPENSKY_ROUTES = Object.freeze(['direct', 'proxy']);
function resolveOpenSkyRoute(raw) {
const value = String(raw || '').trim().toLowerCase();
if (!value) return OPENSKY_ROUTE_DEFAULT;
if (OPENSKY_ROUTES.includes(value)) return value;
console.warn(
`[Relay] Ignoring unknown OPENSKY_ROUTE="${raw}" — expected ${OPENSKY_ROUTES.join(' or ')}; `
+ `using "${OPENSKY_ROUTE_DEFAULT}"`,
);
return OPENSKY_ROUTE_DEFAULT;
}
const OPENSKY_ROUTE_REQUESTED = resolveOpenSkyRoute(process.env.OPENSKY_ROUTE);
const OPENSKY_PROXY_AUTH = process.env.OPENSKY_PROXY_AUTH || process.env.PROXY_URL || '';
// Fail OPEN to the free route: an unconfigured escape hatch must not take OpenSky down.
if (OPENSKY_ROUTE_REQUESTED === 'proxy' && !OPENSKY_PROXY_AUTH) {
console.warn(
'[Relay] OPENSKY_ROUTE=proxy but neither OPENSKY_PROXY_AUTH nor PROXY_URL is set — '
+ 'using the direct route.',
);
}
const OPENSKY_PROXY_ENABLED = OPENSKY_ROUTE_REQUESTED === 'proxy' && !!OPENSKY_PROXY_AUTH;
// The EFFECTIVE route — what requests actually do, which is the only thing an operator
// can act on. Derived from OPENSKY_PROXY_ENABLED rather than tracked separately so the
// two can never disagree: `OPENSKY_ROUTE=proxy` with no credential silently falls back
// to direct, and reporting the REQUESTED value there would send an operator hunting a
// proxy that was never in the request path — in exactly the misconfiguration this
// telemetry exists to diagnose. The requested value stays available for the mismatch.
const OPENSKY_ROUTE = OPENSKY_PROXY_ENABLED ? 'proxy' : 'direct';
const PROXY_URL = process.env.PROXY_URL || ''; // generic residential proxy (US exit) — http://user:pass@host:port or host:port:user:pass (Decodo)
// Tzeva Adom (primary) + OREF (fallback) siren alerts
const TZEVA_ADOM_URL = 'https://api.tzevaadom.co.il/notifications';
const OREF_PROXY_AUTH = process.env.OREF_PROXY_AUTH || ''; // format: user:pass@host:port
const OREF_ALERTS_URL = 'https://www.oref.org.il/WarningMessages/alert/alerts.json';
const OREF_HISTORY_URL = 'https://www.oref.org.il/WarningMessages/alert/History/AlertsHistory.json';
const OREF_POLL_INTERVAL_MS = Math.max(30_000, Number(process.env.OREF_POLL_INTERVAL_MS || 300_000));
const OREF_PROXY_AVAILABLE = !!OREF_PROXY_AUTH;
const SIREN_ALERTS_ENABLED = true; // Tzeva Adom is free, no proxy needed
// Hebrew→English translation dictionaries for siren alerts
const OREF_THREAT_TRANSLATIONS = (() => {
try { return JSON.parse(require('fs').readFileSync(path.join(__dirname, '..', 'data', 'oref-threat-translations-he-en.json'), 'utf8')); }
catch { return {}; }
})();
const OREF_CITY_TRANSLATIONS = (() => {
try { return JSON.parse(require('fs').readFileSync(path.join(__dirname, '..', 'data', 'israeli-localities-he-en.json'), 'utf8')); }
catch { return {}; }
})();
function translateHebrew(text) {
if (!text) return text;
if (OREF_THREAT_TRANSLATIONS[text]) return OREF_THREAT_TRANSLATIONS[text];
if (OREF_CITY_TRANSLATIONS[text]) return OREF_CITY_TRANSLATIONS[text];
let result = text;
for (const [heb, eng] of Object.entries(OREF_THREAT_TRANSLATIONS)) {
if (result.includes(heb)) result = result.replace(heb, eng);
}
return result;
}
function translateCity(city) {
if (!city) return city;
return OREF_CITY_TRANSLATIONS[city] || city;
}
const OREF_DATA_DIR = process.env.OREF_DATA_DIR || '';
const OREF_LOCAL_FILE = (() => {
if (!OREF_DATA_DIR) return '';
try {
const stat = require('fs').statSync(OREF_DATA_DIR);
if (!stat.isDirectory()) { console.warn(`[Relay] OREF_DATA_DIR is not a directory: ${OREF_DATA_DIR}`); return ''; }
} catch { console.warn(`[Relay] OREF_DATA_DIR does not exist: ${OREF_DATA_DIR}`); return ''; }
console.log(`[Relay] OREF local persistence: ${OREF_DATA_DIR}`);
return path.join(OREF_DATA_DIR, 'oref-history.json');
})();
const RELAY_OREF_RATE_LIMIT_MAX = Number.isFinite(Number(process.env.RELAY_OREF_RATE_LIMIT_MAX))
? Number(process.env.RELAY_OREF_RATE_LIMIT_MAX) : 600;
// Fail-closed: refuse to boot without a shared secret. This applies in ALL
// environments (production, dev, self-hosted Docker, etc.) so a self-hosted
// `docker compose up -d` against the published image cannot accidentally
// expose every route. Operators who genuinely need an unauthenticated relay
// must opt in explicitly with the loud `I_UNDERSTAND_THIS_DISABLES_AUTH=true`.
// (#3801 — closes the IS_PRODUCTION_RELAY-only gate that left self-hosted
// deployments wide open.)
if (!RELAY_SHARED_SECRET && !ALLOW_UNAUTHENTICATED_RELAY) {
console.error('[Relay] FATAL: RELAY_SHARED_SECRET is not set.');
console.error('[Relay] Generate a strong random secret and set it on every host running the relay:');
console.error('[Relay] RELAY_SHARED_SECRET="$(openssl rand -hex 32)"');
console.error('[Relay] To explicitly opt out (DEV ONLY — leaves all routes publicly accessible):');
console.error('[Relay] I_UNDERSTAND_THIS_DISABLES_AUTH=true (formerly ALLOW_UNAUTHENTICATED_RELAY=true)');
process.exit(1);
}
// Loud recurring SECURITY warning when auth is effectively disabled. Operators
// who opt out with the bypass var get a bright reminder both at boot and every
// 5 minutes in long-running logs so the no-auth state stays visible.
//
// Auth is effectively disabled ONLY when there's no secret to check. If a
// secret IS set, isAuthorizedRequest() enforces it regardless of the bypass
// flag — the bypass branch only runs when the secret is absent — so we must
// not warn (or report auth.enabled=false on /health) when the secret is set.
const AUTH_EFFECTIVELY_DISABLED = !RELAY_SHARED_SECRET;
if (AUTH_EFFECTIVELY_DISABLED) {
console.warn('[SECURITY] relay is running WITHOUT auth — RELAY_SHARED_SECRET unset and I_UNDERSTAND_THIS_DISABLES_AUTH=true. All non-public routes are reachable by anyone who can hit this port.');
setInterval(() => {
console.warn('[SECURITY] relay STILL running without auth — set RELAY_SHARED_SECRET to lock down non-public routes.');
}, 5 * 60 * 1000).unref();
}
// Separately: if the bypass is set redundantly alongside a real secret, the
// bypass is silently ignored (isAuthorizedRequest enforces the secret). Emit
// a single INFO line so operators can clean up their env without wondering
// why their bypass appears to have no effect.
if (RELAY_SHARED_SECRET && ALLOW_UNAUTHENTICATED_RELAY) {
console.info('[Relay] I_UNDERSTAND_THIS_DISABLES_AUTH=true is ignored — RELAY_SHARED_SECRET is configured and takes precedence. Unset the bypass flag to silence this notice.');
}
// ─────────────────────────────────────────────────────────────
// Upstash Redis REST helpers — persist OREF history across restarts
// ─────────────────────────────────────────────────────────────
const UPSTASH_REDIS_REST_URL = process.env.UPSTASH_REDIS_REST_URL || '';
const UPSTASH_REDIS_REST_TOKEN = process.env.UPSTASH_REDIS_REST_TOKEN || '';
// Self-hosted deployments front Redis with a plain-http REST proxy
// (docker/redis-rest-proxy.mjs) reachable only inside the compose network —
// there's no TLS to terminate and no public exposure, so the https-only
// requirement below (which exists to stop a *real* Upstash bearer token from
// transiting the public internet in cleartext) doesn't apply there. This
// gate stayed https-only after the local proxy shipped, so every seed loop
// in this file silently no-ops against http://redis-rest:80 (see
// SELF_HOSTING.md's redis-rest command allowlist note + 4+ days of
// "[TransitSummary]"/"[CorridorRisk]"/etc. never firing).
// UPSTASH_ALLOW_INSECURE_HTTP is an explicit, off-by-default opt-in (never
// inferred from hostname/URL shape) so a genuine Upstash misconfiguration in
// production can't silently downgrade to plaintext.
const UPSTASH_ALLOW_INSECURE_HTTP = process.env.UPSTASH_ALLOW_INSECURE_HTTP === 'true';
const UPSTASH_ENABLED = !!(
UPSTASH_REDIS_REST_URL &&
UPSTASH_REDIS_REST_TOKEN &&
(UPSTASH_REDIS_REST_URL.startsWith('https://') ||
(UPSTASH_ALLOW_INSECURE_HTTP && UPSTASH_REDIS_REST_URL.startsWith('http://')))
);
// Node's https module can't speak to a plain-http endpoint — resolve the
// matching client once at startup instead of hardcoding https.request at
// each upstash* call site below.
const UPSTASH_HTTP_MODULE = UPSTASH_REDIS_REST_URL.startsWith('http://') ? http : https;
const RELAY_ENV_PREFIX = process.env.RELAY_ENV ? `${process.env.RELAY_ENV}:` : '';
const OREF_REDIS_KEY = `${RELAY_ENV_PREFIX}relay:oref:history:v1`;
const CHROME_UA = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36';
if (UPSTASH_REDIS_REST_URL && !UPSTASH_REDIS_REST_URL.startsWith('https://') && !UPSTASH_ALLOW_INSECURE_HTTP) {
console.warn('[Relay] UPSTASH_REDIS_REST_URL must start with https:// — Redis disabled (set UPSTASH_ALLOW_INSECURE_HTTP=true for a trusted internal http proxy)');
}
if (UPSTASH_ENABLED) {
console.log(`[Relay] Upstash Redis enabled (key: ${OREF_REDIS_KEY}${UPSTASH_REDIS_REST_URL.startsWith('http://') ? ', insecure-http opt-in' : ''})`);
}
function upstashGet(key, onFailure, timeoutMs = 5000) {
return new Promise((resolve) => {
if (!UPSTASH_ENABLED) return resolve(null);
let settled = false;
const finish = (value) => {
if (settled) return;
settled = true;
resolve(value);
};
const fail = (reason) => {
if (settled) return;
settled = true;
if (onFailure) onFailure(reason);
resolve(null);
};
const requestTimeoutMs = Number.isFinite(timeoutMs) && timeoutMs > 0 ? timeoutMs : 5000;
const url = new URL(`/get/${encodeURIComponent(key)}`, UPSTASH_REDIS_REST_URL);
const req = UPSTASH_HTTP_MODULE.request(url, {
method: 'GET',
headers: { Authorization: `Bearer ${UPSTASH_REDIS_REST_TOKEN}` },
timeout: requestTimeoutMs,
}, (resp) => {
if (resp.statusCode < 200 || resp.statusCode >= 300) {
resp.resume();
return fail(`HTTP ${resp.statusCode}`);
}
let data = '';
resp.on('data', (chunk) => { data += chunk; });
resp.on('end', () => {
try {
const parsed = JSON.parse(data);
if (parsed?.result) {
try {
return finish(JSON.parse(parsed.result));
} catch (e) {
return fail(`JSON result parse failed: ${(e && e.message) || e}`);
}
}
finish(null);
} catch (e) {
fail(`JSON response parse failed: ${(e && e.message) || e}`);
}
});
});
req.on('error', (e) => fail((e && e.message) || e));
req.on('timeout', () => { req.destroy(); fail('timeout'); });
req.end();
});
}
function upstashSet(key, value, ttlSeconds) {
return new Promise((resolve) => {
if (!UPSTASH_ENABLED) return resolve(false);
const url = new URL('/', UPSTASH_REDIS_REST_URL);
const body = JSON.stringify(['SET', key, JSON.stringify(value), 'EX', String(ttlSeconds)]);
const req = UPSTASH_HTTP_MODULE.request(url, {
method: 'POST',
headers: {
Authorization: `Bearer ${UPSTASH_REDIS_REST_TOKEN}`,
'Content-Type': 'application/json',
},
timeout: 5000,
}, (resp) => {
let data = '';
resp.on('data', (chunk) => { data += chunk; });
resp.on('end', () => {
try {
const parsed = JSON.parse(data);
resolve(parsed?.result === 'OK');
} catch { resolve(false); }
});
});
req.on('error', () => resolve(false));
req.on('timeout', () => { req.destroy(); resolve(false); });
req.end(body);
});
}
function upstashExpire(key, ttlSeconds) {
return new Promise((resolve) => {
if (!UPSTASH_ENABLED) return resolve(false);
const url = new URL('/', UPSTASH_REDIS_REST_URL);
const body = JSON.stringify(['EXPIRE', key, String(ttlSeconds)]);
const req = UPSTASH_HTTP_MODULE.request(url, {
method: 'POST',
headers: {
Authorization: `Bearer ${UPSTASH_REDIS_REST_TOKEN}`,
'Content-Type': 'application/json',
},
timeout: 5000,
}, (resp) => {
let data = '';
resp.on('data', (chunk) => { data += chunk; });
resp.on('end', () => {
try {
const parsed = JSON.parse(data);
resolve(parsed?.result === 1);
} catch { resolve(false); }
});
});
req.on('error', () => resolve(false));
req.on('timeout', () => { req.destroy(); resolve(false); });
req.end(body);
});
}
function upstashMGet(keys) {
return new Promise((resolve) => {
if (!UPSTASH_ENABLED || keys.length === 0) return resolve([]);
const url = new URL('/pipeline', UPSTASH_REDIS_REST_URL);
const body = JSON.stringify(keys.map((k) => ['GET', k]));
const req = UPSTASH_HTTP_MODULE.request(url, {
method: 'POST',
headers: {
Authorization: `Bearer ${UPSTASH_REDIS_REST_TOKEN}`,
'Content-Type': 'application/json',
},
timeout: 10000,
}, (resp) => {
if (resp.statusCode < 200 || resp.statusCode >= 300) {
resp.resume();
return resolve(keys.map(() => null));
}
let data = '';
resp.on('data', (chunk) => { data += chunk; });
resp.on('end', () => {
try {
const parsed = JSON.parse(data);
resolve(parsed.map((r) => {
if (!r?.result) return null;
try { return JSON.parse(r.result); } catch { return null; }
}));
} catch { resolve(keys.map(() => null)); }
});
});
req.on('error', () => resolve(keys.map(() => null)));
req.on('timeout', () => { req.destroy(); resolve(keys.map(() => null)); });
req.end(body);
});
}
function upstashLpush(key, value) {
return new Promise((resolve) => {
if (!UPSTASH_ENABLED) return resolve(false);
const url = new URL('/', UPSTASH_REDIS_REST_URL);
const serialized = typeof value === 'string' ? value : JSON.stringify(value);
const body = JSON.stringify(['LPUSH', key, serialized]);
const req = UPSTASH_HTTP_MODULE.request(url, {
method: 'POST',
headers: {
Authorization: `Bearer ${UPSTASH_REDIS_REST_TOKEN}`,
'Content-Type': 'application/json',
},
timeout: 5000,
}, (resp) => {
let data = '';
resp.on('data', (chunk) => { data += chunk; });
resp.on('end', () => {
try {
const parsed = JSON.parse(data);
resolve(typeof parsed?.result === 'number' && parsed.result > 0);
} catch { resolve(false); }
});
});
req.on('error', () => resolve(false));
req.on('timeout', () => { req.destroy(); resolve(false); });
req.end(body);
});
}
function upstashSetNx(key, value, ttlSeconds) {
return new Promise((resolve) => {
if (!UPSTASH_ENABLED) return resolve('disabled');
const url = new URL('/', UPSTASH_REDIS_REST_URL);
const serialized = typeof value === 'string' ? value : JSON.stringify(value);
const body = JSON.stringify(['SET', key, serialized, 'NX', 'EX', String(ttlSeconds)]);
const req = UPSTASH_HTTP_MODULE.request(url, {
method: 'POST',
headers: {
Authorization: `Bearer ${UPSTASH_REDIS_REST_TOKEN}`,
'Content-Type': 'application/json',
},
timeout: 5000,
}, (resp) => {
let data = '';
resp.on('data', (chunk) => { data += chunk; });
resp.on('end', () => {
try {
const parsed = JSON.parse(data);
resolve(classifySetNxResult(parsed?.result));
} catch { resolve('error'); }
});
});
req.on('error', () => resolve('error'));
req.on('timeout', () => { req.destroy(); resolve('error'); });
req.end(body);
});
}
// ─────────────────────────────────────────────────────────────
// Boot-seed freshness guard
//
// ais-relay is a long-running HTTP service on proxy.worldmonitor.app that
// Railway recycles frequently (deploys, crashes, OOM). Every seed loop fires an
// IMMEDIATE seed on boot and then schedules a setInterval at its real cadence —
// but the process is usually recycled long before that interval elapses, so the
// boot seed, not the interval, is the de-facto scheduler. During a reboot storm
// that means every upstream gets re-fetched on every boot (~8 min apart in the
// wild, observed 2026-06-06) instead of on its interval: paid credits burned for
// ScrapeCreators, and rate-limit/ban risk for Reddit, Yahoo, CoinGecko, UCDP,
// OpenSky, CelesTrak, USNI, etc.
//
// bootSeedDelayMs gates the immediate boot seed on the existing seed-meta age:
// it runs the seed immediately only when the data is already older than its
// interval (i.e. a refresh is actually due). On a frequently-recycled relay this
// self-throttles the boot seed to the intended cadence. On a long-lived relay,
// startBootSeedLoop schedules the first skipped refresh for the remaining
// freshness window and only then starts the recurring interval, so a restart near
// the end of a long interval does not push the next fetch out by another full
// interval.
//
// It keys on `fetchedAt` ("recently attempted — don't re-attempt"), NOT
// recordCount/status, so a recent FAILED attempt also suppresses the next-boot
// retry. That is deliberate: re-attempting a failing paid upstream on every 8-min
// reboot is the exact abuse being prevented. Seeders that only write seed-meta on
// success leave `fetchedAt` stale on failure and so still retry on boot; the
// in-process retry timers cover stable relays.
//
// `metaKey` is the FULL Redis key the seeder writes (usually `seed-meta:<key>`,
// sometimes a `relay:heartbeat:<key>` for script-delegated seeders). On any
// read/parse failure the guard logs and fails OPEN (returns 0 delay, so the
// caller seeds) so a Redis blip never starves a panel.
async function bootSeedDelayMs(label, metaKey, intervalMs) {
if (UPSTASH_ENABLED && metaKey && intervalMs > 0) {
const meta = await upstashGet(metaKey, (reason) => {
console.warn(`[${label}] Boot freshness check failed (${reason}); seeding`);
});
const fetchedAt = Number(meta && meta.fetchedAt) || 0;
if (fetchedAt > 0) {
const ageMs = Date.now() - fetchedAt;
if (ageMs >= 0 && ageMs < intervalMs) {
const delayMs = intervalMs - ageMs;
console.log(`[${label}] Boot seed delayed — data fresh (age ${Math.round(ageMs / 60000)}min < ${Math.round(intervalMs / 60000)}min interval); next refresh in ${Math.round(delayMs / 60000)}min`);
return delayMs;
}
}
}
return 0;
}
function startBootSeedLoop(label, metaKey, intervalMs, seedFn, onInitialError, onSeedError = onInitialError) {
let intervalStarted = false;
const startInterval = () => {
if (intervalStarted) return;
intervalStarted = true;
setInterval(() => {
seedFn().catch(onSeedError);
}, intervalMs).unref?.();
};
bootSeedDelayMs(label, metaKey, intervalMs).then((delayMs) => {
if (delayMs > 0) {
const timer = setTimeout(() => {
Promise.resolve()
.then(seedFn)
.catch(onInitialError)
.finally(startInterval);
}, delayMs);
timer.unref?.();
return;
}
seedFn().catch(onInitialError);
startInterval();
}).catch((e) => {
onInitialError(e);
seedFn().catch(onInitialError);
startInterval();
});
}
function upstashDel(key) {
return new Promise((resolve) => {
if (!UPSTASH_ENABLED) return resolve(false);
const url = new URL('/', UPSTASH_REDIS_REST_URL);
const body = JSON.stringify(['DEL', key]);
const req = UPSTASH_HTTP_MODULE.request(url, {
method: 'POST',
headers: {
Authorization: `Bearer ${UPSTASH_REDIS_REST_TOKEN}`,
'Content-Type': 'application/json',
},
timeout: 5000,
}, (resp) => {
let data = '';
resp.on('data', (chunk) => { data += chunk; });
resp.on('end', () => {
try { resolve(JSON.parse(data)?.result === 1); } catch { resolve(false); }
});
});
req.on('error', () => resolve(false));
req.on('timeout', () => { req.destroy(); resolve(false); });
req.end(body);
});
}
function upstashEval(script, keys, args) {
return new Promise((resolve) => {
if (!UPSTASH_ENABLED) return resolve(null);
const url = new URL('/', UPSTASH_REDIS_REST_URL);
const body = JSON.stringify(['EVAL', script, String(keys.length), ...keys, ...args.map((arg) => String(arg))]);
const req = UPSTASH_HTTP_MODULE.request(url, {
method: 'POST',
headers: {
Authorization: `Bearer ${UPSTASH_REDIS_REST_TOKEN}`,
'Content-Type': 'application/json',
},
timeout: 5000,
}, (resp) => {
let data = '';
resp.on('error', () => resolve(null));
resp.on('aborted', () => resolve(null));
resp.on('data', (chunk) => { data += chunk; });
resp.on('end', () => {
try { resolve(JSON.parse(data)?.result); } catch { resolve(null); }
});
});
req.on('error', () => resolve(null));
req.on('timeout', () => { req.destroy(); resolve(null); });
req.end(body);
});
}
function upstashReleaseLockIfOwner(key, owner) {
return new Promise((resolve) => {
if (!UPSTASH_ENABLED) return resolve(false);
const url = new URL('/', UPSTASH_REDIS_REST_URL);
const body = JSON.stringify(['EVAL', COMPARE_AND_DELETE_SCRIPT, '1', key, owner]);
const req = UPSTASH_HTTP_MODULE.request(url, {
method: 'POST',
headers: {
Authorization: `Bearer ${UPSTASH_REDIS_REST_TOKEN}`,
'Content-Type': 'application/json',
},
timeout: 5000,
}, (resp) => {
let data = '';
resp.on('data', (chunk) => { data += chunk; });
resp.on('end', () => {
try { resolve(Number(JSON.parse(data)?.result) === 1); } catch { resolve(false); }
});
});
req.on('error', () => resolve(false));
req.on('timeout', () => { req.destroy(); resolve(false); });
req.end(body);
});
}
function upstashPublishXIfLockOwner({ lockKey, owner, snapshotKey, snapshot, pollStateKey, pollState, ttlSeconds, metaKey, meta, metaTtlSeconds }) {
return new Promise((resolve) => {
if (!UPSTASH_ENABLED) return resolve(false);
const url = new URL('/', UPSTASH_REDIS_REST_URL);
const script = [
'if redis.call("get",KEYS[1]) ~= ARGV[1] then return 0 end',
'redis.call("set",KEYS[2],ARGV[2],"EX",ARGV[4])',
'redis.call("set",KEYS[3],ARGV[3],"EX",ARGV[4])',
'if ARGV[5] == "1" then redis.call("set",KEYS[4],ARGV[6],"EX",ARGV[7]) end',
'return 1',
].join(' ');
const body = JSON.stringify([
'EVAL', script, '4', lockKey, snapshotKey, pollStateKey, metaKey,
owner, JSON.stringify(snapshot), JSON.stringify(pollState), String(ttlSeconds),
meta ? '1' : '0', JSON.stringify(meta || {}), String(metaTtlSeconds),
]);
const req = UPSTASH_HTTP_MODULE.request(url, {
method: 'POST',
headers: {
Authorization: `Bearer ${UPSTASH_REDIS_REST_TOKEN}`,
'Content-Type': 'application/json',
},
timeout: 5000,
}, (resp) => {
let data = '';
resp.on('data', (chunk) => { data += chunk; });
resp.on('end', () => {
try { resolve(Number(JSON.parse(data)?.result) === 1); } catch { resolve(false); }
});
});
req.on('error', () => resolve(false));
req.on('timeout', () => { req.destroy(); resolve(false); });
req.end(body);
});
}
// ─────────────────────────────────────────────────────────────
// Seed envelope — canonical { _seed, data } shape. Mirrored from
// scripts/_seed-envelope-source.mjs (ESM; can't be require()'d from CJS).
// Source of truth lives there + api/_seed-envelope.js + server/_shared/seed-envelope.ts.
// Parity enforced by scripts/verify-seed-envelope-parity.mjs.
// ─────────────────────────────────────────────────────────────
function buildEnvelope({ fetchedAt, recordCount, sourceVersion, schemaVersion, state, failedDatasets, errorReason, groupId, data }) {
const _seed = { fetchedAt, recordCount, sourceVersion, schemaVersion, state };
if (failedDatasets != null) _seed.failedDatasets = failedDatasets;
if (errorReason != null) _seed.errorReason = errorReason;
if (groupId != null) _seed.groupId = groupId;
return { _seed, data };
}
// Wrap `data` in a seed envelope and write to Redis at `key` with `ttlSeconds`.
// meta: { fetchedAt?, recordCount, sourceVersion, schemaVersion?, state?, zeroOk?, groupId? }
// - state: omit to derive ('OK_ZERO' when recordCount===0 && zeroOk, else 'OK')
// - schemaVersion: defaults to 1
function envelopeWrite(key, data, ttlSeconds, meta) {
const recordCount = Number(meta?.recordCount ?? 0) || 0;
const state = meta?.state || (recordCount === 0 && meta?.zeroOk ? 'OK_ZERO' : 'OK');
const envelope = buildEnvelope({
fetchedAt: meta?.fetchedAt ?? Date.now(),
recordCount,
sourceVersion: meta?.sourceVersion || 'ais-relay',
schemaVersion: meta?.schemaVersion ?? 1,
state,
groupId: meta?.groupId,
data,
});
return upstashSet(key, envelope, ttlSeconds);
}
// Envelope-aware read. Mirrors server/_shared/redis.ts::getCachedJson semantics:
// returns the bare payload for contract-mode canonical keys ({_seed, data}) and
// passes legacy shapes through unchanged. MUST be used for any seeded canonical
// key — reading raw via upstashGet() on an enveloped key iterates {_seed, data}
// as payload keys and silently corrupts downstream consumers.
async function envelopeRead(key, onFailure) {
const raw = await upstashGet(key, onFailure);
if (raw && typeof raw === 'object' && !Array.isArray(raw) && '_seed' in raw && 'data' in raw) {
return raw.data;
}
return raw;
}
function notifySimpleHash(str) {
let h = 0;
for (let i = 0; i < str.length; i++) h = (Math.imul(31, h) + str.charCodeAt(i)) | 0;
return Math.abs(h).toString(36);
}
function normalizeNotificationCountryCode(raw) {
return countryNameToIso2(raw) ?? undefined;
}
function marketAlertCoalesceKey(assetClass, identifier, direction, severity) {
const stableIdentifier = String(identifier || 'unknown').trim().toLowerCase();
return `market:${assetClass}:${stableIdentifier}:${direction}:${severity}`;
}
function nwsVtec(p) {
const vtec = Array.isArray(p?.parameters?.VTEC) ? p.parameters.VTEC[0] : undefined;
return vtec;
}
/**
* Second copy of a queued rss_alert / market_alert message for API-key webhook
* subscribers. The notification relay RPOPs wm:events:queue destructively, so
* the Shipping v2 worker drains its own list (wm:events:api:v1). One Lua round
* trip pushes, caps and expires it; a failure here never touches the queue
* publish that already succeeded, it only loses the event for API tenants.
*/
function teeApiFanout(eventType, msg) {
if (!UPSTASH_ENABLED) return;
upstashEval(API_FANOUT_TEE_SCRIPT, [API_FANOUT_LIST_KEY], [msg, API_FANOUT_LIST_CAP, API_FANOUT_LIST_TTL_SECONDS])
.then((result) => {
if (Number(result) !== 1) console.warn(`[Notify] API fan-out tee did not confirm (${eventType})`);
})
.catch((e) => console.warn(`[Notify] API fan-out tee failed (${eventType}):`, e?.message || e));
}
/** Quote fields the market.move webhook payload carries; the title stays the human line. */
function marketMoveFields(assetClass, q) {
return {
assetClass,
symbol: String(q.symbol || q.name || '').toUpperCase(),
name: String(q.name || q.symbol || ''),
price: Number.isFinite(q.price) ? q.price : null,
changePct: q.change,
direction: q.change < 0 ? 'down' : 'up',
};
}
async function publishNotificationEvent({ eventType, payload, severity, variant, dedupTtl = 1800 }) {
try {
// Include variant in dedup key so each variant can independently publish the same title
// (e.g. finance and world users both receive an alert for the same headline).
// Slot B: when payload.coalesceKey is set (e.g. NWS VTEC family), key on it
// instead of the title hash so adjacent-zone alerts for the same logical
// event collapse at the publisher (queue stays clean) instead of N times
// per recipient at the relay.
const variantSuffix = variant ? `:${variant}` : '';
const dedupMaterial = buildDedupMaterial(eventType, payload?.title, payload?.coalesceKey);
const dedupKey = `wm:notif:scan-dedup:${eventType}${variantSuffix}:${notifySimpleHash(dedupMaterial)}`;
const dedupResult = await upstashSetNx(dedupKey, '1', dedupTtl);
const dedupDecision = recordDedupOutcome(dedupResult, {
surface: 'ais-relay',
eventType,
severity,
fallbackKey: dedupKey,
fallbackTtlSeconds: dedupTtl,
emitTelemetry: recordNotificationDedupSetNxError,
});
// Returns true once the event is queued or was already queued; callers
// that keep their own edge state retry on false.
if (!dedupDecision.shouldPublish) {
if (!dedupDecision.isDuplicate) return false;
console.log(`[Notify] Dedup hit — ${eventType}: ${String(payload.title ?? '').slice(0, 60)}`);
return true;
}
const publishedAt = Date.now();
const fanout = API_FANOUT_EVENT_TYPES.has(eventType);
// Stable per-story / per-move id for API subscribers (occurrence-events.cjs).
// The relay ignores it; it rides along in the queue message unchanged.
if (fanout && payload && typeof payload === 'object' && !payload.eventId) {
const eventId = occurrenceEventIdFor(eventType, payload, publishedAt);
if (eventId) payload = { ...payload, eventId };
}
const msg = JSON.stringify({ eventType, payload, severity: dedupDecision.severity, ...(variant ? { variant } : {}), publishedAt });
const ok = await upstashLpush('wm:events:queue', msg);
if (ok) {
if (fanout) teeApiFanout(eventType, msg);
console.log(`[Notify] Queued ${dedupDecision.severity} event: ${eventType} — ${String(payload.title ?? '').slice(0, 60)}`);
return true;
} else {
// Rollback the dedup key so the next poll cycle can retry — avoids silent
// suppression for the full dedupTtl when a transient LPUSH fails.
console.warn(`[Notify] LPUSH failed for ${eventType} — rolling back dedup key`);
upstashDel(dedupKey).catch(() => {});
return false;
}
} catch (e) {
console.warn(`[Notify] publishNotificationEvent error (${eventType}):`, e?.message || e);
return false;
}
}
let upstreamSocket = null;
let upstreamReconnectTimer = null;
let upstreamReconnectAt = 0;
let upstreamLastPositionAt = 0;
// Monotonic mirrors. Every freshness DECISION reads these; the wall values above
// exist only so an operator can be told an age or a timestamp.
let upstreamLastPositionMono = 0;
let upstreamSocketOpenedMono = 0;
// The delay the reconnect policy computed for the failure that just occurred.
// `scheduleUpstreamReconnect` consumes it, so the ladder ceiling and the throttle
// escalation are decided once inside shared/ais-watchdog.js rather than re-derived
// from counters here (the two drifting apart is how a stalled feed goes unnoticed).
let upstreamPendingReconnectDelayMs = null;
let relayShuttingDown = false;
// Floors are deliberate: this change exists to stop the relay hammering a
// provider that is rejecting it, so the tunables must not open a config path to
// hammer harder. 50ms/100ms stay well below any production value while leaving
// the ladder compressible under test.
const AIS_RECONNECT_BASE_MS = safeInt(process.env.AIS_RECONNECT_BASE_MS, 5_000, 50);
const AIS_RECONNECT_MAX_MS = safeInt(process.env.AIS_RECONNECT_MAX_MS, 5 * 60 * 1000, 100);
// A 429 on the WebSocket upgrade means the provider is rate-limiting this egress
// IP, not that the stream failed transiently: the rejection lands before the API
// key is ever sent. Knocking every AIS_RECONNECT_MAX_MS keeps ~288 refused
// requests/day inside the provider's sliding window, which can sustain the block
// we are waiting out. Sustained throttling therefore escalates to a much longer
// ceiling; any non-throttle outcome clears it so ordinary disconnects keep the
// responsive schedule.
//
// Clamped at or above the ordinary ceiling: a smaller value would make
// escalation *shorten* the wait and reconnect more aggressively while throttled,
// which is the exact inverse of the intent.
const AIS_THROTTLE_RECONNECT_MAX_MS = Math.max(
AIS_RECONNECT_MAX_MS,
safeInt(process.env.AIS_THROTTLE_RECONNECT_MAX_MS, 30 * 60 * 1000, 100),
);
const AIS_THROTTLE_ESCALATE_AFTER = safeInt(
process.env.AIS_THROTTLE_ESCALATE_AFTER,
5,
1,
);
const AIS_HANDSHAKE_TIMEOUT_MS = safeInt(
process.env.AIS_HANDSHAKE_TIMEOUT_MS,
30_000,
1_000,
);
const AIS_POSITION_FRESHNESS_MS = safeInt(
process.env.AIS_POSITION_FRESHNESS_MS,
5 * 60 * 1000,
1_000,
);
// Silence is REPORTED well before the connection is recycled. The provider allows
// one stream per key, so aborting on the first whiff of staleness would churn the
// only connection we are permitted to hold; telling the operator early costs
// nothing. Stale-but-not-recycled is the `positionStale` health field.
const AIS_POSITION_STALE_MS = safeInt(
process.env.AIS_POSITION_STALE_MS,
Math.floor(AIS_POSITION_FRESHNESS_MS * 0.6),
100,
);
// A refused credential is not a transient failure. Ordinary retrying cannot fix a
// revoked key, and the ladder would otherwise knock hundreds of times a day on a
// provider that has already said no. The slow probe exists only to notice an
// upstream-side mistake; valid data or a credential change clears the state.
const AIS_AUTH_PROBE_MS = safeInt(
process.env.AIS_AUTH_PROBE_MS,
60 * 60 * 1000,
60_000,
);
// The reconnect policy owns: what a failure MEANS, how many consecutive throttles
// have happened, whether the ceiling is escalated, and the auth terminal state.
// The relay keeps owning the transport and the single retry timer.
const aisReconnectPolicy = createAisReconnectPolicy({
// nextBackoffMs is 0-indexed (failures already elapsed); the policy is
// 1-indexed (failures INCLUDING this one), hence the -1. Jitter and the cap stay
// in nextBackoffMs so there is one ladder implementation, not two.
ladder: (attempts, ceilingMs) => nextBackoffMs(attempts - 1, AIS_RECONNECT_BASE_MS, ceilingMs),
maxMs: AIS_RECONNECT_MAX_MS,
throttleCeilingMs: AIS_THROTTLE_RECONNECT_MAX_MS,
escalateAfter: AIS_THROTTLE_ESCALATE_AFTER,
authProbeMs: AIS_AUTH_PROBE_MS,
});
const aisUpstreamMetrics = {
connectionAttempts: 0,
success: 0,
throttle: 0,
terminalFailure: 0,
lastSuccessAt: 0,
lastFailureAt: 0,
lastFailure: null,
};
let upstreamPaused = false;
let upstreamQueue = [];
let upstreamQueueReadIndex = 0;
let upstreamDrainScheduled = false;
const clients = new Set();
let messageCount = 0;
let droppedMessages = 0;
const requestRateBuckets = new Map(); // key: route:ip -> { count, resetAt }
const logThrottleState = new Map(); // key: event key -> timestamp
function isAisThrottleEscalated() {
return aisReconnectPolicy.snapshot().escalated;
}
// Liveness is proven by DATA, never by the socket. `readyState === OPEN` only
// means a handshake succeeded once; AISstream can complete the upgrade and then
// send nothing at all — no frame, no error, no close — which is the stall this
// whole watchdog exists to catch. So readiness is recomputed from the age of the
// last ACCEPTED position, on a monotonic clock, and silence gets its own verdict
// (report at `AIS_POSITION_STALE_MS`, recycle at `AIS_POSITION_FRESHNESS_MS`).
function getAisPositionFreshness(nowMs = Date.now()) {
const positionAgeMonoMs = upstreamLastPositionMono
? Math.max(0, monoNow() - upstreamLastPositionMono)
: null;
return {
currentPositionReady: upstreamSocket?.readyState === WebSocket.OPEN
&& positionAgeMonoMs !== null
&& positionAgeMonoMs <= AIS_POSITION_FRESHNESS_MS,
// Wall age, for operators. The verdict below never uses it.
positionAgeMs: upstreamLastPositionAt
? Math.max(0, nowMs - upstreamLastPositionAt)
: null,
positionStale: positionAgeMonoMs !== null && aisSilenceVerdict({
silentForMs: positionAgeMonoMs,
staleMs: AIS_POSITION_STALE_MS,
recycleAfterMs: AIS_POSITION_FRESHNESS_MS,
}) === 'stale',
};
}
// Safe response: guard against "headers already sent" crashes
function safeEnd(res, statusCode, headers, body) {
if (res.headersSent || res.writableEnded) return false;
try {
res.writeHead(statusCode, headers);
res.end(body);
return true;
} catch {
return;
}
}
const WORLD_BANK_COUNTRY_ALLOWLIST = new Set([
'USA','CHN','JPN','DEU','KOR','GBR','IND','ISR','SGP','TWN',
'FRA','CAN','SWE','NLD','CHE','FIN','IRL','AUS','BRA','IDN',
'ARE','SAU','QAT','BHR','EGY','TUR','MYS','THA','VNM','PHL',
'ESP','ITA','POL','CZE','DNK','NOR','AUT','BEL','PRT','EST',
'MEX','ARG','CHL','COL','ZAF','NGA','KEN',
]);
function normalizeWorldBankCountryCodes(rawValue) {
const parts = String(rawValue || '')
.split(/[;,]/g)
.map((part) => part.trim().toUpperCase())
.filter((part) => /^[A-Z]{3}$/.test(part) && WORLD_BANK_COUNTRY_ALLOWLIST.has(part));
return parts.length > 0 ? parts.join(';') : null;
}
function _acceptsEncoding(header, encoding) {
if (!header) return false;
const tokens = header.split(',');
for (const token of tokens) {
const parts = token.trim().split(';');
if (parts[0].trim().toLowerCase() !== encoding) continue;
const qPart = parts.find(p => p.trim().startsWith('q='));
if (qPart && parseFloat(qPart.trim().substring(2)) === 0) return false;
return true;
}
return false;
}
function _varyHeader(res) {
const existing = String(res.getHeader('vary') || '');
return existing.toLowerCase().includes('accept-encoding')
? existing
: (existing ? `${existing}, Accept-Encoding` : 'Accept-Encoding');
}
// Compress & send a response (Brotli preferred ~15-20% smaller than gzip on JSON)
function sendCompressed(req, res, statusCode, headers, body) {
if (res.headersSent || res.writableEnded) return;
const ae = req.headers['accept-encoding'] || '';
const buf = typeof body === 'string' ? Buffer.from(body) : body;
if (_acceptsEncoding(ae, 'br')) {
zlib.brotliCompress(buf, { params: { [zlib.constants.BROTLI_PARAM_QUALITY]: 4 } }, (err, compressed) => {
if (err || res.headersSent || res.writableEnded) {
safeEnd(res, statusCode, headers, body);
return;
}
safeEnd(res, statusCode, { ...headers, 'Content-Encoding': 'br', 'Vary': _varyHeader(res) }, compressed);
});
} else if (_acceptsEncoding(ae, 'gzip')) {
zlib.gzip(buf, (err, compressed) => {
if (err || res.headersSent || res.writableEnded) {
safeEnd(res, statusCode, headers, body);
return;
}
safeEnd(res, statusCode, { ...headers, 'Content-Encoding': 'gzip', 'Vary': _varyHeader(res) }, compressed);
});
} else {
safeEnd(res, statusCode, headers, body);
}
}
// Pre-compressed response: serve cached gzip/brotli buffer directly (zero CPU per request)
function sendPreGzipped(req, res, statusCode, headers, rawBody, gzippedBody, brotliBody) {
if (res.headersSent || res.writableEnded) return;
const ae = req.headers['accept-encoding'] || '';
if (_acceptsEncoding(ae, 'br') && brotliBody) {
safeEnd(res, statusCode, { ...headers, 'Content-Encoding': 'br', 'Vary': _varyHeader(res) }, brotliBody);
} else if (_acceptsEncoding(ae, 'gzip') && gzippedBody) {
safeEnd(res, statusCode, { ...headers, 'Content-Encoding': 'gzip', 'Vary': _varyHeader(res) }, gzippedBody);
} else {
safeEnd(res, statusCode, headers, rawBody);
}
}
// ─────────────────────────────────────────────────────────────
// Telegram OSINT ingestion (public channels) → Early Signals
// Web-first: runs on this Railway relay process, serves /telegram/feed
// Requires env:
// - TELEGRAM_API_ID
// - TELEGRAM_API_HASH
// - TELEGRAM_SESSION (StringSession)
// ─────────────────────────────────────────────────────────────
function readTelegramNumberEnv(name, fallback, min = 0, max = Number.POSITIVE_INFINITY) {
const raw = process.env[name];
const parsed = raw == null || raw === '' ? fallback : Number(raw);
const finite = Number.isFinite(parsed) ? parsed : fallback;
return Math.min(max, Math.max(min, finite));
}
const TELEGRAM_ENABLED = Boolean(process.env.TELEGRAM_API_ID && process.env.TELEGRAM_API_HASH && process.env.TELEGRAM_SESSION);
const TELEGRAM_POLL_INTERVAL_FLOOR_MS = RELAY_TEST_MODE ? 10 : 15_000;
const TELEGRAM_POLL_INTERVAL_MS = Math.max(
TELEGRAM_POLL_INTERVAL_FLOOR_MS,
Number(process.env.TELEGRAM_POLL_INTERVAL_MS || 60_000),
);
const TELEGRAM_MAX_FEED_ITEMS = Math.max(50, Number(process.env.TELEGRAM_MAX_FEED_ITEMS || 200));
const TELEGRAM_MAX_TEXT_CHARS = Math.max(200, Number(process.env.TELEGRAM_MAX_TEXT_CHARS || 800));
const TELEGRAM_STARTUP_DELAY_MS = readTelegramNumberEnv('TELEGRAM_STARTUP_DELAY_MS', 120_000);
const TELEGRAM_CONNECT_TIMEOUT_MS = readTelegramNumberEnv('TELEGRAM_CONNECT_TIMEOUT_MS', 30_000, 10);
const telegramState = {
client: null,
api: null,
initPromise: null,
channels: [],
cursorByHandle: Object.create(null),
items: [],
lastPollAt: 0,
lastSuccessfulPollAt: 0,
lastError: null,
startedAt: Date.now(),
};
// The feed lives in process memory and a new container waits
// TELEGRAM_STARTUP_DELAY_MS before polling, so every deploy served an empty
// feed for minutes and blanked the panel for every open dashboard. Each good
// poll snapshots the items; a starting container serves that snapshot until
// its own first poll. The TTL bounds how old a restored feed can be.
const TELEGRAM_ITEMS_SNAPSHOT_KEY = 'intelligence:telegram-feed:items:v1';
const TELEGRAM_ITEMS_SNAPSHOT_TTL_SECONDS = 1800;
const TELEGRAM_RESOLVE_CACHE_TTL_MS = 60 * 60 * 1000;
const TELEGRAM_CHANNEL_CACHE_TTL_MS = 30_000;
const TELEGRAM_LOOKUP_CACHE_MAX_ENTRIES = 512;
const TELEGRAM_NEGATIVE_CACHE_TTL_MS = readTelegramNumberEnv('TELEGRAM_NEGATIVE_CACHE_TTL_MS', 300_000, 1_000);
const TELEGRAM_RPC_MAX_CONCURRENCY = Math.floor(readTelegramNumberEnv('TELEGRAM_RPC_MAX_CONCURRENCY', 2, 1, 4));
const TELEGRAM_RPC_MAX_QUEUE = Math.floor(readTelegramNumberEnv('TELEGRAM_RPC_MAX_QUEUE', 32, 1, 128));
const TELEGRAM_RPC_QUEUE_TIMEOUT_MS = readTelegramNumberEnv('TELEGRAM_RPC_QUEUE_TIMEOUT_MS', 5_000, 10);
// The 300ms floor is load-bearing and predates the RPC queue: it used to live in
// the poll loop as Math.max(300, TELEGRAM_RATE_LIMIT_MS). Reusing that env var
// without the floor silently re-pointed an existing production value at a burst
// posture the old code forbade, so keep the floor here. Tests need sub-floor
// spacing to stay fast, and RELAY_TEST_MODE is already this file's test seam.
const TELEGRAM_RPC_MIN_INTERVAL_FLOOR_MS = RELAY_TEST_MODE ? 0 : 300;
const TELEGRAM_RPC_MIN_INTERVAL_MS = readTelegramNumberEnv(
'TELEGRAM_RPC_MIN_INTERVAL_MS',
readTelegramNumberEnv('TELEGRAM_RATE_LIMIT_MS', 800, TELEGRAM_RPC_MIN_INTERVAL_FLOOR_MS),
TELEGRAM_RPC_MIN_INTERVAL_FLOOR_MS,
);
// Backstop for a hostile or absurd upstream FLOOD_WAIT value. The cooldown is
// process-wide and only ever raised, so an uncapped value is an outage.
const TELEGRAM_MAX_FLOOD_WAIT_MS = readTelegramNumberEnv('TELEGRAM_MAX_FLOOD_WAIT_MS', 15 * 60 * 1000, 1_000);
// gramjs disconnect() can hang; the shutdown path already races it against a
// timer. The reconnect path must too, because initTelegramClientIfNeeded awaits
// this promise before anything else.
const TELEGRAM_DISCONNECT_TIMEOUT_MS = readTelegramNumberEnv('TELEGRAM_DISCONNECT_TIMEOUT_MS', 10_000, 10);
// Whole-lookup budget. Individual RPC timeouts compose (queue wait + exec, x3
// for a channel read), so only an outer deadline keeps the relay's worst case
// under the Edge runtime's 25s begin-response ceiling.
const TELEGRAM_LOOKUP_DEADLINE_MS = readTelegramNumberEnv('TELEGRAM_LOOKUP_DEADLINE_MS', 18_000, 100);
// One slow channel is not a dead socket. Only a run of timeouts justifies
// resetting the client that every other caller (and the curated poll) shares.
const TELEGRAM_MAX_CONSECUTIVE_RPC_TIMEOUTS = Math.floor(
readTelegramNumberEnv('TELEGRAM_MAX_CONSECUTIVE_RPC_TIMEOUTS', 3, 1, 20),
);
const TELEGRAM_USERNAME_RE = /^[a-zA-Z][a-zA-Z0-9_]{4,31}$/;
const telegramResolveCache = new Map();
const telegramChannelCache = new Map();
const telegramNegativeCache = new Map();
const telegramLookupInflight = new Map();
const telegramRpcQueue = [];
const telegramRpcActiveEntries = new Set();
let telegramRpcNextStartAt = 0;
let telegramRpcDrainTimer = null;
let telegramFloodWaitUntil = 0;
let telegramDisconnectPromise = null;
let telegramConsecutiveRpcTimeouts = 0;
let telegramTestConnectAttempts = 0;
const orefState = {
lastAlerts: [],
lastAlertsJson: '[]',
lastPollAt: 0,
lastError: null,
historyCount24h: 0,
totalHistoryCount: 0,
history: [],
bootstrapSource: null,
_persistVersion: 0,
_lastPersistedVersion: 0,
_persistInFlight: false,
_alertsCache: null, // { json, gzip, brotli }
_historyCache: null, // { json, gzip, brotli }
};
function loadTelegramChannels() {
// Product-managed curated list lives in repo root under data/ (shared by web + desktop).
// Relay is executed from scripts/, so resolve ../data.
const p = path.join(__dirname, '..', 'data', 'telegram-channels.json');
const set = String(process.env.TELEGRAM_CHANNEL_SET || 'full').toLowerCase();
try {
const raw = JSON.parse(readFileSync(p, 'utf8'));
const bucket = raw?.channels?.[set];
const channels = Array.isArray(bucket) ? bucket : [];
telegramState.channels = channels
.filter(c => c && typeof c.handle === 'string' && c.handle.length > 1)
.map(c => ({
handle: String(c.handle).replace(/^@/, ''),
label: c.label ? String(c.label) : undefined,
topic: c.topic ? String(c.topic) : undefined,
region: c.region ? String(c.region) : undefined,
tier: c.tier != null ? Number(c.tier) : undefined,
enabled: c.enabled !== false,
maxMessages: c.maxMessages != null ? Number(c.maxMessages) : undefined,
}))
.filter(c => c.enabled)
// normalizeTelegramMessage runs every handle through
// sanitizeTelegramUsername, which THROWS on anything outside the strict
// username grammar. That throw lands after getEntity and getMessages have
// already been spent, so an unusable handle would burn 2 RPCs per cycle
// forever and never advance its cursor. Reject it loudly, once, at load.
.filter(c => {
if (TELEGRAM_USERNAME_RE.test(c.handle)) return true;
console.warn(`[Relay] Ignoring Telegram channel with invalid handle: ${JSON.stringify(c.handle)}`);
return false;
});
if (!telegramState.channels.length) {
console.warn(`[Relay] Telegram channel set "${set}" is empty — no channels to poll`);
}
return telegramState.channels;
} catch (e) {
telegramState.channels = [];
telegramState.lastError = `failed to load telegram-channels.json: ${e?.message || String(e)}`;
return [];
}
}
function normalizeTelegramMessage(msg, channel) {
const handle = sanitizeTelegramUsername(channel.handle);
const isSaudiCivilDefense = handle.toLowerCase() === SAUDI_CIVIL_DEFENSE.handle.toLowerCase();
const textRaw = String(msg?.message || '');
const text = textRaw.slice(0, isSaudiCivilDefense
? MAX_POST_CHARS : TELEGRAM_MAX_TEXT_CHARS);
const ts = msg?.date ? new Date(msg.date * 1000).toISOString() : isSaudiCivilDefense ? '' : new Date().toISOString();
return {
id: `${handle}:${msg.id}`,
source: 'telegram',
channel: handle,
channelTitle: channel.label || handle,
url: `https://t.me/${handle}/${msg.id}`,
ts,
text,
textTruncated: text.length < textRaw.length,
topic: channel.topic || 'other',
tags: [channel.region].filter(Boolean),
earlySignal: true,
};
}
function sanitizeTelegramUsername(raw) {
const value = String(raw || '')
.trim()
.replace(/^https?:\/\/t\.me\//i, '')
.replace(/^@+/, '')
.replace(/[?#].*$/, '')
.replace(/\/+$/, '')
.trim();
if (!TELEGRAM_USERNAME_RE.test(value)) {
throw new Error('Invalid Telegram username');
}
return value.toLowerCase();
}
function getCachedTelegramValue(cache, key) {
const cached = cache.get(key);
if (!cached) return null;
if (cached.expiresAt <= Date.now()) {
cache.delete(key);
return null;
}
return cached.value;
}
function setCachedTelegramValue(cache, key, value, ttlMs) {
const now = Date.now();
for (const [cachedKey, cached] of cache) {
if (cached.expiresAt <= now) cache.delete(cachedKey);
}
while (cache.size >= TELEGRAM_LOOKUP_CACHE_MAX_ENTRIES) {
const oldestKey = cache.keys().next().value;
if (oldestKey == null) break;
cache.delete(oldestKey);
}
cache.set(key, { value, expiresAt: Date.now() + ttlMs });
}
function getTelegramErrorStatus(error) {
if (Number.isInteger(error?.statusCode)) return error.statusCode;
// Structured first: a genuine flood carries `seconds` / `errorMessage`. The
// message text is attacker-influenced (it embeds the requested username), so
// it must never be what promotes an error to 429 — see parseTelegramFloodWaitMs.
if (parseTelegramFloodWaitMs(error) > 0) return 429;
const message = String(error?.message || error || '');
if (/^TIMEOUT after \d+ms:/.test(message)) return 504;
if (/invalid telegram username/i.test(message)) return 400;
if (/USERNAME_NOT_OCCUPIED|No user has|No channel has|Cannot find any entity/i.test(message)) return 404;
if (/not active|not installed|invalidated/i.test(message)) return 503;
return 502;
}
// Public-facing copy. The relay's own error strings describe internal state
// ('session invalidated (AUTH_KEY_DUPLICATED)', 'Telegram RPC queue is full',
// raw MTProto/transport text) and used to be returned verbatim to any browser
// holding a free session token. Map to status-shaped copy instead.
//
// 400-for-not-a-public-channel is deliberately folded into 404: distinguishing
// "no such username" from "exists, but is a user account" turned the endpoint
// into a Telegram username-existence oracle, with each probe also spending the
// shared account's tightly-limited resolve budget.
const TELEGRAM_PUBLIC_ERROR_MESSAGES = {
400: 'Invalid Telegram username',
404: 'Public Telegram channel not found',
429: 'Telegram lookup is rate limited',
502: 'Telegram lookup failed',
503: 'Telegram lookup is temporarily unavailable',
504: 'Telegram lookup timed out',
};
function getTelegramPublicErrorMessage(statusCode) {
return TELEGRAM_PUBLIC_ERROR_MESSAGES[statusCode] || 'Telegram lookup failed';
}
function createTelegramStatusError(message, statusCode, retryAfterMs = 0) {
const error = new Error(message);
error.statusCode = statusCode;
if (retryAfterMs > 0) error.retryAfterMs = retryAfterMs;
return error;
}
function getTelegramErrorHeaders(error) {
const headers = { 'Content-Type': 'application/json', 'Cache-Control': 'no-store' };
if (Number(error?.retryAfterMs) > 0) {
headers['Retry-After'] = String(Math.max(1, Math.ceil(error.retryAfterMs / 1000)));
}
return headers;
}
function getCachedTelegramNegative(key) {
const cached = getCachedTelegramValue(telegramNegativeCache, key);
if (!cached) return null;
const retryAfterMs = cached.retryAfterUntil
? Math.max(1, cached.retryAfterUntil - Date.now())
: 0;
return createTelegramStatusError(cached.message, cached.statusCode, retryAfterMs);
}
function setCachedTelegramNegative(key, error) {
const statusCode = getTelegramErrorStatus(error);
// 504 is included so a chronically slow channel backs itself off instead of
// being retried on every 60s panel refresh. 503 stays out: it means "client
// reset / starting up", which is transient relay state, not a bad username.
if (![400, 404, 429, 504].includes(statusCode)) return;
const retryAfterMs = Number(error?.retryAfterMs) || 0;
const ttlMs = statusCode === 429 && retryAfterMs > 0
? retryAfterMs
: TELEGRAM_NEGATIVE_CACHE_TTL_MS;
setCachedTelegramValue(telegramNegativeCache, key, {
message: String(error?.message || error || 'Telegram lookup failed'),
statusCode,
retryAfterUntil: retryAfterMs > 0 ? Date.now() + retryAfterMs : 0,
}, ttlMs);
}
// Derive the flood wait ONLY from structured error fields.
//
// Never parse error.message: GramJS builds its lookup errors by interpolating
// the caller-supplied username verbatim — `No user has "${username}" as
// username` (telegram/client/users.js) — and the username grammar allows
// underscores and digits. A watchlist entry of `flood_wait_99999999` therefore
// used to be read back as a flood duration, and because telegramFloodWaitUntil
// is process-wide and only ever raised, one request could park the whole
// Telegram subsystem (curated poll included) effectively forever.
//
// `seconds` is set by GramJS on FloodWaitError; `errorMessage` is the MTProto
// error CODE (e.g. 'FLOOD_WAIT_42'), never free text, so an anchored match on
// it cannot be spoofed. The cap is the backstop: no single upstream value may
// wedge the process for longer than an operator would tolerate unattended.
function parseTelegramFloodWaitMs(error) {
const seconds = Number(error?.seconds);
let waitMs = Number.isFinite(seconds) && seconds > 0 ? seconds * 1000 : 0;
if (waitMs <= 0) {
const code = typeof error?.errorMessage === 'string' ? error.errorMessage : '';
const match = code.match(/^FLOOD_WAIT_(\d+)$/);
if (match) {
const parsed = Number(match[1]);
if (Number.isFinite(parsed) && parsed > 0) waitMs = parsed * 1000;
}
}
if (!Number.isFinite(waitMs) || waitMs <= 0) return 0;
return Math.min(waitMs, TELEGRAM_MAX_FLOOD_WAIT_MS);
}
function getTelegramFloodWaitRemainingMs(now = Date.now()) {
return Math.max(0, telegramFloodWaitUntil - now);
}
function assertTelegramRpcAllowed() {
const retryAfterMs = getTelegramFloodWaitRemainingMs();
if (retryAfterMs > 0) {
throw createTelegramStatusError('Telegram FLOOD_WAIT cooldown active', 429, retryAfterMs);
}
}
function normalizeTelegramRpcError(error) {
const retryAfterMs = parseTelegramFloodWaitMs(error);
if (retryAfterMs <= 0) return error;
telegramFloodWaitUntil = Math.max(telegramFloodWaitUntil, Date.now() + retryAfterMs);
return createTelegramStatusError(String(error?.message || error), 429, retryAfterMs);
}
function rejectTelegramRpcQueue(error) {
while (telegramRpcQueue.length) {
const entry = telegramRpcQueue.shift();
clearTimeout(entry.waitTimer);
entry.reject(error);
}
}
function drainTelegramRpcQueue() {
if (telegramRpcDrainTimer) return;
const cooldownMs = getTelegramFloodWaitRemainingMs();
if (cooldownMs > 0) {
const error = createTelegramStatusError('Telegram FLOOD_WAIT cooldown active', 429, cooldownMs);
rejectTelegramRpcQueue(error);
return;
}
while (telegramRpcActiveEntries.size < TELEGRAM_RPC_MAX_CONCURRENCY && telegramRpcQueue.length) {
const delayMs = Math.max(0, telegramRpcNextStartAt - Date.now());
if (delayMs > 0) {
telegramRpcDrainTimer = setTimeout(() => {
telegramRpcDrainTimer = null;
drainTelegramRpcQueue();
}, delayMs);
telegramRpcDrainTimer.unref?.();
return;
}
const entry = telegramRpcQueue.shift();
clearTimeout(entry.waitTimer);
telegramRpcActiveEntries.add(entry);
telegramRpcNextStartAt = Date.now() + TELEGRAM_RPC_MIN_INTERVAL_MS;
const operation = Promise.resolve().then(entry.operation);
let completed = false;
let released = false;
let timeout = null;
const release = () => {
if (released) return;
released = true;
telegramRpcActiveEntries.delete(entry);
drainTelegramRpcQueue();
};
const settle = (callback, value) => {
if (completed) return;
completed = true;
if (timeout) clearTimeout(timeout);
callback(value);
release();
};
entry.cancel = error => settle(entry.reject, error);
const entryTimeoutMs = entry.timeoutMs || TELEGRAM_CHANNEL_TIMEOUT_MS;
timeout = setTimeout(() => {
// One slow channel is not a dead socket. Tearing the client down here
// meant an arbitrary user-supplied channel could clear both lookup
// caches, reject every queued RPC and abort the curated poll mid-cycle,
// every 60s. Fail just this entry; only a RUN of timeouts (which implies
// the transport, not the channel) justifies resetting shared state.
const timeoutError = createTelegramStatusError(
`TIMEOUT after ${entryTimeoutMs}ms: ${entry.label}`,
504,
);
telegramConsecutiveRpcTimeouts++;
if (telegramConsecutiveRpcTimeouts >= TELEGRAM_MAX_CONSECUTIVE_RPC_TIMEOUTS) {
console.warn(`[Relay] Telegram RPC timed out ${telegramConsecutiveRpcTimeouts}x consecutively — resetting client`);
destroyTelegramClient(timeoutError, entry);
return;
}
settle(entry.reject, timeoutError);
}, entryTimeoutMs);
timeout.unref?.();
operation.then(
value => {
telegramConsecutiveRpcTimeouts = 0;
settle(entry.resolve, value);
},
error => settle(entry.reject, normalizeTelegramRpcError(error)),
);
}
}
// `priority` reserves the curated poll's place ahead of on-demand user lookups.
// Both share one MTProto session and one global start interval, so without a
// lane the product-managed feed competes on equal FIFO terms with arbitrary
// user traffic and gets pushed past its own queue-wait timeout.
function runTelegramRpc(label, operation, { priority = false, timeoutMs = 0 } = {}) {
assertTelegramRpcAllowed();
if (!priority && telegramRpcQueue.length >= TELEGRAM_RPC_MAX_QUEUE) {
throw createTelegramStatusError('Telegram RPC queue is full', 429, TELEGRAM_RPC_MIN_INTERVAL_MS);
}
return new Promise((resolve, reject) => {
const entry = { label, operation, resolve, reject, waitTimer: null, priority, timeoutMs };
if (priority) {
// No wait timer: the poll must not be evicted from its own queue by
// user-driven congestion. Its work is bounded by the cycle timeout.
const firstNonPriority = telegramRpcQueue.findIndex(queued => !queued.priority);
if (firstNonPriority < 0) telegramRpcQueue.push(entry);
else telegramRpcQueue.splice(firstNonPriority, 0, entry);
} else {
entry.waitTimer = setTimeout(() => {
const index = telegramRpcQueue.indexOf(entry);
if (index < 0) return;
telegramRpcQueue.splice(index, 1);
reject(createTelegramStatusError('Telegram RPC queue wait timed out', 429, TELEGRAM_RPC_MIN_INTERVAL_MS));
}, TELEGRAM_RPC_QUEUE_TIMEOUT_MS);
entry.waitTimer.unref?.();
telegramRpcQueue.push(entry);
}
drainTelegramRpcQueue();
});
}
async function withTelegramLookupSingleFlight(key, operation) {
const cachedError = getCachedTelegramNegative(key);
if (cachedError) throw cachedError;
const inflight = telegramLookupInflight.get(key);
if (inflight) return inflight;
const request = Promise.resolve()
.then(operation)
.catch(error => {
setCachedTelegramNegative(key, error);
throw error;
});
telegramLookupInflight.set(key, request);
try {
return await request;
} finally {
if (telegramLookupInflight.get(key) === request) telegramLookupInflight.delete(key);
}
}
function buildTelegramChannelPreview(entity, fallbackUsername, memberCount = null) {
const username = sanitizeTelegramUsername(entity?.username || fallbackUsername);
const title = String(entity?.title || entity?.firstName || username);
return {
username,
title,
memberCount: Number.isFinite(memberCount) ? memberCount : null,
url: `https://t.me/${username}`,
};
}
function assertTelegramConnectionCurrent(connection) {
if (telegramState.client !== connection.client || telegramState.api !== connection.Api) {
throw createTelegramStatusError('Telegram client reset during request', 503);
}
}
async function resolveTelegramChannelWithConnection(normalized, connection) {
assertTelegramConnectionCurrent(connection);
const cached = getCachedTelegramValue(telegramResolveCache, normalized);
if (cached) return cached;
return withTelegramLookupSingleFlight(`resolve:${normalized}`, async () => {
assertTelegramConnectionCurrent(connection);
const freshCached = getCachedTelegramValue(telegramResolveCache, normalized);
if (freshCached) return freshCached;
const entity = await runTelegramRpc(
`getEntity(${normalized})`,
() => connection.client.getEntity(normalized),
);
const TelegramChannel = connection.Api?.Channel;
if (!TelegramChannel || !(entity instanceof TelegramChannel) || !entity.username) {
// 404, not 400: a distinct status here told the caller that a username
// exists but belongs to a user/basic group, which is a username-existence
// oracle over Telegram paid for out of our own resolve budget.
throw createTelegramStatusError('Only public Telegram channels are supported', 404);
}
let memberCount = null;
if (connection.Api?.channels?.GetFullChannel) {
try {
const full = await runTelegramRpc(
`getFullChannel(${normalized})`,
() => connection.client.invoke(new connection.Api.channels.GetFullChannel({ channel: entity })),
);
const ChannelFull = connection.Api?.ChannelFull;
if (ChannelFull && full?.fullChat instanceof ChannelFull) {
memberCount = full?.fullChat?.participantsCount ?? null;
}
} catch (error) {
console.warn('[Relay] Telegram resolve participants count failed:', error?.message || error);
}
}
assertTelegramConnectionCurrent(connection);
const value = {
preview: buildTelegramChannelPreview(entity, normalized, memberCount),
entity,
};
setCachedTelegramValue(telegramResolveCache, normalized, value, TELEGRAM_RESOLVE_CACHE_TTL_MS);
return value;
});
}
async function resolveTelegramChannel(username) {
const normalized = sanitizeTelegramUsername(username);
const cached = getCachedTelegramValue(telegramResolveCache, normalized);
if (cached) return cached;
const cachedError = getCachedTelegramNegative(`resolve:${normalized}`);
if (cachedError) throw cachedError;
const connection = await initTelegramClientIfNeeded();
return resolveTelegramChannelWithConnection(normalized, connection);
}
async function fetchTelegramChannelFeed(username, limit = 20) {
const normalized = sanitizeTelegramUsername(username);
const safeLimit = Math.max(1, Math.min(50, Number(limit) || 20));
const cacheKey = `${normalized}:${safeLimit}`;
const cached = getCachedTelegramValue(telegramChannelCache, cacheKey);
if (cached) return cached;
return withTelegramLookupSingleFlight(`channel:${cacheKey}`, async () => {
const freshCached = getCachedTelegramValue(telegramChannelCache, cacheKey);
if (freshCached) return freshCached;
const connection = await initTelegramClientIfNeeded();
const { preview, entity } = await resolveTelegramChannelWithConnection(normalized, connection);
assertTelegramConnectionCurrent(connection);
const msgs = await runTelegramRpc(
`getMessages(${normalized})`,
() => connection.client.getMessages(entity, { limit: safeLimit }),
);
assertTelegramConnectionCurrent(connection);
const items = [];
const channel = { handle: preview.username, label: preview.title, topic: 'osint', region: 'watchlist' };
for (const msg of msgs || []) {
if (!msg || !msg.id || !msg.message) continue;
items.push(normalizeTelegramMessage(msg, channel));
}
items.sort((a, b) => (b.ts || '').localeCompare(a.ts || ''));
const value = {
source: 'telegram',
earlySignal: true,
enabled: TELEGRAM_ENABLED,
count: items.length,
updatedAt: new Date().toISOString(),
items,
};
setCachedTelegramValue(telegramChannelCache, cacheKey, value, TELEGRAM_CHANNEL_CACHE_TTL_MS);
return value;
});
}
let telegramPermanentlyDisabled = false;
function destroyTelegramClient(
activeError = createTelegramStatusError('Telegram client not active', 503),
errorEntry = null,
) {
const client = telegramState.client;
telegramState.client = null;
telegramState.api = null;
telegramResolveCache.clear();
telegramChannelCache.clear();
telegramLookupInflight.clear();
if (telegramRpcDrainTimer) clearTimeout(telegramRpcDrainTimer);
telegramRpcDrainTimer = null;
telegramRpcNextStartAt = 0;
telegramConsecutiveRpcTimeouts = 0;
const resetError = createTelegramStatusError('Telegram client reset after RPC timeout', 503);
rejectTelegramRpcQueue(resetError);
for (const entry of [...telegramRpcActiveEntries]) {
entry.cancel(entry === errorEntry ? activeError : resetError);
}
if (!client) return telegramDisconnectPromise;
try {
// Must be bounded: initTelegramClientIfNeeded awaits this promise before
// any other work, so a gramjs disconnect() that never settles would wedge
// every future lookup AND the poll loop permanently (and leak one pending
// poll per cycle once guardedTelegramPoll force-clears its in-flight flag).
// The shutdown path at the bottom of this file already races disconnect
// against a timer for the same reason; the socket is force-destroyed just
// below regardless, so abandoning a hung disconnect is safe.
telegramDisconnectPromise = withTimeout(
Promise.resolve(client.disconnect()),
TELEGRAM_DISCONNECT_TIMEOUT_MS,
'disconnectTelegramClient',
);
} catch (error) {
telegramDisconnectPromise = Promise.reject(error);
}
telegramDisconnectPromise.catch(() => {});
try {
if (client._sender) {
client._sender._reconnecting = false;
client._sender._autoReconnect = false;
if (client._sender._connection) {
try { client._sender._connection.socket?.destroy?.(); } catch {}
try { client._sender._connection.close?.(); } catch {}
}
}
} catch {}
return telegramDisconnectPromise;
}
async function initTelegramClientIfNeeded() {
if (!TELEGRAM_ENABLED) {
throw createTelegramStatusError('Telegram relay not configured', 503);
}
assertTelegramRpcAllowed();
if (telegramState.client && telegramState.api) {
return { client: telegramState.client, Api: telegramState.api };
}
if (telegramPermanentlyDisabled) {
throw createTelegramStatusError(telegramState.lastError || 'Telegram relay not active', 503);
}
if (telegramDisconnectPromise) {
const pendingDisconnect = telegramDisconnectPromise;
try {
await pendingDisconnect;
} catch (error) {
telegramState.lastError = `telegram disconnect failed: ${error?.message || error}`;
console.warn('[Relay] Telegram disconnect failed after forced socket teardown:', telegramState.lastError);
} finally {
// Clear in `finally` so a rejected/abandoned disconnect can never latch
// the variable and block every subsequent reconnect.
if (telegramDisconnectPromise === pendingDisconnect) telegramDisconnectPromise = null;
}
}
const retryAfterMs = (telegramState.startedAt + TELEGRAM_STARTUP_DELAY_MS) - Date.now();
if (retryAfterMs > 0) {
throw createTelegramStatusError('Telegram client startup delay active', 503, retryAfterMs);
}
if (telegramState.initPromise) return telegramState.initPromise;
const initPromise = connectTelegramClient();
telegramState.initPromise = initPromise;
try {
return await initPromise;
} finally {
if (telegramState.initPromise === initPromise) telegramState.initPromise = null;
}
}
async function connectTelegramClient() {
const apiId = parseInt(String(process.env.TELEGRAM_API_ID || ''), 10);
const apiHash = String(process.env.TELEGRAM_API_HASH || '');
const sessionStr = String(process.env.TELEGRAM_SESSION || '');
if (!apiId || !apiHash || !sessionStr) {
throw createTelegramStatusError('Telegram relay not configured', 503);
}
let client;
try {
let Api;
if (RELAY_TEST_MODE && process.env.RELAY_TEST_TELEGRAM === 'true') {
class TestTelegramChannel {
constructor(username) {
this.username = username;
this.title = `Test ${username}`;
}
}
class TestTelegramChannelFull {
constructor(participantsCount) {
this.participantsCount = participantsCount;
}
}
class TestGetFullChannel {
constructor({ channel }) {
this.channel = channel;
}
}
const rpcDelay = async () => {
const delayMs = readTelegramNumberEnv('RELAY_TEST_TELEGRAM_RPC_DELAY_MS', 0);
if (delayMs > 0) await new Promise(resolve => setTimeout(resolve, delayMs));
};
const pendingLookupRejects = new Set();
const maybeFailLookup = username => {
const floodUsername = String(process.env.RELAY_TEST_TELEGRAM_FLOOD_USERNAME || '').toLowerCase();
if (floodUsername && username === floodUsername) {
const seconds = Math.max(1, Number(process.env.RELAY_TEST_TELEGRAM_FLOOD_SECONDS || 3));
// Shaped like a real gramjs FloodWaitError: the duration lives in the
// structured `seconds` / `errorMessage` fields, NOT in free-text
// message. A fake that only set `message` meant every flood test
// exercised a parse branch production never takes.
const error = new Error(`A wait of ${seconds} seconds is required (caused by ResolveUsername)`);
error.seconds = seconds;
error.errorMessage = `FLOOD_WAIT_${seconds}`;
throw error;
}
const invalid = String(process.env.RELAY_TEST_TELEGRAM_INVALID_USERNAMES || '')
.split(',')
.map(value => value.trim().toLowerCase())
.filter(Boolean);
// Real gramjs interpolates the requested username into this message.
// Reproducing that verbatim is what lets a test prove a username like
// `flood_wait_99999999` cannot be read back as a flood duration.
if (invalid.includes(username)) throw new Error(`No user has "${username}" as username`);
};
Api = {
Channel: TestTelegramChannel,
ChannelFull: TestTelegramChannelFull,
...(process.env.RELAY_TEST_TELEGRAM_SKIP_FULL_CHANNEL === 'true'
? {}
: { channels: { GetFullChannel: TestGetFullChannel } }),
};
client = {
connect: async () => {
telegramTestConnectAttempts++;
console.log(`[Relay][TestTelegram] connect ${telegramTestConnectAttempts}`);
const hangAttempts = Math.max(0, Number(process.env.RELAY_TEST_TELEGRAM_CONNECT_HANG_ATTEMPTS || 0));
if (
process.env.RELAY_TEST_TELEGRAM_CONNECT_NEVER === 'true'
|| telegramTestConnectAttempts <= hangAttempts
) {
await new Promise(() => {});
}
const delayMs = Math.max(0, Number(process.env.RELAY_TEST_TELEGRAM_CONNECT_DELAY_MS || 0));
if (delayMs > 0) await new Promise(resolve => setTimeout(resolve, delayMs));
},
disconnect: async () => {
for (const reject of pendingLookupRejects) reject(new Error('TEST_TELEGRAM_DISCONNECTED'));
pendingLookupRejects.clear();
// Models a gramjs disconnect that never settles, which used to wedge
// every later reconnect on an unbounded await.
if (process.env.RELAY_TEST_TELEGRAM_DISCONNECT_NEVER === 'true') {
await new Promise(() => {});
}
const delayMs = readTelegramNumberEnv('RELAY_TEST_TELEGRAM_DISCONNECT_DELAY_MS', 0);
if (delayMs > 0) await new Promise(resolve => setTimeout(resolve, delayMs));
if (process.env.RELAY_TEST_TELEGRAM_DISCONNECT_REJECT === 'true') {
throw new Error('TEST_TELEGRAM_DISCONNECT_FAILED');
}
},
getEntity: async username => {
console.log(`[Relay][TestTelegram] getEntity ${username}`);
maybeFailLookup(username);
const never = String(process.env.RELAY_TEST_TELEGRAM_RPC_NEVER_USERNAMES || '')
.split(',')
.map(value => value.trim().toLowerCase())
.filter(Boolean);
if (never.includes(username)) {
return new Promise((_, reject) => pendingLookupRejects.add(reject));
}
await rpcDelay();
const delayedRejects = String(process.env.RELAY_TEST_TELEGRAM_RPC_REJECT_USERNAMES || '')
.split(',')
.map(value => value.trim().toLowerCase())
.filter(Boolean);
if (delayedRejects.includes('*') || delayedRejects.includes(String(username).toLowerCase())) {
throw new Error('TEST_TELEGRAM_RPC_FAILED');
}
const csvEnv = name => String(process.env[name] || '')
.split(',')
.map(value => value.trim().toLowerCase())
.filter(Boolean);
// Without these two hooks the fake could only ever produce a public
// channel, so the `instanceof Api.Channel` / `!entity.username` gate
// — the only thing rejecting user accounts and private channels —
// was unreachable from the suite and could be deleted while green.
if (csvEnv('RELAY_TEST_TELEGRAM_USER_USERNAMES').includes(username)) {
return { className: 'User', username, firstName: `Test ${username}` };
}
if (csvEnv('RELAY_TEST_TELEGRAM_PRIVATE_USERNAMES').includes(username)) {
const channel = new TestTelegramChannel(username);
channel.username = undefined;
return channel;
}
return new TestTelegramChannel(username);
},
invoke: async request => {
console.log(`[Relay][TestTelegram] getFullChannel ${request.channel.username}`);
const never = String(process.env.RELAY_TEST_TELEGRAM_FULL_NEVER_USERNAMES || '')
.split(',')
.map(value => value.trim().toLowerCase())
.filter(Boolean);
if (never.includes(request.channel.username)) {
return new Promise((_, reject) => pendingLookupRejects.add(reject));
}
await rpcDelay();
return {
fullChat: new TestTelegramChannelFull(
Math.max(0, Number(process.env.RELAY_TEST_TELEGRAM_MEMBER_COUNT || 1234)),
),
};
},
getMessages: async (entity, { limit }) => {
console.log(`[Relay][TestTelegram] getMessages ${limit} ${entity.username}`);
await rpcDelay();
const messages = JSON.parse(process.env.RELAY_TEST_TELEGRAM_MESSAGES || '[]');
return Array.isArray(messages) ? messages.slice(0, limit) : [];
},
};
} else {
const telegram = await import('telegram');
const sessions = await import('telegram/sessions/index.js');
Api = telegram.Api;
client = new telegram.TelegramClient(new sessions.StringSession(sessionStr), apiId, apiHash, {
connectionRetries: 3,
// gramjs defaults this to 60 and SWALLOWS any flood shorter than the
// threshold by sleeping inside _call (telegram/client/users.js) before
// retrying. That sleep outlives TELEGRAM_CHANNEL_TIMEOUT_MS, so a
// routine sub-60s flood surfaced to us as an RPC timeout — bypassing
// the cooldown machinery below entirely and resetting the client
// instead. 0 makes every flood propagate immediately, which is what
// normalizeTelegramRpcError and telegramFloodWaitUntil expect.
floodSleepThreshold: 0,
});
}
await withTimeout(client.connect(), TELEGRAM_CONNECT_TIMEOUT_MS, 'connectTelegramClient');
telegramState.client = client;
telegramState.api = Api;
telegramState.lastError = null;
console.log('[Relay] Telegram client connected');
return { client, Api };
} catch (e) {
const em = e?.message || String(e);
if (e?.code === 'ERR_MODULE_NOT_FOUND' || /Cannot find package|Directory import/.test(em)) {
telegramPermanentlyDisabled = true;
telegramState.lastError = 'telegram package not installed';
console.warn('[Relay] Telegram package not installed — disabling permanently for this session');
throw createTelegramStatusError(telegramState.lastError, 503);
}
// Destroy the locally-created client directly — telegramState.client
// is still null because connect() failed before the assignment. Without
// this, the MTProto sender's autonomous reconnect loop keeps running.
if (client) {
telegramState.client = client;
destroyTelegramClient();
}
if (/AUTH_KEY_DUPLICATED/.test(em)) {
telegramPermanentlyDisabled = true;
telegramState.lastError = 'session invalidated (AUTH_KEY_DUPLICATED) — generate a new TELEGRAM_SESSION';
console.error('[Relay] Telegram session permanently invalidated (AUTH_KEY_DUPLICATED). Generate a new session with: node scripts/telegram/session-auth.mjs');
throw createTelegramStatusError(telegramState.lastError, 503);
}
telegramState.lastError = `telegram init failed: ${em}`;
console.warn('[Relay] Telegram init failed:', telegramState.lastError);
throw e instanceof Error ? e : new Error(em);
}
}
const TELEGRAM_CHANNEL_TIMEOUT_MS = readTelegramNumberEnv('TELEGRAM_CHANNEL_TIMEOUT_MS', 15_000, 10);
const TELEGRAM_POLL_CYCLE_TIMEOUT_MS = 180_000; // 3min max for entire poll cycle
const TELEGRAM_POLL_STUCK_AFTER_MS = RELAY_TEST_MODE
? readTelegramNumberEnv(
'RELAY_TEST_TELEGRAM_POLL_STUCK_AFTER_MS',
TELEGRAM_POLL_CYCLE_TIMEOUT_MS + 30_000,
10,
)
: TELEGRAM_POLL_CYCLE_TIMEOUT_MS + 30_000;
function withTimeout(promise, ms, label) {
return new Promise((resolve, reject) => {
const timer = setTimeout(() => reject(new Error(`TIMEOUT after ${ms}ms: ${label}`)), ms);
promise.then(
v => { clearTimeout(timer); resolve(v); },
e => { clearTimeout(timer); reject(e); }
);
});
}
async function pollTelegramOnce() {
let connection;
try {
connection = await initTelegramClientIfNeeded();
} catch (error) {
if (!telegramState.lastError) telegramState.lastError = error?.message || String(error);
return;
}
const channels = telegramState.channels.length ? telegramState.channels : loadTelegramChannels();
if (!channels.length) return;
const client = connection.client;
const newItems = [];
const pollStart = Date.now();
let channelsPolled = 0;
let channelsFailed = 0;
let mediaSkipped = 0;
for (const channel of channels) {
if (telegramState.client !== client || telegramState.api !== connection.Api) break;
if (Date.now() - pollStart > TELEGRAM_POLL_CYCLE_TIMEOUT_MS) {
console.warn(`[Relay] Telegram poll cycle timeout (${Math.round(TELEGRAM_POLL_CYCLE_TIMEOUT_MS / 1000)}s), polled ${channelsPolled}/${channels.length} channels`);
break;
}
const handle = channel.handle;
const minId = telegramState.cursorByHandle[handle] || 0;
try {
// One queue entry per CHANNEL, not per RPC. The global start interval is
// applied per dequeue, so enqueueing getEntity and getMessages separately
// doubled the pacing cost of a cycle (56 channels x 2 x 800ms = ~90s,
// against a 60s poll interval and a 180s cycle timeout) versus the single
// per-channel sleep this replaced. Priority keeps user lookups from
// interleaving into the middle of the poll's own pacing budget.
const msgs = await runTelegramRpc(
`poll(${handle})`,
async () => {
const entity = await client.getEntity(handle);
assertTelegramConnectionCurrent(connection);
return client.getMessages(entity, {
limit: Math.max(1, Math.min(50, channel.maxMessages || 25)),
minId,
});
},
{ priority: true, timeoutMs: TELEGRAM_CHANNEL_TIMEOUT_MS * 2 },
);
assertTelegramConnectionCurrent(connection);
for (const msg of msgs) {
if (!msg || !msg.id) continue;
if (!msg.message) { mediaSkipped++; continue; }
const item = normalizeTelegramMessage(msg, channel);
newItems.push(item);
// The 200-item feed buffer turns over in ~90 minutes; strike candidates
// are captured here so none age out before the 10-minute extractor runs.
noteMiddleEastStrikeItem(item);
if (!telegramState.cursorByHandle[handle] || msg.id > telegramState.cursorByHandle[handle]) {
telegramState.cursorByHandle[handle] = msg.id;
}
}
channelsPolled++;
} catch (e) {
const em = e?.message || String(e);
channelsFailed++;
telegramState.lastError = `poll ${handle} failed: ${em}`;
console.warn('[Relay] Telegram poll error:', telegramState.lastError);
if (/AUTH_KEY_DUPLICATED/.test(em)) {
telegramPermanentlyDisabled = true;
telegramState.lastError = 'session invalidated (AUTH_KEY_DUPLICATED) — generate a new TELEGRAM_SESSION';
console.error('[Relay] Telegram session permanently invalidated (AUTH_KEY_DUPLICATED). Generate a new session with: node scripts/telegram/session-auth.mjs');
destroyTelegramClient();
break;
}
if (/FLOOD_WAIT/.test(em)) {
const wait = parseInt(em.match(/(\d+)/)?.[1] || '60', 10);
console.warn(`[Relay] Telegram FLOOD_WAIT ${wait}s — stopping poll cycle early`);
break;
}
}
// Rest AFTER the channel, not merely between RPC dispatches. The queue
// paces start-to-start, so leaning on it alone would make this loop's
// effective spacing `max(interval, pair latency)` instead of the
// `pair latency + interval` that main ships — roughly 1.75x the request
// rate against a shared account whose FLOOD_WAIT takes the curated feed
// down with it. Outside the try so a failing channel rests too.
await new Promise(resolve => setTimeout(resolve, TELEGRAM_RPC_MIN_INTERVAL_MS));
}
if (newItems.length) {
const seen = new Set();
telegramState.items = [...newItems, ...telegramState.items]
.filter(item => {
if (seen.has(item.id)) return false;
seen.add(item.id);
return true;
})
.sort((a, b) => (b.ts || '').localeCompare(a.ts || ''))
.slice(0, TELEGRAM_MAX_FEED_ITEMS);
}
telegramState.lastPollAt = Date.now();
// lastPollAt advances on every cycle, including one that a FLOOD_WAIT or a
// dead session ended with zero channels read. Freshness consumers (the
// Middle East strike feed's seed-meta gate) need the last cycle that read data.
if (channelsPolled > 0) telegramState.lastSuccessfulPollAt = telegramState.lastPollAt;
const elapsed = ((Date.now() - pollStart) / 1000).toFixed(1);
console.log(`[Relay] Telegram poll: ${channelsPolled}/${channels.length} channels, ${newItems.length} new msgs, ${telegramState.items.length} total, ${channelsFailed} errors, ${mediaSkipped} media-only skipped (${elapsed}s)`);
if (channelsPolled > 0) {
const rc = telegramState.items.length;
upstashSet(TELEGRAM_ITEMS_SNAPSHOT_KEY, {
savedAt: new Date(telegramState.lastPollAt).toISOString(),
items: telegramState.items,
}, TELEGRAM_ITEMS_SNAPSHOT_TTL_SECONDS).catch(() => {});
// Data key TTL must outlive maxStaleMin (10 min = 600s) by enough
// buffer so health sees hasData=true + stale seed-meta → STALE_SEED.
// If both keys expire together, health jumps straight to EMPTY and
// the stale window is never visible. 1800s (30 min) data vs 900s
// (15 min) meta gives a 15-min STALE_SEED window before EMPTY.
upstashSet('intelligence:telegram-feed:v1', {
count: rc,
updatedAt: new Date().toISOString(),
enabled: true,
}, 1800).catch(() => {});
upstashSet('seed-meta:intelligence:telegram-feed:v1', {
fetchedAt: Date.now(),
recordCount: rc,
}, 900).catch(() => {});
}
}
let telegramPollRun = null;
function guardedTelegramPoll() {
if (telegramPollRun) {
const stuck = Date.now() - telegramPollRun.startedAt;
if (stuck > TELEGRAM_POLL_STUCK_AFTER_MS && !telegramPollRun.warned) {
console.warn(`[Relay] Telegram poll stuck for ${Math.round(stuck / 1000)}s, waiting for the current poll to settle`);
telegramPollRun.warned = true;
}
return false;
}
const run = { startedAt: Date.now(), warned: false };
telegramPollRun = run;
pollTelegramOnce()
.catch(e => console.warn('[Relay] Telegram poll error:', e?.message || e))
.finally(() => {
if (telegramPollRun === run) telegramPollRun = null;
});
}
async function restoreTelegramSnapshot() {
const snapshot = await upstashGet(TELEGRAM_ITEMS_SNAPSHOT_KEY);
const savedAt = Date.parse(snapshot?.savedAt);
// Keep only feed-item shapes: the topic/channel filters and the poll merge
// read fields on every entry, and a bad one would sit there until TTL.
const items = Array.isArray(snapshot?.items)
? snapshot.items.filter(item => item && typeof item === 'object' && typeof item.id === 'string' && item.id)
: [];
if (!items.length || !Number.isFinite(savedAt)) return;
// A poll that finished first holds fresher items; never overwrite it.
if (telegramState.lastPollAt || telegramState.items.length) return;
telegramState.items = items.slice(0, TELEGRAM_MAX_FEED_ITEMS);
// updatedAt reports when these posts were polled. lastSuccessfulPollAt stays
// 0: it gates the strike feed's freshness and no poll has run here yet.
telegramState.lastPollAt = savedAt;
console.log(`[Relay] Telegram feed restored ${telegramState.items.length} items from snapshot saved ${snapshot.savedAt}`);
}
function startTelegramPollLoop() {
if (!TELEGRAM_ENABLED) return;
loadTelegramChannels();
restoreTelegramSnapshot().catch((e) => console.warn('[Relay] Telegram snapshot restore failed:', e?.message || e));
if (TELEGRAM_STARTUP_DELAY_MS > 0) {
console.log(`[Relay] Telegram connect delayed ${TELEGRAM_STARTUP_DELAY_MS}ms (waiting for old container to disconnect)`);
setTimeout(() => {
guardedTelegramPoll();
setInterval(guardedTelegramPoll, TELEGRAM_POLL_INTERVAL_MS).unref?.();
console.log('[Relay] Telegram poll loop started');
}, TELEGRAM_STARTUP_DELAY_MS);
} else {
guardedTelegramPoll();
setInterval(guardedTelegramPoll, TELEGRAM_POLL_INTERVAL_MS).unref?.();
console.log('[Relay] Telegram poll loop started');
}
}
// ─────────────────────────────────────────────────────────────
// Curated X news-account monitoring (Track A / #6654)
// One public X List page in each fixed 15-minute UTC slot.
// Requires the app bearer and the verified public List ID.
// ─────────────────────────────────────────────────────────────
const X_BEARER_TOKEN = String(process.env.X_BEARER_TOKEN || '').trim();
const X_CURATED_LIST_ID = String(process.env.X_CURATED_LIST_ID || '').trim();
const X_CURATED_LIST_CONFIGURED = /^[1-9]\d{0,18}$/.test(X_CURATED_LIST_ID);
const X_ENABLED = Boolean(X_BEARER_TOKEN && X_CURATED_LIST_CONFIGURED);
const X_POLL_INTERVAL_MS = 15 * 60 * 1000;
// `Number('abc')` is NaN, and Math.max(50, NaN) is NaN — which reaches
// mergeAndDedup as `.slice(0, NaN)` and silently publishes an EMPTY feed every
// cycle with no error anywhere. Coerce non-numeric env values to the default.
const X_MAX_FEED_ITEMS = Math.max(50, Number(process.env.X_MAX_FEED_ITEMS) || xNewsAccounts.DEFAULT_MAX_FEED_ITEMS);
const X_MAX_TEXT_CHARS = Math.max(200, Number(process.env.X_MAX_TEXT_CHARS) || 800);
// Reviewed enabled-registry ceiling. Post spend stays one List page per slot
// (96 slots of 5 Posts) plus one 25-Post deletion audit, so member count does
// not add Post reads. Raise this only after re-checking that math. It is not
// the live List size: the List still has to match this registry exactly.
const X_TRACK_A_ACCOUNT_BUDGET = 65;
const X_FEED_CACHE_KEY = 'intelligence:x-feed:v1';
const X_FEED_META_KEY = 'seed-meta:intelligence:x-feed:v1';
const X_FEED_POLL_STATE_KEY = 'intelligence:x-feed:poll-state:v1';
const X_FEED_POLL_LOCK_KEY = 'intelligence:x-feed:poll-lock:v1';
const X_FEED_TTL_SECONDS = 5400;
const X_FEED_META_TTL_SECONDS = 3600;
// Two bounded 15-second X calls plus Redis work fit inside two minutes. A short
// lease lets another replica recover at the next quarter-hour boundary after an
// owner crash instead of losing two slots to the old cadence-sized lease.
const X_FEED_POLL_LOCK_TTL_SECONDS = 120;
// The process-local guard is checked on each scheduled boundary. If a request
// somehow outlives the normal timeout, the next boundary aborts that generation
// before it starts a replacement run.
const X_POLL_STUCK_AFTER_MS = 60_000;
const xPostBudget = createXPostBudget({
evalCommand: upstashEval,
...X_SHARED_COVERAGE_MODEL,
});
const xState = {
accounts: [],
lastMembershipCheckAt: 0,
items: [],
lookupOffset: 0,
// Persisted snapshot version, published to Redis and to /status. NOT the poll
// guard's run counter — see xPollGeneration below for why the two must stay
// apart.
generation: 0,
lastPollAt: 0,
lastHealthyAt: 0,
lastAttemptAt: 0,
lastProviderSuccessAt: 0,
lastAcceptedPublicationAt: 0,
lastAttemptSlot: null,
lastProviderSuccessSlot: null,
lastPublishedSlot: null,
lastCoverage: null,
lastError: null,
rateLimitedUntil: 0,
rateLimitAttempt: 0,
backoffCause: null,
lastDeletionAuditAt: 0,
lastCycleUsage: null,
postBudget: null,
// True when a Redis read failed, so last-good state is present but unreadable.
// Blocks polling/publishing until a clean read (see the cycle's hydrate()).
hydrationFailed: false,
startedAt: Date.now(),
};
// The poll guard's in-process run counter, deliberately NOT xState.generation.
// The guard stamps each run with this value and, in its `.finally`, only clears
// the in-flight flag while the stamp still matches. The cycle's hydrate() runs INSIDE
// a live poll (the lease-conflict and hydration-retry paths) and overwrites the
// persisted snapshot version from Redis — when both meanings shared one field
// that overwrite retired the run the guard was fencing on, so inFlight was never
// cleared, the next tick returned early, and a whole cycle was skipped until
// stuckAfterMs force-cleared it with a misleading "X poll stuck" warning.
let xPollGeneration = 0;
function loadXAccounts() {
const p = path.join(__dirname, '..', 'data', 'x-accounts.json');
try {
const raw = JSON.parse(readFileSync(p, 'utf8'));
const enabledTotal = xNewsAccounts.countEnabledAccounts(raw);
if (enabledTotal > X_TRACK_A_ACCOUNT_BUDGET) {
console.warn(`[Relay] X registry has ${enabledTotal} enabled accounts; Track A budget is ~${X_TRACK_A_ACCOUNT_BUDGET}. Re-run spend math before growing the set.`);
}
xState.accounts = xNewsAccounts.loadXAccounts(raw);
if (!xState.accounts.length) {
console.warn('[Relay] X account registry is empty — no accounts to poll');
}
return xState.accounts;
} catch (e) {
xState.accounts = [];
xState.lastError = `failed to load x-accounts.json: ${e?.message || String(e)}`;
return [];
}
}
// hydrate / publish / pollOnce live in scripts/lib/x-poll-cycle.cjs so a test can
// EXECUTE them. This file has no module.exports and no require.main guard, so
// importing it to reach those functions boots the whole relay — which is why the
// only coverage they ever had was regex-on-source, and why a generation-field
// collision, a lease handoff that dropped a peer's posts, and a stuck-abort
// threshold that outlived the Redis lease all shipped unnoticed. Built once here
// with the relay's real Redis helpers, constants and state object;
// tests/x-poll-cycle.test.mjs drives the same factory with stubs.
const xPollCycle = createXPollCycle({
xState,
xNewsAccounts,
xPostBudget,
loadXAccounts,
upstashGet,
upstashSetNx,
upstashPublishXIfLockOwner,
upstashReleaseLockIfOwner,
// The guard's run counter, never xState.generation — see the comment on
// `let xPollGeneration` above. Passed as an accessor so the cycle module
// cannot reach the module-level mutable itself.
getPollGeneration: () => xPollGeneration,
scheduleRetry: (retryAfterLeaseConflict) => guardedXPoll(retryAfterLeaseConflict),
randomId: () => crypto.randomBytes(4).toString('hex'),
X_ENABLED,
X_BEARER_TOKEN,
X_CURATED_LIST_ID,
X_POLL_INTERVAL_MS,
X_FEED_CACHE_KEY,
X_FEED_META_KEY,
X_FEED_POLL_STATE_KEY,
X_FEED_POLL_LOCK_KEY,
X_FEED_TTL_SECONDS,
X_FEED_META_TTL_SECONDS,
X_FEED_POLL_LOCK_TTL_SECONDS,
X_MAX_FEED_ITEMS,
X_MAX_TEXT_CHARS,
log: (message) => console.log(message),
warn: (message) => console.warn(message),
});
const xPollGuard = createPollGenerationGuard({
poll: (context) => xPollCycle.pollOnce(context),
getGeneration: () => xPollGeneration,
setGeneration: (generation) => { xPollGeneration = generation; },
stuckAfterMs: X_POLL_STUCK_AFTER_MS,
warn: (stuckMs, error) => {
if (error) {
console.warn('[Relay] X poll error:', error?.message || error);
} else {
console.warn(`[Relay] X poll stuck for ${Math.round(stuckMs / 1000)}s — force-clearing in-flight flag`);
}
},
});
function guardedXPoll(retryAfterLeaseConflict = false) {
return xPollGuard.run({ retryAfterLeaseConflict });
}
async function startXPollLoop() {
loadXAccounts();
await xPollCycle.hydrate();
if (!X_ENABLED) {
const missing = [
!X_BEARER_TOKEN ? 'X_BEARER_TOKEN' : null,
!X_CURATED_LIST_CONFIGURED ? 'valid X_CURATED_LIST_ID' : null,
].filter(Boolean);
xState.lastError = `X List poll disabled: missing ${missing.join(' and ')}`;
console.warn(`[Relay] ${xState.lastError}`);
return;
}
const slot = xPollSlot(Date.now(), X_POLL_INTERVAL_MS);
const nextDueAt = Math.max(
xState.lastAttemptSlot === slot.id ? slot.endsAt : Date.now(),
xState.rateLimitedUntil || 0,
);
const startupDelayMs = Math.max(0, nextDueAt - Date.now());
const pollAndScheduleNextSlot = () => {
const started = guardedXPoll();
const activeSlot = xPollSlot(Date.now(), X_POLL_INTERVAL_MS);
const delayMs = started ? Math.max(1000, activeSlot.endsAt - Date.now()) : 1000;
setTimeout(pollAndScheduleNextSlot, delayMs).unref?.();
};
if (startupDelayMs > 0) {
const timer = setTimeout(pollAndScheduleNextSlot, startupDelayMs);
timer.unref?.();
} else {
pollAndScheduleNextSlot();
}
console.log('[Relay] X List poll loop started (fixed 15-minute UTC slots)');
}
// ─────────────────────────────────────────────────────────────
// OREF Siren Alerts (Israel Home Front Command)
// Polls oref.org.il via HTTP CONNECT tunnel through residential proxy (Israel exit)
// ─────────────────────────────────────────────────────────────
function stripBom(text) {
return text.charCodeAt(0) === 0xFEFF ? text.slice(1) : text;
}
// The configured `user:pass` of the OREF proxy, so log lines can drop the
// exact secret rather than guess its shape with a pattern.
const OREF_PROXY_CREDENTIAL = (() => {
const raw = OREF_PROXY_AUTH.replace(/^[a-z][a-z0-9+.-]*:\/\//i, '');
const at = raw.lastIndexOf('@');
return at > 0 ? raw.slice(0, at) : '';
})();
// Strip proxy credentials from an error message. The literal credential goes
// first (it may contain `/`, `@` or whitespace, which no URL pattern can
// anticipate); the `//user:pass@` pattern then covers any credential that
// did not come from our own config.
function redactOrefError(msg, credential) {
let text = String(msg || '');
if (credential) text = text.split(credential).join('<redacted>');
return text.replace(/\/\/[^\s/]*@/g, '//<redacted>@');
}
// What `/oref/alerts` may say about a failed poll. The curl failure text is
// the whole command line, which embeds the proxy URL; no amount of regex
// redaction should be the only thing between that and an unauthenticated
// public route. The redacted detail goes to the server log, the public state
// gets a fixed summary.
const OREF_PUBLIC_POLL_ERROR = 'OREF upstream fetch failed';
function orefPollFailure(context, err, credential) {
const stderr = err?.stderr ? err.stderr.toString().trim() : '';
console.warn(`[Relay] ${context}:`, redactOrefError(stderr || err?.message || String(err), credential));
return OREF_PUBLIC_POLL_ERROR;
}
function orefDateToUTC(dateStr) {
if (!dateStr || !dateStr.includes(' ')) return new Date().toISOString();
const [datePart, timePart] = dateStr.split(' ');
const [y, m, d] = datePart.split('-').map(Number);
const [hh, mm, ss] = timePart.split(':').map(Number);
const fmt = new Intl.DateTimeFormat('en-US', {
timeZone: 'Asia/Jerusalem',
year: 'numeric', month: '2-digit', day: '2-digit',
hour: '2-digit', minute: '2-digit', second: '2-digit', hour12: false,
});
function partsAt(ms) {
const p = Object.fromEntries(fmt.formatToParts(new Date(ms)).map(x => [x.type, x.value]));
return `${p.year}-${p.month}-${p.day} ${p.hour}:${p.minute}:${p.second}`;
}
const base2 = Date.UTC(y, m - 1, d, hh - 2, mm, ss);
const base3 = Date.UTC(y, m - 1, d, hh - 3, mm, ss);
const candidates = [];
if (partsAt(base2) === dateStr) candidates.push(base2);
if (partsAt(base3) === dateStr) candidates.push(base3);
const ms = candidates.length ? Math.min(...candidates) : base2;
return new Date(ms).toISOString();
}
function orefCurlFetch(proxyAuth, url, { toFile } = {}) {
// Use curl via child_process — Node.js TLS fingerprint (JA3) gets blocked by Akamai,
// but curl's fingerprint passes. curl is available on Railway (Linux) and macOS.
// execFileSync avoids shell interpolation — safe with special chars in proxy credentials.
const { execFileSync } = require('child_process');
// Build an optional proxy args block. When no proxy is configured
// (proxyAuth empty), OMIT the -x flag entirely — passing `-x http://`
// (empty host) makes curl fail with "Unsupported proxy syntax".
let proxyArgs = [];
if (proxyAuth && proxyAuth.trim()) {
const proxyUrl = proxyAuth.includes('://') ? proxyAuth : `http://${proxyAuth}`;
proxyArgs = ['-x', proxyUrl];
}
const args = [
'-sS', '--compressed', ...proxyArgs, '--max-time', '15',
'-H', 'Accept: application/json',
'-H', 'Referer: https://www.oref.org.il/',
'-H', 'X-Requested-With: XMLHttpRequest',
];
if (toFile) {
// Write directly to disk — avoids stdout buffer overflow (ENOBUFS) for large responses
args.push('-o', toFile);
args.push(url);
execFileSync('curl', args, { timeout: 20000, stdio: ['pipe', 'pipe', 'pipe'] });
return require('fs').readFileSync(toFile, 'utf8');
}
args.push(url);
const result = execFileSync('curl', args, { encoding: 'utf8', timeout: 20000, stdio: ['pipe', 'pipe', 'pipe'] });
return result;
}
function categorizeOrefThreat(threat) {
const t = (threat || '').toLowerCase();
if (t.includes('missile') || t.includes('טיל') || t.includes('ballistic')) return 'MISSILE';
if (t.includes('rocket') || t.includes('רקט')) return 'ROCKET';
if (t.includes('drone') || t.includes('uav') || t.includes('כטב') || t.includes('hostile aircraft') || t.includes('כלי טיס')) return 'DRONE';
if (t.includes('mortar')) return 'MORTAR';
if (t.includes('infiltration') || t.includes('חדיר') || t.includes('מחבל')) return 'INFILTRATION';
if (t.includes('earthquake') || t.includes('רעידת')) return 'EARTHQUAKE';
if (t.includes('tsunami') || t.includes('צונמי')) return 'TSUNAMI';
if (t.includes('chemical') || t.includes('hazmat') || t.includes('חומרים מסוכנים') || t.includes('רדיולוגי')) return 'HAZMAT';
return 'ALERT';
}
async function tzevaAdomFetchAlerts() {
try {
const resp = await fetch(TZEVA_ADOM_URL, {
headers: { 'User-Agent': 'WorldMonitor/1.0', Accept: 'application/json' },
signal: AbortSignal.timeout(12_000),
});
if (!resp.ok) return null;
const data = await resp.json();
if (!Array.isArray(data) || data.length === 0) return [];
return data.map((alert) => {
const rawThreat = alert.threat || alert.title || '';
const rawCities = Array.isArray(alert.cities) ? alert.cities : (alert.data ? [alert.data] : []);
let translatedThreat = translateHebrew(rawThreat);
const translatedLocations = rawCities.map(translateCity);
// API sometimes puts city name in threat field; detect and move to locations
if (OREF_CITY_TRANSLATIONS[rawThreat]) {
if (!rawCities.includes(rawThreat)) {
translatedLocations.push(OREF_CITY_TRANSLATIONS[rawThreat]);
}
translatedThreat = 'Rocket/Missile Alert';
}
return {
id: alert.notificationId || String(Date.now()),
cat: categorizeOrefThreat(rawThreat),
title: translatedThreat,
titleHe: rawThreat,
data: translatedLocations,
dataHe: rawCities,
desc: alert.desc || '',
date: alert.date || new Date().toISOString(),
source: 'tzeva-adom',
};
});
} catch (err) {
console.warn(`[TzevaAdom] Fetch failed: ${err?.message || err}`);
return null;
}
}
async function orefFetchAlerts() {
let alerts = [];
let source = 'none';
// Primary: Tzeva Adom (free, no proxy needed)
const tzevaAlerts = await tzevaAdomFetchAlerts();
if (tzevaAlerts !== null) {
alerts = tzevaAlerts;
source = 'tzeva-adom';
} else if (OREF_PROXY_AVAILABLE) {
// Fallback: OREF direct (requires Israeli proxy)
try {
const raw = orefCurlFetch(OREF_PROXY_AUTH, OREF_ALERTS_URL);
const cleaned = stripBom(raw).trim();
if (cleaned && cleaned !== '[]' && cleaned !== 'null') {
try {
const parsed = JSON.parse(cleaned);
const orefArr = Array.isArray(parsed) ? parsed : [parsed];
alerts = orefArr.map((a) => ({
...a,
title: translateHebrew(a.title || ''),
titleHe: a.title || '',
data: Array.isArray(a.data) ? a.data.map(translateCity) : a.data ? [translateCity(a.data)] : [],
dataHe: Array.isArray(a.data) ? a.data : a.data ? [a.data] : [],
source: 'oref-direct',
}));
source = 'oref-direct';
} catch { alerts = []; }
}
} catch (err) {
orefState.lastError = orefPollFailure('OREF fallback poll error', err, OREF_PROXY_CREDENTIAL);
}
}
if (source === 'none') {
orefState.lastError = orefState.lastError || 'All siren sources unavailable';
orefState.lastPollAt = Date.now();
console.warn('[Relay] Siren poll: both Tzeva Adom and OREF failed');
orefPreSerializeResponses();
return;
}
try {
const newJson = JSON.stringify(alerts);
const changed = newJson !== orefState.lastAlertsJson;
orefState.lastAlerts = alerts;
orefState.lastAlertsJson = newJson;
orefState.lastPollAt = Date.now();
orefState.lastError = null;
if (changed && alerts.length > 0) {
orefState.history.push({
alerts,
timestamp: new Date().toISOString(),
});
orefState._persistVersion++;
const orefTitle = alerts[0]?.title || 'Siren alert';
const orefLocations = alerts.flatMap(a => Array.isArray(a.data) ? a.data : []);
const orefShown = orefLocations.slice(0, 3);
const orefOverflow = orefLocations.length - orefShown.length;
const orefLocationSuffix = orefShown.length
? ' — ' + orefShown.join(', ') + (orefOverflow > 0 ? ` +${orefOverflow} areas` : '')
: '';
publishNotificationEvent({
eventType: 'oref_siren',
payload: { title: orefTitle + orefLocationSuffix, source: source === 'tzeva-adom' ? 'Tzeva Adom / Pikud HaOref' : 'OREF Pikud HaOref', countryCode: 'IL' },
severity: 'critical',
variant: undefined,
}).catch(e => console.warn('[Notify] OREF publish error:', e?.message));
}
const cutoff = Date.now() - 24 * 60 * 60 * 1000;
orefState.historyCount24h = orefState.history
.filter(h => new Date(h.timestamp).getTime() > cutoff)
.reduce((sum, h) => sum + h.alerts.reduce((s, a) => s + (Array.isArray(a.data) ? a.data.length : 1), 0), 0);
const purgeCutoff = Date.now() - 7 * 24 * 60 * 60 * 1000;
const beforeLen = orefState.history.length;
orefState.history = orefState.history.filter(
h => new Date(h.timestamp).getTime() > purgeCutoff
);
if (orefState.history.length !== beforeLen) orefState._persistVersion++;
orefState.totalHistoryCount = orefState.history.reduce((sum, h) => {
return sum + h.alerts.reduce((s, a) => s + (Array.isArray(a.data) ? a.data.length : 1), 0);
}, 0);
orefPreSerializeResponses();
orefPersistHistory().catch(() => {});
} catch (err) {
orefState.lastError = orefPollFailure('OREF poll error', err, OREF_PROXY_CREDENTIAL);
orefPreSerializeResponses();
}
}
function orefPreSerializeResponses() {
const ts = orefState.lastPollAt ? new Date(orefState.lastPollAt).toISOString() : new Date().toISOString();
const alertsJson = JSON.stringify({
configured: SIREN_ALERTS_ENABLED,
alerts: orefState.lastAlerts || [],
historyCount24h: orefState.historyCount24h,
totalHistoryCount: orefState.totalHistoryCount,
timestamp: ts,
...(orefState.lastError ? { error: orefState.lastError } : {}),
});
orefState._alertsCache = { json: alertsJson, gzip: gzipSyncBuffer(alertsJson), brotli: brotliSyncBuffer(alertsJson) };
const historyJson = JSON.stringify({
configured: SIREN_ALERTS_ENABLED,
history: orefState.history || [],
historyCount24h: orefState.historyCount24h,
totalHistoryCount: orefState.totalHistoryCount,
timestamp: ts,
});
orefState._historyCache = { json: historyJson, gzip: gzipSyncBuffer(historyJson), brotli: brotliSyncBuffer(historyJson) };
}
async function orefBootstrapHistoryFromUpstream() {
const tmpFile = require('path').join(require('os').tmpdir(), `oref-history-${Date.now()}.json`);
let raw;
try {
raw = orefCurlFetch(OREF_PROXY_AUTH, OREF_HISTORY_URL, { toFile: tmpFile });
} finally {
try { require('fs').unlinkSync(tmpFile); } catch {}
}
const cleaned = stripBom(raw).trim();
if (!cleaned || cleaned === '[]') return;
const allRecords = JSON.parse(cleaned);
const records = allRecords.slice(0, 500);
const waves = new Map();
for (const r of records) {
const key = r.alertDate;
if (!waves.has(key)) waves.set(key, []);
waves.get(key).push(r);
}
const history = [];
let totalAlertRecords = 0;
for (const [dateStr, recs] of waves) {
const iso = orefDateToUTC(dateStr);
const byType = new Map();
let typeIdx = 0;
for (const r of recs) {
const k = `${r.category}|${r.title}`;
if (!byType.has(k)) {
byType.set(k, {
id: `${r.category}-${typeIdx++}-${dateStr.replace(/[^0-9]/g, '')}`,
cat: String(r.category),
title: r.title,
data: [],
desc: '',
alertDate: dateStr,
});
}
byType.get(k).data.push(r.data);
totalAlertRecords++;
}
history.push({ alerts: [...byType.values()], timestamp: new Date(iso).toISOString() });
}
history.sort((a, b) => new Date(a.timestamp) - new Date(b.timestamp));
orefState.history = history;
orefState.totalHistoryCount = totalAlertRecords;
const cutoff24h = Date.now() - 24 * 60 * 60 * 1000;
orefState.historyCount24h = history
.filter(h => new Date(h.timestamp).getTime() > cutoff24h)
.reduce((sum, h) => sum + h.alerts.reduce((s, a) => s + (Array.isArray(a.data) ? a.data.length : 1), 0), 0);
orefState.bootstrapSource = 'upstream';
if (history.length > 0) orefState._persistVersion++;
console.log(`[Relay] OREF history bootstrap: ${totalAlertRecords} records across ${history.length} waves`);
orefSaveLocalHistory();
}
const OREF_PERSIST_MAX_WAVES = 200;
const OREF_PERSIST_TTL_SECONDS = 7 * 24 * 60 * 60;
async function orefPersistHistory() {
if (!UPSTASH_ENABLED) return;
if (orefState._persistVersion === orefState._lastPersistedVersion) return;
if (orefState._persistInFlight) return;
orefState._persistInFlight = true;
const versionAtStart = orefState._persistVersion;
try {
let waves = orefState.history;
if (waves.length > OREF_PERSIST_MAX_WAVES) {
console.warn(`[Relay] OREF persist: truncating ${waves.length} waves to ${OREF_PERSIST_MAX_WAVES}`);
waves = waves.slice(-OREF_PERSIST_MAX_WAVES);
}
const payload = {
history: waves,
historyCount24h: orefState.historyCount24h,
totalHistoryCount: orefState.totalHistoryCount,
activeAlertCount: orefState.lastAlerts?.length || 0,
persistedAt: new Date().toISOString(),
};
const ok = await envelopeWrite(OREF_REDIS_KEY, payload, OREF_PERSIST_TTL_SECONDS, { recordCount: waves.length, sourceVersion: 'oref', zeroOk: true });
if (ok) {
orefState._lastPersistedVersion = versionAtStart;
}
// Companion seed-meta:* write — the OREF payload only carries `persistedAt`
// (an ISO string not in extractTimestamp's recognised set), so without this
// key the regional-snapshot freshness classifier would flag the input as
// STALE on every run (#3781). Tracked by api/health.js for staleness alerts.
//
// Gate on `ok`: if the envelope write failed (Upstash 5xx / network blip),
// a successful meta write alone would tell the freshness classifier the
// input is FRESH for data that does not actually exist in Redis. The 7d
// meta TTL is deliberately wider than the 15min maxAgeMin so a brief
// seed gap keeps the meta around for diagnostics; freshness still flips
// STALE off the now-vs-fetchedAt delta.
//
// NOTE (#3781 review): the transit-summaries write at line ~7370 still
// does its meta write unconditionally and has the same failure mode.
// That is a pre-existing bug intentionally left out of scope here; it
// should be fixed in a follow-up PR.
if (ok) {
await upstashSet('seed-meta:relay:oref:history', { fetchedAt: Date.now(), recordCount: waves.length }, 604800);
}
orefSaveLocalHistory();
} finally {
orefState._persistInFlight = false;
}
}
function orefLoadLocalHistory() {
if (!OREF_LOCAL_FILE) return null;
try {
const raw = require('fs').readFileSync(OREF_LOCAL_FILE, 'utf8');
const data = JSON.parse(raw);
if (!Array.isArray(data.history) || data.history.length === 0) return null;
const valid = data.history.every(
h => Array.isArray(h.alerts) && typeof h.timestamp === 'string'
);
if (!valid) return null;
const purgeCutoff = Date.now() - 7 * 24 * 60 * 60 * 1000;
const filtered = data.history.filter(
h => new Date(h.timestamp).getTime() > purgeCutoff
);
if (filtered.length === 0) {
console.log('[Relay] OREF local file data all stale (>7d)');
return null;
}
console.log(`[Relay] OREF local file: ${filtered.length} waves (saved ${data.savedAt || 'unknown'})`);
return filtered;
} catch (err) {
if (err.code !== 'ENOENT') console.warn('[Relay] OREF local file read error:', err.message);
return null;
}
}
function orefSaveLocalHistory() {
if (!OREF_LOCAL_FILE) return;
try {
const fs = require('fs');
let waves = orefState.history;
if (waves.length > OREF_PERSIST_MAX_WAVES) {
waves = waves.slice(-OREF_PERSIST_MAX_WAVES);
}
const payload = JSON.stringify({
history: waves,
historyCount24h: orefState.historyCount24h,
totalHistoryCount: orefState.totalHistoryCount,
savedAt: new Date().toISOString(),
});
const tmpPath = OREF_LOCAL_FILE + '.tmp';
fs.writeFileSync(tmpPath, payload, 'utf8');
fs.renameSync(tmpPath, OREF_LOCAL_FILE);
} catch (err) {
console.warn('[Relay] OREF local file save error:', err.message);
}
}
async function orefBootstrapHistoryWithRetry() {
// Phase 0: local file (Railway volume — instant, no network)
if (OREF_LOCAL_FILE) {
const local = orefLoadLocalHistory();
if (local && local.length > 0) {
const cutoff24h = Date.now() - 24 * 60 * 60 * 1000;
orefState.history = local;
orefState.totalHistoryCount = local.reduce((sum, h) => {
return sum + h.alerts.reduce((s, a) => s + (Array.isArray(a.data) ? a.data.length : 1), 0);
}, 0);
orefState.historyCount24h = local
.filter(h => new Date(h.timestamp).getTime() > cutoff24h)
.reduce((sum, h) => sum + h.alerts.reduce((s, a) => s + (Array.isArray(a.data) ? a.data.length : 1), 0), 0);
const newest = local[local.length - 1];
orefState.lastAlertsJson = JSON.stringify(newest.alerts);
orefState.bootstrapSource = 'local-file';
console.log(`[Relay] OREF history loaded from local file: ${orefState.totalHistoryCount} records across ${local.length} waves`);
return;
}
}
// Phase 1: try Redis first
try {
// envelopeRead unwraps the {_seed, data} shape written by orefPersistHistory()
// at line 1133. Reading raw left cached.history undefined, so OREF state was
// never restored across relay restarts (reported in PR #3139 review).
const cached = await envelopeRead(OREF_REDIS_KEY);
if (cached && Array.isArray(cached.history) && cached.history.length > 0) {
const valid = cached.history.every(
h => Array.isArray(h.alerts) && typeof h.timestamp === 'string'
);
if (valid) {
const cutoff24h = Date.now() - 24 * 60 * 60 * 1000;
const purgeCutoff = Date.now() - 7 * 24 * 60 * 60 * 1000;
const filtered = cached.history.filter(
h => new Date(h.timestamp).getTime() > purgeCutoff
);
if (filtered.length > 0) {
orefState.history = filtered;
orefState.totalHistoryCount = filtered.reduce((sum, h) => {
return sum + h.alerts.reduce((s, a) => s + (Array.isArray(a.data) ? a.data.length : 1), 0);
}, 0);
orefState.historyCount24h = filtered
.filter(h => new Date(h.timestamp).getTime() > cutoff24h)
.reduce((sum, h) => sum + h.alerts.reduce((s, a) => s + (Array.isArray(a.data) ? a.data.length : 1), 0), 0);
const newest = filtered[filtered.length - 1];
orefState.lastAlertsJson = JSON.stringify(newest.alerts);
orefState.bootstrapSource = 'redis';
console.log(`[Relay] OREF history loaded from Redis: ${orefState.totalHistoryCount} records across ${filtered.length} waves (persisted ${cached.persistedAt || 'unknown'})`);
return;
}
console.log('[Relay] OREF Redis data all stale (>7d) — falling through to upstream');
}
}
} catch (err) {
console.warn('[Relay] OREF Redis bootstrap failed:', err?.message || err);
}
// Phase 2: upstream with retry + exponential backoff
const MAX_ATTEMPTS = 3;
const BASE_DELAY_MS = 3000;
for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
try {
await orefBootstrapHistoryFromUpstream();
if (UPSTASH_ENABLED) {
await orefPersistHistory().catch(() => {});
}
console.log(`[Relay] OREF upstream bootstrap succeeded on attempt ${attempt}`);
return;
} catch (err) {
const msg = redactOrefError(err?.message || String(err), OREF_PROXY_CREDENTIAL);
console.warn(`[Relay] OREF upstream bootstrap attempt ${attempt}/${MAX_ATTEMPTS} failed: ${msg}`);
if (attempt < MAX_ATTEMPTS) {
const delay = BASE_DELAY_MS * 2 ** (attempt - 1) + Math.random() * 1000;
await new Promise(r => setTimeout(r, delay));
}
}
}
orefState.bootstrapSource = null;
console.warn('[Relay] OREF bootstrap exhausted all attempts — starting with empty history');
}
async function startOrefPollLoop() {
if (!SIREN_ALERTS_ENABLED) {
console.log('[Relay] Siren alerts disabled');
return;
}
console.log(`[Relay] Siren alerts: primary=Tzeva Adom, fallback=${OREF_PROXY_AVAILABLE ? 'OREF (proxy)' : 'none'}`);
await orefBootstrapHistoryWithRetry();
console.log(`[Relay] OREF bootstrap complete (source: ${orefState.bootstrapSource || 'none'}, redis: ${UPSTASH_ENABLED})`);
orefFetchAlerts().catch(e => console.warn('[Relay] OREF initial poll error:', e?.message || e));
setInterval(() => {
orefFetchAlerts().catch(e => console.warn('[Relay] OREF poll error:', e?.message || e));
}, OREF_POLL_INTERVAL_MS).unref?.();
console.log(`[Relay] OREF poll loop started (interval ${OREF_POLL_INTERVAL_MS}ms)`);
}
// ─────────────────────────────────────────────────────────────
// UCDP GED Events — fetch paginated conflict data, write to Redis
// ─────────────────────────────────────────────────────────────
const UCDP_ACCESS_TOKEN = (process.env.UCDP_ACCESS_TOKEN || process.env.UC_DP_KEY || '').trim();
const UCDP_REDIS_KEY = 'conflict:ucdp-events:v1';
const UCDP_PAGE_SIZE = 1000;
const UCDP_MAX_PAGES = 6;
// GED Candidate discovery/fetch/merge is shared with scripts/seed-ucdp-events.mjs
// so the two UCDP writers cannot drift (they already had, on discovery
// concurrency and probe timeout).
const {
CANDIDATE_MAX_PAGES: UCDP_CANDIDATE_MAX_PAGES,
discoverCandidateVersion: ucdpDiscoverCandidateVersion,
fetchCandidatePages: ucdpFetchCandidatePages,
capWithAnnualFloor: ucdpCapWithAnnualFloor,
candidateContentMeta: ucdpCandidateContentMeta,
hasPreciseUcdpLocation,
} = require('./shared/ucdp-candidate.cjs');
const UCDP_MAX_EVENTS = 2000; // Default capacity. Complete candidate rows plus the annual floor take precedence.
// CII accepts two years, but these newest annual pages and the candidate release
// only cover a 365-day trailing slice. Candidate rows can exceed the default cap.
const UCDP_TRAILING_WINDOW_MS = 365 * 24 * 60 * 60 * 1000;
const UCDP_POLL_INTERVAL_MS = 6 * 60 * 60 * 1000; // 6 hours
const UCDP_TTL_SECONDS = 86400; // 24h safety net
const UCDP_VIOLENCE_TYPE_MAP = { 1: 'UCDP_VIOLENCE_TYPE_STATE_BASED', 2: 'UCDP_VIOLENCE_TYPE_NON_STATE', 3: 'UCDP_VIOLENCE_TYPE_ONE_SIDED' };
function ucdpFetchPage(version, page, timeoutMs = 30000) {
return new Promise((resolve, reject) => {
const pageUrl = new URL(`https://ucdpapi.pcr.uu.se/api/gedevents/${version}?pagesize=${UCDP_PAGE_SIZE}&page=${page}`);
const headers = { Accept: 'application/json', 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36' };
if (UCDP_ACCESS_TOKEN) headers['x-ucdp-access-token'] = UCDP_ACCESS_TOKEN;
const req = https.request(pageUrl, { signal: AbortSignal.timeout(timeoutMs), method: 'GET', headers, timeout: timeoutMs }, (resp) => {
if (resp.statusCode === 401 || resp.statusCode === 403) {
resp.resume();
return reject(new Error(`UCDP ${version} page ${page}: HTTP ${resp.statusCode} — API token required (set UCDP_ACCESS_TOKEN env var)`));
}
if (resp.statusCode < 200 || resp.statusCode >= 300) {
resp.resume();
return reject(new Error(`UCDP ${version} page ${page}: HTTP ${resp.statusCode}`));
}
let data = '';
resp.on('data', (chunk) => { data += chunk; });
resp.on('end', () => {
try {
const parsed = JSON.parse(data);
if (typeof parsed === 'string') return reject(new Error(`UCDP ${version} page ${page}: ${parsed}`));
resolve(parsed);
} catch (e) { reject(e); }
});
});
req.on('error', reject);
req.on('timeout', () => { req.destroy(); reject(new Error('UCDP timeout')); });
req.end();
});
}
// Compare UCDP GED version strings ('26.1' > '25.1' > '24.1', '25.0.6' > '25.0.5')
// numerically segment-by-segment so the NEWEST release sorts first.
function ucdpVersionRank(v) {
return String(v).split('.').map((n) => Number(n) || 0);
}
function ucdpVersionNewer(a, b) {
const ra = ucdpVersionRank(a);
const rb = ucdpVersionRank(b);
for (let i = 0; i < Math.max(ra.length, rb.length); i++) {
const d = (ra[i] || 0) - (rb[i] || 0);
if (d !== 0) return d > 0;
}
return false;
}
async function ucdpDiscoverVersion() {
const year = new Date().getFullYear() - 2000;
const candidates = [...new Set([`${year}.1`, `${year - 1}.1`, '25.1', '24.1'])];
// Probe ALL candidates, then prefer the NEWEST version that returned events.
// Promise.any (first-responder) used to win here, which let an OLDER release
// that merely replied faster win: it froze conflict:ucdp-events:v1 at v24.1
// (2023 data, 889 days old) while v25.1 was available, dropping every event
// outside the CII 2-year conflict recency window and flipping
// /api/health.riskScores to COVERAGE_PARTIAL. allSettled waits for the slowest
// candidate, so the discovery probe uses a tighter 15s timeout (vs the 30s
// full-page default) — a non-existent version that hangs can't stall the
// 6h seed for 30s, while one page is comfortably fetchable in 15s.
const DISCOVER_TIMEOUT_MS = 15000;
const settled = await Promise.allSettled(candidates.map(async (v) => {
const p0 = await ucdpFetchPage(v, 0, DISCOVER_TIMEOUT_MS);
if (!Array.isArray(p0?.Result) || p0.Result.length === 0) throw new Error(`${v}: no results`);
return { version: v, page0: p0 };
}));
const valid = settled.filter((s) => s.status === 'fulfilled').map((s) => s.value);
if (valid.length === 0) {
const reasons = settled.map((s) => s.reason?.message).filter(Boolean).join('; ');
throw new Error(`No valid UCDP GED version found (${reasons})`);
}
// 3-way comparator: equal versions must return 0 (Array.sort contract — an
// inconsistent comparator is undefined behaviour and can reorder
// non-deterministically). Set-dedup makes ties unlikely, but stay spec-correct.
valid.sort((a, b) => (ucdpVersionNewer(a.version, b.version) ? -1 : ucdpVersionNewer(b.version, a.version) ? 1 : 0));
return valid[0];
}
// UCDP also publishes GED Candidate releases monthly ('${year}.0.N'), with "not
// more than a month's lag globally" per UCDP's docs — unlike the ANNUAL release
// above, which is finalized once a year and lags ~7 months behind by the time
// the next one lands. Discovery, paging and merge semantics live in
// scripts/shared/ucdp-candidate.cjs (required above) so this relay and the
// backup cron stay byte-identical in behaviour; only the transport differs.
async function seedUcdpEvents() {
try {
const { version, page0 } = await ucdpDiscoverVersion();
const totalPages = Math.max(1, Number(page0?.TotalPages) || 1);
const newestPage = totalPages - 1;
console.log(`[UCDP] Version ${version}, ${totalPages} total pages`);
const FAILED = Symbol('failed');
const fetches = [];
for (let offset = 0; offset < UCDP_MAX_PAGES && (newestPage - offset) >= 0; offset++) {
const pg = newestPage - offset;
fetches.push(pg === 0 ? Promise.resolve(page0) : ucdpFetchPage(version, pg).catch((err) => {
console.warn(`[UCDP] page ${pg}: ${err.message || err}`);
return FAILED;
}));
}
const pageResults = await Promise.all(fetches);
const allEvents = [];
let latestMs = NaN;
let failedPages = 0;
for (const raw of pageResults) {
if (raw === FAILED) { failedPages++; continue; }
const events = Array.isArray(raw?.Result) ? raw.Result : [];
allEvents.push(...events);
for (const e of events) {
const ms = e?.date_start ? Date.parse(String(e.date_start)) : NaN;
if (Number.isFinite(ms) && (!Number.isFinite(latestMs) || ms > latestMs)) latestMs = ms;
}
}
// If no events from newest pages, extend existing cache TTL instead of overwriting
// with stale/empty data. This preserves the last known good payload.
if (allEvents.length === 0 && failedPages > 0) {
console.warn(`[UCDP] All ${failedPages} newest pages failed, extending existing key TTL (preserving last good data)`);
try { await upstashExpire(UCDP_REDIS_KEY, UCDP_TTL_SECONDS); } catch {}
// Do NOT update seed-meta: health should reflect actual data freshness, not this failed attempt
return;
}
// Merge the newest GED Candidate release on top of the annual base (ADD,
// never replace — a candidate alone is ~1.8k events vs the annual's ~418k).
// The annual base is already in allEvents; append candidate events and dedupe
// by id so a candidate's revision of an event also present in the annual
// release wins (it's the fresher record).
let candidateVersion = null;
let candidateComplete = false;
const candidateIds = new Set();
try {
const candidate = await ucdpDiscoverCandidateVersion(ucdpFetchPage);
if (candidate) {
const merged = await ucdpFetchCandidatePages(ucdpFetchPage, candidate);
// Only claim the candidate version once the release was fetched whole.
// A partial fetch published as `26.0.6` is indistinguishable from a
// complete one downstream, which is how a silently degraded merge would
// look healthy.
candidateComplete = merged.complete && !merged.truncated;
candidateVersion = candidateComplete ? candidate.version : `${candidate.version}+partial`;
if (merged.failedPages > 0) {
console.warn(`[UCDP] candidate ${candidate.version}: ${merged.failedPages} page(s) failed — publishing as partial`);
}
if (merged.truncated) {
console.warn(`[UCDP] candidate ${candidate.version}: ${merged.totalPages} pages exceeds cap ${UCDP_CANDIDATE_MAX_PAGES} — ${merged.totalPages - UCDP_CANDIDATE_MAX_PAGES} page(s) dropped`);
}
for (const e of merged.events) {
if (e?.id != null) candidateIds.add(String(e.id));
const ms = e?.date_start ? Date.parse(String(e.date_start)) : NaN;
if (Number.isFinite(ms) && (!Number.isFinite(latestMs) || ms > latestMs)) latestMs = ms;
}
allEvents.push(...merged.events);
console.log(`[UCDP] Merged candidate ${candidateVersion}: +${merged.events.length} events`);
}
} catch (err) {
console.warn(`[UCDP] Candidate merge skipped: ${err.message || err}`);
}
const byId = new Map();
for (const e of allEvents) {
const id = e?.id != null ? String(e.id) : '';
byId.set(id || Symbol(id), e);
}
const dedupedEvents = [...byId.values()];
const filtered = dedupedEvents.filter((e) => {
if (!Number.isFinite(latestMs)) return true;
const ms = e?.date_start ? Date.parse(String(e.date_start)) : NaN;
return Number.isFinite(ms) && ms >= (latestMs - UCDP_TRAILING_WINDOW_MS);
});
const mapped = filtered.map((e) => ({
id: String(e.id || ''),
dateStart: Date.parse(e.date_start) || 0,
dateEnd: Date.parse(e.date_end) || 0,
location: { latitude: Number(e.latitude) || 0, longitude: Number(e.longitude) || 0 },
country: e.country || '',
hasPreciseLocation: hasPreciseUcdpLocation(e),
locationName: (e.where_coordinates || '').substring(0, 200),
sideA: (e.side_a || '').substring(0, 200),
sideB: (e.side_b || '').substring(0, 200),
deathsBest: Number(e.best) || 0,
deathsLow: Number(e.low) || 0,
deathsHigh: Number(e.high) || 0,
violenceType: UCDP_VIOLENCE_TYPE_MAP[e.type_of_violence] || 'UCDP_VIOLENCE_TYPE_UNSPECIFIED',
sourceOriginal: (e.source_original || '').substring(0, 300),
})).sort((a, b) => b.dateStart - a.dateStart);
// Keep monthly candidate aggregates and the annual conflict-floor history.
const capped = ucdpCapWithAnnualFloor(mapped, (e) => candidateIds.has(e.id), UCDP_MAX_EVENTS);
// Partial success but 0 events after filtering: extend TTL, don't overwrite
if (capped.length === 0) {
console.warn(`[UCDP] 0 events after filtering (failed pages: ${failedPages}), extending existing key TTL`);
try { await upstashExpire(UCDP_REDIS_KEY, UCDP_TTL_SECONDS); } catch {}
return;
}
const payload = { events: capped, fetchedAt: Date.now(), version, candidateVersion, candidateComplete, annualFailedPages: failedPages, totalRaw: allEvents.length, filteredCount: capped.length };
const ok = await envelopeWrite(UCDP_REDIS_KEY, payload, UCDP_TTL_SECONDS, { recordCount: capped.length, sourceVersion: 'ucdp' });
// Content-age trio: api/health.js treats the presence of maxContentAgeMin as
// the opt-in signal and reports STALE_CONTENT once the newest event outruns
// the budget. Without it a silently dead candidate merge is invisible —
// fetchedAt stays fresh and recordCount stays full while the data itself
// falls back to the annual release's ~7-month lag.
await upstashSet('seed-meta:conflict:ucdp-events', {
fetchedAt: Date.now(),
recordCount: capped.length,
candidateVersion,
candidateComplete,
annualFailedPages: failedPages,
...ucdpCandidateContentMeta(capped),
}, 604800);
console.log(`[UCDP] Seeded ${capped.length} events (raw: ${allEvents.length}, candidate: ${candidateVersion || 'none'}, failed pages: ${failedPages}, redis: ${ok ? 'OK' : 'FAIL'})`);
// The bootstrap tier hydrates from the compact projection, which health
// watches under its own seed-meta. The backup cron skips while the canonical
// seed-meta above is fresh, so this writer must refresh the projection too or
// it ages past its budget between backup runs (Sentry WORLDMONITOR-1DC).
// Best-effort, like the backup cron: the canonical key is already published.
if (ok) {
try {
const { compactUcdpDashboardPayload, UCDP_BOOTSTRAP_KEY, UCDP_BOOTSTRAP_META_KEY } = await import('./_ucdp-dashboard.mjs');
const compact = compactUcdpDashboardPayload(payload);
if (!await upstashSet(UCDP_BOOTSTRAP_KEY, compact, UCDP_TTL_SECONDS)) throw new Error('projection SET failed');
if (!await upstashSet(UCDP_BOOTSTRAP_META_KEY, { fetchedAt: Date.now(), recordCount: compact.events.length }, 604800)) {
throw new Error('projection seed-meta SET failed');
}
} catch (err) {
console.warn(`[UCDP] Compact projection write failed: ${err?.message || err}`);
}
}
const newConflicts = capped.filter(e => e.deathsBest >= 10 && !ucdpPrevAlertedIds.has(e.id)).sort((a, b) => b.deathsBest - a.deathsBest);
for (const e of newConflicts.slice(0, 2)) {
ucdpPrevAlertedIds.add(e.id);
const parties = e.sideA && e.sideB ? `${e.sideA.slice(0, 40)} vs ${e.sideB.slice(0, 40)}` : e.sideA || e.sideB || 'Unknown parties';
const countryCode = normalizeNotificationCountryCode(e.country);
publishNotificationEvent({
eventType: 'conflict_escalation',
payload: { title: `${e.country}: ${parties} — ${e.deathsBest} casualties`, source: 'UCDP', ...(countryCode ? { countryCode } : {}) },
severity: e.deathsBest >= 50 ? 'critical' : 'high',
variant: undefined,
dedupTtl: 86400,
}).catch(err => console.warn('[Notify] UCDP publish error:', err?.message));
}
if (ucdpPrevAlertedIds.size > 500) ucdpPrevAlertedIds.clear();
} catch (e) {
console.warn('[UCDP] Seed error:', e?.message || e);
}
}
async function startUcdpSeedLoop() {
if (!UPSTASH_ENABLED) {
console.log('[UCDP] Disabled (no Upstash Redis)');
return;
}
console.log(`[UCDP] Seed loop starting (interval ${UCDP_POLL_INTERVAL_MS / 1000 / 60}min, token: ${UCDP_ACCESS_TOKEN ? 'yes' : 'no'})`);
startBootSeedLoop('UCDP', 'seed-meta:conflict:ucdp-events', UCDP_POLL_INTERVAL_MS, seedUcdpEvents, e => console.warn('[UCDP] Initial seed error:', e?.message || e), e => console.warn('[UCDP] Seed error:', e?.message || e));
}
// ─────────────────────────────────────────────────────────────
// Satellite TLE Seed — CelesTrak NORAD elements → Redis
// ─────────────────────────────────────────────────────────────
const SAT_SEED_INTERVAL_MS = 7_200_000;
const SAT_SEED_TTL = 21_600; // 6h — survives 3 missed cycles before data expires
const SAT_RETRY_MS = 20 * 60 * 1000; // retry 20min after failure instead of waiting 120min
const SAT_GROUPS = ['military', 'resource'];
const SAT_NAME_FILTERS = [
/^YAOGAN/i, /^GAOFEN/i, /^JILIN/i,
/^COSMOS 2[4-9]\d{2}/i,
/^COSMO-SKYMED/i, /^TERRASAR/i, /^PAZ$/i, /^SAR-LUPE/i,
/^WORLDVIEW/i, /^SKYSAT/i, /^PLEIADES/i, /^KOMPSAT/i,
/^SAPPHIRE/i, /^PRAETORIAN/i,
/^SENTINEL/i,
/^CARTOSAT/i,
/^GOKTURK/i, /^RASAT/i,
/^USA[ -]?\d/i,
/^ZIYUAN/i,
];
function satClassify(name) {
const n = name.toUpperCase();
let type = 'military';
if (/COSMO-SKYMED|TERRASAR|PAZ|SAR-LUPE|YAOGAN/i.test(n)) type = 'sar';
else if (/WORLDVIEW|SKYSAT|PLEIADES|KOMPSAT|GAOFEN|JILIN|CARTOSAT|ZIYUAN/i.test(n)) type = 'optical';
else if (/SAPPHIRE|PRAETORIAN|USA|GOKTURK/i.test(n)) type = 'military';
let country = 'OTHER';
if (/^YAOGAN|^GAOFEN|^JILIN|^ZIYUAN/i.test(n)) country = 'CN';
else if (/^COSMOS/i.test(n)) country = 'RU';
else if (/^WORLDVIEW|^SAPPHIRE|^PRAETORIAN|^USA|^SKYSAT/i.test(n)) country = 'US';
else if (/^SENTINEL|^COSMO-SKYMED|^TERRASAR|^SAR-LUPE|^PAZ|^PLEIADES/i.test(n)) country = 'EU';
else if (/^KOMPSAT/i.test(n)) country = 'KR';
else if (/^CARTOSAT/i.test(n)) country = 'IN';
else if (/^GOKTURK|^RASAT/i.test(n)) country = 'TR';
return { type, country };
}
let satSeedInFlight = false;
let satRetryTimer = null;
async function seedSatelliteTLEs() {
if (satSeedInFlight) return;
satSeedInFlight = true;
if (satRetryTimer) { clearTimeout(satRetryTimer); satRetryTimer = null; }
const t0 = Date.now();
try {
const byNorad = new Map();
for (const group of SAT_GROUPS) {
let text;
try {
text = await new Promise((resolve, reject) => {
const url = new URL(`https://celestrak.org/NORAD/elements/gp.php?GROUP=${group}&FORMAT=tle`);
const req = https.request(url, { signal: AbortSignal.timeout(15000), method: 'GET', headers: { 'User-Agent': CHROME_UA }, timeout: 15000 }, (resp) => {
if (resp.statusCode < 200 || resp.statusCode >= 300) {
resp.resume();
return reject(new Error(`CelesTrak ${group}: HTTP ${resp.statusCode}`));
}
let data = '';
let size = 0;
resp.on('data', (chunk) => {
size += chunk.length;
if (size > 2 * 1024 * 1024) { req.destroy(); return reject(new Error(`CelesTrak ${group}: payload > 2MB`)); }
data += chunk;
});
resp.on('end', () => resolve(data));
});
req.on('error', reject);
req.on('timeout', () => { req.destroy(); reject(new Error(`CelesTrak ${group}: timeout`)); });
req.end();
});
} catch (e) {
console.warn(`[Satellites] Skipping group ${group}:`, e?.message || e);
continue;
}
const lines = text.split('\n').map(l => l.trimEnd());
for (let i = 0; i < lines.length - 2; i++) {
const l1 = lines[i + 1];
const l2 = lines[i + 2];
if (!l1.startsWith('1 ') || !l2.startsWith('2 ')) continue;
if (l1.length !== 69 || l2.length !== 69) continue;
const name = lines[i].trim();
const noradId = l1.substring(2, 7).trim();
if (!byNorad.has(noradId)) {
byNorad.set(noradId, { noradId, name, line1: l1, line2: l2 });
}
i += 2;
}
}
const satellites = [];
for (const sat of byNorad.values()) {
if (!SAT_NAME_FILTERS.some(rx => rx.test(sat.name))) continue;
const { type, country } = satClassify(sat.name);
satellites.push({ ...sat, type, country });
}
if (satellites.length === 0) {
console.warn('[Satellites] No matching TLEs found — extending existing key TTL, retrying in 20min');
try { await upstashExpire('intelligence:satellites:tle:v1', SAT_SEED_TTL); } catch {}
satRetryTimer = setTimeout(() => { seedSatelliteTLEs().catch(() => {}); }, SAT_RETRY_MS);
return;
}
const payload = { satellites, fetchedAt: Date.now() };
const ok = await envelopeWrite('intelligence:satellites:tle:v1', payload, SAT_SEED_TTL, { recordCount: satellites.length, sourceVersion: 'celestrak' });
await upstashSet('seed-meta:intelligence:satellites', { fetchedAt: Date.now(), recordCount: satellites.length }, 604800);
console.log(`[Satellites] Seeded ${satellites.length} TLEs (redis: ${ok ? 'OK' : 'FAIL'}) in ${((Date.now() - t0) / 1000).toFixed(1)}s`);
} catch (e) {
console.warn('[Satellites] Seed error:', e?.message || e, '— extending existing key TTL, retrying in 20min');
try { await upstashExpire('intelligence:satellites:tle:v1', SAT_SEED_TTL); } catch {}
satRetryTimer = setTimeout(() => { seedSatelliteTLEs().catch(() => {}); }, SAT_RETRY_MS);
} finally {
satSeedInFlight = false;
}
}
async function startSatelliteSeedLoop() {
if (!UPSTASH_ENABLED) {
console.log('[Satellites] Disabled (no Upstash Redis)');
return;
}
console.log(`[Satellites] Seed loop starting (interval ${SAT_SEED_INTERVAL_MS / 1000 / 60}min)`);
startBootSeedLoop('Satellites', 'seed-meta:intelligence:satellites', SAT_SEED_INTERVAL_MS, seedSatelliteTLEs, e => console.warn('[Satellites] Initial seed error:', e?.message || e), e => console.warn('[Satellites] Seed error:', e?.message || e));
}
// ─────────────────────────────────────────────────────────────
// Market Data Seed — Railway fetches Yahoo/Finnhub → writes to Redis
// so Vercel handlers serve from cache (avoids Yahoo 429 from Vercel IPs)
// ─────────────────────────────────────────────────────────────
const FINNHUB_API_KEY = process.env.FINNHUB_API_KEY || '';
const MARKET_SEED_INTERVAL_MS = 300_000; // 5 min
const MARKET_SEED_TTL = 7200; // 2h — survive extended Yahoo/upstream outages
const _configuredYahooRefreshIntervalMs = Number(process.env.MARKET_YAHOO_REFRESH_INTERVAL_MS);
const MARKET_YAHOO_REFRESH_INTERVAL_MS = Math.max(
MARKET_SEED_INTERVAL_MS,
Number.isFinite(_configuredYahooRefreshIntervalMs) && _configuredYahooRefreshIntervalMs > 0
? _configuredYahooRefreshIntervalMs
: 900_000,
);
const { loadMarketSeedUniverse } = require('./shared/market-seed-universe.cjs');
const _stockCfg = requireShared('stocks.json');
const _stockUniverse = loadMarketSeedUniverse(_stockCfg);
const MARKET_SYMBOLS = _stockUniverse.allSymbols;
const MARKET_AUXILIARY_SYMBOLS = _stockUniverse.auxiliarySymbols;
const MARKET_META = _stockUniverse.metaBySymbol;
const _commodityCfg = requireShared('commodities.json');
const COMMODITY_SYMBOLS = _commodityCfg.commodities.map(c => c.symbol);
const COMMODITY_META = new Map(_commodityCfg.commodities.map(c => [c.symbol, { name: c.name, display: c.display }]));
const SECTOR_SYMBOLS = ['XLK', 'XLF', 'XLE', 'XLV', 'XLY', 'XLI', 'XLP', 'XLU', 'XLB', 'XLRE', 'XLC', 'SMH'];
// Symbols that must come from Yahoo — Finnhub doesn't carry futures (=F),
// major indices, or the exchange-qualified Asian symbols in stocks.json.
const YAHOO_ONLY = new Set([
..._stockCfg.yahooOnly,
...COMMODITY_SYMBOLS.filter(s => s.endsWith('=F') || s.startsWith('^')),
'URA', 'LIT',
// Spot gold and forex pairs (=X suffix) — not on Finnhub
...COMMODITY_SYMBOLS.filter(s => s.endsWith('=X')),
]);
// CommonJS cannot import finiteObservation from _seed-utils.mjs. The relay test
// keeps this boundary guard aligned with the cron seeder.
function _finiteObservation(value) {
if (typeof value !== 'number' && (typeof value !== 'string' || !value.trim())) return null;
const n = Number(value);
return Number.isFinite(n) ? n : null;
}
function _parseYahooChartJson(body) {
try {
const data = JSON.parse(body);
const result = data?.chart?.result?.[0];
const meta = result?.meta;
if (!meta) return null;
const price = _finiteObservation(meta.regularMarketPrice);
if (price == null) return null;
const prevClose = [meta.chartPreviousClose, meta.previousClose]
.map(_finiteObservation)
.find((value) => value != null && value !== 0) ?? price;
const change = prevClose ? ((price - prevClose) / prevClose) * 100 : 0;
const closes = result.indicators?.quote?.[0]?.close;
const sparkline = Array.isArray(closes)
? closes
.map(_finiteObservation)
.filter((value) => value != null)
.map((value) => (value !== 0 ? Number(value.toPrecision(7)) : value))
: [];
return {
price,
change,
sparkline,
awaitingSessionOpen: awaitsSessionOpen(data),
lastTradeAt: meta.regularMarketTime,
chart: data,
};
} catch { return null; }
}
// Two independent Decodo egress pools → two independent cooldowns. Yahoo may
// block one while the other is healthy (2026-04-16: CONNECT blocked, curl OK).
// Sharing state would let one route's outage suppress the working route.
const _YAHOO_PROXY_COOLDOWN_MS = 5 * 60 * 1000;
let _yahooConnectProxyFailCount = 0; // fetchYahooChartDirect via gate.decodo.com (CONNECT)
let _yahooConnectProxyCooldownUntil = 0;
function _fetchYahooChartNoProxy(symbol, query = '') {
return new Promise((resolve) => {
const url = `https://query1.finance.yahoo.com/v8/finance/chart/${encodeURIComponent(symbol)}${query}`;
const req = https.get(url, {
signal: AbortSignal.timeout(10000),
headers: { 'User-Agent': CHROME_UA, Accept: 'application/json' },
timeout: 10000,
}, (resp) => {
if (resp.statusCode !== 200) {
resp.resume();
logThrottled('warn', `market-yahoo-${resp.statusCode}:${symbol}`, `[Market] Yahoo ${symbol} HTTP ${resp.statusCode}`);
return resolve(null);
}
let body = '';
resp.on('data', (chunk) => { body += chunk; });
resp.on('end', () => resolve(_parseYahooChartJson(body)));
});
req.on('error', (err) => { logThrottled('warn', `market-yahoo-err:${symbol}`, `[Market] Yahoo ${symbol} error: ${err.message}`); resolve(null); });
req.on('timeout', () => { req.destroy(); logThrottled('warn', `market-yahoo-timeout:${symbol}`, `[Market] Yahoo ${symbol} timeout`); resolve(null); });
});
}
// A rolled chart's previous close stays the same until the next trade, so one
// daily-chart request per symbol covers a whole weekend of 5-minute cycles.
const _previousSessionCloses = new Map();
async function _withPreviousSessionChange(symbol, quote) {
let cached = _previousSessionCloses.get(symbol);
if (cached?.lastTradeAt !== quote.lastTradeAt) {
await sleep(150);
const daily = await _fetchYahooChartOnce(symbol, PREVIOUS_SESSION_CHART_QUERY);
const close = previousSessionClose(daily?.chart, quote.lastTradeAt);
if (!close) return quote;
cached = { lastTradeAt: quote.lastTradeAt, close };
_previousSessionCloses.set(symbol, cached);
}
return { ...quote, change: ((quote.price - cached.close) / cached.close) * 100 };
}
// History bars before the default chart's first bar never change, so one
// history request per symbol per session covers every 5-minute cycle. After a
// failed request the symbol waits SPARKLINE_HISTORY_RETRY_SECONDS.
const _sparklineHistories = new Map();
const _sparklineHistoryFailedAt = new Map();
async function _commoditySparkline(symbol, quote) {
const firstBarAt = quote.chart?.chart?.result?.[0]?.timestamp?.[0] ?? quote.lastTradeAt;
const cacheable = Number.isFinite(firstBarAt);
let cached = cacheable ? _sparklineHistories.get(symbol) : null;
if (!cached || cached.firstBarAt !== firstBarAt) {
const backingOff = Date.now() - (_sparklineHistoryFailedAt.get(symbol) ?? -Infinity) < SPARKLINE_HISTORY_RETRY_SECONDS * 1000;
let history = null;
if (!backingOff) {
await sleep(150);
history = await _fetchYahooChartOnce(symbol, SPARKLINE_HISTORY_QUERY);
if (history) _sparklineHistoryFailedAt.delete(symbol);
else _sparklineHistoryFailedAt.set(symbol, Date.now());
}
cached = { firstBarAt, chart: history?.chart ?? null };
if (history && cacheable) _sparklineHistories.set(symbol, cached);
if (!history) logThrottled('warn', `market-sparkline-history:${symbol}`, `[Market] ${symbol} sparkline history unavailable; publishing today's bars only`);
}
const { sparkline, ...labels } = buildQuoteSparkline(quote.chart, cached.chart);
return { sparkline: sparkline.map((value) => (value !== 0 ? Number(value.toPrecision(7)) : value)), ...labels };
}
function fetchYahooChartDirect(symbol, query = '') {
return _fetchYahooChartOnce(symbol, query)
.then((quote) => (!query && quote?.awaitingSessionOpen ? _withPreviousSessionChange(symbol, quote) : quote));
}
function _fetchYahooChartOnce(symbol, query = '') {
return _fetchYahooChartNoProxy(symbol, query).then((result) => {
if (result) return result;
if (!PROXY_URL) return null;
if (Date.now() < _yahooConnectProxyCooldownUntil) return null;
const proxy = { ...parseProxyUrl(PROXY_URL), tls: true };
const url = `https://query1.finance.yahoo.com/v8/finance/chart/${encodeURIComponent(symbol)}${query}`;
return fetchViaProxy(url, proxy).then((resp) => {
if (!resp?.ok) {
_yahooConnectProxyFailCount++;
if (_yahooConnectProxyFailCount >= 5) {
_yahooConnectProxyCooldownUntil = Date.now() + _YAHOO_PROXY_COOLDOWN_MS;
_yahooConnectProxyFailCount = 0;
logThrottled('warn', 'market-yahoo-proxy-cooldown', '[Market] Yahoo CONNECT proxy cooldown 5min after 5 failures');
}
return null;
}
_yahooConnectProxyFailCount = 0;
return _parseYahooChartJson(resp.body);
}).catch(() => null);
});
}
// Yahoo quoteSummary now requires a cookie + crumb session. The client caches
// that session, refreshes it once on 401, then cools the whole route down so a
// single auth failure cannot produce 12 direct + 12 proxy retries per cycle.
// The Decodo fallback performs the same authenticated handshake through curl;
// proxy credentials remain process-local and are never included in logs.
const _yahooQuoteSummaryClient = new YahooQuoteSummaryClient({
userAgent: CHROME_UA,
resolveProxyString,
// Yahoo's quote fundamentals cache is populated per residential exit IP, so a
// 200 response can omit trailingPE for a stable subset of ETFs no matter how
// often the same exit is retried. Rotation is the only recovery.
resolveProxyStringForAttempt,
cooldownMs: _YAHOO_PROXY_COOLDOWN_MS,
logger: {
warn(message, { transport }) {
logThrottled('warn', `sector-yahoo-auth-${transport}`, message);
},
},
});
function fetchYahooQuoteSummary(symbol) {
return _yahooQuoteSummaryClient.fetch(symbol);
}
function parseSectorValuation(raw) {
if (!raw) return null;
const num = (v) => typeof v === 'number' && Number.isFinite(v) ? v : null;
const tpe = num(typeof raw.trailingPE === 'string' ? parseFloat(raw.trailingPE) : raw.trailingPE);
const fpe = num(typeof raw.forwardPE === 'string' ? parseFloat(raw.forwardPE) : raw.forwardPE);
const beta = num(typeof raw.beta === 'string' ? parseFloat(raw.beta) : raw.beta);
const ytd = num(typeof raw.ytdReturn === 'string' ? parseFloat(raw.ytdReturn) : raw.ytdReturn);
const y3 = num(typeof raw.threeYearReturn === 'string' ? parseFloat(raw.threeYearReturn) : raw.threeYearReturn);
const y5 = num(typeof raw.fiveYearReturn === 'string' ? parseFloat(raw.fiveYearReturn) : raw.fiveYearReturn);
if (tpe === null && fpe === null) return null;
return { trailingPE: tpe, forwardPE: fpe, beta, ytdReturn: ytd, threeYearReturn: y3, fiveYearReturn: y5 };
}
function fetchFinnhubQuoteDirect(symbol, apiKey) {
return new Promise((resolve) => {
const url = `https://finnhub.io/api/v1/quote?symbol=${encodeURIComponent(symbol)}`;
const req = https.get(url, {
signal: AbortSignal.timeout(10000),
headers: { 'User-Agent': CHROME_UA, Accept: 'application/json', 'X-Finnhub-Token': apiKey },
timeout: 10000,
}, (resp) => {
if (resp.statusCode !== 200) {
resp.resume();
return resolve(null);
}
let body = '';
resp.on('data', (chunk) => { body += chunk; });
resp.on('end', () => {
try {
const data = JSON.parse(body);
if (data.c === 0 && data.h === 0 && data.l === 0) return resolve(null);
resolve({ price: data.c, changePercent: data.dp });
} catch { resolve(null); }
});
});
req.on('error', () => resolve(null));
req.on('timeout', () => { req.destroy(); resolve(null); });
});
}
function sleep(ms) { return new Promise((r) => setTimeout(r, ms)); }
// #4922d closed-market equity gate. Last quote count published by
// seedMarketQuotes — reused to refresh seed-meta freshness while skipping.
let _lastEquityQuoteCount = 0;
let _lastYahooMarketRefreshAt = 0;
// Log once per open↔closed transition, not every 5-minute cycle.
let _equityGateLoggedClosed = false;
const CHINA_COUNTRY_STOCK_SYMBOL = '000001.SS';
// When every tracked equity market is on a non-trading day, skip the equity
// fetch+publish and instead
// keep the last-good equity and companion keys alive: extend their TTLs and
// refresh seed-meta:market:stocks fetchedAt so /api/health (maxStaleMin 30)
// stays green across a 60h+ weekend. Returns true when last-good was
// preserved; false means the keys are missing/expired and the caller must
// fall back to a real fetch to repopulate.
async function maintainClosedMarketEquityKeys() {
const { CHINA_COUNTRY_STOCK_INDEX_KEY } = await chinaCountryStockIndexHelpersPromise;
return maintainClosedMarketEquityKeysWithDeps({
marketSymbols: MARKET_SYMBOLS,
marketSeedTtl: MARKET_SEED_TTL,
lastEquityQuoteCount: _lastEquityQuoteCount,
upstashExpire,
upstashGet,
upstashSet,
nowMs: () => Date.now(),
preserveKeys: [CHINA_COUNTRY_STOCK_INDEX_KEY],
});
}
async function writeChinaCountryStockIndex() {
const {
CHINA_COUNTRY_STOCK_INDEX_KEY,
buildCountryStockIndexSnapshotFromCloses,
} = await chinaCountryStockIndexHelpersPromise;
const chart = await fetchYahooChartDirect(CHINA_COUNTRY_STOCK_SYMBOL, '?range=1mo&interval=1d');
const snapshot = buildCountryStockIndexSnapshotFromCloses(chart?.sparkline, 'CNY');
if (!snapshot) throw new Error('China country index returned insufficient daily closes');
const written = await upstashSet(CHINA_COUNTRY_STOCK_INDEX_KEY, snapshot, MARKET_SEED_TTL);
if (!written) throw new Error('China country index Redis write failed');
}
async function seedMarketQuotes() {
const previousPayloadPromise = envelopeRead('market:stocks-bootstrap:v1');
const freshQuotes = [];
const finnhubSymbols = MARKET_SYMBOLS.filter((s) => !YAHOO_ONLY.has(s));
const yahooSymbols = MARKET_SYMBOLS.filter((s) => YAHOO_ONLY.has(s));
if (FINNHUB_API_KEY && finnhubSymbols.length > 0) {
const results = await Promise.all(finnhubSymbols.map((s) => fetchFinnhubQuoteDirect(s, FINNHUB_API_KEY)));
for (let i = 0; i < finnhubSymbols.length; i++) {
const r = results[i];
const symbol = finnhubSymbols[i];
const meta = MARKET_META.get(symbol);
if (r) freshQuotes.push({ symbol, name: meta?.name || symbol, display: meta?.display || symbol, price: r.price, change: r.changePercent, sparkline: [] });
}
}
const missedFinnhub = FINNHUB_API_KEY
? finnhubSymbols.filter((s) => !freshQuotes.some((q) => q.symbol === s))
: finnhubSymbols;
const yahooPlan = planYahooRefresh({
mandatoryYahooSymbols: yahooSymbols,
everyCycleSymbols: MARKET_AUXILIARY_SYMBOLS.filter((s) => YAHOO_ONLY.has(s)),
missedPrimarySymbols: missedFinnhub,
nowMs: Date.now(),
lastRefreshAt: _lastYahooMarketRefreshAt,
refreshIntervalMs: MARKET_YAHOO_REFRESH_INTERVAL_MS,
});
const allYahoo = yahooPlan.symbols;
if (yahooPlan.due) _lastYahooMarketRefreshAt = Date.now();
const freshCountBeforeYahoo = freshQuotes.length;
for (const s of allYahoo) {
if (freshQuotes.some((q) => q.symbol === s)) continue;
const yahoo = await fetchYahooChartDirect(s);
const meta = MARKET_META.get(s);
if (yahoo) freshQuotes.push({ symbol: s, name: meta?.name || s, display: meta?.display || s, price: yahoo.price, change: yahoo.change, sparkline: yahoo.sparkline });
await sleep(150);
}
if (freshQuotes.length === 0) {
console.warn('[Market] No quotes fetched — skipping Redis write');
return 0;
}
const previousPayload = await previousPayloadPromise;
const previousQuotes = Array.isArray(previousPayload?.quotes) ? previousPayload.quotes : [];
const quotes = mergeLastGoodQuotes(MARKET_SYMBOLS, freshQuotes, previousQuotes);
const retainedCount = quotes.length - freshQuotes.length;
const yahooSuccessCount = freshQuotes.length - freshCountBeforeYahoo;
const coveredByYahoo = finnhubSymbols.every((s) => quotes.some((q) => q.symbol === s));
const skipped = !FINNHUB_API_KEY && !coveredByYahoo;
const redisKey = `market:quotes:v1:${[...MARKET_SYMBOLS].sort().join(',')}`;
// Compute once and thread through every write below so the envelopes'
// _seed.fetchedAt and seed-meta.fetchedAt agree for this one publish,
// instead of each awaited round-trip sampling Date.now() independently.
const fetchedAt = Date.now();
const payload = {
quotes,
finnhubSkipped: skipped,
skipReason: skipped ? 'FINNHUB_API_KEY not configured' : '',
rateLimited: false,
asOf: resolveMergedQuotesAsOf(freshQuotes, quotes, previousPayload?.asOf, fetchedAt),
};
const ok = await envelopeWrite(redisKey, payload, MARKET_SEED_TTL, { fetchedAt, recordCount: quotes.length, sourceVersion: 'market-stocks' });
// Bootstrap-friendly fixed key — frontend hydrates from /api/bootstrap without RPC
const ok2 = await envelopeWrite('market:stocks-bootstrap:v1', payload, MARKET_SEED_TTL, { fetchedAt, recordCount: quotes.length, sourceVersion: 'market-stocks' });
const ok3 = await upstashSet('seed-meta:market:stocks', { fetchedAt, recordCount: quotes.length }, 604800);
if (freshQuotes.some((quote) => quote.symbol === CHINA_COUNTRY_STOCK_SYMBOL)) {
try {
await writeChinaCountryStockIndex();
} catch (err) {
console.warn(`[Market] China country index refresh failed: ${err.message}`);
}
}
_lastEquityQuoteCount = quotes.length;
console.log(`[Market] Seeded ${quotes.length}/${MARKET_SYMBOLS.length} quotes (${freshQuotes.length} fresh, ${retainedCount} retained; Yahoo ${yahooSuccessCount}/${allYahoo.length}, cadence ${MARKET_YAHOO_REFRESH_INTERVAL_MS / 60000}min; redis: ${ok && ok2 && ok3 ? 'OK' : 'PARTIAL'})`);
const movingStocks = quotes.filter(q => Math.abs(q.change ?? 0) >= 5).sort((a, b) => Math.abs(b.change) - Math.abs(a.change));
for (const q of movingStocks.slice(0, 3)) {
const pct = Math.round(q.change);
const dir = q.change < 0 ? 'decline' : 'surge';
const severity = Math.abs(q.change) >= 10 ? 'critical' : 'high';
publishNotificationEvent({
eventType: 'market_alert',
payload: {
title: `${q.symbol}: ${pct > 0 ? '+' : ''}${pct}% ${dir}`,
source: 'Equity Market',
coalesceKey: marketAlertCoalesceKey('equity', q.symbol, dir, severity),
...marketMoveFields('equity', q),
},
severity,
variant: undefined,
dedupTtl: 3600,
}).catch(e => console.warn('[Notify] Market stock publish error:', e?.message));
}
return quotes.length;
}
async function seedCommodityQuotes() {
const quotes = [];
const missing = [];
const toQuote = async (symbol, meta, yahoo) => ({
symbol, name: meta.name, display: meta.display, price: yahoo.price, change: yahoo.change,
...(await _commoditySparkline(symbol, yahoo)),
...crudeOilQuoteProvenance(symbol, 'Yahoo Finance', 'futures', yahoo.lastTradeAt * 1000),
});
for (const s of COMMODITY_SYMBOLS) {
const meta = COMMODITY_META.get(s) || { name: s, display: s };
const yahoo = await fetchYahooChartDirect(s);
if (yahoo) quotes.push(await toQuote(s, meta, yahoo));
else missing.push(s);
await sleep(150);
}
// Retry symbols that failed (Yahoo 429 recovery)
if (missing.length > 0) {
await sleep(3000);
for (const s of missing) {
const meta = COMMODITY_META.get(s);
const yahoo = await fetchYahooChartDirect(s);
if (yahoo) quotes.push(await toQuote(s, meta, yahoo));
await sleep(200);
}
}
if (quotes.length === 0) {
console.warn('[Market] No commodity quotes fetched — extending existing key TTL, skipping write');
try { await upstashExpire('market:commodities-bootstrap:v1', MARKET_SEED_TTL); } catch {}
return 0;
}
const payload = { quotes };
const redisKey = `market:commodities:v1:${[...COMMODITY_SYMBOLS].sort().join(',')}`;
// Compute once and thread through every write below so the envelopes'
// _seed.fetchedAt and seed-meta.fetchedAt agree for this one publish,
// instead of each awaited round-trip sampling Date.now() independently.
const fetchedAt = Date.now();
const ok = await envelopeWrite(redisKey, payload, MARKET_SEED_TTL, { fetchedAt, recordCount: quotes.length, sourceVersion: 'market-commodities' });
// Also write under market:quotes:v1: key — the frontend routes commodities through
// listMarketQuotes RPC, which constructs this key pattern (not market:commodities:v1:)
const quotesKey = `market:quotes:v1:${[...COMMODITY_SYMBOLS].sort().join(',')}`;
const quotesPayload = { quotes, finnhubSkipped: false, skipReason: '', rateLimited: false };
const ok2 = await envelopeWrite(quotesKey, quotesPayload, MARKET_SEED_TTL, { fetchedAt, recordCount: quotes.length, sourceVersion: 'market-commodities' });
// Bootstrap-friendly fixed key — frontend hydrates from /api/bootstrap without RPC
const ok3 = await envelopeWrite('market:commodities-bootstrap:v1', quotesPayload, MARKET_SEED_TTL, { fetchedAt, recordCount: quotes.length, sourceVersion: 'market-commodities' });
const ok4 = await upstashSet('seed-meta:market:commodities', { fetchedAt, recordCount: quotes.length }, 604800);
console.log(`[Market] Seeded ${quotes.length}/${COMMODITY_SYMBOLS.length} commodities (redis: ${ok && ok2 && ok3 && ok4 ? 'OK' : 'PARTIAL'})`);
const movingCommodities = quotes.filter(q => Math.abs(q.change ?? 0) >= 5).sort((a, b) => Math.abs(b.change) - Math.abs(a.change));
for (const q of movingCommodities.slice(0, 3)) {
const pct = Math.round(q.change);
const dir = q.change < 0 ? 'decline' : 'surge';
const severity = Math.abs(q.change) >= 10 ? 'critical' : 'high';
publishNotificationEvent({
eventType: 'market_alert',
payload: {
title: `${q.name || q.symbol}: ${pct > 0 ? '+' : ''}${pct}% ${dir}`,
source: 'Commodity Market',
coalesceKey: marketAlertCoalesceKey('commodity', q.symbol || q.name, dir, severity),
...marketMoveFields('commodity', q),
},
severity,
variant: undefined,
dedupTtl: 3600,
}).catch(e => console.warn('[Notify] Commodity publish error:', e?.message));
}
return quotes.length;
}
async function seedSectorSummary() {
const sectors = [];
if (FINNHUB_API_KEY) {
const results = await Promise.all(SECTOR_SYMBOLS.map((s) => fetchFinnhubQuoteDirect(s, FINNHUB_API_KEY)));
for (let i = 0; i < SECTOR_SYMBOLS.length; i++) {
const r = results[i];
if (r) sectors.push({ symbol: SECTOR_SYMBOLS[i], name: SECTOR_SYMBOLS[i], change: r.changePercent });
}
}
if (sectors.length === 0) {
for (const s of SECTOR_SYMBOLS) {
const yahoo = await fetchYahooChartDirect(s);
if (yahoo) sectors.push({ symbol: s, name: s, change: yahoo.change });
await sleep(150);
}
}
if (sectors.length === 0) {
console.warn('[Market] No sector data fetched — skipping Redis write');
return 0;
}
const {
valuations,
valuationSources,
valuationCount: valCount,
unavailableSymbols,
valuationDiagnostics,
currentValuationCount,
lastGoodFetchedAt,
lastGoodMetricsUsed,
lastGoodValuationSymbols,
} = await collectSectorValuations({
symbols: SECTOR_SYMBOLS,
fetchValue: fetchYahooQuoteSummary,
fetchValueDetailed: (symbol, options) => _yahooQuoteSummaryClient.fetchDetailed(symbol, options),
parseValue: parseSectorValuation,
sleepFn: sleep,
v7UserAgent: CHROME_UA,
v7ResolveProxyString: resolveProxyString,
v7Client: _yahooQuoteSummaryClient,
upstashGet,
upstashSet,
});
const valuationCoverage = buildSectorValuationCoverage({
valuationCount: valCount,
expectedCount: SECTOR_SYMBOLS.length,
fetchedAt: Date.now(),
sources: valuationSources,
unavailableSymbols,
valuationDiagnostics,
currentValuationCount,
lastGoodFetchedAt,
lastGoodMetricsUsed,
lastGoodValuationSymbols,
});
const { payload, meta: sectorMeta } = buildSectorValuationPublication({
sectors,
valuations,
valuationCoverage,
});
const ok = await envelopeWrite('market:sectors:v2', payload, MARKET_SEED_TTL, { recordCount: sectors.length, sourceVersion: 'market-sectors' });
const quotesKey = `market:quotes:v1:${[...SECTOR_SYMBOLS].sort().join(',')}`;
const sectorQuotes = sectors.map((s) => ({
symbol: s.symbol, name: s.name, display: s.name,
price: 0, change: s.change, sparkline: [],
}));
const quotesPayload = { quotes: sectorQuotes, finnhubSkipped: false, skipReason: '', rateLimited: false };
const ok2 = await envelopeWrite(quotesKey, quotesPayload, MARKET_SEED_TTL, { recordCount: sectorQuotes.length, sourceVersion: 'market-sectors' });
const persistedSectorMeta = buildSectorSeedMeta(sectorMeta, ok);
const ok3 = await upstashSet('seed-meta:market:sectors', persistedSectorMeta, 604800);
// valCount includes records replayed from the last-good snapshot, so report
// the live count alongside it: "12/12 (partial)" during a total upstream
// outage reads as healthy coverage to anyone scanning the logs.
const liveCount = currentValuationCount == null ? valCount : currentValuationCount;
const valuationSummary = liveCount === valCount
? `${valCount}/${SECTOR_SYMBOLS.length} valuations`
: `${valCount}/${SECTOR_SYMBOLS.length} valuations (${liveCount} live, ${valCount - liveCount} stale)`;
console.log(`[Market] Seeded ${sectors.length}/${SECTOR_SYMBOLS.length} sectors, ${valuationSummary} (${valuationCoverage.sourceStatus}; redis: ${ok && ok2 && ok3 ? 'OK' : 'PARTIAL'})`);
return sectors.length;
}
// Gulf Quotes — Yahoo Finance (14 symbols: indices, currencies, oil)
const GULF_SYMBOLS = [
{ symbol: '^TASI.SR', name: 'Tadawul All Share', country: 'Saudi Arabia', flag: '\u{1F1F8}\u{1F1E6}', type: 'index' },
{ symbol: 'DFMGI.AE', name: 'Dubai Financial Market', country: 'UAE', flag: '\u{1F1E6}\u{1F1EA}', type: 'index' },
{ symbol: 'UAE', name: 'Abu Dhabi (iShares)', country: 'UAE', flag: '\u{1F1E6}\u{1F1EA}', type: 'index' },
{ symbol: 'QAT', name: 'Qatar (iShares)', country: 'Qatar', flag: '\u{1F1F6}\u{1F1E6}', type: 'index' },
{ symbol: 'GULF', name: 'Gulf Dividend (WisdomTree)', country: 'Kuwait', flag: '\u{1F1F0}\u{1F1FC}', type: 'index' },
{ symbol: '^MSM', name: 'Muscat MSM 30', country: 'Oman', flag: '\u{1F1F4}\u{1F1F2}', type: 'index' },
{ symbol: 'SARUSD=X', name: 'Saudi Riyal', country: 'Saudi Arabia', flag: '\u{1F1F8}\u{1F1E6}', type: 'currency' },
{ symbol: 'AEDUSD=X', name: 'UAE Dirham', country: 'UAE', flag: '\u{1F1E6}\u{1F1EA}', type: 'currency' },
{ symbol: 'QARUSD=X', name: 'Qatari Riyal', country: 'Qatar', flag: '\u{1F1F6}\u{1F1E6}', type: 'currency' },
{ symbol: 'KWDUSD=X', name: 'Kuwaiti Dinar', country: 'Kuwait', flag: '\u{1F1F0}\u{1F1FC}', type: 'currency' },
{ symbol: 'BHDUSD=X', name: 'Bahraini Dinar', country: 'Bahrain', flag: '\u{1F1E7}\u{1F1ED}', type: 'currency' },
{ symbol: 'OMRUSD=X', name: 'Omani Rial', country: 'Oman', flag: '\u{1F1F4}\u{1F1F2}', type: 'currency' },
{ symbol: 'CL=F', name: 'WTI Crude', country: '', flag: '\u{1F6E2}\u{FE0F}', type: 'oil' },
{ symbol: 'BZ=F', name: 'Brent Crude', country: '', flag: '\u{1F6E2}\u{FE0F}', type: 'oil' },
];
const GULF_SEED_TTL = 5400; // 90min — survives 1 missed cycle
async function seedGulfQuotes() {
const quotes = [];
for (const meta of GULF_SYMBOLS) {
const yahoo = await fetchYahooChartDirect(meta.symbol);
if (yahoo) {
quotes.push({
symbol: meta.symbol, name: meta.name, country: meta.country,
flag: meta.flag, type: meta.type,
price: yahoo.price, change: +(yahoo.change).toFixed(2), sparkline: yahoo.sparkline,
});
}
await sleep(150);
}
if (quotes.length === 0) { console.warn('[Gulf] No quotes fetched — skipping'); return 0; }
const payload = { quotes, rateLimited: false };
const ok1 = await envelopeWrite('market:gulf-quotes:v1', payload, GULF_SEED_TTL, { recordCount: quotes.length, sourceVersion: 'market-gulf' });
const ok2 = await upstashSet('seed-meta:market:gulf-quotes', { fetchedAt: Date.now(), recordCount: quotes.length }, 604800);
console.log(`[Gulf] Seeded ${quotes.length}/${GULF_SYMBOLS.length} quotes (redis: ${ok1 && ok2 ? 'OK' : 'PARTIAL'})`);
return quotes.length;
}
// ETF Flows — Yahoo Finance (10 BTC spot ETFs)
const ETF_LIST = [
{ ticker: 'IBIT', issuer: 'BlackRock' }, { ticker: 'FBTC', issuer: 'Fidelity' },
{ ticker: 'ARKB', issuer: 'ARK/21Shares' }, { ticker: 'BITB', issuer: 'Bitwise' },
{ ticker: 'GBTC', issuer: 'Grayscale' }, { ticker: 'HODL', issuer: 'VanEck' },
{ ticker: 'BRRR', issuer: 'Valkyrie' }, { ticker: 'EZBC', issuer: 'Franklin' },
{ ticker: 'BTCO', issuer: 'Invesco' }, { ticker: 'BTCW', issuer: 'WisdomTree' },
];
const ETF_SEED_TTL = 5400; // 90min
function parseEtfChart(chart, ticker, issuer) {
const result = chart?.chart?.result?.[0];
if (!result) return null;
const closes = (result.indicators?.quote?.[0]?.close || []).filter((v) => v != null);
const volumes = (result.indicators?.quote?.[0]?.volume || []).filter((v) => v != null);
if (closes.length < 2) return null;
const price = closes[closes.length - 1];
const prev = closes[closes.length - 2];
const priceChange = prev ? ((price - prev) / prev) * 100 : 0;
const vol = volumes.length > 0 ? volumes[volumes.length - 1] : 0;
const avgVol = volumes.length > 1 ? volumes.slice(0, -1).reduce((a, b) => a + b, 0) / (volumes.length - 1) : vol;
const volumeRatio = avgVol > 0 ? vol / avgVol : 1;
const direction = priceChange > 0.1 ? 'inflow' : priceChange < -0.1 ? 'outflow' : 'neutral';
return { ticker, issuer, price: +price.toFixed(2), priceChange: +priceChange.toFixed(2), volume: vol, avgVolume: Math.round(avgVol), volumeRatio: +volumeRatio.toFixed(2), direction, estFlow: Math.round(vol * price * (priceChange > 0 ? 1 : -1) * 0.1) };
}
async function seedEtfFlows() {
const etfs = [];
for (const { ticker, issuer } of ETF_LIST) {
try {
const raw = await new Promise((resolve) => {
const url = `https://query1.finance.yahoo.com/v8/finance/chart/${encodeURIComponent(ticker)}?range=5d&interval=1d`;
const req = https.get(url, { signal: AbortSignal.timeout(10000), headers: { 'User-Agent': CHROME_UA, Accept: 'application/json' }, timeout: 10000 }, (resp) => {
if (resp.statusCode !== 200) { resp.resume(); return resolve(null); }
let body = '';
resp.on('data', (chunk) => { body += chunk; });
resp.on('end', () => { try { resolve(JSON.parse(body)); } catch { resolve(null); } });
});
req.on('error', () => resolve(null));
req.on('timeout', () => { req.destroy(); resolve(null); });
});
const parsed = raw ? parseEtfChart(raw, ticker, issuer) : null;
if (parsed) etfs.push(parsed);
} catch {}
await sleep(150);
}
if (etfs.length === 0) { console.warn('[ETF] No data fetched — skipping'); return 0; }
const totalVolume = etfs.reduce((s, e) => s + e.volume, 0);
const totalEstFlow = etfs.reduce((s, e) => s + e.estFlow, 0);
const payload = {
timestamp: new Date().toISOString(),
summary: { etfCount: etfs.length, totalVolume, totalEstFlow, netDirection: totalEstFlow > 0 ? 'NET INFLOW' : totalEstFlow < 0 ? 'NET OUTFLOW' : 'NEUTRAL', inflowCount: etfs.filter((e) => e.direction === 'inflow').length, outflowCount: etfs.filter((e) => e.direction === 'outflow').length },
etfs, rateLimited: false,
};
const ok1 = await envelopeWrite('market:etf-flows:v1', payload, ETF_SEED_TTL, { recordCount: etfs.length, sourceVersion: 'market-etf-flows' });
const ok2 = await upstashSet('seed-meta:market:etf-flows', { fetchedAt: Date.now(), recordCount: etfs.length }, 604800);
console.log(`[ETF] Seeded ${etfs.length}/${ETF_LIST.length} ETFs (redis: ${ok1 && ok2 ? 'OK' : 'PARTIAL'})`);
return etfs.length;
}
// Crypto Quotes — CoinGecko → CoinPaprika fallback
const _cryptoCfg = requireShared('crypto.json');
const CRYPTO_IDS = _cryptoCfg.ids;
const CRYPTO_META = _cryptoCfg.meta;
const CRYPTO_PAPRIKA_MAP = _cryptoCfg.coinpaprika;
const CRYPTO_SEED_TTL = 7200; // 2h — 1h buffer over 5min cron cadence (was 1h = 55min buffer)
// Shared CoinPaprika tickers fetcher — direct first, PROXY_URL fallback.
// Cached per ticker for 5 min so crypto, stablecoins, sectors, and token panels
// that run in the same cycle can share overlap without fetching the full catalog.
const _paprikaTickerCache = new Map();
const _PAPRIKA_CACHE_MS = 5 * 60 * 1000;
const _PAPRIKA_FETCH_CONCURRENCY = 4;
async function _fetchCoinPaprikaTickerById(id) {
const url = `https://api.coinpaprika.com/v1/tickers/${encodeURIComponent(id)}?quotes=USD`;
const direct = await cyberHttpGetJson(url, { Accept: 'application/json' }, 15000);
if (direct && typeof direct === 'object' && direct.id) return direct;
if (!PROXY_URL) throw new Error(`CoinPaprika ${id} direct failed and no PROXY_URL configured`);
const proxy = { ...parseProxyUrl(PROXY_URL), tls: true };
const resp = await fetchViaProxy(url, proxy);
if (!resp?.ok) throw new Error(`CoinPaprika ${id} proxy HTTP ${resp?.status || 'unavailable'}`);
const data = JSON.parse(resp.body);
if (!data || typeof data !== 'object' || !data.id) throw new Error(`CoinPaprika ${id} proxy returned invalid ticker`);
return data;
}
async function _fetchCoinPaprikaTickersById(paprikaIds) {
const ids = [...new Set(paprikaIds.filter(Boolean))];
if (ids.length === 0) return [];
const now = Date.now();
const tickers = [];
const misses = [];
for (const id of ids) {
const cached = _paprikaTickerCache.get(id);
if (cached && now - cached.cachedAt < _PAPRIKA_CACHE_MS) {
tickers.push(cached.ticker);
} else {
misses.push(id);
}
}
const results = await allSettledWithConcurrency(misses, _PAPRIKA_FETCH_CONCURRENCY, async (id) => {
const ticker = await _fetchCoinPaprikaTickerById(id);
_paprikaTickerCache.set(id, { ticker, cachedAt: Date.now() });
return ticker;
});
const failures = [];
for (let i = 0; i < results.length; i += 1) {
const result = results[i];
if (result.status === 'fulfilled') {
tickers.push(result.value);
} else {
failures.push(result.reason);
console.warn(`[CoinPaprika] Skipping ${misses[i]}: ${result.reason?.message || result.reason}`);
}
}
if (tickers.length === 0 && failures.length > 0) {
throw new Error(`All ${failures.length} CoinPaprika ticker request(s) failed`);
}
return tickers;
}
async function allSettledWithConcurrency(items, concurrency, mapper) {
const results = new Array(items.length);
let nextIndex = 0;
const workerCount = Math.min(Math.max(1, concurrency), items.length);
await Promise.all(Array.from({ length: workerCount }, async () => {
while (nextIndex < items.length) {
const index = nextIndex;
nextIndex += 1;
try {
results[index] = { status: 'fulfilled', value: await mapper(items[index], index) };
} catch (reason) {
results[index] = { status: 'rejected', reason };
}
}
}));
return results;
}
async function fetchCryptoCoinPaprika() {
const paprikaIds = CRYPTO_IDS.map((id) => CRYPTO_PAPRIKA_MAP[id]).filter(Boolean);
const data = await _fetchCoinPaprikaTickersById(paprikaIds);
const reverseMap = Object.fromEntries(Object.entries(CRYPTO_PAPRIKA_MAP).map(([g, p]) => [p, g]));
return data.map((t) => ({
id: reverseMap[t.id] || t.id, current_price: t.quotes.USD.price,
price_change_percentage_24h: t.quotes.USD.percent_change_24h,
sparkline_in_7d: undefined, symbol: t.symbol.toLowerCase(), name: t.name,
}));
}
// CoinGecko's free Demo and paid Pro plans share the `CG-` key prefix but use
// different hosts + auth headers (a Demo key on the Pro host 400s). Resolve the
// tier explicitly by which env var is set — Pro wins, else Demo, else keyless.
// Mirrors scripts/_seed-utils.mjs `coingeckoEndpoint()`; this relay is CommonJS
// and cannot import the .mjs helper, so the logic is duplicated.
function coingeckoEndpoint() {
const proKey = process.env.COINGECKO_API_KEY;
const demoKey = process.env.COINGECKO_DEMO_API_KEY;
const headers = { Accept: 'application/json' };
if (proKey) {
headers['x-cg-pro-api-key'] = proKey;
return { base: 'https://pro-api.coingecko.com/api/v3', headers };
}
if (demoKey) headers['x-cg-demo-api-key'] = demoKey;
return { base: 'https://api.coingecko.com/api/v3', headers };
}
async function seedCryptoQuotes() {
let data;
try {
const { base, headers } = coingeckoEndpoint();
const url = `${base}/coins/markets?vs_currency=usd&ids=${CRYPTO_IDS.join(',')}&order=market_cap_desc&sparkline=true&price_change_percentage=24h`;
data = await cyberHttpGetJson(url, headers, 15000);
if (!Array.isArray(data) || data.length === 0) throw new Error('CoinGecko returned no data');
} catch (err) {
console.warn(`[Crypto] CoinGecko failed: ${err.message} — trying CoinPaprika`);
try { data = await fetchCryptoCoinPaprika(); } catch (e2) { console.warn(`[Crypto] CoinPaprika also failed: ${e2.message} — skipping`); return 0; }
}
const quotes = [];
for (const id of CRYPTO_IDS) {
const coin = data.find((c) => c.id === id);
if (!coin) continue;
const meta = CRYPTO_META[id];
const prices = coin.sparkline_in_7d?.price;
quotes.push({ name: meta?.name || id, symbol: meta?.symbol || id.toUpperCase(), price: coin.current_price ?? 0, change: coin.price_change_percentage_24h ?? 0, sparkline: prices && prices.length > 24 ? prices.slice(-48) : (prices || []) });
}
if (quotes.length === 0 || quotes.every((q) => q.price === 0)) { console.warn('[Crypto] No valid quotes — skipping'); return 0; }
const ok1 = await envelopeWrite('market:crypto:v1', { quotes }, CRYPTO_SEED_TTL, { recordCount: quotes.length, sourceVersion: 'market-crypto' });
const ok2 = await upstashSet('seed-meta:market:crypto', { fetchedAt: Date.now(), recordCount: quotes.length }, 604800);
console.log(`[Crypto] Seeded ${quotes.length}/${CRYPTO_IDS.length} quotes (redis: ${ok1 && ok2 ? 'OK' : 'PARTIAL'})`);
const movingCrypto = quotes.filter(q => Math.abs(q.change ?? 0) >= 10).sort((a, b) => Math.abs(b.change) - Math.abs(a.change));
for (const q of movingCrypto.slice(0, 3)) {
const pct = Math.round(q.change);
const dir = q.change < 0 ? 'decline' : 'surge';
const severity = Math.abs(q.change) >= 20 ? 'critical' : 'high';
publishNotificationEvent({
eventType: 'market_alert',
payload: {
title: `${q.symbol}: ${pct > 0 ? '+' : ''}${pct}% ${dir}`,
source: 'Crypto Market',
coalesceKey: marketAlertCoalesceKey('crypto', q.symbol || q.name, dir, severity),
...marketMoveFields('crypto', q),
},
severity,
variant: undefined,
dedupTtl: 3600,
}).catch(e => console.warn('[Notify] Crypto publish error:', e?.message));
}
return quotes.length;
}
// Stablecoin Markets — CoinGecko → CoinPaprika fallback
//
// This is the BACKUP writer for market:stablecoins:v1; the standalone
// scripts/seed-stablecoin-markets.mjs is the primary. Both, plus the RPC
// handler that fills gaps the snapshot does not carry, now read one shared
// config — private copies here meant the backup could seed a different coin
// set, or classify the same price differently, than the primary. (#6308)
const _stablecoinCfg = requireShared('stablecoins.json');
const STABLECOIN_IDS = _stablecoinCfg.ids.join(',');
const STABLECOIN_PAPRIKA_MAP = _stablecoinCfg.coinpaprika;
// Row shaping and peg classification live in ONE module shared with the
// primary seeder and the RPC gap path — a private variant here means the
// stored value depends on which writer ran last. (#6319, extends #6308)
const { classifyStablecoin } = requireShared('stablecoin-classifier.cjs');
const STABLECOIN_SEED_TTL = 7200; // 2h — 1h buffer over 5min cron cadence (was 1h = 55min buffer)
async function fetchStablecoinCoinPaprika() {
const ids = STABLECOIN_IDS.split(',');
const paprikaIds = ids.map((id) => STABLECOIN_PAPRIKA_MAP[id]).filter(Boolean);
const data = await _fetchCoinPaprikaTickersById(paprikaIds);
const reverseMap = Object.fromEntries(Object.entries(STABLECOIN_PAPRIKA_MAP).map(([g, p]) => [p, g]));
return data.map((t) => ({
id: reverseMap[t.id] || t.id, current_price: t.quotes.USD.price,
price_change_percentage_24h: t.quotes.USD.percent_change_24h,
price_change_percentage_7d_in_currency: t.quotes.USD.percent_change_7d,
market_cap: t.quotes.USD.market_cap, total_volume: t.quotes.USD.volume_24h,
symbol: t.symbol.toLowerCase(), name: t.name, image: '',
}));
}
async function seedStablecoinMarkets() {
let data;
try {
const { base, headers } = coingeckoEndpoint();
const url = `${base}/coins/markets?vs_currency=usd&ids=${STABLECOIN_IDS}&order=market_cap_desc&sparkline=false&price_change_percentage=7d`;
data = await cyberHttpGetJson(url, headers, 15000);
if (!Array.isArray(data) || data.length === 0) throw new Error('CoinGecko returned no data');
} catch (err) {
console.warn(`[Stablecoin] CoinGecko failed: ${err.message} — trying CoinPaprika`);
try { data = await fetchStablecoinCoinPaprika(); } catch (e2) { console.warn(`[Stablecoin] CoinPaprika also failed: ${e2.message} — skipping`); return 0; }
}
const stablecoins = data.map((coin) => classifyStablecoin(coin));
const totalMarketCap = stablecoins.reduce((s, c) => s + c.marketCap, 0);
const totalVolume24h = stablecoins.reduce((s, c) => s + c.volume24h, 0);
const depeggedCount = stablecoins.filter((c) => c.pegStatus === 'DEPEGGED').length;
const payload = { timestamp: new Date().toISOString(), summary: { totalMarketCap, totalVolume24h, coinCount: stablecoins.length, depeggedCount, healthStatus: depeggedCount === 0 ? 'HEALTHY' : depeggedCount === 1 ? 'CAUTION' : 'WARNING' }, stablecoins };
const ok1 = await envelopeWrite('market:stablecoins:v1', payload, STABLECOIN_SEED_TTL, { recordCount: stablecoins.length, sourceVersion: 'market-stablecoins' });
const ok2 = await upstashSet('seed-meta:market:stablecoins', { fetchedAt: Date.now(), recordCount: stablecoins.length }, 604800);
console.log(`[Stablecoin] Seeded ${stablecoins.length} coins (redis: ${ok1 && ok2 ? 'OK' : 'PARTIAL'})`);
return stablecoins.length;
}
// Crypto Sectors Heatmap — CoinGecko sector averages
const _sectorsCfg = requireShared('crypto-sectors.json');
const SECTORS_LIST = _sectorsCfg.sectors;
const SECTORS_SEED_TTL = 7200; // 2h — 1h buffer over 5min cron cadence (was 1h = 55min buffer)
async function seedCryptoSectors() {
const allIds = [...new Set(SECTORS_LIST.flatMap((s) => s.tokens))];
let data;
try {
const { base, headers } = coingeckoEndpoint();
const url = `${base}/coins/markets?vs_currency=usd&ids=${allIds.join(',')}&order=market_cap_desc&sparkline=false&price_change_percentage=24h`;
data = await cyberHttpGetJson(url, headers, 15000);
if (!Array.isArray(data) || data.length === 0) throw new Error('CoinGecko returned no data');
} catch (err) {
console.warn(`[CryptoSectors] CoinGecko failed: ${err.message} — trying CoinPaprika`);
try {
const paprikaIds = allIds.map((id) => CRYPTO_PAPRIKA_MAP[id]).filter(Boolean);
const paprika = await _fetchCoinPaprikaTickersById(paprikaIds);
data = paprika.map((t) => {
const geckoId = Object.entries(CRYPTO_PAPRIKA_MAP).find(([, p]) => p === t.id)?.[0] || t.id;
return { id: geckoId, price_change_percentage_24h: t.quotes?.USD?.percent_change_24h ?? 0 };
});
if (!data.length) throw new Error('No matching tokens in CoinPaprika');
} catch (e2) { console.warn(`[CryptoSectors] CoinPaprika also failed: ${e2.message} — skipping`); return 0; }
}
const byId = new Map(data.map((c) => [c.id, c.price_change_percentage_24h]));
const sectors = SECTORS_LIST.map((sector) => {
const changes = sector.tokens.map((id) => byId.get(id)).filter((v) => typeof v === 'number' && Number.isFinite(v));
const change = changes.length > 0 ? changes.reduce((a, b) => a + b, 0) / changes.length : 0;
return { id: sector.id, name: sector.name, change };
});
const ok1 = await envelopeWrite('market:crypto-sectors:v1', { sectors }, SECTORS_SEED_TTL, { recordCount: sectors.length, sourceVersion: 'market-crypto-sectors' });
const ok2 = await upstashSet('seed-meta:market:crypto-sectors', { fetchedAt: Date.now(), recordCount: sectors.length }, 604800);
console.log(`[CryptoSectors] Seeded ${sectors.length} sectors (redis: ${ok1 && ok2 ? 'OK' : 'PARTIAL'})`);
return sectors.length;
}
// Token Panels — DeFi, AI, Other — single CoinGecko call writing 3 Redis keys
const _defiCfg = requireShared('defi-tokens.json');
const _aiCfg = requireShared('ai-tokens.json');
const _otherCfg = requireShared('other-tokens.json');
const TOKEN_PANELS_PAPRIKA_MAP = { ..._defiCfg.coinpaprika, ..._aiCfg.coinpaprika, ..._otherCfg.coinpaprika };
const TOKEN_PANELS_SEED_TTL = 3600; // 1h
function _mapTokens(ids, meta, byId) {
const tokens = [];
for (const id of ids) {
const coin = byId.get(id);
if (!coin) continue;
const m = meta[id];
tokens.push({
name: m?.name || coin.name || id,
symbol: m?.symbol || (coin.symbol || id).toUpperCase(),
price: coin.current_price ?? 0,
change24h: coin.price_change_percentage_24h ?? 0,
change7d: coin.price_change_percentage_7d_in_currency ?? 0,
});
}
return tokens;
}
async function fetchTokenPanelsCoinPaprika(allIds) {
const paprikaIds = allIds.map((id) => TOKEN_PANELS_PAPRIKA_MAP[id]).filter(Boolean);
const data = await _fetchCoinPaprikaTickersById(paprikaIds);
const reverseMap = Object.fromEntries(Object.entries(TOKEN_PANELS_PAPRIKA_MAP).map(([g, p]) => [p, g]));
return data.map((t) => ({
id: reverseMap[t.id] || t.id,
current_price: t.quotes.USD.price,
price_change_percentage_24h: t.quotes.USD.percent_change_24h,
price_change_percentage_7d_in_currency: t.quotes.USD.percent_change_7d,
symbol: t.symbol.toLowerCase(),
name: t.name,
}));
}
async function seedTokenPanels() {
const allIds = [...new Set([..._defiCfg.ids, ..._aiCfg.ids, ..._otherCfg.ids])];
let data;
try {
const { base, headers } = coingeckoEndpoint();
const url = `${base}/coins/markets?vs_currency=usd&ids=${allIds.join(',')}&order=market_cap_desc&sparkline=false&price_change_percentage=24h,7d`;
data = await cyberHttpGetJson(url, headers, 15000);
if (!Array.isArray(data) || data.length === 0) throw new Error('CoinGecko returned no data');
} catch (err) {
console.warn(`[TokenPanels] CoinGecko failed: ${err.message} — trying CoinPaprika`);
try { data = await fetchTokenPanelsCoinPaprika(allIds); } catch (e2) { console.warn(`[TokenPanels] CoinPaprika also failed: ${e2.message} — skipping`); return 0; }
}
const byId = new Map(data.map((c) => [c.id, c]));
const panels = [
{ key: 'market:defi-tokens:v1', payload: { tokens: _mapTokens(_defiCfg.ids, _defiCfg.meta, byId) }, sourceVersion: 'market-defi-tokens', label: 'DeFi' },
{ key: 'market:ai-tokens:v1', payload: { tokens: _mapTokens(_aiCfg.ids, _aiCfg.meta, byId) }, sourceVersion: 'market-ai-tokens', label: 'AI' },
{ key: 'market:other-tokens:v1', payload: { tokens: _mapTokens(_otherCfg.ids, _otherCfg.meta, byId) }, sourceVersion: 'market-other-tokens', label: 'Other' },
];
const total = panels.reduce((n, p) => n + p.payload.tokens.length, 0);
if (total === 0) {
console.warn('[TokenPanels] All panels empty after mapping — skipping Redis write to preserve cached data');
return 0;
}
// Write each panel ONLY when it mapped >=1 token. CoinGecko's
// /coins/markets?ids= endpoint returns only the IDs it has data for, so a
// partial response (e.g. it drops the DeFi+AI IDs but keeps Other) maps an
// individual panel to 0 tokens. Writing that empty panel would clobber the
// good cached payload with recordCount=0 — blanking the UI panel AND tripping
// the seed-contract probe's minRecords:1 floor (false 503). Skip the write and
// extend the existing key's TTL so the last-good panel is preserved instead.
const results = [];
for (const p of panels) {
if (p.payload.tokens.length === 0) {
// Preserve last-good by extending the existing key's TTL. upstashExpire
// resolves false when the key is already missing/expired — surface that as
// (TTL-MISS) so the log never implies a cache was preserved when it wasn't
// (the probe will then legitimately read `missing` until the next good write).
const extended = await upstashExpire(p.key, TOKEN_PANELS_SEED_TTL);
if (!extended) console.warn(`[TokenPanels] ${p.key} EXPIRE no-op — key missing/expired, last-good NOT preserved`);
results.push(`${p.label}:skip-empty${extended ? '' : '(TTL-MISS)'}`);
continue;
}
const ok = await envelopeWrite(p.key, p.payload, TOKEN_PANELS_SEED_TTL, { recordCount: p.payload.tokens.length, sourceVersion: p.sourceVersion });
results.push(`${p.label}:${p.payload.tokens.length}${ok ? '' : '(FAIL)'}`);
}
await upstashSet('seed-meta:market:token-panels', { fetchedAt: Date.now(), recordCount: total }, 604800);
console.log(`[TokenPanels] Seeded ${results.join(', ')} (${total} total)`);
return total;
}
let _marketSeedRun = null;
function seedAllMarketData() {
if (_marketSeedRun) {
console.warn('[Market] Prior seed still running — joining it instead of starting an overlapping refresh');
return _marketSeedRun;
}
_marketSeedRun = seedAllMarketDataOnce().finally(() => { _marketSeedRun = null; });
return _marketSeedRun;
}
async function seedAllMarketDataOnce() {
const t0 = Date.now();
// Equity gate (#4922d): shared dead days skip the stocks fetch+publish.
// Crypto (24/7), commodities, gulf, ETF and token panels are untouched.
let q = 0;
let equitySkipped = false;
if (!isMultiMarketEquityTradingDay()) {
equitySkipped = await maintainClosedMarketEquityKeys();
if (equitySkipped) {
if (!_equityGateLoggedClosed) {
console.log(`[Market] Tracked equity markets closed (US session=${getUsEquitySession()}) — skipping equity fetch, extended TTL on last-good keys`);
_equityGateLoggedClosed = true;
}
} else {
console.warn('[Market] Tracked equity markets closed but last-good equity keys missing — fetching anyway');
}
} else if (_equityGateLoggedClosed) {
console.log(`[Market] Tracked equity refresh resumed (US session=${getUsEquitySession()})`);
_equityGateLoggedClosed = false;
}
if (!equitySkipped) q = await seedMarketQuotes();
const c = await seedCommodityQuotes();
const s = await seedSectorSummary();
const g = await seedGulfQuotes();
const e = await seedEtfFlows();
const cr = await seedCryptoQuotes();
const sc = await seedStablecoinMarkets();
const cs = await seedCryptoSectors();
const tp = await seedTokenPanels();
console.log(`[Market] Seed complete: ${q} quotes, ${c} commodities, ${s} sectors, ${g} gulf, ${e} etf, ${cr} crypto, ${sc} stablecoins, ${cs} crypto-sectors, ${tp} token-panels (${((Date.now() - t0) / 1000).toFixed(1)}s)`);
}
async function startMarketDataSeedLoop() {
if (process.env.DISABLE_RELAY_MARKET_SEED) {
console.log('[Market] Relay market seeding disabled via DISABLE_RELAY_MARKET_SEED');
return;
}
if (!UPSTASH_ENABLED) {
console.log('[Market] Disabled (no Upstash Redis)');
return;
}
console.log(`[Market] Seed loop starting (interval ${MARKET_SEED_INTERVAL_MS / 1000 / 60}min, finnhub: ${FINNHUB_API_KEY ? 'yes' : 'no'})`);
startBootSeedLoop('Market', 'seed-meta:market:stocks', MARKET_SEED_INTERVAL_MS, seedAllMarketData, (e) => console.warn('[Market] Initial seed error:', e?.message || e), (e) => console.warn('[Market] Seed error:', e?.message || e));
}
// ─────────────────────────────────────────────────────────────
// Aviation Seed — Railway fetches AviationStack → writes to Redis
// so Vercel handler serves from cache (avoids 114 API calls per miss)
// ─────────────────────────────────────────────────────────────
// AviationStack API key — used only by the /aviationstack live proxy below.
// The aviation + NOTAM background seeds that used to live here were
// consolidated into scripts/seed-aviation.mjs (standalone Railway cron).
const AVIATIONSTACK_API_KEY = process.env.AVIATIONSTACK_API || '';
// In-process dedup sets for non-aviation seed notifications (cyber + UCDP).
// These track IDs already queued within a seed process's lifetime so the
// loops below don't re-notify on every poll. Cleared at 500 entries to
// bound memory. Aviation + NOTAM moved their dedup state to Redis
// (notifications:dedup:aviation:prev-alerted:v1 / notam:prev-closed-state:v1).
const cyberPrevAlertedIds = new Set();
const ucdpPrevAlertedIds = new Set();
// ─────────────────────────────────────────────────────────────
// Cyber Threat Intelligence Seed — Railway fetches IOC feeds → writes to Redis
// so Vercel handler (list-cyber-threats) serves from cache instead of live fetches
// ─────────────────────────────────────────────────────────────
const URLHAUS_AUTH_KEY = process.env.URLHAUS_AUTH_KEY || '';
const OTX_API_KEY = process.env.OTX_API_KEY || '';
const ABUSEIPDB_API_KEY = process.env.ABUSEIPDB_API_KEY || '';
const CYBER_SEED_INTERVAL_MS = 2 * 60 * 60 * 1000; // 2h — matches IOC feed update cadence
const CYBER_SEED_TTL = 21600; // 6h — 3x interval; survives 2 missed cycles before expiry
const CYBER_RETRY_MS = 20 * 60 * 1000;
const CYBER_RPC_KEY = 'cyber:threats:v2'; // must match handler REDIS_CACHE_KEY in list-cyber-threats.ts
const CYBER_BOOTSTRAP_KEY = 'cyber:threats-bootstrap:v2';
const CYBER_MAX_CACHED = 2000;
const CYBER_GEO_MAX = 200;
const CYBER_GEO_CONCURRENCY = 12;
const CYBER_GEO_TIMEOUT_MS = 20_000;
const CYBER_SOURCE_TIMEOUT_MS = 15_000; // longer than Vercel edge budget — OK on Railway
const CYBER_COUNTRY_CENTROIDS = {
US:[39.8,-98.6],CA:[56.1,-106.3],MX:[23.6,-102.6],BR:[-14.2,-51.9],AR:[-38.4,-63.6],
GB:[55.4,-3.4],DE:[51.2,10.5],FR:[46.2,2.2],IT:[41.9,12.6],ES:[40.5,-3.7],
NL:[52.1,5.3],BE:[50.5,4.5],SE:[60.1,18.6],NO:[60.5,8.5],FI:[61.9,25.7],
DK:[56.3,9.5],PL:[51.9,19.1],CZ:[49.8,15.5],AT:[47.5,14.6],CH:[46.8,8.2],
PT:[39.4,-8.2],IE:[53.1,-8.2],RO:[45.9,25.0],HU:[47.2,19.5],BG:[42.7,25.5],
HR:[45.1,15.2],SK:[48.7,19.7],UA:[48.4,31.2],RU:[61.5,105.3],BY:[53.7,28.0],
TR:[39.0,35.2],GR:[39.1,21.8],RS:[44.0,21.0],CN:[35.9,104.2],JP:[36.2,138.3],
KR:[35.9,127.8],IN:[20.6,79.0],PK:[30.4,69.3],BD:[23.7,90.4],ID:[-0.8,113.9],
TH:[15.9,101.0],VN:[14.1,108.3],PH:[12.9,121.8],MY:[4.2,101.9],SG:[1.4,103.8],
TW:[23.7,121.0],HK:[22.4,114.1],AU:[-25.3,133.8],NZ:[-40.9,174.9],
ZA:[-30.6,22.9],NG:[9.1,8.7],EG:[26.8,30.8],KE:[-0.02,37.9],ET:[9.1,40.5],
MA:[31.8,-7.1],DZ:[28.0,1.7],TN:[33.9,9.5],GH:[7.9,-1.0],
SA:[23.9,45.1],AE:[23.4,53.8],IL:[31.0,34.9],IR:[32.4,53.7],IQ:[33.2,43.7],
KW:[29.3,47.5],QA:[25.4,51.2],BH:[26.0,50.6],JO:[30.6,36.2],LB:[33.9,35.9],
CL:[-35.7,-71.5],CO:[4.6,-74.3],PE:[-9.2,-75.0],VE:[6.4,-66.6],
KZ:[48.0,68.0],UZ:[41.4,64.6],GE:[42.3,43.4],AZ:[40.1,47.6],AM:[40.1,45.0],
LT:[55.2,23.9],LV:[56.9,24.1],EE:[58.6,25.0],
HN:[15.2,-86.2],GT:[15.8,-90.2],PA:[8.5,-80.8],CR:[9.7,-84.0],
SN:[14.5,-14.5],CM:[7.4,12.4],CI:[7.5,-5.5],TZ:[-6.4,34.9],UG:[1.4,32.3],
};
const CYBER_THREAT_TYPE_MAP = { c2_server:'CYBER_THREAT_TYPE_C2_SERVER', malware_host:'CYBER_THREAT_TYPE_MALWARE_HOST', phishing:'CYBER_THREAT_TYPE_PHISHING', malicious_url:'CYBER_THREAT_TYPE_MALICIOUS_URL' };
const CYBER_SOURCE_MAP = { feodo:'CYBER_THREAT_SOURCE_FEODO', urlhaus:'CYBER_THREAT_SOURCE_URLHAUS', c2intel:'CYBER_THREAT_SOURCE_C2INTEL', otx:'CYBER_THREAT_SOURCE_OTX', abuseipdb:'CYBER_THREAT_SOURCE_ABUSEIPDB' };
const CYBER_INDICATOR_MAP = { ip:'CYBER_THREAT_INDICATOR_TYPE_IP', domain:'CYBER_THREAT_INDICATOR_TYPE_DOMAIN', url:'CYBER_THREAT_INDICATOR_TYPE_URL' };
const CYBER_SEVERITY_MAP = { low:'CRITICALITY_LEVEL_LOW', medium:'CRITICALITY_LEVEL_MEDIUM', high:'CRITICALITY_LEVEL_HIGH', critical:'CRITICALITY_LEVEL_CRITICAL' };
const CYBER_SEVERITY_RANK = { CRITICALITY_LEVEL_CRITICAL:4, CRITICALITY_LEVEL_HIGH:3, CRITICALITY_LEVEL_MEDIUM:2, CRITICALITY_LEVEL_LOW:1, CRITICALITY_LEVEL_UNSPECIFIED:0 };
function cyberClean(v, max) { if (typeof v !== 'string') return ''; return v.trim().replace(/\s+/g, ' ').slice(0, max || 120); }
function cyberToNum(v) { const n = typeof v === 'number' ? v : parseFloat(String(v ?? '')); return Number.isFinite(n) ? n : null; }
function cyberValidCoords(lat, lon) { return lat !== null && lon !== null && lat >= -90 && lat <= 90 && lon >= -180 && lon <= 180; }
function cyberIsIPv4(v) { if (!/^(\d{1,3}\.){3}\d{1,3}$/.test(v)) return false; return v.split('.').map(Number).every((n) => Number.isInteger(n) && n >= 0 && n <= 255); }
function cyberIsIPv6(v) { return /^[0-9a-f:]+$/i.test(v) && v.includes(':'); }
function cyberIsIp(v) { return cyberIsIPv4(v) || cyberIsIPv6(v); }
function cyberNormCountry(v) { const r = cyberClean(String(v ?? ''), 64); if (!r) return ''; if (/^[a-z]{2}$/i.test(r)) return r.toUpperCase(); return r; }
function cyberToMs(v) {
if (!v) return 0;
const raw = cyberClean(String(v), 80); if (!raw) return 0;
const d1 = new Date(raw); if (!Number.isNaN(d1.getTime())) return d1.getTime();
const d2 = new Date(raw.replace(' UTC', 'Z').replace(' GMT', 'Z').replace(' ', 'T'));
return Number.isNaN(d2.getTime()) ? 0 : d2.getTime();
}
function cyberNormTags(input, max) {
const tags = Array.isArray(input) ? input : typeof input === 'string' ? input.split(/[;,|]/g) : [];
const out = []; const seen = new Set();
for (const t of tags) { const c = cyberClean(String(t ?? ''), 40).toLowerCase(); if (!c || seen.has(c)) continue; seen.add(c); out.push(c); if (out.length >= (max || 8)) break; }
return out;
}
function cyberDjb2(s) { let h = 5381; for (let i = 0; i < s.length; i++) h = ((h << 5) + h + s.charCodeAt(i)) & 0xffffffff; return h; }
function cyberCentroid(cc, seed) {
const c = CYBER_COUNTRY_CENTROIDS[cc ? cc.toUpperCase() : '']; if (!c) return null;
const k = seed || cc;
return { lat: c[0] + (((cyberDjb2(k) & 0xffff) / 0xffff) - 0.5) * 2, lon: c[1] + (((cyberDjb2(k + ':lon') & 0xffff) / 0xffff) - 0.5) * 2 };
}
function cyberSanitize(t) {
const ind = cyberClean(t.indicator, 255); if (!ind) return null;
if ((t.indicatorType || 'ip') === 'ip' && !cyberIsIp(ind)) return null;
return { id: cyberClean(t.id, 255) || `${t.source||'feodo'}:${t.indicatorType||'ip'}:${ind}`, type: t.type||'malicious_url', source: t.source||'feodo', indicator: ind, indicatorType: t.indicatorType||'ip', lat: t.lat??null, lon: t.lon??null, country: t.country||'', severity: t.severity||'medium', malwareFamily: cyberClean(t.malwareFamily, 80), tags: t.tags||[], firstSeen: t.firstSeen||0, lastSeen: t.lastSeen||0 };
}
function cyberDedupe(threats) {
const map = new Map();
for (const t of threats) {
const key = `${t.source}:${t.indicatorType}:${t.indicator}`;
const ex = map.get(key);
if (!ex) { map.set(key, t); continue; }
if ((t.lastSeen || t.firstSeen) >= (ex.lastSeen || ex.firstSeen)) map.set(key, { ...ex, ...t, tags: cyberNormTags([...ex.tags, ...t.tags]) });
}
return Array.from(map.values());
}
function cyberToProto(t) {
return { id: t.id, type: CYBER_THREAT_TYPE_MAP[t.type]||'CYBER_THREAT_TYPE_UNSPECIFIED', source: CYBER_SOURCE_MAP[t.source]||'CYBER_THREAT_SOURCE_UNSPECIFIED', indicator: t.indicator, indicatorType: CYBER_INDICATOR_MAP[t.indicatorType]||'CYBER_THREAT_INDICATOR_TYPE_UNSPECIFIED', location: cyberValidCoords(t.lat, t.lon) ? { latitude: t.lat, longitude: t.lon } : undefined, country: t.country, severity: CYBER_SEVERITY_MAP[t.severity]||'CRITICALITY_LEVEL_UNSPECIFIED', malwareFamily: t.malwareFamily, tags: t.tags, firstSeenAt: t.firstSeen, lastSeenAt: t.lastSeen };
}
function cyberHttpGetJson(url, reqHeaders, timeoutMs) {
return new Promise((resolve) => {
const req = https.get(url, { signal: AbortSignal.timeout(timeoutMs || 10000), headers: { 'User-Agent': CHROME_UA, ...reqHeaders }, timeout: timeoutMs || 10000 }, (resp) => {
if (resp.statusCode < 200 || resp.statusCode >= 300) { resp.resume(); return resolve(null); }
const chunks = [];
resp.on('data', (c) => chunks.push(c));
resp.on('end', () => { try { resolve(JSON.parse(Buffer.concat(chunks).toString())); } catch { resolve(null); } });
});
req.on('error', () => resolve(null));
req.on('timeout', () => { req.destroy(); resolve(null); });
});
}
function cyberHttpGetText(url, reqHeaders, timeoutMs) {
return new Promise((resolve) => {
const req = https.get(url, { signal: AbortSignal.timeout(timeoutMs || 10000), headers: { 'User-Agent': CHROME_UA, ...reqHeaders }, timeout: timeoutMs || 10000 }, (resp) => {
if (resp.statusCode < 200 || resp.statusCode >= 300) { resp.resume(); return resolve(null); }
const chunks = [];
resp.on('data', (c) => chunks.push(c));
resp.on('end', () => resolve(Buffer.concat(chunks).toString()));
});
req.on('error', () => resolve(null));
req.on('timeout', () => { req.destroy(); resolve(null); });
});
}
const CYBER_GEO_CACHE_MAX = 2048;
const cyberGeoCache = new Map();
function cyberGeoCacheSet(ip, geo) {
if (cyberGeoCache.size >= CYBER_GEO_CACHE_MAX) {
cyberGeoCache.delete(cyberGeoCache.keys().next().value);
}
cyberGeoCache.set(ip, geo);
}
async function cyberGeoLookup(ip) {
if (cyberGeoCache.has(ip)) return cyberGeoCache.get(ip);
const d1 = await cyberHttpGetJson(`https://ipinfo.io/${encodeURIComponent(ip)}/json`, {}, 3000);
if (d1?.loc) {
const [latS, lonS] = d1.loc.split(',');
const lat = parseFloat(latS), lon = parseFloat(lonS);
if (cyberValidCoords(lat, lon)) { const r = { lat, lon, country: String(d1.country||'').slice(0,2).toUpperCase() }; cyberGeoCacheSet(ip, r); return r; }
}
const d2 = await cyberHttpGetJson(`https://freeipapi.com/api/json/${encodeURIComponent(ip)}`, {}, 3000);
if (d2) {
const lat = parseFloat(d2.latitude), lon = parseFloat(d2.longitude);
if (cyberValidCoords(lat, lon)) { const r = { lat, lon, country: String(d2.countryCode||d2.countryName||'').slice(0,2).toUpperCase() }; cyberGeoCacheSet(ip, r); return r; }
}
return null;
}
async function cyberHydrateGeo(threats) {
const needsGeo = []; const seen = new Set();
for (const t of threats) {
if (cyberValidCoords(t.lat, t.lon) || t.indicatorType !== 'ip') continue;
const ip = t.indicator.toLowerCase();
if (!cyberIsIp(ip) || seen.has(ip)) continue;
seen.add(ip); needsGeo.push(ip);
}
if (needsGeo.length === 0) return threats;
const queue = [...needsGeo.slice(0, CYBER_GEO_MAX)];
const resolved = new Map();
let timedOut = false;
// timedOut flag stops workers from dequeuing new IPs; in-flight requests may still
// complete up to ~3s after the flag fires (per-request timeout). Acceptable overshoot.
const timeoutId = setTimeout(() => { timedOut = true; }, CYBER_GEO_TIMEOUT_MS);
const workers = Array.from({ length: Math.min(CYBER_GEO_CONCURRENCY, queue.length) }, async () => {
while (queue.length > 0 && !timedOut) {
const ip = queue.shift(); if (!ip) break;
const geo = await cyberGeoLookup(ip);
if (geo) resolved.set(ip, geo);
}
});
try { await Promise.all(workers); } catch { /* ignore */ }
clearTimeout(timeoutId);
return threats.map((t) => {
if (cyberValidCoords(t.lat, t.lon)) return t;
if (t.indicatorType !== 'ip') return t;
const geo = resolved.get(t.indicator.toLowerCase());
if (geo) return { ...t, lat: geo.lat, lon: geo.lon, country: t.country || geo.country };
const cen = cyberCentroid(t.country, t.indicator);
return cen ? { ...t, lat: cen.lat, lon: cen.lon } : t;
});
}
async function cyberFetchFeodo(limit, cutoffMs) {
try {
const payload = await cyberHttpGetJson('https://feodotracker.abuse.ch/downloads/ipblocklist.json', { Accept: 'application/json' }, CYBER_SOURCE_TIMEOUT_MS);
if (!payload) return [];
const records = Array.isArray(payload) ? payload : (Array.isArray(payload?.data) ? payload.data : []);
const out = [];
for (const r of records) {
const ip = cyberClean(r?.ip_address || r?.dst_ip || r?.ip || r?.ioc || r?.host, 80).toLowerCase();
if (!cyberIsIp(ip)) continue;
const status = cyberClean(r?.status || r?.c2_status || '', 30).toLowerCase();
if (status && status !== 'online' && status !== 'offline') continue;
const firstSeen = cyberToMs(r?.first_seen || r?.first_seen_utc || r?.dateadded);
const lastSeen = cyberToMs(r?.last_online || r?.last_seen || r?.last_seen_utc || r?.first_seen || r?.first_seen_utc);
if ((lastSeen || firstSeen) && (lastSeen || firstSeen) < cutoffMs) continue;
const malwareFamily = cyberClean(r?.malware || r?.malware_family || r?.family, 80);
const sev = status === 'online' ? (/emotet|qakbot|trickbot|dridex|ransom/i.test(malwareFamily) ? 'critical' : 'high') : 'medium';
const t = cyberSanitize({ id: `feodo:${ip}`, type: 'c2_server', source: 'feodo', indicator: ip, indicatorType: 'ip', lat: cyberToNum(r?.latitude ?? r?.lat), lon: cyberToNum(r?.longitude ?? r?.lon), country: cyberNormCountry(r?.country || r?.country_code), severity: sev, malwareFamily, tags: cyberNormTags(['botnet', 'c2', ...(r?.tags||[])]), firstSeen, lastSeen });
if (t) { out.push(t); if (out.length >= limit) break; }
}
return out;
} catch (e) { console.warn('[Cyber] Feodo fetch failed:', e?.message || e); return []; }
}
async function cyberFetchUrlhaus(limit, cutoffMs) {
if (!URLHAUS_AUTH_KEY) return [];
try {
const payload = await cyberHttpGetJson(`https://urlhaus-api.abuse.ch/v1/urls/recent/limit/${limit}/`, { Accept: 'application/json', 'Auth-Key': URLHAUS_AUTH_KEY }, CYBER_SOURCE_TIMEOUT_MS);
if (!payload) return [];
const rows = Array.isArray(payload?.urls) ? payload.urls : (Array.isArray(payload?.data) ? payload.data : []);
const out = [];
for (const r of rows) {
const rawUrl = cyberClean(r?.url || r?.ioc || '', 1024);
const status = cyberClean(r?.url_status || r?.status || '', 30).toLowerCase();
if (status && status !== 'online') continue;
const tags = cyberNormTags(r?.tags);
let hostname = ''; try { hostname = new URL(rawUrl).hostname.toLowerCase(); } catch {}
const recordIp = cyberClean(r?.host || r?.ip_address || r?.ip, 80).toLowerCase();
const ipCandidate = cyberIsIp(recordIp) ? recordIp : (cyberIsIp(hostname) ? hostname : '');
const indType = ipCandidate ? 'ip' : (hostname ? 'domain' : 'url');
const indicator = ipCandidate || hostname || rawUrl; if (!indicator) continue;
const firstSeen = cyberToMs(r?.dateadded || r?.firstseen || r?.first_seen);
const lastSeen = cyberToMs(r?.last_online || r?.last_seen || r?.dateadded);
if ((lastSeen || firstSeen) && (lastSeen || firstSeen) < cutoffMs) continue;
const threat = cyberClean(r?.threat || r?.threat_type || '', 40).toLowerCase();
const allTags = tags.join(' ');
const type = (threat.includes('phish') || allTags.includes('phish')) ? 'phishing' : (threat.includes('malware') || threat.includes('payload') || allTags.includes('malware')) ? 'malware_host' : 'malicious_url';
const sev = type === 'phishing' ? 'medium' : (tags.includes('ransomware') || tags.includes('botnet')) ? 'critical' : 'high';
const t = cyberSanitize({ id: `urlhaus:${indType}:${indicator}`, type, source: 'urlhaus', indicator, indicatorType: indType, lat: cyberToNum(r?.latitude ?? r?.lat), lon: cyberToNum(r?.longitude ?? r?.lon), country: cyberNormCountry(r?.country || r?.country_code), severity: sev, malwareFamily: cyberClean(r?.threat, 80), tags, firstSeen, lastSeen });
if (t) { out.push(t); if (out.length >= limit) break; }
}
return out;
} catch (e) { console.warn('[Cyber] URLhaus fetch failed:', e?.message || e); return []; }
}
async function cyberFetchC2Intel(limit) {
try {
const text = await cyberHttpGetText('https://raw.githubusercontent.com/drb-ra/C2IntelFeeds/master/feeds/IPC2s-30day.csv', { Accept: 'text/plain' }, CYBER_SOURCE_TIMEOUT_MS);
if (!text) return [];
const out = [];
for (const line of text.split('\n')) {
if (!line || line.startsWith('#')) continue;
const ci = line.indexOf(','); if (ci < 0) continue;
const ip = cyberClean(line.slice(0, ci), 80).toLowerCase(); if (!cyberIsIp(ip)) continue;
const desc = cyberClean(line.slice(ci + 1), 200);
const malwareFamily = desc.replace(/^Possible\s+/i, '').replace(/\s+C2\s+IP$/i, '').trim() || 'Unknown';
const tags = ['c2']; const descLow = desc.toLowerCase();
if (descLow.includes('cobaltstrike') || descLow.includes('cobalt strike')) tags.push('cobaltstrike');
if (descLow.includes('metasploit')) tags.push('metasploit');
if (descLow.includes('sliver')) tags.push('sliver');
if (descLow.includes('brute ratel') || descLow.includes('bruteratel')) tags.push('bruteratel');
const t = cyberSanitize({ id: `c2intel:${ip}`, type: 'c2_server', source: 'c2intel', indicator: ip, indicatorType: 'ip', lat: null, lon: null, country: '', severity: /cobaltstrike|cobalt.strike|brute.?ratel/i.test(desc) ? 'high' : 'medium', malwareFamily, tags: cyberNormTags(tags), firstSeen: 0, lastSeen: 0 });
if (t) { out.push(t); if (out.length >= limit) break; }
}
return out;
} catch (e) { console.warn('[Cyber] C2Intel fetch failed:', e?.message || e); return []; }
}
async function cyberFetchOtx(limit, days) {
if (!OTX_API_KEY) return [];
try {
const since = new Date(Date.now() - days * 24 * 60 * 60 * 1000).toISOString().slice(0, 10);
const payload = await cyberHttpGetJson(`https://otx.alienvault.com/api/v1/indicators/export?type=IPv4&modified_since=${encodeURIComponent(since)}`, { Accept: 'application/json', 'X-OTX-API-KEY': OTX_API_KEY }, CYBER_SOURCE_TIMEOUT_MS);
if (!payload) return [];
const results = Array.isArray(payload?.results) ? payload.results : (Array.isArray(payload) ? payload : []);
const out = [];
for (const r of results) {
const ip = cyberClean(r?.indicator || r?.ip || '', 80).toLowerCase(); if (!cyberIsIp(ip)) continue;
const tags = cyberNormTags(r?.tags || []);
const t = cyberSanitize({ id: `otx:${ip}`, type: tags.some((tt) => /c2|botnet/.test(tt)) ? 'c2_server' : 'malware_host', source: 'otx', indicator: ip, indicatorType: 'ip', lat: null, lon: null, country: '', severity: tags.some((tt) => /ransomware|apt|c2|botnet/.test(tt)) ? 'high' : 'medium', malwareFamily: cyberClean(r?.title || r?.description || '', 200), tags, firstSeen: cyberToMs(r?.created), lastSeen: cyberToMs(r?.modified || r?.created) });
if (t) { out.push(t); if (out.length >= limit) break; }
}
return out;
} catch (e) { console.warn('[Cyber] OTX fetch failed:', e?.message || e); return []; }
}
async function cyberFetchAbuseIpDb(limit) {
if (!ABUSEIPDB_API_KEY) return [];
try {
const payload = await cyberHttpGetJson(`https://api.abuseipdb.com/api/v2/blacklist?confidenceMinimum=90&limit=${Math.min(limit, 500)}`, { Accept: 'application/json', Key: ABUSEIPDB_API_KEY }, CYBER_SOURCE_TIMEOUT_MS);
if (!payload) return [];
const records = Array.isArray(payload?.data) ? payload.data : [];
const out = [];
for (const r of records) {
const ip = cyberClean(r?.ipAddress || r?.ip || '', 80).toLowerCase(); if (!cyberIsIp(ip)) continue;
const score = cyberToNum(r?.abuseConfidenceScore) ?? 0;
const t = cyberSanitize({ id: `abuseipdb:${ip}`, type: 'malware_host', source: 'abuseipdb', indicator: ip, indicatorType: 'ip', lat: cyberToNum(r?.latitude ?? r?.lat), lon: cyberToNum(r?.longitude ?? r?.lon), country: cyberNormCountry(r?.countryCode || r?.country), severity: score >= 95 ? 'critical' : (score >= 80 ? 'high' : 'medium'), malwareFamily: '', tags: cyberNormTags([`score:${score}`]), firstSeen: 0, lastSeen: cyberToMs(r?.lastReportedAt) });
if (t) { out.push(t); if (out.length >= limit) break; }
}
return out;
} catch (e) { console.warn('[Cyber] AbuseIPDB fetch failed:', e?.message || e); return []; }
}
let cyberSeedInFlight = false;
let cyberRetryTimer = null;
async function seedCyberThreats() {
if (cyberSeedInFlight) return 0;
cyberSeedInFlight = true;
if (cyberRetryTimer) { clearTimeout(cyberRetryTimer); cyberRetryTimer = null; }
const t0 = Date.now();
try {
const days = 14;
const cutoffMs = Date.now() - days * 24 * 60 * 60 * 1000;
const MAX_LIMIT = 1000;
const [feodo, urlhaus, c2intel, otx, abuseipdb] = await Promise.all([
cyberFetchFeodo(MAX_LIMIT, cutoffMs),
cyberFetchUrlhaus(MAX_LIMIT, cutoffMs),
cyberFetchC2Intel(MAX_LIMIT),
cyberFetchOtx(MAX_LIMIT, days),
cyberFetchAbuseIpDb(MAX_LIMIT),
]);
if (feodo.length + urlhaus.length + c2intel.length + otx.length + abuseipdb.length === 0) {
console.warn('[Cyber] All sources returned 0 threats — extending TTL, retrying in 20min');
try { await upstashExpire(CYBER_RPC_KEY, CYBER_SEED_TTL); await upstashExpire(CYBER_BOOTSTRAP_KEY, CYBER_SEED_TTL); } catch {}
cyberRetryTimer = setTimeout(() => { seedCyberThreats().catch(() => {}); }, CYBER_RETRY_MS);
return 0;
}
const combined = cyberDedupe([...feodo, ...urlhaus, ...c2intel, ...otx, ...abuseipdb]);
const hydrated = await cyberHydrateGeo(combined);
const geoCount = hydrated.filter((t) => cyberValidCoords(t.lat, t.lon)).length;
console.log(`[Cyber] Geo resolved: ${geoCount}/${hydrated.length}`);
hydrated.sort((a, b) => {
const aGeo = cyberValidCoords(a.lat, a.lon) ? 0 : 1;
const bGeo = cyberValidCoords(b.lat, b.lon) ? 0 : 1;
if (aGeo !== bGeo) return aGeo - bGeo;
const bySev = (CYBER_SEVERITY_RANK[CYBER_SEVERITY_MAP[b.severity]||'']||0) - (CYBER_SEVERITY_RANK[CYBER_SEVERITY_MAP[a.severity]||'']||0);
return bySev !== 0 ? bySev : (b.lastSeen || b.firstSeen) - (a.lastSeen || a.firstSeen);
});
const threats = hydrated.slice(0, CYBER_MAX_CACHED).map(cyberToProto);
if (threats.length === 0) {
console.warn('[Cyber] No threats after processing — extending TTL, retrying in 20min');
try { await upstashExpire(CYBER_RPC_KEY, CYBER_SEED_TTL); await upstashExpire(CYBER_BOOTSTRAP_KEY, CYBER_SEED_TTL); } catch {}
cyberRetryTimer = setTimeout(() => { seedCyberThreats().catch(() => {}); }, CYBER_RETRY_MS);
return 0;
}
const payload = { threats };
const ok1 = await envelopeWrite(CYBER_RPC_KEY, payload, CYBER_SEED_TTL, { recordCount: threats.length, sourceVersion: 'cyber-threats' });
const ok2 = await envelopeWrite(CYBER_BOOTSTRAP_KEY, payload, CYBER_SEED_TTL, { recordCount: threats.length, sourceVersion: 'cyber-threats' });
const ok3 = await upstashSet('seed-meta:cyber:threats', { fetchedAt: Date.now(), recordCount: threats.length }, 604800);
console.log(`[Cyber] Seeded ${threats.length} threats (feodo:${feodo.length} urlhaus:${urlhaus.length} c2intel:${c2intel.length} otx:${otx.length} abuseipdb:${abuseipdb.length} redis:${ok1 && ok2 && ok3 ? 'OK' : 'PARTIAL'}) in ${((Date.now() - t0) / 1000).toFixed(1)}s`);
const newCyber = hydrated.filter(t =>
(t.severity === 'critical' || t.severity === 'high') && !cyberPrevAlertedIds.has(t.indicator)
);
for (const t of newCyber.slice(0, 2)) {
cyberPrevAlertedIds.add(t.indicator);
const typeLabel = (t.type || 'threat').replace(/_/g, ' ');
const familyTag = t.malwareFamily ? ` (${t.malwareFamily.slice(0, 30)})` : '';
const countryCode = normalizeNotificationCountryCode(t.country);
publishNotificationEvent({
eventType: 'cyber_threat',
payload: { title: `${typeLabel}: ${t.indicator?.slice(0, 50)}${familyTag}`, source: t.source || 'Cyber Intel', ...(countryCode ? { countryCode } : {}) },
severity: t.severity === 'critical' ? 'critical' : 'high',
variant: undefined,
dedupTtl: 43200,
}).catch(err => console.warn('[Notify] Cyber publish error:', err?.message));
}
if (cyberPrevAlertedIds.size > 500) cyberPrevAlertedIds.clear();
return threats.length;
} catch (e) {
console.warn('[Cyber] Seed error:', e?.message || e, '— extending TTL, retrying in 20min');
try { await upstashExpire(CYBER_RPC_KEY, CYBER_SEED_TTL); await upstashExpire(CYBER_BOOTSTRAP_KEY, CYBER_SEED_TTL); } catch {}
cyberRetryTimer = setTimeout(() => { seedCyberThreats().catch(() => {}); }, CYBER_RETRY_MS);
return 0;
} finally {
cyberSeedInFlight = false;
}
}
async function startCyberThreatsSeedLoop() {
if (!UPSTASH_ENABLED) {
console.log('[Cyber] Disabled (no Upstash Redis)');
return;
}
console.log(`[Cyber] Seed loop starting (interval ${CYBER_SEED_INTERVAL_MS / 1000 / 60 / 60}h, urlhaus:${URLHAUS_AUTH_KEY ? 'yes' : 'no'} otx:${OTX_API_KEY ? 'yes' : 'no'} abuseipdb:${ABUSEIPDB_API_KEY ? 'yes' : 'no'})`);
startBootSeedLoop('Cyber', 'seed-meta:cyber:threats', CYBER_SEED_INTERVAL_MS, seedCyberThreats, (e) => console.warn('[Cyber] Initial seed error:', e?.message || e), (e) => console.warn('[Cyber] Seed error:', e?.message || e));
}
// ─────────────────────────────────────────────────────────────
// AI Classification Seed — batch-classify digest titles via LLM → Redis
// Clients get pre-classified items from digest, zero classify-event RPCs
// ─────────────────────────────────────────────────────────────
const CLASSIFY_SEED_INTERVAL_MS = 15 * 60 * 1000;
const CLASSIFY_CACHE_TTL = 86400;
const CLASSIFY_SKIP_TTL = 1800;
const CLASSIFY_BATCH_SIZE = 50;
// Every edition the news digest serves, generated from shared/news-catalog.json
// (scripts/generate-browser-news-catalog.mjs), so a new edition is classified
// without a relay edit.
const CLASSIFY_VARIANTS = requireShared('news-digest-editions.json');
// The waits between editions total CLASSIFY_STAGGER_TOTAL_MS whatever the
// edition count, leaving the rest of the 15-minute interval for classify work;
// a fixed per-edition wait would overrun the interval as editions are added and
// the in-flight guard would then skip every other cycle. Zero under
// RELAY_TEST_MODE so the /__test/seed-classify seam runs a full cycle in
// milliseconds.
const CLASSIFY_STAGGER_TOTAL_MS = 12 * 60 * 1000;
const CLASSIFY_VARIANT_STAGGER_MS = RELAY_TEST_MODE
? 0
: Math.floor(CLASSIFY_STAGGER_TOTAL_MS / Math.max(1, CLASSIFY_VARIANTS.length - 1));
// Relay gates — active only when RELAY_GATES_READY=1 (see Appendix E of docs/internal/news-alerts-enhancements-from-trendradar.md).
// When the flag is set the relay becomes the sole authoritative source of rss_alert events and
// the client /api/notify path is suppressed via VITE_RELAY_GATES_READY on the Vercel side.
const RELAY_GATES_READY = process.env.RELAY_GATES_READY === '1';
const RELAY_RECENCY_MS = 15 * 60 * 1000; // 15 min — matches client-side recency gate
// ── Importance score parity with digest ──────────────────────────────────────
// Source-tier data loaded via requireShared('source-tiers.json'), which resolves
// to either repo-root shared/ OR scripts/shared/ depending on packaging root.
// Both copies are enforced byte-identical by tests/edge-functions.test.mjs
// ('scripts/shared/ stays in sync with shared/') and cross-checked in
// tests/importance-score-parity.test.mjs.
// Formula constants + computeImportanceScore mirror list-feed-digest.ts; parity
// is enforced by tests/importance-score-parity.test.mjs.
const RELAY_SOURCE_TIERS = {
...requireShared('source-tiers.json'),
...requireShared('x-account-source-tiers.json'),
[SAUDI_CIVIL_DEFENSE.name]: SAUDI_CIVIL_DEFENSE.tier,
};
const {
createExplicitTierFourSourceSet,
shouldDropRelaySourceForTier,
} = requireShared('source-tier-policy.cjs');
function relayGetSourceTier(sourceName) {
return RELAY_SOURCE_TIERS[sourceName] ?? 4;
}
// Derived from the tier map so the tier-4 gate and the tier map stay in lockstep.
const RELAY_TIER4_SOURCES = createExplicitTierFourSourceSet(RELAY_SOURCE_TIERS);
const RELAY_SCORE_WEIGHTS = { severity: 0.55, sourceTier: 0.2, corroboration: 0.15, recency: 0.1 };
const RELAY_SEVERITY_SCORES = { critical: 100, high: 75, medium: 50, low: 25, info: 0 };
const RELAY_DIPLOMACY_KEYWORDS = [
'ceasefire', 'truce', 'armistice', 'treaty', 'accord', 'pact', 'diplomatic',
'diplomacy', 'mediate', 'mediator', 'negotiation', 'negotiations', 'negotiate',
'normalization', 'normalisation',
];
const RELAY_FLASHPOINT_SCORING_KEYWORDS = [
'iran', 'tehran', 'russia', 'moscow', 'china', 'beijing', 'taiwan', 'ukraine', 'kyiv',
'north korea', 'pyongyang', 'israel', 'gaza', 'west bank', 'syria', 'damascus',
'yemen', 'hezbollah', 'hamas', 'kremlin', 'pentagon', 'nato', 'wagner',
];
const RELAY_DIPLOMACY_FLASHPOINT_PAIRS = [
['iran', 'deal'],
['iran', 'talks'],
['iran', 'ceasefire'],
['iran', 'treaty'],
['iran', 'accord'],
['iran', 'peace'],
['israel', 'ceasefire'],
['israel', 'truce'],
['israel', 'accord'],
['gaza', 'ceasefire'],
['gaza', 'truce'],
['ukraine', 'ceasefire'],
['ukraine', 'talks'],
['russia', 'talks'],
['russia', 'treaty'],
['hamas', 'truce'],
['hezbollah', 'truce'],
['syria', 'ceasefire'],
['china', 'talks'],
['china', 'accord'],
['taiwan', 'talks'],
['yemen', 'ceasefire'],
['north korea', 'talks'],
['pyongyang', 'talks'],
];
const RELAY_DIPLOMACY_FLASHPOINT_BOOST = 18;
const RELAY_ENTITY_CORROBORATION_SCORE_PER_SOURCE = 4;
function relayNormalizeScoringText(text) {
return String(text || '').toLowerCase().replace(/[^a-z0-9\s]/g, ' ').replace(/\s+/g, ' ').trim();
}
// Word-start containment in normalized text. Mirrors
// shared/brief-filter.js:containsKeywordToken — prevents 'pact' inside
// 'impact' (false positive) while still matching 'iran' inside
// 'iranian' (demonym preserved). PR #3909 review (P2). Keeps the
// relay aligned with digest under tests/importance-score-parity.test.mjs.
function relayContainsKeywordToken(text, kw) {
if (!kw) return false;
const escaped = kw.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
return new RegExp(`(^|\\s)${escaped}`).test(text);
}
function relayHasAnySignal(text, keywords) {
return keywords.some((kw) => relayContainsKeywordToken(text, kw));
}
function relayHasDiplomacyFlashpointSignal(title) {
if (!title) return false;
const text = relayNormalizeScoringText(title);
if (
RELAY_DIPLOMACY_FLASHPOINT_PAIRS.some(([entity, action]) =>
relayContainsKeywordToken(text, entity) && relayContainsKeywordToken(text, action),
)
) {
return true;
}
return relayHasAnySignal(text, RELAY_DIPLOMACY_KEYWORDS) &&
relayHasAnySignal(text, RELAY_FLASHPOINT_SCORING_KEYWORDS);
}
function relayDiplomacyFlashpointBoost(title) {
return relayHasDiplomacyFlashpointSignal(title) ? RELAY_DIPLOMACY_FLASHPOINT_BOOST : 0;
}
function relayEntityCorroborationScore(count) {
const finite = Number.isFinite(count) ? Number(count) : 0;
return Math.min(Math.max(finite, 0), 5) * RELAY_ENTITY_CORROBORATION_SCORE_PER_SOURCE;
}
// Mirrors computeImportanceScore() in list-feed-digest.ts with ONE intentional
// deviation: the relay defensively returns 0 for unknown severity levels
// (`?? 0` on the lookup); the TS digest returns NaN. This defensiveness is
// exercised in tests/importance-score-parity.test.mjs "unknown severity" case.
// Caller responsibility: pass defined values; relay publish site defaults
// corroborationCount → 1 and publishedAt → Date.now() when upstream omits them.
function relayComputeImportanceScore(level, source, corroborationCount, publishedAt, context = {}) {
const tier = relayGetSourceTier(source);
const tierScore = tier === 1 ? 100 : tier === 2 ? 75 : tier === 3 ? 50 : 25;
const corroborationScore = Math.min(corroborationCount, 5) * 20;
const ageMs = Date.now() - publishedAt;
const recencyScore = Math.max(0, 1 - ageMs / (24 * 60 * 60 * 1000)) * 100;
const base = Math.round(
(RELAY_SEVERITY_SCORES[level] ?? 0) * RELAY_SCORE_WEIGHTS.severity +
tierScore * RELAY_SCORE_WEIGHTS.sourceTier +
corroborationScore * RELAY_SCORE_WEIGHTS.corroboration +
recencyScore * RELAY_SCORE_WEIGHTS.recency,
);
return Math.round(
base +
relayDiplomacyFlashpointBoost(context.title) +
relayEntityCorroborationScore(context.entityCorroborationCount),
);
}
const CLASSIFY_VALID_LEVELS = ['critical', 'high', 'medium', 'low', 'info'];
const CLASSIFY_VALID_CATEGORIES = [
'conflict', 'protest', 'disaster', 'diplomatic', 'economic',
'terrorism', 'cyber', 'health', 'environmental', 'military',
'crime', 'infrastructure', 'tech', 'general',
];
const CLASSIFY_SYSTEM_PROMPT = `You classify news headlines by threat level and category.
Return ONLY a JSON array, no other text.
Levels: critical, high, medium, low, info
Categories: conflict, protest, disaster, diplomatic, economic, terrorism, cyber, health, environmental, military, crime, infrastructure, tech, general
Guidelines for LEVEL assignment (geopolitical scope required for critical):
- critical: Active military strikes with international implications, geopolitical mass-casualty events (10+ killed in conflict/terrorism/state action), ceasefire agreements/collapses, nuclear incidents, pandemic declarations, coups, strait/waterway closures
- high: Armed conflict updates, major diplomatic actions, sanctions packages, significant natural disasters, blockades, terrorist attacks, domestic mass-casualty events (mass shootings, industrial disasters)
- medium: Ongoing conflict analysis, economic impact reports, protest movements, regional policy changes, military exercises
- low: Diplomatic meetings, trade discussions, humanitarian aid, election updates, peacekeeping deployments
- info: Opinion/editorial pieces, analysis/explainer articles, historical retrospectives, lifestyle, entertainment, routine local news, tutorials
Key distinction: "critical" requires GEOPOLITICAL scope — events that destabilize international order, threaten cross-border security, or disrupt global systems. Domestic tragedies are "high" unless they trigger international diplomatic responses.
- "8 children killed in mass shooting in Louisiana" → domestic mass-casualty, not geopolitical → high
- "23 killed in fireworks factory explosion in India" → industrial accident → high
- "700 killed in Sudan drone strikes" → geopolitical mass-casualty in active civil war → critical
- "Iran closes Strait of Hormuz" → global trade disruption → critical
- "Guardian view on ceasefire: need real peace" → editorial → info
- "Trump's obsession with energy" → opinion/analysis → info
- "Man killed his estranged wife" → domestic crime → info
- "How to Crack the SAM Database in Kali Linux" → tutorial → info
Do not under-rate "high". The EVENT itself is high even when nobody is hurt and even when the headline reports a vote, an approval or an announcement:
- a sanctions package or sanctions bill passed, signed or imposed
- a major arms sale or weapons transfer approved between states
- a military deployment or force movement ahead of an operation
- an armed attack, raid or clash with deaths, including one that was repelled
- many deaths in state custody or by state action
- a natural disaster that floods, destroys or displaces on a regional scale
Use medium for analysis of or reaction to such an event, not for the event itself.
Input: numbered lines "index|Title"
Output: [{"i":0,"l":"high","c":"conflict"}, ...]
Focus: geopolitical events, conflicts, disasters, diplomacy.
Classify by real-world event severity, not headline sentiment.`;
const NEWS_THREAT_SUMMARY_KEY = 'news:threat:summary:v1';
// The cycle publishes only at its end, about 12 minutes after it starts (four
// 3-minute variant staggers), and a restart discards the cycle in flight. So
// the first write after a restart lands at least 27 minutes after the last
// one; on 2026-10-21 three deploys in 40 minutes left a 48-minute gap, and the
// old 20-minute TTL expired the key twice. Two hours outlasts that, and stays
// above the 60-minute health budget (api/health.js SEED_META.newsThreatSummary)
// so health reads STALE_SEED before the payload can vanish.
const NEWS_THREAT_SUMMARY_TTL = 2 * 60 * 60;
// Country name → ISO2 for threat summary geo-attribution (inline to avoid ESM import)
const THREAT_COUNTRY_NAME_TO_ISO2 = {
'afghanistan':'AF','albania':'AL','algeria':'DZ','angola':'AO','argentina':'AR',
'armenia':'AM','australia':'AU','austria':'AT','azerbaijan':'AZ','bahrain':'BH',
'bangladesh':'BD','belarus':'BY','belgium':'BE','bolivia':'BO','brazil':'BR',
'burkina faso':'BF','burma':'MM','cambodia':'KH','cameroon':'CM','canada':'CA',
'chad':'TD','chile':'CL','china':'CN','colombia':'CO','congo':'CG',
'costa rica':'CR','croatia':'HR','cuba':'CU','cyprus':'CY',
'czech republic':'CZ','czechia':'CZ',
'democratic republic of the congo':'CD','dr congo':'CD','drc':'CD',
'denmark':'DK','djibouti':'DJ','dominican republic':'DO',
'ecuador':'EC','egypt':'EG','el salvador':'SV','eritrea':'ER',
'estonia':'EE','ethiopia':'ET','finland':'FI','france':'FR',
'georgia':'GE','germany':'DE','ghana':'GH','greece':'GR',
'guatemala':'GT','guinea':'GN','haiti':'HT','honduras':'HN','hungary':'HU',
'iceland':'IS','india':'IN','indonesia':'ID','iran':'IR','iraq':'IQ',
'ireland':'IE','israel':'IL','italy':'IT','ivory coast':'CI',
'jamaica':'JM','japan':'JP','jordan':'JO','kazakhstan':'KZ',
'kenya':'KE','kosovo':'XK','kuwait':'KW','kyrgyzstan':'KG',
'laos':'LA','latvia':'LV','lebanon':'LB','libya':'LY','lithuania':'LT',
'mali':'ML','mauritania':'MR','mexico':'MX','moldova':'MD',
'mongolia':'MN','montenegro':'ME','morocco':'MA','mozambique':'MZ',
'myanmar':'MM','namibia':'NA','nepal':'NP','netherlands':'NL',
'new zealand':'NZ','nicaragua':'NI','niger':'NE','nigeria':'NG',
'north korea':'KP','north macedonia':'MK','norway':'NO',
'oman':'OM','pakistan':'PK','palestine':'PS','panama':'PA',
'paraguay':'PY','peru':'PE','philippines':'PH','poland':'PL',
'portugal':'PT','qatar':'QA','romania':'RO','russia':'RU','rwanda':'RW',
'saudi arabia':'SA','senegal':'SN','serbia':'RS','sierra leone':'SL',
'singapore':'SG','slovakia':'SK','slovenia':'SI','somalia':'SO',
'south africa':'ZA','south korea':'KR','south sudan':'SS','spain':'ES',
'sri lanka':'LK','sudan':'SD','sweden':'SE','switzerland':'CH',
'syria':'SY','taiwan':'TW','tajikistan':'TJ','tanzania':'TZ',
'thailand':'TH','togo':'TG','tunisia':'TN','turkey':'TR',
'turkmenistan':'TM','uganda':'UG','ukraine':'UA',
'united arab emirates':'AE','uae':'AE',
'united kingdom':'GB','uk':'GB','united states':'US','usa':'US',
'uruguay':'UY','uzbekistan':'UZ','venezuela':'VE','vietnam':'VN',
'yemen':'YE','zambia':'ZM','zimbabwe':'ZW',
// Key aliases
'tehran':'IR','moscow':'RU','beijing':'CN','kyiv':'UA','pyongyang':'KP',
'tel aviv':'IL','gaza':'PS','damascus':'SY','sanaa':'YE','houthi':'YE',
'kremlin':'RU','pentagon':'US','nato':'','irgc':'IR','hezbollah':'LB',
'hamas':'PS','taliban':'AF','riyadh':'SA','ankara':'TR',
'saudi':'SA','houthis':'YE',
};
// Sort by name length desc so longer multi-word names match first (used for tie-breaking same position)
const THREAT_COUNTRY_NAME_ENTRIES = Object.entries(THREAT_COUNTRY_NAME_TO_ISO2)
.filter(([name, iso2]) => name.length >= 3 && iso2.length === 2)
.sort((a, b) => b[0].length - a[0].length)
.map(([name, iso2]) => ({ name, iso2, regex: new RegExp(`\\b${name.replace(/[.*+?^${}()|[\]\\]/g, '\\$&')}\\b`, 'i') }));
// Returns the single primary affected country — the country appearing immediately after a
// locative preposition or attack verb, which marks the grammatical object/affected entity.
// Returns [] when no such pattern fires (no attribution is better than wrong attribution).
// "UK and US launch strikes on Yemen" → ['YE']
// "US strikes on Yemen condemned by Iran" → ['YE'] (Iran is a reactor, not affected)
// "Yemen says UK and US strikes hit Hodeidah" → [] (Hodeidah is a city, skip)
// "Russia invades Ukraine" → ['UA']
const AFFECTED_PREFIX_RE = /\b(in|on|against|at|into|across|inside|targeting|toward[s]?|invad(?:es?|ed|ing)|attack(?:s|ed|ing)?|bomb(?:s|ed|ing)?|hitt?(?:ing|s)?|strik(?:es?|ing))\s+(?:the\s+)?/gi;
function matchCountryNamesInText(text) {
const lower = text.toLowerCase();
let match;
AFFECTED_PREFIX_RE.lastIndex = 0;
while ((match = AFFECTED_PREFIX_RE.exec(lower)) !== null) {
const afterPfx = lower.slice(match.index + match[0].length);
for (const { name, iso2 } of THREAT_COUNTRY_NAME_ENTRIES) {
if (afterPfx.startsWith(name) && (afterPfx.length === name.length || /\W/.test(afterPfx[name.length]))) {
return [iso2];
}
}
}
return [];
}
// City and region names -> ISO2 (scripts/build-news-places.mjs), longest phrase first.
const NEWS_PLACES = require('./shared/news-places.json');
const NEWS_PLACE_MAX_WORDS = 4;
// Mirrors stripPublisherSuffix in scripts/shared/geo-extract.mjs.
const THREAT_PUBLISHER_CREDIT_WORDS = /\b(?:post|times|news|journal|herald|tribune|guardian|telegraph|gazette|press|reuters|bloomberg|jazeera|bbc|cnn|nbc|cbs|npr|afp|tv|television|radio|daily|weekly|observer|independent|monitor|chronicle|express|mail|standard|insider|magazine|media|network|wire|agency|today|online)\b|\.(?:com|org|net|gov)\b/i;
const THREAT_CREDIT_TAIL_CHARS = 120;
// Every country a lowercased, diacritic-free text names. Place phrases resolve first and are
// blanked, so "new mexico" is US, not MX; country names are then blanked as they match, so
// "sudan" cannot also fire inside "south sudan".
function countriesNamedInText(lower) {
const tokens = [...lower.matchAll(/[a-z]+/g)];
const found = [];
let rest = lower;
for (let i = 0; i < tokens.length; i++) {
for (let n = Math.min(NEWS_PLACE_MAX_WORDS, tokens.length - i); n >= 1; n--) {
const iso2 = NEWS_PLACES[tokens.slice(i, i + n).map((t) => t[0]).join(' ')];
if (!iso2) continue;
if (!found.includes(iso2)) found.push(iso2);
const end = tokens[i + n - 1].index + tokens[i + n - 1][0].length;
rest = rest.slice(0, tokens[i].index) + ' '.repeat(end - tokens[i].index) + rest.slice(end);
i += n - 1;
break;
}
}
for (const { iso2, regex } of THREAT_COUNTRY_NAME_ENTRIES) {
if (!regex.test(rest)) continue;
if (!found.includes(iso2)) found.push(iso2);
rest = rest.replace(new RegExp(regex.source, 'gi'), (m) => ' '.repeat(m.length));
}
return found;
}
// The threat summary feeds CII's information score, where a headline that names no
// attack target still says something about every country it mentions: "Saudi Arabia
// intercepts missile near Red Sea oil hub" has no target and used to count for nobody.
// Alert routing keeps the strict single-country rule above.
function threatSummaryCountries(text) {
const affected = matchCountryNamesInText(text);
if (affected.length > 0) return affected;
// Uppercase US/UK are the countries; lowercase "us" stays the pronoun.
let lower = text.replace(/\bU\.S\.(?:A\.)?/g, 'United States ').replace(/\bU\.K\./g, 'United Kingdom ')
.replace(/\bUS\b/g, 'United States').replace(/\bUK\b/g, 'United Kingdom')
.normalize('NFKD').replace(/\p{Diacritic}/gu, '').toLowerCase();
// Drop a trailing " - The Washington Post" credit so the outlet's country is not counted.
const tailStart = Math.max(0, lower.length - THREAT_CREDIT_TAIL_CHARS);
const credit = /\s[-–—|]\s+([^-–—|]+)$/.exec(lower.slice(tailStart));
if (credit) {
const tail = credit[1].trim();
const isCredit = tail.split(/\s+/).length <= 6
&& (THREAT_PUBLISHER_CREDIT_WORDS.test(tail) || countriesNamedInText(tail).length === 0);
if (isCredit) lower = lower.slice(0, tailStart + credit.index);
}
return countriesNamedInText(lower);
}
// v5 (2026-04-28): bumped from v4 in lockstep with
// server/worldmonitor/intelligence/v1/_shared.ts and
// server/worldmonitor/news/v1/list-feed-digest.ts to evict cache entries
// that landed under the pre-publisher-prefix-fix classifier (PR #3480).
// Brand-prefixed retrospective titles ("CBS News Radio flashback: ...")
// had been promoted to severity=critical via the `invasion` keyword;
// the new brand-prefix branch in _classifier.ts re-rules those rows on
// next touch.
// The relay maintains its own inline helper because .cjs cannot import
// from .ts; the prefix-audit static-analysis test
// (tests/news-classify-cache-prefix-audit.test.mjs) cross-checks all
// three sites.
function classifyCacheKey(title) {
const hash = crypto.createHash('sha256').update(title.toLowerCase()).digest('hex').slice(0, 16);
return `classify:sebuf:v6:${hash}`;
}
// LLM provider fallback chain — mirrors seed-insights.mjs LLM_PROVIDERS
// Order mirrors server/_shared/llm.ts: paid OpenRouter, then two fixed free
// OpenRouter variants.
const CLASSIFY_LLM_PROVIDERS = [
{
name: 'ollama',
envKey: 'OLLAMA_API_URL',
apiUrlFn: (baseUrl) => new URL('/v1/chat/completions', baseUrl).toString(),
model: () => process.env.OLLAMA_MODEL || 'llama3.1:8b',
headers: (_key) => {
const h = { 'Content-Type': 'application/json', 'User-Agent': CHROME_UA };
const apiKey = process.env.OLLAMA_API_KEY;
if (apiKey) h.Authorization = `Bearer ${apiKey}`;
return h;
},
extraBody: { reasoning_effort: 'none' },
timeout: 30000,
},
{
name: 'openrouter',
envKey: 'OPENROUTER_API_KEY',
apiUrl: 'https://openrouter.ai/api/v1/chat/completions',
// Classification only — NOT the shared Flash default. Against 413 blind-judged
// headlines v4-flash raised 51 false critical/high labels for 42 real ones; v4.1
// with the "Do not under-rate high" prompt block raised 15-21 for 41 (three runs).
// Both halves are load-bearing: v4.1 without the block misses 7 of 44 real alerts
// instead of 3, and the block on v4-flash still raises 56 false ones.
// Pinned to that evidence by tests/classify-alert-label-precision.test.mjs;
// re-measure with scripts/eval-classify-labels.mjs before changing either.
model: 'deepseek/deepseek-v4.1-flash',
headers: (key) => ({ Authorization: `Bearer ${key}`, 'Content-Type': 'application/json', 'HTTP-Referer': 'https://worldmonitor.app', 'X-Title': 'World Monitor', 'User-Agent': CHROME_UA }),
extraBody: { reasoning: { enabled: false }, provider: OPENROUTER_PROVIDER_ROUTING },
timeout: 30000,
},
{
name: 'openrouter-free',
envKey: 'OPENROUTER_API_KEY',
apiUrl: 'https://openrouter.ai/api/v1/chat/completions',
model: OPENROUTER_FREE_PRIMARY_MODEL,
headers: (key) => ({ Authorization: `Bearer ${key}`, 'Content-Type': 'application/json', 'HTTP-Referer': 'https://worldmonitor.app', 'X-Title': 'World Monitor', 'User-Agent': CHROME_UA }),
extraBody: { reasoning: { enabled: false }, provider: OPENROUTER_PROVIDER_ROUTING },
timeout: 30000,
},
{
name: 'openrouter-free-backup',
envKey: 'OPENROUTER_API_KEY',
apiUrl: 'https://openrouter.ai/api/v1/chat/completions',
model: OPENROUTER_FREE_BACKUP_MODEL,
headers: (key) => ({ Authorization: `Bearer ${key}`, 'Content-Type': 'application/json', 'HTTP-Referer': 'https://worldmonitor.app', 'X-Title': 'World Monitor', 'User-Agent': CHROME_UA }),
extraBody: { reasoning: { enabled: false }, provider: OPENROUTER_PROVIDER_ROUTING },
timeout: 30000,
},
];
// options.systemPrompt / options.maxTokensPerItem let another index-aligned
// classifier (the Middle East strike extractor) reuse this provider chain.
function classifyFetchLlmSingle(titles, _apiKey, apiUrl, model, headers, extraBody, timeout, maxTextChars = 200, options = {}) {
return new Promise((resolve) => {
const sanitized = titles.map((t) => t.replace(/[\n\r]/g, ' ').replace(/\|/g, '/').slice(0, maxTextChars).trim());
const prompt = sanitized.map((t, i) => `${i}|${t}`).join('\n');
const bodyStr = JSON.stringify({
model,
messages: [
{ role: 'system', content: options.systemPrompt || CLASSIFY_SYSTEM_PROMPT },
{ role: 'user', content: prompt },
],
temperature: 0,
max_tokens: titles.length * (options.maxTokensPerItem || 40),
...extraBody,
});
const parsed = new URL(apiUrl);
const transport = parsed.protocol === 'http:' ? http : https;
const req = transport.request(parsed, {
signal: AbortSignal.timeout(timeout),
method: 'POST',
headers: { ...headers, 'Content-Length': Buffer.byteLength(bodyStr) },
timeout,
}, (resp) => {
if (resp.statusCode < 200 || resp.statusCode >= 300) {
resp.resume();
return resolve(null);
}
let data = '';
resp.on('data', (chunk) => { data += chunk; });
resp.on('end', () => {
try {
const json = JSON.parse(data);
const raw = json?.choices?.[0]?.message?.content?.trim();
if (!raw) return resolve(null);
const match = raw.match(/\[[\s\S]*\]/);
if (!match) return resolve(null);
resolve(JSON.parse(match[0]));
} catch { resolve(null); }
});
});
req.on('error', () => resolve(null));
req.on('timeout', () => { req.destroy(); resolve(null); });
req.end(bodyStr);
});
}
async function classifyFetchLlm(titles, maxTextChars = 200, options = {}) {
for (const provider of CLASSIFY_LLM_PROVIDERS) {
const envVal = process.env[provider.envKey];
if (!envVal) continue;
const apiUrl = provider.apiUrlFn ? provider.apiUrlFn(envVal) : provider.apiUrl;
const model = typeof provider.model === 'function' ? provider.model() : provider.model;
const headers = provider.headers(envVal);
const result = await classifyFetchLlmSingle(titles, envVal, apiUrl, model, headers, provider.extraBody || {}, provider.timeout, maxTextChars, options);
if (result) {
return result;
}
console.warn(`[${options.label || 'Classify'}] ${provider.name} failed, trying next provider...`);
}
return null;
}
// Jev shadow (scripts/lib/jev-classify-relay.cjs): observes the labels the LLM
// chain already cached, and records disagreements. It decides nothing, and is
// inert without TYPESAFE_API_KEY.
const jevShadow = require('./lib/jev-classify-relay.cjs');
const JEV_SHADOW_PUSH_SCRIPT = "redis.call('LPUSH', KEYS[1], ARGV[1]) redis.call('LTRIM', KEYS[1], 0, tonumber(ARGV[2]) - 1) redis.call('EXPIRE', KEYS[1], tonumber(ARGV[3])) return 1";
const observeJevShadow = jevShadow.createShadowObserver({
fetchJevLabel: (title, maxTextChars) => jevShadow.fetchJevLabel(title, maxTextChars, { apiKey: jevShadow.jevApiKey() }),
record: (row) => upstashEval(JEV_SHADOW_PUSH_SCRIPT, [jevShadow.SHADOW_LOG_KEY], [JSON.stringify(row), jevShadow.SHADOW_LOG_MAX, jevShadow.SHADOW_LOG_TTL_S]),
});
let classifyInFlight = false;
async function seedClassifyForVariant(variant, seenTitles) {
const digestUrl = buildClassifyDigestUrl(variant);
const transport = classifyDigestTransport(digestUrl) === 'http' ? http : https;
let digest;
const maxDigestAttempts = CLASSIFY_DIGEST_RETRY_DELAYS_MS.length + 1;
for (let attempt = 0; attempt < maxDigestAttempts; attempt++) {
try {
const resp = await new Promise((resolve, reject) => {
const req = transport.get(digestUrl, {
signal: AbortSignal.timeout(15000),
headers: buildClassifyDigestHeaders({
userAgent: CHROME_UA,
relayKey: RELAY_API_KEY,
}),
timeout: 15000,
}, resolve);
req.on('error', reject);
req.on('timeout', () => { req.destroy(); reject(new Error('timeout')); });
});
if (resp.statusCode !== 200) {
resp.resume();
const retry = classifyDigestRetryDecision({ attempt, status: resp.statusCode });
if (retry.retry) {
console.warn(`[Classify] digest fetch HTTP ${resp.statusCode}; retrying in ${retry.delayMs}ms`);
await new Promise((r) => setTimeout(r, retry.delayMs));
continue;
}
console.warn(formatClassifyDigestFetchFailure(resp.statusCode, RELAY_API_KEY));
return { total: 0, classified: 0, skipped: 0, fetchFailed: true };
}
let body = '';
resp.setEncoding('utf8');
for await (const chunk of resp) body += chunk;
digest = JSON.parse(body);
break;
} catch (e) {
const retry = classifyDigestRetryDecision({ attempt, error: e });
if (retry.retry) {
console.warn(`[Classify] digest fetch error: ${e?.message || e}; retrying in ${retry.delayMs}ms`);
await new Promise((r) => setTimeout(r, retry.delayMs));
continue;
}
console.warn('[Classify] digest fetch error:', e?.message || e);
return { total: 0, classified: 0, skipped: 0, fetchFailed: true };
}
}
if (!digest) {
console.warn('[Classify] digest fetch error: exhausted retries');
return { total: 0, classified: 0, skipped: 0, fetchFailed: true };
}
// #7084: stale RSS titles already had their alert pass when served fresh.
// Exclude only those digest-derived candidates; fresh X candidates remain
// eligible because they are an independent input to the combined pass.
// The digest API answers 200 with `coverage.state: 'unavailable'` and no
// categories when it has no digest to serve; that is no news, not quiet news.
const staleReplay = isStaleDigestReplay(digest) || digest?.coverage?.state === 'unavailable';
if (staleReplay) {
console.log(`[Classify] digest is a stale replay or unavailable (${digest.coverage.staleReason || 'unknown'}, ${digest.coverage.staleAgeSeconds ?? 0}s) — skipping digest-derived candidates for ${variant}`);
}
// Map of title → item metadata; recency gate: skip articles older than 6h.
// The pure helper keeps the stale-RSS plus fresh-X behavior executable in
// tests without importing this boot-on-require relay.
const RECENCY_GATE_MS = 6 * 60 * 60 * 1000;
const classifyNow = Date.now();
const xCandidates = xNewsAccounts.collectXAlertCandidates(
xState.items,
RELAY_SOURCE_TIERS,
classifyNow,
RECENCY_GATE_MS,
);
const allTitles = buildClassifyCandidateMap(digest, xCandidates, variant, classifyNow, RECENCY_GATE_MS);
if (allTitles.size === 0) return { total: 0, classified: 0, skipped: 0, staleReplay };
const titleArr = [...allTitles.keys()];
const cacheKeys = titleArr.map((t) => classifyCacheKey(t));
const cached = await upstashMGet(cacheKeys);
const misses = [];
// byCountry accumulates threat counts while title+level are in scope
const byCountry = {};
const emptyLevel = () => ({ critical: 0, high: 0, medium: 0, low: 0, info: 0 });
// Country-naming titles with no usable label (a cached `_skip`, an unreadable
// entry, or one the LLM skipped below), so an empty byCountry cannot be read
// as a quiet cycle. A title that names no country cannot change byCountry.
let unlabelled = 0;
const countUnlabelled = (title) => { if (threatSummaryCountries(title).length > 0) unlabelled++; };
for (let i = 0; i < titleArr.length; i++) {
const hit = cached[i];
if (!hit) {
misses.push(titleArr[i]);
continue;
}
// Attribute cached hits while we still have the title
let parsed = hit;
if (typeof hit === 'string') { try { parsed = JSON.parse(hit); } catch { countUnlabelled(titleArr[i]); continue; } }
const level = parsed?.level;
if (!CLASSIFY_VALID_LEVELS.includes(level)) { countUnlabelled(titleArr[i]); continue; }
if (seenTitles.has(titleArr[i])) continue;
seenTitles.add(titleArr[i]);
for (const code of threatSummaryCountries(titleArr[i])) {
if (!byCountry[code]) byCountry[code] = emptyLevel();
byCountry[code][level]++;
}
}
if (misses.length === 0) return { total: titleArr.length, classified: 0, skipped: 0, byCountry, staleReplay, unlabelled };
let classified = 0;
let skipped = 0;
const shadow = { asked: 0, answered: 0, agreed: 0, alertFlips: 0 };
for (let b = 0; b < misses.length; b += CLASSIFY_BATCH_SIZE) {
const chunk = misses.slice(b, b + CLASSIFY_BATCH_SIZE);
const llmResult = await classifyFetchLlm(chunk);
if (!Array.isArray(llmResult)) {
for (const title of chunk) {
await upstashSet(classifyCacheKey(title), { level: '_skip', timestamp: Date.now() }, CLASSIFY_SKIP_TTL);
skipped++;
countUnlabelled(title);
}
continue;
}
const classifiedSet = new Set();
const labelled = [];
for (const entry of llmResult) {
const idx = entry?.i;
if (typeof idx !== 'number' || idx < 0 || idx >= chunk.length) continue;
if (classifiedSet.has(idx)) continue;
const level = CLASSIFY_VALID_LEVELS.includes(entry?.l) ? entry.l : null;
const category = CLASSIFY_VALID_CATEGORIES.includes(entry?.c) ? entry.c : null;
if (!level || !category) continue;
classifiedSet.add(idx);
await upstashSet(classifyCacheKey(chunk[idx]), { level, category, timestamp: Date.now() }, CLASSIFY_CACHE_TTL);
classified++;
labelled.push({ title: chunk[idx], level });
// Attribute newly classified title to country stats (global dedup via seenTitles)
if (!seenTitles.has(chunk[idx])) {
seenTitles.add(chunk[idx]);
for (const code of threatSummaryCountries(chunk[idx])) {
if (!byCountry[code]) byCountry[code] = emptyLevel();
byCountry[code][level]++;
}
}
// Notifications are outside seenTitles guard — each variant publishes
// independently, protected by the variant-scoped Redis scan-dedup key.
if (level === 'critical' || level === 'high') {
const meta = allTitles.get(chunk[idx]) ?? {
source: variant,
publishedAt: Date.now(),
corroborationCount: 1,
link: '',
};
// Relay gates: when RELAY_GATES_READY is set the relay enforces source tier and
// recency checks that the client path previously handled.
// Explicit tier-4 keys only — unlisted names (including platform source
// "telegram") are NOT in this set even though getSourceTier() defaults
// them to 4. Any future Telegram alert path must use the public display
// label from shared/telegram-channel-trust.ts (#6600). #6654 should do
// the same for X account labels rather than a generic "x" platform key.
if (shouldDropRelaySourceForTier(RELAY_GATES_READY, meta.source, RELAY_TIER4_SOURCES)) continue;
if (RELAY_GATES_READY) {
const ageMs = Date.now() - (meta.publishedAt ?? 0);
if (meta.publishedAt && ageMs > RELAY_RECENCY_MS) continue;
}
// Recompute importanceScore from the post-LLM level. Publishing the
// digest's pre-LLM keyword-based score would leak a stale value —
// see docs/internal/scoringDiagnostic.md §2.
const importanceScore = relayComputeImportanceScore(
level,
meta.source,
meta.corroborationCount ?? 1,
meta.publishedAt ?? Date.now(),
{
title: chunk[idx],
classSource: 'llm',
// The relay has only exact story-merge corroboration. Entity
// corroboration is a separate digest-side signal computed from
// flashpoint+diplomacy buckets; do not proxy source count here.
entityCorroborationCount: 0,
},
);
publishNotificationEvent({
eventType: 'rss_alert',
payload: {
title: chunk[idx],
source: meta.source,
link: meta.link,
publishedAt: meta.publishedAt,
importanceScore,
corroborationCount: meta.corroborationCount ?? 1,
// Read by API webhook subscribers (news.breaking scope). Plural on
// purpose: the relay scopes rules on payload.countryCode and keeps
// rss_alert permissive-unattributed, so a singular field here would
// start silencing country-scoped Slack rules.
category,
countryCodes: matchCountryNamesInText(chunk[idx]),
},
severity: level,
variant,
}).catch(e => console.warn('[Notify] Classify publish error:', e?.message));
}
}
for (let i = 0; i < chunk.length; i++) {
if (!classifiedSet.has(i)) {
await upstashSet(classifyCacheKey(chunk[i]), { level: '_skip', timestamp: Date.now() }, CLASSIFY_SKIP_TTL);
skipped++;
countUnlabelled(chunk[i]);
}
}
// Last, and read-only: every label above is already cached and published.
const observed = await observeJevShadow(variant, labelled).catch(() => null);
if (observed) {
for (const k of Object.keys(shadow)) shadow[k] += observed[k];
}
}
if (shadow.asked > 0) {
console.log(`[Classify] Jev shadow ${variant}: asked ${shadow.asked}, answered ${shadow.answered}, agreed ${shadow.agreed}, alert flips ${shadow.alertFlips}`);
}
return { total: titleArr.length, classified, skipped, byCountry, staleReplay, unlabelled };
}
async function seedClassify() {
if (classifyInFlight) return;
classifyInFlight = true;
const t0 = Date.now();
try {
const hasAnyProvider = CLASSIFY_LLM_PROVIDERS.some((p) => !!process.env[p.envKey]);
if (!hasAnyProvider) {
console.log('[Classify] Skipped — no LLM provider keys configured');
return;
}
try {
await publishSaudiCivilDefenseAlerts(telegramState.items, {
now: Date.now,
readCache: upstashGet,
writeCache: upstashSet,
classify: (posts) => classifyFetchLlm(posts, MAX_POST_CHARS),
publish: (event) => publishNotificationEvent({
...event,
payload: {
...event.payload,
importanceScore: relayComputeImportanceScore(
event.severity, event.payload.source, 1, event.payload.publishedAt,
{ title: event.payload.title, classSource: 'llm', entityCorroborationCount: 0 },
),
},
}),
});
} catch (e) {
console.warn('[Classify] Saudi Civil Defense alerts failed:', e?.message || e);
}
let totalClassified = 0;
let totalSkipped = 0;
let fetchOk = 0;
let fetchFailed = 0;
let staleReplays = 0;
let unlabelled = 0;
let variantErrors = 0;
const mergedByCountry = {};
const seenTitles = new Set();
for (let v = 0; v < CLASSIFY_VARIANTS.length; v++) {
if (v > 0) await new Promise((r) => setTimeout(r, CLASSIFY_VARIANT_STAGGER_MS));
try {
const stats = await seedClassifyForVariant(CLASSIFY_VARIANTS[v], seenTitles);
if (stats.fetchFailed) fetchFailed += 1;
else fetchOk += 1;
if (stats.staleReplay) staleReplays += 1;
unlabelled += stats.unlabelled || 0;
totalClassified += stats.classified;
totalSkipped += stats.skipped;
console.log(`[Classify] ${CLASSIFY_VARIANTS[v]}: ${stats.total} titles, ${stats.classified} classified, ${stats.skipped} skipped`);
for (const [code, counts] of Object.entries(stats.byCountry || {})) {
if (!mergedByCountry[code]) mergedByCountry[code] = { critical: 0, high: 0, medium: 0, low: 0, info: 0 };
for (const lvl of ['critical', 'high', 'medium', 'low', 'info']) {
mergedByCountry[code][lvl] += counts[lvl] || 0;
}
}
} catch (e) {
variantErrors += 1;
console.warn(`[Classify] ${CLASSIFY_VARIANTS[v]} error:`, e?.message || e);
}
}
if (!shouldWriteClassifySeedMeta({ fetchOk, fetchFailed })) {
console.warn('[Classify] Digest fetch failed for every variant — not writing seed-meta:classify');
return;
}
// Written on every cycle that reached a digest, including one where no
// headline named a country (`byCountry: {}`): the CII reads an empty summary
// as a quiet input and a missing one as a dead input. An empty result is
// quiet only when every headline was fetched fresh and labelled: after a
// failed digest fetch, a variant that threw, a stale or unavailable digest
// (headlines skipped above) or country-naming titles the LLM could not
// label, the countries it would have named are unknown, so the last summary stays and its seed-meta ages. seed-meta advances only
// after the payload lands, so health never vouches for a missing key.
const threatCountryCount = Object.keys(mergedByCountry).length;
const threatSummaryAt = Date.now();
const threatSummaryDegraded = threatCountryCount === 0 && (fetchFailed > 0 || variantErrors > 0 || staleReplays > 0 || unlabelled > 0);
const threatSummaryWritten = !threatSummaryDegraded && await envelopeWrite(NEWS_THREAT_SUMMARY_KEY, { byCountry: mergedByCountry, generatedAt: threatSummaryAt }, NEWS_THREAT_SUMMARY_TTL, { fetchedAt: threatSummaryAt, recordCount: threatCountryCount, zeroOk: true, sourceVersion: 'news-threat-summary' });
if (threatSummaryWritten) {
await upstashSet('seed-meta:news:threat-summary', { fetchedAt: threatSummaryAt, recordCount: threatCountryCount }, 604800);
console.log(`[Classify] Threat summary written for ${threatCountryCount} countries`);
} else if (threatSummaryDegraded) {
console.warn(`[Classify] No country matches from a degraded cycle (${fetchFailed} digest fetch failure(s), ${variantErrors} variant error(s), ${staleReplays} stale replay(s), ${unlabelled} unlabelled title(s)) — keeping the last threat summary`);
} else {
console.warn('[Classify] Threat summary write failed — seed-meta:news:threat-summary not advanced');
}
await upstashSet('seed-meta:classify', { fetchedAt: Date.now(), recordCount: totalClassified }, 604800);
console.log(`[Classify] Done in ${((Date.now() - t0) / 1000).toFixed(1)}s — ${totalClassified} classified, ${totalSkipped} skipped`);
} catch (e) {
console.warn('[Classify] Seed error:', e?.message || e);
} finally {
classifyInFlight = false;
}
}
async function startClassifySeedLoop() {
if (!UPSTASH_ENABLED) {
console.log('[Classify] Disabled (no Upstash Redis)');
return;
}
const activeProviders = CLASSIFY_LLM_PROVIDERS.filter((p) => !!process.env[p.envKey]).map((p) => p.name);
console.log(`[Classify] Seed loop starting (interval ${CLASSIFY_SEED_INTERVAL_MS / 1000 / 60}min, providers:${activeProviders.length ? activeProviders.join(',') : 'none'})`);
startBootSeedLoop('Classify', 'seed-meta:classify', CLASSIFY_SEED_INTERVAL_MS, seedClassify, (e) => console.warn('[Classify] Initial seed error:', e?.message || e), (e) => console.warn('[Classify] Seed error:', e?.message || e));
}
// ─────────────────────────────────────────────────────────────
// News digest warm cycle — requests every (edition, language) digest the
// catalog places a feed in, so every catalog feed is fetched whether or not a
// reader requests its language (scripts/lib/news-digest-warm.cjs).
// ─────────────────────────────────────────────────────────────
const DIGEST_WARM_CELLS = requireShared('news-digest-warm-cells.json');
let digestWarmInFlight = false;
async function seedDigestWarm() {
if (digestWarmInFlight) {
console.log('[DigestWarm] Previous cycle still running, skipping');
return;
}
digestWarmInFlight = true;
const startedAt = Date.now();
try {
const results = await runDigestWarmCycle(DIGEST_WARM_CELLS, {
fetchCell: (cell) => fetchDigestWarmCell(cell, { userAgent: CHROME_UA, relayKey: RELAY_API_KEY }),
});
const summary = summarizeDigestWarmCycle(results, { fetchedAt: Date.now(), durMs: Date.now() - startedAt });
console.log(
`[DigestWarm] ${summary.cells} cells: ${summary.fresh} fresh, ${summary.stale} stale, ${summary.failed} failed ` +
`in ${Math.round(summary.durMs / 1000)}s${summary.failures.length ? ` (${summary.failures.join(', ')})` : ''}`,
);
if (!shouldPublishDigestWarmHeartbeat(summary)) {
console.warn(`[DigestWarm] Fewer than half the cells served news${RELAY_API_KEY ? '' : ' (WORLDMONITOR_RELAY_KEY not set)'} — heartbeat not written`);
return;
}
if (!await upstashSet(DIGEST_WARM_HEARTBEAT_KEY, summary, DIGEST_WARM_HEARTBEAT_TTL_SECONDS)) return;
// Durable, no TTL: health reads the probe as pending until the first
// healthy cycle and strict afterwards (api/health.js ACTIVATION_MARKERS).
await upstashEval("return redis.call('SET', KEYS[1], ARGV[1])", [DIGEST_WARM_ACTIVATION_KEY], [String(summary.fetchedAt)]);
} finally {
digestWarmInFlight = false;
}
}
function startDigestWarmLoop() {
if (!UPSTASH_ENABLED) {
console.log('[DigestWarm] Disabled (no Upstash Redis)');
return;
}
console.log(`[DigestWarm] Warm loop starting (${DIGEST_WARM_CELLS.length} cells, interval ${DIGEST_WARM_INTERVAL_MS / 1000 / 60}min)`);
startBootSeedLoop('DigestWarm', DIGEST_WARM_HEARTBEAT_KEY, DIGEST_WARM_INTERVAL_MS, seedDigestWarm, (e) => console.warn('[DigestWarm] Initial cycle error:', e?.message || e), (e) => console.warn('[DigestWarm] Cycle error:', e?.message || e));
}
// ─────────────────────────────────────────────────────────────
// Middle East strike feed — Telegram + GDELT -> conflict:iran-events:v1
// (logic and tests: scripts/lib/middle-east-strikes.cjs). Replaces the manual
// LiveUAMap dump. Off unless IRAN_EVENTS_ENABLED=true, the same switch every
// reader of the key honours (api/health.js, api/bootstrap.js, ...).
// ─────────────────────────────────────────────────────────────
const MIDDLE_EAST_STRIKES_ENABLED = (process.env.IRAN_EVENTS_ENABLED ?? 'false').toLowerCase() === 'true';
const MIDDLE_EAST_STRIKES_INTERVAL_MS = 10 * 60 * 1000;
// Canonical outlives the 72h window it carries; health's 45-minute seed-meta
// budget (api/health.js SEED_META.iranEvents) is what alarms on a dead loop.
const MIDDLE_EAST_STRIKES_CANONICAL_TTL_S = 72 * 3600;
const MIDDLE_EAST_STRIKES_STORE_TTL_S = 4 * 24 * 3600;
const MIDDLE_EAST_STRIKES_META_TTL_S = 7 * 24 * 3600;
// The 30-day strike-pressure tally (CII v12) outlives its window by 5 days, so
// a relay outage shorter than that loses no history.
const MIDDLE_EAST_STRIKE_PRESSURE_TTL_S = 35 * 24 * 3600;
const MIDDLE_EAST_STRIKES_PENDING_MAX = 1000;
const MIDDLE_EAST_STRIKES_PENDING_MAX_AGE_MS = 12 * 3600 * 1000;
const MIDDLE_EAST_STRIKES_DONE_MAX = 5000;
const middleEastStrikePending = new Map();
// Posts already classified (strike or not): the feed buffer re-offers them
// every cycle for ~90 minutes, and each re-offer would cost a cache read.
const middleEastStrikeDone = new Set();
const MIDDLE_EAST_STRIKES_MAX_MISSES = 3;
const middleEastStrikeMisses = new Map();
let middleEastStrikesInFlight = false;
let middleEastGdeltBulkPromise = null;
function noteMiddleEastStrikeItem(item) {
if (!MIDDLE_EAST_STRIKES_ENABLED || !middleEastStrikes || !item?.id || middleEastStrikePending.has(item.id) || middleEastStrikeDone.has(item.id)) return;
if (!middleEastStrikes.prefilterStrikeText(item.text)) return;
middleEastStrikePending.set(item.id, item);
while (middleEastStrikePending.size > MIDDLE_EAST_STRIKES_PENDING_MAX) {
const evicted = middleEastStrikePending.keys().next().value;
middleEastStrikePending.delete(evicted);
middleEastStrikeMisses.delete(evicted);
}
}
async function loadMiddleEastStrikeGdelt(processed) {
// Lazy and caught: a missing module degrades to a Telegram-only cycle
// instead of an unhandled rejection at relay start.
middleEastGdeltBulkPromise ??= import('./_conflict-gdelt-bulk.mjs');
let bulk;
try {
bulk = await middleEastGdeltBulkPromise;
} catch (e) {
middleEastGdeltBulkPromise = null; // a cached rejection would disable GDELT until restart
console.warn('[MiddleEastStrikes] GDELT module failed to load:', e?.message || e);
return { candidates: [], processed, errors: [String(e?.message || e)], fetched: 0 };
}
try {
return await middleEastStrikes.fetchGdeltStrikeCandidates({ bulk, processed });
} catch (e) {
console.warn('[MiddleEastStrikes] GDELT backstop unavailable:', e?.message || e);
return { candidates: [], processed, errors: [String(e?.message || e)], fetched: 0 };
}
}
async function seedMiddleEastStrikes() {
if (middleEastStrikesInFlight) return;
middleEastStrikesInFlight = true;
const t0 = Date.now();
try {
for (const item of telegramState.items) noteMiddleEastStrikeItem(item);
for (const [id, item] of middleEastStrikePending) {
const ts = Date.parse(item.ts);
if (!Number.isFinite(ts) || t0 - ts > MIDDLE_EAST_STRIKES_PENDING_MAX_AGE_MS) { middleEastStrikePending.delete(id); middleEastStrikeMisses.delete(id); }
}
const pendingItems = [...middleEastStrikePending.values()];
const { status, doneIds, result, pressureStatus } = await middleEastStrikes.publishStrikeCycle({
now: t0,
pendingItems,
readStore: async () => {
let ok = true;
const value = await upstashGet(middleEastStrikes.STORE_KEY, () => { ok = false; });
return { ok, value };
},
writeStore: (store) => upstashSet(middleEastStrikes.STORE_KEY, store, MIDDLE_EAST_STRIKES_STORE_TTL_S),
writeCanonical: (payload, recordCount) => envelopeWrite(middleEastStrikes.CANONICAL_KEY, payload, MIDDLE_EAST_STRIKES_CANONICAL_TTL_S, {
recordCount,
sourceVersion: middleEastStrikes.SOURCE_VERSION,
zeroOk: true,
}),
writeSeedMeta: (meta) => upstashSet(middleEastStrikes.SEED_META_KEY, meta, MIDDLE_EAST_STRIKES_META_TTL_S),
readCache: (key) => upstashGet(key),
writeCache: (key, value, ttl) => upstashSet(key, value, ttl),
classify: (texts) => classifyFetchLlm(texts, middleEastStrikes.CLASSIFY_MAX_TEXT_CHARS, {
systemPrompt: middleEastStrikes.STRIKE_CLASSIFIER_PROMPT,
maxTokensPerItem: 60,
label: 'MiddleEastStrikes',
}),
loadGdelt: loadMiddleEastStrikeGdelt,
telegramPolledAt: telegramState.lastSuccessfulPollAt,
readPressure: async () => {
let ok = true;
const value = await upstashGet(middleEastStrikes.PRESSURE_KEY, () => { ok = false; });
return { ok, value };
},
writePressure: (pressure) => upstashSet(middleEastStrikes.PRESSURE_KEY, pressure, MIDDLE_EAST_STRIKE_PRESSURE_TTL_S),
writePressureMeta: async (meta) => {
if (!await upstashSet(middleEastStrikes.PRESSURE_SEED_META_KEY, meta, MIDDLE_EAST_STRIKES_META_TTL_S)) return false;
// Durable, no TTL: health reads the key as pending until the first
// healthy tally publish and strict afterwards (api/health.js ACTIVATION_MARKERS).
await upstashEval("return redis.call('SET', KEYS[1], ARGV[1])", [middleEastStrikes.PRESSURE_ACTIVATION_KEY], [String(t0)]);
return true;
},
});
for (const id of doneIds) {
middleEastStrikePending.delete(id);
middleEastStrikeMisses.delete(id);
middleEastStrikeDone.add(id);
}
// A post the classifier keeps answering around (it answers the batch but
// never this index) is dropped after a few cycles, so one poison post
// cannot hold the feed unhealthy. A dead provider chain is not counted:
// those posts wait for the classifier to come back.
if (result && !result.classifierFailed) {
for (const item of result.pending) {
const misses = (middleEastStrikeMisses.get(item.id) || 0) + 1;
if (misses < MIDDLE_EAST_STRIKES_MAX_MISSES) {
middleEastStrikeMisses.set(item.id, misses);
continue;
}
console.warn(`[MiddleEastStrikes] Dropping ${item.id}: unclassified after ${misses} answered cycles`);
middleEastStrikePending.delete(item.id);
middleEastStrikeMisses.delete(item.id);
middleEastStrikeDone.add(item.id);
}
}
while (middleEastStrikeDone.size > MIDDLE_EAST_STRIKES_DONE_MAX) {
middleEastStrikeDone.delete(middleEastStrikeDone.values().next().value);
}
if (!result) {
console.warn(`[MiddleEastStrikes] ${status} — cycle skipped, seed-meta not advanced`);
return;
}
const s = result.stats;
const summary = `${s.events} events, ${s.prefiltered} posts (${s.cacheHits} cached, ${s.classified} classified, ${s.retry} retry), ${s.gdeltCandidates} GDELT clusters from ${s.gdeltExportsFetched} exports, ${s.storeCandidates} stored, telegramLive=${s.telegramLive}, pressure=${pressureStatus}`;
if (status === 'ok') console.log(`[MiddleEastStrikes] ${summary} in ${((Date.now() - t0) / 1000).toFixed(1)}s`);
else console.warn(`[MiddleEastStrikes] ${status} — seed-meta not advanced; ${summary}`);
} catch (e) {
console.warn('[MiddleEastStrikes] Seed error:', e?.message || e);
} finally {
middleEastStrikesInFlight = false;
}
}
function startMiddleEastStrikesLoop() {
if (!MIDDLE_EAST_STRIKES_ENABLED) {
console.log('[MiddleEastStrikes] Disabled (IRAN_EVENTS_ENABLED is not true)');
return;
}
if (!middleEastStrikes) {
console.error('[MiddleEastStrikes] Disabled (module failed to load; see the startup error) — health will report STALE');
return;
}
if (!UPSTASH_ENABLED) {
console.log('[MiddleEastStrikes] Disabled (no Upstash Redis)');
return;
}
console.log(`[MiddleEastStrikes] Loop starting (interval ${MIDDLE_EAST_STRIKES_INTERVAL_MS / 60000}min)`);
startBootSeedLoop('MiddleEastStrikes', middleEastStrikes.SEED_META_KEY, MIDDLE_EAST_STRIKES_INTERVAL_MS, seedMiddleEastStrikes,
(e) => console.warn('[MiddleEastStrikes] Initial seed error:', e?.message || e),
(e) => console.warn('[MiddleEastStrikes] Seed error:', e?.message || e));
}
// ─────────────────────────────────────────────────────────────
// Service Statuses Seed — warm-pings Vercel RPC every 15 min
// so service statuses are always cached (TTL is 30 min).
// ─────────────────────────────────────────────────────────────
const SERVICE_STATUSES_SEED_INTERVAL_MS = 15 * 60 * 1000; // 15 min (TTL/2)
const SERVICE_STATUSES_RPC_URL = 'https://api.worldmonitor.app/api/infrastructure/v1/list-service-statuses';
async function seedServiceStatuses() {
try {
const resp = await fetch(SERVICE_STATUSES_RPC_URL, {
method: 'POST',
headers: warmPingHeaders({ 'Content-Type': 'application/json' }),
body: '{}',
signal: AbortSignal.timeout(60_000),
});
if (!resp.ok) {
console.warn(`[ServiceStatuses] Seed ping failed: HTTP ${resp.status}${RELAY_API_KEY ? '' : ' (WORLDMONITOR_RELAY_KEY not set — 401 expected; set it on the relay AND the Vercel api project)'}`);
return;
}
const data = await resp.json();
const count = data?.statuses?.length || 0;
console.log(`[ServiceStatuses] Seed ping OK — ${count} statuses`);
// seed-meta is written by listServiceStatuses handler only when fresh data
// is scraped; writing it here would mark fallback responses as fresh.
} catch (e) {
console.warn('[ServiceStatuses] Seed ping error:', e?.message || e);
}
}
function startServiceStatusesSeedLoop() {
console.log(`[ServiceStatuses] Seed loop starting (interval ${SERVICE_STATUSES_SEED_INTERVAL_MS / 1000 / 60}min)`);
startBootSeedLoop('ServiceStatuses', 'seed-meta:infra:service-statuses', SERVICE_STATUSES_SEED_INTERVAL_MS, seedServiceStatuses, (e) => console.warn('[ServiceStatuses] Initial seed error:', e?.message || e), (e) => console.warn('[ServiceStatuses] Seed error:', e?.message || e));
}
// ─────────────────────────────────────────────────────────────
// Theater Posture Seed — fetches OpenSky directly via localhost
// proxy, computes military postures, writes to Redis.
// Eliminates circular dependency on Vercel RPC.
// ─────────────────────────────────────────────────────────────
const THEATER_POSTURE_SEED_INTERVAL_MS = 600_000; // 10 min
const THEATER_POSTURE_LIVE_KEY = 'theater-posture:sebuf:v1';
const THEATER_POSTURE_STALE_KEY = 'theater_posture:sebuf:stale:v1';
const THEATER_POSTURE_BACKUP_KEY = 'theater-posture:sebuf:backup:v1';
const THEATER_POSTURE_LOCK_KEY = 'seed-lock:theater-posture';
const THEATER_POSTURE_LOCK_TTL_SECONDS = 120;
const THEATER_POSTURE_LIVE_TTL = 1200; // 20 min — must outlive the 10-min seed interval (2x)
const THEATER_POSTURE_STALE_TTL = 86400; // 24h
const THEATER_POSTURE_BACKUP_TTL = 604800; // 7d
const THEATER_MIL_PREFIXES = [
'RCH', 'REACH', 'MOOSE', 'EVAC', 'DUSTOFF', 'PEDRO',
'DUKE', 'HAVOC', 'KNIFE', 'WARHAWK', 'VIPER', 'RAGE', 'FURY',
'SHELL', 'TEXACO', 'ARCO', 'ESSO', 'PETRO',
'SENTRY', 'AWACS', 'MAGIC', 'DISCO', 'DARKSTAR',
'COBRA', 'PYTHON', 'RAPTOR', 'EAGLE', 'HAWK', 'TALON',
'BOXER', 'OMNI', 'TOPCAT', 'SKULL', 'REAPER', 'HUNTER',
'ARMY', 'NAVY', 'USAF', 'USMC', 'USCG',
'CNV', 'EXEC',
'NATO', 'GAF', 'RRF', 'RAF', 'FAF', 'IAF', 'RNLAF', 'BAF', 'DAF', 'HAF', 'PAF',
'SWORD', 'LANCE', 'ARROW', 'SPARTAN',
'RSAF', 'EMIRI', 'UAEAF', 'KAF', 'QAF', 'BAHAF', 'OMAAF',
'IRIAF', 'IRGC',
'TUAF',
'RSD', 'RFF', 'VKS',
'CHN', 'PLAAF', 'PLA',
];
const THEATER_MIL_SHORT_PREFIXES = ['AE', 'RF', 'TF', 'PAT', 'SAM', 'OPS', 'CTF', 'IRG', 'TAF'];
const THEATER_AIRLINE_CODES = new Set([
'SVA', 'QTR', 'THY', 'UAE', 'ETD', 'GFA', 'MEA', 'RJA', 'KAC', 'ELY',
'IAW', 'IRA', 'MSR', 'SYR', 'PGT', 'AXB', 'FDB', 'KNE', 'FAD', 'ADY', 'OMA',
'ABQ', 'ABY', 'NIA', 'FJA', 'SWR', 'HZA', 'OMS', 'EGF', 'NOS', 'SXD',
]);
function theaterIsMilCallsign(callsign) {
if (!callsign) return false;
const cs = callsign.toUpperCase().trim();
for (const prefix of THEATER_MIL_PREFIXES) {
if (cs.startsWith(prefix)) return true;
}
for (const prefix of THEATER_MIL_SHORT_PREFIXES) {
if (cs.startsWith(prefix) && cs.length > prefix.length && /\d/.test(cs.charAt(prefix.length))) return true;
}
if (/^[A-Z]{3}\d{1,2}$/.test(cs)) {
const prefix = cs.slice(0, 3);
if (!THEATER_AIRLINE_CODES.has(prefix)) return true;
}
return false;
}
function theaterDetectAircraftType(callsign) {
if (!callsign) return 'unknown';
const cs = callsign.toUpperCase().trim();
if (/^(SHELL|TEXACO|ARCO|ESSO|PETRO|KC|STRAT)/.test(cs)) return 'tanker';
if (/^(SENTRY|AWACS|MAGIC|DISCO|DARKSTAR|E3|E8|E6)/.test(cs)) return 'awacs';
if (/^(RCH|REACH|MOOSE|EVAC|DUSTOFF|C17|C5|C130|C40)/.test(cs)) return 'transport';
if (/^(HOMER|OLIVE|JAKE|PSEUDO|GORDO|RC|U2|SR)/.test(cs)) return 'reconnaissance';
if (/^(RQ|MQ|REAPER|PREDATOR|GLOBAL)/.test(cs)) return 'drone';
if (/^(DEATH|BONE|DOOM|B52|B1|B2)/.test(cs)) return 'bomber';
if (/^(BOLT|VIPER|RAPTOR|BRONCO|EAGLE|HORNET|FALCON|STRIKE|TANGO|FURY)/.test(cs)) return 'fighter';
return 'unknown';
}
const WINGBITS_MAX_BOX_NM = 2000;
const WINGBITS_VIEWPORT_TILE_DEGREES = 30;
const WINGBITS_MAX_VIEWPORT_AREAS = 36;
const POSTURE_THEATERS = [
{ id: 'iran-theater', bounds: { north: 42, south: 20, east: 65, west: 30 }, thresholds: { elevated: 8, critical: 20 }, strikeIndicators: { minTankers: 2, minAwacs: 1, minFighters: 5 } },
{ id: 'taiwan-theater', bounds: { north: 30, south: 18, east: 130, west: 115 }, thresholds: { elevated: 6, critical: 15 }, strikeIndicators: { minTankers: 1, minAwacs: 1, minFighters: 4 } },
{ id: 'baltic-theater', bounds: { north: 65, south: 52, east: 32, west: 10 }, thresholds: { elevated: 5, critical: 12 }, strikeIndicators: { minTankers: 1, minAwacs: 1, minFighters: 3 } },
{ id: 'blacksea-theater', bounds: { north: 48, south: 40, east: 42, west: 26 }, thresholds: { elevated: 4, critical: 10 }, strikeIndicators: { minTankers: 1, minAwacs: 1, minFighters: 3 } },
{ id: 'korea-theater', bounds: { north: 43, south: 33, east: 132, west: 124 }, thresholds: { elevated: 5, critical: 12 }, strikeIndicators: { minTankers: 1, minAwacs: 1, minFighters: 3 } },
{ id: 'south-china-sea', bounds: { north: 25, south: 5, east: 121, west: 105 }, thresholds: { elevated: 6, critical: 15 }, strikeIndicators: { minTankers: 1, minAwacs: 1, minFighters: 4 } },
{ id: 'east-med-theater', bounds: { north: 37, south: 33, east: 37, west: 25 }, thresholds: { elevated: 4, critical: 10 }, strikeIndicators: { minTankers: 1, minAwacs: 1, minFighters: 3 } },
{ id: 'israel-gaza-theater', bounds: { north: 33, south: 29, east: 36, west: 33 }, thresholds: { elevated: 3, critical: 8 }, strikeIndicators: { minTankers: 1, minAwacs: 1, minFighters: 3 } },
{ id: 'yemen-redsea-theater', bounds: { north: 22, south: 11, east: 54, west: 32 }, thresholds: { elevated: 4, critical: 10 }, strikeIndicators: { minTankers: 1, minAwacs: 1, minFighters: 3 } },
];
// In-memory index of recently-seen Wingbits positions, keyed by ICAO24.
// Populated on every successful bbox response; served for callsign-only lookups.
// Entries expire after 5 minutes (Wingbits data is live, stale beyond that is misleading).
const WINGBITS_POS_INDEX = new Map(); // icao24 -> { position, ts }
const WINGBITS_POS_INDEX_TTL_MS = 5 * 60 * 1000;
function wingbitsIndexUpdate(positions) {
const ts = Date.now();
for (const p of positions) {
if (p.icao24) WINGBITS_POS_INDEX.set(p.icao24, { position: p, ts });
}
// Prune stale entries to prevent unbounded growth.
if (WINGBITS_POS_INDEX.size > 5000) {
const cutoff = ts - WINGBITS_POS_INDEX_TTL_MS;
for (const [k, v] of WINGBITS_POS_INDEX) {
if (v.ts < cutoff) WINGBITS_POS_INDEX.delete(k);
}
}
}
function wingbitsIndexLookupCallsign(callsign) {
const cutoff = Date.now() - WINGBITS_POS_INDEX_TTL_MS;
const results = [];
for (const { position, ts } of WINGBITS_POS_INDEX.values()) {
if (ts < cutoff) continue;
const cs = (position.callsign || '').trim().toUpperCase();
if (cs.includes(callsign)) results.push(position);
}
return results;
}
function buildWingbitsViewportAreas(south, west, north, east) {
const latTiles = Math.max(1, Math.ceil((north - south) / WINGBITS_VIEWPORT_TILE_DEGREES));
const lonTiles = Math.max(1, Math.ceil((east - west) / WINGBITS_VIEWPORT_TILE_DEGREES));
if (latTiles * lonTiles > WINGBITS_MAX_VIEWPORT_AREAS) return null;
const areas = [];
for (let latIndex = 0; latIndex < latTiles; latIndex += 1) {
const tileSouth = south + latIndex * WINGBITS_VIEWPORT_TILE_DEGREES;
const tileNorth = Math.min(north, tileSouth + WINGBITS_VIEWPORT_TILE_DEGREES);
for (let lonIndex = 0; lonIndex < lonTiles; lonIndex += 1) {
const tileWest = west + lonIndex * WINGBITS_VIEWPORT_TILE_DEGREES;
const tileEast = Math.min(east, tileWest + WINGBITS_VIEWPORT_TILE_DEGREES);
const centerLat = (tileSouth + tileNorth) / 2;
const centerLon = (tileWest + tileEast) / 2;
areas.push({
alias: `viewport-${latIndex}-${lonIndex}`,
by: 'box',
la: centerLat,
lo: centerLon,
w: Math.max(1, Math.min((tileEast - tileWest) * 60 * Math.cos(centerLat * Math.PI / 180), WINGBITS_MAX_BOX_NM)),
h: Math.max(1, Math.min((tileNorth - tileSouth) * 60, WINGBITS_MAX_BOX_NM)),
unit: 'nm',
});
}
}
return areas;
}
async function handleWingbitsTrackRequest(req, res) {
const apiKey = process.env.WINGBITS_API_KEY;
if (!apiKey) {
return safeEnd(res, 503, { 'Content-Type': 'application/json' },
JSON.stringify({ error: 'WINGBITS_API_KEY not configured', positions: [] }));
}
const url = new URL(req.url, 'http://localhost');
const params = url.searchParams;
const callsignFilter = (params.get('callsign') || '').trim().toUpperCase();
const laminStr = params.get('lamin');
const lominStr = params.get('lomin');
const lamaxStr = params.get('lamax');
const lomaxStr = params.get('lomax');
// For callsign-only searches (no bbox), serve from the in-memory position index populated
// by recent bbox responses. On index miss, fall back to a global Wingbits API call.
const isBboxMissing = !laminStr || !lominStr || !lamaxStr || !lomaxStr;
if (callsignFilter && isBboxMissing) {
const hits = wingbitsIndexLookupCallsign(callsignFilter);
if (hits.length > 0) {
return sendCompressed(req, res, 200,
{ 'Content-Type': 'application/json', 'Cache-Control': 'no-store' },
JSON.stringify({ positions: hits, source: 'wingbits' }));
}
// Index miss — flight not in any recent viewport. Try regional Wingbits API calls.
// The v1 API rejects boxes larger than ~2000 nm, so we use a set of regional areas
// that together cover high-traffic airspace without exceeding the size limit.
try {
const regionalAreas = [
{ alias: 'europe-mideast', by: 'box', la: 40, lo: 35, w: 2000, h: 2000, unit: 'nm' },
{ alias: 'asia-pacific', by: 'box', la: 25, lo: 120, w: 2000, h: 2000, unit: 'nm' },
{ alias: 'americas', by: 'box', la: 35, lo: -95, w: 2000, h: 2000, unit: 'nm' },
];
const gbResp = await fetch('https://customer-api.wingbits.com/v1/flights', {
method: 'POST',
headers: { 'x-api-key': apiKey, Accept: 'application/json', 'Content-Type': 'application/json', 'User-Agent': CHROME_UA },
body: JSON.stringify(regionalAreas),
signal: AbortSignal.timeout(15_000),
});
if (gbResp.ok) {
const gbData = await gbResp.json();
if (Array.isArray(gbData)) {
const now = Date.now();
const positions = [];
const seenIds = new Set();
for (const areaResult of gbData) {
const flightList = Array.isArray(areaResult.data) ? areaResult.data
: Array.isArray(areaResult.flights) ? areaResult.flights
: Array.isArray(areaResult) ? areaResult : [];
for (const f of flightList) {
const icao24 = f.h || f.icao24 || f.id || '';
if (!icao24 || seenIds.has(icao24)) continue;
const cs = (f.f || f.callsign || f.flight || '').trim().toUpperCase();
if (!cs.includes(callsignFilter)) continue;
seenIds.add(icao24);
positions.push({
icao24,
callsign: (f.f || f.callsign || f.flight || '').trim(),
lat: f.la ?? f.latitude ?? f.lat ?? 0,
lon: f.lo ?? f.longitude ?? f.lon ?? f.lng ?? 0,
altitudeM: (f.ab ?? f.altitude ?? f.alt ?? 0) * 0.3048,
groundSpeedKts: f.gs ?? f.groundSpeed ?? f.speed ?? 0,
trackDeg: f.th ?? f.heading ?? f.track ?? 0,
verticalRate: 0,
onGround: f.og ?? f.gr ?? f.onGround ?? false,
source: 'POSITION_SOURCE_WINGBITS',
observedAt: f.ra ? new Date(f.ra).getTime() : now,
});
}
}
wingbitsIndexUpdate(positions);
logThrottled('log', 'wingbits-callsign-global', `[Wingbits Track] global callsign fallback: ${positions.length} hits for "${callsignFilter}"`);
return sendCompressed(req, res, 200,
{ 'Content-Type': 'application/json', 'Cache-Control': 'no-store' },
JSON.stringify({ positions, source: 'wingbits' }));
}
} else {
const errBody = await gbResp.text().catch(() => '');
console.warn(`[Wingbits Track] Regional callsign fallback error: ${gbResp.status} — ${errBody.slice(0, 200)}`);
}
} catch (err) {
console.warn(`[Wingbits Track] Global callsign fallback failed: ${err?.message || err}`);
}
return sendCompressed(req, res, 200,
{ 'Content-Type': 'application/json', 'Cache-Control': 'no-store' },
JSON.stringify({ positions: [], source: 'wingbits' }));
}
if (isBboxMissing) {
return safeEnd(res, 400, { 'Content-Type': 'application/json' },
JSON.stringify({ error: 'Missing bbox params: lamin, lomin, lamax, lomax', positions: [] }));
}
const lamin = Number(laminStr);
const lomin = Number(lominStr);
const lamax = Number(lamaxStr);
const lomax = Number(lomaxStr);
if (!Number.isFinite(lamin) || !Number.isFinite(lomin) || !Number.isFinite(lamax) || !Number.isFinite(lomax)) {
return safeEnd(res, 400, { 'Content-Type': 'application/json' },
JSON.stringify({ error: 'Invalid bbox params: must be finite numbers', positions: [] }));
}
// Clamp bbox to valid geographic ranges before computing center.
// Map projections can produce slightly out-of-range values; Wingbits rejects la outside [-90,90].
const clampedLamin = Math.max(-90, Math.min(90, lamin));
const clampedLamax = Math.max(-90, Math.min(90, lamax));
const clampedLomin = Math.max(-180, Math.min(180, lomin));
const clampedLomax = Math.max(-180, Math.min(180, lomax));
const south = Math.min(clampedLamin, clampedLamax);
const north = Math.max(clampedLamin, clampedLamax);
const west = Math.min(clampedLomin, clampedLomax);
const east = Math.max(clampedLomin, clampedLomax);
const areas = buildWingbitsViewportAreas(south, west, north, east);
if (!areas) {
return safeEnd(res, 422, { 'Content-Type': 'application/json' }, JSON.stringify({
error: `Viewport requires more than ${WINGBITS_MAX_VIEWPORT_AREAS} Wingbits areas`,
positions: [],
}));
}
try {
const resp = await fetch('https://customer-api.wingbits.com/v1/flights', {
method: 'POST',
headers: { 'x-api-key': apiKey, Accept: 'application/json', 'Content-Type': 'application/json', 'User-Agent': CHROME_UA },
body: JSON.stringify(areas),
signal: AbortSignal.timeout(15_000),
});
if (!resp.ok) {
const errBody = await resp.text().catch(() => '');
console.warn(`[Wingbits Track] API error: ${resp.status} — ${errBody.slice(0, 200)}`);
return safeEnd(res, 502, { 'Content-Type': 'application/json' },
JSON.stringify({ error: `Wingbits API ${resp.status}`, positions: [] }));
}
const data = await resp.json();
if (!Array.isArray(data)) {
console.warn(`[Wingbits Track] Unexpected response shape: ${JSON.stringify(data).slice(0, 200)}`);
return safeEnd(res, 502, { 'Content-Type': 'application/json' },
JSON.stringify({ error: 'Wingbits returned non-array response', positions: [] }));
}
const positions = [];
const seenIds = new Set();
const now = Date.now();
for (const areaResult of data) {
const flightList = Array.isArray(areaResult.data) ? areaResult.data
: Array.isArray(areaResult.flights) ? areaResult.flights
: Array.isArray(areaResult) ? areaResult : [];
for (const f of flightList) {
const icao24 = f.h || f.icao24 || f.id || '';
if (!icao24 || seenIds.has(icao24)) continue;
// For callsign searches, skip non-matching flights early to keep response small.
if (callsignFilter) {
const cs = (f.f || f.callsign || f.flight || '').trim().toUpperCase();
if (!cs.includes(callsignFilter)) continue;
}
const lat = f.la ?? f.latitude ?? f.lat ?? 0;
const lon = f.lo ?? f.longitude ?? f.lon ?? f.lng ?? 0;
// Multi-area requests overlap at tile edges. Deduplicate and retain
// only positions inside the original viewport so a successful
// Wingbits response is complete and authoritative for that bbox.
if (lat < south || lat > north || lon < west || lon > east) continue;
seenIds.add(icao24);
positions.push({
icao24,
callsign: (f.f || f.callsign || f.fl