/**
 * Steam Community Market ingestion. Run after db:migrate:cs-prices.
 *
 * bun scripts/ingest-cs-prices.ts catalog [--limit=500] [--start=0]
 * STEAM_SESSIONID=... STEAM_LOGIN_SECURE=... bun scripts/ingest-cs-prices.ts history [--limit=100] [--start=0]
 * bun scripts/ingest-cs-prices.ts import ./observations.jsonl
 *
 * The first two modes use Steam's public market web endpoints, which are not a
 * contracted API and may rate limit or change. `import` remains provider-neutral.
 */
import { readFile } from 'node:fs/promises';
import postgres from 'postgres';
import { dimensionsFromName, type Dimensions } from '../src/lib/cs-prices/taxonomy';
import { steamDate } from '../src/lib/cs-prices/steam';

if (!process.env.DB_URL) throw new Error('DB_URL is required');
const sql = postgres(process.env.DB_URL, { max: 1 });
const mode = process.argv[2];
const flags = new Map(
	process.argv
		.slice(3)
		.filter((arg) => arg.startsWith('--'))
		.map((arg) => {
			const [key, value] = arg.slice(2).split('=');
			return [key, value ?? ''];
		})
);
const start = Math.max(0, Number(flags.get('start') ?? 0));
const limit = Math.max(1, Number(flags.get('limit') ?? 1_000_000_000));
const delayMs = Math.max(1000, Number(flags.get('delay-ms') ?? 2000));
const sweepTime = new Date();
const knownStreams = new Set<string>();

type Observation = {
	marketHashName: string;
	name?: string;
	source: string;
	currency: string;
	observedAt: string;
	priceCents: number;
	volume?: number | null;
	kind: string;
	dimensions?: Dimensions;
};

function validObservation(value: unknown): Observation {
	if (!value || typeof value !== 'object') throw new Error('Observation must be an object');
	const row = value as Record<string, unknown>;
	for (const key of ['marketHashName', 'source', 'currency', 'observedAt', 'kind']) {
		if (typeof row[key] !== 'string' || !row[key]) throw new Error(`Invalid ${key}`);
	}
	if (!/^[a-z0-9][a-z0-9-]{0,79}$/.test(row.source as string)) throw new Error('Invalid source');
	if (!/^[a-z0-9][a-z0-9-]{0,79}$/.test(row.kind as string)) throw new Error('Invalid kind');
	if (!/^[A-Z]{3}$/.test(row.currency as string)) throw new Error('Invalid currency');
	if (row.name != null && typeof row.name !== 'string') throw new Error('Invalid name');
	if (
		!Number.isInteger(row.priceCents) ||
		(row.priceCents as number) < 0 ||
		(row.priceCents as number) > 2_147_483_647
	)
		throw new Error('Invalid priceCents');
	if (
		row.volume != null &&
		(!Number.isInteger(row.volume) ||
			(row.volume as number) < 0 ||
			(row.volume as number) > 2_147_483_647)
	)
		throw new Error('Invalid volume');
	if (Number.isNaN(Date.parse(row.observedAt as string))) throw new Error('Invalid observedAt');
	if (
		row.dimensions != null &&
		(typeof row.dimensions !== 'object' ||
			Array.isArray(row.dimensions) ||
			Object.values(row.dimensions).some((value) => typeof value !== 'string'))
	)
		throw new Error('Invalid dimensions');
	return row as Observation;
}

