367 lines
19 KiB
Python
367 lines
19 KiB
Python
"""Set-based PostgreSQL read model for the Repair Center."""
|
|
|
|
from collections.abc import Callable, Mapping
|
|
from typing import Any
|
|
|
|
import psycopg
|
|
from psycopg.rows import dict_row
|
|
|
|
from app.repairs.repository import default_connection_factory
|
|
|
|
|
|
ConnectionFactory = Callable[[], psycopg.Connection]
|
|
|
|
|
|
class RepairCenterRepository:
|
|
"""Read-only persistence boundary for consolidated repair operations."""
|
|
|
|
def __init__(self, connection_factory: ConnectionFactory | None = None):
|
|
self._connection_factory = connection_factory or default_connection_factory
|
|
|
|
@staticmethod
|
|
def _ctes(*, show_miners: bool, show_locations: bool) -> str:
|
|
miner_fields = (
|
|
"miner.asset_tag AS asset_id, COALESCE(miner.model, miner.miner_type) AS model, miner.miner_sn, miner.power_sn, "
|
|
"miner.current_mac, host(miner.current_ip) AS current_ip, "
|
|
"miner.status AS runtime_status"
|
|
if show_miners else
|
|
"NULL::text AS asset_id, NULL::text AS model, NULL::text AS miner_sn, NULL::text AS power_sn, "
|
|
"NULL::text AS current_mac, NULL::text AS current_ip, "
|
|
"NULL::text AS runtime_status"
|
|
)
|
|
location_fields = (
|
|
"placement.site_id, placement.site_name, placement.room_id, "
|
|
"placement.room_name, placement.rack_label, placement.shelf_label, "
|
|
"placement.position_id, placement.position_label, placement.position_index"
|
|
if show_locations else
|
|
"NULL::uuid AS site_id, NULL::text AS site_name, NULL::uuid AS room_id, "
|
|
"NULL::text AS room_name, NULL::text AS rack_label, "
|
|
"NULL::text AS shelf_label, NULL::uuid AS position_id, "
|
|
"NULL::text AS position_label, NULL::integer AS position_index"
|
|
)
|
|
return f"""
|
|
WITH latest_legacy AS (
|
|
SELECT DISTINCT ON (event.miner_id)
|
|
event.miner_id, event.event_status, event.event_title,
|
|
event.description, event.event_at, event.created_at
|
|
FROM miner_maintenance_events AS event
|
|
WHERE event.event_type = 'repair'
|
|
ORDER BY event.miner_id, event.event_at DESC,
|
|
event.created_at DESC, event.id DESC
|
|
),
|
|
legacy_first AS (
|
|
SELECT event.miner_id,
|
|
min(COALESCE(event.event_at, event.created_at)) AS opened_at
|
|
FROM miner_maintenance_events AS event
|
|
WHERE event.event_type = 'repair'
|
|
GROUP BY event.miner_id
|
|
),
|
|
legacy_counts AS (
|
|
SELECT event.miner_id,
|
|
count(*)::integer AS repair_event_count,
|
|
count(*) FILTER (WHERE event.event_status IN (
|
|
'open', 'diagnosing', 'waiting_parts', 'in_progress',
|
|
'testing', 'ready_to_shelf', 'needs_followup'
|
|
))::integer AS open_repair_event_count
|
|
FROM miner_maintenance_events AS event
|
|
WHERE event.event_type = 'repair'
|
|
GROUP BY event.miner_id
|
|
),
|
|
latest_modular AS (
|
|
SELECT DISTINCT ON (event.repair_id)
|
|
event.repair_id, event.payload, event.created_at
|
|
FROM repair_events AS event
|
|
ORDER BY event.repair_id, event.sequence DESC, event.id DESC
|
|
),
|
|
initial_note AS (
|
|
SELECT DISTINCT ON (event.repair_id)
|
|
event.repair_id, event.payload ->> 'note' AS note
|
|
FROM repair_events AS event
|
|
WHERE event.event_type = 'INITIAL_NOTE'
|
|
ORDER BY event.repair_id, event.sequence, event.id
|
|
),
|
|
part_rollup AS (
|
|
SELECT part.repair_id,
|
|
count(*) FILTER (WHERE part.voided_at IS NULL)::integer AS parts_count,
|
|
COALESCE((array_agg(
|
|
CASE WHEN part.consumable_description IS NOT NULL
|
|
THEN part.consumable_description
|
|
ELSE concat_ws(' ', part.component_type_snapshot,
|
|
part.component_asset_id_snapshot,
|
|
part.component_manufacturer_serial_snapshot)
|
|
END ORDER BY part.recorded_at, part.id
|
|
) FILTER (WHERE part.voided_at IS NULL))[1:3],
|
|
ARRAY[]::text[]) AS parts_summary
|
|
FROM repair_parts AS part GROUP BY part.repair_id
|
|
),
|
|
placement AS (
|
|
SELECT DISTINCT ON (position.miner_id)
|
|
position.miner_id, site.id AS site_id, site.name AS site_name,
|
|
room.id AS room_id, COALESCE(room.name, position.room_name) AS room_name,
|
|
position.rack_name AS rack_label, position.shelf_name AS shelf_label,
|
|
position.id AS position_id, position.position_label,
|
|
position.position_index
|
|
FROM location_positions AS position
|
|
LEFT JOIN rooms AS room ON room.id = position.room_id
|
|
LEFT JOIN sites AS site ON site.id = room.site_id
|
|
WHERE position.miner_id IS NOT NULL
|
|
ORDER BY position.miner_id, position.id
|
|
),
|
|
modular_rows AS (
|
|
SELECT repair.id AS repair_id, 'modular'::text AS source,
|
|
repair.status = ANY(ARRAY['OPEN', 'IN_PROGRESS',
|
|
'WAITING_FOR_PARTS', 'READY_FOR_TEST']) AS active,
|
|
repair.opened_at, repair.completed_at AS closed_at,
|
|
repair.status AS repair_status, repair.reason,
|
|
COALESCE(NULLIF(trim(miner.manual_details ->
|
|
'office_repair_import' ->> 'reported_error'), ''),
|
|
NULLIF(trim(initial_note.note), ''),
|
|
replace(repair.reason, '_', ' ')) AS reported_problem,
|
|
COALESCE(NULLIF(trim(latest_modular.payload ->> 'note'), ''),
|
|
NULLIF(trim(latest_modular.payload ->> 'notes'), ''),
|
|
NULLIF(trim(latest_modular.payload ->> 'completion_summary'), ''),
|
|
NULLIF(trim(initial_note.note), ''),
|
|
NULLIF(trim(latest_legacy.description), ''),
|
|
NULLIF(trim(location_state.state_notes), '')) AS latest_context,
|
|
COALESCE(latest_modular.created_at, latest_legacy.event_at,
|
|
latest_legacy.created_at, repair.opened_at) AS latest_context_at,
|
|
opened_actor.display_name AS opened_by_display_name,
|
|
miner.id AS miner_id, {miner_fields}, miner.lifecycle_status,
|
|
{location_fields},
|
|
COALESCE(part_rollup.parts_count, 0) AS parts_count,
|
|
COALESCE(part_rollup.parts_summary, ARRAY[]::text[]) AS parts_summary,
|
|
latest_legacy.event_status AS legacy_event_status,
|
|
location_state.asset_state AS legacy_asset_state,
|
|
latest_legacy.event_title AS latest_event_title,
|
|
COALESCE(legacy_counts.repair_event_count, 0) AS repair_event_count,
|
|
COALESCE(legacy_counts.open_repair_event_count, 0) AS open_repair_event_count,
|
|
miner.manual_details -> 'office_repair_import' ->> 'reported_hashrate_ths' AS reported_hashrate_ths,
|
|
miner.manual_details -> 'office_repair_import' ->> 'reported_voltage' AS reported_voltage,
|
|
miner.manual_details -> 'office_repair_import' ->> 'location_note' AS location_note,
|
|
miner.manual_details -> 'office_repair_import' ->> 'source' AS import_source,
|
|
location_state.updated_at AS state_updated_at
|
|
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 latest_modular ON latest_modular.repair_id = repair.id
|
|
LEFT JOIN initial_note ON initial_note.repair_id = repair.id
|
|
LEFT JOIN part_rollup ON part_rollup.repair_id = repair.id
|
|
LEFT JOIN latest_legacy ON latest_legacy.miner_id = miner.id
|
|
LEFT JOIN legacy_counts ON legacy_counts.miner_id = miner.id
|
|
LEFT JOIN miner_location_states AS location_state
|
|
ON location_state.miner_id = miner.id
|
|
LEFT JOIN placement ON placement.miner_id = miner.id
|
|
),
|
|
legacy_rows AS (
|
|
SELECT NULL::uuid AS repair_id, 'legacy'::text AS source,
|
|
true AS active, legacy_first.opened_at, NULL::timestamptz AS closed_at,
|
|
COALESCE(latest_legacy.event_status,
|
|
location_state.asset_state, 'unknown') AS repair_status,
|
|
NULL::text AS reason,
|
|
COALESCE(NULLIF(trim(miner.manual_details ->
|
|
'office_repair_import' ->> 'reported_error'), ''),
|
|
NULLIF(trim(latest_legacy.event_title), '')) AS reported_problem,
|
|
COALESCE(NULLIF(trim(latest_legacy.description), ''),
|
|
NULLIF(trim(location_state.state_notes), '')) AS latest_context,
|
|
COALESCE(latest_legacy.event_at, latest_legacy.created_at,
|
|
location_state.updated_at) AS latest_context_at,
|
|
NULL::text AS opened_by_display_name,
|
|
miner.id AS miner_id, {miner_fields}, miner.lifecycle_status,
|
|
{location_fields}, NULL::integer AS parts_count,
|
|
ARRAY[]::text[] AS parts_summary,
|
|
latest_legacy.event_status AS legacy_event_status,
|
|
location_state.asset_state AS legacy_asset_state,
|
|
latest_legacy.event_title AS latest_event_title,
|
|
COALESCE(legacy_counts.repair_event_count, 0) AS repair_event_count,
|
|
COALESCE(legacy_counts.open_repair_event_count, 0) AS open_repair_event_count,
|
|
miner.manual_details -> 'office_repair_import' ->> 'reported_hashrate_ths' AS reported_hashrate_ths,
|
|
miner.manual_details -> 'office_repair_import' ->> 'reported_voltage' AS reported_voltage,
|
|
miner.manual_details -> 'office_repair_import' ->> 'location_note' AS location_note,
|
|
miner.manual_details -> 'office_repair_import' ->> 'source' AS import_source,
|
|
location_state.updated_at AS state_updated_at
|
|
FROM miner_location_states AS location_state
|
|
JOIN miners AS miner ON miner.id = location_state.miner_id
|
|
LEFT JOIN latest_legacy ON latest_legacy.miner_id = miner.id
|
|
LEFT JOIN legacy_counts ON legacy_counts.miner_id = miner.id
|
|
LEFT JOIN legacy_first ON legacy_first.miner_id = miner.id
|
|
LEFT JOIN placement ON placement.miner_id = miner.id
|
|
WHERE location_state.asset_state = 'repair'
|
|
AND NOT EXISTS (
|
|
SELECT 1 FROM repairs AS active_repair
|
|
WHERE active_repair.miner_id = miner.id
|
|
AND active_repair.status = ANY(ARRAY['OPEN', 'IN_PROGRESS',
|
|
'WAITING_FOR_PARTS', 'READY_FOR_TEST'])
|
|
)
|
|
),
|
|
candidates AS (
|
|
SELECT * FROM modular_rows
|
|
UNION ALL
|
|
SELECT * FROM legacy_rows
|
|
)
|
|
"""
|
|
|
|
@staticmethod
|
|
def _where(filters: Mapping[str, Any]) -> tuple[str, list[Any]]:
|
|
conditions: list[str] = []
|
|
params: list[Any] = []
|
|
scope = filters.get("scope", "active")
|
|
if scope == "active":
|
|
conditions.append("candidate.active")
|
|
elif scope == "historical":
|
|
conditions.append("NOT candidate.active")
|
|
for key, column in (
|
|
("status", "repair_status"),
|
|
("lifecycle_status", "lifecycle_status"),
|
|
("site_id", "site_id"),
|
|
("room_id", "room_id"),
|
|
):
|
|
if filters.get(key):
|
|
conditions.append(f"candidate.{column} = %s")
|
|
params.append(filters[key])
|
|
if filters.get("model"):
|
|
conditions.append("lower(trim(candidate.model)) = lower(trim(%s))")
|
|
params.append(filters["model"])
|
|
if filters.get("query"):
|
|
conditions.append(
|
|
"concat_ws(' ', candidate.repair_id, candidate.miner_id, "
|
|
"candidate.asset_id, candidate.model, candidate.miner_sn, "
|
|
"candidate.current_mac, candidate.current_ip, "
|
|
"candidate.repair_status, candidate.reason, "
|
|
"candidate.reported_problem, candidate.latest_context) ILIKE %s"
|
|
)
|
|
params.append(f"%{filters['query']}%")
|
|
return " AND ".join(conditions) if conditions else "true", params
|
|
|
|
def list_repair_center(
|
|
self,
|
|
filters: Mapping[str, Any],
|
|
*,
|
|
include_miner_details: bool,
|
|
include_location_details: bool,
|
|
) -> dict[str, Any]:
|
|
ctes = self._ctes(
|
|
show_miners=include_miner_details,
|
|
show_locations=include_location_details,
|
|
)
|
|
where_sql, params = self._where(filters)
|
|
item_sql = f"""
|
|
{ctes}
|
|
SELECT candidate.* FROM candidates AS candidate
|
|
WHERE {where_sql}
|
|
ORDER BY candidate.active DESC,
|
|
CASE WHEN candidate.active THEN candidate.opened_at END ASC NULLS LAST,
|
|
candidate.opened_at DESC NULLS LAST,
|
|
candidate.repair_id NULLS LAST, candidate.miner_id
|
|
LIMIT %s OFFSET %s
|
|
"""
|
|
summary_sql = f"""
|
|
{ctes}, filtered AS (
|
|
SELECT * FROM candidates AS candidate WHERE {where_sql}
|
|
)
|
|
SELECT count(*)::integer AS total,
|
|
count(*) FILTER (WHERE active)::integer AS active,
|
|
count(*) FILTER (WHERE NOT active)::integer AS historical,
|
|
COALESCE((SELECT jsonb_object_agg(status_rows.repair_status,
|
|
status_rows.status_count) FROM (
|
|
SELECT repair_status, count(*)::integer AS status_count
|
|
FROM filtered GROUP BY repair_status
|
|
) AS status_rows), '{{}}'::jsonb) AS status_counts,
|
|
COALESCE((SELECT jsonb_object_agg(model_rows.model,
|
|
model_rows.model_count) FROM (
|
|
SELECT model, count(*)::integer AS model_count
|
|
FROM filtered WHERE model IS NOT NULL GROUP BY model
|
|
) AS model_rows), '{{}}'::jsonb) AS model_counts,
|
|
COALESCE((SELECT jsonb_object_agg(error_rows.error,
|
|
error_rows.error_count) FROM (
|
|
SELECT COALESCE(reported_problem, 'No reported error') AS error,
|
|
count(*)::integer AS error_count
|
|
FROM filtered
|
|
GROUP BY COALESCE(reported_problem, 'No reported error')
|
|
) AS error_rows), '{{}}'::jsonb) AS error_counts
|
|
FROM filtered
|
|
"""
|
|
connection = self._connection_factory()
|
|
try:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
item_sql,
|
|
[*params, int(filters.get("limit", 50)),
|
|
int(filters.get("offset", 0))],
|
|
)
|
|
items = [dict(row) for row in cur.fetchall()]
|
|
cur.execute(summary_sql, params)
|
|
summary = dict(cur.fetchone())
|
|
options = {"sites": [], "rooms": []}
|
|
if include_location_details:
|
|
cur.execute(
|
|
f"""
|
|
{ctes}
|
|
SELECT jsonb_build_object(
|
|
'sites', COALESCE((SELECT jsonb_agg(site_rows
|
|
ORDER BY lower(site_rows.name), site_rows.id)
|
|
FROM (SELECT DISTINCT site_id AS id, site_name AS name
|
|
FROM candidates WHERE site_id IS NOT NULL) site_rows),
|
|
'[]'::jsonb),
|
|
'rooms', COALESCE((SELECT jsonb_agg(room_rows
|
|
ORDER BY lower(room_rows.name), room_rows.id)
|
|
FROM (SELECT DISTINCT room_id AS id, site_id,
|
|
room_name AS name FROM candidates
|
|
WHERE room_id IS NOT NULL) room_rows), '[]'::jsonb)
|
|
) AS options
|
|
"""
|
|
)
|
|
options = cur.fetchone()["options"]
|
|
finally:
|
|
connection.close()
|
|
return {"items": items, "summary": summary, "location_options": options}
|
|
|
|
def list_miner_repairs(self, miner_id) -> list[dict[str, Any]]:
|
|
"""Return a bounded, exact repair history for one durable miner."""
|
|
ctes = self._ctes(show_miners=False, show_locations=False)
|
|
connection = self._connection_factory()
|
|
try:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
f"""
|
|
{ctes}
|
|
SELECT candidate.* FROM candidates AS candidate
|
|
WHERE candidate.miner_id = %s
|
|
ORDER BY candidate.active DESC,
|
|
candidate.opened_at DESC NULLS LAST,
|
|
candidate.repair_id NULLS LAST
|
|
LIMIT 50
|
|
""",
|
|
(miner_id,),
|
|
)
|
|
return [dict(row) for row in cur.fetchall()]
|
|
finally:
|
|
connection.close()
|
|
|
|
def summarize_active_repairs(self) -> dict[str, Any]:
|
|
"""Return exact active counts without loading Repair Center rows."""
|
|
ctes = self._ctes(show_miners=False, show_locations=False)
|
|
connection = self._connection_factory()
|
|
try:
|
|
with connection.cursor(row_factory=dict_row) as cur:
|
|
cur.execute(
|
|
f"""
|
|
{ctes}, active_rows AS (
|
|
SELECT repair_status FROM candidates WHERE active
|
|
)
|
|
SELECT count(*)::integer AS active,
|
|
COALESCE((SELECT jsonb_object_agg(
|
|
status_rows.repair_status,
|
|
status_rows.status_count
|
|
) FROM (
|
|
SELECT repair_status,
|
|
count(*)::integer AS status_count
|
|
FROM active_rows
|
|
GROUP BY repair_status
|
|
) AS status_rows), '{{}}'::jsonb) AS status_counts
|
|
FROM active_rows
|
|
"""
|
|
)
|
|
return dict(cur.fetchone())
|
|
finally:
|
|
connection.close()
|