cx-ui/triagelib/settings.py

"""User settings: suppression rules and comms identity.

Suppression rules replace hardcoded judgement calls. The internal-organisation
check used to be baked into the screening code, which meant the one person who
knew about it was whoever read the source. Now it ships as an editable default
rule that says what it does and why, and anyone can add their own.

A rule matches when *every* condition it sets matches (AND); within one
condition, any listed value matches (OR). So "type is error AND organisation
contains modal" is one rule with two conditions.
"""
from __future__ import annotations

import json
import os
import re
import threading
import time
import uuid
from typing import Any, Optional

SETTINGS_DIR = os.path.expanduser(os.environ.get("CX_TRIAGE_HOME", "~/.cx-triage"))
SETTINGS_PATH = os.path.join(SETTINGS_DIR, "settings.json")

# Conditions a rule can set. Every one is a substring match, case-insensitive,
# except `kind` and `region` which are exact.
CONDITIONS = {
    "kind": "Alert type",
    "organization": "Organisation contains",
    "instance_name": "VM name contains",
    "host": "Host contains",
    "region": "Region",
    "status": "Status is",
}

DEFAULT_RULES: list[dict[str, Any]] = [
    {
        "id": "builtin-internal-orgs",
        "name": "Internal NexGen organisations",
        "enabled": True,
        "reason": "Owned by an internal test or platform organisation, not a customer.",
        "conditions": {"organization": ["nexgencloud.com"]},
    },
    {
        "id": "builtin-runpod-storage",
        "name": "Runpod storage nodes (Luis)",
        "enabled": True,
        "reason": "Platform-owned storage nodes; SHUTOFF on these is expected and not customer-impacting.",
        "conditions": {"kind": ["shutoff"], "instance_name": ["stor-runpod"]},
    },
]

DEFAULTS: dict[str, Any] = {
    "rules": DEFAULT_RULES,
    "agent_name": "",
    "chronic_days": 3,
}


def _blank(value: Any) -> bool:
    return value is None or str(value).strip() == ""


class Settings:
    """Loaded once, written through on every change."""

    def __init__(self, path: str = SETTINGS_PATH):
        self.path = path
        self._lock = threading.Lock()
        self._data: dict[str, Any] = {}
        self.load()

    # --- persistence -------------------------------------------------------

    def load(self) -> None:
        data = dict(DEFAULTS)
        try:
            with open(self.path, encoding="utf-8") as handle:
                stored = json.load(handle)
            if isinstance(stored, dict):
                data.update(stored)
        except (OSError, json.JSONDecodeError):
            pass
        data["rules"] = [r for r in (data.get("rules") or []) if isinstance(r, dict)]
        with self._lock:
            self._data = data

    def save(self) -> None:
        with self._lock:
            payload = json.dumps(self._data, indent=2, sort_keys=True)
        try:
            os.makedirs(os.path.dirname(self.path), exist_ok=True)
            tmp = f"{self.path}.tmp"
            with open(tmp, "w", encoding="utf-8") as handle:
                handle.write(payload)
            os.replace(tmp, self.path)
        except OSError:
            pass

    # --- accessors ---------------------------------------------------------

    @property
    def rules(self) -> list[dict[str, Any]]:
        with self._lock:
            return [dict(r) for r in self._data.get("rules", [])]

    @property
    def agent_name(self) -> str:
        with self._lock:
            return str(self._data.get("agent_name") or "")

    @property
    def chronic_days(self) -> int:
        with self._lock:
            try:
                return max(1, int(self._data.get("chronic_days") or 3))
            except (TypeError, ValueError):
                return 3

    def to_json(self) -> dict[str, Any]:
        with self._lock:
            return {
                "rules": [dict(r) for r in self._data.get("rules", [])],
                "agent_name": self._data.get("agent_name") or "",
                "chronic_days": self._data.get("chronic_days", 3),
                "conditions": CONDITIONS,
                "path": self.path,
            }

    # --- mutations ---------------------------------------------------------

    def set_general(self, agent_name: Optional[str] = None, chronic_days: Optional[Any] = None) -> None:
        with self._lock:
            if agent_name is not None:
                self._data["agent_name"] = str(agent_name).strip()
            if chronic_days is not None:
                try:
                    self._data["chronic_days"] = max(1, int(chronic_days))
                except (TypeError, ValueError):
                    pass
        self.save()

    def upsert_rule(self, rule: dict[str, Any]) -> dict[str, Any]:
        clean = _normalize_rule(rule)
        with self._lock:
            rules = self._data.setdefault("rules", [])
            for idx, existing in enumerate(rules):
                if existing.get("id") == clean["id"]:
                    rules[idx] = clean
                    break
            else:
                rules.append(clean)
        self.save()
        return clean

    def delete_rule(self, rule_id: str) -> None:
        with self._lock:
            self._data["rules"] = [r for r in self._data.get("rules", []) if r.get("id") != rule_id]
        self.save()

    def toggle_rule(self, rule_id: str, enabled: bool) -> None:
        with self._lock:
            for rule in self._data.get("rules", []):
                if rule.get("id") == rule_id:
                    rule["enabled"] = bool(enabled)
        self.save()


