vz-kore-adapter/korechain_poster.py

#!/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()