Skip to main content

​
Overview

Webhooks enable real-time notifications for critical e-commerce events including payments, order status changes, and shipping updates. The system receives webhooks from payment gateways and shipping providers to automate order processing.

​
Webhook Types

​
Payment Webhooks

Receive payment status updates from integrated payment gateways:
  • Payment Success - Order payment completed successfully
  • Payment Failed - Payment attempt failed
  • Refund Initiated - Refund process started
  • Refund Completed - Refund successfully processed

​
Shipping Webhooks

Receive shipping status updates from logistics partners:
  • Order Shipped - Order dispatched from warehouse
  • In Transit - Order in delivery
  • Out for Delivery - Order out for final delivery
  • Delivered - Order successfully delivered
  • Delivery Failed - Delivery attempt unsuccessful

​
User Activity Webhooks

Receive notifications for user-related events:
  • Points Update - User loyalty points changed (see TssPointHistoryController.py:1)
  • Transfer Completed - Money transfer webhook events (see UserController.py:1)

​
Webhook Endpoints

​
Payment Gateway Webhook

POST /api/v2/webhook/payment
Handles payment gateway callbacks for order payment status updates.

​
Shipping Provider Webhook

POST /api/v2/webhook/shipping
Receives shipping status updates from logistics partners.

​
Gmetri Points Webhook

POST /api/v2/webhook/gmetri-points
Handles customer points updates from Gmetri integration.

​
Webhook Security

​
Authentication

Webhooks require proper authentication to prevent unauthorized access: 
import hmac
import hashlib

def verify_webhook_signature(payload, signature, secret):
    """
    Verify webhook signature to ensure authenticity
    """
    expected_signature = hmac.new(
        secret.encode(),
        payload.encode(),
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(signature, expected_signature)

​
IP Whitelisting

For enhanced security, configure IP whitelisting for webhook sources: 
ALLOWED_WEBHOOK_IPS = [
    '192.168.1.100',
    '10.0.0.50',
]

def check_webhook_ip(req):
    client_ip = req.env.get('REMOTE_ADDR')
    if client_ip not in ALLOWED_WEBHOOK_IPS:
        raise falcon.HTTPForbidden()

​
Webhook Payload Examples

​
Payment Success Webhook

{
  "event": "payment.success",
  "order_id": "ORD-2026-12345",
  "payment_id": "pay_abc123xyz",
  "amount": 2499.00,
  "currency": "INR",
  "status": "captured",
  "method": "card",
  "timestamp": "2026-03-08T10:30:00Z",
  "signature": "sha256_signature_here"
}

​
Shipping Status Webhook

{
  "event": "shipment.delivered",
  "order_id": "ORD-2026-12345",
  "awb": "AWB123456789",
  "status": "delivered",
  "delivered_at": "2026-03-08T15:45:00Z",
  "location": {
    "city": "Mumbai",
    "state": "Maharashtra",
    "pincode": "400001"
  },
  "recipient": "John Doe",
  "signature_url": "https://cdn.example.com/signatures/xyz.jpg"
}

​
Points Update Webhook

{
  "event": "points.updated",
  "user_id": 12345,
  "transferId": "TXN-ABC123",
  "points": 500,
  "type": "credit",
  "reason": "order_completion",
  "timestamp": "2026-03-08T12:00:00Z"
}

​
Processing Webhooks

​
Webhook Handler Implementation

import falcon
import json
from controllers.OrderController import update_order_status
from utils.logger import get_logger

logger = get_logger()

class PaymentWebhook:
    def on_post(self, req, resp):
        """
        Handle payment webhook callbacks
        """
        try:
            # Parse webhook data
            webhook_data = req.context['data']
            event = webhook_data.get('event')
            order_id = webhook_data.get('order_id')
            
            # Verify webhook signature
            signature = req.headers.get('X-Webhook-Signature')
            if not verify_signature(webhook_data, signature):
                raise falcon.HTTPUnauthorized()
            
            # Process based on event type
            if event == 'payment.success':
                update_order_status(order_id, 'paid')
                logger.info(f"Payment webhook processed: {order_id}")
            
            resp.status = falcon.HTTP_200
            resp.text = json.dumps({"status": "success"})
            
        except Exception as e:
            logger.error(f"Webhook processing error: {str(e)}")
            raise falcon.HTTPBadRequest()

​
Asynchronous Processing

For time-consuming operations, process webhooks asynchronously: 
from libs.celery import app

@app.task
def process_delivery_webhook(webhook_data):
    """
    Process delivery webhook in background
    """
    order_id = webhook_data.get('order_id')
    
    # Update order status
    update_order_status(order_id, 'delivered')
    
    # Send delivery notification
    send_delivered_mail_and_sms(order_id)
    
    # Update inventory
    mark_order_complete(order_id)

​
Retry Mechanism

Implement retry logic for failed webhook processing: 
from settings import OZONETEL_CALLBACK_MAX_RETRIES

@app.task(bind=True, max_retries=OZONETEL_CALLBACK_MAX_RETRIES)
def retry_webhook_processing(self, webhook_data):
    try:
        process_webhook(webhook_data)
    except Exception as exc:
        # Retry with exponential backoff
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)

