Enqueue alert scripts from incidents

app/main.py

@@ -23,8 +23,6 @@ CORE_SERVICES = {
     "appfactory-monitor": "http://appfactory-monitor:9300/health",
 }
 
-TERMINAL_JOB_STATUSES = {"success", "failed", "cancelled"}
-
 app = FastAPI(title="AppFactory Monitor")
 
 _monitor_thread = None
@@ -40,7 +38,6 @@ def get_connection():
 
 def get_app_services():
     con = get_connection()
-
     rows = con.execute(
         """
         SELECT id, health_url
@@ -49,7 +46,6 @@ def get_app_services():
         ORDER BY id
         """
     ).fetchall()
-
     con.close()
 
     services = {}
@@ -68,7 +64,6 @@ def get_app_services():
 
 def cleanup_old_health_records():
     con = get_connection()
-
     con.execute(
         """
         DELETE FROM service_health
@@ -76,7 +71,6 @@ def cleanup_old_health_records():
         """,
         (f"-{HEALTH_RETENTION_DAYS} days",),
     )
-
     con.commit()
     con.close()
 
@@ -111,12 +105,158 @@ def save_health_result(
         ),
     )
 
-    health_id = cur.lastrowid
+    health_id = int(cur.lastrowid)
-
     con.commit()
     con.close()
 
-    return int(health_id)
+    return health_id
+
+
+def enqueue_alerts(
+    event_type: str,
+    service_id: str,
+    incident_id: int | None,
+    payload: dict,
+):
+    con = get_connection()
+
+    rules = con.execute(
+        """
+        SELECT *
+        FROM alert_rules
+        WHERE is_enabled = 1
+          AND event_type = ?
+          AND (service_id IS NULL OR service_id = '' OR service_id = ?)
+        ORDER BY id
+        """,
+        (
+            event_type,
+            service_id,
+        ),
+    ).fetchall()
+
+    for rule in rules:
+        alert_payload = dict(payload)
+        alert_payload["rule_id"] = rule["id"]
+        alert_payload["rule_name"] = rule["name"]
+        alert_payload["script_name"] = rule["script_name"]
+
+        alert_cur = con.execute(
+            """
+            INSERT INTO alert_events (
+                rule_id,
+                event_type,
+                service_id,
+                incident_id,
+                status,
+                payload_json,
+                created_at
+            )
+            VALUES (?, ?, ?, ?, 'queued', ?, CURRENT_TIMESTAMP)
+            """,
+            (
+                rule["id"],
+                event_type,
+                service_id,
+                incident_id,
+                json.dumps(alert_payload, ensure_ascii=False),
+            ),
+        )
+
+        alert_event_id = int(alert_cur.lastrowid)
+
+        job_payload = {
+            "alert_event_id": alert_event_id,
+            "rule_id": rule["id"],
+            "script_name": rule["script_name"],
+            "event_type": event_type,
+            "service_id": service_id,
+            "incident_id": incident_id,
+            "alert_payload": alert_payload,
+        }
+
+        job_cur = con.execute(
+            """
+            INSERT INTO jobs (
+                type,
+                target_type,
+                target_id,
+                status,
+                source,
+                created_by_username,
+                payload_json,
+                created_at
+            )
+            VALUES (
+                'run_alert_script',
+                'alert_rule',
+                ?,
+                'queued',
+                'alerting',
+                'system',
+                ?,
+                CURRENT_TIMESTAMP
+            )
+            """,
+            (
+                rule["script_name"],
+                json.dumps(job_payload, ensure_ascii=False),
+            ),
+        )
+
+        job_id = int(job_cur.lastrowid)
+
+        con.execute(
+            """
+            UPDATE alert_events
+            SET job_id = ?
+            WHERE id = ?
+            """,
+            (
+                job_id,
+                alert_event_id,
+            ),
+        )
+
+        con.execute(
+            """
+            INSERT INTO audit_events (
+                username,
+                action,
+                target_type,
+                target_id,
+                source,
+                metadata,
+                created_at
+            )
+            VALUES (
+                'system',
+                'alert.enqueued',
+                'alert_rule',
+                ?,
+                'monitor',
+                ?,
+                CURRENT_TIMESTAMP
+            )
+            """,
+            (
+                str(rule["id"]),
+                json.dumps(
+                    {
+                        "alert_event_id": alert_event_id,
+                        "job_id": job_id,
+                        "event_type": event_type,
+                        "service_id": service_id,
+                        "incident_id": incident_id,
+                        "script_name": rule["script_name"],
+                    },
+                    ensure_ascii=False,
+                ),
+            ),
+        )
+
+    con.commit()
+    con.close()
 
 
 def get_open_incident(service_id: str):
