fix(conflict): restore keyless GDELT fallback

scripts/_conflict-gdelt-bulk.mjs

@@ -0,0 +1,237 @@
+// Resilient GDELT conflict-event fallback using the official 15-minute bulk
+// event export. The DOC API is aggressively per-IP throttled; the bulk stream
+// is a single global stream and therefore remains usable when country-by-country
+// DOC queries all return 429.
+
+import { createHash } from 'node:crypto';
+import { inflateRawSync } from 'node:zlib';
+import { GDELT_COUNTRY_NAMES, gdeltSeenDateToIso } from './_conflict-gdelt.mjs';
+import { allSettledWithConcurrency } from './_seed-utils.mjs';
+
+const GDELT_STORAGE_ORIGIN = 'https://storage.googleapis.com/data.gdeltproject.org';
+export const GDELT_MASTER_FILELIST_URL = `${GDELT_STORAGE_ORIGIN}/gdeltv2/masterfilelist.txt`;
+export const GDELT_MAX_EXPORT_ZIP_BYTES = 5_000_000;
+export const GDELT_MAX_EXPORT_CSV_BYTES = 30_000_000;
+
+const MASTER_TAIL_BYTES = 16_384;
+const RECENT_EXPORT_COUNT = 8;
+const EXPORT_FETCH_CONCURRENCY = 4;
+const REQUEST_TIMEOUT_MS = 20_000;
+export const GDELT_BULK_WORST_NETWORK_MS = REQUEST_TIMEOUT_MS
+  * (1 + Math.ceil(RECENT_EXPORT_COUNT / EXPORT_FETCH_CONCURRENCY));
+const USER_AGENT = 'WorldMonitor/1.0 (+https://www.worldmonitor.app)';
+const MATERIAL_VIOLENCE_ROOT_CODES = new Set(['18', '19', '20']);
+
+// GDELT ActionGeo_CountryCode uses FIPS 10-4 rather than ISO-2.
+// Palestine can appear as either Gaza (GZ) or West Bank (WE).
+export const GDELT_FIPS_TO_ISO2 = Object.freeze({
+  AF: 'AF', SY: 'SY', UP: 'UA', SU: 'SD', OD: 'SS', SO: 'SO', CG: 'CD',
+  BM: 'MM', YM: 'YE', ET: 'ET', IZ: 'IQ', GZ: 'PS', WE: 'PS', LY: 'LY',
+  ML: 'ML', UV: 'BF', NG: 'NE', NI: 'NG', CM: 'CM', MZ: 'MZ', HA: 'HT',
+});
+
+function boundedPositiveInteger(value, label, max) {
+  const parsed = Number(value);
+  if (!Number.isSafeInteger(parsed) || parsed <= 0 || parsed > max) {
+    throw new Error(`invalid GDELT ${label}: ${value}`);
+  }
+  return parsed;
+}
+
+function parseExportDescriptorLine(exportLine) {
+  const [sizeRaw, md5Raw, urlRaw, ...extra] = exportLine.split(/\s+/);
+  if (!sizeRaw || !md5Raw || !urlRaw || extra.length) {
+    throw new Error('malformed GDELT event export manifest line');
+  }
+  const size = boundedPositiveInteger(sizeRaw, 'event export size', GDELT_MAX_EXPORT_ZIP_BYTES);
+  const md5 = md5Raw.toLowerCase();
+  if (!/^[a-f0-9]{32}$/.test(md5)) throw new Error('invalid GDELT event export checksum');
+
+  const url = new URL(urlRaw);
+  if (!['http:', 'https:'].includes(url.protocol) || url.hostname !== 'data.gdeltproject.org' || url.port) {
+    throw new Error(`untrusted GDELT event export URL: ${urlRaw}`);
+  }
+  const match = url.pathname.match(/^\/gdeltv2\/(\d{14})\.export\.CSV\.zip$/);
+  if (!match || url.search || url.hash) throw new Error(`invalid GDELT event export path: ${urlRaw}`);
+
+  return {
+    size,
+    md5,
+    url: `${GDELT_STORAGE_ORIGIN}${url.pathname}`,
+    exportTimestamp: match[1],
+  };
+}
+
+export function parseGdeltRecentExports(manifest, limit = RECENT_EXPORT_COUNT) {
+  const descriptors = [];
+  for (const line of String(manifest || '').split(/\r?\n/)) {
+    const trimmed = line.trim();
+    if (!/\.export\.CSV\.zip$/i.test(trimmed)) continue;
+    try {
+      descriptors.push(parseExportDescriptorLine(trimmed));
+    } catch (error) {
+      // A suffix range can begin mid-line. Ignore that incomplete fragment,
+      // but fail closed for any full-looking descriptor that violates the
+      // checksum/size/URL allowlist.
+      if (!/^\d+\s+[a-f0-9]{32}\s+/i.test(trimmed)) continue;
+      throw error;
+    }
+  }
+  if (!descriptors.length) throw new Error('GDELT master manifest tail has no valid event exports');
+  return descriptors
+    .sort((a, b) => a.exportTimestamp.localeCompare(b.exportTimestamp))
+    .slice(-Math.max(1, limit));
+}
+
+export function extractGdeltExportCsv(zipBytes) {
+  const zip = Buffer.isBuffer(zipBytes) ? zipBytes : Buffer.from(zipBytes || []);
+  if (zip.length < 30 || zip.readUInt32LE(0) !== 0x04034b50) {
+    throw new Error('invalid GDELT event export ZIP header');
+  }
+  const flags = zip.readUInt16LE(6);
+  if (flags & 0x1) throw new Error('encrypted GDELT event export ZIP is unsupported');
+  if (flags & 0x8) throw new Error('streaming GDELT event export ZIP is unsupported');
+
+  const method = zip.readUInt16LE(8);
+  const compressedSize = boundedPositiveInteger(
+    zip.readUInt32LE(18),
+    'ZIP compressed size',
+    GDELT_MAX_EXPORT_ZIP_BYTES,
+  );
+  const uncompressedSize = boundedPositiveInteger(
+    zip.readUInt32LE(22),
+    'ZIP uncompressed size',
+    GDELT_MAX_EXPORT_CSV_BYTES,
+  );
+  const filenameLength = zip.readUInt16LE(26);
+  const extraLength = zip.readUInt16LE(28);
+  const dataStart = 30 + filenameLength + extraLength;
+  const dataEnd = dataStart + compressedSize;
+  if (dataStart > zip.length || dataEnd > zip.length) throw new Error('truncated GDELT event export ZIP');
+
+  const filename = zip.subarray(30, 30 + filenameLength).toString('utf8');
+  if (!/^\d{14}\.export\.CSV$/.test(filename)) {
+    throw new Error(`unexpected GDELT event export filename: ${filename}`);
+  }
+
+  const compressed = zip.subarray(dataStart, dataEnd);
+  const csv = method === 8
+    ? inflateRawSync(compressed, { maxOutputLength: GDELT_MAX_EXPORT_CSV_BYTES })
+    : (method === 0 ? Buffer.from(compressed) : null);
+  if (!csv) throw new Error(`unsupported GDELT event export ZIP compression method: ${method}`);
+  if (csv.length !== uncompressedSize) {
+    throw new Error(`GDELT event export size mismatch: expected ${uncompressedSize}, got ${csv.length}`);
+  }
+  return csv.toString('utf8');
+}
+
+function sourceDomain(sourceUrl) {
+  try {
+    return new URL(sourceUrl).hostname;
+  } catch {
+    return '';
+  }
+}
+
+export function mapGdeltExportToConflictEvents(csv) {
+  const events = [];
+  const seen = new Set();
+  for (const line of String(csv || '').split(/\r?\n/)) {
+    if (!line) continue;
+    const fields = line.split('\t');
+    if (fields.length < 61 || fields[25] !== '1' || fields[29] !== '4') continue;
+    if (!MATERIAL_VIOLENCE_ROOT_CODES.has(fields[28])) continue;
+
+    const iso2 = GDELT_FIPS_TO_ISO2[fields[53]];
+    const country = GDELT_COUNTRY_NAMES[iso2];
+    const id = fields[0];
+    const eventDate = gdeltSeenDateToIso(fields[59]);
+    if (!id || seen.has(id) || !country || !eventDate) continue;
+    seen.add(id);
+
+    const url = fields[60] || '';
+    events.push({
+      id: `gdelt-event-${id}`,
+      eventType: `GDELT ${fields[26] || fields[28] || 'material conflict'}`,
+      country,
+      event_date: eventDate,
+      occurredAt: Date.parse(eventDate) || 0,
+      source: sourceDomain(url),
+      url,
+    });
+  }
+  return events;
+}
+
+async function fetchBoundedBuffer(fetchImpl, url, maxBytes, extraHeaders = {}) {
+  const response = await fetchImpl(url, {
+    headers: { Accept: '*/*', 'User-Agent': USER_AGENT, ...extraHeaders },
+    signal: AbortSignal.timeout(REQUEST_TIMEOUT_MS),
+  });
+  if (!response.ok) throw new Error(`GDELT bulk HTTP ${response.status} for ${url}`);
+  const declaredLength = Number(response.headers.get('content-length'));
+  if (Number.isFinite(declaredLength) && declaredLength > maxBytes) {
+    throw new Error(`GDELT bulk response exceeds ${maxBytes} bytes`);
+  }
+  if (!response.body) throw new Error(`GDELT bulk response has no body for ${url}`);
+  const chunks = [];
+  let total = 0;
+  for await (const chunk of response.body) {
+    total += chunk.byteLength;
+    if (total > maxBytes) {
+      throw new Error(`GDELT bulk response exceeds ${maxBytes} bytes`);
+    }
+    chunks.push(Buffer.from(chunk));
+  }
+  return Buffer.concat(chunks, total);
+}
+
+export async function fetchGdeltBulkConflictEvents({ fetchImpl = globalThis.fetch } = {}) {
+  const manifestBytes = await fetchBoundedBuffer(
+    fetchImpl,
+    GDELT_MASTER_FILELIST_URL,
+    MASTER_TAIL_BYTES,
+    { Range: `bytes=-${MASTER_TAIL_BYTES}` },
+  );
+  const descriptors = parseGdeltRecentExports(manifestBytes.toString('utf8'));
+  const results = await allSettledWithConcurrency(
+    descriptors,
+    EXPORT_FETCH_CONCURRENCY,
+    async (descriptor) => {
+      const zipBytes = await fetchBoundedBuffer(fetchImpl, descriptor.url, GDELT_MAX_EXPORT_ZIP_BYTES);
+      if (zipBytes.length !== descriptor.size) {
+        throw new Error(`download size mismatch: expected ${descriptor.size}, got ${zipBytes.length}`);
+      }
+      const actualMd5 = createHash('md5').update(zipBytes).digest('hex');
+      if (actualMd5 !== descriptor.md5) throw new Error('checksum mismatch');
+      return {
+        events: mapGdeltExportToConflictEvents(extractGdeltExportCsv(zipBytes)),
+        exportTimestamp: descriptor.exportTimestamp,
+      };
+    },
+  );
+
+  const successful = results.filter(result => result.status === 'fulfilled');
+  if (!successful.length) {
+    const sample = results.slice(0, 3).map(result => result.reason?.message || result.reason).join(', ');
+    throw new Error(`all recent GDELT event exports failed${sample ? `: ${sample}` : ''}`);
+  }
+  const events = [];
+  const seen = new Set();
+  for (const result of successful) {
+    for (const event of result.value.events) {
+      if (seen.has(event.id)) continue;
+      seen.add(event.id);
+      events.push(event);
+    }
+  }
+  return {
+    events,
+    exportTimestamp: successful
+      .map(result => result.value.exportTimestamp)
+      .sort()
+      .at(-1),
+    exportsRequested: descriptors.length,
+    exportsSucceeded: successful.length,
+  };
+}

