feat: house tracker with home worker, alerts and listing map
Collects Pararius rent/buy listings into Postgres via a worker that runs on a residential connection, sends Discord/ntfy alerts for new listings matching rules, and shows an admin-only map, market stats and worker log. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_013HfF5tnChi8tkCx9svJP6D
This commit is contained in:
128
src/lib/server/db/schema/house-tracker.ts
Normal file
128
src/lib/server/db/schema/house-tracker.ts
Normal file
@@ -0,0 +1,128 @@
|
||||
import {
|
||||
bigint,
|
||||
boolean,
|
||||
check,
|
||||
doublePrecision,
|
||||
index,
|
||||
integer,
|
||||
jsonb,
|
||||
pgTable,
|
||||
text,
|
||||
uniqueIndex
|
||||
} from 'drizzle-orm/pg-core';
|
||||
import { sql } from 'drizzle-orm';
|
||||
import { id, timestamptz } from './helpers';
|
||||
|
||||
// Created by scripts/house-tracker-migration.sql; keep both in sync.
|
||||
export const houseListings = pgTable(
|
||||
'house_listings',
|
||||
{
|
||||
id: id(),
|
||||
source: text('source').notNull(),
|
||||
sourceId: text('source_id').notNull(),
|
||||
externalId: text('external_id'),
|
||||
url: text('url').notNull(),
|
||||
offer: text('offer').notNull(),
|
||||
propertyType: text('property_type').notNull(),
|
||||
title: text('title'),
|
||||
street: text('street'),
|
||||
postalCode: text('postal_code'),
|
||||
city: text('city'),
|
||||
citySlug: text('city_slug'),
|
||||
neighbourhood: text('neighbourhood'),
|
||||
priceEur: integer('price_eur'),
|
||||
areaM2: integer('area_m2'),
|
||||
rooms: integer('rooms'),
|
||||
bedrooms: integer('bedrooms'),
|
||||
bathrooms: integer('bathrooms'),
|
||||
constructionYear: integer('construction_year'),
|
||||
energyLabel: text('energy_label'),
|
||||
interior: text('interior'),
|
||||
depositEur: integer('deposit_eur'),
|
||||
availableFrom: text('available_from'),
|
||||
contractType: text('contract_type'),
|
||||
agent: text('agent'),
|
||||
imageUrl: text('image_url'),
|
||||
description: text('description'),
|
||||
features: jsonb('features').$type<Record<string, string>>().notNull().default({}),
|
||||
lat: doublePrecision('lat'),
|
||||
lng: doublePrecision('lng'),
|
||||
offeredAt: timestamptz('offered_at'),
|
||||
firstSeenAt: timestamptz('first_seen_at').notNull().defaultNow(),
|
||||
firstSeenLane: text('first_seen_lane').notNull().default('slow'),
|
||||
lastSeenAt: timestamptz('last_seen_at').notNull().defaultNow(),
|
||||
detailFetchedAt: timestamptz('detail_fetched_at'),
|
||||
removedAt: timestamptz('removed_at'),
|
||||
alertedAt: timestamptz('alerted_at'),
|
||||
status: text('status'),
|
||||
notes: text('notes')
|
||||
},
|
||||
(table) => [
|
||||
check('house_listings_offer_check', sql`offer IN ('rent', 'buy')`),
|
||||
check('house_listings_price_eur_check', sql`price_eur IS NULL OR price_eur >= 0`),
|
||||
uniqueIndex('house_listings_source_idx').on(table.source, table.sourceId),
|
||||
index('house_listings_active_idx')
|
||||
.on(table.offer, table.firstSeenAt.desc())
|
||||
.where(sql`removed_at IS NULL`),
|
||||
index('house_listings_city_idx').on(table.source, table.offer, table.citySlug),
|
||||
index('house_listings_details_idx')
|
||||
.on(table.firstSeenAt.desc())
|
||||
.where(sql`detail_fetched_at IS NULL`)
|
||||
]
|
||||
);
|
||||
|
||||
export const housePriceHistory = pgTable(
|
||||
'house_price_history',
|
||||
{
|
||||
listingId: integer('listing_id')
|
||||
.notNull()
|
||||
.references(() => houseListings.id, { onDelete: 'cascade' }),
|
||||
observedAt: timestamptz('observed_at').notNull().defaultNow(),
|
||||
priceEur: integer('price_eur').notNull()
|
||||
},
|
||||
(table) => [index('house_price_history_listing_idx').on(table.listingId, table.observedAt)]
|
||||
);
|
||||
|
||||
export const houseSearches = pgTable('house_searches', {
|
||||
searchUrl: text('search_url').primaryKey(),
|
||||
source: text('source').notNull(),
|
||||
lane: text('lane').notNull(),
|
||||
firstSyncedAt: timestamptz('first_synced_at').notNull().defaultNow(),
|
||||
lastSyncedAt: timestamptz('last_synced_at').notNull().defaultNow(),
|
||||
lastFound: integer('last_found').notNull().default(0)
|
||||
});
|
||||
|
||||
export const houseWorkerRuns = pgTable(
|
||||
'house_worker_runs',
|
||||
{
|
||||
id: bigint('id', { mode: 'number' }).primaryKey().generatedAlwaysAsIdentity(),
|
||||
source: text('source').notNull(),
|
||||
lane: text('lane').notNull(),
|
||||
url: text('url').notNull(),
|
||||
status: text('status').notNull(),
|
||||
httpStatus: integer('http_status'),
|
||||
found: integer('found'),
|
||||
durationMs: integer('duration_ms'),
|
||||
message: text('message'),
|
||||
createdAt: timestamptz('created_at').notNull().defaultNow()
|
||||
},
|
||||
(table) => [index('house_worker_runs_created_idx').on(table.createdAt.desc())]
|
||||
);
|
||||
|
||||
export const houseAlertRules = pgTable(
|
||||
'house_alert_rules',
|
||||
{
|
||||
id: id(),
|
||||
name: text('name').notNull(),
|
||||
enabled: boolean('enabled').notNull().default(true),
|
||||
offer: text('offer').notNull().default('rent'),
|
||||
minPrice: integer('min_price'),
|
||||
maxPrice: integer('max_price'),
|
||||
minArea: integer('min_area'),
|
||||
minBedrooms: integer('min_bedrooms'),
|
||||
cities: text('cities').array().notNull().default([]),
|
||||
propertyTypes: text('property_types').array().notNull().default([]),
|
||||
createdAt: timestamptz('created_at').notNull().defaultNow()
|
||||
},
|
||||
() => [check('house_alert_rules_offer_check', sql`offer IN ('rent', 'buy')`)]
|
||||
);
|
||||
@@ -10,3 +10,4 @@ export * from './animaldex';
|
||||
export * from './lists';
|
||||
export * from './money-tracker';
|
||||
export * from './cs-prices';
|
||||
export * from './house-tracker';
|
||||
|
||||
360
src/lib/server/house-tracker.ts
Normal file
360
src/lib/server/house-tracker.ts
Normal file
@@ -0,0 +1,360 @@
|
||||
// House tracker ingest: the home worker POSTs what it scraped, this stores it
|
||||
// and sends alerts. Nothing here talks to the listing sites.
|
||||
import { timingSafeEqual } from 'node:crypto';
|
||||
import { env } from '$env/dynamic/private';
|
||||
import { error } from '@sveltejs/kit';
|
||||
import { and, desc, eq, inArray, isNull, lt, sql } from 'drizzle-orm';
|
||||
import { db } from '$lib/server/db';
|
||||
import {
|
||||
houseAlertRules,
|
||||
houseListings,
|
||||
housePriceHistory,
|
||||
houseSearches,
|
||||
houseWorkerRuns
|
||||
} from '$lib/server/db/schema';
|
||||
import {
|
||||
alertMessage,
|
||||
matchesRule,
|
||||
type AlertMessage,
|
||||
type AlertRule
|
||||
} from '$lib/house-tracker/alerts';
|
||||
import { listingRef } from '$lib/house-tracker/pararius';
|
||||
import type {
|
||||
DetailsRequest,
|
||||
ListingCard,
|
||||
ReportRequest,
|
||||
SyncRequest,
|
||||
SyncResponse
|
||||
} from '$lib/house-tracker/types';
|
||||
|
||||
export function requireWorker(request: Request) {
|
||||
const expected = env.HOUSE_WORKER_TOKEN ?? '';
|
||||
if (expected.length < 32) error(503, 'HOUSE_WORKER_TOKEN is not configured');
|
||||
const given = (request.headers.get('authorization') ?? '').replace(/^Bearer\s+/i, '');
|
||||
const a = Buffer.from(given);
|
||||
const b = Buffer.from(expected);
|
||||
if (a.length !== b.length || !timingSafeEqual(a, b)) error(401, 'Invalid worker token');
|
||||
}
|
||||
|
||||
function citySlug(card: Pick<ListingCard, 'url'>): string | null {
|
||||
return listingRef(card.url)?.citySlug ?? null;
|
||||
}
|
||||
|
||||
type Tx = Parameters<Parameters<typeof db.transaction>[0]>[0];
|
||||
|
||||
async function recordPrice(
|
||||
tx: Tx,
|
||||
listingId: number,
|
||||
previous: number | null,
|
||||
next: number | null
|
||||
) {
|
||||
if (next == null || next === previous) return;
|
||||
await tx.insert(housePriceHistory).values({ listingId, priceEur: next });
|
||||
}
|
||||
|
||||
export async function syncCards(input: SyncRequest): Promise<SyncResponse> {
|
||||
const sourceIds = input.cards.map((card) => card.sourceId);
|
||||
const toAlert: { id: number; message: AlertMessage }[] = [];
|
||||
const newIds: string[] = [];
|
||||
|
||||
const rules = (await db
|
||||
.select()
|
||||
.from(houseAlertRules)
|
||||
.where(eq(houseAlertRules.enabled, true))) as AlertRule[];
|
||||
|
||||
await db.transaction(async (tx) => {
|
||||
const [search] = await tx
|
||||
.select({ firstSyncedAt: houseSearches.firstSyncedAt })
|
||||
.from(houseSearches)
|
||||
.where(eq(houseSearches.searchUrl, input.searchUrl));
|
||||
const seeding = !search;
|
||||
await tx
|
||||
.insert(houseSearches)
|
||||
.values({
|
||||
searchUrl: input.searchUrl,
|
||||
source: input.source,
|
||||
lane: input.lane,
|
||||
lastFound: input.cards.length
|
||||
})
|
||||
.onConflictDoUpdate({
|
||||
target: houseSearches.searchUrl,
|
||||
set: { lastSyncedAt: sql`now()`, lastFound: input.cards.length, lane: input.lane }
|
||||
});
|
||||
|
||||
const existing = sourceIds.length
|
||||
? await tx
|
||||
.select({
|
||||
id: houseListings.id,
|
||||
sourceId: houseListings.sourceId,
|
||||
priceEur: houseListings.priceEur
|
||||
})
|
||||
.from(houseListings)
|
||||
.where(
|
||||
and(eq(houseListings.source, input.source), inArray(houseListings.sourceId, sourceIds))
|
||||
)
|
||||
: [];
|
||||
const known = new Map(existing.map((row) => [row.sourceId, row]));
|
||||
|
||||
for (const card of input.cards) {
|
||||
const previous = known.get(card.sourceId);
|
||||
if (previous) {
|
||||
await tx
|
||||
.update(houseListings)
|
||||
.set({
|
||||
lastSeenAt: sql`now()`,
|
||||
removedAt: null,
|
||||
url: card.url,
|
||||
priceEur: card.priceEur ?? previous.priceEur,
|
||||
title: sql`coalesce(${houseListings.title}, ${card.title ?? null})`,
|
||||
postalCode: sql`coalesce(${houseListings.postalCode}, ${card.postalCode ?? null})`,
|
||||
neighbourhood: sql`coalesce(${houseListings.neighbourhood}, ${card.neighbourhood ?? null})`,
|
||||
areaM2: sql`coalesce(${houseListings.areaM2}, ${card.areaM2 ?? null}::int)`,
|
||||
rooms: sql`coalesce(${houseListings.rooms}, ${card.rooms ?? null}::int)`,
|
||||
agent: sql`coalesce(${houseListings.agent}, ${card.agent ?? null})`,
|
||||
imageUrl: sql`coalesce(${houseListings.imageUrl}, ${card.imageUrl ?? null})`
|
||||
})
|
||||
.where(eq(houseListings.id, previous.id));
|
||||
await recordPrice(tx, previous.id, previous.priceEur, card.priceEur ?? null);
|
||||
continue;
|
||||
}
|
||||
|
||||
const [row] = await tx
|
||||
.insert(houseListings)
|
||||
.values({
|
||||
source: input.source,
|
||||
sourceId: card.sourceId,
|
||||
url: card.url,
|
||||
offer: card.offer,
|
||||
propertyType: card.propertyType,
|
||||
title: card.title,
|
||||
postalCode: card.postalCode,
|
||||
city: card.city,
|
||||
citySlug: citySlug(card),
|
||||
neighbourhood: card.neighbourhood,
|
||||
priceEur: card.priceEur,
|
||||
areaM2: card.areaM2,
|
||||
rooms: card.rooms,
|
||||
interior: card.interior,
|
||||
agent: card.agent,
|
||||
imageUrl: card.imageUrl,
|
||||
// Seeded listings were already online; their detection delay means nothing.
|
||||
firstSeenLane: seeding ? 'seed' : input.lane
|
||||
})
|
||||
.onConflictDoNothing()
|
||||
.returning({ id: houseListings.id });
|
||||
if (!row) continue;
|
||||
newIds.push(card.sourceId);
|
||||
await recordPrice(tx, row.id, null, card.priceEur ?? null);
|
||||
if (seeding) continue;
|
||||
const listing = { ...card, citySlug: citySlug(card), source: input.source };
|
||||
const rule = rules.find((candidate) => matchesRule(listing, candidate));
|
||||
if (rule) toAlert.push({ id: row.id, message: alertMessage(listing, rule) });
|
||||
}
|
||||
|
||||
if (input.completedSweep) {
|
||||
await tx
|
||||
.update(houseListings)
|
||||
.set({ removedAt: sql`now()` })
|
||||
.where(
|
||||
and(
|
||||
eq(houseListings.source, input.source),
|
||||
eq(houseListings.offer, input.completedSweep.offer),
|
||||
eq(houseListings.citySlug, input.completedSweep.city),
|
||||
isNull(houseListings.removedAt),
|
||||
lt(houseListings.lastSeenAt, new Date(input.completedSweep.sweepStartedAt))
|
||||
)
|
||||
);
|
||||
}
|
||||
|
||||
await tx.insert(houseWorkerRuns).values({
|
||||
source: input.source,
|
||||
lane: input.lane,
|
||||
url: input.searchUrl,
|
||||
status: 'ok',
|
||||
found: input.cards.length,
|
||||
durationMs: Math.max(0, Date.now() - Date.parse(input.startedAt)),
|
||||
message: seeding ? 'first sync: seeded without alerts' : null
|
||||
});
|
||||
});
|
||||
|
||||
let alerted = 0;
|
||||
for (const { id, message } of toAlert) {
|
||||
if (await sendAlert(message)) {
|
||||
alerted++;
|
||||
await db
|
||||
.update(houseListings)
|
||||
.set({ alertedAt: sql`now()` })
|
||||
.where(eq(houseListings.id, id));
|
||||
}
|
||||
}
|
||||
|
||||
// Fast lane: detail pages for this search first. Otherwise slowly backfill.
|
||||
const pending = await db
|
||||
.select({ sourceId: houseListings.sourceId, url: houseListings.url })
|
||||
.from(houseListings)
|
||||
.where(
|
||||
and(
|
||||
eq(houseListings.source, input.source),
|
||||
isNull(houseListings.detailFetchedAt),
|
||||
isNull(houseListings.removedAt),
|
||||
sourceIds.length && input.lane === 'fast'
|
||||
? inArray(houseListings.sourceId, sourceIds)
|
||||
: undefined
|
||||
)
|
||||
)
|
||||
.orderBy(desc(houseListings.firstSeenAt))
|
||||
.limit(input.lane === 'fast' ? 10 : 5);
|
||||
|
||||
return { newIds, needDetails: pending, alerted };
|
||||
}
|
||||
|
||||
export async function saveDetails(input: DetailsRequest) {
|
||||
let saved = 0;
|
||||
await db.transaction(async (tx) => {
|
||||
for (const detail of input.details) {
|
||||
const [previous] = await tx
|
||||
.select({ id: houseListings.id, priceEur: houseListings.priceEur })
|
||||
.from(houseListings)
|
||||
.where(
|
||||
and(eq(houseListings.source, input.source), eq(houseListings.sourceId, detail.sourceId))
|
||||
);
|
||||
const values = {
|
||||
externalId: detail.externalId,
|
||||
url: detail.url,
|
||||
offer: detail.offer,
|
||||
propertyType: detail.propertyType,
|
||||
title: detail.title,
|
||||
street: detail.street,
|
||||
postalCode: detail.postalCode,
|
||||
city: detail.city,
|
||||
citySlug: citySlug(detail),
|
||||
neighbourhood: detail.neighbourhood,
|
||||
priceEur: detail.priceEur,
|
||||
areaM2: detail.areaM2,
|
||||
rooms: detail.rooms,
|
||||
bedrooms: detail.bedrooms,
|
||||
bathrooms: detail.bathrooms,
|
||||
constructionYear: detail.constructionYear,
|
||||
energyLabel: detail.energyLabel,
|
||||
interior: detail.interior,
|
||||
depositEur: detail.depositEur,
|
||||
availableFrom: detail.availableFrom,
|
||||
contractType: detail.contractType,
|
||||
agent: detail.agent,
|
||||
imageUrl: detail.imageUrl,
|
||||
description: detail.description,
|
||||
features: detail.features ?? {},
|
||||
lat: detail.lat,
|
||||
lng: detail.lng,
|
||||
offeredAt: detail.offeredAt ? new Date(detail.offeredAt) : null,
|
||||
detailFetchedAt: sql`now()`,
|
||||
lastSeenAt: sql`now()`
|
||||
};
|
||||
if (previous) {
|
||||
await tx.update(houseListings).set(values).where(eq(houseListings.id, previous.id));
|
||||
await recordPrice(tx, previous.id, previous.priceEur, detail.priceEur ?? null);
|
||||
} else {
|
||||
const [row] = await tx
|
||||
.insert(houseListings)
|
||||
.values({ ...values, source: input.source, sourceId: detail.sourceId })
|
||||
.returning({ id: houseListings.id });
|
||||
await recordPrice(tx, row.id, null, detail.priceEur ?? null);
|
||||
}
|
||||
saved++;
|
||||
}
|
||||
});
|
||||
return { saved };
|
||||
}
|
||||
|
||||
export async function recordReport(input: ReportRequest) {
|
||||
// One "challenged" notification per 6 hours is plenty.
|
||||
const [recent] =
|
||||
input.status === 'challenged'
|
||||
? await db
|
||||
.select({ id: houseWorkerRuns.id })
|
||||
.from(houseWorkerRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(houseWorkerRuns.status, 'challenged'),
|
||||
sql`${houseWorkerRuns.createdAt} > now() - interval '6 hours'`
|
||||
)
|
||||
)
|
||||
.limit(1)
|
||||
: [];
|
||||
await db.insert(houseWorkerRuns).values({
|
||||
source: input.source,
|
||||
lane: input.lane,
|
||||
url: input.url,
|
||||
status: input.status,
|
||||
httpStatus: input.httpStatus,
|
||||
found: input.found,
|
||||
durationMs: input.durationMs,
|
||||
message: input.message
|
||||
});
|
||||
// Keep a month of history; runs are only for spotting trouble.
|
||||
await db
|
||||
.delete(houseWorkerRuns)
|
||||
.where(lt(houseWorkerRuns.createdAt, sql`now() - interval '30 days'`));
|
||||
if (input.status === 'challenged' && !recent) {
|
||||
await sendAlert({
|
||||
title: '⚠️ House worker was challenged',
|
||||
body: `${input.source} answered ${input.httpStatus ?? '?'} on ${input.url}. The worker is backing off.`,
|
||||
url: input.url
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
/** Discord webhook and/or ntfy. Returns true when at least one channel accepted it. */
|
||||
export async function sendAlert(message: AlertMessage): Promise<boolean> {
|
||||
const jobs: Promise<boolean>[] = [];
|
||||
if (env.HOUSE_DISCORD_WEBHOOK_URL) {
|
||||
jobs.push(
|
||||
fetch(env.HOUSE_DISCORD_WEBHOOK_URL, {
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json' },
|
||||
body: JSON.stringify({
|
||||
username: 'house tracker',
|
||||
embeds: [
|
||||
{
|
||||
title: message.title.slice(0, 256),
|
||||
url: message.url,
|
||||
description: message.body.slice(0, 2000),
|
||||
color: 0x2f9e44
|
||||
}
|
||||
]
|
||||
}),
|
||||
signal: AbortSignal.timeout(8000)
|
||||
}).then(
|
||||
(response) => response.ok,
|
||||
() => false
|
||||
)
|
||||
);
|
||||
}
|
||||
if (env.HOUSE_NTFY_URL) {
|
||||
const url = new URL(env.HOUSE_NTFY_URL);
|
||||
const topic = url.pathname.replace(/^\/+|\/+$/g, '');
|
||||
jobs.push(
|
||||
fetch(url.origin, {
|
||||
method: 'POST',
|
||||
headers: {
|
||||
'content-type': 'application/json',
|
||||
...(env.HOUSE_NTFY_TOKEN ? { authorization: `Bearer ${env.HOUSE_NTFY_TOKEN}` } : {})
|
||||
},
|
||||
body: JSON.stringify({
|
||||
topic,
|
||||
title: message.title,
|
||||
message: message.body,
|
||||
click: message.url,
|
||||
priority: 4,
|
||||
tags: ['house']
|
||||
}),
|
||||
signal: AbortSignal.timeout(8000)
|
||||
}).then(
|
||||
(response) => response.ok,
|
||||
() => false
|
||||
)
|
||||
);
|
||||
}
|
||||
if (!jobs.length) return false;
|
||||
return (await Promise.all(jobs)).some(Boolean);
|
||||
}
|
||||
Reference in New Issue
Block a user