import { db } from "@/app/db/db";
import { bulkAddressUploads } from "@/app/db/schema/bulk_address_uploads";
import {
  CATEGORY_TABLES,
  CATEGORY_VALUES,
  LANDLORD_CATEGORIES,
  isLandlordCategory,
} from "@/lib/landlordOverrides/categories";
import { tasks } from "@/app/db/schema/tasks/tasks";
import { subTasks } from "@/app/db/schema/tasks/subtask";
import { and, count, desc, eq, inArray, sql } from "drizzle-orm";
import type { BulkUpload, BulkUploadStatus, ProgressView, TaskSummary } from "./types";
import { validateColumnMapping, classifierMapping } from "./columnFields";
import { classifierCsvKey } from "./s3Keys";
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<BulkUploadStatus> = new Set([
  "ready_for_processing",
  "mapping_complete",
]);

type FastApiTriggerArgs = {
  endpoint: string;
  payload: Record<string, unknown>;
  sessionToken: string | undefined;
};

type FastApiTriggerResult = { ok: true } | { ok: false; status: number; message: string };

async function triggerFastApiPipeline(args: FastApiTriggerArgs): Promise<FastApiTriggerResult> {
  const url = process.env.FASTAPI_API_URL;
  const key = process.env.FASTAPI_API_KEY;
  if (!url || !key) {
    console.error("FASTAPI_API_URL or FASTAPI_API_KEY not set");
    return { ok: false, status: 500, message: "Server misconfiguration" };
  }

  try {
    const res = await fetch(`${url}${args.endpoint}`, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        "x-api-key": key,
        Authorization: `Bearer ${args.sessionToken}`,
      },
      body: JSON.stringify(args.payload),
    });
    if (!res.ok) {
      const errText = await res.text().catch(() => "");
      console.error(`FastAPI ${args.endpoint} failed:`, res.status, errText);
      return { ok: false, status: 502, message: "Pipeline trigger failed" };
    }
    return { ok: true };
  } catch (err) {
    console.error(`Failed to reach FastAPI ${args.endpoint}:`, err);
    return { ok: false, status: 502, message: "Pipeline trigger failed" };
  }
}

async function loadById(uploadId: string): Promise<BulkUpload | null> {
  const [row] = await db
    .select()
    .from(bulkAddressUploads)
    .where(eq(bulkAddressUploads.id, uploadId))
    .limit(1);
  return row ?? null;
}

export async function listForPortfolio(portfolioId: string): Promise<BulkUpload[]> {
  return db
    .select()
    .from(bulkAddressUploads)
    .where(eq(bulkAddressUploads.portfolioId, portfolioId))
    .orderBy(desc(bulkAddressUploads.createdAt));
}

async function loadTaskSummary(taskId: string): Promise<TaskSummary | null> {
  const [row] = await db
    .select({
      id: tasks.id,
      taskSource: tasks.taskSource,
      status: tasks.status,
      service: tasks.service,
      jobStarted: tasks.jobStarted,
      jobCompleted: tasks.jobCompleted,
      updatedAt: tasks.updatedAt,
      totalSubtasks: count(subTasks.id),
      completedSubtasks: sql<number>`count(case when lower(${subTasks.status}) in ('completed', 'complete') then 1 end)::int`,
      failedSubtasks: sql<number>`count(case when lower(${subTasks.status}) in ('failed', 'failure', 'error') then 1 end)::int`,
      addressTotal: sql<number>`count(case when (${subTasks.service} = 'address2uprn' or ${subTasks.service} is null) and ${subTasks.id} is not null then 1 end)::int`,
      addressCompleted: sql<number>`count(case when (${subTasks.service} = 'address2uprn' or ${subTasks.service} is null) and lower(${subTasks.status}) in ('completed', 'complete') then 1 end)::int`,
      addressFailed: sql<number>`count(case when (${subTasks.service} = 'address2uprn' or ${subTasks.service} is null) and lower(${subTasks.status}) in ('failed', 'failure', 'error') then 1 end)::int`,
      classifierTotal: sql<number>`count(case when ${subTasks.service} = 'landlord_description_overrides' then 1 end)::int`,
      classifierCompleted: sql<number>`count(case when ${subTasks.service} = 'landlord_description_overrides' and lower(${subTasks.status}) in ('completed', 'complete') then 1 end)::int`,
      classifierFailed: sql<number>`count(case when ${subTasks.service} = 'landlord_description_overrides' and lower(${subTasks.status}) in ('failed', 'failure', 'error') then 1 end)::int`,
    })
    .from(tasks)
    .leftJoin(subTasks, eq(subTasks.taskId, tasks.id))
    .where(eq(tasks.id, taskId))
    .groupBy(tasks.id)
    .limit(1);
  return row ?? null;
}

