Add korechain_poster.py
This commit is contained in:
204
korechain_poster.py
Normal file
204
korechain_poster.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user