diff --git a/korechain_poster.py b/korechain_poster.py new file mode 100644 index 0000000..af28067 --- /dev/null +++ b/korechain_poster.py @@ -0,0 +1,204 @@ +#!/usr/bin/env python3 +""" +Vector Zulu — KoreChain Poster +Reads KoreID payloads from Redis queue and posts them to KoreChain via KoreOracle API. +Preserves full KoreID traceability per spec — never strips hash, signature, or chain reference. +""" + +import os, json, time, logging, requests +from datetime import datetime, timezone +import redis +import psycopg2 +import urllib3 +urllib3.disable_warnings() + +logging.basicConfig( + level=logging.INFO, + format='%(asctime)s [KORE-POSTER] %(levelname)s %(message)s' +) +log = logging.getLogger(__name__) + +# ── Config ──────────────────────────────────────────────────────────────────── + +CERT = os.environ.get('CERT_PATH', '/certs/vector-zulu-client.fullchain.pem') +KEY = os.environ.get('KEY_PATH', '/certs/vector-zulu-client.key') +API_BASE = 'https://api.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')) + + +def load_jwt(): + secrets_path = '/app/.secrets' + if os.path.exists(secrets_path): + with open(secrets_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 None + + +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 post_to_korechain(payload: dict, jwt: str) -> bool: + """ + Post a KoreID payload to KoreChain via KoreOracle API. + Returns True on success, False on failure. + """ + url = f'{API_BASE}/api/external/event' + + headers = { + 'Authorization': f'Bearer {jwt}', + 'X-Tenant-Id': 'vector-zulu', + 'Content-Type': 'application/json', + } + + try: + resp = requests.post( + url, + cert=(CERT, KEY), + verify=False, + headers=headers, + json=payload, + timeout=30 + ) + + if resp.status_code in (200, 201, 202): + log.info(f"Posted KoreID {payload['koreid']} → HTTP {resp.status_code}") + return True + else: + log.warning(f"KoreChain rejected {payload['koreid']}: HTTP {resp.status_code} — {resp.text[:200]}") + return False + + except Exception as e: + log.error(f"Post error for {payload.get('koreid','?')}: {e}") + return False + + +def store_posted(payload: dict, pg, status: str, response_koreid: str = None): + """Record posted event in Postgres for audit trail.""" + 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['koreid'], + payload['guid'], + payload['source']['event_type'], + payload['source'].get('token', ''), + payload['source'].get('token', ''), + payload['proof']['hash'], + payload.get('value', 0), + status, + response_koreid, + datetime.now(timezone.utc), + )) + pg.commit() + except Exception as e: + log.error(f"DB store error: {e}") + + +class KoreChainPoster: + + def __init__(self, redis_client, pg_conn): + self.r = redis_client + self.pg = pg_conn + + def process_one(self, jwt: str) -> bool: + """Process one item from the queue. Returns True if item was processed.""" + 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} (attempt {retries + 1}/{MAX_RETRIES})") + + success = post_to_korechain(payload, jwt) + + if success: + store_posted(payload, self.pg, 'POSTED') + log.info(f"✓ {koreid} posted successfully") + else: + retries += 1 + if retries < MAX_RETRIES: + payload['_retries'] = retries + # Re-queue with delay via sorted set + retry_at = time.time() + RETRY_DELAY + self.r.zadd('korechain:retry_queue', {json.dumps(payload): retry_at}) + log.warning(f"Retry {retries}/{MAX_RETRIES} for {koreid} in {RETRY_DELAY}s") + else: + # Dead letter + self.r.lpush(DEAD_LETTER, raw) + store_posted(payload, self.pg, 'DEAD_LETTER') + log.error(f"Dead letter: {koreid} after {MAX_RETRIES} attempts") + + return True + + def process_retries(self, jwt: str): + """Move any due retries back to main queue.""" + now = time.time() + due = self.r.zrangebyscore('korechain:retry_queue', 0, now) + for item in due: + self.r.lpush(QUEUE_KEY, item) + self.r.zrem('korechain:retry_queue', item) + log.info("Moved retry back to main queue") + + def run(self): + log.info("KoreChain poster starting...") + log.info(f"Queue: {QUEUE_KEY} | Max retries: {MAX_RETRIES} | Retry delay: {RETRY_DELAY}s") + + while True: + jwt = load_jwt() + if not jwt: + log.error("No JWT — waiting 60s") + time.sleep(60) + continue + + self.process_retries(jwt) + + processed = True + while processed: + processed = self.process_one(jwt) + + time.sleep(POLL_INTERVAL) + + +if __name__ == '__main__': + r = get_redis() + pg = get_pg() + poster = KoreChainPoster(r, pg) + poster.run()