export async function getProgressView(uploadId: string): Promise<ProgressView | null> {
  const upload = await loadById(uploadId);
  if (!upload) return null;
  const task = upload.taskId ? await loadTaskSummary(upload.taskId) : null;
  return { upload, task };
}

// Persist the multi-entry building-part detection (ADR-0004). Computed once at
// start-address-matching from the already-parsed rows; read back on the
// awaiting_review surface. Only this column is touched, so the later
// status/taskId update leaves it intact.
export async function saveMultiEntrySummary(
  uploadId: string,
  summary: MultiEntrySummary,
): Promise<void> {
  await db
    .update(bulkAddressUploads)
    .set({ multiEntrySummary: summary })
    .where(eq(bulkAddressUploads.id, uploadId));
}

// Classifier field -> resolved enum, keyed by the normalized description that
// the classifier persisted. Empty inner maps / absent fields mean "not
// classified (yet)". Read-only (ADR-0004, issue #298).
export type SampleClassifications = Record<string, Record<string, string>>;

// Look up the classifier's resolved enums for one field's descriptions, via the
// category registry (all nine landlord categories). Returns null for a field
// that is not a landlord category.
async function lookupOverrides(
  field: string,
  portfolioId: bigint,
  descriptions: string[],
): Promise<Array<{ description: string; value: string }> | null> {
  if (!isLandlordCategory(field)) return null;
  const { table } = CATEGORY_TABLES[field];
  return db
    .select({ description: table.description, value: table.value })
    .from(table)
    .where(
      and(eq(table.portfolioId, portfolioId), inArray(table.description, descriptions)),
    ) as Promise<Array<{ description: string; value: string }>>;
}

// The classifier's enums for the review samples' entries, joined by the
// normalized description (exact match — the summary stored it the way the
// classifier persists it, so no re-normalization here). Read-only. Covers the
// Step 1 verify sample AND every per-count ordering sample, since smaller-count
// panels may show descriptions the largest-count sample doesn't (ADR-0004
// amendment).
export async function getSampleClassifications(
  uploadId: string,
): Promise<SampleClassifications> {
  const upload = await loadById(uploadId);
  const summary = upload?.multiEntrySummary;
  if (!upload || !summary || !summary.sample) return {};

  // Gather distinct descriptions per field across all samples.
  const allSamples = [summary.sample, ...Object.values(summary.samplesByCount ?? {})];
  const descriptionsByField: Record<string, Set<string>> = {};
  for (const sample of allSamples) {
    for (const column of sample.columns) {
      const set = (descriptionsByField[column.field] ??= new Set<string>());
      for (const e of column.entries) set.add(e.description);
    }
  }

  const portfolioId = BigInt(upload.portfolioId);
  const result: SampleClassifications = {};
  for (const [field, descSet] of Object.entries(descriptionsByField)) {
    const descriptions = [...descSet];
    if (descriptions.length === 0) continue;
    const rows = await lookupOverrides(field, portfolioId, descriptions);
    if (!rows) continue;
    result[field] = Object.fromEntries(rows.map((r) => [r.description, r.value]));
  }
  return result;
}

