`src/jobs/listing-lifecycle-sweep.ts`

107 lines
/**
 * 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,
  ensureBaselineSnapshot,
  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));
  }

  // TK-10155 fix 1: per-source LAST FULL RESCAN time (max started_at over mode='full' runs).
  // A source with no full run at all is absent from this map → lastFullRescanAt=null → its
  // listings are NEVER aged into gone-from-feed withdrawal (incremental-only sources).
  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 snapshots = 0;
  let baselinesBackfilled = 0; // TK-10155 Cody FIX-FIRST: snapshots created for listings that had none
  let skippedNoRef = 0;
  // # listings on incremental-only sources (no full rescan) that are structurally NOT eligible
  // for gone-from-feed withdrawal — surfaced in the sweep output for transparency.
  let incrementalNotEligible = 0;

  for (const l of rows.rows) {
    const ref = sourceMax.get(l.source);
    if (!ref) { skippedNoRef++; continue; }
    // null when the source has never had a full rescan (incremental-only) → gone-from-feed
    // withdrawal is suppressed for this listing.
    const lastFullRescanAt = lastFullRescan.get(l.source) ?? null;
    if (lastFullRescanAt === null) incrementalNotEligible++;
    const ctx = { sourceMaxLastSeen: ref, lastFullRescanAt, 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++;

    // TK-10155 Cody FIX-FIRST gate: guarantee EVERY listing has >=1 preserved snapshot, so the
    // "data survives after the listing vanishes" promise is a design property, not a side-effect of
    // "nothing ever deletes a listing." Closes the gap where a listing whose first-see snapshot
    // FAILED and that derives straight to 'active' (not a snapshot state) would sit with NO snapshot.
    // No-op once any snapshot exists — so this only writes for the leaked cases, not every sweep.
    if (await ensureBaselineSnapshot(query, l)) baselinesBackfilled++;
  }

  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}`);
  console.log(`[lifecycle-sweep] baseline snapshots backfilled (had none): ${baselinesBackfilled}`);
  if (skippedNoRef) console.log(`[lifecycle-sweep] skipped (no source ref): ${skippedNoRef}`);
  // TK-10155 fix 1 transparency line.
  console.log(
    `[lifecycle-sweep] ${incrementalNotEligible} listings on incremental-only sources are NOT ` +
    `eligible for gone-from-feed withdrawal (needs a full rescan).`,
  );

  await pool.end();
}

main().catch((e) => {
  console.error(e);
  process.exit(1);
});