feat: Resilient Webhook Ingestion System for Frontend External Triggers

backend/.env.example

@@ -61,6 +61,7 @@ STELLAR_ISSUER_SECRET_KEY=""
 # ------------------------------------------
 # Default certificate validity period in days
 CERTIFICATE_VALIDITY_DAYS=365
+WEBHOOK_SECRET="your-webhook-secret-change-in-production"
 
 # IPFS/Storage Configuration (optional - for certificate metadata)
 # IPFS_API_KEY=""

backend/src/routes/index.ts

@@ -11,6 +11,7 @@ import learningRoutes from './learning/learning.routes.js';
 import studentsRouter from './students.js';
 import blockchainRouter from '../blockchain/balance.js';
 import auditRouter from './audit.js';
+import webhookRouter from './webhooks.js';
 
 const router = Router();
 
@@ -30,4 +31,6 @@ router.use('/audit', auditRouter);
 // Blockchain routes
 router.use('/blockchain', blockchainRouter);
 
+router.use('/webhooks', webhookRouter);
+
 export default router;

backend/src/routes/webhooks.ts

@@ -0,0 +1,28 @@
+import { Request, Response, Router } from 'express';
+import { verifySignature } from '../utils/signature.js';
+import { enqueueWebhook } from '../services/queue.service.js';
+import logger from '../utils/logger.js';
+
+const router = Router();
+const WEBHOOK_SECRET = process.env.WEBHOOK_SECRET || 'webhook-secret';
+
+router.post('/ingest', async (req: Request, res: Response) => {
+  const signature = req.headers['x-webhook-signature'] as string;
+  const payload = JSON.stringify(req.body);
+
+  if (!signature || !verifySignature(payload, signature, WEBHOOK_SECRET)) {
+    logger.warn('Invalid webhook signature');
+    return res.status(401).json({ error: 'Invalid signature' });
+  }
+
+  try {
+    // Immediately enqueue and return 200 OK
+    await enqueueWebhook(req.body);
+    res.status(200).json({ status: 'accepted' });
+  } catch (error) {
+    logger.error('Failed to enqueue webhook:', error);
+    res.status(500).json({ error: 'Internal server error' });
+  }
+});
+
+export default router;

backend/src/services/queue.service.ts

@@ -0,0 +1,30 @@
+import redis from '../utils/redis.js';
+import logger from '../utils/logger.js';
+
+const WEBHOOK_QUEUE = 'webhooks:queue';
+const WEBHOOK_DLQ = 'webhooks:dlq';
+
+export const enqueueWebhook = async (payload: any): Promise<void> => {
+  await redis.lpush(WEBHOOK_QUEUE, JSON.stringify({
+    ...payload,
+    enqueuedAt: Date.now(),
+    retries: 0
+  }));
+};
+
+export const dequeueWebhook = async (): Promise<any | null> => {
+  const data = await redis.brpop(WEBHOOK_QUEUE, 0); // Block until data is available
+  if (data) {
+    return JSON.parse(data[1]);
+  }
+  return null;
+};
+
+export const enqueueDLQ = async (payload: any, error: string): Promise<void> => {
+  await redis.lpush(WEBHOOK_DLQ, JSON.stringify({
+    ...payload,
+    failedAt: Date.now(),
+    error
+  }));
+  logger.error(`Webhook moved to DLQ: ${error}`);
+};

backend/src/services/webhookWorker.ts

@@ -0,0 +1,46 @@
+import { dequeueWebhook, enqueueWebhook, enqueueDLQ } from './queue.service.js';
+import logger from '../utils/logger.js';
+
+const MAX_RETRIES = 5;
+
+const processWebhookLogic = async (payload: any) => {
+  // Implement actual business logic here
+  // e.g., update frontend state, database, etc.
+  logger.info(`Processing webhook: ${JSON.stringify(payload)}`);
+
+  // Simulate some logic
+  if (payload.shouldFail) {
+    throw new Error('Simulated processing failure');
+  }
+};
+
+export const startWorker = async () => {
+  logger.info('Webhook worker started');
+
+  while (true) {
+    try {
+      const webhook = await dequeueWebhook();
+      if (!webhook) continue;
+
+      try {
+        await processWebhookLogic(webhook);
+      } catch (error) {
+        if (webhook.retries < MAX_RETRIES) {
+          webhook.retries += 1;
+          const backoff = Math.pow(2, webhook.retries) * 1000;
+          logger.warn(`Retrying webhook in ${backoff}ms (Attempt ${webhook.retries})`);
+
+          setTimeout(async () => {
+            await enqueueWebhook(webhook);
+          }, backoff);
+        } else {
+          await enqueueDLQ(webhook, error instanceof Error ? error.message : 'Unknown error');
+        }
+      }
+    } catch (error) {
+      logger.error('Worker loop error:', error);
+      // Wait a bit before continuing to avoid tight loop on error
+      await new Promise(resolve => setTimeout(resolve, 5000));
+    }
+  }
+};

backend/src/utils/redis.ts

@@ -0,0 +1,22 @@
+import { Redis } from 'ioredis';
+import logger from './logger.js';
+
+const redisUrl = process.env.REDIS_URL || 'redis://localhost:6379';
+
+const redis = new Redis(redisUrl, {
+  maxRetriesPerRequest: 3,
+  retryStrategy: (times) => {
+    const delay = Math.min(times * 50, 2000);
+    return delay;
+  },
+});
+
+redis.on('error', (err) => {
+  logger.error('Redis error:', err);
+});
+
+redis.on('connect', () => {
+  logger.info('Connected to Redis');
+});
+
+export default redis;

backend/src/utils/signature.ts

@@ -0,0 +1,19 @@
+import crypto from 'crypto';
+
+export const verifySignature = (
+  payload: string,
+  signature: string,
+  secret: string
+): boolean => {
+  const hmac = crypto.createHmac('sha256', secret);
+  const digest = hmac.update(payload).digest('hex');
+
+  try {
+    return crypto.timingSafeEqual(
+      Buffer.from(signature, 'hex'),
+      Buffer.from(digest, 'hex')
+    );
+  } catch (error) {
+    return false;
+  }
+};