spb/datacenterindex

1/** Country statistics (population, GDP, electricity, renewable share) → countries columns + stats jsonb + provenance. */2import { sql } from "@dci/db";3import type { NormalizedCountryStat } from "@dci/core";4import { addRef, bump, knownCountries, type IngestContext, type Tx } from "./common.js";5import { writeProvenance } from "./provenance.js";67export async function ingestCountryStat(tx: Tx, ctx: IngestContext, c: NormalizedCountryStat): Promise<void> {8  const iso2 = (c.iso2 ?? c.key).trim().toUpperCase();9  const known = await knownCountries(tx, ctx);10  if (!known.has(iso2)) throw new Error(`country ${iso2}: unknown ISO2 (seed countries first)`);11  const year = c.year ?? null;12  const existing = (await tx.execute(sql`select population, gdp_usd, electricity_twh, renewable_share, stats_year, stats from countries where iso2 = ${iso2}`))[0]!;13  const existingYear = existing.stats_year == null ? null : Number(existing.stats_year);14  // newer (or same-year) statistics win; an older vintage only fills blanks15  const newer = year == null || existingYear == null || year >= existingYear;16  const pick = (incoming: number | null | undefined, current: unknown) => (incoming == null ? (current as number | null) : newer || current == null ? incoming : (current as number));17  const population = pick(c.population, existing.population == null ? null : Number(existing.population));18  const gdp = pick(c.gdpUsd, existing.gdp_usd);19  const twh = pick(c.electricityTwh, existing.electricity_twh);20  const ren = pick(c.renewableShare, existing.renewable_share);21  const stats = { ...((existing.stats as Record<string, unknown>) ?? {}) };22  const indicators = { ...((stats.indicators as Record<string, unknown>) ?? {}) };23  for (const [k, v] of Object.entries({ population: c.population, gdpUsd: c.gdpUsd, electricityTwh: c.electricityTwh, renewableShare: c.renewableShare })) {24    if (v == null) continue;25    indicators[k] = { value: v, year, sourceId: c.provenance.sourceId, url: c.provenance.url, observedAt: ctx.now };26  }27  stats.indicators = indicators;28  const changed = population !== (existing.population == null ? null : Number(existing.population)) || gdp !== existing.gdp_usd || twh !== existing.electricity_twh || ren !== existing.renewable_share;29  await tx.execute(sql`update countries set population = ${population}, gdp_usd = ${gdp}, electricity_twh = ${twh}, renewable_share = ${ren},30    stats_year = ${newer && year != null ? year : existingYear}, stats = ${JSON.stringify(stats)}::jsonb, updated_at = now() where iso2 = ${iso2}`);31  if (changed) ctx.stats.updated++;32  else ctx.stats.unchanged++;33  bump(ctx, "country");34  addRef(ctx, "country", iso2);35  await writeProvenance(36    tx,37    ctx,38    "country",39    iso2,40    [41      { field: "population", value: c.population ?? null },42      { field: "gdpUsd", value: c.gdpUsd ?? null },43      { field: "electricityTwh", value: c.electricityTwh ?? null },44      { field: "renewableShare", value: c.renewableShare ?? null },45      { field: "statsYear", value: year },46    ].map((f) => ({ ...f, provenance: c.provenance })),47    c.key,48  );49}50