// Step 4 (§9): history and culture per place (NLD, GM0513 Gouda, CUW, USA, KOR).
//  - owid:      long-run population, life expectancy and urbanisation (OWID grapher CSVs)
//  - religion:  CBS 37944 (NL 1849–2018), 82904NED (NL 2010→), 71124NED (Gouda census 1930),
//               Pew Research Center 2010/2020 via OWID
//  - offices:   heads of government / state, governor, mayor (Wikidata P39 statements)
//  - turnout:   US and South Korean national elections (Wikidata P1867 / P1868)
//  - people:    best-known people per place (Wikidata, ranked by sitelinks)
//  - holidays:  public holidays (Nager.Date API)
//  - unesco:    World Heritage sites (UNESCO WHC open data, whc001)
//  - curated:   key events, cuisine/exports and transcribed series (scripts/stats/history-events.json)
import { readFileSync } from 'node:fs';
import { getJson, getText, http, insertChunks, log, run, sql, writeSeries } from './common';
import type { ObsRow, SeriesRow } from './common';
import { parseCsv } from './owid';

export const HISTORY_PLACES = ['NLD', 'GM0513', 'CUW', 'USA', 'KOR'] as const;
const year = (y: number) => `${String(y).padStart(4, '0')}-01-01`;
const errText = (e: unknown) => (e instanceof Error ? e.message : String(e));

// ── OWID long-run series ─────────────────────────────────────────────────────────────────
// Curaçao population and life expectancy already exist as owid:cw:* (UN WPP 1950→), and
// Maddison GDP per person as owid:maddison:*; those are reused, not duplicated.
const OWID_SERIES = [
	{
		slug: 'population',
		key: 'population',
		title: 'Population (long run)',
		unit: 'people',
		from: 1500,
		regions: ['NLD', 'USA', 'KOR', 'WLD'],
		source: 'HYDE, Gapminder and UN WPP, via Our World in Data'
	},
	{
		slug: 'life-expectancy',
		key: 'life-expectancy',
		title: 'Life expectancy at birth (long run)',
		unit: 'years',
		from: 1500,
		regions: ['NLD', 'USA', 'KOR', 'WLD'],
		source: 'Riley (2005), Zijdeman et al. (2015), HMD and UN WPP, via Our World in Data'
	},
	{
		slug: 'urbanization-last-500-years',
		key: 'urban-long',
		title: 'Share of people living in urban areas (long run)',
		unit: '%',
		from: 1500,
		regions: ['NLD', 'USA', 'KOR', 'WLD'],
		source: 'HYDE (2023), via Our World in Data'
	},
	{
		slug: 'share-of-population-urban',
		key: 'urban',
		title: 'Share of people living in urban areas',
		unit: '%',
		from: 1950,
		regions: ['NLD', 'USA', 'KOR', 'CUW', 'WLD'],
		source: 'UN World Urbanization Prospects, via Our World in Data'
	}
];
const owidRegion = (code: string, entity: string) =>
	code === 'OWID_WRL' || entity === 'World' ? 'WLD' : code;

async function owidJob(note: (w: string) => void) {
	const series: SeriesRow[] = [];
	const obs: ObsRow[] = [];
	for (const s of OWID_SERIES) {
		try {
			const url = `https://ourworldindata.org/grapher/${s.slug}.csv?csvType=full&useColumnShortNames=true`;
			const [header, ...lines] = parseCsv(await getText(url));
			const [e, c, y] = ['entity', 'code', 'year'].map((n) => header.indexOf(n));
			const v = header.length - 1;
			let n = 0;
			for (const l of lines) {
				if (l.length <= v || l[v] === '') continue;
				const region = owidRegion(l[c], l[e]);
				const yr = Number(l[y]);
				if (!s.regions.includes(region) || yr < s.from) continue;
				const id = `owid:hist:${s.key}:${region}`;
				if (!series.some((x) => x.series_id === id))
					series.push({
						series_id: id,
						source_id: 'owid',
						source_code: `${s.slug}/${region}`,
						title: s.title,
						unit: s.unit,
						freq: 'A',
						region_code: region,
						theme: 'history',
						meta: { url: `https://ourworldindata.org/grapher/${s.slug}`, source: s.source }
					});
				obs.push({ series_id: id, period: year(yr), value: Number(l[v]), status: null });
				n++;
			}
			if (!n) throw new Error('no rows for the requested places');
		} catch (error) {
			note(`owid ${s.slug}: ${errText(error)}`);
		}
	}
	return writeSeries(series, obs);
}

