Webhook Events & Integration

Configuration

Method 1: Merchant Dashboard 1. Log into your Piaxis merchant dashboard 2. Navigate to Settings → Webhooks 3. Add your webhook endpoint URLs 4. Select the events you want to receive 5. Configure retry and timeout settings

Method 2: API Configuration

POST /api/webhooks/endpoints

Create a new webhook endpoint.

Example Request:

POST /api/webhooks/endpoints HTTP/1.1
Host: api.gopiaxis.com
api-key: YOUR_API_KEY
Content-Type: application/json

{
  "url": "https://yourapp.com/webhooks/piaxis",
  "events": [
    "payment.completed",
    "escrow.created",
    "escrow.term_fulfilled",
    "escrow.released"
  ],
  "description": "Production webhook endpoint",
  "secret": "your_webhook_secret_key",
  "enabled": true
}

Response:

{
  "id": "wh_1234567890abcdef",
  "url": "https://yourapp.com/webhooks/piaxis",
  "events": ["payment.completed", "escrow.created", "escrow.term_fulfilled", "escrow.released"],
  "description": "Production webhook endpoint",
  "enabled": true,
  "created": "2024-01-15T10:30:00Z",
  "updated": "2024-01-15T10:30:00Z"
}

Endpoint Requirements

Your webhook endpoint must:

1. Accept HTTP POST requests with JSON payloads

2. Respond with HTTP 2xx status codes (200-299) for successful processing

3. Respond within 30 seconds to avoid timeout

4. Handle duplicate events idempotently

5. Verify webhook signatures for security

6. Use HTTPS for production environments

Example Endpoint Structure:

from flask import Flask, request, jsonify
import hmac
import hashlib

app = Flask(__name__)

@app.route('/webhooks/piaxis', methods=['POST'])
def handle_piaxis_webhook():
    # Verify the webhook signature
    if not verify_webhook_signature(request.data, request.headers.get('X-Piaxis-Signature')):
        return jsonify({'error': 'Invalid signature'}), 401

    # Parse the event data
    event_data = request.get_json()
    event_type = event_data.get('event')

    # Process the event
    try:
        process_webhook_event(event_type, event_data)
        return jsonify({'status': 'success'}), 200
    except Exception as e:
        # Log the error but return 200 to prevent retries for invalid data
        app.logger.error(f"Webhook processing error: {str(e)}")
        return jsonify({'status': 'error', 'message': str(e)}), 200

Payment Events

Example: Payment Completed

{
  "event": "payment.completed",
  "event_id": "evt_1234567890abcdef",
  "created": "2024-02-10T12:00:00Z",
  "api_version": "2024-01-01",
  "data": {
    "payment_id": "f530533e-3761-4cde-9c9d-88c5be6493bb",
    "amount": "5000.00",
    "currency": "UGX",
    "status": "completed",
    "payment_method": "mtn",
    "reference": "REF123456",
    "description": "Service payment",
    "customer": {
      "email": "[email protected]",
      "phone_number": "+256700000000"
    },
    "merchant_id": "096b723a-45c5-4957-94d7-747835136265",
    "created_at": "2024-02-10T11:45:00Z",
    "completed_at": "2024-02-10T12:00:00Z",
    "metadata": {
      "order_id": "ORD-789",
      "customer_type": "premium"
    }
  }
}

Example: Payment Failed

{
  "event": "payment.failed",
  "event_id": "evt_abcdef1234567890",
  "created": "2024-02-10T12:05:00Z",
  "api_version": "2024-01-01",
  "data": {
    "payment_id": "f530533e-3761-4cde-9c9d-88c5be6493bb",
    "amount": "5000.00",
    "currency": "UGX",
    "status": "failed",
    "payment_method": "mtn",
    "error": {
      "code": "INSUFFICIENT_FUNDS",
      "message": "Insufficient funds in customer account",
      "provider_code": "MW001",
      "provider_message": "Your account balance is too low"
    },
    "customer": {
      "email": "[email protected]",
      "phone_number": "+256700000000"
    },
    "merchant_id": "096b723a-45c5-4957-94d7-747835136265",
    "created_at": "2024-02-10T11:45:00Z",
    "failed_at": "2024-02-10T12:05:00Z"
  }
}

