← back to Homesonspec

apps/workers/src/pipeline.ts

345 lines

import { mkdir, writeFile } from "node:fs/promises";
import { join } from "node:path";
import {
  prisma,
  type EntityType,
  type Prisma,
  type SourceRegistry,
} from "@homesonspec/database";
import { validatePayload, type ExtractedRecord } from "@homesonspec/schemas";
import { canonicalKey } from "@homesonspec/shared";
import {
  ALL_RULES,
  runValidation,
  VALIDATOR_VERSION,
  type RuleDb,
  type StagedCandidate,
} from "@homesonspec/validation";
import type { RawPage, SourceAdapter } from "@homesonspec/collectors-common";
import { publishStagedRecord } from "@homesonspec/publisher";
import { recordSourceRun } from "./verify";

export { publishStagedRecord };

/**
 * Pipeline stages. Each is a pure-ish function over (adapter, db) — the
 * queue merely transports invocations; the CLI and tests call these
 * directly. Raw and published tables stay strictly separate: nothing here
 * writes a published table except publishStagedRecord, and that only after
 * a deterministic `publishable` verdict (or explicit human approval).
 */

// Anchor to the repo root — the CLI may run with cwd anywhere in the workspace.
const REPO_ROOT = join(import.meta.dirname, "../../..");
const SNAPSHOT_DIR = process.env.SNAPSHOT_DIR ?? join(REPO_ROOT, "var/snapshots");

const ENTITY_TYPE_MAP: Record<ExtractedRecord["entityType"], EntityType> = {
  community: "COMMUNITY",
  floor_plan: "FLOOR_PLAN",
  inventory_home: "INVENTORY_HOME",
  incentive: "INCENTIVE",
  sales_office: "SALES_OFFICE",
  lot: "LOT",
};

// ── Stage 1: fetch → raw snapshots ─────────────────────────────────────

export async function fetchStage(adapter: SourceAdapter, source: SourceRegistry) {
  const pages: { page: RawPage; snapshotId: string; changed: boolean }[] = [];
  const dir = join(SNAPSHOT_DIR, source.key);
  await mkdir(dir, { recursive: true });

  const isFixture = source.collectionMethod === "FIXTURE" || source.collectionMethod === "SYNTHETIC";
  for await (const page of adapter.fetch({
    mode: isFixture ? "fixture" : "live",
    fixtureDir: join(REPO_ROOT, "fixtures/raw-pages", adapter.key.replace(/-fixtures$/, "")),
    registry: {
      key: source.key,
      collectionMethod: source.collectionMethod,
      rateLimitRpm: source.rateLimitRpm,
      mediaRights: source.mediaRights,
    },
  })) {
    const existing = await prisma.rawSnapshot.findUnique({
      where: { sourceId_contentHash: { sourceId: source.id, contentHash: page.contentHash } },
    });
    if (existing) {
      // Same bytes already snapshotted — source unchanged for this url.
      pages.push({ page, snapshotId: existing.id, changed: false });
      continue;
    }
    const storagePath = join(dir, `${page.contentHash}${page.contentType.includes("json") ? ".json" : ".html"}`);
    await writeFile(storagePath, page.body);
    const snapshot = await prisma.rawSnapshot.create({
      data: {
        sourceId: source.id,
        url: page.url,
        retrievedAt: new Date(page.retrievedAt),
        contentHash: page.contentHash,
        storagePath,
        contentType: page.contentType,
        httpStatus: 200,
      },
    });
    pages.push({ page, snapshotId: snapshot.id, changed: true });
  }
  return pages;
}

// ── Stage 2: extract → staged records + evidence ───────────────────────