async function itemId(
	name: string,
	displayName = name,
	iconUrl?: string,
	itemType?: string,
	extraDimensions: Dimensions = {}
) {
	const [row] = await sql<{ id: number }[]>`
		INSERT INTO cs_market_items (market_hash_name, name, icon_url, item_type)
		VALUES (${name}, ${displayName}, ${iconUrl ?? null}, ${itemType ?? null})
		ON CONFLICT (market_hash_name) DO UPDATE SET
			name = EXCLUDED.name,
			icon_url = COALESCE(EXCLUDED.icon_url, cs_market_items.icon_url),
			item_type = COALESCE(EXCLUDED.item_type, cs_market_items.item_type),
			last_seen_at = now()
		RETURNING id`;
	const inferred = dimensionsFromName(name);
	if (itemType) inferred.steam_type = itemType;
	for (const [key, value] of Object.entries(inferred)) {
		if (!/^[a-z][a-z0-9_]*$/.test(key) || !value) continue;
		await sql`
			INSERT INTO cs_item_dimensions (item_id, key, value) VALUES (${row.id}, ${key}, ${value})
			ON CONFLICT (item_id, key) DO NOTHING`;
	}
	for (const [key, value] of Object.entries(extraDimensions)) {
		if (!/^[a-z][a-z0-9_]*$/.test(key) || !value) continue;
		await sql`
			INSERT INTO cs_item_dimensions (item_id, key, value) VALUES (${row.id}, ${key}, ${value})
			ON CONFLICT (item_id, key) DO UPDATE SET value = EXCLUDED.value`;
	}
	return row.id;
}

async function saveObservation(row: Observation, knownItemId?: number) {
	const item =
		knownItemId ??
		(await itemId(row.marketHashName, row.name, undefined, undefined, row.dimensions));
	await ensureStream(row.source, row.currency, row.kind);
	await sql`
		INSERT INTO cs_price_observations
			(item_id, source, currency, observed_at, price_cents, volume, kind)
		VALUES (${item}, ${row.source}, ${row.currency}, ${new Date(row.observedAt)},
			${row.priceCents}, ${row.volume ?? null}, ${row.kind})
		ON CONFLICT (item_id, source, currency, observed_at, kind) DO UPDATE SET
			price_cents = EXCLUDED.price_cents,
			volume = EXCLUDED.volume`;
}

async function ensureStream(source: string, currency: string, kind: string) {
	const key = `${source}\0${currency}\0${kind}`;
	if (knownStreams.has(key)) return;
	await sql`INSERT INTO cs_price_streams (source, currency, kind)
		VALUES (${source}, ${currency}, ${kind}) ON CONFLICT DO NOTHING`;
	knownStreams.add(key);
}

async function saveHistoryRows(
	item: { id: number; market_hash_name: string },
	prices: [string, number, string][]
) {
	await ensureStream('steam-market', 'USD', 'sale-median');
	const rows = prices
		.filter(([, price]) => Number.isFinite(price) && price >= 0)
		.map(([date, price, volume]) => ({
			item_id: item.id,
			source: 'steam-market',
			currency: 'USD',
			observed_at: new Date(steamDate(date)),
			price_cents: Math.round(price * 100),
			volume: Number(volume.replaceAll(',', '')) || 0,
			kind: 'sale-median'
		}));
	for (let offset = 0; offset < rows.length; offset += 500) {
		await sql`INSERT INTO cs_price_observations ${sql(rows.slice(offset, offset + 500))}
			ON CONFLICT (item_id, source, currency, observed_at, kind) DO UPDATE SET
				price_cents = EXCLUDED.price_cents, volume = EXCLUDED.volume`;
	}
}

async function steamJson<T>(url: URL, authenticated = false): Promise<T> {
	for (let attempt = 0; attempt < 5; attempt++) {
		const cookie = authenticated
			? `sessionid=${process.env.STEAM_SESSIONID ?? ''}; steamLoginSecure=${process.env.STEAM_LOGIN_SECURE ?? ''}`
			: undefined;
		const response = await fetch(url, {
			headers: {
				'User-Agent': 'cs-prices/1.0 personal historical archive',
				...(cookie ? { Cookie: cookie } : {})
			}
		});
		if (response.ok) {
			const json = await response.json();
			if (!json || json.success === false) throw new Error(`Steam rejected ${url.pathname}`);
			return json as T;
		}
		if (response.status !== 429 && response.status < 500)
			throw new Error(`Steam HTTP ${response.status}`);
		await new Promise((resolve) => setTimeout(resolve, delayMs * 2 ** attempt));
	}
	throw new Error(`Steam request failed after retries: ${url.pathname}`);
}

