`apps/workers/src/pipeline.itest.ts`

172 lines
import { beforeAll, describe, expect, it } from "vitest";
import { prisma } from "@homesonspec/database";
import { meridianHomesAdapter } from "@homesonspec/collector-meridian-homes";
import { publishStagedRecord } from "@homesonspec/publisher";
import { runPipeline, validateStage } from "./pipeline";

/**
 * Integration: the full fixture pipeline against homesonspec_test.
 * Proves raw ≠ published separation, verdict gating, idempotence,
 * community auto-creation from staged records, and review-queue flow.
 */

async function resetDb() {
  // Truncate in FK-safe order.
  const tables = [
    "ValidationEvent", "ReviewItem", "SourceEvidence", "StagedRecord", "RawSnapshot",
    "ChangeEvent", "CorrectionReport", "Lead", "Incentive", "InventoryHome", "Lot",
    "SalesOffice", "FloorPlan", "Community", "Division", "SourceRegistry", "Builder",
  ];
  for (const table of tables) {
    await prisma.$executeRawUnsafe(`TRUNCATE TABLE "${table}" CASCADE`);
  }
}

beforeAll(async () => {
  expect(process.env.DATABASE_URL).toMatch(/homesonspec_test/);
  await resetDb();
  // Minimal FK targets: builder + fixture source. NO community — the
  // pipeline must create it from the staged community record.
  const builder = await prisma.builder.create({
    data: { slug: "meridian-homes", name: "Meridian Homes", isDemo: true },
  });
  await prisma.sourceRegistry.create({
    data: {
      key: "meridian-homes-fixtures",
      name: "Meridian Homes (fixture pages)",
      builderId: builder.id,
      collectionMethod: "FIXTURE",
      mediaRights: "NONE",
    },
  });
});

describe("fixture pipeline end-to-end", () => {
  it("publishes community + homes + incentives from fixtures", async () => {
    const summary = await runPipeline(meridianHomesAdapter);
    expect(summary.pages).toBe(5);
    expect(summary.staged).toBe(6);
    expect(summary.published).toBe(6);
    expect(summary.rejected).toBe(0);
    expect(summary.errors).toEqual([]);

    // Community was CREATED by the publish stage from the staged record.
    const community = await prisma.community.findFirst({ where: { name: "Cedar Bend" } });
    expect(community).not.toBeNull();
    expect(community!.city).toBe("Leander");
    expect(community!.state).toBe("TX");
    expect(community!.isDemo).toBe(true);

    const homes = await prisma.inventoryHome.findMany({ where: { status: "PUBLISHED" } });
    expect(homes).toHaveLength(3);
    for (const home of homes) {
      expect(home.verificationLabel).toBe("DEMONSTRATION"); // FIXTURE method
      expect(home.isDemo).toBe(true);
    }

    // Null-price home stays null — never guessed.
    const silverleaf = homes.find((h) => h.street === "124 Silverleaf Pass");
    expect(silverleaf?.price).toBeNull();

    // Evidence re-pointed at published entities.
    const evidence = await prisma.sourceEvidence.count({
      where: { entityType: "INVENTORY_HOME", entityId: { not: null } },
    });
    expect(evidence).toBeGreaterThan(0);
  });

  it("is idempotent — unchanged bytes stage nothing new", async () => {
    const summary = await runPipeline(meridianHomesAdapter);
    expect(summary.changed).toBe(0);
    expect(summary.staged).toBe(0);
    expect(await prisma.inventoryHome.count()).toBe(3);
  });

  it("publish stage refuses non-validated records (only-writer invariant)", async () => {
    const source = await prisma.sourceRegistry.findUniqueOrThrow({
      where: { key: "meridian-homes-fixtures" },
    });
    const staged = await prisma.stagedRecord.create({
      data: {
        sourceId: source.id,
        entityType: "INVENTORY_HOME",
        canonicalKey: "not-validated-key",
        payload: {},
        hints: { builderSlug: "meridian-homes", communityName: "Cedar Bend" },
        extractorVersion: "test",
        status: "EXTRACTED",
      },
    });
    await expect(publishStagedRecord(staged.id)).rejects.toThrow(/refusing to publish/);
    expect(await prisma.inventoryHome.count()).toBe(3); // nothing leaked
  });

  it("big price change → needs_review + ReviewItem; approval publishes + ChangeEvent", async () => {
    const source = await prisma.sourceRegistry.findUniqueOrThrow({
      where: { key: "meridian-homes-fixtures" },
    });
    const juniper = await prisma.inventoryHome.findFirstOrThrow({
      where: { street: "118 Juniper Draw" },
    });
    const fv = (value: string | number | null, raw: string | null) => ({
      value, raw, evidenceText: raw, sourceUrl: "fixture://test", confidence: value === null ? 0 : 1,
    });
    // Same canonical identity, price dropped 20%.
    const newPrice = Math.round(Number(juniper.price) * 0.8);
    const staged = await prisma.stagedRecord.create({
      data: {
        sourceId: source.id,
        entityType: "INVENTORY_HOME",
        canonicalKey: juniper.canonicalKey,
        payload: {
          street: fv(juniper.street, juniper.street),
          city: fv("Leander", "Leander"),
          state: fv("TX", "TX"),
          zip: fv("78641", "78641"),
          price: fv(newPrice, `$${newPrice}`),
          beds: fv(4, "4 Beds"),
          bathsTotal: fv(3, "3 Baths"),
          sqft: fv(2415, "2,415 sq ft"),
          lat: fv(30.5791, "30.5791"),
          lon: fv(-97.8519, "-97.8519"),
        },
        hints: {
          builderSlug: "meridian-homes",
          communityName: "Cedar Bend",
          address: juniper.street,
          lotNumber: "118",
          builderInventoryId: "MH-1118",
          lat: 30.5791,
          lon: -97.8519,
          planName: "The Juniper",
        },
        extractorVersion: "test",
        status: "EXTRACTED",
      },
    });

    const { verdict } = await validateStage(staged.id, "meridian-homes-fixtures");
    expect(verdict).toBe("needs_review");
    const review = await prisma.reviewItem.findFirst({
      where: { stagedRecordId: staged.id, status: "OPEN" },
    });
    expect(review).not.toBeNull();
    expect(review!.reason).toMatch(/price.big-change/);

    // Price unchanged until a human approves.
    const before = await prisma.inventoryHome.findUniqueOrThrow({ where: { id: juniper.id } });
    expect(Number(before.price)).toBe(Number(juniper.price));

    // Human approval → publish updates the home + records a ChangeEvent.
    await prisma.stagedRecord.update({ where: { id: staged.id }, data: { status: "APPROVED" } });
    await publishStagedRecord(staged.id);
    const after = await prisma.inventoryHome.findUniqueOrThrow({ where: { id: juniper.id } });
    expect(Number(after.price)).toBe(newPrice);
    const change = await prisma.changeEvent.findFirst({
      where: { entityType: "INVENTORY_HOME", entityId: juniper.id, field: "price" },
    });
    expect(change).not.toBeNull();
    expect(change!.newValue).toBe(String(newPrice));
  });
});