Webhooks

curl -X POST https://api.sourcesync.ai/v1/webhooks \
  -H "Authorization: Bearer YOUR_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "url": "https://your-domain.com/webhooks",
    "name": "My Webhook"
  }'

{
  "success": true,
  "message": "Created the webhook successfully",
  "data": {
    "webhook": {
      "id": "webhook-123",
      "name": "My Webhook",
      "url": "https://your-domain.com/webhooks",
      "signingKey": "your-signing-key",
      "status": "ACTIVE",
      "createdAt": "2024-03-14T12:34:56.789Z",
      "updatedAt": "2024-03-14T12:34:56.789Z"
    }
  }
}

app.post('/v1/webhooks', express.json(), async (req, res) => {
  try {
    // 1. Extract headers and verify timestamp
    const signature = req.headers['x-sourcesync-signature']
    const timestamp = req.headers['x-sourcesync-timestamp']

    if (!signature || !timestamp) {
      return res.status(400).json({ error: 'Missing required headers' })
    }

    // Check timestamp freshness (prevent replay attacks)
    const eventTime = new Date(timestamp).getTime()
    const currentTime = Date.now()
    if (currentTime - eventTime > 5 * 60 * 1000) {
      // 5 minutes
      return res.status(400).json({ error: 'Webhook expired' })
    }

    // 2. Verify signature
    const isValid = await verifyWebhookSignature(
      req.body,
      signature,
      timestamp,
      process.env.WEBHOOK_SIGNING_KEY,
    )

    if (!isValid) {
      return res.status(401).json({ error: 'Invalid signature' })
    }

    // 3. Respond immediately - this is crucial!
    res.status(200).json({ received: true })

    // 4. Process webhook asynchronously (after responding)
    const { event, data, requestId } = req.body
    processWebhookAsync(requestId, event, data)
  } catch (error) {
    console.error('Webhook error:', error)
    return res.status(500).json({ error: 'Internal server error' })
  }
})

async function processWebhookAsync(
  requestId: string,
  event: string,
  data: any,
) {
  // Check if we've already processed this requestId
  if (await hasProcessedWebhook(requestId)) {
    console.log(`Ignoring duplicate webhook: ${requestId}`)
    return
  }

  // Process the webhook based on event type
  switch (event) {
    case 'DOCUMENTS_READY':
      await handleDocumentsReady(data.documentIds)
      break
    // Handle other events...
  }

  // Mark this requestId as processed
  await markWebhookAsProcessed(requestId)
}

{
  "requestId": "unique-request-id",
  "event": "DOCUMENTS_READY",  // One of the document events
  "data": {
    "documentIds": ["doc-1", "doc-2"]  // Array of affected document IDs
  },
  "tenantId": "tenant-123",  // or null
  "namespaceId": "namespace-456",
  "organizationId": "org-789",
  "timestamp": "2024-03-14T12:34:56.789Z"
}

{
  "requestId": "unique-request-id",
  "event": "INGEST_JOB_RUN_COMPLETED",  // One of the ingestion job events
  "data": {
    "ingestJobRunId": "job-123"  // ID of the ingestion job run
  },
  "tenantId": "tenant-123",  // or null
  "namespaceId": "namespace-456",
  "organizationId": "org-789",
  "timestamp": "2024-03-14T12:34:56.789Z"
}

// TypeScript implementation
async function verifyWebhookSignature(
  payload: any,
  signature: string | null,
  timestamp: string | null,
  signingKey: string,
): Promise<boolean> {
  try {
    if (!signature || !timestamp) {
      return false
    }

    // Check if the timestamp is too old
    const eventTimestamp = new Date(timestamp).getTime()
    const currentTimestamp = Date.now()
    const timeDiff = Math.abs(currentTimestamp - eventTimestamp)
    const fiveMinutesInMs = 5 * 60 * 1000

    if (timeDiff > fiveMinutesInMs) {
      console.error('Webhook timestamp is too old')
      return false
    }

    // Recreate the signed string
    const payloadWithTimestamp = JSON.stringify({ payload, timestamp })

    // Convert message and key to ArrayBuffer
    const encoder = new TextEncoder()
    const messageBuffer = encoder.encode(payloadWithTimestamp)
    const keyBuffer = encoder.encode(signingKey)

    // Import the key
    const cryptoKey = await crypto.subtle.importKey(
      'raw',
      keyBuffer,
      { name: 'HMAC', hash: 'SHA-256' },
      false,
      ['verify'],
    )

    // Convert signature from hex to ArrayBuffer
    const signatureBuffer = hexToArrayBuffer(signature)

    // Verify the signature
    return await crypto.subtle.verify(
      'HMAC',
      cryptoKey,
      signatureBuffer,
      messageBuffer,
    )
  } catch (error) {
    console.error('Signature verification failed:', error)
    return false
  }
}

// Helper function to convert hex string to ArrayBuffer
function hexToArrayBuffer(hexString: string): ArrayBuffer {
  const matches = hexString.match(/.{1,2}/g) || []
  return new Uint8Array(matches.map((byte) => parseInt(byte, 16))).buffer
}

