214 lines
8.3 KiB
Python
214 lines
8.3 KiB
Python
#!/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()
|