Merge pull request #59 from Mrwicks00/feature/durable-job-queue-bullmq

feat: migrate auto-rebalancer and analytics to durable job queue (fix…

Author: ritik4ever

backend/.env.example

@@ -68,6 +68,16 @@ MIN_REBALANCE_INTERVAL=86400000
 # Maximum auto-rebalances per day per portfolio
 MAX_AUTO_REBALANCES_PER_DAY=3
 
+# ============================================
+# JOB QUEUE CONFIGURATION (BullMQ + Redis)
+# ============================================
+
+# Redis connection URL for BullMQ durable job queue (issue #38)
+# Format: redis://[user:password@]host[:port][/db]
+# Defaults to redis://localhost:6379 if not set
+# Required for production – when unset, queue-backed scheduling is skipped gracefully
+REDIS_URL=redis://localhost:6379
+
 # ============================================
 # CIRCUIT BREAKER CONFIGURATION
 # ============================================

backend/package-lock.json

@@ -14,10 +14,12 @@
     "dependencies": {
         "@stellar/stellar-sdk": "^12.0.1",
         "better-sqlite3": "^12.6.2",
+        "bullmq": "^5.69.3",
         "cors": "^2.8.5",
         "dotenv": "^16.3.1",
         "express": "^4.18.2",
         "express-rate-limit": "^7.4.1",
+        "ioredis": "^5.9.3",
         "node-cron": "^3.0.3",
         "nodemailer": "^8.0.1",
         "pg": "^8.11.3",
@@ -31,9 +33,9 @@
         "@types/node-cron": "^3.0.11",
         "@types/nodemailer": "^7.0.10",
         "@types/pg": "^8.10.9",
+        "@types/supertest": "^2.0.12",
         "@types/ws": "^8.18.1",
         "supertest": "^6.3.3",
-        "@types/supertest": "^2.0.12",
         "tsx": "^4.1.4",
         "typescript": "^5.2.2",
         "vitest": "^4.0.18"

backend/package.json

@@ -11,6 +11,7 @@ import { logger } from '../utils/logger.js'
 import { requireAdmin } from '../middleware/auth.js'
 import { blockDebugInProduction } from '../middleware/debugGate.js'
 import { writeRateLimiter } from '../middleware/rateLimit.js'
+import { getQueueMetrics } from '../queue/queueMetrics.js'
 
 const router = Router()
 const stellarService = new StellarService()
@@ -1416,4 +1417,32 @@ router.get('/debug/auto-rebalancer-test', blockDebugInProduction, async (req, re
     }
 })
 
+// ================================
+// QUEUE HEALTH ROUTE
+// ================================
+
+/**
+ * GET /api/queue/health
+ * Returns BullMQ queue depths and Redis connectivity status.
+ * Used for worker health monitoring and alerting (issue #38).
+ */
+router.get('/queue/health', async (req, res) => {
+    try {
+        const metrics = await getQueueMetrics()
+        const httpStatus = metrics.redisConnected ? 200 : 503
+        res.status(httpStatus).json({
+            success: metrics.redisConnected,
+            ...metrics,
+            timestamp: new Date().toISOString(),
+        })
+    } catch (error) {
+        res.status(500).json({
+            success: false,
+            error: getErrorMessage(error),
+            redisConnected: false,
+            timestamp: new Date().toISOString(),
+        })
+    }
+})
+
 export { router as portfolioRouter }

backend/src/api/routes.ts

@@ -10,6 +10,12 @@ import { RebalancingService } from './monitoring/rebalancer.js'
 import { AutoRebalancerService } from './services/autoRebalancer.js'
 import { logger } from './utils/logger.js'
 import { databaseService } from './services/databaseService.js'
+import { isRedisAvailable, logQueueStartup } from './queue/connection.js'
+import { startQueueScheduler } from './queue/scheduler.js'
+import { startPortfolioCheckWorker, stopPortfolioCheckWorker } from './queue/workers/portfolioCheckWorker.js'
+import { startRebalanceWorker, stopRebalanceWorker } from './queue/workers/rebalanceWorker.js'
+import { startAnalyticsSnapshotWorker, stopAnalyticsSnapshotWorker } from './queue/workers/analyticsSnapshotWorker.js'
+import { closeAllQueues } from './queue/queues.js'
 
 const app = express()
 const port = process.env.PORT || 3001
@@ -62,6 +68,9 @@ app.use((req, res, next) => {
 
 app.use(globalRateLimiter)
 
+// Create auto-rebalancer instance
+const autoRebalancer = new AutoRebalancerService()
+
 // Health check endpoint
 app.get('/health', (req, res) => {
     res.json({
@@ -139,14 +148,12 @@ app.get('/', (req, res) => {
             health: '/health',
             corsTest: '/test/cors',
             coinGeckoTest: '/test/coingecko',
-            autoRebalancerStatus: '/api/auto-rebalancer/status'
+            autoRebalancerStatus: '/api/auto-rebalancer/status',
+            queueHealth: '/api/queue/health'
         }
     })
 })
 
-// Create auto-rebalancer instance
-const autoRebalancer = new AutoRebalancerService()
-
 // Mount API routes
 app.use('/api', portfolioRouter)
 app.use('/', portfolioRouter)
@@ -161,7 +168,8 @@ app.use((req, res) => {
         availableEndpoints: {
             health: '/health',
             api: '/api/*',
-            autoRebalancer: '/api/auto-rebalancer/*'
+            autoRebalancer: '/api/auto-rebalancer/*',
+            queueHealth: '/api/queue/health'
         }
     })
 })
@@ -194,30 +202,49 @@ wss.on('connection', (ws) => {
     })
 })
 