// Descriptions still classified `Unknown` per field, portfolio-wide (ADR-0006).
// `Unknown` is the classifier's "couldn't decide" marker; v2 treats it as
// never-final, so the Finalise gate blocks until the user maps every one to a
// real value (and the finaliser fails loudly if any slips through). Portfolio-
// wide is the right scope under the one-real-upload assumption (ADR-0006).
export type UnknownOverrides = Record<string, string[]>;

const UNKNOWN_VALUE = "Unknown";

async function unknownForField(field: string, portfolioId: bigint): Promise<string[]> {
  if (!isLandlordCategory(field)) return [];
  const { table } = CATEGORY_TABLES[field];
  const rows = (await db
    .select({ description: table.description })
    .from(table)
    .where(
      and(eq(table.portfolioId, portfolioId), eq(table.value, UNKNOWN_VALUE)),
    )) as Array<{ description: string }>;
  return rows.map((r) => r.description);
}

export async function getUnknownOverrides(portfolioId: string): Promise<UnknownOverrides> {
  const pid = BigInt(portfolioId);
  const result: UnknownOverrides = {};
  for (const field of LANDLORD_CATEGORIES) {
    const descriptions = await unknownForField(field, pid);
    if (descriptions.length > 0) result[field] = descriptions;
  }
  return result;
}

// Upsert one user override (source='user') into the right category table via the
// registry. The unique (portfolio_id, description) drives the conflict. Sets
// source='user' so the classifier's `ON CONFLICT … WHERE source='classifier'`
// never re-clobbers it. No-op for a field that is not a landlord category
// (setClassificationOverride validates the field before calling).
async function upsertUserOverride(
  field: string,
  portfolioId: bigint,
  description: string,
  value: string,
): Promise<void> {
  if (!isLandlordCategory(field)) return;
  const { table } = CATEGORY_TABLES[field];
  await db
    .insert(table)
    .values({ portfolioId, description, value, source: "user" })
    .onConflictDoUpdate({
      target: [table.portfolioId, table.description],
      set: { value, source: "user", updatedAt: new Date() },
    });
}

export type SetClassificationOutcome =
  | { kind: "ok" }
  | { kind: "invalid"; reason: string };

// Correct one classification, persisted as a user override (ADR-0004, issue
// #299). Keyed by (portfolio, description), so it changes the mapping for every
// row with that description portfolio-wide. The description is normalized to
// match the classifier's stored key.
export async function setClassificationOverride(
  portfolioId: string,
  field: string,
  description: string,
  value: string,
): Promise<SetClassificationOutcome> {
  if (!isLandlordCategory(field))
    return { kind: "invalid", reason: `Unknown category '${field}'` };
  const allowed = CATEGORY_VALUES[field];
  if (!allowed.includes(value))
    return { kind: "invalid", reason: `'${value}' is not a valid ${field}` };

  const normalized = description.trim().toLowerCase();
  if (!normalized) return { kind: "invalid", reason: "Empty description" };

  await upsertUserOverride(field, BigInt(portfolioId), normalized, value);
  return { kind: "ok" };
}

export type SetOrderingOutcome =
  | { kind: "ok"; upload: BulkUpload }
  | { kind: "not_found" }
  | { kind: "wrong_state"; current: string }
  | { kind: "not_multi_entry" }
  | { kind: "invalid_ordering"; reason: string };

