`apps/workers/src/verify.ts`

255 lines
import { prisma, type Freshness, type VerificationLabel } from "@homesonspec/database";

/**
 * Stage-3 verification engine — the maintenance layer that runs OVER already
 * published inventory on a schedule. Ingestion (pipeline.ts) gets a home in the
 * door; this keeps the marketplace honest afterward:
 *   • freshness decay        — verified data ages FRESH → AGING → REVIEW → STALE
 *   • source-health auto-pause — repeated failures trip a source to PAUSED
 *   • cross-source dedup      — same address from two sources → keep the best
 *
 * The pure decision functions (computeFreshness, labelRank, pickDedupeWinner)
 * carry no DB and are unit-tested directly; the exported async ops wrap them.
 */

export const HOUR_MS = 3_600_000;

// Default cadence when a home has no registered source.
export const DEFAULT_INTERVALS = { freshHrs: 24, agingHrs: 72, staleHrs: 168 };

/** Pure: freshness from age vs a source's thresholds. Null verify time = must review. */
export function computeFreshness(
  lastVerifiedAt: Date | null,
  freshHrs: number,
  agingHrs: number,
  staleHrs: number,
  now: Date,
): Freshness {
  if (!lastVerifiedAt) return "REVIEW_REQUIRED";
  const ageHrs = (now.getTime() - lastVerifiedAt.getTime()) / HOUR_MS;
  if (ageHrs <= freshHrs) return "FRESH";
  if (ageHrs <= agingHrs) return "AGING";
  if (ageHrs <= staleHrs) return "REVIEW_REQUIRED";
  return "STALE";
}

// Authority ranking — higher wins a dedup tie. DEMONSTRATION never outranks real data.
const LABEL_RANK: Record<VerificationLabel, number> = {
  VERIFIED_FROM_BUILDER: 6,
  BUILDER_WEBSITE: 5,
  BUILDER_SUBMITTED: 4,
  NEEDS_REVERIFICATION: 3,
  NO_LONGER_AVAILABLE: 2,
  DEMONSTRATION: 1,
};
export const labelRank = (l: VerificationLabel) => LABEL_RANK[l] ?? 0;

export interface DedupeCandidate {
  id: string;
  verificationLabel: VerificationLabel;
  lastVerifiedAt: Date | null;
  createdAt: Date;
}

/** Pure: from a set of homes at the same address, choose the one record to keep. */
export function pickDedupeWinner<T extends DedupeCandidate>(group: T[]): T {
  return [...group].sort((a, b) => {
    const r = labelRank(b.verificationLabel) - labelRank(a.verificationLabel);
    if (r !== 0) return r;
    const av = a.lastVerifiedAt?.getTime() ?? 0;
    const bv = b.lastVerifiedAt?.getTime() ?? 0;
    if (bv !== av) return bv - av;
    return b.createdAt.getTime() - a.createdAt.getTime();
  })[0]!;
}

// ── Freshness recompute ────────────────────────────────────────────────────

// Labels that represent an active positive verification — these get demoted to
// NEEDS_REVERIFICATION once a home goes stale.
const REVERIFIABLE = new Set<VerificationLabel>([
  "VERIFIED_FROM_BUILDER",
  "BUILDER_WEBSITE",
  "BUILDER_SUBMITTED",
]);

export async function recomputeFreshness(now = new Date()) {
  const homes = await prisma.inventoryHome.findMany({
    where: { status: "PUBLISHED", isDemo: false },
    select: {
      id: true,
      lastVerifiedAt: true,
      freshness: true,
      verificationLabel: true,
      source: { select: { freshIntervalHours: true, agingIntervalHours: true, staleIntervalHours: true } },
    },
  });

  const byTarget: Record<Freshness, string[]> = {
    FRESH: [], AGING: [], REVIEW_REQUIRED: [], STALE: [], INACTIVE: [],
  };
  const relabel: string[] = [];

  for (const h of homes) {
    const fresh = h.source?.freshIntervalHours ?? DEFAULT_INTERVALS.freshHrs;
    const aging = h.source?.agingIntervalHours ?? DEFAULT_INTERVALS.agingHrs;
    const stale = h.source?.staleIntervalHours ?? DEFAULT_INTERVALS.staleHrs;
    const target = computeFreshness(h.lastVerifiedAt, fresh, aging, stale, now);
    if (target !== h.freshness) byTarget[target].push(h.id);
    if (target === "STALE" && REVERIFIABLE.has(h.verificationLabel)) relabel.push(h.id);
  }

  let changed = 0;
  for (const [target, ids] of Object.entries(byTarget)) {
    if (ids.length === 0) continue;
    await prisma.inventoryHome.updateMany({ where: { id: { in: ids } }, data: { freshness: target as Freshness } });
    changed += ids.length;
  }
  if (relabel.length > 0) {
    await prisma.inventoryHome.updateMany({
      where: { id: { in: relabel } },
      data: { verificationLabel: "NEEDS_REVERIFICATION" },
    });
  }

  return {
    scanned: homes.length,
    changed,
    relabeled: relabel.length,
    toFresh: byTarget.FRESH.length,
    toAging: byTarget.AGING.length,
    toReview: byTarget.REVIEW_REQUIRED.length,
    toStale: byTarget.STALE.length,
  };
}