// ── Religion ─────────────────────────────────────────────────────────────────────────────
type CbsRow = Record<string, string | number | null>;
const cbsRows = async (table: string, filter = '') =>
	(
		await getJson<{ value: CbsRow[] }>(
			`https://opendata.cbs.nl/ODataApi/odata/${table}/TypedDataSet?$format=json${filter ? `&$filter=${encodeURIComponent(filter)}` : ''}`
		)
	).value;
const cbsYear = (p: unknown) => Number(String(p).slice(0, 4));

const RELIGION_37944: Record<string, [string, string]> = {
	GeenKerkelijkeGezindte_1: ['none', 'No religious affiliation'],
	RoomsKatholiek_2: ['catholic', 'Roman Catholic'],
	ProtestantseKerkInNederland_3: ['pkn', 'Protestant Church in the Netherlands (PKN)'],
	NederlandsHervormd_4: ['hervormd', 'Dutch Reformed (Hervormd)'],
	Gereformeerd_5: ['gereformeerd', 'Reformed (Gereformeerd)'],
	OverigeKerkelijkeGezindten_6: ['other', 'Other religion']
};
const RELIGION_82904: Record<string, [string, string]> = {
	A008467: ['none', 'No religious affiliation'],
	A008466: ['catholic', 'Roman Catholic'],
	A050076: ['protestant', 'Protestant (all)'],
	A008465: ['pkn', 'Protestant Church in the Netherlands (PKN)'],
	A008464: ['hervormd', 'Dutch Reformed (Hervormd)'],
	A008463: ['gereformeerd', 'Reformed (Gereformeerd)'],
	A047804: ['islam', 'Islam'],
	A008468: ['other', 'Other religion']
};
const RELIGION_1930: Record<string, string> = {
	KGNH: 'Dutch Reformed (Hervormd)',
	KGRK: 'Roman Catholic',
	KGGK: 'Reformed (Gereformeerd)',
	KGCG: 'Christian Reformed',
	KGRE: 'Remonstrant',
	KGEL: 'Lutheran',
	KGDO: 'Mennonite',
	KGNI: 'Jewish (Ashkenazi)',
	KGPI: 'Jewish (Portuguese)',
	KGGN: 'No religious affiliation'
};
const PEW: Record<string, string> = {
	christians: 'Christians',
	muslims: 'Muslims',
	jews: 'Jews',
	hindus: 'Hindus',
	buddhists: 'Buddhists',
	other_religions: 'Other religions',
	religiously_unaffiliated: 'Religiously unaffiliated'
};