Escrow Events

Example: Escrow Created

{
  "event": "escrow.created",
  "event_id": "evt_escrow_created_123",
  "created": "2024-02-10T10:00:00Z",
  "api_version": "2024-01-01",
  "data": {
    "escrow_id": "7680cfa9-b0cf-43a7-974b-108b89bd5ebe",
    "amount": "50000.00",
    "currency": "UGX",
    "status": "created",
    "payment_method": "mtn",
    "sender": {
      "email": "[email protected]",
      "phone_number": "+256700000000"
    },
    "recipient": {
      "email": "[email protected]",
      "phone_number": "+256700000001"
    },
    "terms": [
      {
        "term_id": "07a17756-a965-452a-9f0a-48ef55c38131",
        "type": "delivery_confirmation",
        "description": "Confirm product delivery",
        "fulfilled_by": "buyer",
        "status": "pending"
      },
      {
        "term_id": "08b18867-b076-563b-0a1b-59fg66d49242",
        "type": "service_completion",
        "description": "Complete installation service",
        "fulfilled_by": "seller",
        "status": "pending"
      }
    ],
    "auto_release_date": "2024-02-17T10:00:00Z",
    "merchant_id": "096b723a-45c5-4957-94d7-747835136265",
    "created_at": "2024-02-10T10:00:00Z",
    "metadata": {
      "order_id": "ORD-456",
      "product_category": "electronics"
    }
  }
}

Example: Escrow Term Fulfilled

{
  "event": "escrow.term_fulfilled",
  "event_id": "evt_term_fulfilled_456",
  "created": "2024-02-12T14:30:00Z",
  "api_version": "2024-01-01",
  "data": {
    "escrow_id": "7680cfa9-b0cf-43a7-974b-108b89bd5ebe",
    "term_id": "07a17756-a965-452a-9f0a-48ef55c38131",
    "term_type": "delivery_confirmation",
    "term_description": "Confirm product delivery",
    "fulfilled_by": "buyer",
    "fulfilled_at": "2024-02-12T14:30:00Z",
    "fulfillment_data": {
      "confirmation_code": "DEL123456",
      "notes": "Package received in good condition",
      "images": ["https://cdn.gopiaxis.com/images/delivery_proof_1.jpg"]
    },
    "remaining_terms": [
      {
        "term_id": "08b18867-b076-563b-0a1b-59fg66d49242",
        "type": "service_completion",
        "status": "pending"
      }
    ],
    "escrow_status": "partially_fulfilled"
  }
}

Disbursement Events

Example: Disbursement Item Completed

{
  "event": "disbursement.item_completed",
  "event_id": "evt_disbursement_item_789",
  "created": "2024-02-10T16:45:00Z",
  "api_version": "2024-01-01",
  "data": {
    "disbursement_id": "disb_1234567890abcdef",
    "item_id": "item_abcdef1234567890",
    "recipient": {
      "phone_number": "+256700000000",
      "name": "John Doe"
    },
    "amount": "10000.00",
    "currency": "UGX",
    "payment_method": "mtn",
    "status": "completed",
    "reference": "DISB-REF-001",
    "transaction_id": "txn_mtn_987654321",
    "completed_at": "2024-02-10T16:45:00Z",
    "batch_info": {
      "total_items": 50,
      "completed_items": 25,
      "failed_items": 2,
      "pending_items": 23
    }
  }
}

Account & System Events

Signature Verification

Every webhook request includes a signature in the X-Piaxis-Signature header. Always verify this signature to ensure the webhook came from Piaxis and hasn’t been tampered with.

