from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime

from sqlalchemy import func, select, update
from sqlalchemy.orm import Session

from app.core.errors import AppError
from app.models.booking import Booking
from app.models.customer import Customer
from app.models.data_quality import DataQualityIssue
from app.models.inspection import Inspection
from app.models.maintenance import MaintenanceRecord
from app.models.vehicle import Vehicle
from app.schemas import CurrentUser
from app.services.audit import record_audit_event
from app.services.data_quality_common import has_open_issue as _has_open_issue
from app.services.data_quality_common import issue_due_at
from app.services.data_quality_common import load_open_issue as _load_open_issue
from app.services.data_quality_common import new_scan_ref as _new_scan_ref
from app.services.data_quality_duplicate_scan import scan_duplicate_customers
from app.services.data_quality_odometer import (
    open_odometer_regression_issue as open_odometer_regression_issue,
)
from app.services.data_quality_odometer import (
    resolve_odometer_regression as resolve_odometer_regression,
)
from app.services.vehicle_status import (
    RECOMMENDATION_CODE_NO_CONFLICT,
    VehicleStatusRecommendation,
    compute_recommendation_token,
    evaluate_vehicle_status,
    gather_vehicle_status_facts,
)

REQUIRED_CUSTOMER_FIELDS = ("first_name", "last_name")
REQUIRED_VEHICLE_FIELDS = ("registration_number", "make", "model", "location")
DATA_QUALITY_SCAN_LOCK_ID = 6_138_493_717_091_029_491


@dataclass
class ScanResult:
    created: dict[str, int] = field(default_factory=dict)

    def bump(self, rule_type: str) -> None:
        self.created[rule_type] = self.created.get(rule_type, 0) + 1


def _open_issue(
    db: Session,
    scan: ScanResult,
    *,
    rule_type: str,
    entity_type: str,
    entity_id: uuid.UUID,
    severity: str,
    summary: str,
    entity_ref: str,
    related_refs: list[str],
    signals: list[dict] | None = None,
    evidence_extra: dict | None = None,
) -> None:
    if _has_open_issue(db, rule_type, entity_type, entity_id):
        return
    now = datetime.now(UTC)

    # Reintroduced evidence creates a new issue rather than silently reopening the old
    # one, but it stays linked to whatever decision was made last time so an operator
    # doesn't re-litigate from a blank slate.
    previous = db.scalar(
        select(DataQualityIssue)
        .where(
            DataQualityIssue.rule_type == rule_type,
            DataQualityIssue.entity_type == entity_type,
            DataQualityIssue.entity_id == entity_id,
            DataQualityIssue.status != "open",
        )
        .order_by(DataQualityIssue.detected_at.desc())
    )
    # `summary` is kept as a technical-fallback string (shown only under "Technical
    # details"); `signals` is the stable, localizable structure the frontend renders as
    # the primary evidence -- see docs/fleet-ops-correction/current-gap-audit.md §2/§6.
    evidence: dict = {
        "summary": summary,
        "entity_ref": entity_ref,
        "related_refs": related_refs,
        "signals": signals or [],
    }
    evidence.update(evidence_extra or {})
    if previous is not None:
        evidence["reopened_from"] = previous.public_ref
        evidence["previous_decision"] = previous.status

    issue = DataQualityIssue(
        public_ref=_new_scan_ref("DQ-SCAN"),
        rule_type=rule_type,
        entity_type=entity_type,
        entity_id=entity_id,
        severity=severity,
        status="open",
        evidence_json=evidence,
        proposed_action_json={},
        detected_at=now,
        due_at=issue_due_at(now, severity),
    )
    db.add(issue)
    db.flush()
    scan.bump(rule_type)


