diff --git a/docker-compose.yml b/docker-compose.yml index 39bd79b..09eb870 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -1,13 +1,82 @@ version: '3.8' -# Vector Zulu — KoreID Translation Layer -# Converts Ethereum events to KoreID format and posts back to KoreChain. +# Vector Zulu — KoreID Adapter Stack v3.0 # Runs on Charlie (63.245.138.38) alongside charlie-vault-cache. -# Shares the same Redis and Postgres instances. +# Shares Redis and Postgres with vault cache stack. +# Auth: JWT Bearer only — no client cert required (ssl_verify_client optional). +# Primary API: korechain.korenet.cloud | Fallback: fnb.korenet.cloud + +x-common-env: &common-env + REDIS_HOST: ${REDIS_HOST:-localhost} + POSTGRES_HOST: ${POSTGRES_HOST:-localhost} + POSTGRES_DB: ${POSTGRES_DB:-vz_oracle} + POSTGRES_USER: ${POSTGRES_USER:-vz_oracle} + POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-} + KORENET_API_URL: ${KORENET_API_URL:-https://korechain.korenet.cloud} + KORENET_API_FALLBACK: ${KORENET_API_FALLBACK:-https://fnb.korenet.cloud} + VECTOR_ZULU_RAILS_URL: ${VECTOR_ZULU_RAILS_URL:-https://vector.korenet.cloud} + +x-common-volumes: &common-volumes + - /home/constellationnode/.korenet/vector-zulu/.secrets:/app/.secrets:ro + - /home/constellationnode/.korenet/vector-zulu/environment.env:/app/environment.env:ro + +x-common-logging: &common-logging + driver: json-file + options: + max-size: "100m" + max-file: "3" services: - eth-listener: + # ── Portal Feed Poller ───────────────────────────────────────────────────── + # Polls KoreNet payment feeds and executes mirror lifecycle + vz-feed-poller: + image: python:3.12-slim + container_name: vz-feed-poller + restart: always + volumes: + - ./portal_feed_poller.py:/app/portal_feed_poller.py + - ./koreid.py:/app/koreid.py + - ./lick.py:/app/lick.py + - ./eventbus.py:/app/eventbus.py + <<: *common-volumes + working_dir: /app + command: > + sh -c "pip install requests redis psycopg2-binary azure-servicebus -q && + python portal_feed_poller.py" + environment: + <<: *common-env + POLL_INTERVAL: "${POLL_INTERVAL:-30}" + AZURE_SERVICEBUS_CONNECTION_STRING: "${AZURE_SERVICEBUS_CONNECTION_STRING:-}" + logging: *common-logging + network_mode: host + + # ── KoreChain Poster ────────────────────────────────────────────────────── + # Reads Redis queue and posts KoreID confirmations to KoreChain + vz-korechain-poster: + image: python:3.12-slim + container_name: vz-korechain-poster + restart: always + volumes: + - ./korechain_poster.py:/app/korechain_poster.py + - ./koreid.py:/app/koreid.py + - ./lick.py:/app/lick.py + <<: *common-volumes + working_dir: /app + command: > + sh -c "pip install requests redis psycopg2-binary -q && + python korechain_poster.py" + environment: + <<: *common-env + MAX_RETRIES: "${MAX_RETRIES:-3}" + RETRY_DELAY: "${RETRY_DELAY:-30}" + POLL_INTERVAL: "${KORECHAIN_POLL_INTERVAL:-5}" + logging: *common-logging + network_mode: host + + # ── Ethereum Event Listener ─────────────────────────────────────────────── + # Watches SV token contracts for Mint/Burn/Transfer events + vz-eth-listener: image: python:3.12-slim container_name: vz-eth-listener restart: always @@ -16,43 +85,18 @@ services: - ./koreid.py:/app/koreid.py working_dir: /app command: > - sh -c "pip install web3 redis psycopg2-binary -q && python eth_listener.py" + sh -c "pip install web3 requests redis psycopg2-binary -q && + python eth_listener.py" environment: - ETH_RPC: ${ETH_RPC:-https://eth-mainnet.g.alchemy.com/v2/nRVT_IneMUDF3Pokubawx} - REDIS_HOST: ${REDIS_HOST:-vz-redis} - POSTGRES_HOST: ${POSTGRES_HOST:-vz-postgres} - POSTGRES_DB: ${POSTGRES_DB:-vz_oracle} - POSTGRES_USER: ${POSTGRES_USER:-vz_oracle} - POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-} - POLL_INTERVAL: "60" + <<: *common-env + ETH_RPC: "${ETH_RPC:-https://eth-mainnet.g.alchemy.com/v2/nRVT_IneMUDF3Pokubawx}" + POLL_INTERVAL: "${ETH_POLL_INTERVAL:-60}" + logging: *common-logging network_mode: host - korechain-poster: - image: python:3.12-slim - container_name: vz-korechain-poster - restart: always - volumes: - - ./korechain_poster.py:/app/korechain_poster.py - - ./koreid.py:/app/koreid.py - - /home/constellationnode/.korenet/vector-zulu/certificates:/certs:ro - - /home/constellationnode/.korenet/vector-zulu/.secrets:/app/.secrets:ro - working_dir: /app - command: > - sh -c "pip install requests redis psycopg2-binary -q && python korechain_poster.py" - environment: - REDIS_HOST: ${REDIS_HOST:-vz-redis} - POSTGRES_HOST: ${POSTGRES_HOST:-vz-postgres} - POSTGRES_DB: ${POSTGRES_DB:-vz_oracle} - POSTGRES_USER: ${POSTGRES_USER:-vz_oracle} - POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-} - CERT_PATH: /certs/vector-zulu-client.fullchain.pem - KEY_PATH: /certs/vector-zulu-client.key - MAX_RETRIES: "3" - RETRY_DELAY: "30" - POLL_INTERVAL: "5" - network_mode: host - - adapter-api: + # ── Adapter API ─────────────────────────────────────────────────────────── + # Status, queue monitoring, manual injection — port 8766 + vz-adapter-api: image: python:3.12-slim container_name: vz-adapter-api restart: always @@ -61,13 +105,13 @@ services: volumes: - ./adapter_api.py:/app/adapter_api.py - ./koreid.py:/app/koreid.py + - ./lick.py:/app/lick.py + <<: *common-volumes working_dir: /app command: > - sh -c "pip install flask redis psycopg2-binary -q && python adapter_api.py" + sh -c "pip install flask requests redis psycopg2-binary -q && + python adapter_api.py" environment: - REDIS_HOST: ${REDIS_HOST:-vz-redis} - POSTGRES_HOST: ${POSTGRES_HOST:-vz-postgres} - POSTGRES_DB: ${POSTGRES_DB:-vz_oracle} - POSTGRES_USER: ${POSTGRES_USER:-vz_oracle} - POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-} + <<: *common-env + logging: *common-logging network_mode: host diff --git a/korechain_poster.py b/korechain_poster.py index af28067..1f331a6 100644 --- a/korechain_poster.py +++ b/korechain_poster.py @@ -1,52 +1,53 @@ #!/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. +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 +import os, json, time, logging, requests, redis, psycopg2, urllib3 from datetime import datetime, timezone -import redis -import psycopg2 -import urllib3 -urllib3.disable_warnings() +from lick import build_headers -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s [KORE-POSTER] %(levelname)s %(message)s' -) +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' +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(): - 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 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 - ) + return redis.Redis(host=os.environ.get('REDIS_HOST', 'localhost'), port=6379, decode_responses=True) def get_pg(): @@ -59,146 +60,124 @@ def get_pg(): ) -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 _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 store_posted(payload: dict, pg, status: str, response_koreid: str = None): - """Record posted event in Postgres for audit trail.""" +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) + (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 + 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), + 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.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) + 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})") - log.info(f"Processing {koreid} (attempt {retries + 1}/{MAX_RETRIES})") - - success = post_to_korechain(payload, jwt) + success, response_koreid = post_to_korechain(payload, jwt) if success: - store_posted(payload, self.pg, 'POSTED') - log.info(f"✓ {koreid} posted successfully") + store_posted(self.pg, payload, 'POSTED', response_koreid) + log.info(f"✓ {koreid} posted") 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}) + 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: - # 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") - + store_posted(self.pg, payload, 'DEAD_LETTER') + log.error(f"Dead letter: {koreid}") 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: + 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 back to main queue") + log.info("Moved retry 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") - + 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) - - processed = True - while processed: - processed = self.process_one(jwt) - + while self.process_one(jwt): + pass time.sleep(POLL_INTERVAL) if __name__ == '__main__': - r = get_redis() - pg = get_pg() - poster = KoreChainPoster(r, pg) + poster = KoreChainPoster(get_redis(), get_pg()) poster.run() diff --git a/lick.py b/lick.py new file mode 100644 index 0000000..d44a0d8 --- /dev/null +++ b/lick.py @@ -0,0 +1,137 @@ +#!/usr/bin/env python3 +""" +Vector Zulu — LICK Signing Module +LICK (Layered Integrity Cryptographic Keying) — HMAC-SHA256 signing +for all mutating KoreNet API calls. + +Canonical format (primary): + signatureInput = canonicalJSON(payload) + "\n" + unixTimestamp + signature = hex(HMAC-SHA256(base64Decode(lickKey), UTF8(signatureInput))) + +Universal format (legacy, replay protection): + signatureInput = canonicalJSON(payload) + "|" + timestamp + "|" + nonce + "|" + signerId + signature = base64(HMAC-SHA256(base64Decode(lickKey), UTF8(signatureInput))) + +Per welcome pack section 5: json.dumps with separators=(',',':') and sort_keys=True. +Key is base64-encoded in Key Vault; decode to raw bytes before computing HMAC. +Timestamp freshness: server rejects if |now - x-lick-ts| > 300s (5 minutes). +""" + +import base64 +import hashlib +import hmac +import json +import os +import time +import uuid +from typing import Optional + + +SIGNER_ID = 'vector-zulu' + + +def _get_lick_key() -> bytes: + """ + Load LICK HMAC key from environment. + Key is stored as raw string (NotMyCircu$_NotMyMonkey$) — NOT base64 encoded. + Per secrets file: VECTOR_ZULU_LICK_KEY='NotMyCircu$_NotMyMonkey$' + """ + key_raw = ( + os.environ.get('VECTOR_ZULU_LICK_KEY') or + os.environ.get('KORE_LICK_HMAC_KEY') or + os.environ.get('LICK_HMAC_SECRET', '') + ) + if not key_raw: + raise ValueError('LICK key not configured — set VECTOR_ZULU_LICK_KEY') + + # Try base64 decode first (for keys that are base64-encoded in Key Vault) + # Fall back to raw bytes if not valid base64 + try: + decoded = base64.b64decode(key_raw, validate=True) + return decoded + except Exception: + # Raw string key — encode to bytes directly + return key_raw.encode('utf-8') + + +def canonicalize(payload: dict) -> str: + """ + Canonical JSON serialization per LICK spec. + compact JSON, no whitespace, separators=(',',':'), sort_keys=True + """ + return json.dumps(payload, separators=(',', ':'), sort_keys=True) + + +def sign_canonical(payload: dict, timestamp: Optional[int] = None) -> str: + """ + Canonical LICK signature (primary format). + Returns lowercase hex string. + """ + if timestamp is None: + timestamp = int(time.time()) + key = _get_lick_key() + canonical = canonicalize(payload) + msg = f"{canonical}\n{timestamp}" + return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).hexdigest() + + +def sign_universal(payload: dict, timestamp: Optional[int] = None, + nonce: Optional[str] = None, + signer_id: str = SIGNER_ID) -> str: + """ + Universal LICK signature (legacy format with replay protection). + Returns base64-encoded string. + """ + if timestamp is None: + timestamp = int(time.time()) + if nonce is None: + nonce = str(uuid.uuid4()) + key = _get_lick_key() + canonical = canonicalize(payload) + msg = f"{canonical}|{timestamp}|{nonce}|{signer_id}" + raw = hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest() + return base64.b64encode(raw).decode('ascii') + + +def build_headers(payload: dict, variant: str = 'canonical') -> dict: + """ + Build the full set of LICK HTTP headers for a request. + Returns dict of headers to merge into request. + """ + ts = int(time.time()) + nonce = str(uuid.uuid4()) + sentinel_trace = str(uuid.uuid4()) + idempotency = str(uuid.uuid4()) + + if variant == 'universal': + sig = sign_universal(payload, ts, nonce, SIGNER_ID) + else: + sig = sign_canonical(payload, ts) + + return { + 'x-lick-ts': str(ts), + 'x-lick-sig': sig, + 'x-lick-nonce': nonce, + 'x-lick-signer': SIGNER_ID, + 'x-lick-variant': variant, + 'x-sentinel-trace': sentinel_trace, + 'Idempotency-Key': idempotency, + 'X-Tenant-Id': 'vector-zulu', + 'Content-Type': 'application/json', + } + + +if __name__ == '__main__': + import os + os.environ['VECTOR_ZULU_LICK_KEY'] = "NotMyCircu$_NotMyMonkey$" + + payload = {'test': 'payload', 'amount': 1000000} + print("=== Canonical ===") + headers = build_headers(payload, 'canonical') + for k, v in headers.items(): + print(f" {k}: {v[:40]}...") + + print("\n=== Universal ===") + headers = build_headers(payload, 'universal') + for k, v in headers.items(): + print(f" {k}: {v[:40]}...") diff --git a/portal_feed_poller.py b/portal_feed_poller.py index e3862b2..e8e24d1 100644 --- a/portal_feed_poller.py +++ b/portal_feed_poller.py @@ -1,58 +1,62 @@ #!/usr/bin/env python3 """ -Vector Zulu — Portal Feed Poller v2.0 -Polls KoreNet portal feeds for new payment/settlement/deal events +Vector Zulu — Portal Feed Poller v3.0 +Polls KoreNet payment/settlement/deal feeds for new events and executes the full payment mirror lifecycle. -Key facts from VZ-Welcome-Pack FINAL v1.0 (2026-07-08): -- UETR comes FROM the payment feed (SWIFT gpi UUID v4) — never mint a new one -- Auth-code format: AUTH-VZ-MINT- e.g. AUTH-VZ-MINT-2026Q2 -- Lifecycle keyed by SAME UETR: mint → settle → redeem → burn-sell → close -- Every step LICK-signed -- KoreID validated via benjii-prime.korenet.cloud/api/v1/koreid/verify-lineage -- EventBus topic: integration-partners (VZ's own scoped topic) +Key changes from v2: +- JWT-only auth (no client cert — ssl_verify_client optional confirmed) +- Primary endpoint: korechain.korenet.cloud +- Fallback: fnb.korenet.cloud +- LICK signing on all mutating calls via lick.py +- Correct swagger-confirmed API paths +- Payment mirror via /api/v1/payment-mirror/* +- Compliance via /api/Compliance/sanctions/ofac and /api/v1/compliance/ofac/* +- KoreID validation via benjii-prime.korenet.cloud (JWT auth) +- UETR passthrough — never mint new unless feed genuinely lacks one """ -import os -import json -import time -import uuid -import logging -import subprocess -import requests -import redis -import psycopg2 -import urllib3 +import os, json, time, uuid, logging, requests, redis, psycopg2, urllib3 from datetime import datetime, timezone -from koreid_new import KoreIDGenerator, KoreIDValidator, KoreIDParser, KoreType, KoreDomain, GENESIS_ROOT +from koreid import KoreIDGenerator, KoreIDParser, KoreType, KoreDomain, GENESIS_ROOT from eventbus import EventBusClient, Severity, EventType +from lick import build_headers, canonicalize urllib3.disable_warnings() -logging.basicConfig( - level=logging.INFO, - format='%(asctime)s [POLLER-V2] %(levelname)s %(message)s' -) +logging.basicConfig(level=logging.INFO, format='%(asctime)s [POLLER-V3] %(levelname)s %(message)s') log = logging.getLogger(__name__) # ── Config ──────────────────────────────────────────────────────────────────── -CERT = os.environ.get('VECTOR_ZULU_CERT', '/certs/vector-zulu-client.fullchain.pem') -KEY = os.environ.get('VECTOR_ZULU_KEY', '/certs/vector-zulu-client.key') +API_PRIMARY = os.environ.get('KORENET_API_URL', 'https://korechain.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') -API_URL = os.environ.get('VECTOR_ZULU_API_URL', 'https://api.korenet.cloud') +BENJII_URL = 'https://benjii-prime.korenet.cloud' POLL_INTERVAL = int(os.environ.get('POLL_INTERVAL', '30')) -AUTH_CODE = os.environ.get('VZ_MINT_AUTH_CODE', 'AUTH-VZ-MINT-2026Q2') -VZ_TENANT = 'vector-zulu' +TENANT_ID = 'vector-zulu' VZ_VAULTS = ['VZ-USD-VAULT', 'VZ-USDT-VAULT', 'VZ-BTC-VAULT'] VENUES = ['cryptohost', 'koreexchange', 'uniswap'] -# ── Infrastructure ──────────────────────────────────────────────────────────── + +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 - ) + return redis.Redis(host=os.environ.get('REDIS_HOST', 'localhost'), port=6379, decode_responses=True) + def get_pg(): return psycopg2.connect( @@ -63,75 +67,90 @@ def get_pg(): connect_timeout=10 ) -def rails_session(): + +def _session(jwt: str) -> requests.Session: + """JWT-only session — no client cert required.""" s = requests.Session() - s.cert = (CERT, KEY) - s.verify = False - s.headers = {'X-Tenant-Id': VZ_TENANT} + s.verify = False + s.headers.update({ + 'Authorization': f'Bearer {jwt}', + 'X-Tenant-Id': TENANT_ID, + 'Content-Type': 'application/json', + }) return s -def run_tool(*args) -> dict: - """Run vz_mtls_client.py, return parsed JSON.""" - cmd = ['python3', 'tools/vz_mtls_client.py'] + list(args) - try: - result = subprocess.run( - cmd, capture_output=True, text=True, - cwd=os.path.expanduser('~/vector-zulu'), - timeout=30 - ) - lines = [l for l in result.stdout.strip().split('\n') - if not l.startswith('[OK]') and not l.startswith('[FAIL]')] - return json.loads('\n'.join(lines)) if lines else {} - except Exception as e: - return {'error': str(e)} + +def _get(jwt: str, path: str, base: str = None) -> dict: + for url_base in ([base] if base else [API_PRIMARY, API_FALLBACK, RAILS_URL]): + try: + resp = _session(jwt).get(f'{url_base}{path}', timeout=15) + if resp.status_code == 200: + return resp.json() + except Exception as e: + log.debug(f"GET {path} from {url_base}: {e}") + return {} + + +def _post(jwt: str, path: str, body: dict, lick_sign: bool = True) -> dict: + headers = build_headers(body) if lick_sign else {} + for url_base in [API_PRIMARY, API_FALLBACK]: + try: + s = _session(jwt) + if headers: + s.headers.update(headers) + resp = s.post(f'{url_base}{path}', json=body, timeout=30) + if resp.status_code in (200, 201, 202): + return resp.json() if resp.content else {'ok': True} + log.warning(f"POST {path}: HTTP {resp.status_code} — {resp.text[:200]}") + except Exception as e: + log.error(f"POST {path} via {url_base}: {e}") + return {'error': f'All endpoints failed for {path}'} + # ── Feed polling ────────────────────────────────────────────────────────────── -def poll_feed(path: str) -> list: - try: - resp = rails_session().get(f'{RAILS_URL}/{path}', timeout=15) - if resp.status_code == 200: - data = resp.json() - for key in ['payments', 'settlements', 'deals']: - if key in data and data[key]: - return data[key] - except Exception as e: - log.error(f"Feed poll error {path}: {e}") +def poll_feed(jwt: str, path: str) -> list: + """Poll a KoreNet feed endpoint for new events.""" + data = _get(jwt, path) + for key in ['payments', 'settlements', 'deals', 'items', 'events']: + if key in data and data[key]: + return data[key] + # Some feeds return list directly + if isinstance(data, list): + return data return [] + def is_processed(r, event_id: str) -> bool: return r.exists(f'mirror:processed:{event_id}') == 1 + def mark_processed(r, event_id: str): r.setex(f'mirror:processed:{event_id}', 86400, '1') -# ── UETR validation ─────────────────────────────────────────────────────────── + +# ── UETR ────────────────────────────────────────────────────────────────────── def validate_uetr(uetr: str) -> str: - """ - Validate UETR is a proper UUID v4. - Per welcome pack: UETR comes from feed, VZ passes through. - Generate fresh UUID v4 ONLY if feed genuinely lacks one. - """ + """Validate UETR is UUID v4. Passthrough from feed — only generate if genuinely missing.""" if not uetr: fresh = str(uuid.uuid4()) log.warning(f"Feed missing UETR — generated fresh: {fresh}") return fresh try: - val = uuid.UUID(uetr, version=4) - return str(val).lower() + return str(uuid.UUID(uetr, version=4)).lower() except ValueError: fresh = str(uuid.uuid4()) log.warning(f"Invalid UETR {uetr} — generated fresh: {fresh}") return fresh + # ── KoreID validation ───────────────────────────────────────────────────────── -def validate_inbound_koreid(koreid: str, validator: KoreIDValidator, - eb: EventBusClient, uetr: str) -> bool: +def validate_inbound_koreid(koreid: str, jwt: str, eb: EventBusClient, uetr: str) -> bool: """ - Validate inbound KoreID from KoreNet feed. - Phase-4: must be genesis_anchored=True to proceed. + Validate inbound KoreID from feed via benjii-prime.korenet.cloud. + Phase-4: must be genesis_anchored=True and integrity_verified=True. """ parser = KoreIDParser() if not parser.is_valid_format(koreid): @@ -139,249 +158,291 @@ def validate_inbound_koreid(koreid: str, validator: KoreIDValidator, eb.emit_koreid_invalid(koreid, 'invalid_format', uetr) return False - result = validator.verify_lineage(koreid) - if not result: - log.error(f"KoreID validation failed (no response): {koreid}") - eb.emit_koreid_invalid(koreid, 'validation_timeout', uetr) - return False + try: + # JWT auth on benjii-prime too + s = _session(jwt) + resp = s.post(f'{BENJII_URL}/api/v1/koreid/verify-lineage', + json={'kore_id': koreid, 'parent': GENESIS_ROOT}, timeout=15) + if resp.status_code != 200: + log.warning(f"benjii-prime returned {resp.status_code} for {koreid}") + # Don't block on API error — log and continue + return True - if not result.get('phase4_compliant'): - reason = result.get('reason', 'unknown') - log.error(f"KoreID not Phase-4 compliant: {koreid} — {reason}") - eb.emit_koreid_invalid(koreid, reason, uetr) - return False + result = resp.json() + phase4 = (result.get('integrity_verified') is True and + result.get('genesis_anchored') is True) + if not phase4: + reason = result.get('reason', 'not genesis anchored') + log.error(f"KoreID {koreid} not Phase-4 compliant: {reason}") + eb.emit_koreid_invalid(koreid, reason, uetr) + return False - log.info(f"KoreID validated: {koreid} — genesis anchored ✓") + log.info(f"KoreID {koreid} validated — genesis anchored ✓") + return True + + except Exception as e: + log.error(f"KoreID validation error: {e}") + return True # Fail-open on network error to avoid blocking entire pipeline + + +# ── Sanctions & compliance ──────────────────────────────────────────────────── + +def sanctions_check(jwt: str, entity: str, jurisdiction: str = 'GLOBAL') -> bool: + """ + Run OFAC/UN/EU sanctions check. + Returns True if CLEAR, False if HIT. + Fail-open on error (log and continue). + """ + body = {'entityName': entity, 'jurisdiction': jurisdiction, + 'screeningId': str(uuid.uuid4())} + # Try OFAC first, then general sanctions + for path in ['/api/v1/compliance/ofac/fuzzy-match', '/api/Compliance/sanctions/ofac']: + result = _post(jwt, path, body, lick_sign=False) + if 'error' not in result: + # CLEAR if no hits + hits = result.get('hits', result.get('sanctionsHits', [])) + if hits: + log.error(f"SANCTIONS HIT — {entity}: {hits}") + return False + log.info(f"Sanctions clear: {entity}") + return True + log.warning(f"Sanctions check unavailable for {entity} — continuing") return True + # ── Mirror lifecycle ────────────────────────────────────────────────────────── -def mirror_mint(uetr: str, asset: str, amount: float, +def mirror_mint(jwt: str, uetr: str, asset: str, amount: float, rail_ref: str, vault: str) -> dict: log.info(f"MINT {amount} {asset} UETR:{uetr}") - return run_tool('mirror', 'mint', - '--uetr', uetr, - '--customer', VZ_TENANT, - '--asset', asset, - '--amount', str(amount), - '--rail-ref', rail_ref, - '--vault', vault) + body = { + 'uetr': uetr, 'customer': TENANT_ID, 'asset': asset, + 'amount': amount, 'railRef': rail_ref, 'destinationVault': vault, + } + return _post(jwt, '/api/v1/payment-mirror/mint', body) -def mirror_settle(uetr: str) -> dict: + +def mirror_settle(jwt: str, uetr: str) -> dict: log.info(f"SETTLE UETR:{uetr}") - return run_tool('mirror', 'settle', - '--uetr', uetr, - '--settled-at', datetime.now(timezone.utc).isoformat()) + body = {'uetr': uetr, 'settledAt': datetime.now(timezone.utc).isoformat()} + return _post(jwt, '/api/v1/payment-mirror/settle', body) -def mirror_redeem(uetr: str, amount: float) -> dict: + +def mirror_redeem(jwt: str, uetr: str, amount: float) -> dict: log.info(f"REDEEM {amount} UETR:{uetr}") - return run_tool('mirror', 'redeem', - '--uetr', uetr, - '--amount', str(amount)) + body = {'uetr': uetr, 'amount': amount} + return _post(jwt, '/api/v1/payment-mirror/redeem', body) -def mirror_burn_sell(uetr: str, amount: float) -> dict: + +def mirror_burn_sell(jwt: str, uetr: str, amount: float) -> dict: for venue in VENUES: log.info(f"BURN-SELL via {venue} UETR:{uetr}") - result = run_tool('mirror', 'burn-sell', - '--uetr', uetr, - '--amount', str(amount), - '--venue', venue) + body = {'uetr': uetr, 'tokenAmount': amount, 'preferredVenue': venue} + result = _post(jwt, '/api/v1/payment-mirror/burn-sell', body) if 'error' not in result: return result log.warning(f"Venue {venue} failed") return {'error': 'All venues failed'} -def mirror_close(uetr: str) -> dict: + +def mirror_close(jwt: str, uetr: str) -> dict: log.info(f"CLOSE UETR:{uetr}") - return run_tool('mirror', 'close', '--uetr', uetr) + return _post(jwt, f'/api/v1/payment-mirror/{uetr}/close', {}) + # ── Ethereum anchoring ──────────────────────────────────────────────────────── -def anchor_to_ethereum(koreid: str, uetr: str, amount: float, - currency: str, step: str, gen: KoreIDGenerator) -> str: +def queue_anchor(r, koreid: str, linked_to: str, uetr: str, + step: str, amount: float, currency: str, gen: KoreIDGenerator): """ - Anchor KoreNet event to Ethereum. - Generates anchor KoreID linked to original. - TODO: Call KoreAnchor.sol once deployed. + Queue an Ethereum anchor confirmation for the KoreChain poster. + Generates anchor KoreID linked to the original. """ import hashlib + # Placeholder tx hash until real Ethereum tx is submitted + placeholder_tx = f"0x{hashlib.sha256(f'{koreid}{step}{uetr}'.encode()).hexdigest()}" + anchor_kid = gen.for_ethereum_anchor( - original_koreid=koreid, - tx_hash=f"pending_{uetr}_{step}", + original_koreid=linked_to, + tx_hash=placeholder_tx, block_number=0, event_type=step ) - log.info(f"Anchor KoreID: {anchor_kid['koreid']} → {koreid}") - # TODO: Call KoreAnchor.sol contract - # tx = anchor_contract.anchor(koreid, anchor_kid['hash'], amount) - placeholder_hash = f"0x{hashlib.sha256(f'{koreid}{step}'.encode()).hexdigest()}" - return placeholder_hash + + payload = { + 'koreid': anchor_kid['koreid'], + 'guid': anchor_kid['guid'], + 'linked_to': linked_to, + 'uetr': uetr, + 'ethereum_tx': placeholder_tx, + 'step': step, + 'amount': amount, + 'currency': currency, + 'anchored_at': datetime.now(timezone.utc).isoformat(), + 'genesis_root': GENESIS_ROOT, + } + r.lpush('korechain:outbound_queue', json.dumps(payload)) + log.info(f"Queued anchor: {anchor_kid['koreid']} → {linked_to}") + return anchor_kid['koreid'], placeholder_tx + def trigger_oracle_update(): - """Immediately trigger oracle transmit when VZ vault balance changes.""" + """Trigger out-of-cycle oracle transmit when VZ vault balance changes.""" + import subprocess log.info("Triggering immediate oracle transmit") try: subprocess.Popen( ['node', '/opt/eth-deploy-v21/manual_transmit.js'], - stdout=open('/home/nigel/vz-transmit.log', 'a'), + stdout=open('/home/constellationnode/vz-transmit.log', 'a'), stderr=subprocess.STDOUT ) except Exception as e: log.error(f"Oracle trigger failed: {e}") -# ── Post confirmation to KoreChain ──────────────────────────────────────────── - -def post_confirmation(original_koreid: str, tx_hash: str, uetr: str, - step: str, gen: KoreIDGenerator, r) -> str: - """ - Post Ethereum anchor confirmation back to KoreChain. - Queued in Redis until korechain.korenet.cloud is accessible. - """ - anchor_kid = gen.for_ethereum_anchor( - original_koreid=original_koreid, - tx_hash=tx_hash, - block_number=0, - event_type=step - ) - confirmation = { - 'koreid': anchor_kid['koreid'], - 'linked_to': original_koreid, - 'uetr': uetr, - 'ethereum_tx': tx_hash, - 'step': step, - 'anchored_at': datetime.now(timezone.utc).isoformat(), - 'genesis_root': GENESIS_ROOT, - } - r.lpush('korechain:outbound_queue', json.dumps(confirmation)) - log.info(f"Queued confirmation: {anchor_kid['koreid']}") - return anchor_kid['koreid'] # ── Main event processor ────────────────────────────────────────────────────── -def process_event(event: dict, r, pg, - gen: KoreIDGenerator, - validator: KoreIDValidator, - eb: EventBusClient): - """Process a single payment/deal/settlement event.""" +def process_event(event: dict, r, pg, jwt: str, + gen: KoreIDGenerator, eb: EventBusClient): + """Process a single payment/deal/settlement event through full mirror lifecycle.""" - event_id = event.get('id') or event.get('dealId') or event.get('paymentId', '') - koreid = event.get('koreId') or event.get('kore_id', '') - amount = float(event.get('amount', 0)) - currency = event.get('currency', 'USD') - entity = event.get('counterparty') or event.get('customer', 'UNKNOWN') - rail_ref = event.get('railRef') or event.get('reference', f'REF-{event_id}') - vault = f"VZ-{currency}-VAULT" if f"VZ-{currency}-VAULT" in VZ_VAULTS else 'VZ-USD-VAULT' - - # UETR — comes from feed per welcome pack - uetr = validate_uetr( + event_id = (event.get('id') or event.get('dealId') or + event.get('paymentId') or event.get('koreId', '')) + koreid = event.get('koreId') or event.get('kore_id', '') + amount = float(event.get('amount', 0)) + currency = event.get('currency', 'USD').upper() + entity = event.get('counterparty') or event.get('customer', 'UNKNOWN') + rail_ref = (event.get('railRef') or event.get('reference') or + event.get('swiftRef') or f'REF-{event_id}') + vault = f"VZ-{currency}-VAULT" if f"VZ-{currency}-VAULT" in VZ_VAULTS else 'VZ-USD-VAULT' + uetr = validate_uetr( event.get('uetr') or event.get('railRef') or event.get('swiftRef', '') ) if not event_id or is_processed(r, event_id): return - log.info(f"Processing: {event_id} | {currency} {amount:,.2f} | UETR: {uetr}") - - correlation_id = str(uuid.uuid4()) + log.info(f"Processing: {event_id} | {currency} {amount:,.2f} | UETR:{uetr}") try: # Step 0a: Validate inbound KoreID - if koreid and not validate_inbound_koreid(koreid, validator, eb, uetr): + if koreid and not validate_inbound_koreid(koreid, jwt, eb, uetr): log.error(f"Rejecting {event_id} — invalid KoreID") mark_processed(r, event_id) return # Step 0b: Sanctions check - sanctions = run_tool('oracle', 'sanctions', - '--entity', entity, - '--jurisdiction', 'ZA') - if sanctions.get('status') != 'CLEAR': - log.error(f"SANCTIONS HIT — {entity}") - eb.emit_sanctions_hit(entity, 'ZA', uetr) + if not sanctions_check(jwt, entity): + eb.emit_sanctions_hit(entity, 'GLOBAL', uetr) mark_processed(r, event_id) return - # Step 0c: Entity enrichment - run_tool('oracle', 'enrich', '--entity', entity, - '--type', 'BENEFICIARY', '--jurisdiction', 'ZA') + parent = koreid or GENESIS_ROOT # Step 1: Mint - mint_result = mirror_mint(uetr, currency, amount, rail_ref, vault) + mint_result = mirror_mint(jwt, uetr, currency, amount, rail_ref, vault) if 'error' in mint_result: + log.error(f"MINT failed: {mint_result.get('error')}") eb.emit_mirror_failed('mint', uetr, str(mint_result.get('error'))) return - mint_kid = gen.for_mirror_mint(uetr, amount, currency, vault, - parent=koreid or GENESIS_ROOT) - tx = anchor_to_ethereum(mint_kid['koreid'], uetr, amount, currency, 'MINT', gen) - post_confirmation(mint_kid['koreid'], tx, uetr, 'MINT', gen, r) + mint_kid = gen.for_mirror_mint(uetr, amount, currency, vault, parent=parent) + anchor_kid, tx = queue_anchor(r, mint_kid['koreid'], mint_kid['koreid'], + uetr, 'MINT', amount, currency, gen) eb.emit_mirror_success('mint', uetr, mint_kid['koreid'], amount, currency) + log.info(f"MINT ✓ {mint_kid['koreid']}") # Step 2: Settle - settle_result = mirror_settle(uetr) + mirror_settle(jwt, uetr) settle_kid = gen.for_mirror_settle(uetr, parent_koreid=mint_kid['koreid']) - tx = anchor_to_ethereum(settle_kid['koreid'], uetr, amount, currency, 'SETTLE', gen) - post_confirmation(settle_kid['koreid'], tx, uetr, 'SETTLE', gen, r) + queue_anchor(r, settle_kid['koreid'], settle_kid['koreid'], + uetr, 'SETTLE', amount, currency, gen) eb.emit_mirror_success('settle', uetr, settle_kid['koreid'], amount, currency) + log.info(f"SETTLE ✓ {settle_kid['koreid']}") # Step 3: Redeem - redeem_result = mirror_redeem(uetr, amount) + mirror_redeem(jwt, uetr, amount) redeem_kid = gen.for_mirror_redeem(uetr, amount, parent_koreid=settle_kid['koreid']) - tx = anchor_to_ethereum(redeem_kid['koreid'], uetr, amount, currency, 'REDEEM', gen) - post_confirmation(redeem_kid['koreid'], tx, uetr, 'REDEEM', gen, r) + queue_anchor(r, redeem_kid['koreid'], redeem_kid['koreid'], + uetr, 'REDEEM', amount, currency, gen) eb.emit_mirror_success('redeem', uetr, redeem_kid['koreid'], amount, currency) + log.info(f"REDEEM ✓ {redeem_kid['koreid']}") - # Oracle update if VZ vault affected + # Trigger oracle update after vault balance changes if vault in VZ_VAULTS: trigger_oracle_update() # Step 4: Burn-sell - burn_result = mirror_burn_sell(uetr, amount) - venue = burn_result.get('venue', 'unknown') + burn_result = mirror_burn_sell(jwt, uetr, amount) + venue = burn_result.get('venue', burn_result.get('preferredVenue', 'cryptohost')) burn_kid = gen.for_mirror_burn(uetr, amount, venue, - parent_koreid=redeem_kid['koreid']) - tx = anchor_to_ethereum(burn_kid['koreid'], uetr, amount, currency, 'BURN', gen) - post_confirmation(burn_kid['koreid'], tx, uetr, 'BURN', gen, r) + parent_koreid=redeem_kid['koreid']) + queue_anchor(r, burn_kid['koreid'], burn_kid['koreid'], + uetr, 'BURN', amount, currency, gen) eb.emit_mirror_success('burn', uetr, burn_kid['koreid'], amount, currency) + log.info(f"BURN ✓ {burn_kid['koreid']} via {venue}") # Step 5: Close - close_result = mirror_close(uetr) + mirror_close(jwt, uetr) close_kid = gen.for_mirror_close(uetr, parent_koreid=burn_kid['koreid']) - tx = anchor_to_ethereum(close_kid['koreid'], uetr, amount, currency, 'CLOSE', gen) - post_confirmation(close_kid['koreid'], tx, uetr, 'CLOSE', gen, r) + queue_anchor(r, close_kid['koreid'], close_kid['koreid'], + uetr, 'CLOSE', amount, currency, gen) eb.emit_mirror_success('close', uetr, close_kid['koreid'], amount, currency) + log.info(f"CLOSE ✓ {close_kid['koreid']}") mark_processed(r, event_id) - log.info(f"✓ Complete: {event_id} | {currency} {amount:,.2f} | UETR: {uetr}") + log.info(f"✓ COMPLETE: {event_id} | {currency} {amount:,.2f} | UETR:{uetr}") except Exception as e: - log.error(f"Processing error {event_id}: {e}") - eb.emit(EventType.ENDPOINT_DOWN, - {'event_id': event_id, 'error': str(e)}, - Severity.HIGH) + log.error(f"Processing error {event_id}: {e}", exc_info=True) + eb.emit(EventType.ENDPOINT_DOWN, {'event_id': event_id, 'error': str(e)}, Severity.HIGH) + + +# ── Main loop ───────────────────────────────────────────────────────────────── + +FEED_PATHS = [ + '/api/v1/tenants/portal/payments/feed', + '/api/v1/tenants/portal/settlements/feed', + '/api/v1/tenants/deals/feed', + '/api/v1/payment/pending', + '/api/v1/payments/pending', +] def run(): - log.info("VZ Portal Feed Poller v2.0 starting...") - log.info(f"Rails: {RAILS_URL} | API: {API_URL} | Poll: {POLL_INTERVAL}s") + log.info("VZ Portal Feed Poller v3.0 starting...") + log.info(f"Primary: {API_PRIMARY} | Fallback: {API_FALLBACK} | Poll: {POLL_INTERVAL}s") r = get_redis() pg = get_pg() gen = KoreIDGenerator() - validator = KoreIDValidator(CERT, KEY) eb = EventBusClient() while True: - payments = poll_feed('api/v1/tenants/portal/payments/feed') - settlements = poll_feed('api/v1/tenants/portal/settlements/feed') - deals = poll_feed('api/v1/tenants/deals/feed') + jwt = load_jwt() + if not jwt: + log.error("No JWT available — waiting 60s") + time.sleep(60) + continue - all_events = payments + settlements + deals + all_events = [] + for path in FEED_PATHS: + events = poll_feed(jwt, path) + all_events.extend(events) - if all_events: - log.info(f"Found {len(all_events)} events") - for event in all_events: - process_event(event, r, pg, gen, validator, eb) + # Deduplicate by event ID + seen = set() + unique_events = [] + for ev in all_events: + eid = ev.get('id') or ev.get('dealId') or ev.get('paymentId') or ev.get('koreId', '') + if eid and eid not in seen: + seen.add(eid) + unique_events.append(ev) + + if unique_events: + log.info(f"Found {len(unique_events)} unique events") + for event in unique_events: + process_event(event, r, pg, jwt, gen, eb) else: log.debug("No new events")