Webhook - Open Market Api

Webhook Integration

Open Market uses webhooks to notify your application when a payment is successful. Only successful transactions trigger webhook notifications. This real-time notification system allows you to automatically update your application’s state, fulfill orders, or trigger other business processes.

Webhook Authentication

Webhooks are authenticated using your private key. Each webhook request includes a signature in the opm-signature header that you should validate to ensure the webhook is legitimate. This prevents unauthorized parties from sending fake webhook events to your endpoint.

Webhook Payload

When a payment is successful, we’ll send a POST request to your configured webhook URL with the following JSON payload:

{
  "transaction_id": "tr_123456789",
  "reference": "ORDER123ABC",
  "meta_data": {
    "order_id": "ORD_123",
    "customer_id": "CUS_456",
    "product_ids": ["PROD_789", "PROD_012"]
  },
  "status": "completed",
  "transaction_type": "payment",
  "amount": "5000.00",
  "currency": "XOF",
  "buyer_number": "+22961234567",
  "buyer_name": "John Doe",
  "buyer_email": "john@example.com",
  "seller_email": "merchant@example.com",
  "created_at": "2024-01-20T14:30:00Z",
  "updated_at": "2024-01-20T14:32:00Z"
}

Handling Webhooks

Here’s an example of how to handle and validate webhooks:

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

// Middleware to parse JSON bodies
app.use(express.json());

// Utility function to validate signature
function validateSignature(payload, signature, privateKey) {
  const hmac = crypto.createHmac('sha256', privateKey);
  const expectedSignature = hmac.update(JSON.stringify(payload)).digest('hex');
  return crypto.timingSafeEqual(
    Buffer.from(signature),
    Buffer.from(expectedSignature)
  );
}

// Database model for processed webhooks (example using Mongoose)
const ProcessedWebhook = mongoose.model('ProcessedWebhook', {
  transaction_id: { type: String, unique: true },
  processed_at: { type: Date, default: Date.now }
});

// Webhook handler
app.post('/webhook', async (req, res) => {
  try {
    const signature = req.headers['opm-signature'];
    const payload = req.body;
    const privateKey = process.env.OPM_PRIVATE_KEY;

    // 1. Validate signature
    if (!validateSignature(payload, signature, privateKey)) {
      console.error('Invalid signature for transaction:', payload.transaction_id);
      return res.status(400).send('Invalid signature');
    }

    // 2. Check for duplicate webhook
    const existingWebhook = await ProcessedWebhook.findOne({
      transaction_id: payload.transaction_id
    });

    if (existingWebhook) {
      console.log('Duplicate webhook received:', payload.transaction_id);
      return res.status(200).send('Webhook already processed');
    }

    // 3. Process the webhook asynchronously
    res.status(200).send('Webhook received');

    // 4. Business logic
    await processWebhookData(payload);

    // 5. Mark webhook as processed
    await ProcessedWebhook.create({
      transaction_id: payload.transaction_id
    });

  } catch (error) {
    console.error('Webhook processing error:', error);
    // Still return 200 to acknowledge receipt
    if (!res.headersSent) {
      res.status(200).send('Webhook received with errors');
    }
    
    // Store failed webhook for retry
    await storeFailedWebhook(payload, error);
  }
});

// Business logic processing
async function processWebhookData(payload) {
  const {
    transaction_id,
    reference,
    meta_data,
    status,
    amount,
    buyer_email,
    buyer_name
  } = payload;

  // Update order status
  await Orders.findOneAndUpdate(
    { reference },
    {
      payment_status: status,
      payment_confirmed: true,
      transaction_id
    }
  );

  // Process meta data
  if (meta_data?.order_id) {
    await fulfillOrder(meta_data.order_id);
  }

  // Send confirmation email
  await sendPaymentConfirmation({
    email: buyer_email,
    name: buyer_name,
    amount,
    reference
  });

  // Additional business logic...
}

// Failed webhook storage
async function storeFailedWebhook(payload, error) {
  await FailedWebhooks.create({
    transaction_id: payload.transaction_id,
    payload: payload,
    error: error.message,
    created_at: new Date(),
    retry_count: 0
  });
}

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

from flask import Flask, request, jsonify
import hmac
import hashlib
import json
import os
from datetime import datetime
from typing import Dict, Any
from sqlalchemy import create_engine, Column, String, DateTime, Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

app = Flask(__name__)

# Database setup
Base = declarative_base()
engine = create_engine('sqlite:///webhooks.db')
Session = sessionmaker(bind=engine)

class ProcessedWebhook(Base):
    __tablename__ = 'processed_webhooks'
    transaction_id = Column(String, primary_key=True)
    processed_at = Column(DateTime, default=datetime.utcnow)