Signature Algorithm: - HMAC-SHA256 of the raw request body - Signed with your webhook secret key - Prefixed with sha256=

Python Implementation:

import hmac
import hashlib

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    """
    Verify webhook signature

    Args:
        payload: Raw request body as bytes
        signature: X-Piaxis-Signature header value
        secret: Your webhook secret key

    Returns:
        True if signature is valid, False otherwise
    """
    if not signature.startswith('sha256='):
        return False

    expected_signature = 'sha256=' + hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(expected_signature, signature)

# Usage in Flask
from flask import Flask, request

@app.route('/webhooks/piaxis', methods=['POST'])
def handle_webhook():
    signature = request.headers.get('X-Piaxis-Signature')
    if not verify_webhook_signature(request.data, signature, WEBHOOK_SECRET):
        return {'error': 'Invalid signature'}, 401

    # Process webhook...
    return {'status': 'success'}, 200

Node.js Implementation:

const crypto = require('crypto');

function verifyWebhookSignature(payload, signature, secret) {
    if (!signature.startsWith('sha256=')) {
        return false;
    }

    const expectedSignature = 'sha256=' + crypto
        .createHmac('sha256', secret)
        .update(payload, 'utf8')
        .digest('hex');

    return crypto.timingSafeEqual(
        Buffer.from(expectedSignature),
        Buffer.from(signature)
    );
}

// Usage in Express
const express = require('express');
const app = express();

// Use raw body parser for webhooks
app.use('/webhooks/piaxis', express.raw({ type: 'application/json' }));

app.post('/webhooks/piaxis', (req, res) => {
    const signature = req.headers['x-piaxis-signature'];

    if (!verifyWebhookSignature(req.body, signature, process.env.WEBHOOK_SECRET)) {
        return res.status(401).json({ error: 'Invalid signature' });
    }

    const event = JSON.parse(req.body);
    // Process webhook...

    res.json({ status: 'success' });
});

IP Allowlisting

For additional security, you can allowlist Piaxis’s webhook IP addresses:

Production IPs: - 52.214.123.45 - 52.214.123.46 - 52.214.123.47

Sandbox IPs: - 52.214.123.100 - 52.214.123.101

Implementation Example:

ALLOWED_IPS = [
    '52.214.123.45',
    '52.214.123.46',
    '52.214.123.47'
]

def is_ip_allowed(ip_address):
    return ip_address in ALLOWED_IPS

@app.route('/webhooks/piaxis', methods=['POST'])
def handle_webhook():
    client_ip = request.environ.get('HTTP_X_FORWARDED_FOR') or request.environ.get('REMOTE_ADDR')

    if not is_ip_allowed(client_ip):
        return {'error': 'IP not allowed'}, 403

    # Continue with signature verification and processing...

Complete Flask Implementation

from flask import Flask, request, jsonify
import hmac
import hashlib
import json
import logging
from datetime import datetime

app = Flask(__name__)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Configuration
WEBHOOK_SECRET = 'your_webhook_secret_here'
PROCESSED_EVENTS = set()  # Simple in-memory deduplication

def verify_webhook_signature(payload: bytes, signature: str, secret: str) -> bool:
    if not signature or not signature.startswith('sha256='):
        return False

    expected_signature = 'sha256=' + hmac.new(
        secret.encode('utf-8'),
        payload,
        hashlib.sha256
    ).hexdigest()

    return hmac.compare_digest(expected_signature, signature)

def process_payment_event(event_type: str, data: dict):
    """Process payment-related events"""
    payment_id = data['payment_id']

    if event_type == 'payment.completed':
        # Update order status, send confirmation email, etc.
        logger.info(f"Payment {payment_id} completed for amount {data['amount']} {data['currency']}")
        # Your business logic here

    elif event_type == 'payment.failed':
        # Handle failed payment, notify customer, retry logic, etc.
        error_code = data.get('error', {}).get('code')
        logger.warning(f"Payment {payment_id} failed with error: {error_code}")
        # Your error handling logic here

