08:29

files/nats_registration_listener.py

@@ -22,6 +22,7 @@ from typing import Optional, Dict, Any, Tuple, Set
 from urllib.parse import urlencode
 from urllib.request import Request, urlopen
 from urllib.error import URLError, HTTPError
+import base64  # NEW: for RabbitMQ Basic Auth
 
 import nats
 
@@ -33,6 +34,19 @@ NB_URL = "http://netbox.gt-tiso.ikeja.co.za"  # Base URL
 NB_TOKEN = "7648e4f5ee370cda7834682e61b47c2ee8e95623"  # keep as provided
 NB_TIMEOUT = 3.0  # seconds per HTTP GET
 
+# =========================
+# RabbitMQ hardcoded config (TTL + DLX path)
+# =========================
+RMQ_HOST = "10.210.12.2"
+RMQ_PORT = 15672
+RMQ_USER = "admin"
+RMQ_PASS = "change_me"
+RMQ_VHOST = "app"
+RMQ_EXCHANGE_HOLDING = "deviceconfig.holding"  # publish here with per-message TTL
+RMQ_ROUTING_KEY = "deviceconfig"               # routed by DLX to live flow
+RMQ_TIMEOUT = 3.0
+RMQ_DELAY_MS = 10000  # 10 seconds
+
 # Product -> Tag slug mapping (future-proof for fox200 later)
 PRODUCT_TAG_SLUG = {
     "fox100": "fox100-auto-upgrade-latest",
@@ -143,6 +157,29 @@ def http_get_json(url: str, params: Optional[Dict[str, str]] = None, headers: Op
         return None, 597
 
 
+# NEW: RabbitMQ management API POST helper (basic auth; JSON in/out)
+def http_post_json(url: str, payload_obj: Dict[str, Any], user: Optional[str] = None, password: Optional[str] = None, timeout: float = RMQ_TIMEOUT):
+    body = json.dumps(payload_obj).encode("utf-8")
+    headers = {"Content-Type": "application/json"}
+    if user and password:
+        token = base64.b64encode(f"{user}:{password}".encode("utf-8")).decode("ascii")
+        headers["Authorization"] = f"Basic {token}"
+    req = Request(url, data=body, headers=headers, method="POST")
+    try:
+        with urlopen(req, timeout=timeout) as resp:
+            data = resp.read()
+            try:
+                return json.loads(data.decode("utf-8", errors="replace")), resp.status
+            except Exception:
+                return None, resp.status
+    except HTTPError as e:
+        return None, getattr(e, "code", 599)
+    except URLError:
+        return None, 598
+    except Exception:
+        return None, 597
+
+
 async def nb_problem(log_status, msg: str):
     """Increment counter and log a problem line."""
     global NB_PROBLEM_COUNTER
@@ -314,6 +351,28 @@ async def main():
                         # Beep to draw attention
                         bell_prefix = "\a" * 5
                         action_suffix = f" action=ok, i'm ready to schedule this device {host if host else dev_id} upgrade"
+
+                        # NEW: Schedule delayed RabbitMQ message (TTL + DLX; 10s)
+                        if host:
+                            try:
+                                rmq_url = f"http://{RMQ_HOST}:{RMQ_PORT}/api/exchanges/{RMQ_VHOST}/{RMQ_EXCHANGE_HOLDING}/publish"
+                                rmq_body = {
+                                    "properties": {
+                                        "content_type": "application/json",
+                                        "expiration": str(RMQ_DELAY_MS),
+                                    },
+                                    "routing_key": RMQ_ROUTING_KEY,
+                                    "payload": json.dumps({
+                                        "inscope_device": host,
+                                        "task_name": "sot-updater",
+                                    }),
+                                    "payload_encoding": "string",
+                                }
+                                resp, code = http_post_json(rmq_url, rmq_body, user=RMQ_USER, password=RMQ_PASS, timeout=RMQ_TIMEOUT)
+                                if code != 200:
+                                    await log_status(f"[{ts()}] rmq: publish delayed http={code} host={host} delay_ms={RMQ_DELAY_MS}")
+                            except Exception as e:
+                                await log_status(f"[{ts()}] rmq: unexpected error host={host!r} err={e!r}")
             except Exception as e:
                 await nb_problem(log_status, f"[{ts()}] nb: unexpected error mac={mac!r} err={e!r}")
             netbox_time_ms = (time.perf_counter() - nb_start) * 1000