`src/ingest/listings/engine.ts`

221 lines
/**
 * M-B3 firm↔listing pilot ingest engine.
 *
 * Per site: openRun → robots gate → discover ≤CAP listing URLs from the public sitemap →
 * for each: robots-recheck → politeFetch (UA + 2–4s jitter) → parse FACTS ONLY from
 * JSON-LD → resolve firm_id by source host → resolve county region_id by centroid →
 * upsert into `listing` (UNIQUE(source, source_id)) → closeRun.
 *
 * Fail-loud: 0 parsed listings = failed run + non-zero exit. A robots-disallowed listings
 * path = the whole site is SKIPPED and recorded as a failed run with an explicit note.
 *
 *   npm run ingest:listings -- coldwellbanker
 *   npm run ingest:listings -- realtytexas
 *   npm run ingest:listings -- all
 */
import { pool, query } from '../../../db/pool.ts';
import { openRun, closeRun } from '../run.ts';
import type { ListingAdapter } from './types.ts';
import { coldwellbanker } from './coldwellbanker.ts';
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)

async function upsertListing(
  source: string,
  f: import('./types.ts').ListingFacts,
  firmId: number | null,
  firmName: string | null,
  regionId: number | null,
): Promise<void> {
  // 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())
     ON CONFLICT (source, source_id) DO UPDATE SET
       firm_id  = COALESCE(EXCLUDED.firm_id, listing.firm_id),
       region_id= COALESCE(EXCLUDED.region_id, listing.region_id),
       address  = EXCLUDED.address,
       price    = EXCLUDED.price,
       beds     = COALESCE(EXCLUDED.beds, listing.beds),
       baths    = COALESCE(EXCLUDED.baths, listing.baths),
       sqft     = COALESCE(EXCLUDED.sqft, listing.sqft),
       lat      = COALESCE(EXCLUDED.lat, listing.lat),
       lng      = COALESCE(EXCLUDED.lng, listing.lng),
       url      = EXCLUDED.url,
       status   = EXCLUDED.status,
       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],
  );

  // TK-10139: record every broker/firm-of-record observation for this address (append-only,
  // no-ops when unchanged). Non-fatal — a recorder failure must never break listing ingest.
  try {
    // NOTE: the `listing` table stores the full "street, city, state, zip" in `address`
    // (no separate city column), so we key on the address alone (city: null) to stay
    // consistent with db/backfill-broker-of-record.ts and avoid a spurious re-observation.
    await recordBrokerOfRecord(query, {
      address: f.address,
      city: null,
      state_code: f.state,
      firm_id: firmId,
      firm_name: firmName,
      source,
      source_url: f.url,
    });
  } 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) {
    // Loud + identifiable (TK-10155 Cycle-3): a swallowed first-see failure is why an ingest's
    // listing_count can exceed the new-event count by 1+. Not fatal — the next sweep re-derives
    // state for every listing and writes the missing 'new' event + snapshot as its safety net.
    console.error(`[${source}] lifecycle first-see record FAILED for source_id=${f.sourceId} (sweep will recover): ${e?.message || e}`);
  }
}