def process_escrow_event(event_type: str, data: dict):
    """Process escrow-related events"""
    escrow_id = data['escrow_id']

    if event_type == 'escrow.created':
        # Send notifications to buyer and seller
        logger.info(f"New escrow {escrow_id} created for amount {data['amount']}")

    elif event_type == 'escrow.term_fulfilled':
        # Update fulfillment status, check if all terms completed
        term_id = data['term_id']
        logger.info(f"Term {term_id} fulfilled for escrow {escrow_id}")

    elif event_type == 'escrow.released':
        # Notify recipient, update order status, send receipts
        logger.info(f"Escrow {escrow_id} released to recipient")

def process_disbursement_event(event_type: str, data: dict):
    """Process disbursement-related events"""
    if event_type == 'disbursement.item_completed':
        # Update recipient records, send notifications
        item_id = data['item_id']
        recipient = data['recipient']['phone_number']
        logger.info(f"Disbursement item {item_id} completed for {recipient}")

@app.route('/webhooks/piaxis', methods=['POST'])
def handle_piaxis_webhook():
    # Verify signature
    signature = request.headers.get('X-Piaxis-Signature')
    if not verify_webhook_signature(request.data, signature, WEBHOOK_SECRET):
        logger.warning("Invalid webhook signature")
        return jsonify({'error': 'Invalid signature'}), 401

    try:
        # Parse event data
        event_data = request.get_json()
        event_type = event_data.get('event')
        event_id = event_data.get('event_id')
        data = event_data.get('data', {})

        # Idempotency check
        if event_id in PROCESSED_EVENTS:
            logger.info(f"Event {event_id} already processed, skipping")
            return jsonify({'status': 'already_processed'}), 200

        # Log the event
        logger.info(f"Processing webhook event: {event_type} (ID: {event_id})")

        # Route to appropriate handler
        if event_type.startswith('payment.'):
            process_payment_event(event_type, data)
        elif event_type.startswith('escrow.'):
            process_escrow_event(event_type, data)
        elif event_type.startswith('disbursement.'):
            process_disbursement_event(event_type, data)
        else:
            logger.warning(f"Unknown event type: {event_type}")

        # Mark as processed
        PROCESSED_EVENTS.add(event_id)

        return jsonify({'status': 'success'}), 200

    except Exception as e:
        logger.error(f"Error processing webhook: {str(e)}")
        # Return 200 to prevent retries for invalid data
        return jsonify({'status': 'error', 'message': str(e)}), 200

@app.route('/webhooks/test', methods=['GET'])
def webhook_test():
    """Test endpoint to verify webhook URL is accessible"""
    return jsonify({'status': 'webhook_endpoint_active', 'timestamp': datetime.now().isoformat()})

if __name__ == '__main__':
    app.run(debug=True, port=5000)

Complete Express.js Implementation

const express = require('express');
const crypto = require('crypto');
const app = express();

// Configuration
const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET || 'your_webhook_secret_here';
const processedEvents = new Set(); // Simple in-memory deduplication

// Middleware for webhook endpoint (need raw body for signature verification)
app.use('/webhooks/piaxis', express.raw({ type: 'application/json' }));
app.use(express.json()); // For other endpoints

function verifyWebhookSignature(payload, signature, secret) {
    if (!signature || !signature.startsWith('sha256=')) {
        return false;
    }

    const expectedSignature = 'sha256=' + crypto
        .createHmac('sha256', secret)
        .update(payload, 'utf8')
        .digest('hex');

    return crypto.timingSafeEqual(
        Buffer.from(expectedSignature),
        Buffer.from(signature)
    );
}

