from __future__ import annotations import uuid from datetime import UTC, datetime from sqlalchemy import select from sqlalchemy.orm import Session from app.core.errors import AppError from app.models.booking import Booking from app.models.data_quality import DataQualityIssue from app.models.inspection import Inspection from app.models.vehicle import Vehicle from app.schemas import CurrentUser, ResolveOdometerRegressionRequest from app.services.audit import record_audit_event from app.services.data_quality_common import issue_due_at, load_open_issue, new_scan_ref def _odometer_fingerprint( *, source_type: str, later_ref: str, later_km: int, ) -> dict[str, str | int]: """Stable identity for one reviewed regression, independent of issue IDs.""" return { "source_type": source_type, "later_ref": later_ref, "later_km": later_km, } def _was_odometer_regression_retained( db: Session, *, vehicle_id: uuid.UUID, fingerprint: dict[str, str | int], ) -> bool: reviewed = db.scalars( select(DataQualityIssue).where( DataQualityIssue.rule_type == "odometer_regression", DataQualityIssue.entity_type == "vehicle", DataQualityIssue.entity_id == vehicle_id, DataQualityIssue.status == "resolved", ) ).all() return any( issue.evidence_json.get("resolution_decision") == "retain_canonical" and fingerprint in issue.evidence_json.get("retained_odometer_fingerprints", []) for issue in reviewed ) def open_odometer_regression_issue( db: Session, *, vehicle: Vehicle, reading_ref: str, reading_km: int, canonical_km: int, source_type: str, related_refs: list[str], correctable_booking_refs: list[str] | None = None, canonical_ref: str | None = None, detected_at: datetime | None = None, public_ref: str | None = None, actor_label: str | None = None, actor_type: str = "user", correlation_id: uuid.UUID | None = None, ) -> DataQualityIssue | None: """Open one explainable DQ-03 issue while the caller holds the vehicle lock.""" if reading_km >= canonical_km: return None earlier_ref = canonical_ref or vehicle.public_ref fingerprint = _odometer_fingerprint( source_type=source_type, later_ref=reading_ref, later_km=reading_km, ) # A manager's explicit "retain canonical" decision acknowledges this exact source # fact. Reopening it on every scheduled scan would create churn; only changed or new # evidence (and therefore a different fingerprint) is actionable again. if _was_odometer_regression_retained( db, vehicle_id=vehicle.id, fingerprint=fingerprint, ): return None existing = db.scalar( select(DataQualityIssue) .where( DataQualityIssue.rule_type == "odometer_regression", DataQualityIssue.entity_type == "vehicle", DataQualityIssue.entity_id == vehicle.id, DataQualityIssue.status == "open", ) .with_for_update() .execution_options(populate_existing=True) ) if existing is not None: before_related_refs = list(existing.evidence_json.get("related_refs", [])) before_correctable_refs = list(existing.evidence_json.get("correctable_booking_refs", [])) new_signal = { "code": "odometer.regression", "source_type": source_type, "params": { "later_ref": reading_ref, "later_km": reading_km, "earlier_ref": earlier_ref, "earlier_km": canonical_km, }, } existing_signals = list(existing.evidence_json.get("signals", [])) signal_was_new = not any( isinstance(signal, dict) and isinstance(signal.get("params"), dict) and _odometer_fingerprint( source_type=str( signal.get("source_type", existing.evidence_json.get("source_type")) ), later_ref=str(signal["params"].get("later_ref")), later_km=signal["params"].get("later_km"), ) == fingerprint for signal in existing_signals if isinstance(signal, dict) and isinstance(signal.get("params"), dict) and isinstance(signal["params"].get("later_km"), int) ) if signal_was_new: existing_signals.append(new_signal) merged_related_refs = list( dict.fromkeys([*before_related_refs, *(related_refs if signal_was_new else [])]) ) merged_correctable_refs = list( dict.fromkeys([*before_correctable_refs, *(correctable_booking_refs or [])]) ) if ( not signal_was_new and merged_related_refs == before_related_refs and merged_correctable_refs == before_correctable_refs ): return existing source_types = list(existing.evidence_json.get("source_types", [])) previous_source_type = existing.evidence_json.get("source_type") if ( isinstance(previous_source_type, str) and previous_source_type != "multiple" and previous_source_type not in source_types ): source_types.append(previous_source_type) if source_type not in source_types: source_types.append(source_type) existing.evidence_json = { **existing.evidence_json, "summary": ( f"{source_type.title()} {reading_ref} recorded {reading_km} km, below " f"the canonical {canonical_km} km for {vehicle.public_ref}." ), "related_refs": merged_related_refs, "correctable_booking_refs": merged_correctable_refs, "source_type": source_type if len(source_types) == 1 else "multiple", "source_types": source_types, "signals": existing_signals, } if actor_label is not None and ( merged_related_refs != before_related_refs or merged_correctable_refs != before_correctable_refs or signal_was_new ): record_audit_event( db, actor_type=actor_type, actor_label=actor_label, action="data_quality_issue_evidence_updated", entity_type="data_quality_issue", entity_id=existing.id, correlation_id=correlation_id, before={ "related_refs": before_related_refs, "correctable_booking_refs": before_correctable_refs, }, after={ "related_refs": merged_related_refs, "correctable_booking_refs": merged_correctable_refs, }, metadata={"vehicle_ref": vehicle.public_ref, "reading_ref": reading_ref}, ) return existing now = detected_at or datetime.now(UTC) previous = db.scalar( select(DataQualityIssue) .where( DataQualityIssue.rule_type == "odometer_regression", DataQualityIssue.entity_type == "vehicle", DataQualityIssue.entity_id == vehicle.id, DataQualityIssue.status != "open", ) .order_by(DataQualityIssue.detected_at.desc()) ) evidence = { "summary": ( f"{source_type.title()} {reading_ref} recorded {reading_km} km, below " f"the canonical {canonical_km} km for {vehicle.public_ref}." ), "entity_ref": vehicle.public_ref, "related_refs": related_refs, "correctable_booking_refs": correctable_booking_refs or [], "source_type": source_type, "source_types": [source_type], "signals": [ { "code": "odometer.regression", "source_type": source_type, "params": { "later_ref": reading_ref, "later_km": reading_km, "earlier_ref": earlier_ref, "earlier_km": canonical_km, }, } ], } if previous is not None: evidence["reopened_from"] = previous.public_ref evidence["previous_decision"] = previous.status issue = DataQualityIssue( public_ref=public_ref or new_scan_ref("DQ-ODO"), rule_type="odometer_regression", entity_type="vehicle", entity_id=vehicle.id, severity="medium", status="open", evidence_json=evidence, proposed_action_json={}, detected_at=now, due_at=issue_due_at(now, "medium"), ) db.add(issue) db.flush() if actor_label is not None: record_audit_event( db, actor_type=actor_type, actor_label=actor_label, action="data_quality_issue_created", entity_type="data_quality_issue", entity_id=issue.id, correlation_id=correlation_id, after={"status": "open", "rule_type": "odometer_regression"}, metadata={"vehicle_ref": vehicle.public_ref, "reading_ref": reading_ref}, ) return issue def resolve_odometer_regression( db: Session, public_ref: str, body: ResolveOdometerRegressionRequest, actor: CurrentUser ) -> DataQualityIssue: # Read the routing data without a row lock first. The canonical mutation order is # booking -> vehicle -> issue -> inspection everywhere, matching checkout/return. # Locking the issue before the booking creates a resolver-vs-return deadlock. issue_snapshot = load_open_issue(db, public_ref, lock=False) if issue_snapshot.rule_type != "odometer_regression": raise AppError( "NOT_AN_ODOMETER_ISSUE", "This issue is not an odometer_regression issue.", status_code=409, ) def correctable_refs(issue: DataQualityIssue) -> list[str]: if "correctable_booking_refs" in issue.evidence_json: refs = issue.evidence_json.get("correctable_booking_refs", []) else: # Backward compatibility for issues created before source-aware evidence. refs = [ ref for ref in issue.evidence_json.get("related_refs", []) if isinstance(ref, str) and ref.startswith("BK-") ] return [ref for ref in refs if isinstance(ref, str)] booking: Booking | None = None if body.decision != "retain_canonical": if body.booking_ref not in correctable_refs(issue_snapshot): 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, ) 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 ) vehicle = db.scalar( select(Vehicle).where(Vehicle.id == issue_snapshot.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 ) # A return or scan may have appended evidence while we waited for the domain locks. # Lock and refresh the issue only now, then revalidate every decision against that # current evidence instead of resolving a stale snapshot. issue = load_open_issue(db, public_ref) if issue.rule_type != "odometer_regression" or issue.entity_id != vehicle.id: raise AppError( "ISSUE_CHANGED", "The issue changed while the correction was being prepared. Review it again.", status_code=409, ) if booking is not None: if booking.vehicle_id != vehicle.id or booking.public_ref not in correctable_refs(issue): raise AppError( "INVALID_BOOKING_REFERENCE", "booking_ref must be one of this issue's related bookings.", status_code=422, ) correlation_id = uuid.uuid4() if body.decision == "retain_canonical": retained_fingerprints: list[dict[str, str | int]] = [] fallback_source_type = str(issue.evidence_json.get("source_type", "unknown")) for signal in issue.evidence_json.get("signals", []): if not isinstance(signal, dict) or signal.get("code") != "odometer.regression": continue params = signal.get("params") if not isinstance(params, dict): continue source_type = signal.get("source_type", fallback_source_type) later_ref = params.get("later_ref") later_km = params.get("later_km") earlier_ref = params.get("earlier_ref") earlier_km = params.get("earlier_km") if ( isinstance(source_type, str) and isinstance(later_ref, str) and isinstance(later_km, int) and isinstance(earlier_ref, str) and isinstance(earlier_km, int) ): retained_fingerprints.append( _odometer_fingerprint( source_type=source_type, later_ref=later_ref, later_km=later_km, ) ) issue.evidence_json = { **issue.evidence_json, "resolution_decision": "retain_canonical", "retained_odometer_fingerprints": retained_fingerprints, } 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, "retained_fingerprint_count": len(retained_fingerprints), }, ) else: assert booking is not None and body.corrected_odometer_km is not None # 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, ) # A returned booking has two persisted representations of the same reading. # Correct every return inspection for the selected booking, including historical # seed rows whose old issue evidence did not yet cite the inspection explicitly. related_inspections = db.scalars( select(Inspection) .where(Inspection.booking_id == booking.id, Inspection.type == "return") .with_for_update() ).all() before = { "booking_end_odometer_km": booking.end_odometer_km, "vehicle_odometer_km": vehicle.odometer_km, "inspection_odometer_km": { inspection.public_ref: inspection.odometer_km for inspection in related_inspections }, } booking.end_odometer_km = body.corrected_odometer_km for inspection in related_inspections: inspection.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, "inspection_odometer_km": { inspection.public_ref: inspection.odometer_km for inspection in related_inspections }, }, metadata={"issue_ref": issue.public_ref, "booking_ref": booking.public_ref}, ) has_remaining_evidence = False if booking is not None: target_refs = {booking.public_ref, *(item.public_ref for item in related_inspections)} before_related = [ ref for ref in issue.evidence_json.get("related_refs", []) if isinstance(ref, str) ] before_correctable = correctable_refs(issue) before_signals = [ signal for signal in issue.evidence_json.get("signals", []) if isinstance(signal, dict) ] remaining_signals = [ signal for signal in before_signals if not ( isinstance(signal.get("params"), dict) and signal["params"].get("later_ref") in target_refs ) ] remaining_related = [ref for ref in before_related if ref not in target_refs] remaining_correctable = [ref for ref in before_correctable if ref != booking.public_ref] has_remaining_evidence = bool(remaining_signals or remaining_correctable) issue.evidence_json = { **issue.evidence_json, "summary": ( "Additional odometer regression evidence remains for review." if has_remaining_evidence else issue.evidence_json.get("summary", "Odometer reading corrected.") ), "related_refs": remaining_related, "correctable_booking_refs": remaining_correctable, "signals": remaining_signals, } if has_remaining_evidence: record_audit_event( db, actor_type="user", actor_label=actor.display_name, action="data_quality_issue_partially_resolved", entity_type="data_quality_issue", entity_id=issue.id, correlation_id=correlation_id, before={ "related_refs": before_related, "correctable_booking_refs": before_correctable, "signal_count": len(before_signals), }, after={ "related_refs": remaining_related, "correctable_booking_refs": remaining_correctable, "signal_count": len(remaining_signals), }, metadata={ "decision": body.decision, "booking_ref": booking.public_ref, "note": body.note, }, ) if not has_remaining_evidence: 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