Webhooks

curl "https://api.sandbox.paygentic.io/v0/webhooks" \
  -H "Authorization: Bearer sk_test_YOUR_API_KEY"

const response = await fetch('https://api.sandbox.paygentic.io/v0/webhooks', {
  headers: { 'Authorization': 'Bearer sk_test_YOUR_API_KEY' },
});

const { enabled } = await response.json();

import requests

response = requests.get(
    "https://api.sandbox.paygentic.io/v0/webhooks",
    headers={"Authorization": "Bearer sk_test_YOUR_API_KEY"},
)
enabled = response.json()["enabled"]

curl -X PATCH "https://api.sandbox.paygentic.io/v0/webhooks" \
  -H "Authorization: Bearer sk_test_YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{ "enabled": false }'

const response = await fetch('https://api.sandbox.paygentic.io/v0/webhooks', {
  method: 'PATCH',
  headers: {
    'Authorization': 'Bearer sk_test_YOUR_API_KEY',
    'Content-Type': 'application/json',
  },
  body: JSON.stringify({ enabled: false }),
});

const webhooks = await response.json();

import requests

response = requests.patch(
    "https://api.sandbox.paygentic.io/v0/webhooks",
    headers={"Authorization": "Bearer sk_test_YOUR_API_KEY"},
    json={"enabled": False},
)
webhooks = response.json()

curl "https://api.sandbox.paygentic.io/v0/webhooks/portal" \
  -H "Authorization: Bearer sk_test_YOUR_API_KEY"

const response = await fetch('https://api.sandbox.paygentic.io/v0/webhooks/portal', {
  headers: { 'Authorization': 'Bearer sk_test_YOUR_API_KEY' },
});

const { url, expiresAt } = await response.json();

import requests

response = requests.get(
    "https://api.sandbox.paygentic.io/v0/webhooks/portal",
    headers={"Authorization": "Bearer sk_test_YOUR_API_KEY"},
)
portal = response.json()

npm install svix

pip install svix

go get github.com/svix/svix-webhooks/go

gem install svix

import { Webhook } from 'svix';

function verifyWebhook(body, headers, secret) {
  const wh = new Webhook(secret);
  
  try {
    // Verify the webhook - throws on error
    const payload = wh.verify(body, headers);
    return { verified: true, payload };
  } catch (err) {
    return { verified: false, error: err.message };
  }
}

// Important: Use the raw request body
// Many frameworks parse JSON automatically - you need the raw string
app.use('/webhooks', express.raw({ type: 'application/json' }));

from svix.webhooks import Webhook

def verify_webhook(body, headers, secret):
    wh = Webhook(secret)
    
    try:
        # Verify the webhook - throws on error
        payload = wh.verify(body, headers)
        return {"verified": True, "payload": payload}
    except Exception as e:
        return {"verified": False, "error": str(e)}

# Important: Use the raw request body
# In Flask: request.data
# In Django: request.body
# In FastAPI: await request.body()

// Register the webhook route before the CSRF middleware,
// so the CSRF check never runs for it
app.post('/webhooks/paygentic', express.raw({ type: 'application/json' }), handleWebhook);

app.use(csrfProtection);

from django.views.decorators.csrf import csrf_exempt

@csrf_exempt
def webhook_handler(request):
    # Handle webhook
    pass

import express from 'express';
import { Webhook } from 'svix';
import Redis from 'ioredis';

const app = express();
const redis = new Redis(process.env.REDIS_URL);
const webhookSecret = process.env.PAYGENTIC_WEBHOOK_SECRET;

// Webhook handler with all best practices
app.post('/webhooks/paygentic',
  express.raw({ type: 'application/json' }),
  async (req, res) => {
    const svixId = req.headers['svix-id'];
    const svixTimestamp = req.headers['svix-timestamp'];
    const svixSignature = req.headers['svix-signature'];

    // Check required headers
    if (!svixId || !svixTimestamp || !svixSignature) {
      return res.status(400).send('Missing required headers');
    }

    try {
      // Verify webhook signature
      const wh = new Webhook(webhookSecret);
      const payload = wh.verify(req.body, {
        'svix-id': svixId,
        'svix-timestamp': svixTimestamp,
        'svix-signature': svixSignature
      });

      // Check for duplicate processing (idempotency)
      const processed = await redis.get(`webhook:${svixId}`);
      if (processed) {
        console.log(`Webhook ${svixId} already processed`);
        return res.status(200).send('Already processed');
      }

      // Mark as processed (with 24 hour expiry)
      await redis.setex(`webhook:${svixId}`, 86400, 'processed');

      // Respond immediately
      res.status(200).send('OK');

      // Process asynchronously
      setImmediate(async () => {
        try {
          await processWebhookEvent(payload);
        } catch (error) {
          console.error('Error processing webhook:', error);
          // Send to error tracking service
          // Consider implementing a retry queue
        }
      });

    } catch (err) {
      console.error('Webhook verification failed:', err);
      return res.status(400).send('Invalid webhook');
    }
  }
);

