← back to Homesonspec
generalize pipeline: hints column on StagedRecord, publisher resolves builder/community from hints, label derives from collection method, live/fixture mode from registry
73a5db3eeddffe8988164fc87fe81c3666ec6a24 · 2026-07-22 11:39:54 -0700 · Steve Abrams
Files touched
M apps/workers/src/pipeline.tsA packages/database/prisma/migrations/20260722183952_staged_hints/migration.sqlM packages/database/prisma/schema.prismaM packages/publisher/src/index.ts
Diff
commit 73a5db3eeddffe8988164fc87fe81c3666ec6a24
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Wed Jul 22 11:39:54 2026 -0700
generalize pipeline: hints column on StagedRecord, publisher resolves builder/community from hints, label derives from collection method, live/fixture mode from registry
---
apps/workers/src/pipeline.ts | 30 +---
.../20260722183952_staged_hints/migration.sql | 5 +
packages/database/prisma/schema.prisma | 1 +
packages/publisher/src/index.ts | 190 +++++++++++++++++----
4 files changed, 168 insertions(+), 58 deletions(-)
diff --git a/apps/workers/src/pipeline.ts b/apps/workers/src/pipeline.ts
index a34c5f4..cae9862 100644
--- a/apps/workers/src/pipeline.ts
+++ b/apps/workers/src/pipeline.ts
@@ -48,8 +48,9 @@ export async function fetchStage(adapter: SourceAdapter, source: SourceRegistry)
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: "fixture",
+ mode: isFixture ? "fixture" : "live",
fixtureDir: join(REPO_ROOT, "fixtures/raw-pages", adapter.key.replace(/-fixtures$/, "")),
registry: {
key: source.key,
@@ -66,7 +67,7 @@ export async function fetchStage(adapter: SourceAdapter, source: SourceRegistry)
pages.push({ page, snapshotId: existing.id, changed: false });
continue;
}
- const storagePath = join(dir, `${page.contentHash}.html`);
+ const storagePath = join(dir, `${page.contentHash}${page.contentType.includes("json") ? ".json" : ".html"}`);
await writeFile(storagePath, page.body);
const snapshot = await prisma.rawSnapshot.create({
data: {
@@ -113,6 +114,7 @@ export async function extractStage(
entityType: ENTITY_TYPE_MAP[normalized.entityType],
canonicalKey: key,
payload: normalized.fields as unknown as Prisma.InputJsonValue,
+ hints: normalized.canonicalHints as unknown as Prisma.InputJsonValue,
extractorVersion: adapter.version,
status: "EXTRACTED",
},
@@ -187,7 +189,7 @@ export async function validateStage(stagedRecordId: string, sourceKey: string) {
const candidate: StagedCandidate = {
entityType: REVERSE_ENTITY_MAP[staged.entityType]!,
canonicalKey: staged.canonicalKey,
- canonicalHints: reconstructHints(staged.payload as Record<string, { value: unknown }>, sourceKey),
+ canonicalHints: (staged.hints ?? { builderSlug: "" }) as StagedCandidate["canonicalHints"],
fields: staged.payload as StagedCandidate["fields"],
};
@@ -225,28 +227,6 @@ export async function validateStage(stagedRecordId: string, sourceKey: string) {
return { verdict, status };
}
-/** Hints were computed at extract time from the record itself; rebuild them from the payload. */
-function reconstructHints(
- payload: Record<string, { value: unknown }>,
- _sourceKey: string,
-): StagedCandidate["canonicalHints"] {
- const get = (field: string) => (payload[field]?.value ?? null) as string | null;
- const num = (field: string) => (payload[field]?.value ?? null) as number | null;
- return {
- builderSlug: "meridian-homes", // v1: single fixture builder; live adapters carry this in payload
- communityName: get("communityName") ?? "Cedar Bend",
- address: get("street"),
- lotNumber: get("lotNumber"),
- builderInventoryId: get("builderInventoryId"),
- lat: num("lat"),
- lon: num("lon"),
- planName: get("planName"),
- };
-}
-
-// ── Stage 4: publish lives in @spechomes/publisher (the ONLY writer
-// of published tables) — re-exported here for pipeline callers. ────
-
// ── Orchestration: run every stage for one adapter ─────────────────────
export async function runPipeline(adapter: SourceAdapter) {
diff --git a/packages/database/prisma/migrations/20260722183952_staged_hints/migration.sql b/packages/database/prisma/migrations/20260722183952_staged_hints/migration.sql
new file mode 100644
index 0000000..57b6895
--- /dev/null
+++ b/packages/database/prisma/migrations/20260722183952_staged_hints/migration.sql
@@ -0,0 +1,5 @@
+-- DropIndex
+DROP INDEX "Community_searchText_trgm_idx";
+
+-- AlterTable
+ALTER TABLE "StagedRecord" ADD COLUMN "hints" JSONB;
diff --git a/packages/database/prisma/schema.prisma b/packages/database/prisma/schema.prisma
index d4f50eb..0e8f506 100644
--- a/packages/database/prisma/schema.prisma
+++ b/packages/database/prisma/schema.prisma
@@ -418,6 +418,7 @@ model StagedRecord {
entityType EntityType
canonicalKey String
payload Json // per-field FieldValue envelope {value,raw,evidenceText,sourceUrl,confidence}
+ hints Json? // CanonicalHints captured at extract time (builderSlug, communityName, address…)
extractorVersion String
status StagedStatus @default(EXTRACTED)
publishedEntityId String?
diff --git a/packages/publisher/src/index.ts b/packages/publisher/src/index.ts
index 4488494..95d70b7 100644
--- a/packages/publisher/src/index.ts
+++ b/packages/publisher/src/index.ts
@@ -1,4 +1,4 @@
-import { prisma } from "@spechomes/database";
+import { prisma, type CollectionMethod, type VerificationLabel } from "@spechomes/database";
import { normalizeAddress } from "@spechomes/shared";
/**
@@ -8,7 +8,43 @@ import { normalizeAddress } from "@spechomes/shared";
* validators passed) or APPROVED (human review). Any other status throws.
* Collectors, extractors, and normalizers never touch published tables;
* they end at StagedRecord, and this function is the single bridge.
+ *
+ * Builder + community resolution comes from the hints captured at extract
+ * time. Community records publish (upsert) real Community rows, so an
+ * adapter's discovery order — communities before homes — makes the FK
+ * integrity rule pass naturally.
*/
+
+interface Hints {
+ builderSlug: string;
+ communityName?: string | null;
+ address?: string | null;
+ lotNumber?: string | null;
+ builderInventoryId?: string | null;
+ lat?: number | null;
+ lon?: number | null;
+ planName?: string | null;
+}
+
+/** Verification label + demo flag derive from HOW the data was collected. */
+export function labelForMethod(method: CollectionMethod): { label: VerificationLabel; isDemo: boolean } {
+ switch (method) {
+ case "FEED":
+ return { label: "VERIFIED_FROM_BUILDER", isDemo: false };
+ case "CRAWL":
+ return { label: "BUILDER_WEBSITE", isDemo: false };
+ case "MANUAL":
+ return { label: "BUILDER_SUBMITTED", isDemo: false };
+ case "FIXTURE":
+ case "SYNTHETIC":
+ return { label: "DEMONSTRATION", isDemo: true };
+ }
+}
+
+function slugify(text: string): string {
+ return text.toLowerCase().replace(/[^a-z0-9]+/g, "-").replace(/^-|-$/g, "");
+}
+
export async function publishStagedRecord(stagedRecordId: string) {
const staged = await prisma.stagedRecord.findUniqueOrThrow({
where: { id: stagedRecordId },
@@ -21,15 +57,63 @@ export async function publishStagedRecord(stagedRecordId: string) {
}
const payload = staged.payload as Record<string, { value: unknown; sourceUrl?: string }>;
+ const hints = (staged.hints ?? {}) as unknown as Hints;
const value = <T>(field: string): T | null => (payload[field]?.value ?? null) as T | null;
+ const { label, isDemo } = labelForMethod(staged.source.collectionMethod);
- let publishedEntityId: string | null = null;
+ if (!hints.builderSlug) {
+ throw new Error(`staged record ${staged.id} has no builderSlug hint — cannot publish`);
+ }
+ const builder = await prisma.builder.findUniqueOrThrow({ where: { slug: hints.builderSlug } });
- if (staged.entityType === "INVENTORY_HOME") {
- const builder = await prisma.builder.findUniqueOrThrow({ where: { slug: "meridian-homes" } });
- const community = await prisma.community.findFirstOrThrow({
- where: { name: "Cedar Bend", builderId: builder.id },
+ const findCommunity = async () => {
+ if (!hints.communityName) throw new Error(`staged record ${staged.id} has no communityName hint`);
+ return prisma.community.findFirstOrThrow({
+ where: { name: hints.communityName, builderId: builder.id },
});
+ };
+
+ let publishedEntityId: string | null = null;
+
+ if (staged.entityType === "COMMUNITY") {
+ const name = value<string>("name");
+ if (!name) throw new Error(`staged community ${staged.id} has no name`);
+ const city = value<string>("city");
+ const state = value<string>("state");
+ const zip = value<string>("zip");
+ const existing = await prisma.community.findFirst({ where: { name, builderId: builder.id } });
+ const data = {
+ name,
+ street: value<string>("street"),
+ city: city ?? "",
+ state: state ?? "",
+ zip: zip ?? "",
+ county: value<string>("county"),
+ metro: value<string>("metro"),
+ lat: value<number>("lat"),
+ lon: value<number>("lon"),
+ hoaFeeMonthly: value<number>("hoaFeeMonthly"),
+ hoaRequired: value<number>("hoaFeeMonthly") !== null ? true : null,
+ schoolDistrict: value<string>("schoolDistrict"),
+ ageRestricted: value<boolean>("ageRestricted") ?? false,
+ searchText: [name, city, state, zip, value<string>("metro"), value<string>("county")]
+ .filter(Boolean)
+ .join(" "),
+ sourceUrl: payload.name?.sourceUrl ?? null,
+ isDemo,
+ };
+ const community = existing
+ ? await prisma.community.update({ where: { id: existing.id }, data })
+ : await prisma.community.create({
+ data: {
+ ...data,
+ builderId: builder.id,
+ slug: await uniqueCommunitySlug(name, builder.slug),
+ },
+ });
+ publishedEntityId = community.id;
+ } else if (staged.entityType === "INVENTORY_HOME") {
+ const community = await findCommunity();
const priorPrice = await prisma.inventoryHome.findUnique({
where: { canonicalKey: staged.canonicalKey },
select: { price: true },
@@ -43,13 +127,13 @@ export async function publishStagedRecord(stagedRecordId: string) {
canonicalKey: staged.canonicalKey,
communityId: community.id,
builderId: builder.id,
- builderInventoryId: value<string>("builderInventoryId"),
+ builderInventoryId: value<string>("builderInventoryId") ?? hints.builderInventoryId ?? null,
street,
city: value<string>("city") ?? community.city,
state: value<string>("state") ?? community.state,
zip: value<string>("zip") ?? community.zip,
normalizedAddress: street ? normalizeAddress(street) : null,
- lotNumber: value<string>("lotNumber"),
+ lotNumber: value<string>("lotNumber") ?? hints.lotNumber ?? null,
lat: value<number>("lat"),
lon: value<number>("lon"),
price,
@@ -65,17 +149,25 @@ export async function publishStagedRecord(stagedRecordId: string) {
: null,
status: "PUBLISHED",
freshness: "FRESH",
- verificationLabel: "DEMONSTRATION", // fixture source — never a production claim
+ verificationLabel: label,
lastVerifiedAt: new Date(),
sourceId: staged.sourceId,
- sourceUrl: payload.street?.sourceUrl ?? null,
- isDemo: true,
+ sourceUrl: payload.street?.sourceUrl ?? payload.price?.sourceUrl ?? null,
+ isDemo,
},
update: {
previousPrice: priorPrice?.price ?? undefined,
price,
+ beds: value<number>("beds"),
+ bathsTotal: value<number>("bathsTotal"),
+ sqft: value<number>("sqft"),
+ constructionStatus: (value<string>("constructionStatus") as never) ?? undefined,
+ estCompletionDate: value<string>("estCompletionDate")
+ ? new Date(value<string>("estCompletionDate")!)
+ : undefined,
lastVerifiedAt: new Date(),
freshness: "FRESH",
+ verificationLabel: label,
updatedAt: new Date(),
},
});
@@ -94,31 +186,55 @@ export async function publishStagedRecord(stagedRecordId: string) {
});
}
} else if (staged.entityType === "INCENTIVE") {
- const builder = await prisma.builder.findUniqueOrThrow({ where: { slug: "meridian-homes" } });
- const community = await prisma.community.findFirstOrThrow({
- where: { name: "Cedar Bend", builderId: builder.id },
- });
+ const community = await findCommunity();
const expiresAt = value<string>("expiresAt");
- const incentive = await prisma.incentive.create({
- data: {
- scope: "COMMUNITY",
- builderId: builder.id,
- communityId: community.id,
- title: value<string>("title") ?? "Untitled offer",
- description: value<string>("description"),
- valueUsd: value<number>("valueUsd"),
- expiresAt: expiresAt ? new Date(expiresAt) : null,
- // Spec rule: incentives need an expiration OR an explicit label.
- evergreenLabel: expiresAt ? null : "expiration not provided",
- sourceId: staged.sourceId,
- isDemo: true,
- },
+ const title = value<string>("title") ?? "Untitled offer";
+ const existing = await prisma.incentive.findFirst({
+ where: { communityId: community.id, title, sourceId: staged.sourceId },
});
+ const data = {
+ description: value<string>("description"),
+ valueUsd: value<number>("valueUsd"),
+ expiresAt: expiresAt ? new Date(expiresAt) : null,
+ // Spec rule: incentives need an expiration OR an explicit label.
+ evergreenLabel: expiresAt ? null : "expiration not provided",
+ isDemo,
+ };
+ const incentive = existing
+ ? await prisma.incentive.update({ where: { id: existing.id }, data })
+ : await prisma.incentive.create({
+ data: {
+ ...data,
+ scope: "COMMUNITY",
+ builderId: builder.id,
+ communityId: community.id,
+ title,
+ sourceId: staged.sourceId,
+ },
+ });
publishedEntityId = incentive.id;
- } else if (staged.entityType === "COMMUNITY") {
- // Fixture communities are pre-seeded (FK target); publishing refreshes verification only.
- const community = await prisma.community.findFirst({ where: { name: value<string>("name") ?? "" } });
- publishedEntityId = community?.id ?? null;
+ } else if (staged.entityType === "FLOOR_PLAN") {
+ const community = await findCommunity();
+ const name = value<string>("name") ?? hints.planName;
+ if (!name) throw new Error(`staged floor plan ${staged.id} has no name`);
+ const existing = await prisma.floorPlan.findFirst({ where: { communityId: community.id, name } });
+ const data = {
+ beds: value<number>("beds"),
+ bathsFull: value<number>("bathsFull"),
+ bathsHalf: value<number>("bathsHalf"),
+ sqft: value<number>("sqft"),
+ stories: value<number>("stories"),
+ garageSpaces: value<number>("garageSpaces"),
+ basePrice: value<number>("basePrice"),
+ sourceUrl: payload.name?.sourceUrl ?? null,
+ isDemo,
+ };
+ const plan = existing
+ ? await prisma.floorPlan.update({ where: { id: existing.id }, data })
+ : await prisma.floorPlan.create({
+ data: { ...data, communityId: community.id, builderId: builder.id, name, planMediaRights: "none" },
+ });
+ publishedEntityId = plan.id;
}
await prisma.stagedRecord.update({
@@ -134,3 +250,11 @@ export async function publishStagedRecord(stagedRecordId: string) {
}
return publishedEntityId;
}
+
+async function uniqueCommunitySlug(name: string, builderSlug: string): Promise<string> {
+ const base = slugify(name);
+ if (!(await prisma.community.findUnique({ where: { slug: base } }))) return base;
+ const withBuilder = `${slugify(builderSlug)}-${base}`;
+ if (!(await prisma.community.findUnique({ where: { slug: withBuilder } }))) return withBuilder;
+ return `${withBuilder}-${Date.now().toString(36)}`;
+}
← 7c746cc fix Leaflet SSR: keep MapView out of the shared-ui barrel, s
·
back to Homesonspec
·
integration tests (pipeline e2e, only-writer invariant, revi 0ba8822 →