MobilityOps/backend/app/services/data_quality.py

from __future__ import annotations

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

from sqlalchemy import 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.vehicle import Vehicle
from app.schemas import CurrentUser, ResolveOdometerRegressionRequest
from app.services.audit import record_audit_event

REQUIRED_CUSTOMER_FIELDS = ("first_name", "last_name")
REQUIRED_VEHICLE_FIELDS = ("registration_number", "make", "model", "location")
DUPLICATE_THRESHOLD = 70


@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 _normalize(value: str | None) -> str:
    return (value or "").strip().lower()


def _has_open_issue(db: Session, rule_type: str, entity_type: str, entity_id: uuid.UUID) -> bool:
    return (
        db.scalar(
            select(DataQualityIssue.id).where(
                DataQualityIssue.rule_type == rule_type,
                DataQualityIssue.entity_type == entity_type,
                DataQualityIssue.entity_id == entity_id,
                DataQualityIssue.status == "open",
            )
        )
        is not None
    )


def _next_public_ref(db: Session, prefix: str) -> str:
    existing = db.execute(select(DataQualityIssue.public_ref)).scalars().all()
    numbers = [
        int(ref.rsplit("-", 1)[-1])
        for ref in existing
        if ref.startswith(f"{prefix}-") and ref.rsplit("-", 1)[-1].isdigit()
    ]
    next_number = (max(numbers) + 1) if numbers else 1
    return f"{prefix}-{next_number:04d}"


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],
) -> 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())
    )
    evidence: dict = {
        "summary": summary,
        "entity_ref": entity_ref,
        "related_refs": related_refs,
    }
    if previous is not None:
        evidence["reopened_from"] = previous.public_ref
        evidence["previous_decision"] = previous.status

    issue = DataQualityIssue(
        public_ref=_next_public_ref(db, "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,
    )
    db.add(issue)
    db.flush()
    scan.bump(rule_type)


def _scan_duplicate_customers(db: Session, scan: ScanResult) -> None:
    customers = list(
        db.scalars(select(Customer).where(Customer.merged_into_customer_id.is_(None))).all()
    )
    customers.sort(key=lambda c: c.public_ref)

    for i, a in enumerate(customers):
        for b in customers[i + 1 :]:
            score = 0
            signals = []
            if _normalize(a.email) and _normalize(a.email) == _normalize(b.email):
                score += 60
                signals.append("exact email")
            if _normalize(a.phone) and _normalize(a.phone) == _normalize(b.phone):
                score += 50
                signals.append("exact phone")
            if _normalize(a.postal_code) and _normalize(a.postal_code) == _normalize(b.postal_code):
                score += 10
                signals.append("exact postal code")
            name_a = f"{_normalize(a.first_name)} {_normalize(a.last_name)}"
            name_b = f"{_normalize(b.first_name)} {_normalize(b.last_name)}"
            ratio = SequenceMatcher(None, name_a, name_b).ratio()
            if ratio >= 0.5:
                score += round(ratio * 30)
                signals.append("similar name")

            if score >= DUPLICATE_THRESHOLD:
                _open_issue(
                    db,
                    scan,
                    rule_type="possible_duplicate_customer",
                    entity_type="customer",
                    entity_id=a.id,
                    severity="high",
                    summary="; ".join(signals) + f" (score {score})",
                    entity_ref=a.public_ref,
                    related_refs=[b.public_ref],
                )


def _scan_missing_required_fields(db: Session, scan: ScanResult) -> None:
    for customer in db.scalars(
        select(Customer).where(Customer.merged_into_customer_id.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=[],
            )

    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=[],
            )


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],
                    )


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

    open_high_by_vehicle = {
        row[0]
        for row in db.execute(
            select(DataQualityIssue.entity_id).where(
                DataQualityIssue.entity_type == "vehicle",
                DataQualityIssue.status == "open",
                DataQualityIssue.severity == "high",
            )
        ).all()
    }

    for vehicle in vehicles:
        has_active_booking = vehicle.id in active_by_vehicle
        reason = None
        if vehicle.operational_status == "available" and has_active_booking:
            reason = "marked available while an active booking exists"
        elif vehicle.operational_status == "rented" and not has_active_booking:
            reason = "marked rented without an active booking"
        elif vehicle.operational_status == "available" and vehicle.id in open_high_by_vehicle:
            reason = "marked available while a high-severity quality issue is open"
        elif vehicle.operational_status == "maintenance" and has_active_booking:
            reason = "marked maintenance while an active booking exists"

        if reason:
            _open_issue(
                db,
                scan,
                rule_type="vehicle_status_conflict",
                entity_type="vehicle",
                entity_id=vehicle.id,
                severity="high",
                summary=f"Vehicle {reason}",
                entity_ref=vehicle.public_ref,
                related_refs=[],
            )