export async function extractStage(
  adapter: SourceAdapter,
  source: SourceRegistry,
  page: RawPage,
  snapshotId: string,
) {
  const { records, errors } = adapter.extract(page);
  const stagedIds: string[] = [];

  for (const record of records) {
    const normalized = adapter.normalize ? await adapter.normalize(record) : record;

    const shape = validatePayload(normalized);
    if (!shape.ok) {
      // Shape failure is an extraction bug, not bad data — log, don't stage.
      errors.push({ url: page.url, reason: `payload schema: ${shape.issues.join("; ")}` });
      continue;
    }

    const key = canonicalKey(normalized.canonicalHints);
    const entityType = ENTITY_TYPE_MAP[normalized.entityType];
    const payloadData = {
      entityType,
      payload: normalized.fields as unknown as Prisma.InputJsonValue,
      hints: normalized.canonicalHints as unknown as Prisma.InputJsonValue,
      extractorVersion: adapter.version,
      status: "EXTRACTED" as const,
    };
    // Upsert on (source, snapshot, home) so re-extracting the same snapshot (e.g.
    // FORCE_REEXTRACT to backfill a new field) updates the row instead of piling
    // up duplicate staged/evidence rows.
    const staged = await prisma.stagedRecord.upsert({
      where: { sourceId_snapshotId_canonicalKey: { sourceId: source.id, snapshotId, canonicalKey: key } },
      create: { sourceId: source.id, snapshotId, canonicalKey: key, ...payloadData },
      update: payloadData,
    });
    stagedIds.push(staged.id);

    // Replace this row's evidence rather than append (idempotent re-extract). Wrapped in a
    // transaction so a concurrent re-extract of the same source (e.g. a FORCE_REEXTRACT backfill
    // running alongside a normal loop sweep) can't interleave delete/create and DOUBLE the evidence
    // rows — the delete+create is atomic. [yolo iter-5 contrarian fix]
    await prisma.$transaction([
      prisma.sourceEvidence.deleteMany({ where: { stagedRecordId: staged.id } }),
      prisma.sourceEvidence.createMany({
        data: Object.entries(normalized.fields).map(([field, envelope]) => ({
          stagedRecordId: staged.id,
          field,
          sourceUrl: envelope.sourceUrl,
          retrievedAt: new Date(page.retrievedAt),
          contentHash: page.contentHash,
          extractorVersion: adapter.version,
          rawValue: envelope.raw,
          normalizedValue: envelope.value === null ? null : String(envelope.value),
          evidenceText: envelope.evidenceText,
          confidence: envelope.confidence,
        })),
      }),
    ]);
  }
  return { stagedIds, errors };
}

// ── Stage 3: validate → ValidationEvents + verdict ─────────────────────

const ruleDb: RuleDb = {
  async activeHomeExistsAtAddress(normalizedAddress, zip, excludeCanonicalKey) {
    const found = await prisma.inventoryHome.findFirst({
      where: {
        normalizedAddress,
        ...(zip ? { zip } : {}),
        status: "PUBLISHED",
        canonicalKey: { not: excludeCanonicalKey },
      },
      select: { id: true },
    });
    return found !== null;
  },
  async publishedPriceForCanonicalKey(key) {
    const home = await prisma.inventoryHome.findUnique({
      where: { canonicalKey: key },
      select: { price: true },
    });
    return home?.price ? Number(home.price) : null;
  },
  async communityExists(name, builderSlug) {
    const found = await prisma.community.findFirst({
      where: { name, builder: { slug: builderSlug } },
      select: { id: true },
    });
    return found !== null;
  },
  async builderExists(slug) {
    const found = await prisma.builder.findUnique({ where: { slug }, select: { id: true } });
    return found !== null;
  },
};

const REVERSE_ENTITY_MAP: Record<string, ExtractedRecord["entityType"]> = {
  COMMUNITY: "community",
  FLOOR_PLAN: "floor_plan",
  INVENTORY_HOME: "inventory_home",
  INCENTIVE: "incentive",
  SALES_OFFICE: "sales_office",
  LOT: "lot",
};

export async function validateStage(stagedRecordId: string, sourceKey: string) {
  const staged = await prisma.stagedRecord.findUniqueOrThrow({ where: { id: stagedRecordId } });

  const candidate: StagedCandidate = {
    entityType: REVERSE_ENTITY_MAP[staged.entityType]!,
    canonicalKey: staged.canonicalKey,
    canonicalHints: (staged.hints ?? { builderSlug: "" }) as StagedCandidate["canonicalHints"],
    fields: staged.payload as StagedCandidate["fields"],
  };

  const { events, verdict } = await runValidation(candidate, ALL_RULES, {
    db: ruleDb,
    sourceKey,
    now: new Date(),
  });

  // Idempotent re-validation: drop this row's prior events + open review items first.
  await prisma.validationEvent.deleteMany({ where: { stagedRecordId: staged.id } });
  await prisma.reviewItem.deleteMany({ where: { stagedRecordId: staged.id } });
  await prisma.validationEvent.createMany({
    data: events.map((event) => ({
      stagedRecordId: staged.id,
      ruleId: event.ruleId,
      severity: event.severity,
      passed: event.passed,
      message: event.message,
      details: (event.details ?? undefined) as Prisma.InputJsonValue | undefined,
      validatorVersion: VALIDATOR_VERSION,
    })),
  });

  const status =
    verdict === "publishable" ? "VALIDATED" : verdict === "needs_review" ? "NEEDS_REVIEW" : "REJECTED";
  await prisma.stagedRecord.update({ where: { id: staged.id }, data: { status } });

  if (verdict === "needs_review") {
    const failing = events.filter((e) => !e.passed && e.severity === "review");
    await prisma.reviewItem.create({
      data: {
        stagedRecordId: staged.id,
        reason: failing.map((e) => `${e.ruleId}: ${e.message}`).join(" | ") || "review required",
      },
    });
  }
  return { verdict, status };
}