function processPaymentEvent(eventType, data) {
    const paymentId = data.payment_id;

    switch (eventType) {
        case 'payment.completed':
            console.log(`Payment ${paymentId} completed for ${data.amount} ${data.currency}`);
            // Your business logic here
            break;

        case 'payment.failed':
            const errorCode = data.error?.code;
            console.warn(`Payment ${paymentId} failed with error: ${errorCode}`);
            // Your error handling logic here
            break;
    }
}

function processEscrowEvent(eventType, data) {
    const escrowId = data.escrow_id;

    switch (eventType) {
        case 'escrow.created':
            console.log(`New escrow ${escrowId} created for ${data.amount}`);
            break;

        case 'escrow.term_fulfilled':
            const termId = data.term_id;
            console.log(`Term ${termId} fulfilled for escrow ${escrowId}`);
            break;

        case 'escrow.released':
            console.log(`Escrow ${escrowId} released to recipient`);
            break;
    }
}

function processDisbursementEvent(eventType, data) {
    if (eventType === 'disbursement.item_completed') {
        const itemId = data.item_id;
        const recipient = data.recipient.phone_number;
        console.log(`Disbursement item ${itemId} completed for ${recipient}`);
    }
}

app.post('/webhooks/piaxis', (req, res) => {
    const signature = req.headers['x-piaxis-signature'];

    // Verify signature
    if (!verifyWebhookSignature(req.body, signature, WEBHOOK_SECRET)) {
        console.warn('Invalid webhook signature');
        return res.status(401).json({ error: 'Invalid signature' });
    }

    try {
        // Parse event data
        const eventData = JSON.parse(req.body);
        const eventType = eventData.event;
        const eventId = eventData.event_id;
        const data = eventData.data || {};

        // Idempotency check
        if (processedEvents.has(eventId)) {
            console.log(`Event ${eventId} already processed, skipping`);
            return res.json({ status: 'already_processed' });
        }

        // Log the event
        console.log(`Processing webhook event: ${eventType} (ID: ${eventId})`);

        // Route to appropriate handler
        if (eventType.startsWith('payment.')) {
            processPaymentEvent(eventType, data);
        } else if (eventType.startsWith('escrow.')) {
            processEscrowEvent(eventType, data);
        } else if (eventType.startsWith('disbursement.')) {
            processDisbursementEvent(eventType, data);
        } else {
            console.warn(`Unknown event type: ${eventType}`);
        }

        // Mark as processed
        processedEvents.add(eventId);

        res.json({ status: 'success' });

    } catch (error) {
        console.error('Error processing webhook:', error.message);
        // Return 200 to prevent retries for invalid data
        res.json({ status: 'error', message: error.message });
    }
});

app.get('/webhooks/test', (req, res) => {
    res.json({
        status: 'webhook_endpoint_active',
        timestamp: new Date().toISOString()
    });
});

const PORT = process.env.PORT || 3000;
app.listen(PORT, () => {
    console.log(`Webhook server running on port ${PORT}`);
});

Delivery Guarantees

Piaxis implements the following delivery guarantees:

• Retry Policy: Failed webhook deliveries are retried with exponential backoff

• Retry Schedule: 1s, 5s, 25s, 125s, 625s (up to 5 attempts)

• Timeout: 30 seconds per attempt

• Success Criteria: HTTP status codes 200-299

Retry Headers:

Piaxis includes retry information in webhook headers:

X-Piaxis-Delivery-ID: del_1234567890abcdef
X-Piaxis-Delivery-Attempt: 3
X-Piaxis-Event-ID: evt_abcdef1234567890
X-Piaxis-Webhook-ID: wh_fedcba0987654321

Handling Failures

Best Practices for Error Handling:

