From a30a78051bab323235fb0cb801d80a277d27317d Mon Sep 17 00:00:00 2001 From: VZ Build Date: Wed, 15 Jul 2026 03:15:38 +0000 Subject: [PATCH] Add PACS.002 listener, CAMT.054 listener, ULL tracker - ISO 20022 full loop --- camt054_listener.py | 295 ++++++++++++++++++++++++++++++++++++++++++ pacs002_listener.py | 282 ++++++++++++++++++++++++++++++++++++++++ portal_feed_poller.py | 22 ++++ ull_tracker.py | 265 +++++++++++++++++++++++++++++++++++++ 4 files changed, 864 insertions(+) create mode 100644 camt054_listener.py create mode 100644 pacs002_listener.py create mode 100644 ull_tracker.py diff --git a/camt054_listener.py b/camt054_listener.py new file mode 100644 index 0000000..1114223 --- /dev/null +++ b/camt054_listener.py @@ -0,0 +1,295 @@ +#!/usr/bin/env python3 +""" +Vector Zulu — CAMT.054 Debit/Credit Notification Listener +Polls KoreNet for CAMT.054 bank-to-customer debit/credit notifications. + +CAMT.054 is the ISO 20022 notification that tells us when our account +has been debited (outbound payment executed) or credited (inbound payment received). +This closes the loop on outbound PACS.008 sends and inbound payment events. + +Runs on api.korenet.cloud (requires AKS pods up). +Also checks CAMT.053 end-of-day statements for reconciliation. +""" + +import os, json, time, logging, requests, redis, psycopg2, urllib3 +from datetime import datetime, timezone, timedelta +from lick import build_headers + +urllib3.disable_warnings() +logging.basicConfig(level=logging.INFO, + format='%(asctime)s [CAMT054] %(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('CAMT054_POLL_INTERVAL', '60')) +TENANT_ID = 'vector-zulu' +VZ_VAULTS = ['VZ-USD-VAULT', 'VZ-USDT-VAULT', 'VZ-BTC-VAULT'] +WF_ACCOUNT = '8618159373' + + +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 fetch_camt054(account_id: str, jwt: str) -> list: + """Fetch CAMT.054 debit/credit notifications for an account.""" + session = _session(jwt) + notifications = [] + + for base in [API_PRIMARY, API_FALLBACK]: + try: + resp = session.get(f'{base}/api/camt/054/{account_id}', timeout=15) + if resp.status_code == 200: + data = resp.json() + entries = (data.get('entries') or data.get('notifications') or + data.get('transactions') or []) + if entries: + log.info(f"CAMT.054 {account_id}: {len(entries)} entries") + notifications.extend(entries) + break + except Exception as e: + log.debug(f"CAMT.054 error via {base}: {e}") + + return notifications + + +def fetch_camt053(account_id: str, jwt: str, date: str = None) -> dict: + """ + Fetch CAMT.053 end-of-day statement for an account. + date format: YYYY-MM-DD, defaults to yesterday + """ + if not date: + date = (datetime.now(timezone.utc) - timedelta(days=1)).strftime('%Y-%m-%d') + + session = _session(jwt) + for base in [API_PRIMARY, API_FALLBACK]: + try: + resp = session.get(f'{base}/api/camt/053/{account_id}', + params={'date': date}, timeout=15) + if resp.status_code == 200: + log.info(f"CAMT.053 {account_id} for {date}: received") + return resp.json() + except Exception as e: + log.debug(f"CAMT.053 error via {base}: {e}") + return {} + + +def process_notification(entry: dict, r, pg, jwt: str): + """ + Process a single CAMT.054 notification entry. + Credit = inbound payment received + Debit = outbound payment executed + """ + entry_type = (entry.get('creditDebitIndicator') or + entry.get('type') or '').upper() + amount = float(entry.get('amount', {}).get('value', 0) if + isinstance(entry.get('amount'), dict) else + entry.get('amount', 0)) + currency = (entry.get('amount', {}).get('currency', 'USD') if + isinstance(entry.get('amount'), dict) else + entry.get('currency', 'USD')) + uetr = entry.get('uetr') or entry.get('endToEndId') or '' + reference = entry.get('reference') or entry.get('remittanceInfo') or '' + value_date = entry.get('valueDate') or entry.get('bookingDate') or '' + entry_ref = entry.get('entryReference') or entry.get('id') or '' + + # Skip if already processed + if r.exists(f'camt054:processed:{entry_ref}'): + return + + log.info(f"CAMT.054 {entry_type}: {currency} {amount:,.2f} " + f"UETR:{uetr} ref:{reference}") + + if entry_type == 'CRDT': + # Credit — inbound payment received + # Trigger mirror lifecycle if UETR not already processed + if uetr and not r.exists(f'mirror:processed:{uetr}'): + log.info(f"New inbound credit — queuing for mirror lifecycle: {uetr}") + r.lpush('camt054:inbound_credits', json.dumps({ + 'uetr': uetr, 'amount': amount, 'currency': currency, + 'reference': reference, 'value_date': value_date, + 'entry_ref': entry_ref, 'source': 'CAMT054', + })) + + elif entry_type == 'DBIT': + # Debit — outbound payment executed + # Confirms our PACS.008 was funded + if uetr: + log.info(f"Outbound debit confirmed via CAMT.054: {uetr}") + # Update PACS.002 pending if exists + r.zadd('pacs002:camt_confirmed', {uetr: time.time()}) + + # Store in Postgres + try: + with pg.cursor() as cur: + cur.execute(""" + INSERT INTO camt054_entries + (entry_ref, uetr, entry_type, amount, currency, + reference, value_date, raw_entry, processed_at) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (entry_ref) DO NOTHING + """, (entry_ref, uetr, entry_type, amount, currency, + reference, value_date, json.dumps(entry), + datetime.now(timezone.utc))) + pg.commit() + except Exception as e: + log.error(f"DB error for entry {entry_ref}: {e}") + try: pg.rollback() + except: pass + + # Mark processed + r.setex(f'camt054:processed:{entry_ref}', 86400 * 7, '1') + + +def init_schema(pg): + """Create CAMT tables if not exists.""" + try: + with pg.cursor() as cur: + cur.execute(""" + CREATE TABLE IF NOT EXISTS camt054_entries ( + id BIGSERIAL PRIMARY KEY, + entry_ref TEXT NOT NULL UNIQUE, + uetr TEXT, + entry_type TEXT, + amount NUMERIC(28,8), + currency TEXT, + reference TEXT, + value_date TEXT, + raw_entry JSONB, + processed_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + CREATE INDEX IF NOT EXISTS idx_camt054_uetr ON camt054_entries(uetr); + CREATE INDEX IF NOT EXISTS idx_camt054_type ON camt054_entries(entry_type); + CREATE INDEX IF NOT EXISTS idx_camt054_date ON camt054_entries(value_date); + + CREATE TABLE IF NOT EXISTS camt053_statements ( + id BIGSERIAL PRIMARY KEY, + account_id TEXT NOT NULL, + statement_date TEXT NOT NULL, + opening_balance NUMERIC(28,8), + closing_balance NUMERIC(28,8), + currency TEXT, + entry_count INT, + raw_statement JSONB, + received_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE(account_id, statement_date) + ); + """) + pg.commit() + log.info("CAMT schema ready") + except Exception as e: + log.error(f"Schema init error: {e}") + pg.rollback() + + +class Camt054Listener: + + def __init__(self): + self.r = get_redis() + self.pg = get_pg() + init_schema(self.pg) + log.info(f"CAMT.054 Listener starting — polling every {POLL_INTERVAL}s") + + def poll_all_accounts(self, jwt: str): + """Poll CAMT.054 for all VZ accounts.""" + accounts = VZ_VAULTS + [WF_ACCOUNT] + for account_id in accounts: + try: + entries = fetch_camt054(account_id, jwt) + for entry in entries: + process_notification(entry, self.r, self.pg, jwt) + except Exception as e: + log.error(f"Error polling CAMT.054 for {account_id}: {e}") + + def fetch_eod_statements(self, jwt: str): + """Fetch end-of-day CAMT.053 statements for reconciliation.""" + # Only run once per day — check if already fetched today + today = datetime.now(timezone.utc).strftime('%Y-%m-%d') + yesterday = (datetime.now(timezone.utc) - timedelta(days=1)).strftime('%Y-%m-%d') + + if self.r.exists(f'camt053:fetched:{yesterday}'): + return + + for account_id in VZ_VAULTS + [WF_ACCOUNT]: + try: + stmt = fetch_camt053(account_id, jwt, yesterday) + if stmt: + with self.pg.cursor() as cur: + cur.execute(""" + INSERT INTO camt053_statements + (account_id, statement_date, opening_balance, + closing_balance, currency, entry_count, raw_statement) + VALUES (%s, %s, %s, %s, %s, %s, %s) + ON CONFLICT (account_id, statement_date) DO NOTHING + """, ( + account_id, yesterday, + stmt.get('openingBalance', 0), + stmt.get('closingBalance', 0), + stmt.get('currency', 'USD'), + stmt.get('entryCount', 0), + json.dumps(stmt) + )) + self.pg.commit() + log.info(f"CAMT.053 stored for {account_id} / {yesterday}") + except Exception as e: + log.error(f"CAMT.053 error for {account_id}: {e}") + + self.r.setex(f'camt053:fetched:{yesterday}', 86400 * 2, '1') + + def run(self): + while True: + jwt = load_jwt() + if not jwt: + log.error("No JWT — waiting 60s") + time.sleep(60) + continue + + self.poll_all_accounts(jwt) + self.fetch_eod_statements(jwt) + time.sleep(POLL_INTERVAL) + + +if __name__ == '__main__': + listener = Camt054Listener() + listener.run() diff --git a/pacs002_listener.py b/pacs002_listener.py new file mode 100644 index 0000000..01cea05 --- /dev/null +++ b/pacs002_listener.py @@ -0,0 +1,282 @@ +#!/usr/bin/env python3 +""" +Vector Zulu — PACS.002 Payment Status Listener +Polls KoreNet for PACS.002 payment status confirmations on outbound PACS.008 sends. + +Status progression per KoreNet ULL: + POSTED → SETTLED → SYNCED → ANCHORED → VERIFIED + +Runs on api.korenet.cloud (requires AKS pods up). +Polls Redis for pending UETRs, checks PACS.002 status, updates Postgres. +Queues KoreID anchor on confirmation. +""" + +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 [PACS002] %(levelname)s %(message)s') +log = logging.getLogger(__name__) + +# When api.korenet.cloud is back up, this is the primary +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('PACS002_POLL_INTERVAL', '15')) +PENDING_KEY = 'pacs002:pending_uetrs' # Redis sorted set — score = submitted_at +CONFIRMED_KEY = 'pacs002:confirmed_uetrs' +TENANT_ID = 'vector-zulu' +MAX_AGE_HOURS = 48 # Stop polling after 48 hours + + +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_pacs002_status(uetr: str, jwt: str) -> dict: + """ + Check PACS.002 status for a UETR. + Tries multiple endpoints in order of preference. + """ + session = _session(jwt) + + # Primary: PACS.002 dedicated endpoint (api.korenet.cloud) + for base in [API_PRIMARY, API_FALLBACK]: + for path in [ + f'/api/swift/pacs002/status/{uetr}', + f'/api/swift/gpi/track/{uetr}', + f'/api/ull/transaction/{uetr}', + f'/api/v1/treasury/transfers/{uetr}', + ]: + try: + resp = session.get(f'{base}{path}', timeout=10) + if resp.status_code == 200: + data = resp.json() + log.debug(f"PACS.002 {uetr} via {base}{path}: {data}") + return {'source': f'{base}{path}', 'data': data} + except Exception as e: + log.debug(f"Error checking {base}{path}: {e}") + + return {} + + +def extract_status(result: dict) -> str: + """Extract normalised status from various response formats.""" + if not result: + return 'UNKNOWN' + data = result.get('data', {}) + + # ULL format + ull_status = data.get('CoreStatus') or data.get('KoreStatus') or data.get('ullStatus') + if ull_status: + return ull_status.upper() + + # GPI format + gpi_status = data.get('transactionStatus') or data.get('gpiStatus') + if gpi_status: + return gpi_status.upper() + + # PACS.002 format + p002_status = data.get('status') or data.get('paymentStatus') + if p002_status: + return p002_status.upper() + + # Treasury transfer format + t_status = data.get('status') or data.get('movementStatus') + if t_status: + return t_status.upper() + + return 'UNKNOWN' + + +def register_pending(r, uetr: str, koreid: str, amount: float, + currency: str, reference: str): + """Register a UETR for PACS.002 monitoring.""" + payload = json.dumps({ + 'uetr': uetr, + 'koreid': koreid, + 'amount': amount, + 'currency': currency, + 'reference': reference, + 'submitted_at': datetime.now(timezone.utc).isoformat(), + }) + r.zadd(PENDING_KEY, {payload: time.time()}) + log.info(f"Registered UETR {uetr} for PACS.002 monitoring") + + +def update_pg_status(pg, uetr: str, status: str, pacs002_data: dict): + """Update Postgres with PACS.002 confirmation status.""" + try: + with pg.cursor() as cur: + cur.execute(""" + INSERT INTO pacs002_confirmations + (uetr, status, pacs002_data, confirmed_at) + VALUES (%s, %s, %s, %s) + ON CONFLICT (uetr) DO UPDATE SET + status = EXCLUDED.status, + pacs002_data = EXCLUDED.pacs002_data, + confirmed_at = EXCLUDED.confirmed_at + """, (uetr, status, json.dumps(pacs002_data), + datetime.now(timezone.utc))) + pg.commit() + except Exception as e: + log.error(f"DB update error for {uetr}: {e}") + try: pg.rollback() + except: pass + + +def init_schema(pg): + """Create PACS.002 confirmations table if not exists.""" + try: + with pg.cursor() as cur: + cur.execute(""" + CREATE TABLE IF NOT EXISTS pacs002_confirmations ( + id BIGSERIAL PRIMARY KEY, + uetr TEXT NOT NULL UNIQUE, + status TEXT NOT NULL DEFAULT 'PENDING', + pacs002_data JSONB, + submitted_at TIMESTAMPTZ, + confirmed_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() + ); + CREATE INDEX IF NOT EXISTS idx_pacs002_status ON pacs002_confirmations(status); + CREATE INDEX IF NOT EXISTS idx_pacs002_uetr ON pacs002_confirmations(uetr); + """) + pg.commit() + log.info("PACS.002 schema ready") + except Exception as e: + log.error(f"Schema init error: {e}") + pg.rollback() + + +class Pacs002Listener: + + TERMINAL_STATUSES = {'SETTLED', 'VERIFIED', 'ANCHORED', 'COMPLETED', + 'ACSC', 'ACCC', 'ACSP'} # ISO 20022 acceptance codes + REJECTED_STATUSES = {'REJECTED', 'RJCT', 'FAILED', 'RETURNED', 'RTRN'} + + def __init__(self): + self.r = get_redis() + self.pg = get_pg() + init_schema(self.pg) + log.info(f"PACS.002 Listener starting — polling every {POLL_INTERVAL}s") + log.info(f"Primary: {API_PRIMARY} | Fallback: {API_FALLBACK}") + + def process_pending(self, jwt: str): + """Check all pending UETRs for PACS.002 confirmation.""" + now = time.time() + max_age = MAX_AGE_HOURS * 3600 + + # Get all pending UETRs + items = self.r.zrangebyscore(PENDING_KEY, 0, '+inf', withscores=True) + + if not items: + log.debug("No pending UETRs") + return + + log.info(f"Checking {len(items)} pending UETRs") + + for raw, submitted_at in items: + try: + item = json.loads(raw) + uetr = item['uetr'] + age = now - submitted_at + + # Expire after MAX_AGE_HOURS + if age > max_age: + log.warning(f"UETR {uetr} expired after {MAX_AGE_HOURS}h — moving to dead letter") + self.r.zadd('pacs002:expired', {raw: now}) + self.r.zrem(PENDING_KEY, raw) + update_pg_status(self.pg, uetr, 'EXPIRED', item) + continue + + # Check status + result = check_pacs002_status(uetr, jwt) + status = extract_status(result) + + log.info(f"UETR {uetr} — status: {status} (age: {age/60:.1f}m)") + + if status in self.TERMINAL_STATUSES: + log.info(f"✓ UETR {uetr} CONFIRMED: {status}") + self.r.zadd(CONFIRMED_KEY, {raw: now}) + self.r.zrem(PENDING_KEY, raw) + update_pg_status(self.pg, uetr, status, result.get('data', {})) + + # Queue KoreChain anchor update with confirmed status + from datetime import datetime, timezone + anchor_update = { + 'koreid': item.get('koreid', ''), + 'uetr': uetr, + 'step': f'PACS002_{status}', + 'amount': item.get('amount', 0), + 'currency': item.get('currency', 'USD'), + 'linked_to': item.get('koreid', 'GENESIS-07'), + 'anchored_at': datetime.now(timezone.utc).isoformat(), + 'genesis_root': 'GENESIS-07', + 'pacs002_status': status, + } + self.r.lpush('korechain:outbound_queue', json.dumps(anchor_update)) + log.info(f"Queued PACS.002 confirmation anchor for {uetr}") + + elif status in self.REJECTED_STATUSES: + log.error(f"✗ UETR {uetr} REJECTED: {status}") + self.r.zrem(PENDING_KEY, raw) + update_pg_status(self.pg, uetr, status, result.get('data', {})) + + except Exception as e: + log.error(f"Error processing 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.process_pending(jwt) + time.sleep(POLL_INTERVAL) + + +if __name__ == '__main__': + listener = Pacs002Listener() + listener.run() diff --git a/portal_feed_poller.py b/portal_feed_poller.py index e8e24d1..677bd76 100644 --- a/portal_feed_poller.py +++ b/portal_feed_poller.py @@ -22,6 +22,15 @@ from koreid import KoreIDGenerator, KoreIDParser, KoreType, KoreDomain, GENESIS_ from eventbus import EventBusClient, Severity, EventType from lick import build_headers, canonicalize +# Import ISO 20022 trackers (fail gracefully if not available) +try: + from pacs002_listener import register_pending as pacs002_register + from ull_tracker import register_transaction as ull_register + ISO20022_TRACKING = True +except ImportError: + ISO20022_TRACKING = False + log.warning("ISO 20022 trackers not available — PACS.002/ULL tracking disabled") + urllib3.disable_warnings() logging.basicConfig(level=logging.INFO, format='%(asctime)s [POLLER-V3] %(levelname)s %(message)s') log = logging.getLogger(__name__) @@ -29,6 +38,7 @@ log = logging.getLogger(__name__) # ── Config ──────────────────────────────────────────────────────────────────── API_PRIMARY = os.environ.get('KORENET_API_URL', 'https://korechain.korenet.cloud') +API_ISO20022 = os.environ.get('KORENET_API_URL_FULL', 'https://api.korenet.cloud') # PACS/CAMT/ULL API_FALLBACK = os.environ.get('KORENET_API_FALLBACK', 'https://fnb.korenet.cloud') RAILS_URL = os.environ.get('VECTOR_ZULU_RAILS_URL', 'https://vector.korenet.cloud') BENJII_URL = 'https://benjii-prime.korenet.cloud' @@ -212,6 +222,18 @@ def sanctions_check(jwt: str, entity: str, jurisdiction: str = 'GLOBAL') -> bool # ── Mirror lifecycle ────────────────────────────────────────────────────────── +def register_iso20022(r, uetr: str, koreid: str, amount: float, currency: str): + """Register transaction with PACS.002 and ULL trackers.""" + if not ISO20022_TRACKING: + return + try: + pacs002_register(r, uetr, koreid, amount, currency, f'VZ-{koreid}') + ull_register(r, uetr, koreid, amount, currency) + log.info(f"Registered {uetr} for PACS.002 + ULL tracking") + except Exception as e: + log.error(f"ISO 20022 registration error: {e}") + + def mirror_mint(jwt: str, uetr: str, asset: str, amount: float, rail_ref: str, vault: str) -> dict: log.info(f"MINT {amount} {asset} UETR:{uetr}") diff --git a/ull_tracker.py b/ull_tracker.py new file mode 100644 index 0000000..a591905 --- /dev/null +++ b/ull_tracker.py @@ -0,0 +1,265 @@ +#!/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()