From d64b1d9bc5d27bd9f4a28922a82c0c2278be47f1 Mon Sep 17 00:00:00 2001 From: Ramkumar <153575188+fuzziecoder@users.noreply.github.com> Date: Wed, 25 Feb 2026 09:06:17 -0500 Subject: [PATCH] Add BullMQ background job scaffolding and Swagger API docs --- .env.example | 15 ++++ backend/README.md | 22 ++++++ backend/db.js | 46 +++++++++++ backend/env.js | 119 ++++++++++++++++++++++------- backend/jobs.js | 187 +++++++++++++++++++++++++++++++++++++++++++++ backend/openapi.js | 146 +++++++++++++++++++++++++++++++++++ backend/server.js | 75 +++++++++++++++++- 7 files changed, 582 insertions(+), 28 deletions(-) create mode 100644 backend/jobs.js create mode 100644 backend/openapi.js diff --git a/.env.example b/.env.example index 830be13..373f388 100644 --- a/.env.example +++ b/.env.example @@ -9,3 +9,18 @@ VITE_GEMINI_API_KEY=your_gemini_api_key_here # These are only used when mockApi.ts is active VITE_ADMIN_PASSWORD=your_admin_password_here VITE_USER_PASSWORD=your_user_password_here + +# Backend server +PORT=4000 +CORS_ALLOW_ORIGIN=* +AUTH_TOKEN_SECRET=replace_with_a_long_random_secret +AUTH_TOKEN_TTL_SECONDS=43200 +LOGIN_RATE_LIMIT_MAX_ATTEMPTS=5 +LOGIN_RATE_LIMIT_WINDOW_MS=900000 +LOGIN_RATE_LIMIT_BLOCK_MS=900000 + +# BullMQ + Redis +REDIS_HOST=127.0.0.1 +REDIS_PORT=6379 +REDIS_PASSWORD= +EVENT_REMINDER_BEFORE_HOURS=2 diff --git a/backend/README.md b/backend/README.md index 430e5cb..48c898f 100644 --- a/backend/README.md +++ b/backend/README.md @@ -47,11 +47,31 @@ Server starts at `http://localhost:4000` by default. - `AUTH_TOKEN_TTL_SECONDS` - `CORS_ALLOW_ORIGIN` +### Background jobs (BullMQ + Redis) + +- The backend initializes BullMQ queues for: + - email notification jobs (`email-notifications`) when a new order is created. + - scheduled reminder jobs (`spot-reminders`) for upcoming spots/events. + - recurring cleanup jobs (`expired-spot-cleanup`) for expired events. +- Redis connection settings: + - `REDIS_HOST` (default `127.0.0.1`) + - `REDIS_PORT` (default `6379`) + - `REDIS_PASSWORD` (optional) +- Reminder timing: + - `EVENT_REMINDER_BEFORE_HOURS` (default `2`) +- If BullMQ or Redis dependencies are unavailable, backend continues to run with jobs disabled and logs a warning. + +### API Documentation (Swagger/OpenAPI) + +- OpenAPI JSON is available at `GET /api/docs/openapi.json`. +- Swagger UI is available at `GET /api/docs`. ## Available endpoints - `GET /api/health` - `POST /api/auth/login` +- `GET /api/docs` +- `GET /api/docs/openapi.json` - `GET /api/catalog` - `GET /api/catalog/:category` (`drinks`, `food`, `cigarettes`) - `GET /api/spots` @@ -60,6 +80,8 @@ Server starts at `http://localhost:4000` by default. - `POST /api/orders` (auth required) - `GET /api/bills/:spotId` (admin only) - `DELETE /api/users/:userId` (admin only; removes the user and all related records) +- `POST /api/jobs/reminders/run` (admin only; manually queue reminder jobs) +- `POST /api/jobs/cleanup/run` (admin only; manually queue expired-event cleanup) ## Example login payload diff --git a/backend/db.js b/backend/db.js index cea7338..34e5835 100644 --- a/backend/db.js +++ b/backend/db.js @@ -197,6 +197,23 @@ export const database = { .sort((a, b) => b.date.localeCompare(a.date)); }, + getSpotsBetween({ fromInclusive, toInclusive }) { + const fromValue = fromInclusive ? new Date(fromInclusive).getTime() : Number.NEGATIVE_INFINITY; + const toValue = toInclusive ? new Date(toInclusive).getTime() : Number.POSITIVE_INFINITY; + + return state.spots + .filter((spot) => { + const timestamp = new Date(spot.date).getTime(); + return timestamp >= fromValue && timestamp <= toValue; + }) + .map((spot) => ({ + id: spot.id, + location: spot.location, + date: spot.date, + hostUserId: spot.host_user_id, + })); + }, + getOrders({ spotId, userId }) { const orders = state.orders .filter((order) => !spotId || order.spot_id === spotId) @@ -308,6 +325,35 @@ export const database = { orderCount: summaryRows.length, }; }, + + cleanupExpiredSpots(referenceDate = new Date().toISOString()) { + const referenceTimestamp = new Date(referenceDate).getTime(); + const expiredSpotIds = new Set( + state.spots + .filter((spot) => new Date(spot.date).getTime() < referenceTimestamp) + .map((spot) => spot.id) + ); + + if (expiredSpotIds.size === 0) { + return { removedSpotCount: 0, removedOrderCount: 0, removedOrderItemCount: 0 }; + } + + const previousOrderCount = state.orders.length; + const previousOrderItemCount = state.order_items.length; + + state.spots = state.spots.filter((spot) => !expiredSpotIds.has(spot.id)); + state.orders = state.orders.filter((order) => !expiredSpotIds.has(order.spot_id)); + const activeOrderIds = new Set(state.orders.map((order) => order.id)); + state.order_items = state.order_items.filter((item) => activeOrderIds.has(item.order_id)); + + persist(); + + return { + removedSpotCount: expiredSpotIds.size, + removedOrderCount: previousOrderCount - state.orders.length, + removedOrderItemCount: previousOrderItemCount - state.order_items.length, + }; + }, }; export { dbPath }; diff --git a/backend/env.js b/backend/env.js index 1a2890d..5262b50 100644 --- a/backend/env.js +++ b/backend/env.js @@ -1,30 +1,95 @@ -import dotenv from 'dotenv'; -import { z } from 'zod'; - -dotenv.config(); - -const envSchema = z.object({ - VITE_SUPABASE_URL: z.string().url(), - VITE_SUPABASE_ANON_KEY: z.string().min(10), - PORT: z.string().optional(), - AUTH_TOKEN_SECRET: z.string().min(16).optional(), - AUTH_TOKEN_TTL_SECONDS: z.string().regex(/^\d+$/).optional(), - CORS_ALLOW_ORIGIN: z.string().optional(), - LOGIN_RATE_LIMIT_MAX_ATTEMPTS: z.string().regex(/^\d+$/).optional(), - LOGIN_RATE_LIMIT_WINDOW_MS: z.string().regex(/^\d+$/).optional(), - LOGIN_RATE_LIMIT_BLOCK_MS: z.string().regex(/^\d+$/).optional(), -}); - -const result = envSchema.safeParse(process.env); - -if (!result.success) { - console.error('\n❌ Invalid environment configuration:\n'); - - result.error.errors.forEach((err) => { - console.error(`- ${err.path.join('.')}: ${err.message}`); +import { existsSync, readFileSync } from 'node:fs'; +import { resolve } from 'node:path'; + +const ENV_PATH = resolve(process.cwd(), '.env'); + +const parseEnvLine = (line) => { + const trimmed = line.trim(); + if (!trimmed || trimmed.startsWith('#')) { + return null; + } + + const separatorIndex = trimmed.indexOf('='); + if (separatorIndex < 0) { + return null; + } + + const key = trimmed.slice(0, separatorIndex).trim(); + let value = trimmed.slice(separatorIndex + 1).trim(); + + if ( + (value.startsWith('"') && value.endsWith('"')) || + (value.startsWith("'") && value.endsWith("'")) + ) { + value = value.slice(1, -1); + } + + return { key, value }; +}; + +const loadDotEnv = () => { + if (!existsSync(ENV_PATH)) { + return; + } + + const contents = readFileSync(ENV_PATH, 'utf-8'); + contents.split(/\r?\n/).forEach((line) => { + const parsed = parseEnvLine(line); + if (!parsed || process.env[parsed.key] !== undefined) { + return; + } + + process.env[parsed.key] = parsed.value; }); +}; + +const isIntegerString = (value) => typeof value === 'string' && /^\d+$/.test(value); + +const validateEnv = () => { + const errors = []; + const { VITE_SUPABASE_URL, VITE_SUPABASE_ANON_KEY } = process.env; + + if (!VITE_SUPABASE_URL) { + errors.push('VITE_SUPABASE_URL is required'); + } else { + try { + new URL(VITE_SUPABASE_URL); + } catch { + errors.push('VITE_SUPABASE_URL must be a valid URL'); + } + } + + if (!VITE_SUPABASE_ANON_KEY || VITE_SUPABASE_ANON_KEY.length < 10) { + errors.push('VITE_SUPABASE_ANON_KEY is required and must be at least 10 characters'); + } + + const numericKeys = [ + 'AUTH_TOKEN_TTL_SECONDS', + 'LOGIN_RATE_LIMIT_MAX_ATTEMPTS', + 'LOGIN_RATE_LIMIT_WINDOW_MS', + 'LOGIN_RATE_LIMIT_BLOCK_MS', + 'REDIS_PORT', + ]; + + numericKeys.forEach((key) => { + const value = process.env[key]; + if (value !== undefined && !isIntegerString(value)) { + errors.push(`${key} must be an integer string`); + } + }); + + if (process.env.AUTH_TOKEN_SECRET && process.env.AUTH_TOKEN_SECRET.length < 16) { + errors.push('AUTH_TOKEN_SECRET must be at least 16 characters when provided'); + } + + if (errors.length > 0) { + console.error('\n❌ Invalid environment configuration:\n'); + errors.forEach((error) => console.error(`- ${error}`)); + process.exit(1); + } +}; - process.exit(1); -} +loadDotEnv(); +validateEnv(); -export default result.data; +export default process.env; diff --git a/backend/jobs.js b/backend/jobs.js new file mode 100644 index 0000000..c7eb6fd --- /dev/null +++ b/backend/jobs.js @@ -0,0 +1,187 @@ +import { database } from './db.js'; + +const redisConfig = { + host: process.env.REDIS_HOST || '127.0.0.1', + port: Number(process.env.REDIS_PORT || 6379), + password: process.env.REDIS_PASSWORD || undefined, +}; + +const safeLog = (message, details) => { + if (details) { + console.log(`[jobs] ${message}`, details); + return; + } + + console.log(`[jobs] ${message}`); +}; + +const safeError = (message, error) => { + console.error(`[jobs] ${message}`, error?.message || error); +}; + +const buildReminderPayload = (spot) => ({ + spotId: spot.id, + location: spot.location, + date: spot.date, + hostUserId: spot.hostUserId, +}); + +export const createJobSystem = async () => { + let Queue; + let Worker; + let QueueEvents; + let IORedis; + + try { + ({ Queue, Worker, QueueEvents } = await import('bullmq')); + ({ default: IORedis } = await import('ioredis')); + } catch (error) { + safeError( + 'BullMQ/ioredis not available. Background jobs are disabled until dependencies are installed.', + error + ); + + return { + enabled: false, + async enqueueEmailNotification() {}, + async enqueueSpotReminders() {}, + async enqueueExpiredSpotCleanup() {}, + async shutdown() {}, + }; + } + + const connection = new IORedis(redisConfig); + + const emailQueue = new Queue('email-notifications', { connection }); + const reminderQueue = new Queue('spot-reminders', { connection }); + const cleanupQueue = new Queue('expired-spot-cleanup', { connection }); + + const emailEvents = new QueueEvents('email-notifications', { connection }); + const reminderEvents = new QueueEvents('spot-reminders', { connection }); + const cleanupEvents = new QueueEvents('expired-spot-cleanup', { connection }); + + emailEvents.on('completed', ({ jobId }) => safeLog(`Email job completed: ${jobId}`)); + reminderEvents.on('completed', ({ jobId }) => safeLog(`Reminder job completed: ${jobId}`)); + cleanupEvents.on('completed', ({ jobId }) => safeLog(`Cleanup job completed: ${jobId}`)); + + emailEvents.on('failed', ({ jobId, failedReason }) => safeError(`Email job failed (${jobId})`, failedReason)); + reminderEvents.on('failed', ({ jobId, failedReason }) => safeError(`Reminder job failed (${jobId})`, failedReason)); + cleanupEvents.on('failed', ({ jobId, failedReason }) => safeError(`Cleanup job failed (${jobId})`, failedReason)); + + const emailWorker = new Worker( + 'email-notifications', + async (job) => { + const { toUserId, subject, message } = job.data; + safeLog('Sending email notification', { toUserId, subject, message }); + return { delivered: true, sentAt: new Date().toISOString() }; + }, + { connection } + ); + + const reminderWorker = new Worker( + 'spot-reminders', + async (job) => { + safeLog('Sending scheduled event reminder', job.data); + return { delivered: true, sentAt: new Date().toISOString() }; + }, + { connection } + ); + + const cleanupWorker = new Worker( + 'expired-spot-cleanup', + async () => { + const result = database.cleanupExpiredSpots(); + safeLog('Expired events cleanup finished', result); + return result; + }, + { connection } + ); + + const enqueueEmailNotification = async ({ toUserId, subject, message }) => + emailQueue.add( + 'send-email-notification', + { toUserId, subject, message }, + { removeOnComplete: true, removeOnFail: 100, attempts: 3 } + ); + + const enqueueSpotReminders = async ({ beforeHours = 2 } = {}) => { + const now = Date.now(); + const reminderWindowEnd = now + beforeHours * 60 * 60 * 1000; + const spots = database.getSpotsBetween({ + fromInclusive: new Date(now).toISOString(), + toInclusive: new Date(reminderWindowEnd).toISOString(), + }); + + await Promise.all( + spots.map((spot) => + reminderQueue.add( + 'send-event-reminder', + buildReminderPayload(spot), + { + jobId: `reminder:${spot.id}:${beforeHours}`, + removeOnComplete: true, + removeOnFail: 100, + attempts: 2, + } + ) + ) + ); + + return { queuedReminders: spots.length }; + }; + + const enqueueExpiredSpotCleanup = async () => + cleanupQueue.add('cleanup-expired-events', {}, { removeOnComplete: true, removeOnFail: 100 }); + + await cleanupQueue.add( + 'cleanup-expired-events-recurring', + {}, + { + jobId: 'cleanup-expired-events-recurring', + repeat: { every: 60 * 60 * 1000 }, + removeOnComplete: true, + removeOnFail: 100, + } + ); + + await reminderQueue.add( + 'send-event-reminders-recurring', + { beforeHours: Number(process.env.EVENT_REMINDER_BEFORE_HOURS || 2) }, + { + jobId: 'send-event-reminders-recurring', + repeat: { every: 30 * 60 * 1000 }, + removeOnComplete: true, + removeOnFail: 100, + } + ); + + reminderWorker.on('completed', async (job) => { + if (job?.name === 'send-event-reminders-recurring') { + await enqueueSpotReminders({ beforeHours: job.data.beforeHours }); + } + }); + + const shutdown = async () => { + await Promise.allSettled([ + emailWorker.close(), + reminderWorker.close(), + cleanupWorker.close(), + emailQueue.close(), + reminderQueue.close(), + cleanupQueue.close(), + emailEvents.close(), + reminderEvents.close(), + cleanupEvents.close(), + ]); + + await connection.quit(); + }; + + return { + enabled: true, + enqueueEmailNotification, + enqueueSpotReminders, + enqueueExpiredSpotCleanup, + shutdown, + }; +}; diff --git a/backend/openapi.js b/backend/openapi.js new file mode 100644 index 0000000..eb1dce7 --- /dev/null +++ b/backend/openapi.js @@ -0,0 +1,146 @@ +const buildOpenApiSpec = (port) => ({ + openapi: '3.0.3', + info: { + title: 'Brocode Party Update API', + version: '1.0.0', + description: + 'Backend API for authentication, catalog, events, orders, billing, and maintenance jobs.', + }, + servers: [{ url: `http://localhost:${port}` }], + tags: [ + { name: 'Health' }, + { name: 'Auth' }, + { name: 'Catalog' }, + { name: 'Spots' }, + { name: 'Orders' }, + { name: 'Bills' }, + { name: 'Users' }, + { name: 'Jobs' }, + ], + components: { + securitySchemes: { + BearerAuth: { + type: 'http', + scheme: 'bearer', + bearerFormat: 'Token', + }, + }, + }, + paths: { + '/api/health': { get: { tags: ['Health'], summary: 'Health check', responses: { 200: { description: 'OK' } } } }, + '/api/auth/login': { + post: { + tags: ['Auth'], + summary: 'Authenticate user and return token', + requestBody: { + required: true, + content: { + 'application/json': { + schema: { + type: 'object', + required: ['username', 'password'], + properties: { + username: { type: 'string' }, + password: { type: 'string' }, + }, + }, + }, + }, + }, + responses: { 200: { description: 'Authenticated' }, 401: { description: 'Invalid credentials' } }, + }, + }, + '/api/catalog': { get: { tags: ['Catalog'], summary: 'Get catalog', responses: { 200: { description: 'Catalog' } } } }, + '/api/catalog/{category}': { + get: { + tags: ['Catalog'], + summary: 'Get catalog category', + parameters: [{ name: 'category', in: 'path', required: true, schema: { type: 'string' } }], + responses: { 200: { description: 'Category items' }, 404: { description: 'Unknown category' } }, + }, + }, + '/api/spots': { get: { tags: ['Spots'], summary: 'Get events/spots', responses: { 200: { description: 'Spots' } } } }, + '/api/orders': { + get: { + tags: ['Orders'], + security: [{ BearerAuth: [] }], + summary: 'List orders', + parameters: [ + { name: 'spotId', in: 'query', schema: { type: 'string' } }, + { name: 'userId', in: 'query', schema: { type: 'string' } }, + ], + responses: { 200: { description: 'Orders' }, 401: { description: 'Unauthorized' } }, + }, + post: { + tags: ['Orders'], + security: [{ BearerAuth: [] }], + summary: 'Create order', + responses: { 201: { description: 'Created' }, 400: { description: 'Validation error' } }, + }, + }, + '/api/orders/{orderId}': { + get: { + tags: ['Orders'], + security: [{ BearerAuth: [] }], + summary: 'Get order by id', + parameters: [{ name: 'orderId', in: 'path', required: true, schema: { type: 'string' } }], + responses: { 200: { description: 'Order' }, 404: { description: 'Not found' } }, + }, + }, + '/api/bills/{spotId}': { + get: { + tags: ['Bills'], + security: [{ BearerAuth: [] }], + summary: 'Get bill summary for a spot', + parameters: [{ name: 'spotId', in: 'path', required: true, schema: { type: 'string' } }], + responses: { 200: { description: 'Bill' }, 403: { description: 'Forbidden' } }, + }, + }, + '/api/users/{userId}': { + delete: { + tags: ['Users'], + security: [{ BearerAuth: [] }], + summary: 'Delete a user and related records', + parameters: [{ name: 'userId', in: 'path', required: true, schema: { type: 'string' } }], + responses: { 200: { description: 'Deleted' }, 403: { description: 'Forbidden' } }, + }, + }, + '/api/jobs/reminders/run': { + post: { + tags: ['Jobs'], + summary: 'Trigger reminder enqueue job now', + responses: { 202: { description: 'Reminder enqueue accepted' } }, + }, + }, + '/api/jobs/cleanup/run': { + post: { + tags: ['Jobs'], + summary: 'Trigger expired events cleanup job now', + responses: { 202: { description: 'Cleanup enqueue accepted' } }, + }, + }, + }, +}); + +const buildSwaggerHtml = () => ` + + + + + Brocode API Docs + + + +
+ + + +`; + +export { buildOpenApiSpec, buildSwaggerHtml }; diff --git a/backend/server.js b/backend/server.js index 5d5532a..e578a9e 100644 --- a/backend/server.js +++ b/backend/server.js @@ -2,6 +2,8 @@ import { createServer } from 'node:http'; import { createHmac, timingSafeEqual } from 'node:crypto'; import { URL } from 'node:url'; import { database, dbPath } from './db.js'; +import { createJobSystem } from './jobs.js'; +import { buildOpenApiSpec, buildSwaggerHtml } from './openapi.js'; import "./env.js"; const port = Number(process.env.PORT || 4000); @@ -13,6 +15,8 @@ const AUTH_TOKEN_SECRET = process.env.AUTH_TOKEN_SECRET || 'brocode-dev-secret-c const AUTH_TOKEN_TTL_SECONDS = Number(process.env.AUTH_TOKEN_TTL_SECONDS || 60 * 60 * 12); const CORS_ALLOW_ORIGIN = process.env.CORS_ALLOW_ORIGIN || '*'; const loginAttempts = new Map(); +const SWAGGER_HTML = buildSwaggerHtml(); +const jobSystem = await createJobSystem(); const getLoginKey = (req, username) => { const forwardedFor = req.headers['x-forwarded-for']; @@ -146,6 +150,14 @@ const sendJson = (res, statusCode, body) => { res.end(JSON.stringify(body)); }; +const sendHtml = (res, statusCode, html) => { + res.writeHead(statusCode, { + 'Content-Type': 'text/html; charset=utf-8', + 'Access-Control-Allow-Origin': CORS_ALLOW_ORIGIN, + }); + res.end(html); +}; + const readBody = (req) => new Promise((resolve, reject) => { let data = ''; @@ -181,7 +193,24 @@ const server = createServer(async (req, res) => { } if (method === 'GET' && path === '/api/health') { - sendJson(res, 200, { status: 'ok', service: 'brocode-backend', timestamp: new Date().toISOString() }); + sendJson(res, 200, { + status: 'ok', + service: 'brocode-backend', + timestamp: new Date().toISOString(), + jobs: { + enabled: jobSystem.enabled, + }, + }); + return; + } + + if (method === 'GET' && path === '/api/docs/openapi.json') { + sendJson(res, 200, buildOpenApiSpec(port)); + return; + } + + if (method === 'GET' && path === '/api/docs') { + sendHtml(res, 200, SWAGGER_HTML); return; } @@ -326,6 +355,13 @@ const server = createServer(async (req, res) => { } const newOrder = database.createOrder({ spotId, userId, items }); + + await jobSystem.enqueueEmailNotification({ + toUserId: userId, + subject: `Order placed for ${spotId}`, + message: `Your order ${newOrder.id} with total ₹${newOrder.totalAmount} has been received.`, + }); + sendJson(res, 201, newOrder); return; } catch (error) { @@ -375,10 +411,47 @@ const server = createServer(async (req, res) => { return; } + if (method === 'POST' && path === '/api/jobs/reminders/run') { + const authedUser = getUserFromAuthHeader(req.headers.authorization); + if (!authedUser || authedUser.role !== 'admin') { + sendJson(res, 403, { error: 'Forbidden' }); + return; + } + + const result = await jobSystem.enqueueSpotReminders({ + beforeHours: Number(process.env.EVENT_REMINDER_BEFORE_HOURS || 2), + }); + sendJson(res, 202, { accepted: true, ...result, jobsEnabled: jobSystem.enabled }); + return; + } + + if (method === 'POST' && path === '/api/jobs/cleanup/run') { + const authedUser = getUserFromAuthHeader(req.headers.authorization); + if (!authedUser || authedUser.role !== 'admin') { + sendJson(res, 403, { error: 'Forbidden' }); + return; + } + + await jobSystem.enqueueExpiredSpotCleanup(); + sendJson(res, 202, { accepted: true, jobsEnabled: jobSystem.enabled }); + return; + } + sendJson(res, 404, { error: 'Route not found' }); }); server.listen(port, () => { console.log(`Backend API running on http://localhost:${port}`); console.log(`Using local database at: ${dbPath}`); + console.log(`Swagger docs available at http://localhost:${port}/api/docs`); +}); + +process.on('SIGINT', async () => { + await jobSystem.shutdown(); + process.exit(0); +}); + +process.on('SIGTERM', async () => { + await jobSystem.shutdown(); + process.exit(0); });