// ── Orchestration: run every stage for one adapter ─────────────────────

export async function runPipeline(adapter: SourceAdapter) {
  const source = await prisma.sourceRegistry.findUniqueOrThrow({ where: { key: adapter.key } });
  if (!source.active) throw new Error(`source ${source.key} is inactive — refusing to run`);

  const startedAt = Date.now();
  const summary = {
    pages: 0,
    changed: 0,
    staged: 0,
    published: 0,
    needsReview: 0,
    rejected: 0,
    errors: [] as { url: string; reason: string }[],
  };

  // STREAMING: each page is snapshotted, extracted, validated, and published
  // before the next fetch — memory stays flat across multi-thousand-page
  // sweeps, and a mid-run failure loses nothing already processed.
  const dir = join(SNAPSHOT_DIR, source.key);
  await mkdir(dir, { recursive: true });
  const isFixture = source.collectionMethod === "FIXTURE" || source.collectionMethod === "SYNTHETIC";

  for await (const page of adapter.fetch({
    mode: isFixture ? "fixture" : "live",
    fixtureDir: join(REPO_ROOT, "fixtures/raw-pages", adapter.key.replace(/-fixtures$/, "")),
    registry: {
      key: source.key,
      collectionMethod: source.collectionMethod,
      rateLimitRpm: source.rateLimitRpm,
      mediaRights: source.mediaRights,
    },
  })) {
    summary.pages += 1;
    const existing = await prisma.rawSnapshot.findUnique({
      where: { sourceId_contentHash: { sourceId: source.id, contentHash: page.contentHash } },
    });
    // FORCE_REEXTRACT=1 re-extracts an already-snapshotted page against its
    // existing snapshot — used to backfill newly-added fields (e.g. images) onto
    // homes whose source bytes haven't changed. Default: skip unchanged pages.
    const force = process.env.FORCE_REEXTRACT === "1";
    if (existing && !force) continue; // unchanged bytes → nothing new to extract
    summary.changed += 1;

    let snapshotId: string;
    if (existing) {
      snapshotId = existing.id;
    } else {
      const storagePath = join(dir, `${page.contentHash}${page.contentType.includes("json") ? ".json" : ".html"}`);
      await writeFile(storagePath, page.body);
      const snapshot = await prisma.rawSnapshot.create({
        data: {
          sourceId: source.id,
          url: page.url,
          retrievedAt: new Date(page.retrievedAt),
          contentHash: page.contentHash,
          storagePath,
          contentType: page.contentType,
          httpStatus: 200,
        },
      });
      snapshotId = snapshot.id;
    }

    const { stagedIds, errors } = await extractStage(adapter, source, page, snapshotId);
    summary.errors.push(...errors);
    summary.staged += stagedIds.length;

    for (const stagedId of stagedIds) {
      const { verdict } = await validateStage(stagedId, source.key);
      if (verdict === "publishable") {
        await publishStagedRecord(stagedId);
        summary.published += 1;
      } else if (verdict === "needs_review") {
        summary.needsReview += 1;
      } else {
        summary.rejected += 1;
      }
    }
  }

  // Log the run + roll source health forward (Stage-3). A clean run resets any
  // failure streak; errors mark the source DEGRADED. Outright throws are recorded
  // as failures by the caller (verify-cli / pipeline runner), which is what trips
  // the auto-pause.
  await recordSourceRun(source.key, {
    ok: true,
    kind: "pipeline",
    pages: summary.pages,
    changed: summary.changed,
    published: summary.published,
    needsReview: summary.needsReview,
    rejected: summary.rejected,
    errorCount: summary.errors.length,
    durationMs: Date.now() - startedAt,
  });

  return summary;
}