feat: add control board replacement workflow

backend/app/repairs/service.py

@@ -2,8 +2,13 @@
 
 from uuid import UUID
 
+import re
+
 from app.repairs.exceptions import (
     ActiveRepairConflict,
+    ControlBoardReplacementActorRequired,
+    ControlBoardReplacementConflict,
+    InvalidMacAddress,
     IncompatibleMinerLifecycle,
     InconsistentRepairState,
     RepairActiveMismatch,
@@ -17,6 +22,9 @@ from app.repairs.exceptions import (
 )
 from app.repairs.models import (
     BeginRepairRequest,
+    ControlBoardReplacedPayload,
+    ControlBoardReplacementRequest,
+    ControlBoardReplacementResult,
     BeginRepairResult,
     InitialRepairNotePayload,
     RepairActor,
@@ -143,6 +151,155 @@ class RepairService:
             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,