414 lines
14 KiB
Python
414 lines
14 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 UniqueViolation
|
|
from psycopg.rows import dict_row
|
|
from psycopg.types.json import Jsonb
|
|
|
|
from app.config import settings
|
|
from app.repairs.exceptions import (
|
|
ActiveRepairConflict,
|
|
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 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, lifecycle_status
|
|
FROM miners
|
|
WHERE id = %s
|
|
FOR UPDATE
|
|
""",
|
|
(miner_id,),
|
|
)
|
|
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 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]
|