diff --git a/backend/.env.example b/backend/.env.example index 1081d944..5933c2b0 100644 --- a/backend/.env.example +++ b/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="" diff --git a/backend/src/routes/index.ts b/backend/src/routes/index.ts index a886c30a..c7f649b4 100644 --- a/backend/src/routes/index.ts +++ b/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; diff --git a/backend/src/routes/webhooks.ts b/backend/src/routes/webhooks.ts new file mode 100644 index 00000000..4f6bdcc9 --- /dev/null +++ b/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; diff --git a/backend/src/services/queue.service.ts b/backend/src/services/queue.service.ts new file mode 100644 index 00000000..8422444a --- /dev/null +++ b/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 => { + await redis.lpush(WEBHOOK_QUEUE, JSON.stringify({ + ...payload, + enqueuedAt: Date.now(), + retries: 0 + })); +}; + +export const dequeueWebhook = async (): Promise => { + 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 => { + await redis.lpush(WEBHOOK_DLQ, JSON.stringify({ + ...payload, + failedAt: Date.now(), + error + })); + logger.error(`Webhook moved to DLQ: ${error}`); +}; diff --git a/backend/src/services/webhookWorker.ts b/backend/src/services/webhookWorker.ts new file mode 100644 index 00000000..c7f3613b --- /dev/null +++ b/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)); + } + } +}; diff --git a/backend/src/utils/redis.ts b/backend/src/utils/redis.ts new file mode 100644 index 00000000..b83b6bae --- /dev/null +++ b/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; diff --git a/backend/src/utils/signature.ts b/backend/src/utils/signature.ts new file mode 100644 index 00000000..f51a4878 --- /dev/null +++ b/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; + } +};