// Persist the user-confirmed building-part ordering (ADR-0004, amended
// 2026-06-05). Allowed only at awaiting_review and only when the upload is
// multi-entry. Each distinct count ≥ 2 needs its own permutation; the UI confirms
// one count at a time, so we MERGE the supplied permutations into any already
// stored, validate each is a bijection, and only mark `confirmed` once EVERY
// detected count has a permutation (which gates Finalise).
export async function setMultiEntryOrdering(
  uploadId: string,
  permutations: Record<string, number[]>,
): Promise<SetOrderingOutcome> {
  const upload = await loadById(uploadId);
  if (!upload) return { kind: "not_found" };
  if (upload.status !== "awaiting_review")
    return { kind: "wrong_state", current: upload.status };

  const summary = upload.multiEntrySummary;
  // A sample now exists for non-multi-entry uploads too (Step 1's verify
  // sample), so "is multi-entry" is largestCount >= 2, not "has a sample".
  if (!summary || summary.largestCount < 2)
    return { kind: "not_multi_entry" };

  for (const [count, permutation] of Object.entries(permutations)) {
    if (permutation.length !== Number(count) || !isPermutation(permutation))
      return { kind: "invalid_ordering", reason: `Ordering for ${count} parts is not a valid arrangement.` };
  }

  // Merge with any counts confirmed earlier, then decide whether every detected
  // count (the keys of countDistribution, all ≥ 2) now has a permutation.
  const merged = { ...(upload.multiEntryOrdering?.permutations ?? {}), ...permutations };
  const requiredCounts = Object.keys(summary.countDistribution);
  const confirmed = requiredCounts.every(
    (c) => Array.isArray(merged[c]) && merged[c].length === Number(c),
  );

  const [updated] = await db
    .update(bulkAddressUploads)
    .set({ multiEntryOrdering: { permutations: merged, confirmed } })
    .where(eq(bulkAddressUploads.id, uploadId))
    .returning();
  if (!updated) return { kind: "not_found" };
  return { kind: "ok", upload: updated };
}

export type SetVerifyAckOutcome =
  | { kind: "ok"; upload: BulkUpload }
  | { kind: "not_found" }
  | { kind: "wrong_state"; current: string };

// Record the user's "Verify classification" acknowledgement (ADR-0004 Step 1).
// Allowed only at awaiting_review. Gates Finalise whenever the upload has
// classifier columns — independent of multi-entry, hence its own column rather
// than a flag on multiEntryOrdering.
export async function setVerifyAck(uploadId: string): Promise<SetVerifyAckOutcome> {
  const upload = await loadById(uploadId);
  if (!upload) return { kind: "not_found" };
  if (upload.status !== "awaiting_review")
    return { kind: "wrong_state", current: upload.status };

  const [updated] = await db
    .update(bulkAddressUploads)
    .set({ verifyAck: true })
    .where(eq(bulkAddressUploads.id, uploadId))
    .returning();
  if (!updated) return { kind: "not_found" };
  return { kind: "ok", upload: updated };
}

export type SetMappingOutcome =
  | { kind: "ok"; upload: BulkUpload }
  | { kind: "not_found" }
  | { kind: "invalid_status"; current: string }
  | { kind: "invalid_mapping"; reason: string };

export async function setColumnMapping(
  uploadId: string,
  mapping: Record<string, string>,
): Promise<SetMappingOutcome> {
  const upload = await loadById(uploadId);
  if (!upload) return { kind: "not_found" };
  if (!REMAP_ALLOWED.has(upload.status as BulkUploadStatus))
    return { kind: "invalid_status", current: upload.status };

  const reason = validateColumnMapping(mapping);
  if (reason) return { kind: "invalid_mapping", reason };

  const [updated] = await db
    .update(bulkAddressUploads)
    .set({ columnMapping: mapping, status: "mapping_complete" })
    .where(eq(bulkAddressUploads.id, uploadId))
    .returning();
  if (!updated) return { kind: "not_found" };
  return { kind: "ok", upload: updated };
}

export type LoadForAddressMatchingOutcome =
  | { kind: "ok"; upload: BulkUpload }
  | { kind: "not_found" }
  | { kind: "wrong_state"; current: string }
  | { kind: "missing_mapping" };

