from __future__ import annotations import uuid from datetime import UTC, datetime 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 REF_PREFIX = "INSP" def _next_public_ref(db: Session) -> str: existing = db.execute(select(Inspection.public_ref)).scalars().all() return f"{REF_PREFIX}-{len(existing) + 1:04d}" def _derive_vehicle_status(body: RegisterReturnRequest, vehicle: Vehicle, new_odometer: int) -> str: if body.damage_reported or body.technical_warning: return "blocked" if new_odometer >= vehicle.next_service_km: return "maintenance" return "cleaning" 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 = db.scalar(select(Booking).where(Booking.public_ref == booking_ref).with_for_update()) if booking is None: raise AppError("BOOKING_NOT_FOUND", "Booking not found.", status_code=404) vehicle = db.scalar(select(Vehicle).where(Vehicle.id == booking.vehicle_id).with_for_update()) # 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() inspection = Inspection( public_ref=_next_public_ref(db), 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 odometer_regression = body.end_odometer_km < vehicle.odometer_km quality_issue_ref: str | None = None canonical_odometer = vehicle.odometer_km if not odometer_regression: canonical_odometer = body.end_odometer_km vehicle.odometer_km = canonical_odometer else: 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"{vehicle.odometer_km} km." ), "entity_ref": vehicle.public_ref, "related_refs": [booking.public_ref, inspection.public_ref], }, proposed_action_json={}, detected_at=now, ) db.add(issue) db.flush() quality_issue_ref = issue.public_ref resulting_status = _derive_vehicle_status(body, vehicle, canonical_odometer) 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, }, ) 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") 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": attention_reasons, }, "aggregate_ref": booking.public_ref, }, occurred_at=now, delivery_status="pending", attempts=0, ) db.add(event) 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, } response_body = { "booking_ref": booking.public_ref, "vehicle_ref": vehicle.public_ref, "inspection_ref": inspection.public_ref, "resulting_vehicle_status": resulting_status, "odometer_regression": odometer_regression, "quality_issue_ref": quality_issue_ref, "workflow_event_id": str(event.event_id), "next_booking_risk": 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