def _scan_missing_required_fields(db: Session, scan: ScanResult) -> None:
    # Anonymised customers have had their contact data removed on purpose; flagging
    # them as "missing required field" would only be resolvable by re-entering PII.
    for customer in db.scalars(
        select(Customer).where(
            Customer.merged_into_customer_id.is_(None),
            Customer.anonymized_at.is_(None),
        )
    ).all():
        missing = [f for f in REQUIRED_CUSTOMER_FIELDS if not getattr(customer, f)]
        if not customer.email and not customer.phone:
            missing.append("email_or_phone")
        if missing:
            _open_issue(
                db,
                scan,
                rule_type="missing_required_field",
                entity_type="customer",
                entity_id=customer.id,
                severity="low",
                summary=f"Missing: {', '.join(missing)}",
                entity_ref=customer.public_ref,
                related_refs=[],
                signals=[{"code": "missing_field", "params": {"field": f}} for f in missing],
            )

    for vehicle in db.scalars(select(Vehicle).where(Vehicle.active.is_(True))).all():
        missing = [f for f in REQUIRED_VEHICLE_FIELDS if not getattr(vehicle, f)]
        if missing:
            _open_issue(
                db,
                scan,
                rule_type="missing_required_field",
                entity_type="vehicle",
                entity_id=vehicle.id,
                severity="low",
                summary=f"Missing: {', '.join(missing)}",
                entity_ref=vehicle.public_ref,
                related_refs=[],
                signals=[{"code": "missing_field", "params": {"field": f}} for f in missing],
            )


def _scan_booking_overlaps(db: Session, scan: ScanResult) -> None:
    vehicles = db.scalars(select(Vehicle)).all()
    bookings_by_vehicle: dict[uuid.UUID, list[Booking]] = {}
    for booking in db.scalars(
        select(Booking).where(Booking.status.in_(["reserved", "active"]))
    ).all():
        bookings_by_vehicle.setdefault(booking.vehicle_id, []).append(booking)

    vehicle_by_id = {v.id: v for v in vehicles}
    for vehicle_id, bookings in bookings_by_vehicle.items():
        bookings.sort(key=lambda b: b.starts_at)
        for i, first in enumerate(bookings):
            for second in bookings[i + 1 :]:
                if second.starts_at < first.ends_at and first.starts_at < second.ends_at:
                    vehicle = vehicle_by_id[vehicle_id]
                    _open_issue(
                        db,
                        scan,
                        rule_type="booking_overlap",
                        entity_type="vehicle",
                        entity_id=vehicle_id,
                        severity="high",
                        summary=f"Overlapping bookings {first.public_ref} and {second.public_ref}",
                        entity_ref=vehicle.public_ref,
                        related_refs=[first.public_ref, second.public_ref],
                        signals=[
                            {
                                "code": "overlap.reserved_bookings",
                                "params": {"refs": [first.public_ref, second.public_ref]},
                            }
                        ],
                    )


def _scan_vehicle_status_conflicts(db: Session, scan: ScanResult) -> None:
    # Uses the same shared evaluator as the preview/apply flow (app.services.vehicle_status)
    # so detection and resolution can never structurally disagree -- see
    # docs/fleet-ops-correction/vehicle-status-decision-table.md.
    vehicles = db.scalars(select(Vehicle)).all()
    for vehicle in vehicles:
        facts = gather_vehicle_status_facts(db, vehicle)
        recommendation = evaluate_vehicle_status(vehicle, facts)
        if recommendation.recommendation_code == RECOMMENDATION_CODE_NO_CONFLICT:
            continue

        signals = [{"code": recommendation.recommendation_code, "params": facts.as_dict()}]
        summary = (
            f"Recommended status: {recommendation.recommended_status}"
            if recommendation.recommended_status
            else "Manual review required: active rental conflicts with a blocking condition"
        )
        _open_issue(
            db,
            scan,
            rule_type="vehicle_status_conflict",
            entity_type="vehicle",
            entity_id=vehicle.id,
            severity="high",
            summary=summary,
            entity_ref=vehicle.public_ref,
            related_refs=[
                *facts.active_booking_refs,
                *(ref for pair in facts.overlapping_booking_pairs for ref in pair),
            ],
            signals=signals,
        )