def _scan_odometer_regressions(db: Session, scan: ScanResult) -> None:
    # The seed dataset's vehicle.odometer_km is generated independently of booking
    # history, so comparing every historical booking against it produces near-universal
    # false positives. Instead check the booking sequence's own internal consistency:
    # each vehicle's completed bookings should show a non-decreasing odometer reading.
    vehicles = {v.id: v for v in db.scalars(select(Vehicle)).all()}
    bookings_by_vehicle: dict[uuid.UUID, list[Booking]] = {}
    for booking in db.scalars(
        select(Booking).where(
            Booking.status == "returned", Booking.end_odometer_km.is_not(None)
        )
    ).all():
        bookings_by_vehicle.setdefault(booking.vehicle_id, []).append(booking)

    for vehicle_id, bookings in bookings_by_vehicle.items():
        bookings.sort(key=lambda b: b.ends_at)
        for earlier, later in zip(bookings, bookings[1:], strict=False):
            # The query above filters end_odometer_km IS NOT NULL, so both are ints here.
            assert earlier.end_odometer_km is not None
            assert later.end_odometer_km is not None
            if later.end_odometer_km < earlier.end_odometer_km:
                vehicle = vehicles[vehicle_id]
                _open_issue(
                    db,
                    scan,
                    rule_type="odometer_regression",
                    entity_type="vehicle",
                    entity_id=vehicle_id,
                    severity="medium",
                    summary=(
                        f"Booking {later.public_ref} recorded {later.end_odometer_km} km, "
                        f"below the {earlier.end_odometer_km} km recorded by earlier "
                        f"booking {earlier.public_ref}."
                    ),
                    entity_ref=vehicle.public_ref,
                    related_refs=[earlier.public_ref, later.public_ref],
                )
                break


def run_scan(db: Session, *, actor: CurrentUser | None = None) -> ScanResult:
    scan = ScanResult()
    _scan_duplicate_customers(db, scan)
    _scan_missing_required_fields(db, scan)
    _scan_odometer_regressions(db, scan)
    _scan_booking_overlaps(db, scan)
    _scan_vehicle_status_conflicts(db, scan)
    if actor is not None:
        record_audit_event(
            db,
            actor_type="user",
            actor_label=actor.display_name,
            action="data_quality_scan_run",
            entity_type="system",
            metadata={"created": scan.created},
        )
    db.commit()
    return scan


