appfactory-portal/app/db/alerting.py

from typing import Any

from app.db.database import get_connection
from app.db.migrations import run_migrations


def get_alert_rules():
    run_migrations()
    con = get_connection()

    rows = con.execute(
        """
        SELECT
            r.id,
            r.name,
            r.description,
            r.event_type,
            r.service_id,
            r.script_name,
            COALESCE(r.is_enabled, 1) AS is_enabled,
            r.created_at,
            r.updated_at,
            e.id AS last_event_id,
            e.status AS last_event_status,
            e.created_at AS last_event_at
        FROM alert_rules r
        LEFT JOIN alert_events e ON e.id = (
            SELECT id
            FROM alert_events
            WHERE rule_id = r.id
            ORDER BY created_at DESC, id DESC
            LIMIT 1
        )
        ORDER BY r.name
        """
    ).fetchall()

    con.close()
    return [dict(row) for row in rows]


def get_alert_rule(rule_id: int):
    run_migrations()
    con = get_connection()

    row = con.execute(
        """
        SELECT
            id,
            name,
            description,
            event_type,
            service_id,
            script_name,
            COALESCE(is_enabled, 1) AS is_enabled,
            created_at,
            updated_at
        FROM alert_rules
        WHERE id = ?
        """,
        (rule_id,),
    ).fetchone()

    con.close()
    return dict(row) if row else None


def create_alert_rule(metadata: dict[str, Any]) -> int:
    run_migrations()
    con = get_connection()

    cur = con.execute(
        """
        INSERT INTO alert_rules (
            name,
            description,
            event_type,
            service_id,
            script_name,
            is_enabled,
            created_at,
            updated_at
        )
        VALUES (?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
        """,
        (
            metadata.get("name"),
            metadata.get("description") or None,
            metadata.get("event_type"),
            metadata.get("service_id") or None,
            metadata.get("script_name"),
            1 if metadata.get("is_enabled") else 0,
        ),
    )
    rule_id = cur.lastrowid
    con.commit()
    con.close()
    return rule_id


def update_alert_rule(rule_id: int, metadata: dict[str, Any]) -> None:
    run_migrations()
    con = get_connection()

    con.execute(
        """
        UPDATE alert_rules
        SET name = ?,
            description = ?,
            event_type = ?,
            service_id = ?,
            script_name = ?,
            is_enabled = ?,
            updated_at = CURRENT_TIMESTAMP
        WHERE id = ?
        """,
        (
            metadata.get("name"),
            metadata.get("description") or None,
            metadata.get("event_type"),
            metadata.get("service_id") or None,
            metadata.get("script_name"),
            1 if metadata.get("is_enabled") else 0,
            rule_id,
        ),
    )

    con.commit()
    con.close()


def set_alert_rule_enabled(rule_id: int, is_enabled: bool) -> None:
    run_migrations()
    con = get_connection()

    con.execute(
        """
        UPDATE alert_rules
        SET is_enabled = ?,
            updated_at = CURRENT_TIMESTAMP
        WHERE id = ?
        """,
        (1 if is_enabled else 0, rule_id),
    )

    con.commit()
    con.close()


def delete_alert_rule(rule_id: int) -> bool:
    run_migrations()
    con = get_connection()

    cur = con.execute("DELETE FROM alert_rules WHERE id = ?", (rule_id,))
    deleted = cur.rowcount == 1
    con.commit()
    con.close()
    return deleted


def get_alert_events(limit: int = 200):
    run_migrations()
    con = get_connection()

    rows = con.execute(
        """
        SELECT
            e.id,
            e.rule_id,
            r.name AS rule_name,
            e.event_type,
            e.service_id,
            e.incident_id,
            e.status,
            e.job_id,
            e.error_text,
            e.payload_json,
            e.created_at,
            e.processed_at
        FROM alert_events e
        LEFT JOIN alert_rules r ON r.id = e.rule_id
        ORDER BY e.created_at DESC, e.id DESC
        LIMIT ?
        """,
        (limit,),
    ).fetchall()

    con.close()
    return [dict(row) for row in rows]


def get_alert_event(event_id: int):
    run_migrations()
    con = get_connection()

    row = con.execute(
        """
        SELECT
            e.id,
            e.rule_id,
            r.name AS rule_name,
            e.event_type,
            e.service_id,
            e.incident_id,
            e.status,
            e.job_id,
            e.error_text,
            e.payload_json,
            e.created_at,
            e.processed_at
        FROM alert_events e
        LEFT JOIN alert_rules r ON r.id = e.rule_id
        WHERE e.id = ?
        """,
        (event_id,),
    ).fetchone()

    con.close()
    return dict(row) if row else None


def get_rule_alert_events(rule_id: int, limit: int = 20):
    run_migrations()
    con = get_connection()

    rows = con.execute(
        """
        SELECT
            id,
            rule_id,
            event_type,
            service_id,
            incident_id,
            status,
            job_id,
            error_text,
            created_at,
            processed_at
        FROM alert_events
        WHERE rule_id = ?
        ORDER BY created_at DESC, id DESC
        LIMIT ?
        """,
        (rule_id, limit),
    ).fetchall()

    con.close()
    return [dict(row) for row in rows]