Add DSH key-alarm subscribe hub for outbound netxops clients.

netx hosts /v1/integrations/dsh-alarm/ws; matched key alerts fan out to dialed-in DSH subscribers while the optional oclaw bridge remains.

Co-authored-by: Cursor <cursoragent@cursor.com>

netx_api/dsh_alarm_hub.py

@@ -0,0 +1,212 @@
+"""DSH / netxops key-alarm subscribe hub (netx is the fixed-IP WebSocket server).
+
+Clients (netxops on each DSH host) dial out to:
+  ws[s]://<netx-host>:<port>/v1/integrations/dsh-alarm/ws
+authenticate with the same API token used for REST, then receive `netx.alarm` events.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import threading
+from datetime import datetime, timezone
+from typing import Any
+
+from fastapi import WebSocket, WebSocketDisconnect
+from starlette.websockets import WebSocketState
+
+from .auth_deps import resolve_user_from_token
+from .auth_scopes import SCOPE_ALARMS_READ, has_scope
+from .config import settings
+from .db import SessionLocal
+
+_log = logging.getLogger("netx.dsh.alarm_hub")
+
+_LOCK = threading.Lock()
+_CLIENTS: set[WebSocket] = set()
+_LOOP: asyncio.AbstractEventLoop | None = None
+_STATS = {
+    "published": 0,
+    "deliver_ok": 0,
+    "deliver_fail": 0,
+    "subscribers": 0,
+}
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def bind_event_loop(loop: asyncio.AbstractEventLoop | None) -> None:
+    """Remember the API process loop so sync publishers can schedule sends."""
+    global _LOOP
+    _LOOP = loop
+
+
+def subscriber_count() -> int:
+    with _LOCK:
+        return len(_CLIENTS)
+
+
+def hub_status() -> dict[str, Any]:
+    with _LOCK:
+        clients = len(_CLIENTS)
+        stats = dict(_STATS)
+    stats["subscribers"] = clients
+    return {
+        "enabled": True,
+        "path": "/v1/integrations/dsh-alarm/ws",
+        "subscribers": clients,
+        "published": int(stats.get("published") or 0),
+        "deliver_ok": int(stats.get("deliver_ok") or 0),
+        "deliver_fail": int(stats.get("deliver_fail") or 0),
+    }
+
+
+def _authorize_token(token: str) -> tuple[bool, str]:
+    raw = str(token or "").strip()
+    if not bool(settings.auth_enabled):
+        return True, "auth_disabled"
+    if not raw:
+        return False, "token_required"
+    db = SessionLocal()
+    try:
+        hit = resolve_user_from_token(db, raw)
+        if hit is None:
+            return False, "invalid_token"
+        user, _via, scopes, _token_id, _jti = hit
+        if has_scope(scopes, SCOPE_ALARMS_READ) or str(user.role or "").lower() == "admin":
+            return True, str(user.username or user.id or "user")
+        return False, "missing_alarms_read_scope"
+    except Exception as exc:  # noqa: BLE001
+        _log.warning("dsh alarm hub auth error: %s", exc)
+        return False, "auth_error"
+    finally:
+        db.close()
+
+
+async def _send_json(ws: WebSocket, payload: dict[str, Any]) -> bool:
+    if ws.client_state != WebSocketState.CONNECTED:
+        return False
+    try:
+        await ws.send_json(payload)
+        return True
+    except Exception:
+        return False
+
+
+async def _broadcast(payload: dict[str, Any]) -> int:
+    with _LOCK:
+        clients = list(_CLIENTS)
+    if not clients:
+        return 0
+    envelope = {
+        "type": "event",
+        "event": "netx.alarm",
+        "ts": _utc_now_iso(),
+        "payload": payload,
+    }
+    dead: list[WebSocket] = []
+    ok = 0
+    for ws in clients:
+        if await _send_json(ws, envelope):
+            ok += 1
+        else:
+            dead.append(ws)
+    if dead:
+        with _LOCK:
+            for ws in dead:
+                _CLIENTS.discard(ws)
+            _STATS["subscribers"] = len(_CLIENTS)
+            _STATS["deliver_fail"] += len(dead)
+        for ws in dead:
+            try:
+                await ws.close()
+            except Exception:
+                pass
+    with _LOCK:
+        _STATS["deliver_ok"] += ok
+    return ok
+
+
+def publish_alarm(payload: dict[str, Any]) -> int:
+    """Fan-out a matched key-alert payload to all connected DSH subscribers.
+
+    Safe to call from sync UME/alarm threads. Returns the number of clients that
+    accepted the message (0 when nobody is subscribed).
+    """
+    if not isinstance(payload, dict):
+        return 0
+    with _LOCK:
+        _STATS["published"] += 1
+        clients = len(_CLIENTS)
+        loop = _LOOP
+    if clients <= 0:
+        return 0
+    if loop is None or not loop.is_running():
+        _log.warning("dsh alarm hub has subscribers but no running event loop")
+        return 0
+
+    future = asyncio.run_coroutine_threadsafe(_broadcast(dict(payload)), loop)
+    try:
+        return int(future.result(timeout=5))
+    except Exception as exc:  # noqa: BLE001
+        _log.warning("dsh alarm hub publish failed: %s", exc)
+        return 0
+
+
+async def dsh_alarm_ws_loop(websocket: WebSocket) -> None:
+    """Accept one netxops subscriber: auth → ping/pong → receive netx.alarm pushes."""
+    await websocket.accept()
+    bind_event_loop(asyncio.get_running_loop())
+    authed = False
+    try:
+        while True:
+            raw = await websocket.receive_text()
+            try:
+                import json
+
+                msg = json.loads(raw)
+            except Exception:
+                await _send_json(websocket, {"type": "error", "error": "invalid_json"})
+                continue
+            if not isinstance(msg, dict):
+                await _send_json(websocket, {"type": "error", "error": "invalid_message"})
+                continue
+            mtype = str(msg.get("type") or "").strip().lower()
+            if not authed:
+                if mtype != "auth":
+                    await _send_json(websocket, {"type": "auth-fail", "error": "auth_required"})
+                    await websocket.close(code=4401)
+                    return
+                ok, detail = _authorize_token(str(msg.get("token") or ""))
+                if not ok:
+                    await _send_json(websocket, {"type": "auth-fail", "error": detail})
+                    await websocket.close(code=4401)
+                    return
+                authed = True
+                with _LOCK:
+                    _CLIENTS.add(websocket)
+                    _STATS["subscribers"] = len(_CLIENTS)
+                await _send_json(
+                    websocket,
+                    {
+                        "type": "auth-ok",
+                        "user": detail,
+                        "ts": _utc_now_iso(),
+                    },
+                )
+                _log.info("dsh alarm hub subscriber connected (%s)", detail)
+                continue
+            if mtype == "ping":
+                await _send_json(websocket, {"type": "pong", "ts": _utc_now_iso()})
+                continue
+            await _send_json(websocket, {"type": "error", "error": f"unknown_type:{mtype}"})
+    except WebSocketDisconnect:
+        return
+    finally:
+        with _LOCK:
+            _CLIENTS.discard(websocket)
+            _STATS["subscribers"] = len(_CLIENTS)
+        _log.info("dsh alarm hub subscriber disconnected")

netx_api/integrations_router.py

@@ -5,17 +5,24 @@ from __future__ import annotations
 import time
 from typing import Any
 
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, WebSocket
 from sqlalchemy import text as sql_text
 from sqlalchemy.orm import Session
 
 from .config import settings
 from .db import get_db
+from .dsh_alarm_hub import dsh_alarm_ws_loop, hub_status
 from .oclaw_alarm_forwarder import forwarder_status
 
 router = APIRouter(tags=["health"])
 
 
+@router.websocket("/v1/integrations/dsh-alarm/ws")
+async def dsh_alarm_subscribe(websocket: WebSocket) -> None:
+    """netxops dials out here to receive matched key-alert pushes."""
+    await dsh_alarm_ws_loop(websocket)
+
+
 @router.get("/health/live", status_code=200)
 def health_live() -> dict[str, str]:
     """Process liveness — no DB or upstream checks."""
@@ -130,4 +137,9 @@ def integrations_status(db: Session = Depends(get_db)) -> dict:
             "forwarder": fwd,
         }
 
-    return {"netx_api": netx_api, "db": db_status, "oclaw_bridge": oclaw_status}
+    return {
+        "netx_api": netx_api,
+        "db": db_status,
+        "oclaw_bridge": oclaw_status,
+        "dsh_alarm_hub": hub_status(),
+    }

netx_api/key_alert_forward.py

@@ -10,6 +10,7 @@ from sqlalchemy.orm import Session
 from .key_alert_matcher import match_key_alert_rule
 from .models import UmeInventoryNE, UmeKeyAlertForwardLog
 from .oclaw_alarm_forwarder import enqueue_alarm_forward, is_forwarder_operational
+from .dsh_alarm_hub import publish_alarm
 from .ume_sync_service import (
     _derive_ne_id_from_alarm,
     _pick,
@@ -69,8 +70,6 @@ def maybe_forward_key_alert(
     alarm_key: str,
     action: str,
 ) -> bool:
-    if not is_forwarder_operational():
-        return False
     rule = match_key_alert_rule(db, norm=norm, action=action)
     if rule is None:
         return False
@@ -96,10 +95,23 @@ def maybe_forward_key_alert(
     payload["ne"] = _ne_payload(db, ne_id)
     payload["rule_key"] = str(rule.notification_id or "")
 
-    queued = enqueue_alarm_forward(payload)
+    hub_sent = publish_alarm(payload)
-    if not queued:
+    oclaw_queued = False
+    if is_forwarder_operational():
+        oclaw_queued = bool(enqueue_alarm_forward(payload))
+    if hub_sent <= 0 and not oclaw_queued:
         return False
 
+    # Hub-only delivery already reached DSH clients — mark ok for dedup.
+    # Oclaw path stays pending until the bridge records a result.
+    delivered_ok = hub_sent > 0 and not oclaw_queued
+    status = []
+    if hub_sent > 0:
+        status.append(f"dsh_hub:{hub_sent}")
+    if oclaw_queued:
+        status.append("oclaw_queued")
+    status_text = ",".join(status) if status else "queued"
+
     row = existing
     if row is None:
         row = UmeKeyAlertForwardLog(
@@ -108,16 +120,16 @@ def maybe_forward_key_alert(
             rule_key=str(rule.notification_id or ""),
             notification_id=notification_id_from_norm(norm),
             forwarded_at=_utc_now_naive(),
-            oclaw_ok=0,
+            oclaw_ok=1 if delivered_ok else 0,
-            error="queued",
+            error="" if delivered_ok else status_text,
         )
         db.add(row)
     else:
         row.notification_id = notification_id_from_norm(norm)
         row.rule_key = str(rule.notification_id or "")
         row.forwarded_at = _utc_now_naive()
-        row.oclaw_ok = 0
+        row.oclaw_ok = 1 if delivered_ok else 0
-        row.error = "queued"
+        row.error = "" if delivered_ok else status_text
     try:
         db.commit()
     except IntegrityError:

netx_api/main.py

@@ -45,10 +45,14 @@ _BOOT_MONO = time.monotonic()
 
 @asynccontextmanager
 async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
+    import asyncio
+
     from .app_startup import run_api_startup
     from .app_shutdown import shutdown_runtime
+    from .dsh_alarm_hub import bind_event_loop
 
     run_api_startup()
+    bind_event_loop(asyncio.get_running_loop())
     try:
         yield
     finally: