"""Klikit webhook receiver — minimal reference (Python + Flask)."""
import hashlib
import hmac
import logging
import os
import threading
from flask import Flask, jsonify, request
SECRET = os.environ.get("KLIKIT_WEBHOOK_SECRET")
if not SECRET:
raise SystemExit("KLIKIT_WEBHOOK_SECRET not set")
SECRET_BYTES = SECRET.encode("utf-8")
app = Flask(__name__)
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(message)s")
# Replace with your own persistent store (Redis SETNX, Postgres unique index, etc).
_seen_event_ids: set[str] = set()
_seen_lock = threading.Lock()
@app.post("/webhooks/klikit")
def receive():
sig_header = request.headers.get("x-klikit-signature")
event_id = request.headers.get("x-klikit-event-id")
event_type = request.headers.get("x-klikit-event-type")
if not sig_header or not event_id or not event_type:
return jsonify(error="missing klikit headers"), 400
# IMPORTANT: compute HMAC over the raw body bytes, not the parsed JSON.
raw_body = request.get_data(cache=True)
expected = hmac.new(SECRET_BYTES, raw_body, hashlib.sha256).hexdigest()
if not hmac.compare_digest(sig_header, expected):
return jsonify(error="invalid signature"), 401
with _seen_lock:
if event_id in _seen_event_ids:
return jsonify(status="duplicate-ignored"), 200
_seen_event_ids.add(event_id)
payload = request.get_json(silent=True) or {}
# Process asynchronously so klikit's 10s timeout isn't tied to downstream work.
threading.Thread(
target=_handle_event, args=(event_type, payload), daemon=True
).start()
return jsonify(status="received"), 200
def _handle_event(event_type: str, payload: dict) -> None:
brand_id = payload.get("brand_id")
branch_id = payload.get("branch_id")
orders = payload.get("orders") or []
logging.info(
"[%s] brand=%s branch=%s orders=%d",
event_type, brand_id, branch_id, len(orders),
)
if event_type == "klikit.order.created.v2":
pass # push to POS / kitchen display
elif event_type == "klikit.order.status.updated":
pass # update order record
elif event_type == "klikit.order.cart.updated":
pass # replace cart state
else:
logging.warning("unknown event type: %s", event_type)
if __name__ == "__main__":
app.run(host="0.0.0.0", port=int(os.environ.get("PORT", "8080")))