async function religionJob(note: (w: string) => void) {
	const series: SeriesRow[] = [];
	const obs: ObsRow[] = [];
	const add = (s: Omit<SeriesRow, 'freq' | 'theme'>, period: string, value: number | null) => {
		if (value == null || !Number.isFinite(value)) return;
		if (!series.some((x) => x.series_id === s.series_id))
			series.push({ ...s, freq: 'A', theme: 'culture' });
		obs.push({ series_id: s.series_id, period, value, status: null });
	};
	try {
		for (const r of await cbsRows('37944'))
			for (const [col, [key, title]] of Object.entries(RELIGION_37944))
				add(
					{
						series_id: `cbs:religion:37944:${key}:NL01`,
						source_id: 'cbs',
						source_code: `37944/${col}`,
						title: `${title}, share of adults (18+)`,
						unit: '%',
						region_code: 'NLD',
						meta: {
							url: 'https://opendata.cbs.nl/statline/#/CBS/nl/dataset/37944/table',
							note: 'Censuses until 1971, surveys after'
						}
					},
					year(cbsYear(r.Perioden)),
					r[col] as number | null
				);
	} catch (error) {
		note(`cbs 37944: ${errText(error)}`);
	}
	try {
		for (const r of await cbsRows('82904NED', "Persoonskenmerken eq 'T009002'")) {
			const hit = RELIGION_82904[String(r.KerkelijkeGezindte).trim()];
			if (!hit) continue;
			add(
				{
					series_id: `cbs:religion:82904:${hit[0]}:NL01`,
					source_id: 'cbs',
					source_code: `82904NED/${String(r.KerkelijkeGezindte).trim()}`,
					title: `${hit[1]}, share of people 15+`,
					unit: '%',
					region_code: 'NLD',
					meta: { url: 'https://opendata.cbs.nl/statline/#/CBS/nl/dataset/82904NED/table' }
				},
				year(cbsYear(r.Perioden)),
				r.PersonenKerkelijkeGezindte_1 as number | null
			);
		}
	} catch (error) {
		note(`cbs 82904NED: ${errText(error)}`);
	}
	try {
		const rows = await cbsRows('71124NED', "Regio eq 'GM0513' and Geslacht eq 'GesT'");
		const total = rows.find((r) => String(r.KerkelijkeGezindte).trim() === 'KGTO')?.Bewoners_1;
		if (typeof total !== 'number' || total <= 0) throw new Error('no Gouda total for 1930');
		for (const r of rows) {
			const code = String(r.KerkelijkeGezindte).trim();
			if (!RELIGION_1930[code] || typeof r.Bewoners_1 !== 'number') continue;
			add(
				{
					series_id: `cbs:religion:1930:${code}:GM0513`,
					source_id: 'cbs',
					source_code: `71124NED/${code}`,
					title: `${RELIGION_1930[code]}, share of residents (census 1930)`,
					unit: '%',
					region_code: 'GM0513',
					meta: {
						url: 'https://opendata.cbs.nl/statline/#/CBS/nl/dataset/71124NED/table',
						residents: r.Bewoners_1,
						total
					}
				},
				'1930-01-01',
				Math.round((r.Bewoners_1 / total) * 1000) / 10
			);
		}
	} catch (error) {
		note(`cbs 71124NED: ${errText(error)}`);
	}
	try {
		const [header, ...lines] = parseCsv(
			await getText(
				'https://ourworldindata.org/grapher/share-of-population-by-religious-affiliation.csv?csvType=full&useColumnShortNames=true'
			)
		);
		const [c, y] = [header.indexOf('code'), header.indexOf('year')];
		for (const l of lines) {
			const region = owidRegion(l[c] ?? '', l[0] ?? '');
			if (!['NLD', 'USA', 'KOR', 'CUW', 'WLD'].includes(region)) continue;
			for (const [key, label] of Object.entries(PEW)) {
				const i = header.indexOf(`share__religion_${key}`);
				if (i < 0 || l[i] === '' || l[i] == null) continue;
				add(
					{
						series_id: `pew:religion:${key}:${region}`,
						source_id: 'pew',
						source_code: `religious-composition/${key}/${region}`,
						title: `${label}, share of population`,
						unit: '%',
						region_code: region,
						meta: {
							url: 'https://www.pewresearch.org/religion/feature/religious-composition-by-country-2010-2020/',
							via: 'https://ourworldindata.org/grapher/share-of-population-by-religious-affiliation'
						}
					},
					year(Number(l[y])),
					Number(l[i])
				);
			}
		}
	} catch (error) {
		note(`pew: ${errText(error)}`);
	}
	if (!obs.length) throw new Error('no religion rows');
	return writeSeries(series, obs);
}

