Add adapter_api.py

adapter_api.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""
+Vector Zulu — KoreID Adapter API
+Port 8766 on Charlie.
+Provides status, queue monitoring, and manual event injection.
+"""
+
+import os, json, time, logging
+from datetime import datetime, timezone
+from flask import Flask, jsonify, request, abort
+import redis
+import psycopg2
+from koreid import KoreIDGenerator, wrap_ethereum_event
+
+logging.basicConfig(level=logging.INFO, format='%(asctime)s [ADAPTER-API] %(message)s')
+app = Flask(__name__)
+
+QUEUE_KEY   = 'korechain:outbound_queue'
+DEAD_LETTER = 'korechain:dead_letter'
+
+
+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
+    )
+
+
+@app.route('/health')
+def health():
+    try:
+        r = get_redis()
+        r.ping()
+        queue_len   = r.llen(QUEUE_KEY)
+        retry_len   = r.zcard('korechain:retry_queue')
+        dead_len    = r.llen(DEAD_LETTER)
+        last_block  = r.get('eth_listener:last_block')
+        return jsonify({
+            'status':       'ok',
+            'queue_depth':  queue_len,
+            'retry_queue':  retry_len,
+            'dead_letter':  dead_len,
+            'last_eth_block': last_block,
+            'timestamp':    datetime.now(timezone.utc).isoformat(),
+        })
+    except Exception as e:
+        return jsonify({'status': 'error', 'error': str(e)}), 500
+
+
+@app.route('/queue')
+def queue_status():
+    try:
+        r = get_redis()
+        queue_items = r.lrange(QUEUE_KEY, 0, 9)  # last 10
+        return jsonify({
+            'queue_depth': r.llen(QUEUE_KEY),
+            'recent': [json.loads(i) for i in queue_items],
+        })
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+
+@app.route('/events')
+def recent_events():
+    try:
+        pg = get_pg()
+        with pg.cursor() as cur:
+            cur.execute("""
+                SELECT koreid, guid, event_type, token_symbol,
+                       tx_hash, amount, status, posted_at
+                FROM kore_outbound_events
+                ORDER BY posted_at DESC
+                LIMIT 50
+            """)
+            rows = cur.fetchall()
+        pg.close()
+        return jsonify({'events': [{
+            'koreid':       r[0],
+            'guid':         r[1],
+            'event_type':   r[2],
+            'token':        r[3],
+            'tx_hash':      r[4],
+            'amount':       float(r[5]) if r[5] else 0,
+            'status':       r[6],
+            'posted_at':    r[7].isoformat() if r[7] else None,
+        } for r in rows]})
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+
+@app.route('/generate', methods=['POST'])
+def generate_koreid():
+    """Manually generate a KoreID for testing."""
+    data = request.get_json()
+    if not data:
+        abort(400)
+    try:
+        gen = KoreIDGenerator()
+        result = gen.generate(
+            event_type=data.get('event_type', 'EXTERNAL'),
+            currency=data.get('currency', 'USD'),
+            current_network=data.get('network', 'ETHEREUM'),
+        )
+        return jsonify(result)
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+
+@app.route('/inject', methods=['POST'])
+def inject_event():
+    """
+    Manually inject an Ethereum event into the queue.
+    Useful for testing or replaying missed events.
+    """
+    data = request.get_json()
+    if not data:
+        abort(400)
+    try:
+        r   = get_redis()
+        gen = KoreIDGenerator()
+        payload = wrap_ethereum_event(
+            event_type=data['event_type'],
+            amount=float(data['amount']),
+            currency=data['currency'],
+            tx_hash=data['tx_hash'],
+            block_number=int(data.get('block_number', 0)),
+            from_address=data.get('from_address', '0x0'),
+            to_address=data.get('to_address', '0x0'),
+            token_symbol=data['token_symbol'],
+            koreid_generator=gen,
+            linked_koreid=data.get('linked_koreid'),
+        )
+        r.lpush(QUEUE_KEY, json.dumps(payload))
+        return jsonify({'queued': True, 'koreid': payload['koreid']})
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+
+@app.route('/dead-letter')
+def dead_letter():
+    try:
+        r = get_redis()
+        items = r.lrange(DEAD_LETTER, 0, 19)
+        return jsonify({
+            'count': r.llen(DEAD_LETTER),
+            'items': [json.loads(i) for i in items],
+        })
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+
+@app.route('/dead-letter/retry', methods=['POST'])
+def retry_dead_letter():
+    """Move all dead letter items back to main queue for retry."""
+    try:
+        r = get_redis()
+        count = 0
+        while True:
+            item = r.rpop(DEAD_LETTER)
+            if not item:
+                break
+            payload = json.loads(item)
+            payload['_retries'] = 0  # reset retry count
+            r.lpush(QUEUE_KEY, json.dumps(payload))
+            count += 1
+        return jsonify({'requeued': count})
+    except Exception as e:
+        return jsonify({'error': str(e)}), 500
+
+
+if __name__ == '__main__':
+    app.run(host='0.0.0.0', port=8766, debug=False)