const pause = () => new Promise((resolve) => setTimeout(resolve, delayMs));

async function catalog() {
	let offset = start;
	let processed = 0;
	while (processed < limit) {
		const url = new URL('https://steamcommunity.com/market/search/render/');
		url.search = new URLSearchParams({
			appid: '730',
			currency: '1',
			country: 'US',
			norender: '1',
			count: String(Math.min(100, limit - processed)),
			start: String(offset),
			sort_column: 'name',
			sort_dir: 'asc'
		}).toString();
		const page = await steamJson<{
			total_count: number;
			results: {
				hash_name: string;
				name: string;
				sell_price?: number;
				sell_listings?: number;
				asset_description?: { icon_url?: string; type?: string };
			}[];
		}>(url);
		if (!page.results?.length) break;
		for (const entry of page.results) {
			if (!entry.hash_name) continue;
			const id = await itemId(
				entry.hash_name,
				entry.name || entry.hash_name,
				entry.asset_description?.icon_url,
				entry.asset_description?.type
			);
			if (Number.isSafeInteger(entry.sell_price) && (entry.sell_price ?? -1) >= 0) {
				await saveObservation(
					{
						marketHashName: entry.hash_name,
						source: 'steam-market',
						currency: 'USD',
						observedAt: sweepTime.toISOString(),
						priceCents: entry.sell_price!,
						volume: entry.sell_listings,
						kind: 'lowest-listing'
					},
					id
				);
			}
		}
		processed += page.results.length;
		offset += page.results.length;
		console.log(`Catalog: ${offset}/${page.total_count} listings`);
		if (offset >= page.total_count) break;
		await pause();
	}
}

async function history() {
	if (!process.env.STEAM_SESSIONID || !process.env.STEAM_LOGIN_SECURE)
		throw new Error('STEAM_SESSIONID and STEAM_LOGIN_SECURE are required for Steam history');
	const items = await sql<{ id: number; market_hash_name: string }[]>`
		SELECT id, market_hash_name FROM cs_market_items ORDER BY id LIMIT ${limit} OFFSET ${start}`;
	let done = 0;
	for (const item of items) {
		const url = new URL('https://steamcommunity.com/market/pricehistory/');
		url.search = new URLSearchParams({
			appid: '730',
			market_hash_name: item.market_hash_name,
			currency: '1',
			country: 'US'
		}).toString();
		const data = await steamJson<{ prices: [string, number, string][] }>(url, true);
		await saveHistoryRows(item, data.prices ?? []);
		console.log(
			`History: ${++done}/${items.length} ${item.market_hash_name} (${data.prices?.length ?? 0} points)`
		);
		await pause();
	}
}

async function importFile(path: string) {
	const lines = (await readFile(path, 'utf8')).split(/\r?\n/);
	let imported = 0;
	for (const [index, line] of lines.entries()) {
		if (!line.trim()) continue;
		try {
			await saveObservation(validObservation(JSON.parse(line)));
			imported++;
		} catch (cause) {
			throw new Error(`Invalid observation on line ${index + 1}: ${String(cause)}`, { cause });
		}
	}
	console.log(`Imported ${imported} observations`);
}

try {
	if (mode === 'catalog') await catalog();
	else if (mode === 'history') await history();
	else if (mode === 'import' && process.argv[3]) await importFile(process.argv[3]);
	else throw new Error('Usage: ingest-cs-prices.ts catalog|history|import <file>');
	await sql`ANALYZE cs_market_items, cs_item_dimensions, cs_price_observations`;
} finally {
	await sql.end();
}