// ── Source-health auto-pause ────────────────────────────────────────────────

export interface RunOutcome {
  ok: boolean;
  kind?: string;
  pages?: number;
  changed?: number;
  published?: number;
  needsReview?: number;
  rejected?: number;
  errorCount?: number;
  message?: string;
  durationMs?: number;
}

/**
 * Log a run and roll the source's health forward. A success resets the failure
 * streak (and recovers a paused/degraded source); a failure increments it and,
 * once it reaches maxConsecutiveFailures, trips the source to PAUSED + inactive
 * so runPipeline (which checks `active`) refuses to run it until a human clears it.
 */
export async function recordSourceRun(sourceKey: string, outcome: RunOutcome, now = new Date()) {
  const source = await prisma.sourceRegistry.findUniqueOrThrow({ where: { key: sourceKey } });
  await prisma.sourceRun.create({
    data: {
      sourceId: source.id,
      kind: outcome.kind ?? "pipeline",
      ok: outcome.ok,
      pages: outcome.pages ?? 0,
      changed: outcome.changed ?? 0,
      published: outcome.published ?? 0,
      needsReview: outcome.needsReview ?? 0,
      rejected: outcome.rejected ?? 0,
      errorCount: outcome.errorCount ?? 0,
      message: outcome.message ?? null,
      durationMs: outcome.durationMs ?? null,
      startedAt: now,
    },
  });

  const failures = outcome.ok ? 0 : source.consecutiveFailures + 1;
  const tripped = failures >= source.maxConsecutiveFailures;
  const health = outcome.ok
    ? outcome.errorCount && outcome.errorCount > 0
      ? "DEGRADED"
      : "HEALTHY"
    : tripped
      ? "PAUSED"
      : "FAILING";

  await prisma.sourceRegistry.update({
    where: { id: source.id },
    data: {
      lastRunAt: now,
      lastSuccessAt: outcome.ok ? now : source.lastSuccessAt,
      consecutiveFailures: failures,
      health,
      // Auto-pause disables the source. Recovery is deliberate: only a source that
      // was AUTO-paused (health === PAUSED) re-enables on a clean run. A source an
      // operator manually pulled (active=false while health != PAUSED) stays down —
      // a stray successful run must not silently resurrect it.
      active: outcome.ok ? (source.health === "PAUSED" ? true : source.active) : tripped ? false : source.active,
    },
  });

  return { key: sourceKey, health, consecutiveFailures: failures, autoPaused: !outcome.ok && tripped };
}

// ── Refresh scheduling ──────────────────────────────────────────────────────

/** Sources due for a refresh: active, not paused, and past their fresh interval. */
export async function refreshDue(now = new Date()) {
  const sources = await prisma.sourceRegistry.findMany({
    where: { active: true, health: { not: "PAUSED" } },
    select: { key: true, name: true, lastRunAt: true, freshIntervalHours: true },
  });
  return sources.filter((s) => {
    if (!s.lastRunAt) return true;
    const ageHrs = (now.getTime() - s.lastRunAt.getTime()) / HOUR_MS;
    return ageHrs >= s.freshIntervalHours;
  });
}

// ── Cross-source dedup ──────────────────────────────────────────────────────

/**
 * Find published non-demo homes that share a normalized address + zip (the same
 * physical home surfaced by two sources / crawls). Keep the most authoritative
 * record; deactivate the rest. Dry-run by default — pass { apply: true } to write.
 */
export async function dedupePass(opts: { apply?: boolean } = {}) {
  const homes = await prisma.inventoryHome.findMany({
    where: { status: "PUBLISHED", isDemo: false, normalizedAddress: { not: null } },
    select: {
      id: true,
      normalizedAddress: true,
      zip: true,
      verificationLabel: true,
      lastVerifiedAt: true,
      createdAt: true,
    },
  });

  const groups = new Map<string, typeof homes>();
  for (const h of homes) {
    const key = `${h.normalizedAddress}|${h.zip}`;
    const arr = groups.get(key) ?? [];
    arr.push(h);
    groups.set(key, arr);
  }

  const losers: string[] = [];
  let dupGroups = 0;
  for (const group of groups.values()) {
    if (group.length < 2) continue;
    dupGroups += 1;
    const winner = pickDedupeWinner(group);
    for (const h of group) if (h.id !== winner.id) losers.push(h.id);
  }

  if (opts.apply && losers.length > 0) {
    await prisma.inventoryHome.updateMany({
      where: { id: { in: losers } },
      data: { status: "INACTIVE", freshness: "INACTIVE" },
    });
  }

  return { scanned: homes.length, duplicateGroups: dupGroups, duplicates: losers.length, applied: !!opts.apply };
}