from __future__ import annotations import uuid from dataclasses import dataclass from datetime import UTC, datetime, timedelta from sqlalchemy import select from sqlalchemy.exc import IntegrityError 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.idempotency import IdempotencyRecord from app.models.inspection import Inspection from app.models.outbox import OutboxEvent from app.models.vehicle import Vehicle from app.schemas import CurrentUser, RegisterReturnRequest from app.services.audit import record_audit_event def _new_inspection_ref() -> str: """Generate a collision-resistant public reference without reading mutable counts. Return commands for different bookings can commit concurrently. A count-based reference made those independent transactions race for the same unique value. """ return f"INSP-{uuid.uuid4().hex[:10].upper()}" def _derive_vehicle_status_with_reason( body: RegisterReturnRequest, vehicle: Vehicle, new_odometer: int ) -> tuple[str, str, dict[str, str | int]]: # Stable, localizable codes + params -- the backend never emits prose here. The # frontend renders review.reasonCodes. in the selected locale; the mirrored # raw-English fallback strings live only in status_reason (shown under "Technical # details") for backward compatibility. See docs/fleet-ops-correction/i18n-inventory.md. if body.damage_reported and body.technical_warning: return ( "blocked", "returnBlockedDamageAndTechnical", {}, ) if body.damage_reported: return "blocked", "returnBlockedDamage", {} if body.technical_warning: return "blocked", "returnBlockedTechnicalWarning", {} if new_odometer >= vehicle.next_service_km: return ( "maintenance", "returnServiceThresholdReached", {"threshold_km": vehicle.next_service_km}, ) return "cleaning", "returnRoutedToCleaning", {} _STATUS_REASON_FALLBACK_TEXT: dict[str, str] = { "returnBlockedDamageAndTechnical": ( "Damage and a technical warning were both reported on return." ), "returnBlockedDamage": "Damage was reported on return.", "returnBlockedTechnicalWarning": "A technical warning was reported on return.", "returnServiceThresholdReached": "Odometer reached the service threshold.", "returnRoutedToCleaning": ( "No damage, technical warning or service threshold; routed to cleaning." ), } @dataclass class ReturnEvaluation: canonical_odometer_km: int submitted_odometer_km: int odometer_regression: bool resulting_odometer_km: int resulting_vehicle_status: str status_reason: str status_reason_code: str status_reason_params: dict[str, str | int] would_create_quality_issue: bool attention_reasons: list[str] next_booking_risk: dict | None def evaluate_return( db: Session, booking: Booking, vehicle: Vehicle, body: RegisterReturnRequest, *, now: datetime ) -> ReturnEvaluation: """Pure evaluation of what a return would do. No writes; safe to call from a non-mutating preview endpoint. `register_vehicle_return` uses the same function so preview and commit can never drift apart.""" odometer_regression = body.end_odometer_km < vehicle.odometer_km resulting_odometer_km = vehicle.odometer_km if odometer_regression else body.end_odometer_km resulting_status, status_reason_code, status_reason_params = _derive_vehicle_status_with_reason( body, vehicle, resulting_odometer_km ) status_reason = _STATUS_REASON_FALLBACK_TEXT[status_reason_code] attention_reasons = [] if body.damage_reported: attention_reasons.append("damage_reported") if body.technical_warning: attention_reasons.append("technical_warning") if odometer_regression: attention_reasons.append("odometer_regression") next_booking = db.scalar( select(Booking) .where( Booking.vehicle_id == vehicle.id, Booking.status == "reserved", Booking.starts_at > now, ) .order_by(Booking.starts_at.asc()) ) next_booking_risk = None if next_booking is not None: hours_until = (next_booking.starts_at - now).total_seconds() / 3600 next_booking_risk = { "booking_ref": next_booking.public_ref, "starts_at": next_booking.starts_at.isoformat(), "at_risk": resulting_status != "cleaning" or hours_until < 4, } return ReturnEvaluation( canonical_odometer_km=vehicle.odometer_km, submitted_odometer_km=body.end_odometer_km, odometer_regression=odometer_regression, resulting_odometer_km=resulting_odometer_km, resulting_vehicle_status=resulting_status, status_reason=status_reason, status_reason_code=status_reason_code, status_reason_params=status_reason_params, would_create_quality_issue=odometer_regression, attention_reasons=attention_reasons, next_booking_risk=next_booking_risk, ) def _load_active_booking_and_vehicle( db: Session, booking_ref: str, *, lock: bool ) -> tuple[Booking, Vehicle]: stmt = select(Booking).where(Booking.public_ref == booking_ref) if lock: stmt = stmt.with_for_update() booking = db.scalar(stmt) if booking is None: raise AppError("BOOKING_NOT_FOUND", "Booking not found.", status_code=404) vehicle_stmt = select(Vehicle).where(Vehicle.id == booking.vehicle_id) if lock: vehicle_stmt = vehicle_stmt.with_for_update() vehicle = db.scalar(vehicle_stmt) if vehicle is None: raise AppError( "VEHICLE_NOT_FOUND", "The vehicle for this booking could not be found.", status_code=404 ) return booking, vehicle def preview_vehicle_return( db: Session, booking_ref: str, body: RegisterReturnRequest ) -> tuple[Booking, Vehicle, ReturnEvaluation]: booking, vehicle = _load_active_booking_and_vehicle(db, booking_ref, lock=False) if booking.status != "active": raise AppError( "INVALID_BOOKING_STATE", f"Booking is '{booking.status}', not 'active'; it cannot be returned.", status_code=409, ) evaluation = evaluate_return(db, booking, vehicle, body, now=datetime.now(UTC)) return booking, vehicle, evaluation def register_vehicle_return( db: Session, booking_ref: str, body: RegisterReturnRequest, idempotency_key: str, actor: CurrentUser, ) -> tuple[int, dict]: existing = db.scalar( select(IdempotencyRecord).where(IdempotencyRecord.idempotency_key == idempotency_key) ) if existing is not None: booking = db.get(Booking, existing.booking_id) if booking is None or booking.public_ref != booking_ref: raise AppError( "IDEMPOTENCY_KEY_REUSED", "This idempotency key was already used for a different booking.", status_code=409, ) return existing.response_status, existing.response_body booking, vehicle = _load_active_booking_and_vehicle(db, booking_ref, lock=True) # Re-check after acquiring the row lock: a concurrent identical-key request may have # just committed while we were waiting. existing = db.scalar( select(IdempotencyRecord).where(IdempotencyRecord.idempotency_key == idempotency_key) ) if existing is not None: return existing.response_status, existing.response_body if booking.status != "active": raise AppError( "INVALID_BOOKING_STATE", f"Booking is '{booking.status}', not 'active'; it cannot be returned.", status_code=409, ) now = datetime.now(UTC) correlation_id = uuid.uuid4() evaluation = evaluate_return(db, booking, vehicle, body, now=now) inspection = Inspection( public_ref=_new_inspection_ref(), booking_id=booking.id, vehicle_id=vehicle.id, type="return", fuel_level_percent=body.fuel_level_percent, cleanliness_ok=body.cleanliness_ok, damage_reported=body.damage_reported, technical_warning=body.technical_warning, notes=body.notes, odometer_km=body.end_odometer_km, completed_at=now, completed_by=actor.display_name, ) db.add(inspection) before_vehicle = { "operational_status": vehicle.operational_status, "odometer_km": vehicle.odometer_km, } booking.status = "returned" booking.end_odometer_km = body.end_odometer_km quality_issue_ref: str | None = None if evaluation.odometer_regression: issue = DataQualityIssue( public_ref=f"DQ-RET-{str(inspection.public_ref).split('-')[-1]}", rule_type="odometer_regression", entity_type="vehicle", entity_id=vehicle.id, severity="medium", status="open", evidence_json={ "summary": ( f"Return submitted {body.end_odometer_km} km, below canonical " f"{evaluation.canonical_odometer_km} km." ), "entity_ref": vehicle.public_ref, "related_refs": [booking.public_ref, inspection.public_ref], }, proposed_action_json={}, detected_at=now, due_at=now + timedelta(days=1), ) db.add(issue) db.flush() quality_issue_ref = issue.public_ref resulting_status = evaluation.resulting_vehicle_status vehicle.odometer_km = evaluation.resulting_odometer_km vehicle.operational_status = resulting_status vehicle.version += 1 record_audit_event( db, actor_type="user", actor_label=actor.display_name, action="return_registered", entity_type="booking", entity_id=booking.id, correlation_id=correlation_id, before={"status": "active"}, after={"status": "returned", "end_odometer_km": body.end_odometer_km}, metadata={"idempotency_key": idempotency_key}, ) record_audit_event( db, actor_type="user", actor_label=actor.display_name, action="vehicle_status_changed", entity_type="vehicle", entity_id=vehicle.id, correlation_id=correlation_id, before=before_vehicle, after={ "operational_status": vehicle.operational_status, "odometer_km": vehicle.odometer_km, }, ) event = OutboxEvent( event_id=uuid.uuid4(), event_type="vehicle.returned.v1", aggregate_type="booking", aggregate_id=booking.id, payload_json={ "event_type": "vehicle.returned.v1", "correlation_id": str(correlation_id), "aggregate": { "type": "booking", "id": str(booking.id), "public_ref": booking.public_ref, }, "data": { "vehicle_ref": vehicle.public_ref, "inspection_ref": inspection.public_ref, "resulting_vehicle_status": resulting_status, "attention_reasons": evaluation.attention_reasons, }, "aggregate_ref": booking.public_ref, }, occurred_at=now, delivery_status="pending", attempts=0, ) db.add(event) response_body = { "booking_ref": booking.public_ref, "vehicle_ref": vehicle.public_ref, "inspection_ref": inspection.public_ref, "resulting_vehicle_status": resulting_status, "odometer_regression": evaluation.odometer_regression, "quality_issue_ref": quality_issue_ref, "workflow_event_id": str(event.event_id), "correlation_id": str(correlation_id), "next_booking_risk": evaluation.next_booking_risk, } db.add( IdempotencyRecord( idempotency_key=idempotency_key, booking_id=booking.id, response_status=201, response_body=response_body, ) ) try: db.commit() except IntegrityError: db.rollback() existing = db.scalar( select(IdempotencyRecord).where(IdempotencyRecord.idempotency_key == idempotency_key) ) if existing is not None: return existing.response_status, existing.response_body raise return 201, response_body