`src/server/listing-lifecycle.ts`

326 lines
/**
 * 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;
  /**
   * TK-10155 fix 1 (false-withdrawal guard): the start time of the source's most-recent FULL
   * rescan (MAX(started_at) over source_ingest_run WHERE mode='full'), or null if the source
   * has NEVER had a full rescan (e.g. an incremental-only source like ColdwellBanker's new-day
   * sitemap). Gone-from-feed → stale/withdrawn is ONLY derivable when a full rescan happened
   * AFTER the listing's last_seen — otherwise absence from the feed is structural, not a pull.
   * When null (or older than the listing's last_seen), gone-from-feed is suppressed and the
   * listing stays active/new. Status-based withdrawn/closed is unaffected. Undefined = the
   * legacy behavior (no guard), so old callers still compile.
   */
  lastFullRescanAt?: Date | null;
  /** "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;

  // TK-10155 fix 1: only allow gone-from-feed → stale/withdrawn if the source was FULLY
  // re-scanned AFTER this listing's last_seen. Incremental-only sources (no full rescan, or
  // the last full rescan predates last_seen) CANNOT prove the listing is gone — the feed just
  // never re-surfaces it — so we must NOT fabricate a withdrawal. `undefined` = legacy callers
  // with no guard (unchanged behavior); an explicit `null` = "no full rescan ever" (suppress).
  const eligibleForGoneFromFeed =
    ctx.lastFullRescanAt === undefined
      ? true // legacy path — no ingest-mode signal supplied
      : ctx.lastFullRescanAt != null && lastSeen != null && ctx.lastFullRescanAt >= lastSeen;

  if (eligibleForGoneFromFeed) {
    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;
}

/**
 * TK-10155 (Cody FIX-FIRST gate): guarantee EVERY listing has at least one preserved snapshot,
 * so preservation is a property of the DESIGN — not a side-effect of "nothing ever deletes a
 * listing." Without this, a listing whose first-see snapshot hook FAILED (caught+logged) and
 * that then derives straight to 'active' (e.g. first_seen already > NEW_DAYS) sits in the table
 * with NO snapshot at all; if a future prune job ever hard-deletes it, its data is lost forever
 * (the FK just SET-NULLs an event row that points at nothing). The sweep calls this for every
 * listing: if the listing has ZERO snapshots, capture one 'baseline'. Cheap NOT-EXISTS guard,
 * idempotent (no-op once any snapshot exists), append-only. Returns true iff it wrote one.
 */
export async function ensureBaselineSnapshot(
  db: QueryFn,
  listing: LifecycleListing,
): Promise<boolean> {
  if (listing.id == null) return false;
  const exists = await db<{ n: string }>(
    `SELECT count(*)::text n FROM listing_snapshot WHERE source = $1 AND source_id = $2`,
    [listing.source, listing.source_id],
  );
  if (Number(exists.rows[0]?.n ?? '0') > 0) return false; // already preserved — nothing to do
  const snap = await buildSnapshot(db, Number(listing.id));
  if (!snap) return false;
  await db(
    `INSERT INTO listing_snapshot
       (listing_id, source, source_id, address_key, snapshot, captured_reason)
     VALUES ($1,$2,$3,$4,$5,'baseline')`,
    [
      Number(listing.id),
      listing.source,
      listing.source_id,
      computeAddressKey(listing.address, null) || null,
      JSON.stringify(snap),
    ],
  );
  return true;
}

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 interface RecordLifecycleOpts {
  /**
   * TK-10155 fix 4: when true, still write the lifecycle EVENT but SKIP the state-triggered
   * snapshot. The backfill uses this because it already lays down one 'baseline' snapshot per
   * listing itself — without this, a listing deriving to 'new' would get a SECOND ('new')
   * snapshot, inflating the store (599 listings → 1026 snapshots pre-fix). Normal ingest/sweep
   * callers leave this false so real transitions (withdrawn/closed/stale) still snapshot.
   */
  suppressSnapshot?: boolean;
}

export async function recordLifecycle(
  db: QueryFn,
  listing: LifecycleListing,
  ctx: LifecycleCtx,
  opts: RecordLifecycleOpts = {},
): 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.
  // fix 4: opts.suppressSnapshot lets the backfill skip this (it writes its own baseline snapshot).
  if (!opts.suppressSnapshot && 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';
  }
}