// ── Wikidata ─────────────────────────────────────────────────────────────────────────────
type Binding = Record<string, { value: string } | undefined>;
async function sparql(query: string): Promise<Binding[]> {
	const response = await http(
		`https://query.wikidata.org/sparql?query=${encodeURIComponent(query)}`,
		{ headers: { accept: 'application/sparql-results+json' } }
	);
	return ((await response.json()) as { results: { bindings: Binding[] } }).results.bindings;
}
const qid = (b: { value: string } | undefined) => b?.value.split('/').pop() ?? '';
const day = (b: { value: string } | undefined) => {
	const m = b?.value.match(/^(-?\d{4})-(\d{2})-(\d{2})/);
	return m ? `${m[1]}-${m[2] === '00' ? '01' : m[2]}-${m[3] === '00' ? '01' : m[3]}` : null;
};

const OFFICES = [
	{ place: 'NLD', kind: 'head_of_government', qid: 'Q3058109' },
	{ place: 'NLD', kind: 'head_of_state', qid: 'Q2045066' },
	{ place: 'USA', kind: 'head_of_state', qid: 'Q11696' },
	{ place: 'KOR', kind: 'head_of_state', qid: 'Q6296418' },
	{ place: 'KOR', kind: 'head_of_government', qid: 'Q15407843' },
	{ place: 'CUW', kind: 'head_of_government', qid: 'Q2247822' },
	{ place: 'CUW', kind: 'governor', qid: 'Q2140877' },
	{ place: 'GM0513', kind: 'mayor', qid: 'Q98970128' }
];

async function officesJob(note: (w: string) => void) {
	const rows: Record<string, unknown>[] = [];
	for (const o of OFFICES) {
		try {
			const b = await sparql(`SELECT ?p ?pLabel ?officeLabel ?start ?end
				(GROUP_CONCAT(DISTINCT ?partyName; separator=" / ") AS ?parties) WHERE {
				BIND(wd:${o.qid} AS ?office)
				?p p:P39 ?st. ?st ps:P39 ?office; pq:P580 ?start; wikibase:rank ?rank.
				FILTER(?rank != wikibase:DeprecatedRank)
				OPTIONAL { ?st pq:P582 ?end }
				OPTIONAL { ?p wdt:P102 ?party. ?party rdfs:label ?partyName. FILTER(lang(?partyName) = "en") }
				SERVICE wikibase:label { bd:serviceParam wikibase:language "en,mul,nl,ko,pap". }
			} GROUP BY ?p ?pLabel ?officeLabel ?start ?end`);
			const seen = new Set<string>();
			for (const r of b) {
				const start = day(r.start);
				const key = `${qid(r.p)}|${start}`;
				if (!start || seen.has(key)) continue;
				seen.add(key);
				rows.push({
					place: o.place,
					kind: o.kind,
					office_qid: o.qid,
					office_label: r.officeLabel?.value ?? o.qid,
					person_qid: qid(r.p),
					person_label: r.pLabel?.value ?? qid(r.p),
					party: r.parties?.value || null,
					start_date: start,
					end_date: day(r.end)
				});
			}
		} catch (error) {
			note(`office ${o.qid}: ${errText(error)}`);
		}
	}
	if (!rows.length) throw new Error('no office holders');
	await sql.begin(async (tx) => {
		await tx`delete from stats.hist_office`;
		await insertChunks(tx, 'hist_office', rows);
	});
	return rows.length;
}

const ELECTIONS = [
	{
		cls: 'Q47566',
		id: 'wikidata:turnout:uspres:USA',
		region: 'USA',
		title: 'US presidential election'
	},
	{
		cls: 'Q5518574',
		id: 'wikidata:turnout:krpres:KOR',
		region: 'KOR',
		title: 'South Korean presidential election'
	},
	{
		cls: 'Q5518656',
		id: 'wikidata:turnout:krleg:KOR',
		region: 'KOR',
		title: 'South Korean legislative election'
	}
];