class FailedWebhook(Base):
    __tablename__ = 'failed_webhooks'
    id = Column(Integer, primary_key=True)
    transaction_id = Column(String)
    payload = Column(String)
    error = Column(String)
    retry_count = Column(Integer, default=0)
    created_at = Column(DateTime, default=datetime.utcnow)

Base.metadata.create_all(engine)

def validate_signature(payload: Dict[str, Any], signature: str, private_key: str) -> bool:
    """Validate webhook signature"""
    hmac_obj = hmac.new(
        private_key.encode('utf-8'),
        json.dumps(payload).encode('utf-8'),
        hashlib.sha256
    )
    expected_signature = hmac_obj.hexdigest()
    return hmac.compare_digest(signature, expected_signature)

def process_webhook_data(payload: Dict[str, Any]) -> None:
    """Process webhook business logic"""
    # Extract data
    transaction_id = payload['transaction_id']
    reference = payload['reference']
    meta_data = payload.get('meta_data', {})
    status = payload['status']
    
    # Update order status
    update_order_status(reference, status, transaction_id)
    
    # Process meta data
    if meta_data.get('order_id'):
        fulfill_order(meta_data['order_id'])
    
    # Send confirmation
    send_payment_confirmation(
        email=payload['buyer_email'],
        name=payload['buyer_name'],
        amount=payload['amount'],
        reference=reference
    )

@app.route('/webhook', methods=['POST'])
def handle_webhook():
    try:
        # Get request data
        signature = request.headers.get('opm-signature')
        payload = request.json
        private_key = os.getenv('OPM_PRIVATE_KEY')

        # Validate signature
        if not validate_signature(payload, signature, private_key):
            app.logger.error(f"Invalid signature for transaction: {payload.get('transaction_id')}")
            return 'Invalid signature', 400

        # Check for duplicate webhook
        session = Session()
        existing = session.query(ProcessedWebhook).filter_by(
            transaction_id=payload['transaction_id']
        ).first()

        if existing:
            app.logger.info(f"Duplicate webhook received: {payload['transaction_id']}")
            return 'Webhook already processed', 200

        # Process webhook asynchronously
        from threading import Thread
        Thread(target=async_process_webhook, args=(payload,)).start()

        return 'Webhook received', 200

    except Exception as e:
        app.logger.error(f"Webhook processing error: {str(e)}")
        # Store failed webhook
        store_failed_webhook(payload, str(e))
        return 'Webhook received with errors', 200

def async_process_webhook(payload: Dict[str, Any]) -> None:
    """Process webhook asynchronously"""
    try:
        session = Session()
        
        # Process the webhook
        process_webhook_data(payload)
        
        # Mark as processed
        session.add(ProcessedWebhook(transaction_id=payload['transaction_id']))
        session.commit()
        
    except Exception as e:
        app.logger.error(f"Async processing error: {str(e)}")
        store_failed_webhook(payload, str(e))
    finally:
        session.close()

def store_failed_webhook(payload: Dict[str, Any], error: str) -> None:
    """Store failed webhook for retry"""
    session = Session()
    try:
        failed_webhook = FailedWebhook(
            transaction_id=payload['transaction_id'],
            payload=json.dumps(payload),
            error=error
        )
        session.add(failed_webhook)
        session.commit()
    finally:
        session.close()

if __name__ == '__main__':
    app.run(port=3000)

<?php

require_once 'vendor/autoload.php';

use Monolog\Logger;
use Monolog\Handler\StreamHandler;

class WebhookHandler {
    private $db;
    private $logger;
    private $privateKey;

    public function __construct() {
        // Initialize database connection
        $this->db = new PDO(
            "mysql:host=" . getenv('DB_HOST') . ";dbname=" . getenv('DB_NAME'),
            getenv('DB_USER'),
            getenv('DB_PASS')
        );
        
        // Initialize logger
        $this->logger = new Logger('webhooks');
        $this->logger->pushHandler(new StreamHandler('logs/webhooks.log', Logger::DEBUG));
        
        // Get private key
        $this->privateKey = getenv('OPM_PRIVATE_KEY');
    }

    public function handleWebhook() {
        try {
            // Get request data
            $payload = file_get_contents('php://input');
            $signature = $_SERVER['HTTP_OPM_SIGNATURE'] ?? null;
            $data = json_decode($payload, true);

            // Validate request
            if (!$this->validateRequest($payload, $signature, $data)) {
                return;
            }

            // Check for duplicate
            if ($this->isDuplicate($data['transaction_id'])) {
                http_response_code(200);
                echo 'Webhook already processed';
                return;
            }

            // Send immediate response
            http_response_code(200);
            echo 'Webhook received';
            flush();

            // Process webhook
            $this->processWebhook($data);

            // Mark as processed
            $this->markAsProcessed($data['transaction_id']);

        } catch (Exception $e) {
            $this->logger->error('Webhook processing error: ' . $e->getMessage());
            $this->storeFailedWebhook($data ?? [], $e->getMessage());
            
            if (!headers_sent()) {
                http_response_code(200);
                echo 'Webhook received with errors';
            }
        }
    }

