Update app.py

app.py

@@ -6,12 +6,13 @@ import hashlib
 from collections import deque
 from typing import Any, Dict, Optional
 
+import requests
 from fastapi import FastAPI, Request, Header, HTTPException
 from fastapi.responses import JSONResponse
 import gradio as gr
 
 # -----------------------------------------------------------------------------
-# Configuration (set these as HF Space "Secrets" or "Variables")
+# Configuration (HF Space Variables/Secrets)
 # -----------------------------------------------------------------------------
 WEBHOOK_TOKEN = os.getenv("WEBHOOK_TOKEN", "").strip()
 WEBHOOK_HMAC_SECRET = os.getenv("WEBHOOK_HMAC_SECRET", "").strip()
@@ -22,6 +23,17 @@ MAX_MESSAGES = int(os.getenv("MAX_MESSAGES", "200"))
 PERSIST_TO_FILE = os.getenv("PERSIST_TO_FILE", "0").strip() == "1"
 PERSIST_PATH = os.getenv("PERSIST_PATH", "webhook_messages.jsonl").strip()
 
+# Relay forwarding
+WAKEUP_URL = os.getenv("WAKEUP_URL", "http://163.245.192.149:8080/wakeup").strip().rstrip("/")
+RELAY_TIMEOUT = int(os.getenv("RELAY_TIMEOUT", "30"))
+
+# Dedupe window (seconds). Default 24h.
+SEEN_TTL_SECONDS = int(os.getenv("SEEN_TTL_SECONDS", "86400"))
+MAX_SEEN = int(os.getenv("MAX_SEEN", "5000"))
+
+# Limit concurrent forwards (prevents flood / money burn)
+MAX_INFLIGHT = int(os.getenv("MAX_INFLIGHT", "1"))
+
 # -----------------------------------------------------------------------------
 # In-memory state
 # -----------------------------------------------------------------------------
@@ -29,6 +41,10 @@ MESSAGES = deque(maxlen=MAX_MESSAGES)
 TOTAL_COUNT = 0
 LAST_RECEIVED_TS: Optional[float] = None
 
+# seen fingerprints -> last_seen_ts
+SEEN: Dict[str, float] = {}
+INFLIGHT = 0
+
 
 def _now_ts() -> float:
     return time.time()
@@ -47,7 +63,7 @@ def _safe_json_dumps(obj: Any) -> str:
         return str(obj)
 
 
-def _append_message(payload: Dict[str, Any], meta: Dict[str, Any]) -> None:
+def _append_message(payload: Dict[str, Any], meta: Dict[str, Any], relay: Dict[str, Any]) -> None:
     global TOTAL_COUNT, LAST_RECEIVED_TS
     TOTAL_COUNT += 1
     LAST_RECEIVED_TS = _now_ts()
@@ -57,6 +73,7 @@ def _append_message(payload: Dict[str, Any], meta: Dict[str, Any]) -> None:
         "received_at": _ts_to_str(LAST_RECEIVED_TS),
         "meta": meta,
         "payload": payload,
+        "relay": relay,
     }
 
     MESSAGES.appendleft(entry)
@@ -86,39 +103,97 @@ def _verify_hmac_signature(raw_body: bytes, provided_sig: Optional[str]) -> None
         raise HTTPException(status_code=401, detail="Invalid signature.")
 
 