​
Webhook Logging

All webhook events are logged for auditing and debugging: 
from models.ExternalApi import ExternalApiLogsModel

def log_webhook(webhook_data, status, response):
    """
    Log webhook event to database
    """
    log = ExternalApiLogsModel(
        event_type=webhook_data.get('event'),
        payload=json.dumps(webhook_data),
        status=status,
        response=response,
        created_at=datetime.utcnow()
    )
    session.add(log)
    session.commit()

​
Testing Webhooks

​
Local Testing

Use tools like ngrok to expose your local server for webhook testing: 
# Start ngrok tunnel
ngrok http 8000

# Configure webhook URL in provider dashboard
https://your-ngrok-url.ngrok.io/api/v2/webhook/payment

​
Webhook Simulation

Test webhook handlers with simulated payloads: 
curl -X POST http://localhost:8000/api/v2/webhook/payment \
  -H "Content-Type: application/json" \
  -H "X-Webhook-Signature: test_signature" \
  -d '{
    "event": "payment.success",
    "order_id": "TEST-001",
    "amount": 1000.00,
    "status": "captured"
  }'

​
Webhook Monitoring

Monitor webhook health using:
  • Success Rate - Percentage of successfully processed webhooks
  • Processing Time - Average time to process webhooks
  • Failure Alerts - Notifications for repeated webhook failures
  • Queue Depth - Number of pending webhooks in Celery queues

​
Common Integration Patterns

​
Order Status Updates

# Shipping webhook triggers order status update
Webhook Received → Validate Signature → Update Order Status → Send Notification → Update Inventory

​
Payment Processing

# Payment webhook triggers order confirmation
Payment Webhook → Verify Payment → Update Order → Send Confirmation Email → Trigger Fulfillment

​
Refund Processing

# Refund webhook triggers refund workflow
Refund Webhook → Validate Refund → Update Order → Process Refund → Send Notification

​
Slack Notifications

Webhook events can trigger Slack notifications: 
import requests

def send_slack_webhook(message):
    slack_webhook_url = os.getenv('SLACK_WEBHOOK_URL')
    requests.post(slack_webhook_url, json={'text': message})
See OrderController.py for Slack webhook implementation.

​
Best Practices

Security Considerations:
  1. Always verify webhook signatures
  2. Use HTTPS endpoints only
  3. Implement IP whitelisting when possible
  4. Log all webhook events for audit trails
  5. Use idempotency keys to prevent duplicate processing
  6. Set reasonable timeout values for webhook handlers
  7. Implement retry logic with exponential backoff

Celery Tasks

Background processing for webhook events

Order API

Order management endpoints

Build docs developers (and LLMs) love

Get started for freeTalk to us