    private function validateRequest($payload, $signature, $data) {
        if (!$signature) {
            http_response_code(400);
            echo 'Missing signature';
            return false;
        }

        if (!$data) {
            http_response_code(400);
            echo 'Invalid JSON payload';
            return false;
        }

        if (!$this->validateSignature($payload, $signature)) {
            http_response_code(400);
            echo 'Invalid signature';
            return false;
        }

        return true;
    }

    private function validateSignature($payload, $signature) {
        $expectedSignature = hash_hmac('sha256', $payload, $this->privateKey);
        return hash_equals($signature, $expectedSignature);
    }

    private function isDuplicate($transactionId) {
        $stmt = $this->db->prepare(
            "SELECT id FROM processed_webhooks WHERE transaction_id = ?"
        );
        $stmt->execute([$transactionId]);
        return $stmt->fetch() !== false;
    }

    private function processWebhook($data) {
        // Extract data
        $transactionId = $data['transaction_id'];
        $reference = $data['reference'];
        $metaData = $data['meta_data'] ?? [];
        $status = $data['status'];

        // Update order status
        $this->updateOrderStatus($reference, $status, $transactionId);

        // Process meta data
        if (!empty($metaData['order_id'])) {
            $this->fulfillOrder($metaData['order_id']);
        }

        // Send confirmation
        $this->sendPaymentConfirmation(
            $data['buyer_email'],
            $data['buyer_name'],
            $data['amount'],
            $reference
        );
    }

    private function markAsProcessed($transactionId) {
        $stmt = $this->db->prepare(
            "INSERT INTO processed_webhooks (transaction_id, processed_at) 
             VALUES (?, NOW())"
        );
        $stmt->execute([$transactionId]);
    }

    private function storeFailedWebhook($data, $error) {
        $stmt = $this->db->prepare(
            "INSERT INTO failed_webhooks 
             (transaction_id, payload, error, created_at, retry_count) 
             VALUES (?, ?, ?, NOW(), 0)"
        );
        $stmt->execute([
            $data['transaction_id'] ?? null,
            json_encode($data),
            $error
        ]);
    }

    private function updateOrderStatus($reference, $status, $transactionId) {
        $stmt = $this->db->prepare(
            "UPDATE orders 
             SET payment_status = ?, 
                 payment_confirmed = 1, 
                 transaction_id = ?,
                 updated_at = NOW()
             WHERE reference = ?"
        );
        $stmt->execute([$status, $transactionId, $reference]);
    }

    private function fulfillOrder($orderId) {
        // Implement order fulfillment logic
        $this->logger->info('Fulfilling order: ' . $orderId);
    }

    private function sendPaymentConfirmation($email, $name, $amount, $reference) {
        // Implement email sending logic
        $this->logger->info('Sending payment confirmation to: ' . $email);
    }
}

// Handle webhook
$handler = new WebhookHandler();
$handler->handleWebhook();

Best Practices for Webhooks

Signature Validation

Always validate the webhook signature using your private key to ensure the request is from Open Market.

Idempotency

Implement idempotency checks using the transaction_id to avoid processing the same webhook multiple times.

Error Handling

Implement proper error handling and logging for webhook processing failures.

Quick Response

Return a 200 status code quickly and process the webhook asynchronously if needed.

Retry Mechanism

Implement a retry mechanism in case your server fails to process the webhook.

Webhook Use Cases

Basic Usage - For Beginners

Simple payment confirmation
Update order status
Send confirmation emails
Basic logging of transactions

Intermediate Integration

Order fulfillment automation
Inventory management
Customer notification system
Basic error handling and retries
Simple database storage

Advanced Implementation - For Pros

Distributed systems integration
Queue-based processing
Advanced error handling with retry mechanisms
Load balancing and scaling
Monitoring and alerting systems
Data analytics and reporting
Multi-region deployment

Enterprise Solutions

Microservices architecture
Event-driven systems
High availability setup
Disaster recovery
Compliance and audit logging
Advanced security measures
Performance optimization

Testing Webhooks

During development, you can use tools like ngrok to test webhooks locally:

ngrok http 3000

Then update your webhook URL in the API Settings with the ngrok URL.

Remember to update your webhook URL to your production URL before going live.

​Webhook Integration

​Webhook Authentication

​Webhook Payload

​Handling Webhooks

​Best Practices for Webhooks

​Webhook Use Cases

​Testing Webhooks

Webhook Integration

Webhook Authentication

Webhook Payload

Handling Webhooks

Best Practices for Webhooks

Webhook Use Cases

Testing Webhooks