feat: Configured Webhook Infrastructure for escrow_transactions EventTriggers

metadata/databases/safetrust/tables/escrow_transactions.yaml

@@ -57,4 +57,30 @@ table:
       using:
         foreign_key_constraint_on:
           column: escrow_transaction_id
-          table: escrow_api_calls
+          table: escrow_api_calls
+  event_triggers:
+    - name: escrow_transaction_webhook
+      definition:
+        enable_manual: true
+        insert:
+          columns: '*'
+        update:
+          columns: '*'
+        delete:
+          columns: '*'
+      retry_conf:
+        num_retries: 5
+        interval_sec: 15
+        timeout_sec: 60
+        exponential_backoff: true
+        tolerance_seconds: 21600  # 6 hours tolerance for event processing
+      webhook: '{{WEBHOOK_BASE_URL}}/webhook/escrow-transaction'
+      headers:
+        - name: Authorization
+          value_from_env: WEBHOOK_SECRET
+        - name: Content-Type
+          value: application/json
+        - name: X-Hasura-Event-Id
+          value_from_env: EVENT_ID
+        - name: X-Idempotency-Key
+          value_from_env: EVENT_ID

webhook-service/.gitignore

@@ -0,0 +1,69 @@
+# Node.js dependencies
+node_modules/
+
+# Logs
+logs/
+*.log
+npm-debug.log*
+yarn-debug.log*
+yarn-error.log*
+
+# Environment variables
+.env
+.env.local
+.env.development.local
+.env.test.local
+.env.production.local
+.env.test
+.env.production
+
+# Runtime data
+pids/
+*.pid
+*.seed
+*.pid.lock
+
+# Temporary files
+tmp/
+temp/
+*.tmp
+
+# Cache and build artifacts
+.cache/
+dist/
+build/
+
+# Editor-specific files
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+
+# OS-specific files
+.DS_Store
+Desktop.ini
+Thumbs.db
+
+# Debug files
+debug/
+
+# Coverage reports
+coverage/
+*.lcov
+*.lcov.info
+
+# Dependency directories
+bower_components/
+
+# Compressed files
+*.tgz
+*.lock
+
+# Test output
+test-output/
+*.tap
+
+# Karate reports
+target/
+tests/results/

webhook-service/config/logger.js

@@ -0,0 +1,118 @@
+/**
+ * Centralized logger configuration
+ * This file provides a consistent logger instance for the entire application
+ */
+
+const winston = require('winston');
+const fs = require('fs');
+const path = require('path');
+
+// Create logs directory if it doesn't exist
+const logsDir = path.join(__dirname, '..', 'logs');
+if (!fs.existsSync(logsDir)) {
+  fs.mkdirSync(logsDir, { recursive: true });
+}
+
+// Configure logger
+const logger = winston.createLogger({
+  level: process.env.LOG_LEVEL || 'info',
+  format: winston.format.combine(
+    winston.format.timestamp(),
+    winston.format.json()
+  ),
+  defaultMeta: { service: 'webhook-service' },
+  transports: [
+    // Use a more readable format for console output
+    new winston.transports.Console({
+      format: winston.format.combine(
+        winston.format.colorize(),
+        winston.format.timestamp({ format: 'YYYY-MM-DD HH:mm:ss' }),
+        winston.format.printf(({ level, message, timestamp, service, ...metadata }) => {
+          // Filter out empty objects and the service field
+          const filteredMetadata = Object.entries(metadata)
+            .filter(([_, value]) => {
+              // Keep only non-empty values
+              if (typeof value === 'object' && value !== null) {
+                return Object.keys(value).length > 0;
+              }
+              return value !== undefined && value !== null && value !== '';
+            })
+            .reduce((obj, [key, value]) => {
+              obj[key] = value;
+              return obj;
+            }, {});
+
+          // Only add metadata if there's something to show
+          const metaStr = Object.keys(filteredMetadata).length > 0 
+            ? `\n${JSON.stringify(filteredMetadata, null, 2)}` 
+            : '';
+
+          return `${timestamp} ${level}: ${message}${metaStr}`;
+        })
+      )
+    }),
+    new winston.transports.File({ filename: 'logs/error.log', level: 'error' }),
+    new winston.transports.File({ filename: 'logs/combined.log' })
+  ],
+});
+
+// Create a request logger middleware
+const requestLogger = (req, res, next) => {
+  const start = Date.now();
+
+  // Log request
+  logger.info('Request received', {
+    method: req.method,
+    path: req.path,
+    headers: {
+      'x-hasura-event-id': req.headers['x-hasura-event-id'],
+      'content-type': req.headers['content-type'],
+      'x-idempotency-key': req.headers['x-idempotency-key']
+    }
+  });
+
+  // Capture response
+  const originalSend = res.send;
+  res.send = function(body) {
+    res.responseBody = body;
+    return originalSend.call(this, body);
+  };
+
+  // Log response after request is complete
+  res.on('finish', () => {
+    const duration = Date.now() - start;
+    const level = res.statusCode >= 400 ? 'error' : 'info';
+
+    logger[level]('Response sent', {
+      method: req.method,
+      path: req.path,
+      statusCode: res.statusCode,
+      duration: `${duration}ms`,
+      eventId: req.headers['x-hasura-event-id'] || 'unknown'
+    });
+  });
+
+  next();
+};
+
+// Error handling middleware
+const errorLogger = (err, req, res, next) => {
+  logger.error('Unhandled error', {
+    error: err.message,
+    stack: err.stack,
+    path: req.path,
+    method: req.method,
+    eventId: req.headers['x-hasura-event-id'] || 'unknown'
+  });
+
+  res.status(500).json({
+    error: 'Internal server error',
+    message: err.message
+  });
+};
+
+module.exports = {
+  logger,
+  requestLogger,
+  errorLogger
+}; 