scripts/seed-conflict-intel.mjs

@@ -19,6 +19,7 @@
 import { loadEnvFile, CHROME_UA, runSeed, writeExtraKeyWithMeta, sleep, loadSharedConfig } from './_seed-utils.mjs';
 import { fetchGdeltJson } from './_gdelt-fetch.mjs';
 import { buildGdeltConflictUrl, mapGdeltArticlesToEvents, GDELT_COUNTRY_NAMES } from './_conflict-gdelt.mjs';
+import { fetchGdeltBulkConflictEvents } from './_conflict-gdelt-bulk.mjs';
 
 loadEnvFile(import.meta.url);
 
@@ -48,10 +49,11 @@ export const GDELT_MIN_SUCCESSFUL_COUNTRIES = Math.ceil(CONFLICT_COUNTRIES.lengt
 // in-flight batch may still drain past the cutoff: ≤~100s at the knobs below
 // (15s concurrent direct legs + 4 × 20s SERIALIZED sync proxy curls — curlFetch is
 // execFileSync, so "concurrent" proxy attempts block the event loop one at a time;
-// 92s observed live 2026-07-10). Worst single fetchAll attempt ≈
-// max(HAPI 306s, 120s + 100s) + extra-key writes ≈ ~350s, inside both the 360s
-// lock below and the 480s fetch deadline (lockTtl + margin). Without this cap a
+// 92s observed live 2026-07-10). Worst single fetchAll attempt before the bulk
+// fallback ≈ max(HAPI 306s, 120s + 100s). Without this cap a
 // GDELT brownout ran 5 batches ≈ 375s+ → deadline breach → exit 75 every tick.
+// The bulk fallback runs after those parallel feeds settle, so its 60s bound
+// and 30s publish slack are additive: max(306s, 220s) + 60s + 30s = 396s.
 export const GDELT_SWEEP_BUDGET_MS = 120_000;
 // maxRetries: 0 — a second direct attempt would honor GDELT's Retry-After header
 // (≤60s sleep, _gdelt-fetch.mjs MAX_RETRY_AFTER_MS), blowing any per-batch bound;
@@ -61,10 +63,10 @@ export const GDELT_SWEEP_BUDGET_MS = 120_000;
 export const GDELT_COUNTRY_FETCH_OPTS = Object.freeze({ maxRetries: 0, proxyMaxAttempts: 1 });
 // Lock must outlive the worst legitimate run (runSeed's documented invariant —
 // _seed-utils.mjs: "a healthy seeder is designed never to outlive its own lock");
-// it also sets the fetch deadline (lockTtlMs + 120s margin = 480s). The default
+// it also sets the fetch deadline (lockTtlMs + 120s margin = 540s). The default
 // 120s lock was ALREADY shorter than this seeder's worst case. Cron cadence is
-// 30min, so a hard-crashed run's dangling lock costs at most 6 of those minutes.
-export const ACLED_INTEL_LOCK_TTL_MS = 360_000;
+// 30min, so a hard-crashed run's dangling lock costs at most 7 of those minutes.
+export const ACLED_INTEL_LOCK_TTL_MS = 420_000;
 
 const ISO2_TO_ISO3 = loadSharedConfig('iso2-to-iso3.json');
 
@@ -209,14 +211,11 @@ async function fetchAcledEvents({
 
 // ─── GDELT conflict-events fallback (used when ACLED has no credentials) ───
 // ACLED requires a registered account. When its credentials are absent, keep a
-// near-real-time conflict signal by proxying GDELT DOC 2.0 coverage volume: per
-// priority country, count recent conflict-tagged articles and emit them as synthetic
-// events in the SAME {country, event_date} shape the EMA engine reads
-// (_ema-threat-engine.mjs). Article volume is a coarser proxy than ACLED event counts,
-// but it is keyless, near-real-time, and directionally valid for the per-country
-// escalation EMA. URL/query + article→event mapping live in the import-safe,
-// unit-tested _conflict-gdelt.mjs; this fn owns the throttle-aware fetch via
-// fetchGdeltJson's proxy path (GDELT is per-IP 429-throttled).
+// near-real-time conflict signal from GDELT. The DOC 2.0 path counts recent
+// conflict-tagged articles per priority country and emits synthetic events in the
+// SAME {country, event_date} shape the EMA engine reads (_ema-threat-engine.mjs).
+// When DOC coverage is throttled or yields no events, the official 15-minute bulk
+// event export supplies material-conflict records instead.
 export async function fetchGdeltCountryEvents(cc) {
   if (!GDELT_COUNTRY_NAMES[cc]) {
     return { country: cc, ok: false, events: [], error: 'unknown country code' };
@@ -234,6 +233,7 @@ export async function fetchGdeltCountryEvents(cc) {
 
 export async function fetchGdeltConflictEvents({
   fetchCountryEvents = fetchGdeltCountryEvents,
+  fetchBulkEvents = fetchGdeltBulkConflictEvents,
   pace = sleep,
   now = Date.now,
   deadlineAt,
@@ -269,12 +269,34 @@ export async function fetchGdeltConflictEvents({
     }
     if (i + CONCURRENCY < CONFLICT_COUNTRIES.length) await pace(500); // inter-batch only; no trailing wait
   }
-  if (successfulCountries < GDELT_MIN_SUCCESSFUL_COUNTRIES) {
+  if (successfulCountries < GDELT_MIN_SUCCESSFUL_COUNTRIES || events.length === 0) {
     const sample = failedCountries.slice(0, 6).map(({ country, error }) => `${country}:${error}`).join(', ');
-    throw new Error(
-      `GDELT conflict-events coverage below floor: ${successfulCountries}/${CONFLICT_COUNTRIES.length} countries succeeded ` +
-      `(min ${GDELT_MIN_SUCCESSFUL_COUNTRIES})${sample ? `; failures: ${sample}` : ''}`,
-    );
+    const docFailure = successfulCountries < GDELT_MIN_SUCCESSFUL_COUNTRIES
+      ? `GDELT conflict-events coverage below floor: ${successfulCountries}/${CONFLICT_COUNTRIES.length} countries succeeded ` +
+        `(min ${GDELT_MIN_SUCCESSFUL_COUNTRIES})${sample ? `; failures: ${sample}` : ''}`
+      : `GDELT conflict-events returned zero events across ${successfulCountries}/${CONFLICT_COUNTRIES.length} successful countries`;
+    console.warn(`  ${docFailure}; trying official bulk event export`);
+    try {
+      const bulk = await fetchBulkEvents();
+      if (!bulk?.events?.length) throw new Error('latest export contained no priority-country material-conflict events');
+      console.log(`  GDELT bulk conflict-events fallback: ${bulk.events.length} events from export ${bulk.exportTimestamp}`);
+      return {
+        events: bulk.events,
+        pagination: {
+          countriesTotal: CONFLICT_COUNTRIES.length,
+          countriesSucceeded: CONFLICT_COUNTRIES.length,
+          countriesFailed: 0,
+          minSuccessfulCountries: GDELT_MIN_SUCCESSFUL_COUNTRIES,
+          exportTimestamp: bulk.exportTimestamp,
+          exportsRequested: bulk.exportsRequested,
+          exportsSucceeded: bulk.exportsSucceeded,
+          countriesWithEvents: new Set(bulk.events.map(event => event.country)).size,
+        },
+        source: 'gdelt-bulk',
+      };
+    } catch (bulkError) {
+      throw new Error(`${docFailure}; bulk fallback failed: ${bulkError?.message || bulkError}`);
+    }
   }
   console.log(`  GDELT conflict-events (ACLED fallback): ${events.length} events across ${successfulCountries}/${CONFLICT_COUNTRIES.length} successful country fetches`);
   return {