async function runAdapter(a: ListingAdapter): Promise<{ upserted: number; skipped: number }> {
  const runId = await openRun(a.source, `https://${a.host}`);
  console.log(`[${a.source}] robots gate…`);
  try {
    const allowed = await robotsGate(a.host);
    if (!allowed(a.listingsPathHint)) {
      const note = `robots.txt disallows listings path ${a.listingsPathHint} — SKIPPED`;
      console.log(`[${a.source}] ${note}`);
      await closeRun(runId, 'failed', { upserted: 0, skipped: 0, notes: note });
      return { upserted: 0, skipped: 0 };
    }

    const firmId = await resolveFirmByHost(a.host);
    console.log(`[${a.source}] firm_id resolved: ${firmId ?? 'NONE'}`);
    // firm name for the broker-of-record history (resolved once per site)
    let firmName: string | null = null;
    if (firmId != null) {
      const fr = await query<{ name: string }>(`SELECT name FROM firm WHERE id=$1`, [firmId]);
      firmName = fr.rows[0]?.name ?? null;
    }

    const urls = await a.discover(CAP);
    console.log(`[${a.source}] discovered ${urls.length} listing URLs (cap ${CAP})`);

    let upserted = 0;
    let skipped = 0;
    for (const url of urls) {
      const path = new URL(url).pathname;
      if (!allowed(path)) { skipped++; continue; }
      const { ok, status, body } = await politeFetch(url);
      if (!ok) { console.log(`[${a.source}] fetch ${status} ${url}`); skipped++; continue; }
      const facts = a.parse(url, body);
      if (!facts) { skipped++; continue; }
      const regionId = (await resolveRegion(facts.state, facts.lat, facts.lng))
        ?? (await resolveRegionByZip(facts.zip ?? null));
      await upsertListing(a.source, facts, firmId, firmName, regionId);
      upserted++;
      if (upserted % 5 === 0) console.log(`[${a.source}] upserted ${upserted}…`);
    }

    if (upserted === 0) {
      const note = `0 listings parsed from ${urls.length} URLs — structure drift or bot-block`;
      console.error(`[${a.source}] FAIL: ${note}`);
      await closeRun(runId, 'failed', { upserted: 0, skipped, notes: note });
      return { upserted: 0, skipped };
    }
    await closeRun(runId, 'ok', { upserted, skipped, notes: `firm_id=${firmId}` });

    // TK-10155 fix 1: record this run's INGEST MODE so the lifecycle detector knows whether the
    // source was FULLY re-scanned. A 'full' run here is what lets classifyState later treat a
    // gone-from-feed listing as withdrawn; an 'incremental' run does NOT establish absence.
    // TK-10155 Cycle-2 fix (Cody gate): a 'full' adapter that only covered a FRACTION of its known
    // listings (flaky/truncated rescan, e.g. RealtyTexas 3/46) must NOT stamp a real 'full' marker —
    // that would falsely arm gone-from-feed withdrawal for the un-refetched listings and fabricate
    // withdrawals ~WITHDRAWN_DAYS later. Downgrade to 'partial' (which does NOT arm withdrawal)
    // unless coverage >= FULL_RESCAN_MIN_COVERAGE of the source's known listings.
    // Non-fatal — a ledger failure must never fail the ingest.
    const FULL_RESCAN_MIN_COVERAGE = 0.5;
    let recordedMode: string = a.mode;
    try {
      if (a.mode === 'full') {
        const { rows } = await query<{ n: string }>(
          `SELECT count(*)::text AS n FROM listing WHERE source=$1`,
          [a.source],
        );
        const known = Number(rows?.[0]?.n ?? 0);
        if (known > 0 && upserted < known * FULL_RESCAN_MIN_COVERAGE) {
          recordedMode = 'partial';
          console.warn(
            `[${a.source}] full rescan covered only ${upserted}/${known} (<${FULL_RESCAN_MIN_COVERAGE * 100}%) — recording mode='partial' (withdrawal detection stays disarmed for this source this run)`,
          );
        }
      }
      await query(
        `INSERT INTO source_ingest_run (source, mode, finished_at, listing_count)
         VALUES ($1,$2, now(), $3)`,
        [a.source, recordedMode, upserted],
      );
    } catch (e: any) {
      console.warn(`[${a.source}] source_ingest_run record skipped: ${e?.message || e}`);
    }

    console.log(`[${a.source}] DONE upserted=${upserted} skipped=${skipped} firm_id=${firmId} mode=${recordedMode}`);
    return { upserted, skipped };
  } catch (e: any) {
    await closeRun(runId, 'failed', { notes: String(e.message || e) });
    throw e;
  }
}

async function main() {
  const arg = (process.argv[2] || '').toLowerCase();
  const targets = arg === 'all' || !arg ? ADAPTERS : ADAPTERS.filter((a) => a.source === arg);
  if (!targets.length) {
    console.error(`unknown site '${arg}'. known: ${ADAPTERS.map((a) => a.source).join(', ')}, all`);
    process.exit(2);
  }
  let total = 0;
  for (const a of targets) {
    const r = await runAdapter(a);
    total += r.upserted;
  }
  console.log(`\n[listings] total upserted across ${targets.length} site(s): ${total}`);
  await pool.end();
  if (total === 0) process.exit(1);
}

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