Webhook Examples

import express from 'express';
import crypto from 'crypto';

const app = express();
app.use(express.json());

const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET;

// Verify webhook authenticity
function verifyWebhookSignature(payload, signature) {
  const computedSignature = crypto
    .createHmac('sha256', WEBHOOK_SECRET)
    .update(JSON.stringify(payload))
    .digest('hex');
  return computedSignature === signature;
}

app.post('/webhooks/sns-auctions', (req, res) => {
  const signature = req.headers['x-webhook-signature'];
  const eventId = req.headers['x-webhook-id'];
  const eventType = req.headers['x-webhook-event'];
  const timestamp = req.headers['x-webhook-timestamp'];
  
  const payload = req.body;

  // Verify signature
  if (!verifyWebhookSignature(payload, signature)) {
    console.warn('Invalid webhook signature');
    return res.status(401).json({ error: 'Signature verification failed' });
  }

  // Process event
  console.log(`Received ${eventType} event (ID: ${eventId})`);

  switch (eventType) {
    case 'auction.bid':
      console.log(`New bid on ${payload.domain}:`, payload.data);
      // Your logic here
      break;
      
    case 'auction.settled':
      console.log(`Auction settled: ${payload.auctionId}`);
      // Your logic here
      break;
      
    case 'auction.created':
      console.log(`New auction: ${payload.domain}`);
      // Your logic here
      break;

    case 'auction.bid.extended':
      console.log(`Snipe protection triggered on ${payload.domain}`);
      // Your logic here
      break;

    default:
      console.log(`Ignoring unknown event: ${eventType}`);
  }

  // Return 200 to acknowledge receipt (important!)
  res.status(200).json({ received: true });
});

app.listen(3000, () => {
  console.log('Webhook receiver listening on port 3000');
});

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

app = Flask(__name__)
WEBHOOK_SECRET = os.environ['WEBHOOK_SECRET']

def verify_signature(payload, signature):
    computed = hmac.new(
        WEBHOOK_SECRET.encode(),
        json.dumps(payload).encode(),
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(computed, signature)

@app.route('/webhooks/sns-auctions', methods=['POST'])
def handle_webhook():
    signature = request.headers.get('X-Webhook-Signature')
    event_type = request.headers.get('X-Webhook-Event')
    event_id = request.headers.get('X-Webhook-ID')
    
    payload = request.get_json()

    if not verify_signature(payload, signature):
        print('Invalid webhook signature')
        return {'error': 'Signature verification failed'}, 401

    print(f'Received {event_type} event (ID: {event_id})')

    if event_type == 'auction.bid':
        domain = payload.get('domain')
        bidder = payload.get('data', {}).get('bidder')
        amount = payload.get('data', {}).get('bidAmount')
        print(f'New bid on {domain}: {amount} SOL from {bidder}')
        
    elif event_type == 'auction.settled':
        auction_id = payload.get('auctionId')
        print(f'Auction settled: {auction_id}')
        
    elif event_type == 'auction.created':
        domain = payload.get('domain')
        print(f'New auction: {domain}')

    # Acknowledge receipt
    return {'received': True}, 200

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