Feature/jobs documentation

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
+);
+```

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<string, unknown>;
+  secret?: string; // Optional secret to sign the payload (HMAC)
+}
+
+/**
+ * Queue instance for dispatching webhook deliveries.
+ */
+export const webhookDeliveryQueue = new Queue<WebhookDeliveryPayload>(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<string, unknown>,
+  secret?: string
+): Promise<void> {
+  await webhookDeliveryQueue.add('deliver', { url, payload, secret });
+}
+
+/**
+ * Worker to process webhook deliveries.
+ */
+export const webhookDeliveryWorker = new Worker<WebhookDeliveryPayload>(
+  WEBHOOK_DELIVERY_QUEUE,
+  async (job: Job<WebhookDeliveryPayload>) => {
+    const { url, payload, secret } = job.data;
+
+    logger.info({ jobId: job.id, url }, 'Starting webhook delivery');
+
+    const headers: Record<string, string> = {
+      '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<WebhookDeliveryPayload> | 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');
+  }
+});

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);
+  });
+});