# Python implementation
import hmac
import hashlib
import json
from typing import Any, Dict

def verify_webhook_signature(
    payload: Dict[str, Any],
    signature: str | None,
    timestamp: str | None,
    signing_key: str
) -> bool:
    """Verify that the webhook signature is valid."""
    try:
        if not signature or not timestamp:
          return False

        event_time = datetime.fromtimestamp(timestamp, timezone.utc)
        current_time = datetime.now(timezone.utc)
        if (current_time - event_time).total_seconds() > 300:  # 5 minutes
          return False

        # Recreate the signed string
        payload_with_timestamp = json.dumps({
          "payload": payload,
          "timestamp": timestamp
        },separators=(',', ':'))  # Use compact JSON encoding

        # Create HMAC signature
        computed_signature = hmac.new(
            signing_key.encode('utf-8'),
            payload_with_timestamp.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()

        # Compare signatures using constant-time comparison
        return hmac.compare_digest(computed_signature, signature)
    except Exception as e:
        print(f"Signature verification failed: {e}")
        return False

import express from 'express'
const app = express()

app.post('/v1/webhooks', express.json(), async (req, res) => {
  try {
    // Get headers
    const signature = req.headers['x-sourcesync-signature']
    const timestamp = req.headers['x-sourcesync-timestamp']

    // Verify signature
    const isValid = await verifyWebhookSignature(
      req.body,
      signature,
      timestamp,
      process.env.WEBHOOK_SIGNING_KEY,
    )

    if (!isValid) {
      return res.status(401).json({ error: 'Invalid signature' })
    }

    // Process webhook asynchronously
    const { event, data } = req.body
    processWebhookAsync(event, data)

    // Respond immediately
    res.status(200).json({ received: true })
  } catch (error) {
    console.error('Webhook error:', error)
    res.status(500).json({ error: 'Internal server error' })
  }
})

async function processWebhookAsync(event: string, data: any) {
  switch (event) {
    case 'DOCUMENTS_READY':
      await handleDocumentsReady(data.documentIds)
      break
    case 'DOCUMENTS_ERROR':
      await handleDocumentsError(data.documentIds)
      break
    // Handle other events...
  }
}

from flask import Flask, request, jsonify
import os
import threading
from datetime import datetime, timezone
import json
import hmac
import hashlib

app = Flask(__name__)

@app.route('/webhooks', methods=['POST'])
def webhook_handler():
    try:
        # Get headers
        signature = request.headers.get('x-sourcesync-signature')
        timestamp = request.headers.get('x-sourcesync-timestamp')

        # Verify signature
        is_valid = verify_webhook_signature(
            request.json,
            signature,
            timestamp,
            os.environ.get('WEBHOOK_SIGNING_KEY')
        )

        if not is_valid:
            return jsonify({'error': 'Invalid signature'}), 401

        # Process webhook asynchronously (after responding)
        webhook_data = request.json
        event = webhook_data.get('event')
        data = webhook_data.get('data')

        # Start processing in a separate thread
        threading.Thread(
            target=process_webhook_async,
            args=(event, data)
        ).start()

        # Respond immediately
        return jsonify({'received': True}), 200

    except Exception as e:
        print(f"Webhook error: {e}")
        return jsonify({'error': 'Internal server error'}), 500

def verify_webhook_signature(payload, signature, timestamp, signing_key):
    try:
        if not signature or not timestamp:
            return False

        # Check timestamp freshness
        try:
            event_time = datetime.fromisoformat(timestamp.replace('Z', '+00:00'))
            current_time = datetime.now(timezone.utc)
            if (current_time - event_time).total_seconds() > 300:  # 5 minutes
                print('Webhook timestamp is too old')
                return False
        except ValueError:
            return False

        # Recreate the signed string
        payload_with_timestamp = json.dumps(
            {'payload': payload, 'timestamp': timestamp},
            separators=(',', ':')  # Use compact JSON encoding
        )

        # Create HMAC signature
        computed_signature = hmac.new(
            signing_key.encode('utf-8'),
            payload_with_timestamp.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()

        # Compare signatures using constant-time comparison
        return hmac.compare_digest(computed_signature, signature)
    except Exception as e:
        print(f"Signature verification failed: {e}")
        return False

def process_webhook_async(event, data):
    """Process webhook asynchronously to avoid blocking the response."""
    try:
        if event == 'DOCUMENTS_READY':
            handle_documents_ready(data.get('documentIds', []))
        elif event == 'DOCUMENTS_ERROR':
            handle_documents_error(data.get('documentIds', []))
        # Handle other events...
    except Exception as e:
        print(f"Error processing webhook: {e}")

def handle_documents_ready(document_ids):
    """Handle the DOCUMENTS_READY event."""
    # Your business logic here
    print(f"Processing {len(document_ids)} ready documents")

def handle_documents_error(document_ids):
    """Handle the DOCUMENTS_ERROR event."""
    # Your business logic here
    print(f"Processing {len(document_ids)} documents with errors")

if __name__ == '__main__':
    # Run the Flask app
    app.run(host='0.0.0.0', port=5000)