website/src/lib/house-tracker/ingest.ts

// Writes scraped listings into house.listings with plain postgres.js, for the nightly
// scraper on stan-vm (scripts/house-scrape.ts). No #lib / $app imports.
import type postgres from 'postgres';
import { addressKey, addressKeyFromText, citySlug, DEDUPE_SQL } from './address';
import {
	containsPattern,
	linkHost,
	pickLinkedListing,
	withoutEmpty,
	type LinkCandidate
} from './link-key';
import type { ScrapedListing } from './scraped';

type Sql = postgres.Sql;

export type IngestResult = {
	seeded: boolean;
	inserted: { id: number; listing: ScrapedListing }[];
	updated: number;
	/** Rows stored under another source (a hand-added link) that this scrape took over. */
	adopted: number;
	removed: number;
};

/**
 * Rows not keyed under `source` whose link is on the same host as one of `listings`,
 * for pickLinkedListing. One query per ingest.
 */
export async function linkCandidates(
	sql: Sql | postgres.TransactionSql,
	source: string,
	listings: { url: string | null }[]
): Promise<LinkCandidate[]> {
	const hosts = [...new Set(listings.map((l) => linkHost(l.url)).filter((h) => h != null))];
	if (!hosts.length) return [];
	return sql<LinkCandidate[]>`
		SELECT id, source, source_id AS "sourceId", url, duplicate_of AS "duplicateOf"
		FROM house.listings
		WHERE source <> ${source} AND url ILIKE ANY (${hosts.map(containsPattern)}::text[])`;
}

export async function ingestListings(
	sql: Sql,
	source: string,
	listings: ScrapedListing[],
	opts: { runStartedAt: Date; completed: boolean }
): Promise<IngestResult> {
	// The same listing can show up on two city pages; keep the first.
	const unique = [...new Map(listings.map((l) => [l.sourceId, l])).values()];
	return sql.begin(async (tx) => {
		const [any] = await tx`SELECT 1 FROM house.listings WHERE source = ${source} LIMIT 1`;
		const seeded = !any;
		const ids = unique.map((l) => l.sourceId);
		const existing = ids.length
			? await tx<{ id: number; source_id: string; price_eur: number | null }[]>`
				SELECT id, source_id, price_eur FROM house.listings
				WHERE source = ${source} AND source_id IN ${tx(ids)}`
			: [];
		const known = new Map(existing.map((row) => [row.source_id, row]));
		const inserted: IngestResult['inserted'] = [];
		let updated = 0;
		let adopted = 0;
		// A listing already added by hand (source 'manual' with this link): update that row
		// and give it this source's key, instead of inserting a copy.
		let candidates = await linkCandidates(
			tx,
			source,
			unique.filter((l) => !known.has(l.sourceId))
		);

		for (const l of unique) {
			const columns = {
				url: l.url,
				title: l.title,
				street: l.houseNumber ? `${l.street ?? ''} ${l.houseNumber}`.trim() : l.street,
				postal_code: l.postalCode,
				city: l.city,
				city_slug: citySlug(l.city),
				price_eur: l.priceEur,
				total_price_eur: l.totalPriceEur,
				area_m2: l.areaM2,
				rooms: l.rooms,
				bedrooms: l.bedrooms,
				energy_label: l.energyLabel,
				available_from: l.availableFrom,
				interior: l.interior,
				agent: l.agent,
				image_url: l.imageUrl,
				lat: l.lat,
				lng: l.lng,
				features: tx.json(l.features),
				address_key: addressKey(l.street, l.houseNumber, l.city)
			};
			const previous = known.get(l.sourceId);
			if (previous) {
				await tx`UPDATE house.listings SET ${tx(columns)}, last_seen_at = now(), removed_at = NULL
					WHERE id = ${previous.id}`;
				if (l.priceEur != null && l.priceEur !== previous.price_eur)
					await tx`INSERT INTO house.price_history (listing_id, price_eur) VALUES (${previous.id}, ${l.priceEur})`;
				updated++;
				continue;
			}
			const linked = pickLinkedListing(candidates, { source, sourceId: l.sourceId }, l.url);
			if (linked) {
				candidates = candidates.filter((c) => c.id !== linked.id);
				const [before] = await tx<{ price_eur: number | null }[]>`
					SELECT price_eur FROM house.listings WHERE id = ${linked.id} FOR UPDATE`;
				await tx`UPDATE house.listings
					SET ${tx({ ...withoutEmpty(columns), source, source_id: l.sourceId })},
						last_seen_at = now(), removed_at = NULL
					WHERE id = ${linked.id}`;
				if (l.priceEur != null && l.priceEur !== before?.price_eur)
					await tx`INSERT INTO house.price_history (listing_id, price_eur) VALUES (${linked.id}, ${l.priceEur})`;
				updated++;
				adopted++;
				continue;
			}
			const [row] = await tx<{ id: number }[]>`INSERT INTO house.listings ${tx({
				...columns,
				source,
				source_id: l.sourceId,
				offer: l.offer,
				property_type: l.propertyType,
				first_seen_lane: seeded ? 'seed' : 'nightly',
				// Everything the source offers is already in the search result.
				detail_fetched_at: new Date()
			})} ON CONFLICT (source, source_id) DO NOTHING RETURNING id`;
			if (!row) continue;
			if (l.priceEur != null)
				await tx`INSERT INTO house.price_history (listing_id, price_eur) VALUES (${row.id}, ${l.priceEur})`;
			inserted.push({ id: row.id, listing: l });
		}

		let removed = 0;
		if (opts.completed) {
			const gone = await tx`UPDATE house.listings SET removed_at = now()
				WHERE source = ${source} AND removed_at IS NULL AND last_seen_at < ${opts.runStartedAt}
				RETURNING id`;
			removed = gone.length;
		}
		return { seeded, inserted, updated, adopted, removed };
	});
}

/**
 * Fills address_key for rows written elsewhere (manual CLI, Pararius worker) from their
 * street or title, then marks cross-source copies. Safe to run any time.
 */
export async function dedupe(sql: Sql): Promise<{ keyed: number; copies: number }> {
	const rows = await sql<
		{
			id: number;
			source: string;
			street: string | null;
			title: string | null;
			city: string | null;
		}[]
	>`SELECT id, source, street, title, city FROM house.listings
		WHERE address_key IS NULL AND (street IS NOT NULL OR source = 'manual')`;
	let keyed = 0;
	for (const row of rows) {
		const key =
			addressKeyFromText(row.street, row.city) ??
			(row.source === 'manual' ? addressKeyFromText(row.title, row.city) : null);
		if (!key) continue;
		await sql`UPDATE house.listings SET address_key = ${key} WHERE id = ${row.id}`;
		keyed++;
	}
	for (const statement of DEDUPE_SQL) await sql.unsafe(statement);
	const [{ copies }] = await sql<{ copies: number }[]>`
		SELECT count(*)::int AS copies FROM house.listings WHERE duplicate_of IS NOT NULL`;
	return { keyed, copies };
}