@app.route('/webhooks/piaxis', methods=['POST'])
def handle_webhook():
    try:
        # Process webhook
        process_webhook_event(event_data)
        return jsonify({'status': 'success'}), 200

    except ValidationError as e:
        # Data validation errors - don't retry
        logger.error(f"Webhook validation error: {e}")
        return jsonify({'status': 'error', 'message': str(e)}), 200

    except ExternalServiceError as e:
        # External service errors - can retry
        logger.error(f"External service error: {e}")
        return jsonify({'status': 'error', 'message': str(e)}), 500

    except DatabaseError as e:
        # Database errors - can retry
        logger.error(f"Database error: {e}")
        return jsonify({'status': 'error', 'message': str(e)}), 500

Manual Retry

If you need to manually replay missed webhooks:

POST /api/webhooks/replay

Replay webhook events for a specific time range.

Example Request:

POST /api/webhooks/replay HTTP/1.1
Host: api.gopiaxis.com
api-key: YOUR_API_KEY
Content-Type: application/json

{
  "start_date": "2024-02-10T00:00:00Z",
  "end_date": "2024-02-10T23:59:59Z",
  "event_types": ["payment.completed", "escrow.released"],
  "webhook_endpoint_id": "wh_1234567890abcdef"
}

Local Development Setup

Using ngrok for HTTPS tunneling:

# Install ngrok
npm install -g ngrok

# Start your local webhook server
python webhook_server.py  # or node webhook_server.js

# In another terminal, expose your local server
ngrok http 5000

# Use the ngrok HTTPS URL in your webhook configuration
# Example: https://abc123.ngrok.io/webhooks/piaxis

Environment Variables:

# .env file
WEBHOOK_SECRET=your_webhook_secret_key
piaxis_API_KEY=your_api_key
DATABASE_URL=your_database_url
LOG_LEVEL=info

Test Event Generation

Manual Test Events:

POST /api/webhooks/test

Send test webhook events to your endpoint.

Example Request:

POST /api/webhooks/test HTTP/1.1
Host: api.gopiaxis.com
api-key: YOUR_API_KEY
Content-Type: application/json

{
  "webhook_endpoint_id": "wh_1234567890abcdef",
  "event_type": "payment.completed",
  "test_data": {
    "payment_id": "test_payment_123",
    "amount": "1000.00",
    "currency": "UGX"
  }
}

Webhook Testing Tools:

• Webhook.site: https://webhook.site/ - Inspect webhook payloads

• RequestBin: https://requestbin.com/ - Capture and debug webhooks

• Postman: Create collections for webhook testing

Webhook Analytics

Monitor webhook delivery through your merchant dashboard:

• Delivery Rate: Percentage of successful deliveries

• Response Times: Average response time from your endpoint

• Error Rates: Rate of failed deliveries by error type

• Event Volume: Number of events sent over time

API for Webhook Analytics:

GET /api/webhooks/analytics

Get webhook delivery analytics.

Query Parameters:

• start_date – Start date for analytics period

• end_date – End date for analytics period

• webhook_endpoint_id – Specific webhook endpoint ID

Example Response:

{
  "period": {
    "start": "2024-02-01T00:00:00Z",
    "end": "2024-02-28T23:59:59Z"
  },
  "metrics": {
    "total_events": 1250,
    "successful_deliveries": 1200,
    "failed_deliveries": 50,
    "success_rate": 96.0,
    "average_response_time_ms": 150,
    "retry_rate": 4.0
  },
  "event_breakdown": {
    "payment.completed": 600,
    "payment.failed": 25,
    "escrow.created": 200,
    "escrow.released": 180,
    "disbursement.completed": 245
  }
}

Logging & Alerting

Structured Logging Example:

import json
import logging
from datetime import datetime

def log_webhook_event(event_type, event_id, success, duration_ms, error=None):
    log_data = {
        'timestamp': datetime.now().isoformat(),
        'event_type': event_type,
        'event_id': event_id,
        'success': success,
        'duration_ms': duration_ms,
        'service': 'webhook_handler'
    }

    if error:
        log_data['error'] = str(error)
        logging.error(json.dumps(log_data))
    else:
        logging.info(json.dumps(log_data))

