From 50218c28d55cfb7e1f5d6bddcd0d0eca89de37b4 Mon Sep 17 00:00:00 2001 From: JA Date: Thu, 9 Jul 2026 03:42:40 +0000 Subject: [PATCH] Add eventbus.py --- eventbus.py | 288 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 288 insertions(+) create mode 100644 eventbus.py diff --git a/eventbus.py b/eventbus.py new file mode 100644 index 0000000..7f9ca69 --- /dev/null +++ b/eventbus.py @@ -0,0 +1,288 @@ +#!/usr/bin/env python3 +""" +Vector Zulu — EventBus Emitter +Emits events to KoreNet Azure Service Bus per welcome pack spec. + +Namespace: korenet-servicebus (RG korenetprod) +Topic: integration-partners (VZ's scoped topic) + +Envelope format per section 4 of VZ-Welcome-Pack FINAL v1.0: +{ + "correlationId": "", + "causationId": "", + "tenantId": "", + "source": "vector-zulu", + "severity": "LOW|MEDIUM|HIGH|CRITICAL", + "timestamp": "", + "type": "", + "payload": { ... } +} + +VZ emits to integration-partners only. +KoreNet fans into internal error topics on its side. +""" + +import os +import json +import uuid +import logging +from datetime import datetime, timezone +from enum import Enum +from typing import Optional + +log = logging.getLogger(__name__) + + +# ── Event types ─────────────────────────────────────────────────────────────── + +class EventType: + # Payment mirror + MIRROR_MINT_SUCCESS = 'vz.mirror.mint.success' + MIRROR_MINT_FAILED = 'vz.mirror.mint.failed' + MIRROR_SETTLE_SUCCESS = 'vz.mirror.settle.success' + MIRROR_SETTLE_FAILED = 'vz.mirror.settle.failed' + MIRROR_REDEEM_SUCCESS = 'vz.mirror.redeem.success' + MIRROR_REDEEM_FAILED = 'vz.mirror.redeem.failed' + MIRROR_BURN_SUCCESS = 'vz.mirror.burn.success' + MIRROR_BURN_FAILED = 'vz.mirror.burn.failed' + MIRROR_CLOSE_SUCCESS = 'vz.mirror.close.success' + MIRROR_CLOSE_FAILED = 'vz.mirror.close.failed' + # KoreID + KOREID_VALID = 'vz.koreid.valid' + KOREID_INVALID = 'vz.koreid.invalid' + KOREID_GENESIS_ANCHORED = 'vz.koreid.genesis_anchored' + KOREID_NOT_ANCHORED = 'vz.koreid.not_anchored' + # Oracle + ORACLE_TRANSMIT_SUCCESS = 'vz.oracle.transmit.success' + ORACLE_TRANSMIT_FAILED = 'vz.oracle.transmit.failed' + ORACLE_STALE = 'vz.oracle.stale' + # Vault + VAULT_BALANCE_UPDATED = 'vz.vault.balance.updated' + VAULT_DELTA_DETECTED = 'vz.vault.delta.detected' + VAULT_FROZEN = 'vz.vault.frozen' + # Compliance + SANCTIONS_CLEAR = 'vz.compliance.sanctions.clear' + SANCTIONS_HIT = 'vz.compliance.sanctions.hit' + # Ethereum + ETHEREUM_ANCHOR_SUCCESS = 'vz.ethereum.anchor.success' + ETHEREUM_ANCHOR_FAILED = 'vz.ethereum.anchor.failed' + # Infrastructure + ENDPOINT_DOWN = 'vz.infra.endpoint.down' + ENDPOINT_RESTORED = 'vz.infra.endpoint.restored' + CERT_EXPIRY_WARNING = 'vz.infra.cert.expiry_warning' + + +class Severity(str, Enum): + LOW = 'LOW' + MEDIUM = 'MEDIUM' + HIGH = 'HIGH' + CRITICAL = 'CRITICAL' + + +# ── EventBus client ─────────────────────────────────────────────────────────── + +class EventBusClient: + """ + Azure Service Bus client for VZ's integration-partners topic. + Uses VZ's own scoped identity — no shared connection string. + + Connection is provisioned via VZ's Azure service principal. + Namespace: korenet-servicebus + Topic: integration-partners + """ + + NAMESPACE = 'korenet-servicebus' + TOPIC = 'integration-partners' + TENANT_ID = 'vector-zulu' + + def __init__(self, connection_string: str = None): + """ + connection_string: Azure Service Bus connection string + Delivered via secure provisioning channel, not hardcoded. + Falls back to AZURE_SERVICEBUS_CONNECTION_STRING env var. + """ + self.connection_string = ( + connection_string or + os.environ.get('AZURE_SERVICEBUS_CONNECTION_STRING') + ) + self._client = None + self._sender = None + self._connected = False + + if self.connection_string: + self._connect() + else: + log.warning("EventBus: no connection string — events will be queued locally") + + def _connect(self): + try: + from azure.servicebus import ServiceBusClient as AzureSBClient + self._client = AzureSBClient.from_connection_string(self.connection_string) + self._sender = self._client.get_topic_sender(self.TOPIC) + self._connected = True + log.info(f"EventBus connected: {self.NAMESPACE}/{self.TOPIC}") + except ImportError: + log.warning("azure-servicebus not installed — pip install azure-servicebus") + except Exception as e: + log.error(f"EventBus connection failed: {e}") + + def _build_envelope( + self, + event_type: str, + payload: dict, + severity: Severity = Severity.LOW, + causation_id: str = None, + correlation_id: str = None, + ) -> dict: + """Build the standard VZ event envelope per section 4 of welcome pack.""" + return { + 'correlationId': correlation_id or str(uuid.uuid4()), + 'causationId': causation_id or str(uuid.uuid4()), + 'tenantId': self.TENANT_ID, + 'source': 'vector-zulu', + 'severity': severity.value, + 'timestamp': datetime.now(timezone.utc).isoformat(), + 'type': event_type, + 'payload': payload, + } + + def emit( + self, + event_type: str, + payload: dict, + severity: Severity = Severity.LOW, + causation_id: str = None, + correlation_id: str = None, + ) -> bool: + """ + Emit an event to the integration-partners topic. + Returns True on success, False on failure. + Falls back to local Redis queue if Service Bus unavailable. + """ + envelope = self._build_envelope( + event_type, payload, severity, causation_id, correlation_id + ) + + if self._connected and self._sender: + try: + from azure.servicebus import ServiceBusMessage + msg = ServiceBusMessage( + json.dumps(envelope), + content_type='application/json', + subject=event_type, + ) + self._sender.send_messages(msg) + log.info(f"EventBus emitted: {event_type} [{severity.value}]") + return True + except Exception as e: + log.error(f"EventBus emit failed: {e}") + self._queue_locally(envelope) + return False + else: + self._queue_locally(envelope) + return False + + def _queue_locally(self, envelope: dict): + """Queue event locally when Service Bus unavailable.""" + try: + import redis + r = redis.Redis( + host=os.environ.get('REDIS_HOST', 'localhost'), + port=6379, decode_responses=True + ) + r.lpush('eventbus:outbound_queue', json.dumps(envelope)) + log.info(f"EventBus queued locally: {envelope['type']}") + except Exception as e: + log.error(f"Local queue failed: {e}") + + def close(self): + if self._sender: + self._sender.close() + if self._client: + self._client.close() + + # ── Convenience methods ─────────────────────────────────────────────────── + + def emit_mirror_success(self, step: str, uetr: str, koreid: str, + amount: float, currency: str): + event_map = { + 'mint': EventType.MIRROR_MINT_SUCCESS, + 'settle': EventType.MIRROR_SETTLE_SUCCESS, + 'redeem': EventType.MIRROR_REDEEM_SUCCESS, + 'burn': EventType.MIRROR_BURN_SUCCESS, + 'close': EventType.MIRROR_CLOSE_SUCCESS, + } + self.emit( + event_map.get(step, EventType.MIRROR_MINT_SUCCESS), + {'uetr': uetr, 'koreid': koreid, 'amount': amount, 'currency': currency}, + Severity.LOW + ) + + def emit_mirror_failed(self, step: str, uetr: str, error: str): + event_map = { + 'mint': EventType.MIRROR_MINT_FAILED, + 'settle': EventType.MIRROR_SETTLE_FAILED, + 'redeem': EventType.MIRROR_REDEEM_FAILED, + 'burn': EventType.MIRROR_BURN_FAILED, + 'close': EventType.MIRROR_CLOSE_FAILED, + } + self.emit( + event_map.get(step, EventType.MIRROR_MINT_FAILED), + {'uetr': uetr, 'error': error}, + Severity.HIGH + ) + + def emit_koreid_invalid(self, koreid: str, reason: str, uetr: str = None): + self.emit( + EventType.KOREID_INVALID, + {'koreid': koreid, 'reason': reason, 'uetr': uetr}, + Severity.HIGH + ) + + def emit_sanctions_hit(self, entity: str, jurisdiction: str, uetr: str): + self.emit( + EventType.SANCTIONS_HIT, + {'entity': entity, 'jurisdiction': jurisdiction, 'uetr': uetr}, + Severity.CRITICAL + ) + + def emit_oracle_stale(self, age_minutes: int, balance: float): + self.emit( + EventType.ORACLE_STALE, + {'age_minutes': age_minutes, 'balance': balance}, + Severity.HIGH + ) + + def emit_ethereum_anchor(self, koreid: str, tx_hash: str, + block_number: int, success: bool): + self.emit( + EventType.ETHEREUM_ANCHOR_SUCCESS if success else EventType.ETHEREUM_ANCHOR_FAILED, + {'koreid': koreid, 'tx_hash': tx_hash, 'block_number': block_number}, + Severity.LOW if success else Severity.HIGH + ) + + def emit_endpoint_status(self, endpoint: str, status: str, http_code: int): + self.emit( + EventType.ENDPOINT_RESTORED if status == 'UP' else EventType.ENDPOINT_DOWN, + {'endpoint': endpoint, 'status': status, 'http_code': http_code}, + Severity.LOW if status == 'UP' else Severity.MEDIUM + ) + + +if __name__ == '__main__': + logging.basicConfig(level=logging.INFO) + + # Test without Service Bus connection + eb = EventBusClient() + + print("Testing EventBus emitter...") + eb.emit_mirror_success( + 'mint', + uetr='550e8400-e29b-41d4-a716-446655440000', + koreid='KID-MINT-MIRROR-20260709-75f1c3c9ba36d3af', + amount=10000000, + currency='USD' + ) + eb.emit_sanctions_hit('Test Entity', 'ZA', '550e8400-e29b-41d4-a716-446655440000') + eb.emit_oracle_stale(125, 49918000000) + print("Done — events queued locally (no Service Bus connection)")