export async function loadForAddressMatching(
  uploadId: string,
): Promise<LoadForAddressMatchingOutcome> {
  const upload = await loadById(uploadId);
  if (!upload) return { kind: "not_found" };
  if (upload.status !== "mapping_complete")
    return { kind: "wrong_state", current: upload.status };
  if (!upload.columnMapping) return { kind: "missing_mapping" };
  return { kind: "ok", upload };
}

export type TriggerAddressMatchingOutcome =
  | { kind: "ok"; taskId: string }
  | { kind: "trigger_failed"; status: number; message: string };

export async function triggerAddressMatching(args: {
  uploadId: string;
  s3Uri: string;
  sessionToken: string | undefined;
}): Promise<TriggerAddressMatchingOutcome> {
  const upload = await loadById(args.uploadId);
  if (!upload) return { kind: "trigger_failed", status: 404, message: "Upload not found" };

  const now = new Date();
  const [task] = await db
    .insert(tasks)
    .values({
      taskSource: `Address Onboarding – ${upload.filename}`,
      service: "address2uprn",
      source: "portfolio_id",
      sourceId: upload.portfolioId,
      status: "waiting",
      jobStarted: now,
    })
    .returning();
  const [subTask] = await db
    .insert(subTasks)
    .values({
      taskId: task.id,
      status: "waiting",
      service: SUBTASK_SERVICE.address,
      inputs: JSON.stringify({ bulk_upload_id: args.uploadId }),
    })
    .returning();

  const payload = {
    task_id: task.id,
    sub_task_id: subTask.id,
    s3_uri: args.s3Uri,
  };

  const trigger = await triggerFastApiPipeline({
    endpoint: "/v1/bulk-uploads/trigger-postcode-splitter",
    payload,
    sessionToken: args.sessionToken,
  });
  if (!trigger.ok)
    return { kind: "trigger_failed", status: trigger.status, message: trigger.message };

  await Promise.all([
    db
      .update(bulkAddressUploads)
      .set({ status: "processing", taskId: task.id })
      .where(eq(bulkAddressUploads.id, args.uploadId)),
    db
      .update(tasks)
      .set({ status: "in progress" })
      .where(eq(tasks.id, task.id)),
    db
      .update(subTasks)
      .set({ inputs: JSON.stringify(payload) })
      .where(eq(subTasks.id, subTask.id)),
  ]);
  return { kind: "ok", taskId: task.id };
}

// Co-fires the landlord classifier as a subtask under the address task. Reads a
// dedicated classifier CSV (the classifier columns converted from the upload by
// the start-address-matching route — the address-matching CSV strips the
// description columns), so the lambda always parses a real CSV even for
// .xlsx/.xls uploads. Non-blocking: a trigger failure marks only the classifier
// subtask, so address matching is unaffected. See ADR-0003.
export async function triggerClassifier(args: {
  taskId: string;
  uploadId: string;
  s3Uri: string;
  sessionToken: string | undefined;
}): Promise<void> {
  const upload = await loadById(args.uploadId);
  if (!upload || !upload.columnMapping) return;

  const columnMapping = classifierMapping(upload.columnMapping);
  if (Object.keys(columnMapping).length === 0) return;

  const [subTask] = await db
    .insert(subTasks)
    .values({
      taskId: args.taskId,
      status: "waiting",
      service: SUBTASK_SERVICE.classifier,
      inputs: JSON.stringify({ bulk_upload_id: args.uploadId }),
    })
    .returning();

  const payload = {
    task_id: args.taskId,
    sub_task_id: subTask.id,
    s3_uri: args.s3Uri,
    portfolio_id: Number(upload.portfolioId),
    column_mapping: columnMapping,
  };

  const trigger = await triggerFastApiPipeline({
    endpoint: "/v1/bulk-uploads/trigger-landlord-overrides",
    payload,
    sessionToken: args.sessionToken,
  });

  if (!trigger.ok) {
    await db
      .update(subTasks)
      .set({
        status: "failed",
        outputs: JSON.stringify({ error: trigger.message }),
      })
      .where(eq(subTasks.id, subTask.id));
    return;
  }

  await db
    .update(subTasks)
    .set({ status: "in progress", inputs: JSON.stringify(payload) })
    .where(eq(subTasks.id, subTask.id));
}