def _scan_odometer_regressions(
    db: Session,
    scan: ScanResult,
    *,
    actor_label: str | None,
    actor_type: str,
) -> None:
    # Compare the chronological history with itself rather than every historical reading
    # to today's canonical value. That detects imported checkout/return/maintenance
    # regressions without flagging every legitimate older reading.
    # Runtime checkout/return/maintenance mutations take the vehicle lock before they
    # inspect or append DQ-03 evidence. Taking the same lock here makes scan-vs-command
    # check/merge atomic and prevents a partial-unique race for the open issue.
    vehicles = {
        v.id: v for v in db.scalars(select(Vehicle).order_by(Vehicle.id).with_for_update()).all()
    }
    readings_by_vehicle: dict[uuid.UUID, list[tuple[datetime, str, int, str, str | None]]] = {}
    inspections = db.scalars(select(Inspection)).all()
    return_inspection_booking_ids = {
        inspection.booking_id for inspection in inspections if inspection.type == "return"
    }
    returned_bookings = db.scalars(
        select(Booking).where(Booking.status == "returned", Booking.end_odometer_km.is_not(None))
    ).all()
    booking_ref_by_id = {booking.id: booking.public_ref for booking in returned_bookings}
    for booking in returned_bookings:
        # A real return inspection owns the actual reading timestamp. Using the booking's
        # planned ends_at as a duplicate second reading can make an early return appear to
        # go backwards after a newer real inspection. Keep booking data only as the legacy
        # import fallback when no return inspection exists.
        if booking.id in return_inspection_booking_ids:
            continue
        assert booking.end_odometer_km is not None
        readings_by_vehicle.setdefault(booking.vehicle_id, []).append(
            (
                booking.ends_at,
                booking.public_ref,
                booking.end_odometer_km,
                "booking",
                booking.public_ref,
            )
        )
    for inspection in inspections:
        readings_by_vehicle.setdefault(inspection.vehicle_id, []).append(
            (
                inspection.completed_at,
                inspection.public_ref,
                inspection.odometer_km,
                inspection.type,
                (
                    booking_ref_by_id.get(inspection.booking_id)
                    if inspection.type == "return"
                    else None
                ),
            )
        )
    for record in db.scalars(select(MaintenanceRecord)).all():
        readings_by_vehicle.setdefault(record.vehicle_id, []).append(
            (record.occurred_at, record.public_ref, record.odometer_km, "maintenance", None)
        )

    for vehicle_id, readings in readings_by_vehicle.items():
        readings.sort(key=lambda reading: (reading[0], reading[1]))
        highest = readings[0] if readings else None
        for later in readings[1:]:
            assert highest is not None
            if later[2] < highest[2]:
                vehicle = vehicles[vehicle_id]
                had_open_issue = _has_open_issue(
                    db,
                    "odometer_regression",
                    "vehicle",
                    vehicle_id,
                )
                issue = open_odometer_regression_issue(
                    db,
                    vehicle=vehicle,
                    reading_ref=later[1],
                    reading_km=later[2],
                    canonical_km=highest[2],
                    canonical_ref=highest[1],
                    source_type=later[3],
                    related_refs=list(
                        dict.fromkeys(
                            [highest[1], later[1], *([later[4]] if later[4] is not None else [])]
                        )
                    ),
                    correctable_booking_refs=[later[4]] if later[4] is not None else [],
                    public_ref=_new_scan_ref("DQ-SCAN"),
                    actor_label=actor_label,
                    actor_type=actor_type,
                )
                if issue is not None and not had_open_issue:
                    scan.bump("odometer_regression")
            if later[2] > highest[2]:
                highest = later


def run_scan(
    db: Session,
    *,
    actor_label: str | None = None,
    actor_type: str = "user",
    commit: bool = True,
) -> ScanResult:
    # The check-then-insert work below spans several rules. Serialise whole scans at the
    # database boundary so API and n8n triggers cannot both observe an empty condition.
    db.scalar(select(func.pg_advisory_xact_lock(DATA_QUALITY_SCAN_LOCK_ID)))
    scan = ScanResult()
    scan_duplicate_customers(db, scan, _open_issue)
    _scan_missing_required_fields(db, scan)
    _scan_odometer_regressions(
        db,
        scan,
        actor_label=actor_label,
        actor_type=actor_type,
    )
    _scan_booking_overlaps(db, scan)
    _scan_vehicle_status_conflicts(db, scan)
    if actor_label is not None:
        record_audit_event(
            db,
            actor_type=actor_type,
            actor_label=actor_label,
            action="data_quality_scan_run",
            entity_type="system",
            metadata={"created": scan.created},
        )
    if commit:
        db.commit()
    else:
        db.flush()
    return scan


