← back to Nationalrealestate
TK-10155: listing lifecycle detector + sweep + backfill + ingest first-see hook + lifecycle/withdrawn API (migration 014)
a7cc0b662e6c3c735e19f0d4e703e670526ddb08 · 2026-08-02 23:30:55 -0700 · Steve
Files touched
A db/backfill-listing-lifecycle.tsA db/migrations/014_listing_lifecycle.sqlM package.jsonM src/ingest/listings/engine.tsA src/jobs/listing-lifecycle-sweep.tsA src/server/listing-lifecycle.tsM src/server/listings.ts
Diff
commit a7cc0b662e6c3c735e19f0d4e703e670526ddb08
Author: Steve <steve@designerwallcoverings.com>
Date: Sun Aug 2 23:30:55 2026 -0700
TK-10155: listing lifecycle detector + sweep + backfill + ingest first-see hook + lifecycle/withdrawn API (migration 014)
---
db/backfill-listing-lifecycle.ts | 111 ++++++++++++++
db/migrations/014_listing_lifecycle.sql | 73 +++++++++
package.json | 4 +-
src/ingest/listings/engine.ts | 36 ++++-
src/jobs/listing-lifecycle-sweep.ts | 72 +++++++++
src/server/listing-lifecycle.ts | 252 ++++++++++++++++++++++++++++++++
src/server/listings.ts | 83 ++++++++++-
7 files changed, 626 insertions(+), 5 deletions(-)
diff --git a/db/backfill-listing-lifecycle.ts b/db/backfill-listing-lifecycle.ts
new file mode 100644
index 0000000..21f7e9e
--- /dev/null
+++ b/db/backfill-listing-lifecycle.ts
@@ -0,0 +1,111 @@
+/**
+ * TK-10155: seed the lifecycle system with a full baseline from day one.
+ *
+ * Classifies EVERY current listing into its derived state and writes an initial
+ * lifecycle_event + a baseline listing_snapshot per listing (so the system has a complete
+ * baseline the moment it turns on). Re-runnable: recordLifecycle is append-only and no-ops
+ * when the latest event already matches the derived state, so running twice does not double-seed
+ * (a second run captures no baseline snapshot because the state is unchanged).
+ *
+ * NOTE: this baseline intentionally captures a snapshot for EVERY listing regardless of state
+ * (via captured_reason='baseline'), then lays down the derived-state event. That guarantees a
+ * preserved copy exists for every listing from the start — the withdrawal-snapshot promise.
+ *
+ * npm run lifecycle:backfill
+ * npx tsx db/backfill-listing-lifecycle.ts
+ */
+import 'dotenv/config';
+import { pool, query } from './pool.ts';
+import { computeAddressKey } from '../src/server/broker-of-record.ts';
+import {
+ recordLifecycle,
+ classifyState,
+ type LifecycleListing,
+ type LifecycleState,
+} from '../src/server/listing-lifecycle.ts';
+
+async function main() {
+ const beforeEv = await query<{ n: string }>(`SELECT count(*)::text n FROM listing_lifecycle_event`);
+ const beforeSnap = await query<{ n: string }>(`SELECT count(*)::text n FROM listing_snapshot`);
+ console.log(`[backfill] events before: ${beforeEv.rows[0].n}`);
+ console.log(`[backfill] snapshots before: ${beforeSnap.rows[0].n}`);
+
+ const maxRes = await query<{ source: string; max_last_seen: string }>(
+ `SELECT source, MAX(last_seen) AS max_last_seen FROM listing GROUP BY source`,
+ );
+ const sourceMax = new Map<string, Date>();
+ for (const r of maxRes.rows) if (r.max_last_seen) sourceMax.set(r.source, new Date(r.max_last_seen));
+
+ const rows = await query<LifecycleListing>(
+ `SELECT id, source, source_id, address, price, status, first_seen, last_seen
+ FROM listing ORDER BY id`,
+ );
+
+ const now = new Date();
+ const stateCounts: Record<string, number> = {};
+ let events = 0;
+ let baselineSnaps = 0;
+ let stateSnaps = 0;
+
+ for (const l of rows.rows) {
+ const ref = sourceMax.get(l.source);
+ if (!ref) continue;
+ const ctx = { sourceMaxLastSeen: ref, now };
+ const derived: LifecycleState = classifyState(l, ctx);
+ stateCounts[derived] = (stateCounts[derived] || 0) + 1;
+
+ // 1. baseline snapshot for EVERY listing (only if none exists yet — re-run safe).
+ if (l.id != null) {
+ const exists = await query<{ n: string }>(
+ `SELECT count(*)::text n FROM listing_snapshot WHERE source=$1 AND source_id=$2 AND captured_reason='baseline'`,
+ [l.source, l.source_id],
+ );
+ if (Number(exists.rows[0].n) === 0) {
+ const snap = await query<Record<string, any>>(
+ `SELECT l.*, b.name AS broker_name, f.name AS firm_name,
+ rg.name AS region_name, rg.name AS county_name, rg.state_code AS region_state_code
+ FROM listing l
+ LEFT JOIN broker b ON b.id = l.broker_id
+ LEFT JOIN firm f ON f.id = l.firm_id
+ LEFT JOIN region rg ON rg.id = l.region_id
+ WHERE l.id = $1`,
+ [l.id],
+ );
+ if (snap.rows[0]) {
+ await query(
+ `INSERT INTO listing_snapshot (listing_id, source, source_id, address_key, snapshot, captured_reason)
+ VALUES ($1,$2,$3,$4,$5,'baseline')`,
+ [l.id, l.source, l.source_id, computeAddressKey(l.address, null) || null, JSON.stringify(snap.rows[0])],
+ );
+ baselineSnaps++;
+ }
+ }
+ }
+
+ // 2. derived-state event (+ any state-triggered snapshot via recordLifecycle).
+ const res = await recordLifecycle(query, l, ctx);
+ if (res.eventInserted) events++;
+ if (res.snapshotInserted) stateSnaps++;
+ }
+
+ const afterEv = await query<{ n: string }>(`SELECT count(*)::text n FROM listing_lifecycle_event`);
+ const afterSnap = await query<{ n: string }>(`SELECT count(*)::text n FROM listing_snapshot`);
+
+ console.log('\n[backfill] derived state (all listings):');
+ for (const [s, n] of Object.entries(stateCounts).sort((a, b) => b[1] - a[1])) {
+ console.log(` ${s.padEnd(12)} ${n}`);
+ }
+ console.log(`[backfill] listings scanned: ${rows.rows.length}`);
+ console.log(`[backfill] lifecycle events inserted: ${events}`);
+ console.log(`[backfill] baseline snapshots: ${baselineSnaps}`);
+ console.log(`[backfill] state-trigger snapshots: ${stateSnaps}`);
+ console.log(`[backfill] events after: ${afterEv.rows[0].n}`);
+ console.log(`[backfill] snapshots after: ${afterSnap.rows[0].n}`);
+
+ await pool.end();
+}
+
+main().catch((e) => {
+ console.error(e);
+ process.exit(1);
+});
diff --git a/db/migrations/014_listing_lifecycle.sql b/db/migrations/014_listing_lifecycle.sql
new file mode 100644
index 0000000..3bcdbc2
--- /dev/null
+++ b/db/migrations/014_listing_lifecycle.sql
@@ -0,0 +1,73 @@
+-- TK-10155: listing-lifecycle tracking + withdrawal-snapshot preservation.
+--
+-- Two append-only tables + one "latest state" view. Built on the same discipline as
+-- broker_of_record_history (migration 013): NOTHING here mutates existing listing/catalog
+-- data — new tables/view only, all CREATE ... IF NOT EXISTS so the migration is idempotent
+-- and reversible. No BEGIN/COMMIT here — migrate.ts wraps each file in a transaction.
+--
+-- GOAL (Steve): track every broker listing across its whole lifecycle — new, active, stale,
+-- withdrawn, closed, reappeared — and, critically, PRESERVE A FULL SNAPSHOT of all property
+-- info the moment a listing becomes withdrawn (disappears from the source feed), so the data
+-- survives after the listing vanishes from the vendor.
+
+-- ─────────────────────────────────────────────────────────────────────────────
+-- listing_lifecycle_event — append-only, one row per DETECTED state transition.
+-- The recorder inserts a row only when the derived state DIFFERS from the latest
+-- observation for that listing (or when there is no prior event), so the table is
+-- the change log of every lifecycle update.
+-- ─────────────────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS listing_lifecycle_event (
+ id BIGSERIAL PRIMARY KEY,
+ listing_id BIGINT REFERENCES listing(id), -- nullable: a listing may vanish/be deleted
+ source TEXT,
+ source_id TEXT,
+ address_key TEXT, -- normalized "address, city" (shared normalizer)
+ state TEXT NOT NULL, -- new | active | stale | withdrawn | closed | reappeared
+ prev_state TEXT,
+ price NUMERIC,
+ reason TEXT,
+ observed_at TIMESTAMPTZ NOT NULL DEFAULT now()
+);
+
+CREATE INDEX IF NOT EXISTS idx_lle_listing_observed ON listing_lifecycle_event (listing_id, observed_at DESC);
+CREATE INDEX IF NOT EXISTS idx_lle_addrkey_observed ON listing_lifecycle_event (address_key, observed_at DESC);
+CREATE INDEX IF NOT EXISTS idx_lle_state ON listing_lifecycle_event (state);
+CREATE INDEX IF NOT EXISTS idx_lle_source_sid_obs ON listing_lifecycle_event (source, source_id, observed_at DESC);
+
+-- ─────────────────────────────────────────────────────────────────────────────
+-- listing_snapshot — append-only, the FULL property info preserved at a point in time.
+-- The snapshot JSONB carries the ENTIRE listing row + joined broker_name + firm_name +
+-- region/county + every field available, so a pulled listing's data is still browsable
+-- after it vanishes from the vendor feed.
+-- ─────────────────────────────────────────────────────────────────────────────
+CREATE TABLE IF NOT EXISTS listing_snapshot (
+ id BIGSERIAL PRIMARY KEY,
+ listing_id BIGINT REFERENCES listing(id), -- nullable, same reasoning as above
+ source TEXT,
+ source_id TEXT,
+ address_key TEXT,
+ snapshot JSONB NOT NULL, -- full listing row + broker/firm/region names
+ captured_at TIMESTAMPTZ NOT NULL DEFAULT now(),
+ captured_reason TEXT -- withdrawn | closed | stale | new | periodic | manual
+);
+
+CREATE INDEX IF NOT EXISTS idx_lsnap_source_sid_cap ON listing_snapshot (source, source_id, captured_at DESC);
+CREATE INDEX IF NOT EXISTS idx_lsnap_addrkey_cap ON listing_snapshot (address_key, captured_at DESC);
+CREATE INDEX IF NOT EXISTS idx_lsnap_reason ON listing_snapshot (captured_reason);
+
+-- ─────────────────────────────────────────────────────────────────────────────
+-- listing_current_state — the latest lifecycle state per listing (DISTINCT ON the
+-- newest event per source/source_id). LEFT-JOINed by the listings API so every row
+-- can carry its lifecycle_state + lifecycle_since.
+-- ─────────────────────────────────────────────────────────────────────────────
+CREATE OR REPLACE VIEW listing_current_state AS
+SELECT DISTINCT ON (source, source_id)
+ source,
+ source_id,
+ listing_id,
+ address_key,
+ state,
+ reason,
+ observed_at
+FROM listing_lifecycle_event
+ORDER BY source, source_id, observed_at DESC;
diff --git a/package.json b/package.json
index 622db49..d51c972 100644
--- a/package.json
+++ b/package.json
@@ -29,7 +29,9 @@
"ingest:commercial-briefs": "tsx src/ingest/commercial/briefs.ts",
"ingest:places": "tsx src/ingest/places/seed.ts",
"loop": "tsx src/jobs/hourly_loop.ts",
- "ingest:zipcounty": "tsx src/ingest/zip_county.ts"
+ "ingest:zipcounty": "tsx src/ingest/zip_county.ts",
+ "lifecycle:sweep": "tsx src/jobs/listing-lifecycle-sweep.ts",
+ "lifecycle:backfill": "tsx db/backfill-listing-lifecycle.ts"
},
"dependencies": {
"better-sqlite3": "^13.0.1",
diff --git a/src/ingest/listings/engine.ts b/src/ingest/listings/engine.ts
index ee6cd87..4e35e21 100644
--- a/src/ingest/listings/engine.ts
+++ b/src/ingest/listings/engine.ts
@@ -21,6 +21,7 @@ import { realtytexas } from './realtytexas.ts';
import { weichert } from './weichert.ts';
import { politeFetch, robotsGate, resolveFirmByHost, resolveRegion, resolveRegionByZip } from './shared.ts';
import { recordBrokerOfRecord } from '../../server/broker-of-record.ts';
+import { recordLifecycle } from '../../server/listing-lifecycle.ts';
const ADAPTERS: ListingAdapter[] = [coldwellbanker, realtytexas, weichert];
const CAP = Number(process.env.USRE_LISTINGS_CAP || 100); // per-site cap (env-tunable; was pilot 50)
@@ -32,7 +33,9 @@ async function upsertListing(
firmName: string | null,
regionId: number | null,
): Promise<void> {
- await query(
+ // RETURNING id + (xmax = 0) tells us whether this row was freshly INSERTED (xmax=0) or
+ // updated (an existing row), so the lifecycle hook can capture 'new' + a snapshot on first-see.
+ const up = await query<{ id: number; freshly_inserted: boolean }>(
`INSERT INTO listing
(source, source_id, firm_id, region_id, address, price, beds, baths, sqft, lat, lng, url, status, last_seen)
VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13, NOW())
@@ -48,7 +51,8 @@ async function upsertListing(
lng = COALESCE(EXCLUDED.lng, listing.lng),
url = EXCLUDED.url,
status = EXCLUDED.status,
- last_seen= NOW()`,
+ last_seen= NOW()
+ RETURNING id, (xmax = 0) AS freshly_inserted`,
[source, f.sourceId, firmId, regionId, f.address, f.price, f.beds, f.baths, f.sqft, f.lat, f.lng, f.url, f.status],
);
@@ -70,6 +74,34 @@ async function upsertListing(
} catch (e: any) {
console.warn(`[${source}] broker-of-record record skipped: ${e?.message || e}`);
}
+
+ // TK-10155: snapshot-on-first-see. If this listing was FRESHLY INSERTED, fire recordLifecycle
+ // so its 'new' lifecycle_event + an initial full snapshot are captured immediately (don't wait
+ // for the sweep). Non-fatal — a lifecycle failure must never break listing ingest. The sweep
+ // handles all later transitions (stale/withdrawn/closed) for both new and existing rows.
+ try {
+ const row = up.rows[0];
+ if (row?.freshly_inserted && row.id != null) {
+ // ingest-time reference: this row was just seen, so "now" is the source's most-recent ingest.
+ const now = new Date();
+ await recordLifecycle(
+ query,
+ {
+ id: row.id,
+ source,
+ source_id: f.sourceId,
+ address: f.address,
+ price: f.price ?? null,
+ status: f.status ?? null,
+ first_seen: now,
+ last_seen: now,
+ },
+ { sourceMaxLastSeen: now, now },
+ );
+ }
+ } catch (e: any) {
+ console.warn(`[${source}] lifecycle first-see record skipped: ${e?.message || e}`);
+ }
}
async function runAdapter(a: ListingAdapter): Promise<{ upserted: number; skipped: number }> {
diff --git a/src/jobs/listing-lifecycle-sweep.ts b/src/jobs/listing-lifecycle-sweep.ts
new file mode 100644
index 0000000..c92178e
--- /dev/null
+++ b/src/jobs/listing-lifecycle-sweep.ts
@@ -0,0 +1,72 @@
+/**
+ * TK-10155: listing-lifecycle SWEEP — the batch driver that catches "gone from feed"
+ * withdrawals (the withdrawal-snapshot headline feature).
+ *
+ * Loads every listing plus the per-source MOST-RECENT ingest reference (max last_seen),
+ * runs recordLifecycle() for each, and prints a summary (counts per derived state, # new
+ * lifecycle events written, # snapshots captured). Idempotent — safe to run repeatedly; a
+ * second run with no feed change writes 0 events / 0 snapshots.
+ *
+ * npm run lifecycle:sweep
+ * npx tsx src/jobs/listing-lifecycle-sweep.ts
+ */
+import { pool, query } from '../../db/pool.ts';
+import {
+ recordLifecycle,
+ classifyState,
+ type LifecycleListing,
+ type LifecycleState,
+} from '../server/listing-lifecycle.ts';
+
+async function main() {
+ // per-source most-recent successful ingest reference = max(last_seen) for that source.
+ const maxRes = await query<{ source: string; max_last_seen: string }>(
+ `SELECT source, MAX(last_seen) AS max_last_seen FROM listing GROUP BY source`,
+ );
+ const sourceMax = new Map<string, Date>();
+ for (const r of maxRes.rows) {
+ if (r.max_last_seen) sourceMax.set(r.source, new Date(r.max_last_seen));
+ }
+
+ const rows = await query<LifecycleListing>(
+ `SELECT id, source, source_id, address, price, status, first_seen, last_seen
+ FROM listing
+ ORDER BY id`,
+ );
+
+ const now = new Date();
+ const stateCounts: Record<string, number> = {};
+ let events = 0;
+ let snapshots = 0;
+ let skippedNoRef = 0;
+
+ for (const l of rows.rows) {
+ const ref = sourceMax.get(l.source);
+ if (!ref) { skippedNoRef++; continue; }
+ const ctx = { sourceMaxLastSeen: ref, now };
+
+ // tally the DERIVED state for every listing (the "counts per state" summary).
+ const derived: LifecycleState = classifyState(l, ctx);
+ stateCounts[derived] = (stateCounts[derived] || 0) + 1;
+
+ const res = await recordLifecycle(query, l, ctx);
+ if (res.eventInserted) events++;
+ if (res.snapshotInserted) snapshots++;
+ }
+
+ console.log('\n[lifecycle-sweep] derived state (all listings):');
+ for (const [s, n] of Object.entries(stateCounts).sort((a, b) => b[1] - a[1])) {
+ console.log(` ${s.padEnd(12)} ${n}`);
+ }
+ console.log(`[lifecycle-sweep] listings scanned: ${rows.rows.length}`);
+ console.log(`[lifecycle-sweep] new events written: ${events}`);
+ console.log(`[lifecycle-sweep] snapshots captured: ${snapshots}`);
+ if (skippedNoRef) console.log(`[lifecycle-sweep] skipped (no source ref): ${skippedNoRef}`);
+
+ await pool.end();
+}
+
+main().catch((e) => {
+ console.error(e);
+ process.exit(1);
+});
diff --git a/src/server/listing-lifecycle.ts b/src/server/listing-lifecycle.ts
new file mode 100644
index 0000000..d3ca184
--- /dev/null
+++ b/src/server/listing-lifecycle.ts
@@ -0,0 +1,252 @@
+/**
+ * TK-10155: listing lifecycle state model + detector/recorder.
+ *
+ * Classifies every broker listing into a lifecycle state, records each state TRANSITION
+ * into the append-only listing_lifecycle_event log, and — critically — PRESERVES a FULL
+ * listing_snapshot the moment a listing becomes withdrawn/closed/stale (or on its very first
+ * 'new' event), so the property data survives after the listing vanishes from the vendor feed.
+ *
+ * Append-only discipline mirrors broker-of-record.ts (migration 013): a new event is written
+ * only when the derived state DIFFERS from the latest observation for that listing (or none
+ * exists). Unchanged state = no-op, so it's safe to run repeatedly (idempotent).
+ *
+ * Callable both from server handlers and from ingest/backfill/sweep scripts — takes a db
+ * `query` handle so it doesn't hard-import a pool.
+ *
+ * "gone from feed" = a listing's last_seen is older than the source's MOST-RECENT successful
+ * ingest (the per-source max last_seen) by >= a threshold. That max is the ingest reference —
+ * it means "the feed was refreshed at time T, and this listing wasn't in it."
+ */
+import { computeAddressKey } from './broker-of-record.ts';
+
+type QueryFn = <T extends Record<string, any> = any>(
+ text: string,
+ params?: unknown[],
+) => Promise<{ rows: T[] }>;
+
+// ── Tunable thresholds (named constants) ────────────────────────────────────
+/** first_seen within this many days AND still currently seen ⇒ 'new' */
+export const NEW_DAYS = 7;
+/** not seen for >= this (but < WITHDRAWN_DAYS) ⇒ 'stale' */
+export const STALE_DAYS = 14;
+/** not seen for >= this ⇒ 'withdrawn' (gone from the feed → pulled) */
+export const WITHDRAWN_DAYS = 30;
+
+const DAY_MS = 86_400_000;
+
+export type LifecycleState = 'new' | 'active' | 'stale' | 'withdrawn' | 'closed' | 'reappeared';
+
+// status-string classifiers
+const CLOSED_RE = /sold|closed|off.?market.*sold/i;
+const WITHDRAWN_STATUS_RE = /withdrawn|cancell?ed|expired/i;
+
+export interface LifecycleListing {
+ id?: number | null;
+ source: string;
+ source_id: string;
+ address?: string | null;
+ price?: number | string | null;
+ status?: string | null;
+ first_seen?: string | Date | null;
+ last_seen?: string | Date | null;
+}
+
+export interface LifecycleCtx {
+ /** per-source MOST-RECENT successful ingest reference (max last_seen for the source). */
+ sourceMaxLastSeen: Date;
+ /** "now" reference — defaults to new Date(); injectable for tests. */
+ now?: Date;
+}
+
+function toDate(v: string | Date | null | undefined): Date | null {
+ if (v == null) return null;
+ const d = v instanceof Date ? v : new Date(v);
+ return isNaN(d.getTime()) ? null : d;
+}
+
+/**
+ * classifyState — the pure state model. Returns one of new|active|stale|withdrawn|closed.
+ * ('reappeared' is a transition emitted by the recorder, not by pure classification.)
+ *
+ * - closed: status looks transacted (sold / closed / off-market-sold).
+ * - withdrawn: status looks withdrawn/cancelled/expired, OR gone from the feed
+ * (last_seen older than the source's max last_seen by >= WITHDRAWN_DAYS).
+ * - stale: gone from the feed by >= STALE_DAYS but < WITHDRAWN_DAYS.
+ * - new: first_seen within NEW_DAYS AND still currently seen.
+ * - active: currently seen and not new.
+ */
+export function classifyState(listing: LifecycleListing, ctx: LifecycleCtx): LifecycleState {
+ const status = (listing.status || '').trim();
+
+ // 1. transacted wins outright.
+ if (status && CLOSED_RE.test(status)) return 'closed';
+ // 2. explicit withdrawn/cancelled/expired status.
+ if (status && WITHDRAWN_STATUS_RE.test(status)) return 'withdrawn';
+
+ const ref = ctx.sourceMaxLastSeen;
+ const lastSeen = toDate(listing.last_seen);
+ const firstSeen = toDate(listing.first_seen);
+ const now = ctx.now ?? new Date();
+
+ // gone-from-feed age: how far behind the source's most-recent ingest this listing's
+ // last_seen sits. A listing seen in the latest ingest has age ~0 ⇒ currently seen.
+ const goneDays = lastSeen ? (ref.getTime() - lastSeen.getTime()) / DAY_MS : Infinity;
+
+ if (goneDays >= WITHDRAWN_DAYS) return 'withdrawn';
+ if (goneDays >= STALE_DAYS) return 'stale';
+
+ // currently seen in (or very near) the latest ingest window.
+ const newAgeDays = firstSeen ? (now.getTime() - firstSeen.getTime()) / DAY_MS : Infinity;
+ if (newAgeDays <= NEW_DAYS) return 'new';
+ return 'active';
+}
+
+/** which states preserve a full snapshot when we transition INTO them */
+const SNAPSHOT_STATES = new Set<LifecycleState>(['withdrawn', 'closed', 'stale', 'new']);
+
+const norm = (v: unknown) => (v == null || v === '' ? null : v);
+
+/**
+ * Build the FULL snapshot JSONB for a listing: the entire listing row + joined broker_name,
+ * firm_name, region/county name + state_code. Read straight from the DB so we capture every
+ * field available, not just what the caller happened to pass.
+ */
+async function buildSnapshot(db: QueryFn, listingId: number): Promise<Record<string, any> | null> {
+ const r = await db<Record<string, any>>(
+ `SELECT l.*,
+ b.name AS broker_name,
+ f.name AS firm_name,
+ rg.name AS region_name,
+ rg.name AS county_name,
+ rg.state_code AS region_state_code
+ FROM listing l
+ LEFT JOIN broker b ON b.id = l.broker_id
+ LEFT JOIN firm f ON f.id = l.firm_id
+ LEFT JOIN region rg ON rg.id = l.region_id
+ WHERE l.id = $1`,
+ [listingId],
+ );
+ return r.rows[0] ?? null;
+}
+
+export interface RecordLifecycleResult {
+ state: LifecycleState;
+ prev_state: LifecycleState | null;
+ eventInserted: boolean;
+ snapshotInserted: boolean;
+}
+
+/**
+ * recordLifecycle — compute the current state for a listing; read its latest lifecycle_event;
+ * if the state CHANGED (or no event exists) INSERT a new lifecycle_event (append-only) with
+ * prev_state + reason; AND when transitioning INTO withdrawn|closed|stale|new INSERT a full
+ * listing_snapshot preserving the property info. Idempotent — unchanged state is a no-op.
+ *
+ * A 'reappeared' event is emitted when a listing that was previously stale/withdrawn is seen
+ * again (returns to active/new); no snapshot is captured for reappearance (it's coming BACK,
+ * not being lost).
+ */
+export async function recordLifecycle(
+ db: QueryFn,
+ listing: LifecycleListing,
+ ctx: LifecycleCtx,
+): Promise<RecordLifecycleResult> {
+ let state = classifyState(listing, ctx);
+
+ // latest event for this listing (by source/source_id — stable even if listing_id is null).
+ const prevRes = await db<{ state: LifecycleState }>(
+ `SELECT state FROM listing_lifecycle_event
+ WHERE source = $1 AND source_id = $2
+ ORDER BY observed_at DESC, id DESC
+ LIMIT 1`,
+ [listing.source, listing.source_id],
+ );
+ const prevState: LifecycleState | null = prevRes.rows[0]?.state ?? null;
+
+ // 'reappeared' — was gone (stale/withdrawn), now current again (active/new).
+ if (prevState && (prevState === 'stale' || prevState === 'withdrawn')
+ && (state === 'active' || state === 'new')) {
+ state = 'reappeared';
+ }
+
+ const result: RecordLifecycleResult = {
+ state,
+ prev_state: prevState,
+ eventInserted: false,
+ snapshotInserted: false,
+ };
+
+ // unchanged state ⇒ no-op (append-only, only records transitions).
+ if (prevState === state) return result;
+
+ const addressKey = computeAddressKey(listing.address, null) || null;
+ const price = norm(listing.price) as number | string | null;
+ const reason = reasonFor(state, listing, ctx);
+
+ await db(
+ `INSERT INTO listing_lifecycle_event
+ (listing_id, source, source_id, address_key, state, prev_state, price, reason)
+ VALUES ($1,$2,$3,$4,$5,$6,$7,$8)`,
+ [
+ norm(listing.id) ?? null,
+ listing.source,
+ listing.source_id,
+ addressKey,
+ state,
+ prevState,
+ price,
+ reason,
+ ],
+ );
+ result.eventInserted = true;
+
+ // preserve a FULL snapshot on transitions INTO a snapshot state (incl. first-ever 'new').
+ // 'reappeared' is intentionally NOT a snapshot state — the listing is coming back, not lost.
+ if (SNAPSHOT_STATES.has(state) && listing.id != null) {
+ const snap = await buildSnapshot(db, Number(listing.id));
+ if (snap) {
+ await db(
+ `INSERT INTO listing_snapshot
+ (listing_id, source, source_id, address_key, snapshot, captured_reason)
+ VALUES ($1,$2,$3,$4,$5,$6)`,
+ [
+ Number(listing.id),
+ listing.source,
+ listing.source_id,
+ addressKey,
+ JSON.stringify(snap),
+ state, // captured_reason mirrors the state that triggered preservation
+ ],
+ );
+ result.snapshotInserted = true;
+ }
+ }
+
+ return result;
+}
+
+/** human-readable reason string for an event, tuned per state. */
+function reasonFor(state: LifecycleState, listing: LifecycleListing, ctx: LifecycleCtx): string {
+ const status = (listing.status || '').trim();
+ const lastSeen = toDate(listing.last_seen);
+ const goneDays = lastSeen
+ ? Math.round((ctx.sourceMaxLastSeen.getTime() - lastSeen.getTime()) / DAY_MS)
+ : null;
+ switch (state) {
+ case 'closed':
+ return status ? `status="${status}" matched closed/sold` : 'transacted';
+ case 'withdrawn':
+ return status && WITHDRAWN_STATUS_RE.test(status)
+ ? `status="${status}" matched withdrawn/cancelled/expired`
+ : `gone from feed ${goneDays ?? '?'}d (>= ${WITHDRAWN_DAYS}d)`;
+ case 'stale':
+ return `not seen ${goneDays ?? '?'}d (>= ${STALE_DAYS}d, < ${WITHDRAWN_DAYS}d)`;
+ case 'new':
+ return `first seen within ${NEW_DAYS}d and currently listed`;
+ case 'reappeared':
+ return 'seen again in feed after being stale/withdrawn';
+ case 'active':
+ default:
+ return 'currently listed';
+ }
+}
diff --git a/src/server/listings.ts b/src/server/listings.ts
index dae929d..364d874 100644
--- a/src/server/listings.ts
+++ b/src/server/listings.ts
@@ -12,7 +12,7 @@ import { query } from '../../db/pool.ts';
import { addressKeySql } from './broker-of-record.ts';
export function mountListings(app: Express): void {
- // GET /api/listings?firm=<id>&county=<region_id>&source=&limit=
+ // GET /api/listings?firm=<id>&county=<region_id>&source=&state=<lifecycle>&limit=
app.get('/api/listings', async (req, res) => {
try {
const limit = Math.min(Math.max(Number(req.query.limit) || 500, 1), 2000);
@@ -21,6 +21,8 @@ export function mountListings(app: Express): void {
if (req.query.firm) { params.push(Number(req.query.firm)); conds.push(`l.firm_id = $${params.length}`); }
if (req.query.county) { params.push(Number(req.query.county)); conds.push(`l.region_id = $${params.length}`); }
if (req.query.source) { params.push(String(req.query.source)); conds.push(`l.source = $${params.length}`); }
+ // TK-10155: filter by current lifecycle state (withdrawn|stale|new|active|closed|reappeared).
+ if (req.query.state) { params.push(String(req.query.state)); conds.push(`lcs.state = $${params.length}`); }
const where = conds.length ? 'WHERE ' + conds.join(' AND ') : '';
params.push(limit);
const r = await query(
@@ -31,12 +33,17 @@ export function mountListings(app: Express): void {
-- TK-10139: newest broker/firm of record for this listing's address.
lbor.broker_name AS last_broker_of_record,
lbor.firm_name AS last_firm_of_record,
- lbor.observed_at AS last_broker_observed_at
+ lbor.observed_at AS last_broker_observed_at,
+ -- TK-10155: current lifecycle state + when it was last observed.
+ lcs.state AS lifecycle_state,
+ lcs.observed_at AS lifecycle_since
FROM listing l
LEFT JOIN firm f ON f.id = l.firm_id
LEFT JOIN region rg ON rg.id = l.region_id
LEFT JOIN last_broker_of_record lbor
ON lbor.address_key = ${addressKeySql(`coalesce(l.address,'')`)}
+ LEFT JOIN listing_current_state lcs
+ ON lcs.source = l.source AND lcs.source_id = l.source_id
${where}
ORDER BY l.last_seen DESC NULLS LAST, l.id DESC
LIMIT $${params.length}`,
@@ -48,6 +55,78 @@ export function mountListings(app: Express): void {
}
});
+ // TK-10155: GET /api/listings/:id/lifecycle — full lifecycle history + preserved snapshots
+ // for one listing (newest-first). The headline drill-down: current state, every state
+ // transition, and every preserved property snapshot.
+ app.get('/api/listings/:id/lifecycle', async (req, res) => {
+ try {
+ const id = Number(req.params.id);
+ if (!Number.isInteger(id) || id <= 0) return res.status(400).json({ error: 'bad listing id' });
+ const [events, snaps] = await Promise.all([
+ query(
+ `SELECT id, listing_id, source, source_id, address_key, state, prev_state, price, reason, observed_at
+ FROM listing_lifecycle_event
+ WHERE listing_id = $1
+ ORDER BY observed_at DESC, id DESC`,
+ [id],
+ ),
+ query(
+ `SELECT id, captured_at, captured_reason, snapshot
+ FROM listing_snapshot
+ WHERE listing_id = $1
+ ORDER BY captured_at DESC, id DESC`,
+ [id],
+ ),
+ ]);
+ const current_state = events.rows[0]?.state ?? null;
+ res.json({
+ listing_id: id,
+ current_state,
+ events: events.rows,
+ snapshots: snaps.rows,
+ });
+ } catch (e: any) {
+ res.status(500).json({ error: String(e.message || e) });
+ }
+ });
+
+ // TK-10155: GET /api/withdrawn?limit=N — recently withdrawn listings WITH their preserved
+ // snapshot. The headline feature: everything that got pulled from the feed, full info saved,
+ // newest-withdrawn first. Uses the withdrawn lifecycle_event as the trigger time, and joins
+ // the snapshot that was captured for that withdrawal (falls back to the newest snapshot).
+ app.get('/api/withdrawn', async (req, res) => {
+ try {
+ const limit = Math.min(Math.max(Number(req.query.limit) || 100, 1), 1000);
+ const r = await query(
+ `WITH w AS (
+ -- latest event per listing; keep only those whose current state is withdrawn.
+ SELECT DISTINCT ON (source, source_id)
+ source, source_id, listing_id, state, reason, observed_at
+ FROM listing_lifecycle_event
+ ORDER BY source, source_id, observed_at DESC, id DESC
+ )
+ SELECT w.source, w.source_id, w.listing_id,
+ w.observed_at AS withdrawn_at, w.reason,
+ s.captured_at, s.captured_reason, s.snapshot
+ FROM w
+ LEFT JOIN LATERAL (
+ SELECT captured_at, captured_reason, snapshot
+ FROM listing_snapshot ls
+ WHERE ls.source = w.source AND ls.source_id = w.source_id
+ ORDER BY (ls.captured_reason = 'withdrawn') DESC, ls.captured_at DESC, ls.id DESC
+ LIMIT 1
+ ) s ON true
+ WHERE w.state = 'withdrawn'
+ ORDER BY w.observed_at DESC
+ LIMIT $1`,
+ [limit],
+ );
+ res.json({ count: r.rows.length, rows: r.rows });
+ } catch (e: any) {
+ res.status(500).json({ error: String(e.message || e) });
+ }
+ });
+
// GET /api/listings/stats — count by firm + by source (proves the firm↔listing tie)
app.get('/api/listings/stats', async (_req, res) => {
try {
← 39e9dbe TK-10139: canonical address_key normalizer (strip trailing s
·
back to Nationalrealestate
·
TK-10155: listings.html — color-coded lifecycle state chip + 6193a68 →