export type CombineRetriggerOutcome =
  | { kind: "triggered"; taskId: string; subTaskId: string }
  | { kind: "already_combined" }
  | { kind: "already_dispatched" }
  | { kind: "not_found" }
  | { kind: "missing_task" }
  | { kind: "trigger_failed"; status: number; message: string };

export async function requestCombineRetrigger(args: {
  uploadId: string;
  sessionToken: string | undefined;
}): Promise<CombineRetriggerOutcome> {
  const upload = await loadById(args.uploadId);
  if (!upload) return { kind: "not_found" };
  if (!upload.taskId) return { kind: "missing_task" };
  if (upload.combinedOutputS3Uri) return { kind: "already_combined" };

  // CAS: atomically claim `processing → combining` — the double-dispatch guard
  // (mirrors dispatchFinaliser's awaiting_review → finalising, ADR-0005). Of two
  // rapid "Run Combiner" clicks exactly one flips the row; the loser updates 0
  // rows and bails, so only one combiner subtask is ever dispatched. It also
  // closes the window where status is still `processing` because the backend
  // hasn't written `combining` yet.
  const claimed = await db
    .update(bulkAddressUploads)
    .set({ status: "combining" })
    .where(
      and(
        eq(bulkAddressUploads.id, args.uploadId),
        eq(bulkAddressUploads.status, "processing"),
      ),
    )
    .returning();
  if (claimed.length === 0) return { kind: "already_dispatched" };

  const [subTask] = await db
    .insert(subTasks)
    .values({ taskId: upload.taskId, status: "waiting" })
    .returning();

  const payload = { task_id: upload.taskId, sub_task_id: subTask.id };

  const trigger = await triggerFastApiPipeline({
    endpoint: "/v1/bulk-uploads/trigger-combiner",
    payload,
    sessionToken: args.sessionToken,
  });
  if (!trigger.ok) {
    // Roll the claim back so the user can retry, and fail the subtask.
    await Promise.all([
      db
        .update(bulkAddressUploads)
        .set({ status: "processing" })
        .where(eq(bulkAddressUploads.id, args.uploadId)),
      db
        .update(subTasks)
        .set({
          status: "failed",
          outputs: JSON.stringify({ error: trigger.message }),
        })
        .where(eq(subTasks.id, subTask.id)),
    ]);
    return { kind: "trigger_failed", status: trigger.status, message: trigger.message };
  }

  await db
    .update(subTasks)
    .set({ inputs: JSON.stringify(payload) })
    .where(eq(subTasks.id, subTask.id));

  return { kind: "triggered", taskId: upload.taskId, subTaskId: subTask.id };
}

export type LoadForFinalizeOutcome =
  | { kind: "ready"; upload: BulkUpload }
  | { kind: "already_finalized" }
  | { kind: "not_found" }
  | { kind: "not_yet_combined" }
  | { kind: "wrong_state"; current: string };

export async function loadForFinalize(uploadId: string): Promise<LoadForFinalizeOutcome> {
  const upload = await loadById(uploadId);
  if (!upload) return { kind: "not_found" };
  if (upload.status === "complete") return { kind: "already_finalized" };
  if (upload.status !== "awaiting_review")
    return { kind: "wrong_state", current: upload.status };
  if (!upload.combinedOutputS3Uri) return { kind: "not_yet_combined" };
  return { kind: "ready", upload };
}

