feat(bulk-upload): read flagged combiner rows + save UPRN corrections (ADR-0057)

Server module + two routes for the pre-finalise Addresses tab:
- getFlaggedAddresses: reads the combiner CSV from combinedOutputS3Uri,
  returns rows address2uprn could not confidently match (empty UPRN or a
  non-"matched" status), joined with any saved correction.
- upsertUprnCorrection: upserts a confirmed UPRN / no-UPRN per source_row_id.
- GET .../address-matches, POST .../address-corrections (thin, auth-guarded).

tsc --noEmit clean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/address-corrections/route.ts

@@ -0,0 +1,43 @@
+import { NextRequest, NextResponse } from "next/server";
+import { getServerSession } from "next-auth";
+import { z } from "zod";
+
+import { AuthOptions } from "@/app/api/auth/[...nextauth]/authOptions";
+import { upsertUprnCorrection } from "@/lib/bulkUpload/addressMatches";
+
+// Confirm one flagged combiner row (ADR-0057): either assign a UPRN chosen from
+// OS Places, or assert the address has none. Keyed by `source_row_id` (there is
+// no property.id yet — the finaliser creates identities). Upsert semantics, so
+// re-confirming overwrites. The confirmation is overlaid onto the combiner CSV
+// at dispatchFinaliser.
+const AssignSchema = z.object({
+  sourceRowId: z.string().min(1),
+  uprn: z.string().regex(/^\d+$/, "UPRN must be numeric"),
+  address: z.string().min(1),
+  postcode: z.string().min(1),
+});
+const NoUprnSchema = z.object({
+  sourceRowId: z.string().min(1),
+  markedNoUprn: z.literal(true),
+});
+const BodySchema = z.union([NoUprnSchema, AssignSchema]);
+
+export async function POST(
+  request: NextRequest,
+  { params }: { params: Promise<{ portfolioId: string; uploadId: string }> },
+) {
+  const session = await getServerSession(AuthOptions);
+  if (!session) return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
+
+  const { uploadId } = await params;
+  const parsed = BodySchema.safeParse(await request.json().catch(() => null));
+  if (!parsed.success) {
+    return NextResponse.json(
+      { error: parsed.error.issues[0]?.message ?? "Invalid request" },
+      { status: 400 },
+    );
+  }
+
+  await upsertUprnCorrection(uploadId, parsed.data, session.user?.email ?? undefined);
+  return NextResponse.json({ ok: true }, { status: 200 });
+}

src/app/api/portfolio/[portfolioId]/bulk-uploads/[uploadId]/address-matches/route.ts

@@ -0,0 +1,22 @@
+import { NextRequest, NextResponse } from "next/server";
+import { getServerSession } from "next-auth";
+
+import { AuthOptions } from "@/app/api/auth/[...nextauth]/authOptions";
+import { getFlaggedAddresses } from "@/lib/bulkUpload/addressMatches";
+
+// Read-only: the combiner rows address2uprn could not confidently match
+// (unmatched / ambiguous_duplicate / invalid_postcode), joined with any saved
+// correction. Drives the pre-finalise "Addresses" review tab (ADR-0057); the
+// Finalise gate blocks until every flagged row is resolved.
+export async function GET(
+  _request: NextRequest,
+  { params }: { params: Promise<{ portfolioId: string; uploadId: string }> },
+) {
+  const session = await getServerSession(AuthOptions);
+  if (!session) return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
+
+  const { uploadId } = await params;
+  const flagged = await getFlaggedAddresses(uploadId);
+  const unresolved = flagged.filter((f) => !f.resolved).length;
+  return NextResponse.json({ flagged, unresolved }, { status: 200 });
+}

src/lib/bulkUpload/addressMatches.ts

