#!/usr/bin/env python3 """ Vector Zulu — KoreChain Poster v3.0 Reads KoreID payloads from Redis queue and posts them to KoreChain. Key changes from v2: - Primary endpoint: korechain.korenet.cloud (not api.korenet.cloud) - Auth: JWT Bearer only — no client cert required (ssl_verify_client optional) - LICK signing on all mutating calls - Uses /api/v1/korechain/anchor for Ethereum anchor confirmations - Uses /api/korechain/v1/broadcast for broadcast events - Retry with dead-letter after MAX_RETRIES """ import os, json, time, logging, requests, redis, psycopg2, urllib3 from datetime import datetime, timezone from lick import build_headers # Ethereum anchoring — import conditionally (web3 may not be installed) try: from eth_anchor import get_anchor ETH_ANCHOR_ENABLED = True except ImportError: ETH_ANCHOR_ENABLED = False import logging logging.getLogger(__name__).warning("eth_anchor not available — Ethereum anchoring disabled") urllib3.disable_warnings() logging.basicConfig(level=logging.INFO, format='%(asctime)s [KORE-POSTER] %(levelname)s %(message)s') log = logging.getLogger(__name__) API_PRIMARY = os.environ.get('KORENET_API_URL', 'https://korechain.korenet.cloud') API_FALLBACK = os.environ.get('KORENET_API_FALLBACK', 'https://fnb.korenet.cloud') QUEUE_KEY = 'korechain:outbound_queue' DEAD_LETTER = 'korechain:dead_letter' MAX_RETRIES = int(os.environ.get('MAX_RETRIES', '3')) RETRY_DELAY = int(os.environ.get('RETRY_DELAY', '30')) POLL_INTERVAL = int(os.environ.get('POLL_INTERVAL', '5')) TENANT_ID = 'vector-zulu' 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 _post_to_korechain(payload: dict, jwt: str, path: str) -> tuple: body = {**payload, 'tenantId': TENANT_ID, 'source': 'vector-zulu', 'timestamp': datetime.now(timezone.utc).isoformat()} lick = build_headers(body) session = _session(jwt) session.headers.update(lick) for base_url in [API_PRIMARY, API_FALLBACK]: url = f'{base_url}{path}' try: resp = session.post(url, json=body, timeout=30) if resp.status_code in (200, 201, 202): data = resp.json() if resp.content else {} response_koreid = data.get('koreId') or data.get('koreid') or '' log.info(f"✓ Posted {payload.get('koreid')} via {base_url} → {resp.status_code}") return True, response_koreid else: log.warning(f"Post {payload.get('koreid')} via {base_url}: HTTP {resp.status_code}") except Exception as e: log.error(f"Post error via {base_url}: {e}") return False, '' def post_to_korechain(payload: dict, jwt: str) -> tuple: step = payload.get('step', 'ANCHOR') if step in ('MINT', 'SETTLE', 'REDEEM', 'BURN', 'CLOSE'): path = '/api/korechain/v1/broadcast' else: path = '/api/v1/korechain/anchor' return _post_to_korechain(payload, jwt, path) def store_posted(pg, payload: dict, status: str, response_koreid: str = ''): try: with pg.cursor() as cur: cur.execute(""" INSERT INTO kore_outbound_events (koreid, guid, event_type, currency, token_symbol, tx_hash, amount, status, response_koreid, posted_at) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT (koreid) DO UPDATE SET status = EXCLUDED.status, response_koreid = EXCLUDED.response_koreid, posted_at = EXCLUDED.posted_at """, ( payload.get('koreid', ''), payload.get('guid', ''), payload.get('step', 'ANCHOR'), payload.get('currency', ''), payload.get('token_symbol', ''), payload.get('ethereum_tx', ''), payload.get('amount', 0), status, response_koreid, datetime.now(timezone.utc), )) pg.commit() except Exception as e: log.error(f"DB store error: {e}") try: pg.rollback() except: pass class KoreChainPoster: def __init__(self, redis_client, pg_conn): self.r = redis_client self.pg = pg_conn def process_one(self, jwt: str) -> bool: raw = self.r.rpop(QUEUE_KEY) if not raw: return False try: payload = json.loads(raw) except json.JSONDecodeError as e: log.error(f"Invalid JSON in queue: {e}") return True koreid = payload.get('koreid', 'UNKNOWN') retries = payload.get('_retries', 0) log.info(f"Processing {koreid} step={payload.get('step','ANCHOR')} (attempt {retries+1}/{MAX_RETRIES})") success, response_koreid = post_to_korechain(payload, jwt) if success: store_posted(self.pg, payload, 'POSTED', response_koreid) log.info(f"✓ {koreid} posted") # Anchor to Ethereum mainnet if ETH_ANCHOR_ENABLED: try: anchor = get_anchor() tx_hash = anchor.anchor( kore_id=koreid, uetr=payload.get('uetr', ''), event_type=payload.get('step', 'ANCHOR'), amount=float(payload.get('amount', 0)), currency=payload.get('currency', 'USD'), linked_to=payload.get('linked_to', 'GENESIS-07'), payload=payload, ) if tx_hash and tx_hash != 'already_anchored': log.info(f"⛓ Ethereum anchor tx: {tx_hash}") store_posted(self.pg, {**payload, 'ethereum_tx': tx_hash}, 'ETH_ANCHORED', tx_hash) elif tx_hash == 'already_anchored': log.info(f"⛓ Already on Ethereum: {koreid}") except Exception as e: log.error(f"Ethereum anchor error for {koreid}: {e}") else: retries += 1 if retries < MAX_RETRIES: payload['_retries'] = retries self.r.zadd('korechain:retry_queue', {json.dumps(payload): time.time() + RETRY_DELAY}) log.warning(f"Retry {retries}/{MAX_RETRIES} for {koreid} in {RETRY_DELAY}s") else: self.r.lpush(DEAD_LETTER, raw) store_posted(self.pg, payload, 'DEAD_LETTER') log.error(f"Dead letter: {koreid}") return True def process_retries(self, jwt: str): now = time.time() for item in self.r.zrangebyscore('korechain:retry_queue', 0, now): self.r.lpush(QUEUE_KEY, item) self.r.zrem('korechain:retry_queue', item) log.info("Moved retry to main queue") def run(self): log.info(f"KoreChain Poster v3.0 — Primary: {API_PRIMARY} | Fallback: {API_FALLBACK}") while True: jwt = load_jwt() if not jwt: log.error("No JWT — waiting 60s") time.sleep(60) continue self.process_retries(jwt) while self.process_one(jwt): pass time.sleep(POLL_INTERVAL) if __name__ == '__main__': poster = KoreChainPoster(get_redis(), get_pg()) poster.run()