847 lines
32 KiB
Python
847 lines
32 KiB
Python
"""PostgreSQL persistence for repair workflows."""
|
|
|
|
from collections.abc import Callable, Iterator, Mapping
|
|
from contextlib import contextmanager
|
|
from typing import Any
|
|
from uuid import UUID
|
|
|
|
import psycopg
|
|
from psycopg.errors import CheckViolation, LockNotAvailable, UniqueViolation
|
|
from psycopg.rows import dict_row
|
|
from psycopg.types.json import Jsonb
|
|
|
|
from app.config import settings
|
|
from app.repairs.exceptions import (
|
|
ActiveRepairConflict,
|
|
ControlBoardReplacementConflict,
|
|
MacOwnershipConflict,
|
|
RepairPersistenceError,
|
|
)
|
|
|
|
|
|
ConnectionFactory = Callable[[], psycopg.Connection]
|
|
ACTIVE_REPAIR_STATUSES = (
|
|
"OPEN",
|
|
"IN_PROGRESS",
|
|
"WAITING_FOR_PARTS",
|
|
"READY_FOR_TEST",
|
|
)
|
|
ACTIVE_REPAIR_CONSTRAINT = "uq_repairs_one_active_per_miner"
|
|
|
|
|
|
def default_connection_factory() -> psycopg.Connection:
|
|
return psycopg.connect(
|
|
host=settings.postgres_host,
|
|
port=settings.postgres_port,
|
|
dbname=settings.postgres_db,
|
|
user=settings.postgres_user,
|
|
password=settings.postgres_password,
|
|
connect_timeout=5,
|
|
)
|
|
|
|
|
|
class RepairRepository:
|
|
"""Persistence boundary for explicit atomic repair workflows."""
|
|
|
|
def __init__(
|
|
self,
|
|
connection_factory: ConnectionFactory | None = None,
|
|
*,
|
|
connection: psycopg.Connection | None = None,
|
|
):
|
|
self._connection_factory = connection_factory or default_connection_factory
|
|
self._connection = connection
|
|
|
|
@contextmanager
|
|
def transaction(self) -> Iterator["RepairRepository"]:
|
|
if self._connection is not None:
|
|
yield self
|
|
return
|
|
|
|
connection = self._connection_factory()
|
|
transaction_repository = RepairRepository(
|
|
self._connection_factory,
|
|
connection=connection,
|
|
)
|
|
try:
|
|
yield transaction_repository
|
|
connection.commit()
|
|
except UniqueViolation as exc:
|
|
connection.rollback()
|
|
if exc.diag.constraint_name in {
|
|
"uq_miner_mac_ownerships_owner",
|
|
"uq_miner_network_identities_normalized_mac",
|
|
"uq_miners_current_normalized_mac",
|
|
"uq_control_board_installations_current_miner",
|
|
"uq_control_board_installations_current_mac",
|
|
}:
|
|
raise ControlBoardReplacementConflict(
|
|
"Concurrent control-board replacement or MAC ownership conflict"
|
|
) from exc
|
|
raise
|
|
except CheckViolation as exc:
|
|
connection.rollback()
|
|
if exc.diag.constraint_name == "ck_current_control_board_projection":
|
|
raise ControlBoardReplacementConflict(
|
|
"Control-board projection would contradict authoritative history"
|
|
) from exc
|
|
raise
|
|
except Exception:
|
|
connection.rollback()
|
|
raise
|
|
finally:
|
|
connection.close()
|
|
|
|
@contextmanager
|
|
def _connection_scope(
|
|
self,
|
|
*,
|
|
write: bool = False,
|
|
) -> Iterator[psycopg.Connection]:
|
|
if self._connection is not None:
|
|
yield self._connection
|
|
return
|
|
|
|
connection = self._connection_factory()
|
|
try:
|
|
yield connection
|
|
if write:
|
|
connection.commit()
|
|
except Exception:
|
|
if write:
|
|
connection.rollback()
|
|
raise
|
|
finally:
|
|
connection.close()
|
|
|
|
def lock_miner_for_update(self, miner_id: UUID) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, asset_tag, lifecycle_status, current_mac, control_board_sn,
|
|
firmware_version
|
|
FROM miners
|
|
WHERE id = %s
|
|
FOR UPDATE
|
|
""",
|
|
(miner_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def lock_miner_for_control_board_replacement(
|
|
self, miner_id: UUID
|
|
) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
try:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, asset_tag, lifecycle_status, current_mac,
|
|
control_board_sn, firmware_version
|
|
FROM miners WHERE id = %s FOR UPDATE NOWAIT
|
|
""",
|
|
(miner_id,),
|
|
)
|
|
except LockNotAvailable as exc:
|
|
raise ControlBoardReplacementConflict(
|
|
"A control-board replacement is already in progress"
|
|
) from exc
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def get_active_repair(self, miner_id: UUID) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
miner_id,
|
|
status,
|
|
reason,
|
|
opened_at,
|
|
opened_by,
|
|
completed_at,
|
|
completed_by
|
|
FROM repairs
|
|
WHERE miner_id = %s
|
|
AND status = ANY(%s)
|
|
ORDER BY opened_at, id
|
|
LIMIT 1
|
|
""",
|
|
(miner_id, list(ACTIVE_REPAIR_STATUSES)),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def create_repair(
|
|
self,
|
|
miner_id: UUID,
|
|
reason: str,
|
|
opened_by: UUID,
|
|
) -> dict:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
try:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO repairs (miner_id, reason, opened_by)
|
|
VALUES (%s, %s, %s)
|
|
RETURNING
|
|
id,
|
|
miner_id,
|
|
status,
|
|
reason,
|
|
opened_at,
|
|
opened_by,
|
|
completed_at,
|
|
completed_by
|
|
""",
|
|
(miner_id, reason, opened_by),
|
|
)
|
|
except UniqueViolation as exc:
|
|
if exc.diag.constraint_name == ACTIVE_REPAIR_CONSTRAINT:
|
|
raise ActiveRepairConflict(miner_id) from exc
|
|
raise
|
|
return dict(cur.fetchone())
|
|
|
|
def create_event(
|
|
self,
|
|
repair_id: UUID,
|
|
*,
|
|
sequence: int,
|
|
event_type: str,
|
|
created_by: UUID,
|
|
payload: Mapping[str, Any],
|
|
) -> dict:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO repair_events (
|
|
repair_id,
|
|
sequence,
|
|
event_type,
|
|
created_by,
|
|
payload
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s)
|
|
RETURNING
|
|
id,
|
|
repair_id,
|
|
sequence,
|
|
event_type,
|
|
created_at,
|
|
created_by,
|
|
payload
|
|
""",
|
|
(
|
|
repair_id,
|
|
sequence,
|
|
event_type,
|
|
created_by,
|
|
Jsonb(dict(payload)),
|
|
),
|
|
)
|
|
return dict(cur.fetchone())
|
|
|
|
def set_miner_lifecycle(
|
|
self,
|
|
miner_id: UUID,
|
|
lifecycle_status: str,
|
|
) -> None:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
UPDATE miners
|
|
SET lifecycle_status = %s,
|
|
updated_at = now()
|
|
WHERE id = %s
|
|
""",
|
|
(lifecycle_status, miner_id),
|
|
)
|
|
if cur.rowcount != 1:
|
|
raise RepairPersistenceError(
|
|
"Miner lifecycle transition did not update one record"
|
|
)
|
|
|
|
def get_repair_identity(self, repair_id: UUID) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, miner_id
|
|
FROM repairs
|
|
WHERE id = %s
|
|
""",
|
|
(repair_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def lock_repair_for_update(self, repair_id: UUID) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
miner_id,
|
|
status,
|
|
reason,
|
|
opened_at,
|
|
opened_by,
|
|
completed_at,
|
|
completed_by,
|
|
completion_summary,
|
|
resulting_lifecycle
|
|
FROM repairs
|
|
WHERE id = %s
|
|
FOR UPDATE
|
|
""",
|
|
(repair_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def get_next_event_sequence(self, repair_id: UUID) -> int:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT COALESCE(max(sequence), 0) + 1
|
|
FROM repair_events
|
|
WHERE repair_id = %s
|
|
""",
|
|
(repair_id,),
|
|
)
|
|
return int(cur.fetchone()[0])
|
|
|
|
def get_pre_repair_lifecycle(self, repair_id: UUID) -> str | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT event.payload ->> 'from_lifecycle'
|
|
FROM repair_events AS event
|
|
WHERE event.repair_id = %s
|
|
AND event.event_type = 'REPAIR_STARTED'
|
|
ORDER BY event.sequence
|
|
LIMIT 2
|
|
""",
|
|
(repair_id,),
|
|
)
|
|
rows = cur.fetchall()
|
|
|
|
if len(rows) != 1:
|
|
return None
|
|
lifecycle_status = rows[0][0]
|
|
if lifecycle_status not in {"active", "spare"}:
|
|
return None
|
|
return str(lifecycle_status)
|
|
|
|
def complete_repair(
|
|
self,
|
|
repair_id: UUID,
|
|
*,
|
|
completed_by: UUID,
|
|
completion_summary: str | None,
|
|
resulting_lifecycle: str,
|
|
) -> dict | None:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
UPDATE repairs
|
|
SET status = 'COMPLETED',
|
|
completed_at = now(),
|
|
completed_by = %s,
|
|
completion_summary = %s,
|
|
resulting_lifecycle = %s
|
|
WHERE id = %s
|
|
AND status = ANY(%s)
|
|
RETURNING
|
|
id,
|
|
miner_id,
|
|
status,
|
|
reason,
|
|
opened_at,
|
|
opened_by,
|
|
completed_at,
|
|
completed_by,
|
|
completion_summary,
|
|
resulting_lifecycle
|
|
""",
|
|
(
|
|
completed_by,
|
|
completion_summary,
|
|
resulting_lifecycle,
|
|
repair_id,
|
|
list(ACTIVE_REPAIR_STATUSES),
|
|
),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def lock_active_repair_for_miner(self, miner_id: UUID) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT id, miner_id, status, reason, opened_at, opened_by
|
|
FROM repairs
|
|
WHERE miner_id = %s AND status = ANY(%s)
|
|
ORDER BY opened_at, id
|
|
LIMIT 1
|
|
FOR UPDATE
|
|
""",
|
|
(miner_id, list(ACTIVE_REPAIR_STATUSES)),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def lock_current_control_board(self, miner_id: UUID) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT * FROM miner_control_board_installations
|
|
WHERE miner_id = %s AND removed_at IS NULL
|
|
FOR UPDATE
|
|
""",
|
|
(miner_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def create_legacy_control_board(self, miner: Mapping[str, Any]) -> dict:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
mac_address = miner.get("current_mac")
|
|
normalized_mac = (
|
|
"".join(ch for ch in str(mac_address).lower() if ch in "0123456789abcdef")
|
|
if mac_address else None
|
|
)
|
|
if normalized_mac and len(normalized_mac) == 12:
|
|
mac_address = ":".join(
|
|
normalized_mac[index:index + 2].upper()
|
|
for index in range(0, 12, 2)
|
|
)
|
|
elif normalized_mac:
|
|
normalized_mac = None
|
|
mac_address = None
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO miner_control_board_installations (
|
|
miner_id, record_origin, mac_address, normalized_mac,
|
|
board_serial, firmware_version
|
|
)
|
|
VALUES (%s, 'legacy_projection', %s, %s, %s, %s)
|
|
RETURNING *
|
|
""",
|
|
(miner["id"], mac_address, normalized_mac,
|
|
miner.get("control_board_sn"), miner.get("firmware_version")),
|
|
)
|
|
return dict(cur.fetchone())
|
|
|
|
def claim_mac_ownership(
|
|
self, normalized_mac: str, miner_id: UUID, actor_user_id: UUID
|
|
) -> None:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"SELECT pg_advisory_xact_lock(hashtextextended(%s, 0))",
|
|
(normalized_mac,),
|
|
)
|
|
cur.execute(
|
|
"SELECT miner_id FROM miner_mac_ownerships "
|
|
"WHERE normalized_mac = %s FOR UPDATE",
|
|
(normalized_mac,),
|
|
)
|
|
row = cur.fetchone()
|
|
if row and row["miner_id"] != miner_id:
|
|
raise MacOwnershipConflict(normalized_mac)
|
|
if row is None:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO miner_mac_ownerships (
|
|
normalized_mac, miner_id, claimed_at, claimed_by
|
|
) VALUES (%s, %s, now(), %s)
|
|
""",
|
|
(normalized_mac, miner_id, actor_user_id),
|
|
)
|
|
|
|
def close_control_board(
|
|
self, installation_id: UUID, *, repair_id: UUID, actor_user_id: UUID,
|
|
disposition: str
|
|
) -> dict:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
UPDATE miner_control_board_installations
|
|
SET removed_at = now(), removed_by = %s,
|
|
removed_repair_id = %s, removed_disposition = %s
|
|
WHERE id = %s AND removed_at IS NULL
|
|
RETURNING *
|
|
""",
|
|
(actor_user_id, repair_id, disposition, installation_id),
|
|
)
|
|
row = cur.fetchone()
|
|
if row is None:
|
|
raise RepairPersistenceError(
|
|
"Current control-board installation was not closed"
|
|
)
|
|
return dict(row)
|
|
|
|
def create_control_board_installation(
|
|
self, *, miner_id: UUID, repair_id: UUID, mac_address: str,
|
|
normalized_mac: str, board_serial: str | None, board_model: str | None,
|
|
board_revision: str | None, firmware_version: str | None,
|
|
replacement_reason: str, notes: str | None, actor_user_id: UUID
|
|
) -> dict:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO miner_control_board_installations (
|
|
miner_id, record_origin, installed_repair_id, mac_address,
|
|
normalized_mac, board_serial, board_model, board_revision,
|
|
firmware_version, replacement_reason, installed_at,
|
|
installed_by, notes
|
|
) VALUES (
|
|
%s, 'replacement', %s, %s, %s, %s, %s, %s, %s, %s,
|
|
now(), %s, %s
|
|
) RETURNING *
|
|
""",
|
|
(miner_id, repair_id, mac_address, normalized_mac, board_serial,
|
|
board_model, board_revision, firmware_version,
|
|
replacement_reason, actor_user_id, notes),
|
|
)
|
|
return dict(cur.fetchone())
|
|
|
|
def set_current_control_board(
|
|
self, miner_id: UUID, *, mac_address: str, board_serial: str | None
|
|
) -> None:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
UPDATE miners SET current_mac = %s, control_board_sn = %s,
|
|
updated_at = now() WHERE id = %s
|
|
""",
|
|
(mac_address, board_serial, miner_id),
|
|
)
|
|
if cur.rowcount != 1:
|
|
raise RepairPersistenceError(
|
|
"Current miner control-board projection was not updated"
|
|
)
|
|
|
|
def ensure_network_identity_history(
|
|
self, miner_id: UUID, *, mac_address: str, normalized_mac: str
|
|
) -> None:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO miner_network_identities (
|
|
miner_id, mac_address, source, is_current,
|
|
first_seen_at, last_seen_at, associated_at
|
|
)
|
|
SELECT %s, %s, 'legacy_projection', false, NULL, NULL, now()
|
|
WHERE NOT EXISTS (
|
|
SELECT 1 FROM miner_network_identities
|
|
WHERE miner_id = %s
|
|
AND regexp_replace(lower(mac_address), '[^0-9a-f]', '', 'g') = %s
|
|
)
|
|
""",
|
|
(miner_id, mac_address, miner_id, normalized_mac),
|
|
)
|
|
|
|
def set_current_network_identity(
|
|
self, miner_id: UUID, *, mac_address: str, normalized_mac: str
|
|
) -> dict:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"UPDATE miner_network_identities SET is_current = false, "
|
|
"updated_at = now() WHERE miner_id = %s AND is_current",
|
|
(miner_id,),
|
|
)
|
|
cur.execute(
|
|
"""
|
|
UPDATE miner_network_identities
|
|
SET is_current = true, associated_at = now(), updated_at = now()
|
|
WHERE miner_id = %s
|
|
AND regexp_replace(lower(mac_address), '[^0-9a-f]', '', 'g') = %s
|
|
RETURNING *
|
|
""",
|
|
(miner_id, normalized_mac),
|
|
)
|
|
row = cur.fetchone()
|
|
if row is None:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO miner_network_identities (
|
|
miner_id, mac_address, source, is_current,
|
|
first_seen_at, last_seen_at, associated_at
|
|
) VALUES (%s, %s, 'control_board_replacement', true,
|
|
NULL, NULL, now())
|
|
RETURNING *
|
|
""",
|
|
(miner_id, mac_address),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row)
|
|
|
|
def lock_component_for_update(self, component_id: UUID) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
id,
|
|
component_type,
|
|
asset_id,
|
|
manufacturer,
|
|
model,
|
|
manufacturer_serial
|
|
FROM components
|
|
WHERE id = %s
|
|
FOR UPDATE
|
|
""",
|
|
(component_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def create_repair_part(
|
|
self,
|
|
*,
|
|
repair_part_id: UUID,
|
|
repair_id: UUID,
|
|
consumable_description: str | None,
|
|
quantity: object | None,
|
|
unit: str | None,
|
|
component: Mapping[str, Any] | None,
|
|
notes: str | None,
|
|
recorded_event: Mapping[str, Any],
|
|
) -> dict:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO repair_parts (
|
|
id,
|
|
repair_id,
|
|
consumable_description,
|
|
quantity,
|
|
unit,
|
|
component_id,
|
|
component_type_snapshot,
|
|
component_asset_id_snapshot,
|
|
component_manufacturer_snapshot,
|
|
component_model_snapshot,
|
|
component_manufacturer_serial_snapshot,
|
|
notes,
|
|
recorded_by,
|
|
recorded_at,
|
|
recorded_event_id
|
|
)
|
|
VALUES (
|
|
%s, %s, %s, %s, %s, %s, %s, %s,
|
|
%s, %s, %s, %s, %s, %s, %s
|
|
)
|
|
RETURNING *
|
|
""",
|
|
(
|
|
repair_part_id,
|
|
repair_id,
|
|
consumable_description,
|
|
quantity,
|
|
unit,
|
|
component.get("id") if component else None,
|
|
component.get("component_type") if component else None,
|
|
component.get("asset_id") if component else None,
|
|
component.get("manufacturer") if component else None,
|
|
component.get("model") if component else None,
|
|
component.get("manufacturer_serial") if component else None,
|
|
notes,
|
|
recorded_event["created_by"],
|
|
recorded_event["created_at"],
|
|
recorded_event["id"],
|
|
),
|
|
)
|
|
return dict(cur.fetchone())
|
|
|
|
def get_repair_part(
|
|
self,
|
|
repair_id: UUID,
|
|
repair_part_id: UUID,
|
|
) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT part.*, event.sequence AS recorded_event_sequence
|
|
FROM repair_parts AS part
|
|
JOIN repair_events AS event
|
|
ON event.id = part.recorded_event_id
|
|
WHERE part.repair_id = %s AND part.id = %s
|
|
""",
|
|
(repair_id, repair_part_id),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def get_repair_parts(self, repair_id: UUID) -> list[dict] | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
repair.id AS existing_repair_id,
|
|
part.*,
|
|
event.sequence AS recorded_event_sequence
|
|
FROM repairs AS repair
|
|
LEFT JOIN repair_parts AS part
|
|
ON part.repair_id = repair.id
|
|
LEFT JOIN repair_events AS event
|
|
ON event.id = part.recorded_event_id
|
|
WHERE repair.id = %s
|
|
ORDER BY event.sequence ASC, part.id ASC
|
|
""",
|
|
(repair_id,),
|
|
)
|
|
rows = [dict(row) for row in cur.fetchall()]
|
|
|
|
if not rows:
|
|
return None
|
|
return [row for row in rows if row.get("id") is not None]
|
|
|
|
def lock_repair_part_for_update(
|
|
self,
|
|
repair_id: UUID,
|
|
repair_part_id: UUID,
|
|
) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT *
|
|
FROM repair_parts
|
|
WHERE repair_id = %s AND id = %s
|
|
FOR UPDATE
|
|
""",
|
|
(repair_id, repair_part_id),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def void_repair_part(
|
|
self,
|
|
repair_id: UUID,
|
|
repair_part_id: UUID,
|
|
*,
|
|
reason: str,
|
|
void_event: Mapping[str, Any],
|
|
) -> dict | None:
|
|
with self._connection_scope(write=True) as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
UPDATE repair_parts
|
|
SET voided_by = %s,
|
|
voided_at = %s,
|
|
void_reason = %s,
|
|
void_event_id = %s
|
|
WHERE repair_id = %s
|
|
AND id = %s
|
|
AND voided_at IS NULL
|
|
RETURNING *
|
|
""",
|
|
(
|
|
void_event["created_by"],
|
|
void_event["created_at"],
|
|
reason,
|
|
void_event["id"],
|
|
repair_id,
|
|
repair_part_id,
|
|
),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def get_repair_detail(self, repair_id: UUID) -> dict | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
repair.id AS repair_id,
|
|
repair.miner_id,
|
|
miner.asset_tag AS miner_asset_id,
|
|
repair.status,
|
|
repair.reason,
|
|
repair.opened_at,
|
|
opened_actor.id AS opened_by_user_id,
|
|
opened_actor.display_name AS opened_by_display_name,
|
|
repair.completed_at AS closed_at,
|
|
closed_actor.id AS closed_by_user_id,
|
|
closed_actor.display_name AS closed_by_display_name,
|
|
miner.lifecycle_status AS miner_lifecycle,
|
|
(
|
|
SELECT max(event.sequence)
|
|
FROM repair_events AS event
|
|
WHERE event.repair_id = repair.id
|
|
) AS latest_event_sequence
|
|
FROM repairs AS repair
|
|
JOIN miners AS miner ON miner.id = repair.miner_id
|
|
JOIN users AS opened_actor
|
|
ON opened_actor.id = repair.opened_by
|
|
LEFT JOIN users AS closed_actor
|
|
ON closed_actor.id = repair.completed_by
|
|
WHERE repair.id = %s
|
|
""",
|
|
(repair_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
return dict(row) if row else None
|
|
|
|
def get_repair_events(
|
|
self,
|
|
repair_id: UUID,
|
|
) -> list[dict] | None:
|
|
with self._connection_scope() as connection:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
repair.id AS repair_id,
|
|
event.id AS event_id,
|
|
event.sequence,
|
|
event.event_type,
|
|
actor.id AS actor_user_id,
|
|
actor.display_name AS actor_display_name,
|
|
event.created_at AS occurred_at,
|
|
event.payload
|
|
FROM repairs AS repair
|
|
LEFT JOIN (
|
|
repair_events AS event
|
|
JOIN users AS actor
|
|
ON actor.id = event.created_by
|
|
) ON event.repair_id = repair.id
|
|
WHERE repair.id = %s
|
|
ORDER BY event.sequence ASC
|
|
""",
|
|
(repair_id,),
|
|
)
|
|
rows = [dict(row) for row in cur.fetchall()]
|
|
|
|
if not rows:
|
|
return None
|
|
return [row for row in rows if row["event_id"] is not None]
|