def defer_issue(db: Session, public_ref: str, actor: CurrentUser) -> DataQualityIssue:
    issue = _load_open_issue(db, public_ref)
    issue.status = "deferred"
    issue.resolved_at = datetime.now(UTC)
    issue.resolved_by = actor.display_name
    record_audit_event(
        db,
        actor_type="user",
        actor_label=actor.display_name,
        action="data_quality_issue_deferred",
        entity_type="data_quality_issue",
        entity_id=issue.id,
        before={"status": "open"},
        after={"status": "deferred"},
    )
    db.commit()
    return issue


def reject_issue(db: Session, public_ref: str, actor: CurrentUser) -> DataQualityIssue:
    issue = _load_open_issue(db, public_ref)
    issue.status = "rejected"
    issue.resolved_at = datetime.now(UTC)
    issue.resolved_by = actor.display_name
    record_audit_event(
        db,
        actor_type="user",
        actor_label=actor.display_name,
        action="data_quality_issue_rejected",
        entity_type="data_quality_issue",
        entity_id=issue.id,
        before={"status": "open"},
        after={"status": "rejected"},
    )
    db.commit()
    return issue


def provide_missing_fields(
    db: Session, public_ref: str, fields: dict[str, str], actor: CurrentUser
) -> DataQualityIssue:
    issue = _load_open_issue(db, public_ref)
    if issue.rule_type != "missing_required_field":
        raise AppError(
            "NOT_A_MISSING_FIELD_ISSUE",
            "This issue is not a missing-required-field issue.",
            status_code=409,
        )

    entity: Customer | Vehicle | None
    if issue.entity_type == "customer":
        entity = db.get(Customer, issue.entity_id)
        allowed = {*REQUIRED_CUSTOMER_FIELDS, "email", "phone"}
    elif issue.entity_type == "vehicle":
        entity = db.get(Vehicle, issue.entity_id)
        allowed = set(REQUIRED_VEHICLE_FIELDS)
    else:
        raise AppError(
            "UNSUPPORTED_ENTITY",
            f"Cannot provide fields for entity type '{issue.entity_type}'.",
            status_code=409,
        )
    if entity is None:
        raise AppError(
            "ENTITY_NOT_FOUND", "The underlying record could not be found.", status_code=404
        )

    invalid = set(fields) - allowed
    if invalid:
        raise AppError(
            "INVALID_FIELD",
            f"Fields not permitted here: {', '.join(sorted(invalid))}.",
            status_code=422,
        )
    if not fields:
        raise AppError(
            "NO_FIELDS_PROVIDED", "At least one field must be provided.", status_code=422
        )

    before = {f: getattr(entity, f) for f in allowed}
    for field_name, value in fields.items():
        if not value.strip():
            raise AppError("EMPTY_VALUE", f"Field '{field_name}' cannot be blank.", status_code=422)
        setattr(entity, field_name, value.strip())
    after = {f: getattr(entity, f) for f in allowed}

    correlation_id = uuid.uuid4()
    record_audit_event(
        db,
        actor_type="user",
        actor_label=actor.display_name,
        action="data_quality_fields_provided",
        entity_type=issue.entity_type,
        entity_id=entity.id,
        correlation_id=correlation_id,
        before=before,
        after=after,
        metadata={"issue_ref": issue.public_ref},
    )

    if isinstance(entity, Customer):
        missing = [f for f in REQUIRED_CUSTOMER_FIELDS if not getattr(entity, f)]
        if not entity.email and not entity.phone:
            missing.append("email_or_phone")
    else:
        missing = [f for f in REQUIRED_VEHICLE_FIELDS if not getattr(entity, f)]

    if not missing:
        issue.status = "resolved"
        issue.resolved_at = datetime.now(UTC)
        issue.resolved_by = actor.display_name
        record_audit_event(
            db,
            actor_type="user",
            actor_label=actor.display_name,
            action="data_quality_issue_resolved",
            entity_type="data_quality_issue",
            entity_id=issue.id,
            correlation_id=correlation_id,
            before={"status": "open"},
            after={"status": "resolved"},
        )
    else:
        issue.evidence_json = {**issue.evidence_json, "summary": f"Missing: {', '.join(missing)}"}

    db.commit()
    return issue


