From 6743cd8baed534f1a553c09b423075c477c781e6 Mon Sep 17 00:00:00 2001 From: Jun-te Kim Date: Tue, 7 Jul 2026 16:09:21 +0000 Subject: [PATCH] feat(bulk-upload): gate finalise on unresolved addresses + overlay corrections (ADR-0057) dispatchFinaliser now: - blocks (before the CAS claim) when any flagged combiner row is still unresolved -> new "unresolved_addresses" outcome; finalize route returns 409. - overlays the user's confirmed UPRNs onto the combiner CSV (buildConfirmedCombinerUri writes a sibling .confirmed.csv) and points the finaliser at it, so a confirmed UPRN drives identity + property_overrides in one pass. No corrections -> original URI unchanged. tsc --noEmit clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- .../bulk-uploads/[uploadId]/finalize/route.ts | 8 +++ src/lib/bulkUpload/addressMatches.ts | 69 +++++++++++++++++++ src/lib/bulkUpload/server.ts | 23 ++++++- 3 files changed, 99 insertions(+), 1 deletion(-) diff --git a/src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/finalize/route.ts b/src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/finalize/route.ts index f326c65c..c5ecc423 100644 --- a/src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/finalize/route.ts +++ b/src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/finalize/route.ts @@ -35,6 +35,14 @@ export async function POST( return NextResponse.json({ error: "Combiner not finished" }, { status: 409 }); case "missing_task": return NextResponse.json({ error: "Upload has no task to finalise" }, { status: 409 }); + case "unresolved_addresses": + return NextResponse.json( + { + error: `${result.count} address${result.count === 1 ? "" : "es"} still need a UPRN. Resolve them on the Addresses tab before finalising.`, + unresolvedAddresses: result.count, + }, + { status: 409 }, + ); case "wrong_state": return NextResponse.json( { error: `Upload not ready to finalize (state: ${result.current})` }, diff --git a/src/lib/bulkUpload/addressMatches.ts b/src/lib/bulkUpload/addressMatches.ts index 198a4c0b..8d3213ab 100644 --- a/src/lib/bulkUpload/addressMatches.ts +++ b/src/lib/bulkUpload/addressMatches.ts @@ -121,6 +121,75 @@ export async function countUnresolvedFlaggedAddresses(uploadId: string): Promise return flagged.filter((f) => !f.resolved).length; } +const ADDRESS_COL = "address2uprn_address"; +const OVERLAY_STATUS_CONFIRMED = "user_confirmed"; +const OVERLAY_STATUS_NO_UPRN = "user_no_uprn"; + +/** + * Overlay the user's saved corrections onto the combiner CSV and write the + * result to a sibling `…​.confirmed.csv`, returning its S3 URI (ADR-0057). The + * finaliser reads this so a confirmed UPRN drives both the identity insert and + * property_overrides in one pass. No corrections ⇒ the original URI is returned + * unchanged (no rewrite). A `markedNoUprn` row keeps an empty UPRN (accepted + * terminal state — the finaliser leaves it as a no-UPRN property). + */ +export async function buildConfirmedCombinerUri( + uploadId: string, + combinedOutputS3Uri: string, +): Promise { + const corrections = await db + .select() + .from(bulkUploadUprnCorrections) + .where(eq(bulkUploadUprnCorrections.uploadId, uploadId)); + if (corrections.length === 0) return combinedOutputS3Uri; + const byRow = new Map(corrections.map((c) => [c.sourceRowId, c])); + + const { bucket, key } = parseS3Uri(combinedOutputS3Uri); + const s3 = createRetrofitDataS3Client(); + const result = await s3.getObject({ Bucket: bucket, Key: key }).promise(); + const body = result.Body?.toString("utf-8"); + if (!body) return combinedOutputS3Uri; + + const wb = XLSX.read(body, { type: "string" }); + const sheet = wb.Sheets[wb.SheetNames[0]]; + // Array-of-arrays so every column + its order is preserved exactly; we patch + // only the address2uprn cells of corrected rows. + const aoa = XLSX.utils.sheet_to_json(sheet, { + header: 1, + defval: "", + raw: false, + blankrows: false, + }); + const headers = (aoa[0] ?? []).map((h) => String(h)); + const iRow = headers.indexOf(SOURCE_ROW_ID_COLUMN); + const iUprn = headers.indexOf(UPRN_COL); + const iStatus = headers.indexOf(STATUS_COL); + const iAddr = headers.indexOf(ADDRESS_COL); + if (iRow < 0 || iUprn < 0) return combinedOutputS3Uri; // not an address2uprn CSV + + for (let r = 1; r < aoa.length; r++) { + const line = aoa[r]; + const srid = String(line[iRow] ?? "").trim(); + const c = byRow.get(srid); + if (!c) continue; + if (c.markedNoUprn) { + line[iUprn] = ""; + if (iStatus >= 0) line[iStatus] = OVERLAY_STATUS_NO_UPRN; + } else { + line[iUprn] = c.uprn ?? ""; + if (iStatus >= 0) line[iStatus] = OVERLAY_STATUS_CONFIRMED; + if (iAddr >= 0 && c.address) line[iAddr] = c.address; + } + } + + const csv = XLSX.utils.sheet_to_csv(XLSX.utils.aoa_to_sheet(aoa)); + const confirmedKey = `${key.replace(/\.csv$/i, "")}.confirmed.csv`; + await s3 + .putObject({ Bucket: bucket, Key: confirmedKey, Body: csv, ContentType: "text/csv" }) + .promise(); + return `s3://${bucket}/${confirmedKey}`; +} + export type UprnCorrectionInput = | { sourceRowId: string; diff --git a/src/lib/bulkUpload/server.ts b/src/lib/bulkUpload/server.ts index ca303f9b..843156df 100644 --- a/src/lib/bulkUpload/server.ts +++ b/src/lib/bulkUpload/server.ts @@ -20,6 +20,10 @@ import { retrofitDataS3Bucket } from "@/app/utils/s3"; import { SUBTASK_SERVICE } from "./types"; import type { MultiEntrySummary } from "./multiEntry"; import { isPermutation } from "./multiEntry"; +import { + buildConfirmedCombinerUri, + countUnresolvedFlaggedAddresses, +} from "./addressMatches"; const REMAP_ALLOWED: ReadonlySet = new Set([ "ready_for_processing", @@ -677,6 +681,7 @@ export type DispatchFinaliserOutcome = | { kind: "not_yet_combined" } | { kind: "wrong_state"; current: string } | { kind: "missing_task" } + | { kind: "unresolved_addresses"; count: number } | { kind: "trigger_failed"; status: number; message: string }; // Dispatch the async bulk_upload_finaliser (ADR-0005). Replaces the old @@ -705,6 +710,14 @@ export async function dispatchFinaliser(args: { const upload = guarded.upload; if (!upload.taskId) return { kind: "missing_task" }; + // ADR-0057: every combiner row address2uprn could not confidently match must + // be resolved on the Addresses tab (a chosen UPRN or an explicit no-UPRN) + // before finalise — otherwise a withheld row would land as an unintended + // no-UPRN property. Gate before the CAS claim so a blocked upload stays in + // awaiting_review. + const unresolved = await countUnresolvedFlaggedAddresses(args.uploadId); + if (unresolved > 0) return { kind: "unresolved_addresses", count: unresolved }; + // CAS: atomically claim the dispatch. Only the request that flips // awaiting_review → finalising proceeds; a concurrent one updates 0 rows. const claimed = await db @@ -748,10 +761,18 @@ export async function dispatchFinaliser(args: { ? `s3://${retrofitDataS3Bucket()}/${classifierCsvKey(upload.portfolioId, args.uploadId)}` : null; + // Overlay the user's confirmed UPRNs onto the combiner CSV and point the + // finaliser at the corrected copy (ADR-0057). No corrections ⇒ the original + // URI is returned unchanged. + const confirmedS3Uri = await buildConfirmedCombinerUri( + args.uploadId, + upload.combinedOutputS3Uri!, + ); + const payload = { task_id: upload.taskId, sub_task_id: subTask.id, - s3_uri: upload.combinedOutputS3Uri, + s3_uri: confirmedS3Uri, portfolio_id: Number(upload.portfolioId), bulk_upload_id: args.uploadId, classifier_s3_uri: classifierS3Uri,