Add eventbus.py
This commit is contained in:
288
eventbus.py
Normal file
288
eventbus.py
Normal file
@@ -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": "<uuid>",
|
||||
"causationId": "<uuid>",
|
||||
"tenantId": "<tenant>",
|
||||
"source": "vector-zulu",
|
||||
"severity": "LOW|MEDIUM|HIGH|CRITICAL",
|
||||
"timestamp": "<ISO8601 UTC>",
|
||||
"type": "<event-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)")
|
||||
Reference in New Issue
Block a user