Alerting Recommendations:

• Alert on webhook failure rate > 5%

• Alert on average response time > 5 seconds

• Alert on consecutive failures > 10

• Monitor for missing expected events

Performance

1. Async Processing: Process webhooks asynchronously to respond quickly

2. Database Optimization: Use efficient queries and indexes

3. Caching: Cache frequently accessed data

4. Rate Limiting: Implement rate limiting on your webhook endpoints

from celery import Celery
from flask import Flask, request, jsonify

app = Flask(__name__)
celery = Celery('webhook_processor')

@celery.task
def process_webhook_async(event_type, event_data):
    # Async processing logic here
    pass

@app.route('/webhooks/piaxis', methods=['POST'])
def handle_webhook():
    # Verify signature and basic validation
    if not verify_webhook_signature(...):
        return jsonify({'error': 'Invalid signature'}), 401

    event_data = request.get_json()

    # Queue for async processing
    process_webhook_async.delay(event_data['event'], event_data)

    # Respond immediately
    return jsonify({'status': 'queued'}), 200

Security

1. Always verify signatures - Never process unverified webhooks

2. Use HTTPS only - Webhooks contain sensitive data

3. Implement IP allowlisting - Additional security layer

4. Validate event data - Don’t trust webhook data blindly

5. Log security events - Monitor for suspicious activity

Reliability

1. Idempotency: Handle duplicate events gracefully

2. Error recovery: Implement proper error handling and recovery

3. Monitoring: Set up comprehensive monitoring and alerting

4. Backup processing: Have fallback mechanisms for critical events

# Idempotency example using database
def process_webhook_idempotent(event_id, event_data):
    # Check if already processed
    if ProcessedEvent.objects.filter(event_id=event_id).exists():
        return {'status': 'already_processed'}

    try:
        # Process the event
        result = process_event(event_data)

        # Mark as processed
        ProcessedEvent.objects.create(
            event_id=event_id,
            processed_at=datetime.now(),
            status='success'
        )

        return result

    except Exception as e:
        # Log error but mark as processed to prevent retries
        ProcessedEvent.objects.create(
            event_id=event_id,
            processed_at=datetime.now(),
            status='error',
            error_message=str(e)
        )
        raise

Common Issues

1. Webhook Signature Verification Failing

Problem: Webhook signature verification always fails

Solutions:
- Ensure you're using the raw request body (not parsed JSON)
- Check that your webhook secret is correct
- Verify the signature format (should start with 'sha256=')
- Use timing-safe comparison functions

2. High Webhook Failure Rate

Problem: Many webhook deliveries are failing

Solutions:
- Check endpoint response times (should be < 30 seconds)
- Verify your endpoint returns 2xx status codes
- Ensure HTTPS certificate is valid
- Check for rate limiting on your server

3. Missing Webhook Events

Problem: Expected webhook events are not being received

Solutions:
- Verify webhook endpoint configuration in dashboard
- Check that event types are enabled for your endpoint
- Use webhook replay API to resend missed events
- Monitor webhook delivery logs in dashboard

4. Duplicate Event Processing

Problem: Same webhook event processed multiple times

Solutions:
- Implement idempotency using event_id
- Use database transactions for critical operations
- Store processed event IDs in cache/database
- Handle race conditions properly

Debug Tools

Webhook URL Validation:

# Test webhook endpoint accessibility
curl -X POST https://yourapp.com/webhooks/piaxis \
  -H "Content-Type: application/json" \
  -H "X-Piaxis-Signature: sha256=test" \
  -d '{"test": "data"}'

Signature Verification Test:

# Test signature generation
import hmac
import hashlib

payload = '{"event": "test"}'
secret = 'your_webhook_secret'

signature = 'sha256=' + hmac.new(
    secret.encode(),
    payload.encode(),
    hashlib.sha256
).hexdigest()

print(f"Test signature: {signature}")