amitis55/vz-kore-adapter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add PACS.002 listener, CAMT.054 listener, ULL tracker - ISO 20022 full loop

Browse Source
This commit is contained in:
VZ Build
2026-07-15 03:15:38 +00:00
parent 6c38390461
commit a30a78051b
4 changed files with 864 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

295
camt054_listener.py Normal file
View File

@@ -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()

282
pacs002_listener.py Normal file
View File

@@ -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()

22
portal_feed_poller.py
View File

@@ -22,6 +22,15 @@ from koreid import KoreIDGenerator, KoreIDParser, KoreType, KoreDomain, GENESIS_
from eventbus import EventBusClient, Severity, EventType from eventbus import EventBusClient, Severity, EventType
from lick import build_headers, canonicalize 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() urllib3.disable_warnings()
logging.basicConfig(level=logging.INFO, format='%(asctime)s [POLLER-V3] %(levelname)s %(message)s') logging.basicConfig(level=logging.INFO, format='%(asctime)s [POLLER-V3] %(levelname)s %(message)s')
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -29,6 +38,7 @@ log = logging.getLogger(__name__)
# ── Config ──────────────────────────────────────────────────────────────────── # ── Config ────────────────────────────────────────────────────────────────────
API_PRIMARY = os.environ.get('KORENET_API_URL', 'https://korechain.korenet.cloud') 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') 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') RAILS_URL = os.environ.get('VECTOR_ZULU_RAILS_URL', 'https://vector.korenet.cloud')
BENJII_URL = 'https://benjii-prime.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 ────────────────────────────────────────────────────────── # ── 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, def mirror_mint(jwt: str, uetr: str, asset: str, amount: float,
rail_ref: str, vault: str) -> dict: rail_ref: str, vault: str) -> dict:
log.info(f"MINT {amount} {asset} UETR:{uetr}") log.info(f"MINT {amount} {asset} UETR:{uetr}")

265
ull_tracker.py Normal file
View File

@@ -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()