webhook-service/config/webhook.js

@@ -0,0 +1,59 @@
+/**
+ * Webhook configuration settings
+ * This file centralizes all webhook-related configuration
+ */
+
+module.exports = {
+  // Retry configuration
+  retry: {
+    maxAttempts: parseInt(process.env.WEBHOOK_MAX_ATTEMPTS || '5'),
+    initialDelay: parseInt(process.env.WEBHOOK_INITIAL_DELAY || '15000'), // 15 seconds
+    maxDelay: parseInt(process.env.WEBHOOK_MAX_DELAY || '3600000'), // 1 hour
+    backoffFactor: parseFloat(process.env.WEBHOOK_BACKOFF_FACTOR || '2.0'),
+    jitter: parseFloat(process.env.WEBHOOK_JITTER || '0.1') // 10% jitter
+  },
+
+  // Timeout settings
+  timeout: {
+    request: parseInt(process.env.WEBHOOK_REQUEST_TIMEOUT || '30000'), // 30 seconds
+    processing: parseInt(process.env.WEBHOOK_PROCESSING_TIMEOUT || '60000') // 60 seconds
+  },
+
+  // Authentication
+  auth: {
+    headerName: process.env.WEBHOOK_AUTH_HEADER || 'Authorization',
+    secretEnvVar: 'WEBHOOK_SECRET'
+  },
+
+  // Alerting thresholds
+  alerts: {
+    consecutiveFailures: parseInt(process.env.WEBHOOK_ALERT_FAILURES || '3'),
+    errorRateThreshold: parseFloat(process.env.WEBHOOK_ERROR_RATE_THRESHOLD || '0.1') // 10% error rate
+  },
+
+  // Logging
+  logging: {
+    level: process.env.WEBHOOK_LOG_LEVEL || 'info',
+    includeHeaders: process.env.WEBHOOK_LOG_HEADERS === 'true',
+    includeBody: process.env.WEBHOOK_LOG_BODY === 'true',
+    sensitiveFields: [
+      'password',
+      'token',
+      'secret',
+      'authorization',
+      'api_key'
+    ]
+  },
+
+  // Rate limiting
+  rateLimit: {
+    windowMs: parseInt(process.env.WEBHOOK_RATE_LIMIT_WINDOW || '60000'), // 1 minute
+    maxRequests: parseInt(process.env.WEBHOOK_RATE_LIMIT_MAX || '100') // 100 requests per minute
+  },
+
+  // Endpoints
+  endpoints: {
+    escrowTransaction: '/webhook/escrow-transaction',
+    health: '/webhook/health'
+  }
+}; 

webhook-service/migrations/2025122611447_create_webhook_processed_events.sql

@@ -0,0 +1,45 @@
+-- Create table for tracking processed webhook events
+CREATE TABLE IF NOT EXISTS webhook_processed_events (
+  id SERIAL PRIMARY KEY,
+  event_id TEXT NOT NULL,
+  status TEXT NOT NULL,
+  details JSONB,
+  processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+-- Create index for fast lookups by event_id
+CREATE UNIQUE INDEX IF NOT EXISTS webhook_processed_events_event_id_idx ON webhook_processed_events (event_id);
+
+-- Add column for webhook_status to escrow_transactions if it doesn't exist
+DO $$ 
+BEGIN
+  IF NOT EXISTS (
+    SELECT FROM information_schema.columns 
+    WHERE table_name = 'escrow_transactions' AND column_name = 'webhook_status'
+  ) THEN
+    ALTER TABLE escrow_transactions ADD COLUMN webhook_status TEXT;
+  END IF;
+END $$;
+
+-- Add column for webhook_attempts to escrow_transactions if it doesn't exist
+DO $$ 
+BEGIN
+  IF NOT EXISTS (
+    SELECT FROM information_schema.columns 
+    WHERE table_name = 'escrow_transactions' AND column_name = 'webhook_attempts'
+  ) THEN
+    ALTER TABLE escrow_transactions ADD COLUMN webhook_attempts INTEGER DEFAULT 0;
+  END IF;
+END $$;
+
+-- Add column for last_webhook_attempt to escrow_transactions if it doesn't exist
+DO $$ 
+BEGIN
+  IF NOT EXISTS (
+    SELECT FROM information_schema.columns 
+    WHERE table_name = 'escrow_transactions' AND column_name = 'last_webhook_attempt'
+  ) THEN
+    ALTER TABLE escrow_transactions ADD COLUMN last_webhook_attempt TIMESTAMPTZ;
+  END IF;
+END $$;