+def _norm(v: Any) -> str:
+    if v is None:
+        return ""
+    if isinstance(v, str):
+        s = v.strip()
+        if s.lower() in ("null", "none"):
+            return ""
+        return s
+    return str(v)
+
+
+def compute_fingerprint(payload: Dict[str, Any]) -> str:
+    """
+    Payload-based fingerprint (MVP).
+    When you have a file download URL, upgrade to sha256(file_bytes).
+    """
+    parts = {
+        "event_type": _norm(payload.get("event_type")),
+        "rid": _norm(payload.get("rid")),
+        "file_type": _norm(payload.get("file_type")),
+        "case_vendor_rid": _norm(payload.get("case_vendor_rid")),
+        "case_rid": _norm(payload.get("case_rid")),
+        "date_modified": _norm(payload.get("date_modified")),
+    }
+    raw = json.dumps(parts, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(raw).hexdigest()
+
+
+def _gc_seen(now: float) -> None:
+    # TTL cleanup
+    cutoff = now - SEEN_TTL_SECONDS
+    dead = [k for k, ts in SEEN.items() if ts < cutoff]
+    for k in dead:
+        SEEN.pop(k, None)
+
+    # size cap cleanup (drop oldest)
+    if len(SEEN) > MAX_SEEN:
+        items = sorted(SEEN.items(), key=lambda kv: kv[1])  # oldest first
+        for k, _ in items[: len(SEEN) - MAX_SEEN]:
+            SEEN.pop(k, None)
+
+
 # -----------------------------------------------------------------------------
-# FastAPI app (webhook listener)
+# FastAPI app
 # -----------------------------------------------------------------------------
-app = FastAPI(title="Webhook Listener", version="1.0.0")
+app = FastAPI(title="Inserio Relay", version="1.0.0")
 
 
 @app.get("/health")
 def health():
     return {
         "ok": True,
+        "service": "inserio-relay",
+        "post": ["/webhook/", "/wakeup/"],
+        "forward_to": f"{WAKEUP_URL}/wakeup",
+        "max_inflight": MAX_INFLIGHT,
+        "seen_ttl_seconds": SEEN_TTL_SECONDS,
         "total_count": TOTAL_COUNT,
         "last_received_at": _ts_to_str(LAST_RECEIVED_TS),
     }
 
 
-# IMPORTANT: define both with and without trailing slash
-@app.post("/wakeup")
+@app.get("/")
+def root():
+    # A simple JSON so HF "App" tab doesn't show Not Found
+    return {
+        "ok": True,
+        "service": "inserio-relay",
+        "post": ["/webhook/", "/wakeup/"],
+        "forward_to": f"{WAKEUP_URL}/wakeup",
+        "max_inflight": MAX_INFLIGHT,
+        "seen_ttl_seconds": SEEN_TTL_SECONDS,
+    }
+
+
 @app.post("/wakeup/")
-@app.post("/webhook")
 @app.post("/webhook/")
 async def webhook(
     request: Request,
-    x_sa_signature: Optional[str] = Header(default=None),
-    x_sa_token: Optional[str] = Header(default=None),
+    x_sa_signature: Optional[str] = Header(default=None, convert_underscores=False),
+    x_sa_token: Optional[str] = Header(default=None, convert_underscores=False),
 ):
+    global INFLIGHT
+
     raw_body = await request.body()
 
-    # Token auth (optional)
+    # Auth (optional)
     if WEBHOOK_TOKEN:
         token_value = request.headers.get(WEBHOOK_TOKEN_HEADER) or x_sa_token
         _verify_token(token_value)
 
-    # HMAC auth (optional)
     if WEBHOOK_HMAC_SECRET:
         sig_value = request.headers.get("X-SA-Signature") or x_sa_signature
         _verify_hmac_signature(raw_body, sig_value)
@@ -131,6 +206,89 @@ async def webhook(
     except Exception:
         payload = {"_raw_body": raw_body.decode("utf-8", errors="replace")}
 
+    now = _now_ts()
+    _gc_seen(now)
+
+    # Enforce default state
+    if not _norm(payload.get("sa_processing_state")):
+        payload["sa_processing_state"] = "new"
+
+    # Fingerprint
+    fp = _norm(payload.get("sa_fingerprint"))
+    if not fp:
+        fp = compute_fingerprint(payload)
+        payload["sa_fingerprint"] = fp
+
+    # Dedupe check
+    duplicate = fp in SEEN
+    if not duplicate:
+        SEEN[fp] = now
+
+    # Flood guard
+    if INFLIGHT >= MAX_INFLIGHT:
+        relay_info = {
+            "ok": True,
+            "duplicate": duplicate,
+            "fingerprint": fp,
+            "forwarded": False,
+            "reason": "rate_limited_max_inflight",
+        }
+        meta = {
+            "client": request.client.host if request.client else None,
+            "path": str(request.url.path),
+            "headers": {
+                "user-agent": request.headers.get("user-agent"),
+                "content-type": request.headers.get("content-type"),
+            },
+        }
+        _append_message(payload=payload, meta=meta, relay=relay_info)
+        return JSONResponse(status_code=200, content=relay_info)
+
+    # Forward (even if duplicate? NO. If duplicate, don’t forward.)
+    forward_status = None
+    forward_body = None
+    forward_url = f"{WAKEUP_URL}/wakeup"
+
+    if duplicate:
+        relay_info = {
+            "ok": True,
+            "duplicate": True,
+            "fingerprint": fp,
+            "forwarded": False,
+            "reason": "duplicate",
+        }
+        meta = {
+            "client": request.client.host if request.client else None,
+            "path": str(request.url.path),
+            "headers": {
+                "user-agent": request.headers.get("user-agent"),
+                "content-type": request.headers.get("content-type"),
+            },
+        }
+        _append_message(payload=payload, meta=meta, relay=relay_info)
+        return JSONResponse(status_code=200, content=relay_info)
+
+    # Forward non-duplicate
+    INFLIGHT += 1
+    try:
+        r = requests.post(forward_url, json=payload, timeout=RELAY_TIMEOUT)
+        forward_status = r.status_code
+        forward_body = r.text[:2000]
+    except requests.RequestException as e:
+        forward_status = 502
+        forward_body = f"relay_failed: {e}"
+    finally:
+        INFLIGHT -= 1
+
+    relay_info = {
+        "ok": True,
+        "duplicate": False,
+        "fingerprint": fp,
+        "forward_url": forward_url,
+        "forward_status": forward_status,
+        "forward_body": forward_body,
+    }
+
     meta = {
         "client": request.client.host if request.client else None,
         "path": str(request.url.path),
@@ -142,22 +300,13 @@ async def webhook(
         },
     }
 
-    _append_message(payload=payload, meta=meta)
+    _append_message(payload=payload, meta=meta, relay=relay_info)
 
-    return JSONResponse(
-        status_code=200,
-        content={
-            "ok": True,
-            "message": "Webhook received.",
-            "total_count": TOTAL_COUNT,
-            "last_received_at": _ts_to_str(LAST_RECEIVED_TS),
-            "echo": payload,
-        },
-    )
+    return JSONResponse(status_code=200, content=relay_info)
 
 
 # -----------------------------------------------------------------------------
-# Gradio UI (simple dashboard)
+# Gradio dashboard
 # -----------------------------------------------------------------------------
 def ui_status() -> str:
     return _safe_json_dumps(
@@ -166,6 +315,13 @@ def ui_status() -> str:
             "last_received_at": _ts_to_str(LAST_RECEIVED_TS),
             "buffer_size": len(MESSAGES),
             "max_buffer": MAX_MESSAGES,
+            "relay": {
+                "forward_url": f"{WAKEUP_URL}/wakeup",
+                "timeout": RELAY_TIMEOUT,
+                "max_inflight": MAX_INFLIGHT,
+                "seen_size": len(SEEN),
+                "seen_ttl_seconds": SEEN_TTL_SECONDS,
+            },
             "auth": {
                 "token_enabled": bool(WEBHOOK_TOKEN),
                 "token_header": WEBHOOK_TOKEN_HEADER,
@@ -173,8 +329,7 @@ def ui_status() -> str:
             },
             "endpoints": {
                 "health": "/health",
-                "webhook_post": "/wakeup",
-                "webhook_post_alias": "/webhook",
+                "post": ["/wakeup/", "/webhook/"],
             },
         }
     )
@@ -194,12 +349,9 @@ def ui_clear() -> str:
     return "Cleared in-memory messages."
 
 
-with gr.Blocks(title="Webhook Listener Dashboard") as demo:
-    gr.Markdown("# Webhook Listener Dashboard")
-    gr.Markdown(
-        "Send a POST request with JSON to `/wakeup` or `/webhook` and it will appear below.\n\n"
-        "This is for payload debugging / confirming delivery."
-    )
+with gr.Blocks(title="Inserio Relay Dashboard") as demo:
+    gr.Markdown("# Inserio Relay Dashboard")
+    gr.Markdown("POST JSON to `/webhook/` (or `/wakeup/`). It will be logged and forwarded to the wakeup server.")
 
     with gr.Row():
         status_box = gr.Code(label="Status", value=ui_status(), language="json")
@@ -208,7 +360,7 @@ with gr.Blocks(title="Webhook Listener Dashboard") as demo:
     with gr.Row():
         limit = gr.Slider(1, 200, value=25, step=1, label="Recent message limit")
 
-    recent_box = gr.Code(label="Recent payloads (newest first)", value=ui_recent(25), language="json")
+    recent_box = gr.Code(label="Recent messages (newest first)", value=ui_recent(25), language="json")
 
     def refresh(limit_val: int):
         return ui_status(), ui_recent(limit_val)