"""Business rules and orchestration for repair workflows.""" from decimal import Decimal from uuid import UUID, uuid4 import re from app.repairs.exceptions import ( ActiveRepairConflict, ControlBoardReplacementActorRequired, ControlBoardReplacementConflict, InvalidMacAddress, IncompatibleMinerLifecycle, InconsistentRepairState, RepairActiveMismatch, RepairActorRequired, RepairCompletionActorRequired, RepairMinerNotFound, RepairMinerNotInRepair, RepairNotActive, RepairNotFound, RepairPersistenceError, RepairPartActorRequired, RepairPartAlreadyVoided, RepairPartComponentNotFound, RepairPartNotFound, RepairPartValidationError, ) from app.repairs.models import ( BeginRepairRequest, ControlBoardReplacedPayload, ControlBoardReplacementRequest, ControlBoardReplacementResult, BeginRepairResult, InitialRepairNotePayload, RepairActor, RepairCompleteRequest, RepairCompletedPayload, RepairPartComponentSnapshot, RepairPartConsumable, RepairPartDetail, RepairPartList, RepairPartRecordRequest, RepairPartVoidRequest, RepairDetail, RepairTimeline, RepairTimelineEvent, RepairStartedPayload, ) from app.repairs.repository import RepairRepository ELIGIBLE_LIFECYCLE_STATUSES = frozenset({"active", "spare"}) ACTIVE_REPAIR_STATUSES = frozenset( {"OPEN", "IN_PROGRESS", "WAITING_FOR_PARTS", "READY_FOR_TEST"} ) REPAIR_LIFECYCLE_STATUS = "in_repair" class RepairService: """Explicit operational repair workflows independent of FastAPI.""" def __init__(self, repository: RepairRepository): self.repository = repository @staticmethod def normalize_note(note: str | None) -> str | None: if note is None: return None normalized = str(note).strip() return normalized or None @staticmethod def _map_repair_detail(repair: dict) -> RepairDetail: closed_by = None if repair["closed_by_user_id"] is not None: closed_by = RepairActor( user_id=repair["closed_by_user_id"], display_name=repair["closed_by_display_name"], ) return RepairDetail( repair_id=repair["repair_id"], miner_id=repair["miner_id"], miner_asset_id=repair["miner_asset_id"], status=repair["status"], reason=repair["reason"], opened_at=repair["opened_at"], opened_by=RepairActor( user_id=repair["opened_by_user_id"], display_name=repair["opened_by_display_name"], ), closed_at=repair["closed_at"], closed_by=closed_by, miner_lifecycle=repair["miner_lifecycle"], latest_event_sequence=repair["latest_event_sequence"], ) def begin_repair( self, request: BeginRepairRequest, *, actor_user_id: UUID | None, ) -> BeginRepairResult: if actor_user_id is None: raise RepairActorRequired() note = self.normalize_note(request.notes) with self.repository.transaction() as repository: miner = repository.lock_miner_for_update(request.miner_id) if not miner: raise RepairMinerNotFound(request.miner_id) lifecycle_status = str(miner["lifecycle_status"]) active_repair = repository.get_active_repair(request.miner_id) if active_repair: raise ActiveRepairConflict(request.miner_id) if lifecycle_status == REPAIR_LIFECYCLE_STATUS: raise InconsistentRepairState(request.miner_id) if lifecycle_status not in ELIGIBLE_LIFECYCLE_STATUSES: raise IncompatibleMinerLifecycle( request.miner_id, lifecycle_status, ) repair = repository.create_repair( request.miner_id, request.reason, actor_user_id, ) started_payload = RepairStartedPayload( reason=request.reason, from_lifecycle=lifecycle_status, ) repository.create_event( repair["id"], sequence=1, event_type="REPAIR_STARTED", created_by=actor_user_id, payload=started_payload.model_dump(mode="json"), ) if note is not None: note_payload = InitialRepairNotePayload(note=note) repository.create_event( repair["id"], sequence=2, event_type="INITIAL_NOTE", created_by=actor_user_id, payload=note_payload.model_dump(mode="json"), ) repository.set_miner_lifecycle( request.miner_id, REPAIR_LIFECYCLE_STATUS, ) return BeginRepairResult( repair_id=repair["id"], miner_id=repair["miner_id"], status=repair["status"], reason=repair["reason"], opened_at=repair["opened_at"], opened_by=repair["opened_by"], ) @staticmethod def normalize_optional_text(value: str | None) -> str | None: if value is None: return None normalized = str(value).strip() return normalized or None @staticmethod def normalize_mac_address(value: str) -> tuple[str, str]: raw = str(value).strip() if not raw or re.fullmatch(r"[0-9A-Fa-f:\-.\s]+", raw) is None: raise InvalidMacAddress() normalized = re.sub(r"[^0-9A-Fa-f]", "", raw).lower() if len(normalized) != 12: raise InvalidMacAddress() first_octet = int(normalized[:2], 16) if normalized == "000000000000": raise InvalidMacAddress("All-zero MAC addresses are not allowed") if normalized == "ffffffffffff": raise InvalidMacAddress("Broadcast MAC addresses are not allowed") if first_octet & 1: raise InvalidMacAddress("Multicast MAC addresses are not allowed") canonical = ":".join( normalized[index:index + 2].upper() for index in range(0, 12, 2) ) return canonical, normalized def replace_control_board( self, miner_id: UUID, request: ControlBoardReplacementRequest, *, actor_user_id: UUID | None ) -> ControlBoardReplacementResult: if actor_user_id is None: raise ControlBoardReplacementActorRequired() new_mac, new_mac_normalized = self.normalize_mac_address( request.new_mac_address ) new_serial = self.normalize_optional_text( request.new_control_board_serial ) board_model = self.normalize_optional_text(request.board_model) board_revision = self.normalize_optional_text(request.board_revision) firmware_version = self.normalize_optional_text(request.firmware_version) notes = self.normalize_note(request.notes) with self.repository.transaction() as repository: miner = repository.lock_miner_for_control_board_replacement(miner_id) if miner is None: raise RepairMinerNotFound(miner_id) if str(miner["lifecycle_status"]) != REPAIR_LIFECYCLE_STATUS: raise RepairMinerNotInRepair( miner_id, miner["lifecycle_status"] ) repair = repository.lock_active_repair_for_miner(miner_id) if repair is None: raise ControlBoardReplacementConflict( "Miner has no active repair" ) current_board = repository.lock_current_control_board(miner_id) if current_board is None: if not miner.get("current_mac") and not miner.get("control_board_sn"): raise ControlBoardReplacementConflict( "Miner has no current control-board identity to replace" ) current_board = repository.create_legacy_control_board(miner) old_mac = current_board.get("mac_address") or miner.get("current_mac") old_mac_canonical = None old_mac_normalized = None if old_mac: try: old_mac_canonical, old_mac_normalized = ( self.normalize_mac_address(str(old_mac)) ) except InvalidMacAddress: old_mac_canonical = str(old_mac) if old_mac_normalized == new_mac_normalized: raise ControlBoardReplacementConflict( "New MAC address is the same as the current MAC address" ) if old_mac_normalized is not None: repository.ensure_network_identity_history( miner_id, mac_address=old_mac_canonical, normalized_mac=old_mac_normalized ) repository.claim_mac_ownership( new_mac_normalized, miner_id, actor_user_id ) previous_board = repository.close_control_board( current_board["id"], repair_id=repair["id"], actor_user_id=actor_user_id, disposition=request.removed_board_disposition, ) installation = repository.create_control_board_installation( miner_id=miner_id, repair_id=repair["id"], mac_address=new_mac, normalized_mac=new_mac_normalized, board_serial=new_serial, board_model=board_model, board_revision=board_revision, firmware_version=firmware_version, replacement_reason=request.replacement_reason, notes=notes, actor_user_id=actor_user_id, ) repository.set_current_network_identity( miner_id, mac_address=new_mac, normalized_mac=new_mac_normalized ) repository.set_current_control_board( miner_id, mac_address=new_mac, board_serial=new_serial ) next_sequence = repository.get_next_event_sequence(repair["id"]) payload = ControlBoardReplacedPayload( old_mac_address=old_mac_canonical, new_mac_address=new_mac, old_control_board_serial=previous_board.get("board_serial"), new_control_board_serial=new_serial, old_board_model=previous_board.get("board_model"), new_board_model=board_model, old_board_revision=previous_board.get("board_revision"), new_board_revision=board_revision, new_firmware_version=firmware_version, removed_board_disposition=request.removed_board_disposition, replacement_reason=request.replacement_reason, notes=notes, ) event = repository.create_event( repair["id"], sequence=next_sequence, event_type="CONTROL_BOARD_REPLACED", created_by=actor_user_id, payload=payload.model_dump(mode="json", exclude_none=False), ) return ControlBoardReplacementResult( miner_id=miner_id, asset_id=miner.get("asset_tag"), repair_id=repair["id"], installation_id=installation["id"], event_id=event["id"], event_sequence=event["sequence"], current_mac=new_mac, control_board_serial=new_serial, replaced_at=event["created_at"], replaced_by=actor_user_id, ) def complete_repair( self, repair_id: UUID, request: RepairCompleteRequest, *, actor_user_id: UUID | None, ) -> RepairDetail: if actor_user_id is None: raise RepairCompletionActorRequired() completion_summary = self.normalize_note(request.completion_summary) with self.repository.transaction() as repository: repair_identity = repository.get_repair_identity(repair_id) if repair_identity is None: raise RepairNotFound(repair_id) miner_id = repair_identity["miner_id"] # Begin Repair also locks the miner first. Keeping this order makes # competing lifecycle workflows serialize on the durable asset. miner = repository.lock_miner_for_update(miner_id) if miner is None: raise RepairMinerNotFound(miner_id) repair = repository.lock_repair_for_update(repair_id) if repair is None: raise RepairNotFound(repair_id) if repair["miner_id"] != miner_id: raise RepairActiveMismatch(repair_id, miner_id) if repair["status"] not in ACTIVE_REPAIR_STATUSES: raise RepairNotActive(repair_id, repair["status"]) active_repair = repository.get_active_repair(miner_id) if active_repair is None or active_repair["id"] != repair_id: raise RepairActiveMismatch(repair_id, miner_id) lifecycle_status = str(miner["lifecycle_status"]) if lifecycle_status != REPAIR_LIFECYCLE_STATUS: raise RepairMinerNotInRepair(miner_id, lifecycle_status) next_sequence = repository.get_next_event_sequence(repair_id) pre_repair_lifecycle = repository.get_pre_repair_lifecycle( repair_id ) completed = repository.complete_repair( repair_id, completed_by=actor_user_id, completion_summary=completion_summary, resulting_lifecycle=request.target_lifecycle, ) if completed is None: raise RepairNotActive(repair_id, repair["status"]) repository.set_miner_lifecycle( miner_id, request.target_lifecycle, ) payload = RepairCompletedPayload( resulting_miner_lifecycle=request.target_lifecycle, pre_repair_lifecycle=pre_repair_lifecycle, completion_summary=completion_summary, ) repository.create_event( repair_id, sequence=next_sequence, event_type="REPAIR_COMPLETED", created_by=actor_user_id, payload=payload.model_dump(mode="json", exclude_none=True), ) detail = repository.get_repair_detail(repair_id) if detail is None: raise RepairPersistenceError( "Completed repair could not be read in its transaction" ) return self._map_repair_detail(detail) @staticmethod def _normalize_repair_part_text( value: str | None, *, field_name: str, max_length: int, ) -> str | None: if value is None: return None normalized = str(value).strip() if not normalized: raise RepairPartValidationError( f"{field_name} must not be blank when supplied" ) if len(normalized) > max_length: raise RepairPartValidationError( f"{field_name} must not exceed {max_length} characters" ) return normalized @staticmethod def _validate_consumable_quantity(quantity: Decimal, unit: str) -> Decimal: normalized = Decimal(quantity) if normalized <= 0: raise RepairPartValidationError("quantity must be greater than zero") significant = normalized.normalize() fractional_places = max(-significant.as_tuple().exponent, 0) if fractional_places > 3 or normalized >= Decimal("1000000000"): raise RepairPartValidationError( "quantity must fit NUMERIC(12,3)" ) if unit == "each" and normalized != normalized.to_integral_value(): raise RepairPartValidationError( "quantity for unit 'each' must be a whole number" ) return normalized @staticmethod def _map_repair_part(part: dict) -> RepairPartDetail: if part["consumable_description"] is not None: consumable = RepairPartConsumable( description=part["consumable_description"], quantity=part["quantity"], unit=part["unit"], ) component_snapshot = None entry_kind = "consumable" else: consumable = None component_snapshot = RepairPartComponentSnapshot( component_type=part["component_type_snapshot"], asset_id=part["component_asset_id_snapshot"], manufacturer=part["component_manufacturer_snapshot"], model=part["component_model_snapshot"], manufacturer_serial=( part["component_manufacturer_serial_snapshot"] ), ) entry_kind = "tracked_component" return RepairPartDetail( repair_part_id=part["id"], repair_id=part["repair_id"], entry_kind=entry_kind, consumable=consumable, component_id=part["component_id"], component_snapshot=component_snapshot, notes=part["notes"], recorded_by=part["recorded_by"], recorded_at=part["recorded_at"], recorded_event_id=part["recorded_event_id"], voided_by=part["voided_by"], voided_at=part["voided_at"], void_reason=part["void_reason"], void_event_id=part["void_event_id"], ) @staticmethod def _lock_active_repair( repository: RepairRepository, repair_id: UUID, ) -> tuple[dict, dict]: identity = repository.get_repair_identity(repair_id) if identity is None: raise RepairNotFound(repair_id) miner_id = identity["miner_id"] miner = repository.lock_miner_for_update(miner_id) if miner is None: raise RepairMinerNotFound(miner_id) repair = repository.lock_repair_for_update(repair_id) if repair is None: raise RepairNotFound(repair_id) if repair["miner_id"] != miner_id: raise RepairActiveMismatch(repair_id, miner_id) if repair["status"] not in ACTIVE_REPAIR_STATUSES: raise RepairNotActive(repair_id, repair["status"]) active_repair = repository.get_active_repair(miner_id) if active_repair is None or active_repair["id"] != repair_id: raise RepairActiveMismatch(repair_id, miner_id) lifecycle_status = str(miner["lifecycle_status"]) if lifecycle_status != REPAIR_LIFECYCLE_STATUS: raise RepairMinerNotInRepair(miner_id, lifecycle_status) return miner, repair def record_repair_part( self, repair_id: UUID, request: RepairPartRecordRequest, *, actor_user_id: UUID | None, ) -> RepairPartDetail: if actor_user_id is None: raise RepairPartActorRequired() notes = self._normalize_repair_part_text( request.notes, field_name="notes", max_length=4000, ) description = None quantity = None unit = None if request.consumable is not None: description = self._normalize_repair_part_text( request.consumable.description, field_name="consumable.description", max_length=255, ) quantity = self._validate_consumable_quantity( request.consumable.quantity, request.consumable.unit, ) unit = request.consumable.unit with self.repository.transaction() as repository: self._lock_active_repair(repository, repair_id) component = None if request.tracked_component_uuid is not None: component = repository.lock_component_for_update( request.tracked_component_uuid ) if component is None: raise RepairPartComponentNotFound( request.tracked_component_uuid ) repair_part_id = uuid4() next_sequence = repository.get_next_event_sequence(repair_id) if component is None: payload = { "repair_part_id": str(repair_part_id), "entry_kind": "consumable", "description": description, "quantity": format(quantity, "f"), "unit": unit, "notes": notes, } else: payload = { "repair_part_id": str(repair_part_id), "entry_kind": "tracked_component", "component_id": str(component["id"]), "component_type": component["component_type"], "component_asset_id": component["asset_id"], "component_manufacturer": component["manufacturer"], "component_model": component["model"], "component_manufacturer_serial": ( component["manufacturer_serial"] ), "notes": notes, } event = repository.create_event( repair_id, sequence=next_sequence, event_type="REPAIR_PART_RECORDED", created_by=actor_user_id, payload=payload, ) repository.create_repair_part( repair_part_id=repair_part_id, repair_id=repair_id, consumable_description=description, quantity=quantity, unit=unit, component=component, notes=notes, recorded_event=event, ) part = repository.get_repair_part(repair_id, repair_part_id) if part is None: raise RepairPersistenceError( "Recorded repair part could not be read in its transaction" ) return self._map_repair_part(part) def list_repair_parts(self, repair_id: UUID) -> RepairPartList: parts = self.repository.get_repair_parts(repair_id) if parts is None: raise RepairNotFound(repair_id) return RepairPartList( repair_id=repair_id, entries=[self._map_repair_part(part) for part in parts], ) def void_repair_part( self, repair_id: UUID, repair_part_id: UUID, request: RepairPartVoidRequest, *, actor_user_id: UUID | None, ) -> RepairPartDetail: if actor_user_id is None: raise RepairPartActorRequired() reason = self._normalize_repair_part_text( request.reason, field_name="reason", max_length=1000, ) assert reason is not None with self.repository.transaction() as repository: self._lock_active_repair(repository, repair_id) part = repository.lock_repair_part_for_update( repair_id, repair_part_id, ) if part is None: raise RepairPartNotFound(repair_part_id) if part["voided_at"] is not None: raise RepairPartAlreadyVoided(repair_part_id) next_sequence = repository.get_next_event_sequence(repair_id) event = repository.create_event( repair_id, sequence=next_sequence, event_type="REPAIR_PART_VOIDED", created_by=actor_user_id, payload={ "repair_part_id": str(repair_part_id), "reason": reason, }, ) voided = repository.void_repair_part( repair_id, repair_part_id, reason=reason, void_event=event, ) if voided is None: raise RepairPartAlreadyVoided(repair_part_id) part = repository.get_repair_part(repair_id, repair_part_id) if part is None: raise RepairPersistenceError( "Voided repair part could not be read in its transaction" ) return self._map_repair_part(part) def get_repair(self, repair_id: UUID) -> RepairDetail: repair = self.repository.get_repair_detail(repair_id) if repair is None: raise RepairNotFound(repair_id) return self._map_repair_detail(repair) def get_repair_timeline(self, repair_id: UUID) -> RepairTimeline: events = self.repository.get_repair_events(repair_id) if events is None: raise RepairNotFound(repair_id) return RepairTimeline( repair_id=repair_id, events=[ RepairTimelineEvent( event_id=event["event_id"], sequence=event["sequence"], event_type=event["event_type"], actor=RepairActor( user_id=event["actor_user_id"], display_name=event["actor_display_name"], ), occurred_at=event["occurred_at"], payload=event["payload"], ) for event in events ], )