def resolve_booking_overlap(
    db: Session, public_ref: str, booking_ref: str, note: str | None, actor: CurrentUser
) -> DataQualityIssue:
    issue = _load_open_issue(db, public_ref)
    if issue.rule_type != "booking_overlap":
        raise AppError(
            "NOT_AN_OVERLAP_ISSUE", "This issue is not a booking_overlap issue.", status_code=409
        )
    related_refs = issue.evidence_json.get("related_refs", [])
    if booking_ref not in related_refs:
        raise AppError(
            "INVALID_BOOKING_REFERENCE",
            "booking_ref must be one of this issue's overlapping bookings.",
            status_code=422,
        )
    booking = db.scalar(select(Booking).where(Booking.public_ref == booking_ref).with_for_update())
    if booking is None:
        raise AppError("BOOKING_NOT_FOUND", "The booking to block was not found.", status_code=404)
    if booking.status not in ("reserved", "active"):
        raise AppError(
            "BOOKING_NOT_ACTIVE",
            f"Booking is '{booking.status}'; only a reserved or active booking can be blocked.",
            status_code=409,
        )

    before = {"status": booking.status}
    booking.status = "blocked"

    # Verify the minimal safe resolution actually removed the conflict: no two
    # reserved/active bookings for this vehicle should still overlap. The session has
    # autoflush disabled, so exclude the just-blocked booking by id rather than relying
    # on the in-memory status change being visible to this query.
    remaining = db.scalars(
        select(Booking).where(
            Booking.vehicle_id == booking.vehicle_id,
            Booking.status.in_(["reserved", "active"]),
            Booking.public_ref.in_(related_refs),
            Booking.id != booking.id,
        )
    ).all()
    for i, first in enumerate(remaining):
        for second in remaining[i + 1 :]:
            if second.starts_at < first.ends_at and first.starts_at < second.ends_at:
                raise AppError(
                    "OVERLAP_STILL_PRESENT",
                    "Blocking this booking did not remove the overlap; another commitment remains.",
                    status_code=409,
                )

    correlation_id = uuid.uuid4()
    record_audit_event(
        db,
        actor_type="user",
        actor_label=actor.display_name,
        action="data_quality_booking_blocked",
        entity_type="booking",
        entity_id=booking.id,
        correlation_id=correlation_id,
        before=before,
        after={"status": booking.status},
        metadata={"issue_ref": issue.public_ref, "note": note},
    )

    issue.status = "resolved"
    issue.resolved_at = datetime.now(UTC)
    issue.resolved_by = actor.display_name
    record_audit_event(
        db,
        actor_type="user",
        actor_label=actor.display_name,
        action="data_quality_issue_resolved",
        entity_type="data_quality_issue",
        entity_id=issue.id,
        correlation_id=correlation_id,
        before={"status": "open"},
        after={"status": "resolved"},
    )
    db.commit()
    return issue


def _load_vehicle_status_conflict_issue(
    db: Session, public_ref: str, *, lock: bool = True
) -> DataQualityIssue:
    issue = _load_open_issue(db, public_ref, lock=lock)
    if issue.rule_type != "vehicle_status_conflict":
        raise AppError(
            "NOT_A_STATUS_CONFLICT_ISSUE",
            "This issue is not a vehicle_status_conflict issue.",
            status_code=409,
        )
    return issue


