""" Production webhook server using FastAPI + Gunicorn/Uvicorn Receives Square webhooks with signature verification """ import os import json import asyncio from typing import Dict, Any from fastapi import FastAPI, Request, Response, status from fastapi.responses import JSONResponse import dotenv # Load environment variables dotenv.load_dotenv() # Square webhook validation from square.utils.webhooks_helper import verify_signature # Configuration SQUARE_WEBHOOK_SIGNATURE_KEY = os.getenv("SANDBOX_SQUARE_WEBHOOK_INVENTORY_ID") SQUARE_WEBHOOK_URL = os.getenv("SQUARE_WEBHOOK_URL") # In-memory store for processed event IDs (idempotency) processed_event_ids = set() MAX_STORED_EVENT_IDS = 10000 # Create FastAPI app app = FastAPI(title="StudSpider Webhook Server") @app.get("/") async def root(): """Landing page""" return { "service": "StudSpider Webhook Server", "status": "running", "endpoints": {"webhook": "/hook (POST)", "health": "/health (GET)"}, } @app.get("/health") async def health_check(): """Health check endpoint for monitoring""" return { "status": "healthy", "signature_verification": ( "enabled" if SQUARE_WEBHOOK_SIGNATURE_KEY else "disabled" ), } @app.post("/hook") async def webhook_handler(request: Request): """ Handle Square webhook events Square sends POST requests with: - x-square-hmacsha256-signature header - JSON body with event data - event_id for idempotency """ # Read raw body for signature verification body = await request.body() body_str = body.decode("utf-8") # Get signature header signature_header = request.headers.get("x-square-hmacsha256-signature", "") # Verify Square webhook signature if SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL: if not signature_header: print( f"[ERROR] Missing x-square-hmacsha256-signature header from {request.client.host}" ) return JSONResponse( status_code=status.HTTP_401_UNAUTHORIZED, content={"error": "Missing signature header"}, ) try: is_valid = verify_signature( request_body=body_str, signature_header=signature_header, signature_key=SQUARE_WEBHOOK_SIGNATURE_KEY, notification_url=SQUARE_WEBHOOK_URL, ) if not is_valid: print(f"[ERROR] Invalid webhook signature from {request.client.host}") return JSONResponse( status_code=status.HTTP_401_UNAUTHORIZED, content={"error": "Invalid signature"}, ) print(f"[INFO] ✓ Webhook signature verified from {request.client.host}") except Exception as e: print(f"[ERROR] Signature verification failed: {e}") return JSONResponse( status_code=status.HTTP_401_UNAUTHORIZED, content={"error": "Signature verification failed"}, ) else: print( f"[WARN] Webhook signature verification DISABLED - configure SQUARE_WEBHOOK_SIGNATURE_KEY" ) # Parse JSON data try: data = json.loads(body_str) except json.JSONDecodeError as e: print(f"[ERROR] Invalid JSON from {request.client.host}: {e}") return JSONResponse( status_code=status.HTTP_400_BAD_REQUEST, content={"error": "Invalid JSON"} ) # Check idempotency (event_id) event_id = data.get("event_id") if event_id: if event_id in processed_event_ids: print( f"[INFO] Duplicate event {event_id} - already processed (idempotency)" ) return JSONResponse( status_code=status.HTTP_200_OK, content={ "status": "success", "message": "Event already processed", "event_id": event_id, }, ) # Add to processed set processed_event_ids.add(event_id) # Prevent memory growth if len(processed_event_ids) > MAX_STORED_EVENT_IDS: processed_event_ids.pop() # Get event info for quick response webhook_type = data.get("type", "unknown") event_id_val = data.get("event_id", "none") merchant_id = data.get("merchant_id", "unknown") # RESPOND IMMEDIATELY - Square requires fast response (< 2 seconds) response_data = { "status": "success", "message": "Webhook received", "event_id": event_id_val, } # Process webhook in background (async task) asyncio.create_task( process_webhook( webhook_type, event_id_val, merchant_id, data, request.client.host ) ) # Return immediate response return JSONResponse(status_code=status.HTTP_200_OK, content=response_data) async def process_webhook( webhook_type: str, event_id: str, merchant_id: str, data: Dict[str, Any], client_ip: str, ): """ Process webhook data asynchronously after responding to Square This runs in the background and doesn't block the HTTP response """ try: print("\n" + "=" * 60) print(f"[WEBHOOK] Received at {asyncio.get_event_loop().time()}") print(f"[WEBHOOK] From: {client_ip}") print(f"[WEBHOOK] Type: {webhook_type}") print(f"[WEBHOOK] Event ID: {event_id}") print(f"[WEBHOOK] Merchant: {merchant_id}") print(f"[WEBHOOK] Full payload:") print(json.dumps(data, indent=2)) print("=" * 60 + "\n") # Process based on webhook type if webhook_type == "inventory.count.updated": print(f"[PROCESS] Processing inventory count update...") inventory_counts = ( data.get("data", {}).get("object", {}).get("inventory_counts", []) ) for count in inventory_counts: catalog_id = count.get("catalog_object_id") quantity = count.get("quantity") state = count.get("state") print( f"[PROCESS] Catalog: {catalog_id}, Qty: {quantity}, State: {state}" ) elif webhook_type == "catalog.version.updated": print(f"[PROCESS] Processing catalog version update...") # Add your catalog processing logic here else: print(f"[PROCESS] Webhook type '{webhook_type}' - no handler configured") print(f"[PROCESS] ✓ Webhook processing complete for {event_id}") except Exception as e: print(f"[ERROR] Failed to process webhook {event_id}: {e}") if __name__ == "__main__": # For development/testing - use uvicorn directly import uvicorn print("\n" + "=" * 60) print("StudSpider Webhook Server - Development Mode") print("=" * 60) print( f"Signature verification: {'ENABLED' if SQUARE_WEBHOOK_SIGNATURE_KEY else 'DISABLED'}" ) print(f"Webhook URL: {SQUARE_WEBHOOK_URL or 'NOT CONFIGURED'}") print("\nFor production, use Gunicorn:") print( " gunicorn webhook_server:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:7433" ) print("=" * 60 + "\n") uvicorn.run(app, host="0.0.0.0", port=7433, log_level="info")