diff --git a/backend/src/jobs/README.md b/backend/src/jobs/README.md new file mode 100644 index 00000000..ac8eab95 --- /dev/null +++ b/backend/src/jobs/README.md @@ -0,0 +1,42 @@ +# Background Jobs & Workers + +This directory contains the background processing architecture for the Stellar Tipz real-time off-chain backend. We use [BullMQ](https://docs.bullmq.io/) backed by **Redis** to reliably manage job queues, schedules, and retries. + +## Directory Structure +- `index.ts`: The central export point for all configured queues and workers. +- `webhookDelivery.ts`: Manages the `webhook-delivery` queue, which safely dispatches HTTP POST webhooks to external clients with automatic HMAC signing, timeout handling, and exponential backoff on transient failures. + +## 1. Queues +Queues are responsible for holding jobs until they are processed. They are initialized utilizing the shared Redis connection located in `src/db/redis.ts`. + +### Best Practices for Queues: +- **Idempotency:** Ensure that the data payload submitted to a queue is deterministic. Do not pass complex class instances; instead, pass scalar IDs and pure JSON objects. +- **Backoff & Retries:** Configure queues with standard failure handling. E.g., exponential backoff (`delay: 2000`, `attempts: 5`). + +## 2. Workers +Workers actively listen to Queues and process jobs as they arrive. +In a production environment, you may scale workers independently of the main API server to increase throughput. + +### How to Run Workers Locally +For local development, workers are instantiated directly in the application runtime via `src/jobs/index.ts` alongside the Express API server. + +When you start the local dev server, the workers will automatically begin processing: +```bash +npm run dev +``` +*(Make sure your local Redis instance is running via `docker compose -f backend/docker-compose.yml up -d`)* + +### Error Handling +Workers should **throw** an Error (`throw new Error(...)`) whenever a job fails due to an external factor (e.g., a non-2xx HTTP status from a webhook). Throwing an error natively leverages BullMQ's automatic retry logic. +Listen for the `failed` event on your worker to log issues via the shared `logger`. + +## 3. Schedules (Cron Jobs) +Scheduled or recurring tasks (e.g., daily cleanup, stale tip sweeps) can be implemented using BullMQ's [Repeatable Jobs](https://docs.bullmq.io/guide/jobs/repeatable). +To schedule a recurring job, use the `repeat` option when adding it to the queue: +```typescript +await myQueue.add( + 'daily-cleanup', + { }, + { repeat: { pattern: '0 0 * * *' } } // Every midnight +); +``` diff --git a/backend/src/jobs/webhookDelivery.ts b/backend/src/jobs/webhookDelivery.ts new file mode 100644 index 00000000..a533a14d --- /dev/null +++ b/backend/src/jobs/webhookDelivery.ts @@ -0,0 +1,106 @@ +import { Queue, Worker, Job } from 'bullmq'; +import { redis } from '../db/redis.js'; +import { logger } from '../common/utils/logger.js'; +import crypto from 'node:crypto'; + +export const WEBHOOK_DELIVERY_QUEUE = 'webhook-delivery'; + +export interface WebhookDeliveryPayload { + url: string; + payload: Record; + secret?: string; // Optional secret to sign the payload (HMAC) +} + +/** + * Queue instance for dispatching webhook deliveries. + */ +export const webhookDeliveryQueue = new Queue(WEBHOOK_DELIVERY_QUEUE, { + connection: redis, + defaultJobOptions: { + attempts: 5, + backoff: { + type: 'exponential', + delay: 2000, + }, + removeOnComplete: { + age: 3600, // keep completed jobs for 1 hour + count: 1000, // keep at most 1000 completed jobs + }, + removeOnFail: { + age: 24 * 3600, // keep failed jobs for 24 hours + }, + }, +}); + +/** + * Helper to schedule a webhook delivery. + */ +export async function scheduleWebhookDelivery( + url: string, + payload: Record, + secret?: string +): Promise { + await webhookDeliveryQueue.add('deliver', { url, payload, secret }); +} + +/** + * Worker to process webhook deliveries. + */ +export const webhookDeliveryWorker = new Worker( + WEBHOOK_DELIVERY_QUEUE, + async (job: Job) => { + const { url, payload, secret } = job.data; + + logger.info({ jobId: job.id, url }, 'Starting webhook delivery'); + + const headers: Record = { + 'Content-Type': 'application/json', + 'User-Agent': 'Stellar-Tipz-Webhook-Bot/1.0', + }; + + const body = JSON.stringify(payload); + + // If a secret is provided, sign the payload with HMAC SHA256 + if (secret) { + const hmac = crypto.createHmac('sha256', secret); + hmac.update(body); + headers['X-Signature'] = `sha256=${hmac.digest('hex')}`; + } + + try { + const controller = new AbortController(); + const timeout = setTimeout(() => controller.abort(), 10_000); // 10 second timeout + + const response = await fetch(url, { + method: 'POST', + headers, + body, + signal: controller.signal, + }); + + clearTimeout(timeout); + + if (!response.ok) { + // Throwing an error automatically triggers BullMQ backoff retry + throw new Error(`HTTP Error: ${response.status} ${response.statusText}`); + } + + logger.info({ jobId: job.id, url, status: response.status }, 'Webhook delivered successfully'); + } catch (error: unknown) { + logger.warn({ jobId: job.id, url, error: error instanceof Error ? error.message : error }, 'Webhook delivery failed'); + throw error; + } + }, + { + connection: redis, + concurrency: 5, // Process up to 5 webhooks concurrently + } +); + +webhookDeliveryWorker.on('failed', (job: Job | undefined, err: Error) => { + if (job) { + logger.error({ jobId: job.id, url: job.data.url, err: err.message }, 'Webhook job failed permanently or retrying'); + } else { + logger.error({ err: err.message }, 'Webhook worker error'); + } +}); diff --git a/backend/tests/jobs/webhookDelivery.test.ts b/backend/tests/jobs/webhookDelivery.test.ts new file mode 100644 index 00000000..fb313bcd --- /dev/null +++ b/backend/tests/jobs/webhookDelivery.test.ts @@ -0,0 +1,118 @@ +import { describe, it, expect, vi, beforeEach, afterAll, beforeAll } from 'vitest'; +import { + webhookDeliveryQueue, + webhookDeliveryWorker, + scheduleWebhookDelivery +} from '../../src/jobs/webhookDelivery.js'; +import { redis } from '../../src/db/redis.js'; + +// Mock fetch globally +const fetchMock = vi.fn(); +global.fetch = fetchMock; + +describe('Webhook Delivery Job', () => { + beforeAll(async () => { + // Wait for worker to be ready + await webhookDeliveryWorker.waitUntilReady(); + }); + + afterAll(async () => { + await webhookDeliveryQueue.close(); + await webhookDeliveryWorker.close(); + await redis.quit(); + }); + + beforeEach(async () => { + fetchMock.mockReset(); + // Clear the queue before each test + await webhookDeliveryQueue.drain(); + }); + + it('successfully delivers a webhook and processes the job', async () => { + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + json: async () => ({ success: true }), + } as Response); + + const url = 'https://example.com/webhook'; + const payload = { event: 'test_event', data: { id: 123 } }; + + // Schedule the delivery + await scheduleWebhookDelivery(url, payload); + + // Wait for the job to complete + const completedJob = await new Promise((resolve) => { + webhookDeliveryWorker.once('completed', (job) => { + resolve(job); + }); + }); + + expect(completedJob).toBeDefined(); + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(fetchMock).toHaveBeenCalledWith( + url, + expect.objectContaining({ + method: 'POST', + body: JSON.stringify(payload), + headers: expect.objectContaining({ + 'Content-Type': 'application/json', + 'User-Agent': 'Stellar-Tipz-Webhook-Bot/1.0', + }), + }) + ); + }); + + it('includes an HMAC signature when a secret is provided', async () => { + fetchMock.mockResolvedValueOnce({ + ok: true, + status: 200, + json: async () => ({ success: true }), + } as Response); + + const url = 'https://example.com/webhook-secure'; + const payload = { event: 'secure_event' }; + const secret = 'my-super-secret'; + + await scheduleWebhookDelivery(url, payload, secret); + + await new Promise((resolve) => { + webhookDeliveryWorker.once('completed', (job) => { + resolve(job); + }); + }); + + expect(fetchMock).toHaveBeenCalledTimes(1); + expect(fetchMock).toHaveBeenCalledWith( + url, + expect.objectContaining({ + headers: expect.objectContaining({ + 'X-Signature': expect.stringMatching(/^sha256=[0-9a-f]{64}$/), + }), + }) + ); + }); + + it('fails the job and triggers a retry when the webhook endpoint returns a non-2xx status', async () => { + fetchMock.mockResolvedValueOnce({ + ok: false, + status: 500, + statusText: 'Internal Server Error', + } as Response); + + const url = 'https://example.com/webhook-fail'; + const payload = { event: 'fail_event' }; + + await scheduleWebhookDelivery(url, payload); + + // Wait for the job to fail + const failedJob = await new Promise((resolve) => { + webhookDeliveryWorker.once('failed', (job) => { + resolve(job); + }); + }); + + expect(failedJob).toBeDefined(); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); +});