def _normalize_rule(rule: dict[str, Any]) -> dict[str, Any]:
    conditions: dict[str, list[str]] = {}
    for field, values in (rule.get("conditions") or {}).items():
        if field not in CONDITIONS:
            continue
        if isinstance(values, str):
            values = [v.strip() for v in values.split(",")]
        cleaned = [str(v).strip() for v in (values or []) if str(v).strip()]
        if cleaned:
            conditions[field] = cleaned
    return {
        "id": str(rule.get("id") or f"rule-{uuid.uuid4().hex[:8]}"),
        "name": str(rule.get("name") or "Untitled rule").strip(),
        "enabled": bool(rule.get("enabled", True)),
        "reason": str(rule.get("reason") or "").strip(),
        "conditions": conditions,
        "created": rule.get("created") or time.strftime("%Y-%m-%d"),
    }


# --- matching ---------------------------------------------------------------

def _alert_field(alert: Any, field: str) -> str:
    if field == "kind":
        return str(getattr(alert, "kind", ""))
    if field == "organization":
        return f"{getattr(alert, 'org_id', '')} {getattr(alert, 'org_name', '')}"
    if field == "instance_name":
        return str(getattr(alert, "instance_name", ""))
    if field == "host":
        return str(getattr(alert, "host", ""))
    if field == "region":
        return f"{getattr(alert, 'region', '')} {getattr(alert, 'region_label', '')}"
    if field == "status":
        return str(getattr(alert, "status", ""))
    return ""


def _condition_matches(field: str, values: list[str], alert: Any) -> bool:
    actual = _alert_field(alert, field).lower()
    if field in ("kind", "status"):
        return any(actual == str(v).strip().lower() for v in values)
    if field == "region":
        return any(str(v).strip().lower() in actual for v in values)
    return any(str(v).strip().lower() in actual for v in values)


def rule_matches(rule: dict[str, Any], alert: Any) -> bool:
    """Every condition in the rule must match (AND)."""
    conditions = rule.get("conditions") or {}
    if not conditions:
        return False  # an empty rule would swallow the whole queue
    return all(_condition_matches(f, v, alert) for f, v in conditions.items())


def first_match(settings: Settings, alert: Any) -> Optional[dict[str, Any]]:
    for rule in settings.rules:
        if rule.get("enabled") and rule_matches(rule, alert):
            return rule
    return None


def preview(settings: Settings, rule: dict[str, Any], alerts: list[Any]) -> list[dict[str, Any]]:
    """Which currently-firing alerts a rule would hide - shown before saving."""
    clean = _normalize_rule(rule)
    hits = []
    for alert in alerts:
        if rule_matches(clean, alert):
            hits.append({
                "kind": alert.kind,
                "title": alert.title,
                "instance_name": alert.instance_name,
                "host": alert.host,
                "org_name": alert.org_name,
                "region": alert.region,
            })
    return hits