#!/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)")