#!/usr/bin/env python3 """ Vector Zulu — ULL Transaction Tracker Tracks transactions through the KoreNet Universal Ledger Layer. ULL status progression: POSTED → SETTLED → SYNCED → ANCHORED → VERIFIED Monitors all VZ transactions to ensure full fabric compliance. Queues KoreChain anchor at each stage transition. Triggers Ethereum anchor on VERIFIED. Runs on api.korenet.cloud (requires AKS pods up). """ import os, json, time, logging, requests, redis, psycopg2, urllib3 from datetime import datetime, timezone from lick import build_headers urllib3.disable_warnings() logging.basicConfig(level=logging.INFO, format='%(asctime)s [ULL] %(levelname)s %(message)s') log = logging.getLogger(__name__) API_PRIMARY = os.environ.get('KORENET_API_URL_FULL', 'https://api.korenet.cloud') API_FALLBACK = os.environ.get('KORENET_API_URL', 'https://korechain.korenet.cloud') POLL_INTERVAL = int(os.environ.get('ULL_POLL_INTERVAL', '30')) TRACKING_KEY = 'ull:tracking' # Redis hash — uetr → {koreid, status, last_checked} TENANT_ID = 'vector-zulu' # ULL status order — for detecting progression STATUS_ORDER = ['POSTED', 'SETTLED', 'SYNCED', 'ANCHORED', 'VERIFIED'] TERMINAL_STATUSES = {'VERIFIED', 'FAILED', 'REJECTED', 'RECALLED'} def load_jwt() -> str: jwt = os.environ.get('VECTOR_ZULU_JWT', '') if jwt: return jwt for path in ['/app/.secrets', '/home/constellationnode/.korenet/vector-zulu/.secrets', os.path.expanduser('~/.korenet/vector-zulu/.secrets')]: if os.path.exists(path): with open(path) as f: for line in f: if 'VECTOR_ZULU_JWT=' in line: val = line.split('VECTOR_ZULU_JWT=', 1)[1].strip().strip('"').strip("'") if val: return val return '' def get_redis(): return redis.Redis( host=os.environ.get('REDIS_HOST', 'localhost'), port=6379, decode_responses=True) def get_pg(): return psycopg2.connect( host=os.environ.get('POSTGRES_HOST', 'localhost'), dbname=os.environ.get('POSTGRES_DB', 'vz_oracle'), user=os.environ.get('POSTGRES_USER', 'vz_oracle'), password=os.environ.get('POSTGRES_PASSWORD', ''), connect_timeout=10) def _session(jwt: str) -> requests.Session: s = requests.Session() s.verify = False s.headers.update({ 'Authorization': f'Bearer {jwt}', 'X-Tenant-Id': TENANT_ID, 'Content-Type': 'application/json', }) return s def check_ull_status(uetr: str, jwt: str) -> dict: """Check ULL status for a UETR.""" session = _session(jwt) for base in [API_PRIMARY, API_FALLBACK]: for path in [ f'/api/ull/transaction/{uetr}', f'/api/v1/ull/transaction/{uetr}', f'/api/v1/ledger/transaction/{uetr}', ]: try: resp = session.get(f'{base}{path}', timeout=10) if resp.status_code == 200: return resp.json() except Exception as e: log.debug(f"ULL check error via {base}{path}: {e}") return {} def register_transaction(r, uetr: str, koreid: str, amount: float, currency: str, event_type: str = 'ACH_TRANSFER'): """Register a transaction for ULL tracking.""" payload = json.dumps({ 'uetr': uetr, 'koreid': koreid, 'amount': amount, 'currency': currency, 'event_type': event_type, 'status': 'POSTED', 'registered_at': datetime.now(timezone.utc).isoformat(), }) r.hset(TRACKING_KEY, uetr, payload) log.info(f"ULL tracking registered: {uetr} | {koreid}") def init_schema(pg): """Create ULL tracking table.""" try: with pg.cursor() as cur: cur.execute(""" CREATE TABLE IF NOT EXISTS ull_transactions ( id BIGSERIAL PRIMARY KEY, uetr TEXT NOT NULL UNIQUE, koreid TEXT, event_type TEXT, amount NUMERIC(28,8), currency TEXT, ull_status TEXT NOT NULL DEFAULT 'POSTED', core_status TEXT, kore_status TEXT, last_checked TIMESTAMPTZ, verified_at TIMESTAMPTZ, raw_ull JSONB, created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() ); CREATE INDEX IF NOT EXISTS idx_ull_status ON ull_transactions(ull_status); CREATE INDEX IF NOT EXISTS idx_ull_uetr ON ull_transactions(uetr); """) pg.commit() log.info("ULL schema ready") except Exception as e: log.error(f"Schema init: {e}") pg.rollback() def update_pg_ull(pg, uetr: str, status: str, data: dict): try: with pg.cursor() as cur: cur.execute(""" INSERT INTO ull_transactions (uetr, ull_status, core_status, kore_status, last_checked, raw_ull, verified_at) VALUES (%s, %s, %s, %s, %s, %s, %s) ON CONFLICT (uetr) DO UPDATE SET ull_status = EXCLUDED.ull_status, core_status = EXCLUDED.core_status, kore_status = EXCLUDED.kore_status, last_checked = EXCLUDED.last_checked, raw_ull = EXCLUDED.raw_ull, verified_at = CASE WHEN EXCLUDED.ull_status = 'VERIFIED' THEN EXCLUDED.verified_at ELSE ull_transactions.verified_at END """, ( uetr, status, data.get('CoreStatus'), data.get('KoreStatus'), datetime.now(timezone.utc), json.dumps(data), datetime.now(timezone.utc) if status == 'VERIFIED' else None )) pg.commit() except Exception as e: log.error(f"ULL DB error for {uetr}: {e}") try: pg.rollback() except: pass class UllTracker: def __init__(self): self.r = get_redis() self.pg = get_pg() init_schema(self.pg) log.info(f"ULL Tracker starting — polling every {POLL_INTERVAL}s") log.info(f"Primary: {API_PRIMARY} | Fallback: {API_FALLBACK}") def check_all(self, jwt: str): """Check ULL status for all tracked transactions.""" tracking = self.r.hgetall(TRACKING_KEY) if not tracking: log.debug("No tracked transactions") return log.info(f"Checking ULL for {len(tracking)} transactions") for uetr, raw in tracking.items(): try: item = json.loads(raw) koreid = item.get('koreid', '') prev_status = item.get('status', 'POSTED') # Skip already terminal if prev_status in TERMINAL_STATUSES: continue ull_data = check_ull_status(uetr, jwt) if not ull_data: log.debug(f"No ULL data for {uetr} — api.korenet.cloud may be down") continue # Extract status new_status = (ull_data.get('KoreStatus') or ull_data.get('CoreStatus') or ull_data.get('status') or prev_status).upper() if new_status != prev_status: log.info(f"ULL progression: {uetr} {prev_status} → {new_status}") # Update Redis item['status'] = new_status item['last_checked'] = datetime.now(timezone.utc).isoformat() self.r.hset(TRACKING_KEY, uetr, json.dumps(item)) # Update Postgres update_pg_ull(self.pg, uetr, new_status, ull_data) # Queue KoreChain anchor for status transition anchor = { 'koreid': f'{koreid}-ULL-{new_status}', 'linked_to': koreid, 'uetr': uetr, 'step': f'ULL_{new_status}', 'amount': item.get('amount', 0), 'currency': item.get('currency', 'USD'), 'anchored_at': datetime.now(timezone.utc).isoformat(), 'genesis_root': 'GENESIS-07', 'ull_status': new_status, } self.r.lpush('korechain:outbound_queue', json.dumps(anchor)) log.info(f"Queued ULL transition anchor: {uetr} → {new_status}") # On VERIFIED — trigger full Ethereum anchor if new_status == 'VERIFIED': log.info(f"✓ VERIFIED: {uetr} | {koreid} — full fabric loop complete") # Remove from active tracking self.r.hdel(TRACKING_KEY, uetr) else: log.debug(f"ULL {uetr}: no change ({new_status})") update_pg_ull(self.pg, uetr, new_status, ull_data) except Exception as e: log.error(f"ULL error for {uetr}: {e}", exc_info=True) def run(self): while True: jwt = load_jwt() if not jwt: log.error("No JWT — waiting 60s") time.sleep(60) continue self.check_all(jwt) time.sleep(POLL_INTERVAL) if __name__ == '__main__': tracker = UllTracker() tracker.run()