def preview_vehicle_status_recommendation(
    db: Session, public_ref: str
) -> tuple[DataQualityIssue, Vehicle, VehicleStatusRecommendation, str]:
    """Non-mutating: computes and returns the recommendation only. Never resolves the
    issue, never writes an audit event, never queues automation -- safe to call as often
    as the UI needs (e.g. every time the panel is opened) with zero side effects."""
    issue = _load_vehicle_status_conflict_issue(db, public_ref, lock=False)
    vehicle = db.scalar(select(Vehicle).where(Vehicle.id == issue.entity_id))
    if vehicle is None:
        raise AppError(
            "VEHICLE_NOT_FOUND", "The vehicle for this issue was not found.", status_code=404
        )
    facts = gather_vehicle_status_facts(db, vehicle, exclude_issue_id=issue.id)
    recommendation = evaluate_vehicle_status(vehicle, facts)
    token = compute_recommendation_token(vehicle, facts)
    return issue, vehicle, recommendation, token


def apply_recommended_status(
    db: Session, public_ref: str, actor: CurrentUser, expected_token: str
) -> tuple[DataQualityIssue, str, str]:
    issue = _load_vehicle_status_conflict_issue(db, public_ref)
    # Lock the vehicle row for the remainder of this transaction so a concurrent apply
    # (or return/checkout) can't race between our fact-gathering and the write below.
    vehicle = db.scalar(select(Vehicle).where(Vehicle.id == issue.entity_id).with_for_update())
    if vehicle is None:
        raise AppError(
            "VEHICLE_NOT_FOUND", "The vehicle for this issue was not found.", status_code=404
        )

    facts = gather_vehicle_status_facts(db, vehicle, exclude_issue_id=issue.id)
    recommendation = evaluate_vehicle_status(vehicle, facts)
    current_token = compute_recommendation_token(vehicle, facts)

    if current_token != expected_token:
        raise AppError(
            "RECOMMENDATION_STALE",
            "The underlying facts changed since this recommendation was shown; "
            "review the recommendation again before applying it.",
            status_code=409,
        )
    if recommendation.manual_review_required or not recommendation.safe_to_apply:
        raise AppError(
            "MANUAL_REVIEW_REQUIRED",
            "This vehicle's state requires manual review; no automatic status change is safe.",
            status_code=409,
        )
    if recommendation.recommended_status is None:
        raise AppError(
            "NO_CONFLICT_DETECTED",
            "The current vehicle state no longer conflicts; nothing to apply.",
            status_code=409,
        )
    new_status = recommendation.recommended_status
    reason_code = recommendation.recommendation_code

    before = {"operational_status": vehicle.operational_status}
    vehicle.operational_status = new_status
    vehicle.version += 1

    # Re-validate against the same shared evaluator, over freshly-gathered facts, that
    # applying this change actually leaves no conflict -- never trust the pre-computed
    # recommendation alone for the post-condition.
    post_facts = gather_vehicle_status_facts(db, vehicle, exclude_issue_id=issue.id)
    post_check = evaluate_vehicle_status(vehicle, post_facts)
    if post_check.recommendation_code not in (RECOMMENDATION_CODE_NO_CONFLICT,):
        raise AppError(
            "CONFLICT_STILL_PRESENT",
            "Applying the recommended status did not resolve the conflict.",
            status_code=409,
        )

    correlation_id = uuid.uuid4()
    record_audit_event(
        db,
        actor_type="user",
        actor_label=actor.display_name,
        action="data_quality_status_applied",
        entity_type="vehicle",
        entity_id=vehicle.id,
        correlation_id=correlation_id,
        before=before,
        after={"operational_status": vehicle.operational_status},
        metadata={"issue_ref": issue.public_ref, "reason_code": reason_code},
    )

    issue.status = "resolved"
    issue.resolved_at = datetime.now(UTC)
    issue.resolved_by = actor.display_name
    record_audit_event(
        db,
        actor_type="user",
        actor_label=actor.display_name,
        action="data_quality_issue_resolved",
        entity_type="data_quality_issue",
        entity_id=issue.id,
        correlation_id=correlation_id,
        before={"status": "open"},
        after={"status": "resolved"},
    )
    db.commit()
    return issue, new_status, reason_code


