website/src/lib/server/stats.ts

// Read side of the public /stats section: normalised series from stats.series/obs, source
// attributions and data freshness from stats.fetch_run. Everything here is open data.
import { client } from '#lib/server/db/index.js';

// Data changes at most hourly; keep page loads cheap with a short in-process cache.
const TTL_MS = 10 * 60 * 1000;
const memo = new Map<string, { at: number; value: Promise<unknown> }>();
export function cached<T>(key: string, fn: () => Promise<T>): Promise<T> {
	const hit = memo.get(key);
	if (hit && Date.now() - hit.at < TTL_MS) return hit.value as Promise<T>;
	const value = fn().catch((error) => {
		memo.delete(key);
		throw error;
	});
	memo.set(key, { at: Date.now(), value });
	return value;
}

export type StatPoint = [period: string, value: number];
export type StatSeries = {
	id: string;
	title: string;
	unit: string;
	freq: string;
	region: string | null;
	sourceId: string;
	points: StatPoint[];
	/** First forecast year (IMF WEO vintages); points from that year on are projections. */
	forecastFrom?: number;
};
export type SeriesRequest = {
	id: string;
	/** Only observations on or after this date (YYYY-MM-DD). */
	from?: string;
	/** Average daily data per month or week to keep payloads small. */
	resample?: 'M' | 'W';
};

/** Loads the requested series (missing ids are simply absent from the result). */
export async function loadSeries(
	requests: (string | SeriesRequest)[]
): Promise<Record<string, StatSeries>> {
	const reqs = requests.map((r) => (typeof r === 'string' ? { id: r } : r));
	if (!reqs.length) return {};
	const meta = await client<
		{
			series_id: string;
			title: string;
			unit: string;
			freq: string;
			region_code: string | null;
			source_id: string;
			forecast_from: string | null;
		}[]
	>`select series_id, title, unit, freq, region_code, source_id, meta->>'forecastFrom' as forecast_from
		from stats.series
		where series_id = any(${reqs.map((r) => r.id)})`;
	const out: Record<string, StatSeries> = {};
	await Promise.all(
		meta.map(async (m) => {
			const req = reqs.find((r) => r.id === m.series_id)!;
			const from = req.from ?? '0001-01-01';
			const rows = req.resample
				? await client<{ period: string; value: number }[]>`
					select to_char(date_trunc(${req.resample === 'M' ? 'month' : 'week'}, period), 'YYYY-MM-DD') as period,
						avg(value)::float8 as value
					from stats.obs where series_id = ${m.series_id} and period >= ${from} and value is not null
					group by 1 order by 1`
				: await client<{ period: string; value: number }[]>`
					select to_char(period, 'YYYY-MM-DD') as period, value from stats.obs
					where series_id = ${m.series_id} and period >= ${from} and value is not null order by period`;
			out[m.series_id] = {
				id: m.series_id,
				title: m.title,
				unit: m.unit,
				freq: m.freq,
				region: m.region_code,
				sourceId: m.source_id,
				...(m.forecast_from ? { forecastFrom: Number(m.forecast_from) } : {}),
				points: rows.map((r) => [r.period, Math.round(r.value * 1e6) / 1e6])
			};
		})
	);
	return out;
}

export type StatSource = {
	id: string;
	name: string;
	attribution: string;
	licence: string;
	baseUrl: string;
};
export async function loadSources(): Promise<Record<string, StatSource>> {
	const rows = await client<
		{ source_id: string; name: string; attribution: string; licence: string; base_url: string }[]
	>`
		select source_id, name, attribution, licence, base_url from stats.source order by source_id`;
	return Object.fromEntries(
		rows.map((r) => [
			r.source_id,
			{
				id: r.source_id,
				name: r.name,
				attribution: r.attribution,
				licence: r.licence,
				baseUrl: r.base_url
			}
		])
	);
}

export type Freshness = {
	sourceId: string;
	name: string;
	lastOk: string | null;
	lastRun: string | null;
	lastRunOk: boolean | null;
	rows: number | null;
	series: number;
	observations: number;
	latestPeriod: string | null;
};
export async function loadFreshness(): Promise<Freshness[]> {
	const rows = await client<
		{
			source_id: string;
			name: string;
			last_ok: Date | string | null;
			last_run: Date | string | null;
			last_run_ok: boolean | null;
			rows: number | null;
			series: number;
			observations: number;
			latest_period: string | null;
		}[]
	>`
		select s.source_id, s.name,
			(select max(finished_at) from stats.fetch_run r where r.source_id = s.source_id and r.ok) as last_ok,
			l.started_at as last_run, l.ok as last_run_ok, l.rows,
			coalesce(c.series, 0)::int as series, coalesce(c.observations, 0)::int as observations,
			to_char(c.latest, 'YYYY-MM-DD') as latest_period
		from stats.source s
		left join lateral (select started_at, ok, rows from stats.fetch_run r
			where r.source_id = s.source_id order by started_at desc limit 1) l on true
		left join lateral (select count(distinct se.series_id) as series, count(o.*) as observations, max(o.period) as latest
			from stats.series se left join stats.obs o on o.series_id = se.series_id
			where se.source_id = s.source_id) c on true
		order by s.source_id`;
	return rows.map((r) => ({
		sourceId: r.source_id,
		name: r.name,
		lastOk: r.last_ok ? new Date(r.last_ok).toISOString() : null,
		lastRun: r.last_run ? new Date(r.last_run).toISOString() : null,
		lastRunOk: r.last_run_ok,
		rows: r.rows,
		series: r.series,
		observations: r.observations,
		latestPeriod: r.latest_period
	}));
}

/** Rows of a raw CBS table for one measure, grouped by a dimension (e.g. kwb per buurt). */
export async function loadCbsCross(
	table: string,
	measures: string[],
	dim: string,
	codePrefix: string[]
): Promise<{ code: string; title: string; values: Record<string, number | null> }[]> {
	const rows = await client<
		{ code: string; title: string; measure: string; value: number | null }[]
	>`
		select o.dims->>${dim} as code, coalesce(c.title, o.dims->>${dim}) as title, o.measure, o.value
		from stats.cbs_obs o
		left join stats.cbs_code c on c.table_id = o.table_id and c.dimension = ${dim} and c.code = o.dims->>${dim}
		where o.table_id = ${table} and o.measure = any(${measures})
		  and left(o.dims->>${dim}, 6) = any(${codePrefix})`;
	const byCode = new Map<
		string,
		{ code: string; title: string; values: Record<string, number | null> }
	>();
	for (const r of rows) {
		const entry = byCode.get(r.code) ?? { code: r.code, title: r.title, values: {} };
		entry.values[r.measure] = r.value;
		byCode.set(r.code, entry);
	}
	return [...byCode.values()].sort((a, b) => a.code.localeCompare(b.code));
}

export type LatestValue = { period: string; value: number };
/** Latest observed (non-forecast) value per series; missing ids are absent. */
export async function loadLatest(ids: string[]): Promise<Record<string, LatestValue>> {
	if (!ids.length) return {};
	const rows = await client<{ series_id: string; period: string; value: number }[]>`
		select distinct on (series_id) series_id, to_char(period, 'YYYY-MM-DD') as period, value
		from stats.obs
		where series_id = any(${ids}) and value is not null and status is distinct from 'forecast'
		order by series_id, period desc`;
	return Object.fromEntries(rows.map((r) => [r.series_id, { period: r.period, value: r.value }]));
}