async function turnoutJob(note: (w: string) => void) {
	const series: SeriesRow[] = [];
	const obs: ObsRow[] = [];
	for (const e of ELECTIONS) {
		try {
			const b = await sparql(`SELECT ?e ?d ?elig ?cast WHERE {
				?e wdt:P31 wd:${e.cls}; wdt:P585 ?d; wdt:P1867 ?elig; wdt:P1868 ?cast. }`);
			const points = new Map<string, number>();
			for (const r of b) {
				const d = day(r.d);
				const elig = Number(r.elig?.value);
				const cast = Number(r.cast?.value);
				// Skip indirect elections (e.g. 1948 by the National Assembly) and bad data.
				if (!d || !(elig > 100_000) || !(cast > 0) || cast > elig) continue;
				points.set(d, Math.round((cast / elig) * 1000) / 10);
			}
			if (!points.size) continue;
			series.push({
				series_id: e.id,
				source_id: 'wikidata',
				source_code: e.cls,
				title: `Turnout, ${e.title}`,
				unit: '%',
				freq: 'A',
				region_code: e.region,
				theme: 'elections',
				meta: {
					url: `https://www.wikidata.org/wiki/${e.cls}`,
					note: 'Ballots cast (P1868) / eligible voters (P1867) as recorded on Wikidata; elections without both are missing'
				}
			});
			for (const [period, value] of points)
				obs.push({ series_id: e.id, period, value, status: null });
		} catch (error) {
			note(`turnout ${e.cls}: ${errText(error)}`);
		}
	}
	return writeSeries(series, obs);
}

const PEOPLE = [
	{ place: 'NLD', where: 'VALUES ?c { wd:Q55 wd:Q29999 wd:Q170072 } ?p wdt:P27 ?c.', min: 60 },
	{ place: 'USA', where: '?p wdt:P27 wd:Q30.', min: 160 },
	{ place: 'KOR', where: 'VALUES ?c { wd:Q884 wd:Q28179 wd:Q28233 } ?p wdt:P27 ?c.', min: 50 },
	{ place: 'GM0513', where: '?p wdt:P19/wdt:P131* wd:Q84125.', min: 0 },
	{ place: 'CUW', where: '?p wdt:P19/wdt:P131* wd:Q25279.', min: 0 }
];

async function peopleJob(note: (w: string) => void) {
	const rows: Record<string, unknown>[] = [];
	for (const p of PEOPLE) {
		try {
			const b = await sparql(`SELECT ?p ?n WHERE {
				${p.where} ?p wdt:P31 wd:Q5; wikibase:sitelinks ?n. FILTER(?n >= ${p.min})
			} ORDER BY DESC(?n) LIMIT 60`);
			const top = [...new Map(b.map((r) => [qid(r.p), Number(r.n?.value)])).entries()].slice(0, 25);
			if (!top.length) continue;
			const detail =
				await sparql(`SELECT ?p ?pLabel ?pDescription (MIN(YEAR(?b)) AS ?by) (MIN(YEAR(?d)) AS ?dy)
				(SAMPLE(?en) AS ?enwiki) (SAMPLE(?nl) AS ?nlwiki) WHERE {
				VALUES ?p { ${top.map(([q]) => `wd:${q}`).join(' ')} }
				OPTIONAL { ?p wdt:P569 ?b } OPTIONAL { ?p wdt:P570 ?d }
				OPTIONAL { ?en schema:about ?p; schema:isPartOf <https://en.wikipedia.org/> }
				OPTIONAL { ?nl schema:about ?p; schema:isPartOf <https://nl.wikipedia.org/> }
				SERVICE wikibase:label { bd:serviceParam wikibase:language "en,mul,nl,ko". }
			} GROUP BY ?p ?pLabel ?pDescription`);
			const info = new Map(detail.map((r) => [qid(r.p), r]));
			top.forEach(([q, n], i) => {
				const r = info.get(q);
				rows.push({
					place: p.place,
					person_qid: q,
					label: r?.pLabel?.value ?? q,
					description: r?.pDescription?.value ?? null,
					birth_year: r?.by ? Number(r.by.value) : null,
					death_year: r?.dy ? Number(r.dy.value) : null,
					sitelinks: n,
					url: r?.enwiki?.value ?? r?.nlwiki?.value ?? `https://www.wikidata.org/wiki/${q}`,
					rank: i + 1
				});
			});
		} catch (error) {
			note(`people ${p.place}: ${errText(error)}`);
		}
	}
	if (!rows.length) throw new Error('no people');
	await sql.begin(async (tx) => {
		await tx`delete from stats.hist_person`;
		await insertChunks(tx, 'hist_person', rows);
	});
	return rows.length;
}