@@ -0,0 +1,177 @@
+import { eq } from "drizzle-orm";
+import * as XLSX from "xlsx";
+
+import { db } from "@/app/db/db";
+import { bulkAddressUploads } from "@/app/db/schema/bulk_address_uploads";
+import { bulkUploadUprnCorrections } from "@/app/db/schema/bulk_upload_uprn_corrections";
+import { createRetrofitDataS3Client } from "@/app/utils/s3";
+import { SOURCE_ROW_ID_COLUMN } from "./s3Keys";
+
+// Columns address2uprn writes onto each combiner row (ADR-0057). `status` is the
+// per-row outcome; a withheld/ambiguous/no-match row has an empty `uprn`.
+const UPRN_COL = "address2uprn_uprn";
+const STATUS_COL = "address2uprn_status";
+const LEXISCORE_COL = "address2uprn_lexiscore";
+
+// A combiner row is confidently matched only when address2uprn kept a UPRN AND
+// did not flag it. Everything else — unmatched, ambiguous_duplicate,
+// invalid_postcode, error, or a legacy row with no status but no UPRN — needs
+// the user to resolve it on the Addresses tab before finalise.
+function isConfidentlyMatched(row: Record<string, string>): boolean {
+  const uprn = (row[UPRN_COL] ?? "").trim();
+  const status = (row[STATUS_COL] ?? "").trim();
+  if (!uprn) return false;
+  return status === "" || status === "matched";
+}
+
+export interface FlaggedAddressRow {
+  sourceRowId: string;
+  address: string;
+  postcode: string;
+  status: string;
+  lexiscore: number | null;
+  // The candidate address2uprn found but did not trust (null when none).
+  suggestedUprn: string | null;
+  // The user's confirmation (from bulk_upload_uprn_corrections), if any.
+  resolvedUprn: string | null;
+  resolvedAddress: string | null;
+  markedNoUprn: boolean;
+  // Resolved = user picked a UPRN OR asserted there is none.
+  resolved: boolean;
+}
+
+function parseS3Uri(uri: string): { bucket: string; key: string } {
+  const url = new URL(uri); // s3://bucket/key...
+  const key = url.pathname.startsWith("/") ? url.pathname.slice(1) : url.pathname;
+  if (!url.hostname || !key) throw new Error(`Malformed S3 URI: ${uri}`);
+  return { bucket: url.hostname, key };
+}
+
+async function readCombinerRows(s3Uri: string): Promise<Record<string, string>[]> {
+  const { bucket, key } = parseS3Uri(s3Uri);
+  const s3 = createRetrofitDataS3Client();
+  const result = await s3.getObject({ Bucket: bucket, Key: key }).promise();
+  const body = result.Body?.toString("utf-8");
+  if (!body) return [];
+  const wb = XLSX.read(body, { type: "string" });
+  const sheet = wb.Sheets[wb.SheetNames[0]];
+  // Every cell as a string so a numeric UPRN never arrives as a float.
+  return XLSX.utils.sheet_to_json<Record<string, string>>(sheet, { defval: "", raw: false });
+}
+
+function buildAddress(row: Record<string, string>): string {
+  return ["Address 1", "Address 2", "Address 3"]
+    .map((c) => (row[c] ?? "").trim())
+    .filter(Boolean)
+    .join(", ");
+}
+
+/**
+ * The combiner rows that need user confirmation, joined with any saved
+ * correction. Reads the combiner CSV from `combinedOutputS3Uri` (ADR-0057).
+ * Returns [] until the combiner has run.
+ */
+export async function getFlaggedAddresses(
+  uploadId: string,
+): Promise<FlaggedAddressRow[]> {
+  const [row] = await db
+    .select({ combinedOutputS3Uri: bulkAddressUploads.combinedOutputS3Uri })
+    .from(bulkAddressUploads)
+    .where(eq(bulkAddressUploads.id, uploadId));
+  if (!row?.combinedOutputS3Uri) return [];
+
+  const [rows, corrections] = await Promise.all([
+    readCombinerRows(row.combinedOutputS3Uri),
+    db
+      .select()
+      .from(bulkUploadUprnCorrections)
+      .where(eq(bulkUploadUprnCorrections.uploadId, uploadId)),
+  ]);
+
+  const correctionByRow = new Map(corrections.map((c) => [c.sourceRowId, c]));
+
+  return rows
+    .filter((r) => !isConfidentlyMatched(r))
+    .map((r) => {
+      const sourceRowId = (r[SOURCE_ROW_ID_COLUMN] ?? "").trim();
+      const uprn = (r[UPRN_COL] ?? "").trim();
+      const rawScore = (r[LEXISCORE_COL] ?? "").trim();
+      const lexiscore = rawScore === "" ? null : Number(rawScore);
+      const c = correctionByRow.get(sourceRowId);
+      const resolvedUprn = c?.uprn ?? null;
+      const markedNoUprn = c?.markedNoUprn ?? false;
+      return {
+        sourceRowId,
+        address: buildAddress(r),
+        postcode: (r["postcode"] ?? "").trim(),
+        status: (r[STATUS_COL] ?? (uprn ? "matched" : "unmatched")).trim(),
+        lexiscore: lexiscore !== null && Number.isFinite(lexiscore) ? lexiscore : null,
+        suggestedUprn: uprn || null,
+        resolvedUprn,
+        resolvedAddress: c?.address ?? null,
+        markedNoUprn,
+        resolved: resolvedUprn !== null || markedNoUprn,
+      };
+    });
+}
+
+/** Count of flagged rows the user has not yet resolved — gates Finalise. */
+export async function countUnresolvedFlaggedAddresses(uploadId: string): Promise<number> {
+  const flagged = await getFlaggedAddresses(uploadId);
+  return flagged.filter((f) => !f.resolved).length;
+}
+
+export type UprnCorrectionInput =
+  | {
+      sourceRowId: string;
+      uprn: string;
+      address: string;
+      postcode: string;
+      markedNoUprn?: false;
+    }
+  | { sourceRowId: string; markedNoUprn: true };
+
+/** Upsert one combiner row's confirmation (assign a UPRN, or mark no-UPRN). */
+export async function upsertUprnCorrection(
+  uploadId: string,
+  input: UprnCorrectionInput,
+  userId?: string,
+): Promise<void> {
+  const values = input.markedNoUprn
+    ? {
+        uploadId,
+        sourceRowId: input.sourceRowId,
+        uprn: null,
+        address: null,
+        postcode: null,
+        markedNoUprn: true as const,
+        userId: userId ?? null,
+      }
+    : {
+        uploadId,
+        sourceRowId: input.sourceRowId,
+        uprn: input.uprn,
+        address: input.address,
+        postcode: input.postcode,
+        markedNoUprn: false as const,
+        userId: userId ?? null,
+      };
+
+  await db
+    .insert(bulkUploadUprnCorrections)
+    .values(values)
+    .onConflictDoUpdate({
+      target: [
+        bulkUploadUprnCorrections.uploadId,
+        bulkUploadUprnCorrections.sourceRowId,
+      ],
+      set: {
+        uprn: values.uprn,
+        address: values.address,
+        postcode: values.postcode,
+        markedNoUprn: values.markedNoUprn,
+        userId: values.userId,
+        updatedAt: new Date(),
+      },
+    });
+}