266 lines
9.7 KiB
Python
266 lines
9.7 KiB
Python
#!/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()
|