def _load_open_issue(db: Session, public_ref: str) -> DataQualityIssue:
    issue = db.scalar(
        select(DataQualityIssue).where(DataQualityIssue.public_ref == public_ref)
    )
    if issue is None:
        raise AppError("ISSUE_NOT_FOUND", "Data quality issue not found.", status_code=404)
    if issue.status != "open":
        raise AppError(
            "ISSUE_NOT_OPEN",
            f"Issue is '{issue.status}', not 'open'.",
            status_code=409,
        )
    return issue


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_odometer_regression(
    db: Session, public_ref: str, body: ResolveOdometerRegressionRequest, actor: CurrentUser
) -> DataQualityIssue:
    issue = _load_open_issue(db, public_ref)
    if issue.rule_type != "odometer_regression":
        raise AppError(
            "NOT_AN_ODOMETER_ISSUE",
            "This issue is not an odometer_regression issue.",
            status_code=409,
        )
    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
        )

    correlation_id = uuid.uuid4()

    if body.decision == "retain_canonical":
        record_audit_event(
            db,
            actor_type="user",
            actor_label=actor.display_name,
            action="data_quality_odometer_retained",
            entity_type="vehicle",
            entity_id=vehicle.id,
            correlation_id=correlation_id,
            metadata={"issue_ref": issue.public_ref, "canonical_odometer_km": vehicle.odometer_km},
        )
    else:
        related_refs = issue.evidence_json.get("related_refs", [])
        if body.booking_ref not in related_refs:
            raise AppError(
                "INVALID_BOOKING_REFERENCE",
                "booking_ref must be one of this issue's related bookings.",
                status_code=422,
            )
        if body.corrected_odometer_km is None:
            raise AppError(
                "CORRECTED_VALUE_REQUIRED",
                "corrected_odometer_km is required when correcting a reading.",
                status_code=422,
            )
        # Never silently lower the canonical odometer: a correction must be at or above
        # the current canonical value, otherwise it would just create a new regression.
        if body.corrected_odometer_km < vehicle.odometer_km:
            raise AppError(
                "CORRECTION_BELOW_CANONICAL",
                (
                    f"Corrected value {body.corrected_odometer_km} km is still below the "
                    f"canonical {vehicle.odometer_km} km; it would not resolve the regression."
                ),
                status_code=422,
            )
        booking = db.scalar(
            select(Booking).where(Booking.public_ref == body.booking_ref).with_for_update()
        )
        if booking is None:
            raise AppError(
                "BOOKING_NOT_FOUND", "The booking to correct was not found.", status_code=404
            )

        before = {
            "booking_end_odometer_km": booking.end_odometer_km,
            "vehicle_odometer_km": vehicle.odometer_km,
        }
        booking.end_odometer_km = body.corrected_odometer_km
        vehicle.odometer_km = body.corrected_odometer_km
        vehicle.version += 1
        record_audit_event(
            db,
            actor_type="user",
            actor_label=actor.display_name,
            action="data_quality_odometer_corrected",
            entity_type="vehicle",
            entity_id=vehicle.id,
            correlation_id=correlation_id,
            before=before,
            after={
                "booking_end_odometer_km": booking.end_odometer_km,
                "vehicle_odometer_km": vehicle.odometer_km,
            },
            metadata={"issue_ref": issue.public_ref, "booking_ref": booking.public_ref},
        )

    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"},
        metadata={"decision": body.decision, "note": body.note},
    )
    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 _recommend_vehicle_status(
    operational_status: str, has_active_booking: bool, has_open_high_issue: bool
) -> tuple[str, str] | None:
    """The single authoritative recommendation function for vehicle_status_conflict,
    mirroring the exact conditions `_scan_vehicle_status_conflicts` flags."""
    if operational_status == "available" and has_active_booking:
        return "rented", "An active booking exists; the vehicle should be marked rented."
    if operational_status == "rented" and not has_active_booking:
        return "available", "No active booking exists; the vehicle should be marked available."
    if operational_status == "available" and has_open_high_issue:
        return "blocked", "A high-severity quality issue is open; the vehicle should be blocked."
    if operational_status == "maintenance" and has_active_booking:
        return (
            "rented",
            "An active booking exists despite the maintenance status; it should be rented.",
        )
    return None


def apply_recommended_status(
    db: Session, public_ref: str, actor: CurrentUser
) -> tuple[DataQualityIssue, str, str]:
    issue = _load_open_issue(db, public_ref)
    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,
        )
    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
        )

    has_active_booking = (
        db.scalar(
            select(Booking.id).where(Booking.vehicle_id == vehicle.id, Booking.status == "active")
        )
        is not None
    )
    has_open_high_issue = (
        db.scalar(
            select(DataQualityIssue.id).where(
                DataQualityIssue.entity_type == "vehicle",
                DataQualityIssue.entity_id == vehicle.id,
                DataQualityIssue.status == "open",
                DataQualityIssue.severity == "high",
                DataQualityIssue.id != issue.id,
            )
        )
        is not None
    )
    recommendation = _recommend_vehicle_status(
        vehicle.operational_status, has_active_booking, has_open_high_issue
    )
    if recommendation is None:
        raise AppError(
            "NO_CONFLICT_DETECTED",
            "The current vehicle state no longer conflicts; nothing to apply.",
            status_code=409,
        )
    new_status, reason = recommendation

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

    # Re-validate: the same recommendation function must find no further conflict.
    if _recommend_vehicle_status(new_status, has_active_booking, has_open_high_issue) is not None:
        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": reason},
    )

    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


MERGEABLE_FIELDS = ("first_name", "last_name", "email", "phone", "postal_code", "city")


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)

    survivor = db.scalar(select(Customer).where(Customer.public_ref == survivor_ref))
    loser = db.scalar(select(Customer).where(Customer.public_ref == loser_ref))
    if survivor is None or loser is None:
        raise AppError(
            "CUSTOMER_NOT_FOUND", "One of the customers could not be found.", status_code=404
        )

    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
            )
        setattr(survivor, field_name, value)

    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,
    }