minerops/backend/alembic/versions/0007_identity_and_roles.py

"""add application roles and permission matrix

Revision ID: 0007_identity_and_roles
Revises: 0006_operational_schema_baseline
Create Date: 2026-07-27
"""

from typing import Sequence, Union

from alembic import op


revision: str = "0007_identity_and_roles"
down_revision: Union[str, None] = "0006_operational_schema_baseline"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


PERMISSIONS = (
    ("dashboard.view", "View fleet dashboards"),
    ("miners.view", "View miners and telemetry"),
    ("miners.update", "Update miner inventory and details"),
    ("miners.control", "Run approved miner control operations"),
    ("repairs.view", "View repair queues and history"),
    ("repairs.manage", "Manage repairs, notes, parts, and status"),
    ("locations.view", "View rooms, racks, shelves, and placements"),
    ("locations.manage", "Manage layouts and miner placements"),
    ("agents.view", "View site agents and command status"),
    ("agents.manage", "Queue and manage site-agent work"),
    ("system_health.view", "View system health and retention status"),
    ("users.manage", "Create and manage MinerOps accounts"),
    ("credentials.manage", "Manage protected miner credentials"),
    ("scan_ranges.manage", "Manage scan ranges"),
    ("scans.run", "Run discovery and polling operations"),
    ("exports.create", "Export operational data"),
    ("audit.view", "View audit history"),
)

ROLE_PERMISSIONS = {
    "Manager": (
        "dashboard.view",
        "miners.view",
        "miners.update",
        "miners.control",
        "repairs.view",
        "repairs.manage",
        "locations.view",
        "locations.manage",
        "agents.view",
        "agents.manage",
        "system_health.view",
        "scan_ranges.manage",
        "scans.run",
        "exports.create",
        "audit.view",
    ),
    "Technician": (
        "dashboard.view",
        "miners.view",
        "miners.update",
        "miners.control",
        "repairs.view",
        "repairs.manage",
        "locations.view",
        "locations.manage",
        "agents.view",
        "scans.run",
        "exports.create",
    ),
    "Read Only": (
        "dashboard.view",
        "miners.view",
        "repairs.view",
        "locations.view",
        "agents.view",
        "system_health.view",
    ),
}


def upgrade() -> None:
    connection = op.get_bind()

    for key, description in PERMISSIONS:
        connection.exec_driver_sql(
            """
            INSERT INTO permissions (key, description)
            VALUES (%s, %s)
            ON CONFLICT (key) DO UPDATE
            SET description = EXCLUDED.description
            """,
            (key, description),
        )

    connection.exec_driver_sql(
        """
        INSERT INTO roles (name, description)
        VALUES
            ('Manager', 'Operational manager with fleet and workflow control'),
            ('Read Only', 'Read-only access to operational views')
        ON CONFLICT (name) DO UPDATE
        SET description = EXCLUDED.description
        """
    )

    connection.exec_driver_sql(
        """
        UPDATE roles
        SET description = CASE name
            WHEN 'Admin' THEN 'Full MinerOps administrator'
            WHEN 'Technician' THEN 'Technician for repair and fleet operations'
            ELSE description
        END
        WHERE name IN ('Admin', 'Technician')
        """
    )

    connection.exec_driver_sql(
        """
        UPDATE users
        SET role_id = (SELECT id FROM roles WHERE name = 'Read Only')
        WHERE role_id IN (
            SELECT id FROM roles WHERE name IN ('Viewer', 'Auditor')
        )
        """
    )
    connection.exec_driver_sql(
        "DELETE FROM role_permissions WHERE role_id IN (SELECT id FROM roles WHERE name IN ('Viewer', 'Auditor'))"
    )
    connection.exec_driver_sql("DELETE FROM roles WHERE name IN ('Viewer', 'Auditor')")

    connection.exec_driver_sql(
        "DELETE FROM role_permissions WHERE role_id IN (SELECT id FROM roles WHERE name IN ('Admin', 'Manager', 'Technician', 'Read Only'))"
    )
    connection.exec_driver_sql(
        """
        INSERT INTO role_permissions (role_id, permission_id)
        SELECT roles.id, permissions.id
        FROM roles
        CROSS JOIN permissions
        WHERE roles.name = 'Admin'
        ON CONFLICT DO NOTHING
        """
    )

    for role_name, keys in ROLE_PERMISSIONS.items():
        connection.exec_driver_sql(
            """
            INSERT INTO role_permissions (role_id, permission_id)
            SELECT roles.id, permissions.id
            FROM roles
            JOIN permissions ON permissions.key = ANY(%s)
            WHERE roles.name = %s
            ON CONFLICT DO NOTHING
            """,
            (list(keys), role_name),
        )


def downgrade() -> None:
    raise RuntimeError("0007 changes live role assignments and is intentionally irreversible")