289 lines
10 KiB
Python
289 lines
10 KiB
Python
#!/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)")
|