`db/backfill-listing-lifecycle.ts`

122 lines
/**
 * 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 captures EXACTLY ONE snapshot per listing (captured_reason='baseline'),
 * then lays down the derived-state event with suppressSnapshot=true so recordLifecycle does NOT
 * fire a SECOND ('new') snapshot on top of it (fix 4 — that double-count made 599 listings write
 * 1026 snapshots). One preserved copy per 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));

  // fix 1: last full rescan per source (null = incremental-only → no gone-from-feed withdrawal).
  const fullRes = await query<{ source: string; last_full: string }>(
    `SELECT source, MAX(started_at) AS last_full FROM source_ingest_run WHERE mode='full' GROUP BY source`,
  );
  const lastFullRescan = new Map<string, Date>();
  for (const r of fullRes.rows) if (r.last_full) lastFullRescan.set(r.source, new Date(r.last_full));

  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, lastFullRescanAt: lastFullRescan.get(l.source) ?? null, 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 only. fix 4: suppressSnapshot=true so we do NOT also fire the
    //    state-triggered ('new') snapshot — step 1 already wrote exactly one 'baseline' snapshot
    //    per listing. Previously this double-counted (599 listings → 1026 snapshots).
    const res = await recordLifecycle(query, l, ctx, { suppressSnapshot: true });
    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);
});