MERGEABLE_FIELDS = ("first_name", "last_name", "email", "phone", "postal_code", "city")
# Mirrors the column lengths in app/models/customer.py so an override can never fail with
# a database DataError (500) instead of a validation error.
_MERGEABLE_FIELD_MAX_LENGTH = {
    "first_name": 80,
    "last_name": 80,
    "email": 200,
    "phone": 40,
    "postal_code": 20,
    "city": 120,
}


def merge_customers(
    db: Session,
    public_ref: str,
    survivor_ref: str,
    field_overrides: dict[str, str] | None,
    actor: CurrentUser,
) -> dict:
    issue = _load_open_issue(db, public_ref)
    if issue.rule_type != "possible_duplicate_customer":
        raise AppError(
            "NOT_A_DUPLICATE_ISSUE",
            "This issue is not a possible-duplicate-customer issue.",
            status_code=409,
        )

    entity_ref = issue.evidence_json.get("entity_ref")
    related_refs = issue.evidence_json.get("related_refs", [])
    candidate_refs = {entity_ref, *related_refs}
    if survivor_ref not in candidate_refs:
        raise AppError(
            "INVALID_SURVIVOR",
            "The survivor reference must be one of the two customers in this issue.",
            status_code=422,
            details={"candidates": sorted(candidate_refs)},
        )
    loser_ref = next(ref for ref in candidate_refs if ref != survivor_ref)

    # Lock both rows in a deterministic order (by public_ref) so two concurrent merges
    # touching the same customers serialise instead of deadlocking or double-merging.
    survivor = None
    loser = None
    for ref in sorted((survivor_ref, loser_ref)):
        customer = db.scalar(select(Customer).where(Customer.public_ref == ref).with_for_update())
        if ref == survivor_ref:
            survivor = customer
        else:
            loser = customer
    if survivor is None or loser is None:
        raise AppError(
            "CUSTOMER_NOT_FOUND", "One of the customers could not be found.", status_code=404
        )
    if survivor.merged_into_customer_id is not None or loser.merged_into_customer_id is not None:
        raise AppError(
            "CUSTOMER_ALREADY_MERGED",
            "One of the customers has already been merged into another record.",
            status_code=409,
        )

    before = {
        "survivor": {f: getattr(survivor, f) for f in MERGEABLE_FIELDS},
        "loser": {f: getattr(loser, f) for f in MERGEABLE_FIELDS},
    }

    for field_name, value in (field_overrides or {}).items():
        if field_name not in MERGEABLE_FIELDS:
            raise AppError(
                "INVALID_FIELD_OVERRIDE", f"Field '{field_name}' cannot be merged.", status_code=422
            )
        cleaned = value.strip() if isinstance(value, str) else value
        max_length = _MERGEABLE_FIELD_MAX_LENGTH[field_name]
        if not cleaned or len(cleaned) > max_length:
            raise AppError(
                "INVALID_FIELD_OVERRIDE",
                f"Field '{field_name}' must be 1 to {max_length} characters.",
                status_code=422,
            )
        setattr(survivor, field_name, cleaned)

    rewired = db.execute(
        update(Booking).where(Booking.customer_id == loser.id).values(customer_id=survivor.id)
    )
    rewired_count: int = rewired.rowcount  # type: ignore[attr-defined]
    loser.merged_into_customer_id = survivor.id

    issue.status = "resolved"
    issue.resolved_at = datetime.now(UTC)
    issue.resolved_by = actor.display_name

    record_audit_event(
        db,
        actor_type="user",
        actor_label=actor.display_name,
        action="customer_merged",
        entity_type="customer",
        entity_id=survivor.id,
        before=before,
        after={"survivor": {f: getattr(survivor, f) for f in MERGEABLE_FIELDS}},
        metadata={
            "loser_ref": loser_ref,
            "survivor_ref": survivor_ref,
            "rewired_bookings": rewired_count,
        },
    )
    db.commit()

    return {
        "issue_ref": issue.public_ref,
        "survivor_ref": survivor_ref,
        "loser_ref": loser_ref,
        "rewired_bookings": rewired_count,
    }