-// Start existing rebalancing service
+// Start existing rebalancing service (now queue-backed, no cron)
 try {
     const rebalancingService = new RebalancingService(wss)
     rebalancingService.start()
-    console.log('[LEGACY-REBALANCER] Legacy rebalancing service started')
+    console.log('[REBALANCING-SERVICE] Monitoring service started (queue-backed)')
 } catch (error) {
-    console.error('Failed to start legacy rebalancing service:', error)
+    console.error('Failed to start rebalancing service:', error)
 }
 
 // Start server
-server.listen(port, () => {
+server.listen(port, async () => {
     console.log(`🚀 Server running on port ${port}`)
     console.log(`Environment: ${process.env.NODE_ENV || 'development'}`)
     console.log(`CoinGecko API Key: ${!!process.env.COINGECKO_API_KEY ? 'SET' : 'NOT SET'}`)
 
-    // Start automatic rebalancing service
+    // ── BullMQ / Redis setup ────────────────────────────────────────────────
+    const redisAvailable = await isRedisAvailable()
+    logQueueStartup(redisAvailable)
+
+    if (redisAvailable) {
+        // Start all three workers
+        startPortfolioCheckWorker()
+        startRebalanceWorker()
+        startAnalyticsSnapshotWorker()
+
+        // Register repeatable jobs (scheduler)
+        try {
+            await startQueueScheduler()
+            console.log('[SCHEDULER] ✅ Queue scheduler registered')
+        } catch (err) {
+            console.error('[SCHEDULER] ❌ Failed to register scheduler:', err)
+        }
+    }
+
+    // ── Auto-rebalancer (queue-backed) ──────────────────────────────────────
     const shouldStartAutoRebalancer =
         process.env.NODE_ENV === 'production' ||
         process.env.ENABLE_AUTO_REBALANCER === 'true'
 
     if (shouldStartAutoRebalancer) {
         try {
             console.log('[AUTO-REBALANCER] Starting automatic rebalancing service...')
-            autoRebalancer.start()
+            await autoRebalancer.start()
             console.log('[AUTO-REBALANCER] ✅ Automatic rebalancing service started successfully')
 
             // Broadcast to WebSocket clients
@@ -243,10 +270,11 @@ server.listen(port, () => {
     console.log(`  CORS Test: http://localhost:${port}/test/cors`)
     console.log(`  CoinGecko Test: http://localhost:${port}/test/coingecko`)
     console.log(`  Auto-Rebalancer Status: http://localhost:${port}/api/auto-rebalancer/status`)
+    console.log(`  Queue Health: http://localhost:${port}/api/queue/health`)
 })
 
 // Graceful shutdown
-const gracefulShutdown = (signal: string) => {
+const gracefulShutdown = async (signal: string) => {
     console.log(`\n[SHUTDOWN] ${signal} received, shutting down gracefully...`)
 
     // Stop auto-rebalancer
@@ -257,6 +285,26 @@ const gracefulShutdown = (signal: string) => {
         console.error('[SHUTDOWN] Error stopping auto-rebalancer:', error)
     }
 
+    // Stop BullMQ workers
+    try {
+        await Promise.all([
+            stopPortfolioCheckWorker(),
+            stopRebalanceWorker(),
+            stopAnalyticsSnapshotWorker(),
+        ])
+        console.log('[SHUTDOWN] BullMQ workers stopped')
+    } catch (error) {
+        console.error('[SHUTDOWN] Error stopping BullMQ workers:', error)
+    }
+
+    // Close BullMQ queues
+    try {
+        await closeAllQueues()
+        console.log('[SHUTDOWN] BullMQ queues closed')
+    } catch (error) {
+        console.error('[SHUTDOWN] Error closing queues:', error)
+    }
+
     // Close database connection
     try {
         databaseService.close()