// ── Holidays and UNESCO ──────────────────────────────────────────────────────────────────
const NAGER: Record<string, string> = { NL: 'NLD', US: 'USA', KR: 'KOR', CW: 'CUW' };

async function holidaysJob(note: (w: string) => void) {
	const rows: Record<string, unknown>[] = [];
	const y = new Date().getUTCFullYear();
	for (const [cc, place] of Object.entries(NAGER))
		for (const yr of [y, y + 1]) {
			try {
				const list = await getJson<
					{ date: string; localName: string; name: string; global: boolean; types: string[] }[]
				>(`https://date.nager.at/api/v3/PublicHolidays/${yr}/${cc}`);
				const seen = new Set<string>();
				for (const h of list) {
					const key = `${h.date}|${h.name}`;
					if (seen.has(key)) continue;
					seen.add(key);
					rows.push({
						place,
						date: h.date,
						name: h.name,
						local_name: h.localName,
						global: h.global,
						types: h.types
					});
				}
			} catch (error) {
				note(`nager ${cc} ${yr}: ${errText(error)}`);
			}
		}
	if (!rows.length) throw new Error('no holidays');
	await sql.begin(async (tx) => {
		await tx`delete from stats.hist_holiday`;
		await insertChunks(tx, 'hist_holiday', rows);
	});
	return rows.length;
}

type WhcRecord = {
	id_no: string | number;
	name_en: string;
	iso_codes: string | string[] | null;
	category: string;
	date_inscribed: string | number | null;
	danger: string | boolean | null;
	short_description_en: string | null;
	geo_point_2d: { lat: number; lon: number } | null;
	main_image_url: string | null;
};

async function unescoJob() {
	const records = await getJson<WhcRecord[]>(
		'https://data.unesco.org/api/explore/v2.1/catalog/datasets/whc001/exports/json?select=id_no,name_en,iso_codes,category,date_inscribed,danger,short_description_en,geo_point_2d,main_image_url'
	);
	const rows: Record<string, unknown>[] = [];
	for (const r of records) {
		const codes = (
			Array.isArray(r.iso_codes)
				? r.iso_codes
				: (String(r.iso_codes ?? '').match(/[A-Za-z]{2}/g) ?? [])
		).map((c) => c.toUpperCase());
		const places = new Set<string>();
		for (const c of codes) {
			// Willemstad is listed under the Netherlands (Kingdom); split by longitude.
			if (c === 'NL') places.add((r.geo_point_2d?.lon ?? 0) < -30 ? 'CUW' : 'NLD');
			else if (c === 'CW') places.add('CUW');
			else if (c === 'US') places.add('USA');
			else if (c === 'KR') places.add('KOR');
		}
		if (!places.size) continue;
		rows.push({
			id_no: Number(r.id_no),
			name: r.name_en,
			places: [...places],
			category: r.category,
			year_inscribed: r.date_inscribed ? Number(String(r.date_inscribed).slice(0, 4)) : null,
			danger: r.danger === true || r.danger === 'True',
			description: r.short_description_en
				? r.short_description_en.replace(/<[^>]+>/g, '').trim()
				: null,
			lat: r.geo_point_2d?.lat ?? null,
			lon: r.geo_point_2d?.lon ?? null,
			image_url: r.main_image_url
		});
	}
	if (rows.length < 10) throw new Error(`only ${rows.length} sites`);
	await sql.begin(async (tx) => {
		await tx`delete from stats.unesco_site`;
		await insertChunks(tx, 'unesco_site', rows);
	});
	return rows.length;
}

