Webhooks Deep Dive

const processedEvents = new Set(); // In production, use Redis/database
 
async function processWebhook(event) {
  const eventId = event.id;
  
  // Check if already processed
  if (processedEvents.has(eventId)) {
    console.log(`Event ${eventId} already processed, skipping`);
    return;
  }
  
  try {
    // Process event
    await handleEvent(event);
    
    // Mark as processed
    processedEvents.add(eventId);
    
    // In production, persist to database
    await db.saveProcessedEvent(eventId, new Date());
    
  } catch (error) {
    console.error(`Error processing event ${eventId}:`, error);
    // Don't mark as processed - allow retry
    throw error;
  }
}

policy.{resourceType}.executed

{
  "id": "evt_abc123xyz",
  "type": "order.completed",
  "version": "v1",
  "occurred_at": "2025-11-13T15:30:00Z",
  "source": "koywe.api",
  "environment": "production",
  "organization_id": "org_xyz789",
  "merchant_id": "mrc_abc123",
  "data": {
    "orderId": "ord_123456",
    "type": "PAYIN",
    "status": "COMPLETED",
    "amountIn": 50000,
    "originCurrencySymbol": "COP",
    "destinationCurrencySymbol": "COP",
    "externalId": "order-12345",
    "contactId": "cnt_customer1",
    "dates": {
      "confirmationDate": "2025-11-13T15:29:00Z",
      "paymentDate": "2025-11-13T15:29:30Z",
      "deliveryDate": "2025-11-13T15:30:00Z"
    }
  }
}

async function handleEvent(event) {
  const { type, data } = event;
 
  switch (type) {
    case 'order.created':
      await onOrderCreated(data);
      break;
 
    case 'order.approved':
      await onOrderApproved(data);
      break;
 
    case 'order.processing':
      await onOrderProcessing(data);
      break;
 
    case 'order.paid':
      await onOrderPaid(data);
      break;
 
    case 'order.completed':
      await onOrderCompleted(data);
      break;
 
    case 'order.failed':
      await onOrderFailed(data);
      break;
 
    case 'order.expired':
      await onOrderExpired(data);
      break;
 
    case 'order.canceled':
      await onOrderCanceled(data);
      break;
 
    case 'order.refunded':
      await onOrderRefunded(data);
      break;
 
    case 'order.updated':
      await onOrderUpdated(data);
      break;
 
    default:
      // Other families (merchant.*, merchant.kyb.*, invitation.*,
      // policy.approval.*, policy.*.executed, bank_income.*, webhook.ping)
      // are handled in their own routers.
      console.warn(`Unhandled event type: ${type}`);
  }
}
 
async function onOrderCompleted(data) {
  // CRITICAL: This is where you fulfill orders
  
  if (data.type === 'PAYIN') {
    // Customer paid, fulfill order
    await fulfillOrder(data.externalId);
    await sendConfirmationEmail(data.externalId);
    await updateInventory(data.externalId);
  } else if (data.type === 'PAYOUT') {
    // Payout completed, mark as paid
    await markInvoiceAsPaid(data.externalId);
    await notifyProvider(data.externalId);
  }
}
 
async function onOrderFailed(data) {
  // Handle failure
  await markOrderFailed(data.externalId);
  await sendFailureNotification(data.externalId);
  
  // Log for investigation
  console.error('Order failed:', {
    orderId: data.orderId,
    externalId: data.externalId,
    type: data.type
  });
}

app.post('/webhooks/koywe', express.raw({ type: 'application/json' }), async (req, res) => {
  // Convert raw body to string once
  const rawBody = req.body.toString();
  
  // 1. Verify signature (fast)
  if (!verifySignature(rawBody, req.headers['koywe-signature'])) {
    return res.status(401).send('Invalid signature');
  }
  
  // Parse JSON from string
  const event = JSON.parse(rawBody);
  
  // 2. Check idempotency (fast)
  if (await isEventProcessed(event.id)) {
    return res.status(200).send('Already processed');
  }
  
  // 3. Respond immediately
  res.status(200).send('OK');
  
  // 4. Process asynchronously (AFTER responding)
  setImmediate(async () => {
    try {
      await processWebhook(event);
      await markEventProcessed(event.id);
    } catch (error) {
      console.error('Webhook processing error:', error);
      await logWebhookError(event, error);
    }
  });
});

async function processWebhook(event) {
  const eventId = event.id;
  
  // Use database for persistence (not in-memory)
  const alreadyProcessed = await db.isEventProcessed(eventId);
  
  if (alreadyProcessed) {
    console.log(`Event ${eventId} already processed`);
    return; // Return success without re-processing
  }
  
  // Process event
  await handleEvent(event);
  
  // Mark as processed AFTER successful handling
  await db.markEventProcessed(eventId, {
    processedAt: new Date(),
    eventType: event.type,
    orderId: event.data.orderId
  });
}

async function getWebhookDeliveries(token, orgId, orderId) {
  const response = await axios.get(
    `https://api.koywe.com/api/v1/organizations/${orgId}/webhook-events`,
    {
      params: { orderId },
      headers: { 'Authorization': `Bearer ${token}` }
    }
  );
  
  return response.data;
}
 
// Usage
const deliveries = await getWebhookDeliveries(token, orgId, 'ord_123');
deliveries.forEach(d => {
  console.log(`Attempt ${d.attempt}: ${d.status} (${d.responseCode})`);
  console.log(`  Delivered at: ${d.deliveredAt}`);
  console.log(`  Response time: ${d.responseTime}ms`);
});

async function replayWebhook(token, orgId, eventId) {
  const response = await axios.post(
    `https://api.koywe.com/api/v1/organizations/${orgId}/webhook-events/{eventId}/replay`,
    {},
    {
      headers: { 'Authorization': `Bearer ${token}` }
    }
  );
  
  console.log('Webhook replayed');
}