export type DispatchFinaliserOutcome =
  | { kind: "ok"; taskId: string; subTaskId: string }
  | { kind: "not_found" }
  | { kind: "already_finalized" }
  | { 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
// synchronous property insert + markFinalized. Order matters:
//   1. loadForFinalize — rich guards (combined output present, awaiting_review).
//   2. CAS claim `awaiting_review → finalising` — the double-dispatch guard:
//      of two simultaneous clicks exactly one updates a row; the loser gets 409.
//   3. create the finaliser subtask under the upload's existing Task + POST the
//      trigger. On trigger failure, revert the status so the user can retry and
//      mark the subtask failed. The backend writes the terminal complete/failed.
export async function dispatchFinaliser(args: {
  uploadId: string;
  sessionToken: string | undefined;
}): Promise<DispatchFinaliserOutcome> {
  const guarded = await loadForFinalize(args.uploadId);
  switch (guarded.kind) {
    case "not_found":
      return { kind: "not_found" };
    case "already_finalized":
      return { kind: "already_finalized" };
    case "not_yet_combined":
      return { kind: "not_yet_combined" };
    case "wrong_state":
      return { kind: "wrong_state", current: guarded.current };
  }
  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
    .update(bulkAddressUploads)
    .set({ status: "finalising" })
    .where(
      and(
        eq(bulkAddressUploads.id, args.uploadId),
        eq(bulkAddressUploads.status, "awaiting_review"),
      ),
    )
    .returning();
  if (claimed.length === 0) {
    const current = await loadById(args.uploadId);
    if (current?.status === "complete") return { kind: "already_finalized" };
    return { kind: "wrong_state", current: current?.status ?? "unknown" };
  }

  const [subTask] = await db
    .insert(subTasks)
    .values({
      taskId: upload.taskId,
      status: "waiting",
      service: SUBTASK_SERVICE.finaliser,
      inputs: JSON.stringify({ bulk_upload_id: args.uploadId }),
    })
    .returning();

  // v2 (ADR-0006): the finaliser also writes property_overrides for UPRN-matched
  // rows, which needs the classifier CSV (raw descriptions, joined to the
  // combiner output by `source_row_id`) and the confirmed building-part ordering.
  // Both are derivable here — we already hold the upload row, and dispatch runs
  // after the user confirms ordering, so the value is final.
  //   - classifier_s3_uri: null when no classifier columns were mapped (no
  //     classifier CSV was written; the finaliser then writes no overrides).
  //   - multi_entry_ordering: permutations keyed by entry-count; {} when the
  //     upload is not multi-entry (every cell is a single building part → part 0).
  const classifierMap = classifierMapping(upload.columnMapping ?? {});
  const classifierS3Uri =
    Object.keys(classifierMap).length > 0
      ? `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: confirmedS3Uri,
    portfolio_id: Number(upload.portfolioId),
    bulk_upload_id: args.uploadId,
    classifier_s3_uri: classifierS3Uri,
    multi_entry_ordering: upload.multiEntryOrdering?.permutations ?? {},
    // classifier category → source CSV header, so the finaliser knows which
    // classifier-CSV column feeds each override_component (ADR-0006).
    column_mapping: classifierMap,
  };

  const trigger = await triggerFastApiPipeline({
    endpoint: "/v1/bulk-uploads/trigger-finaliser",
    payload,
    sessionToken: args.sessionToken,
  });

  if (!trigger.ok) {
    // Roll the claim back so the user can retry, and fail the subtask.
    await Promise.all([
      db
        .update(bulkAddressUploads)
        .set({ status: "awaiting_review" })
        .where(eq(bulkAddressUploads.id, args.uploadId)),
      db
        .update(subTasks)
        .set({
          status: "failed",
          outputs: JSON.stringify({ error: trigger.message }),
        })
        .where(eq(subTasks.id, subTask.id)),
    ]);
    return { kind: "trigger_failed", status: trigger.status, message: trigger.message };
  }

  await db
    .update(subTasks)
    .set({ status: "in progress", inputs: JSON.stringify(payload) })
    .where(eq(subTasks.id, subTask.id));

  return { kind: "ok", taskId: upload.taskId, subTaskId: subTask.id };
}