// ── Curated events, facts and transcribed series ─────────────────────────────────────────
type Curated = {
	events: {
		place: string;
		slug: string;
		year: number;
		year_end: number | null;
		date_label: string | null;
		title: string;
		summary: string;
		theme: string;
		source_url: string;
	}[];
	facts: Record<string, unknown>[];
	series: {
		series_id: string;
		title: string;
		unit: string;
		region_code: string;
		source_url: string;
		note: string;
		points?: [number, number][];
		shares?: Record<string, number>;
		year?: number;
	}[];
};

async function curatedJob() {
	const data = JSON.parse(
		readFileSync(new URL('./history-events.json', import.meta.url), 'utf8')
	) as Curated;
	for (const e of data.events)
		if (!e.source_url?.startsWith('https://'))
			throw new Error(`event ${e.place}/${e.slug} has no source`);
	await sql.begin(async (tx) => {
		await tx`delete from stats.hist_event`;
		await insertChunks(tx, 'hist_event', data.events);
		await tx`delete from stats.hist_fact`;
		await insertChunks(tx, 'hist_fact', data.facts);
	});
	const series: SeriesRow[] = [];
	const obs: ObsRow[] = [];
	for (const s of data.series) {
		const source = s.series_id.startsWith('cbscw:') ? 'cbscw' : 'history';
		const meta = { url: s.source_url, note: s.note };
		if (s.points) {
			series.push({
				series_id: s.series_id,
				source_id: source,
				source_code: s.series_id,
				title: s.title,
				unit: s.unit,
				freq: 'A',
				region_code: s.region_code,
				theme: 'history',
				meta
			});
			for (const [y, v] of s.points)
				obs.push({ series_id: s.series_id, period: year(y), value: v, status: null });
		}
		for (const [label, v] of Object.entries(s.shares ?? {})) {
			const id = `${s.series_id}:${label.toLowerCase()}`;
			series.push({
				series_id: id,
				source_id: source,
				source_code: id,
				title: `${s.title}: ${label}`,
				unit: s.unit,
				freq: 'A',
				region_code: s.region_code,
				theme: 'culture',
				meta: { ...meta, label }
			});
			obs.push({ series_id: id, period: year(s.year ?? 2023), value: v, status: null });
		}
	}
	await writeSeries(series, obs);
	return data.events.length + data.facts.length + obs.length;
}

const JOBS = {
	curated: ['history', curatedJob],
	owid: ['owid', owidJob],
	religion: ['cbs', religionJob],
	offices: ['wikidata', officesJob],
	turnout: ['wikidata', turnoutJob],
	people: ['wikidata', peopleJob],
	holidays: ['nager', holidaysJob],
	unesco: ['unesco', unescoJob]
} as const;

export async function history(opts: { only?: string; job: string }) {
	const wanted = opts.only ? opts.only.split(',') : Object.keys(JOBS);
	let failed = 0;
	for (const name of wanted) {
		const entry = JOBS[name as keyof typeof JOBS];
		if (!entry) throw new Error(`unknown history job ${name} (${Object.keys(JOBS).join(', ')})`);
		const [sourceId, fn] = entry;
		try {
			await run(sourceId, `history:${name}`, (note) => fn(note));
		} catch {
			failed++;
		}
	}
	log(`history: ${wanted.length - failed}/${wanted.length} jobs ok`);
	if (failed) throw new Error(`${failed} history jobs failed`);
}