@@ -135,7 +275,6 @@ def get_open_incident(service_id: str):
     ).fetchone()
 
     con.close()
-
     return row
 
 
@@ -148,7 +287,7 @@ def open_incident(service_id: str, health_id: int, status: str, error_text: str
 
     con = get_connection()
 
-    con.execute(
+    cur = con.execute(
         """
         INSERT INTO service_incidents (
             service_id,
@@ -168,6 +307,14 @@ def open_incident(service_id: str, health_id: int, status: str, error_text: str
         ),
     )
 
+    incident_id = int(cur.lastrowid)
+
+    metadata = {
+        "incident_id": incident_id,
+        "health_id": health_id,
+        "status": status,
+    }
+
     con.execute(
         """
         INSERT INTO audit_events (
@@ -191,19 +338,28 @@ def open_incident(service_id: str, health_id: int, status: str, error_text: str
         """,
         (
             service_id,
-            json.dumps(
+            json.dumps(metadata, ensure_ascii=False),
-                {
-                    "health_id": health_id,
-                    "status": status,
-                },
-                ensure_ascii=False,
-            ),
         ),
     )
 
     con.commit()
     con.close()
 
+    enqueue_alerts(
+        event_type="incident.opened",
+        service_id=service_id,
+        incident_id=incident_id,
+        payload={
+            "incident_id": incident_id,
+            "service_id": service_id,
+            "health_id": health_id,
+            "health_status": status,
+            "title": title,
+            "description": description,
+            "error_text": error_text,
+        },
+    )
+
 
 def resolve_incident(service_id: str, health_id: int):
     incident = get_open_incident(service_id)
@@ -211,6 +367,8 @@ def resolve_incident(service_id: str, health_id: int):
     if not incident:
         return
 
+    incident_id = int(incident["id"])
+
     con = get_connection()
 
     con.execute(
@@ -227,10 +385,19 @@ def resolve_incident(service_id: str, health_id: int):
         """,
         (
             health_id,
-            incident["id"],
+            incident_id,
         ),
     )
 
+    row = con.execute(
+        """
+        SELECT *
+        FROM service_incidents
+        WHERE id = ?
+        """,
+        (incident_id,),
+    ).fetchone()
+
     con.execute(
         """
         INSERT INTO audit_events (
@@ -256,7 +423,7 @@ def resolve_incident(service_id: str, health_id: int):
             service_id,
             json.dumps(
                 {
-                    "incident_id": incident["id"],
+                    "incident_id": incident_id,
                     "health_id": health_id,
                 },
                 ensure_ascii=False,
@@ -267,6 +434,22 @@ def resolve_incident(service_id: str, health_id: int):
     con.commit()
     con.close()
 
+    enqueue_alerts(
+        event_type="incident.resolved",
+        service_id=service_id,
+        incident_id=incident_id,
+        payload={
+            "incident_id": incident_id,
+            "service_id": service_id,
+            "health_id": health_id,
+            "title": row["title"] if row else None,
+            "description": row["description"] if row else None,
+            "started_at": row["started_at"] if row else None,
+            "ended_at": row["ended_at"] if row else None,
+            "duration_seconds": row["duration_seconds"] if row else None,
+        },
+    )
+
 
 def update_incident_state(service_id: str, health_id: int, status: str, error_text: str | None):
     if status == "healthy":