async function processWebhookEvent(payload) {
  const { eventType, data } = payload;
  
  console.log(`Processing event: ${eventType}`);
  
  switch (eventType) {
    case 'customer.created.v0':
      await handleCustomerCreated(data);
      break;

    case 'customer.creation_failed.v0':
      await handleCustomerCreationFailed(data);
      break;

    case 'subscription.created.v0':
      await handleSubscriptionCreated(data);
      break;

    case 'subscription.activated.v0':
      await handleSubscriptionActivated(data);
      break;

    case 'subscription.cancelled.v0':
      await handleSubscriptionCancelled(data);
      break;

    case 'subscription.updated.v0':
      await handleSubscriptionUpdated(data);
      break;

    case 'payment.completed.v0':
      await handlePaymentCompleted(data);
      break;

    case 'payment.failed.v0':
      await handlePaymentFailed(data);
      break;

    case 'payment.expired.v0':
      await handlePaymentExpired(data);
      break;

    case 'source_event.pending.v0':
      await handleSourceEventPending(data);
      break;

    default:
      console.log(`Unhandled event type: ${eventType}`);
  }
}

// Event handlers
async function handleCustomerCreated(data) {
  console.log('New customer created:', data.customerId);
  // Your business logic here
}

async function handleSubscriptionCreated(data) {
  console.log('New subscription created:', data.subscriptionId);
  // Your business logic here
}

async function handleSubscriptionActivated(data) {
  console.log('Subscription activated:', data.subscriptionId);
  // Provision access, activate features, etc.
  // Your business logic here
}

async function handleSourceEventPending(data) {
  console.log('Source event requires manual processing:', data.sourceEventId);
  // Your business logic here
}

// ... implement other handlers

from flask import Flask, request, jsonify
import os
import json
import redis
from svix.webhooks import Webhook
from threading import Thread

app = Flask(__name__)
redis_client = redis.from_url(os.environ.get('REDIS_URL'))
webhook_secret = os.environ.get('PAYGENTIC_WEBHOOK_SECRET')

@app.route('/webhooks/paygentic', methods=['POST'])
def handle_webhook():
    svix_id = request.headers.get('svix-id')
    svix_timestamp = request.headers.get('svix-timestamp')
    svix_signature = request.headers.get('svix-signature')
    
    # Check required headers
    if not all([svix_id, svix_timestamp, svix_signature]):
        return 'Missing required headers', 400
    
    try:
        # Verify webhook signature
        wh = Webhook(webhook_secret)
        payload = wh.verify(request.data, {
            'svix-id': svix_id,
            'svix-timestamp': svix_timestamp,
            'svix-signature': svix_signature
        })
        
        # Check for duplicate processing (idempotency)
        if redis_client.get(f"webhook:{svix_id}"):
            print(f"Webhook {svix_id} already processed")
            return 'Already processed', 200
        
        # Mark as processed (with 24 hour expiry)
        redis_client.setex(f"webhook:{svix_id}", 86400, 'processed')
        
        # Process asynchronously
        thread = Thread(target=process_webhook_event, args=(payload,))
        thread.daemon = True
        thread.start()
        
        # Respond immediately
        return 'OK', 200
        
    except Exception as e:
        print(f"Webhook verification failed: {e}")
        return 'Invalid webhook', 400

def process_webhook_event(payload):
    event_type = payload.get('eventType')
    data = payload.get('data')
    
    print(f"Processing event: {event_type}")
    
    try:
        if event_type == 'customer.created.v0':
            handle_customer_created(data)
        elif event_type == 'customer.creation_failed.v0':
            handle_customer_creation_failed(data)
        elif event_type == 'subscription.created.v0':
            handle_subscription_created(data)
        elif event_type == 'subscription.activated.v0':
            handle_subscription_activated(data)
        elif event_type == 'subscription.cancelled.v0':
            handle_subscription_cancelled(data)
        elif event_type == 'payment.completed.v0':
            handle_payment_completed(data)
        elif event_type == 'payment.failed.v0':
            handle_payment_failed(data)
        elif event_type == 'payment.expired.v0':
            handle_payment_expired(data)
        elif event_type == 'source_event.pending.v0':
            handle_source_event_pending(data)
        else:
            print(f"Unhandled event type: {event_type}")
    except Exception as e:
        print(f"Error processing webhook: {e}")
        # Send to error tracking service
        # Consider implementing a retry queue

# Event handlers
def handle_customer_created(data):
    print(f"New customer created: {data.get('customerId')}")
    # Your business logic here

def handle_subscription_created(data):
    print(f"New subscription created: {data.get('subscriptionId')}")
    # Your business logic here

def handle_subscription_activated(data):
    print(f"Subscription activated: {data.get('subscriptionId')}")
    # Provision access, activate features, etc.
    # Your business logic here

def handle_source_event_pending(data):
    print(f"Source event requires manual processing: {data.get('sourceEventId')}")
    # Your business logic here

# ... implement other handlers

if __name__ == '__main__':
    app.run(debug=False)