feat: add Redis-based job queue with BullMQ

[mchestr]

Summary

• Add Redis-based background job queue system using BullMQ for processing async tasks
• Falls back to legacy PostgreSQL-based polling when Redis is not configured
• Admin dashboard at /admin/queues for monitoring jobs and queue status
• Scheduler support for repeating jobs (e.g., watchlist sync every N minutes)

Changes

• New queue system (lib/queue/) - BullMQ client, worker, connection management
• Admin UI (app/admin/queues/, components/admin/queues/) - Job table, queue stats, controls
• Watchlist sync - Uses queue when Redis available, triggers via admin or user action
• Instrumentation - Queue worker auto-starts with graceful shutdown handling
• Config - New env vars: REDIS_URL, ENABLE_QUEUE_WORKER, QUEUE_WORKER_CONCURRENCY

Test plan

• Verify app works without Redis (falls back to legacy polling)
• Configure Redis and verify queue worker starts
• Test admin queue dashboard displays jobs correctly
• Trigger watchlist sync and verify job is queued/processed
• Test graceful shutdown stops worker cleanly

🤖 Generated with Claude Code

[claude]

Claude finished @mchestr's task in 3m 17s —— View job

---

Code Review: Redis-based Job Queue with BullMQ

I've completed a comprehensive review of this PR. Great work addressing the previous feedback! Here's my updated assessment:

---

✅ What Was Fixed (Since Last Review)

The commit "fix: address security and testing review comments" (1406c50) successfully addressed several critical issues:

1. ✅ Security Improvements:

   • Added rate limiting (30 requests/min per admin) to all queue admin actions
   • Added audit logging for all queue operations (actions/admin/queue.ts:31-64)
   • Fixed sensitive data logging - now only logs safe fields like userId and triggeredBy (lib/queue/client.ts:86-93)

2. ✅ Input Validation:

   • Added isRegisteredJobType() validation for job types (actions/admin/queue.ts:261)
   • Improved error messages to be user-friendly instead of exposing internals

3. ✅ Error Handling:

   • All error messages now use generic text to avoid leaking implementation details
   • Proper error propagation with logger.error calls

4. ✅ Test Infrastructure:

   • Updated admin nav test for new Queues menu item (components/tests/admin-shared.test.tsx)
   • Added data-testid attributes to UI components for E2E testing

---

⚠️ Critical Issues Remaining

1. Multiple SIGINT/SIGTERM Handlers (Race Condition) 🚨

Location: lib/instrumentation/node.ts:89-90, 164-165, 266-267

Issue: Three separate shutdown handlers are registered:

• Queue worker shutdown (line 89-90)
• Watchlist polling shutdown (line 164-165)
• Discord bot shutdown (line 266-267)

Problem: When SIGINT/SIGTERM is received, all three handlers execute in parallel. If one fails, the others may not complete, leading to:

• Unclosed Redis connections
• Orphaned worker processes
• Data loss from incomplete job processing

Impact: High - could cause resource leaks in production

Recommendation: Create a single coordinated shutdown handler that calls all cleanup functions sequentially with proper error handling. This ensures orderly shutdown even if one subsystem fails.

---

2. Missing Test Coverage 🧪

Issue: No unit or E2E tests found for the queue system itself (searched for *queue*.test.ts and *queue*.spec.ts patterns).

What's Missing:

• Unit tests for lib/queue/client.ts (job scheduling, queue operations)
• Unit tests for lib/queue/worker.ts (job processing, error handling, retry logic)
• Unit tests for lib/queue/connection.ts (Redis connection management)
• Integration tests for job processors (lib/queue/jobs/watchlist-sync.ts)
• E2E tests for admin queue dashboard (/admin/queues)

Impact: High - without tests, regression bugs could be introduced during refactoring

Recommendation: Add comprehensive test coverage before merging. Use Redis Mock for unit tests to avoid external dependencies.

---

3. Job ID Collision Risk ⚠️

Location: lib/queue/client.ts:80

const jobId = options?.jobId ?? `${jobType}:${Date.now()}`

Problem: Date.now() has millisecond precision. Multiple jobs added in the same millisecond will have the same ID, causing collisions.

Impact: Medium - more likely under high load (e.g., bulk operations)

Fix Options:

1. Add UUID suffix: ${jobType}:${Date.now()}-${randomUUID()}
2. Let BullMQ auto-generate IDs (remove jobId if not provided)

---

4. Non-null Assertion on job.id ⚠️

Location: lib/queue/client.ts:95, 193, 223

return job.id!  // Line 95
jobId: job.id!  // Lines 193, 223

Problem: BullMQ's job.id type is string | undefined. Using ! assertion assumes it's always defined, but this could fail if a job is created but not yet persisted.

Impact: Medium - could cause runtime errors

Fix: Add proper null check:

if (!job.id) {
  throw new Error('Job was created but has no ID')
}
return job.id

---

💡 Recommendations & Improvements

5. Lazy Connection Pattern (Informational)

Location: lib/queue/connection.ts:22, 49-71

The Redis connection uses lazyConnect: true but never explicitly calls .connect(). BullMQ will auto-connect on first use, which is intentional for avoiding connections during build time.

Recommendation: Add a JSDoc comment explaining this design choice to prevent confusion:

/**
 * Get or create the shared Redis connection
 * Note: Uses lazy initialization with BullMQ auto-connect pattern.
 * The connection is established automatically when first accessed by BullMQ.
 */

---

6. Static UI Refresh

Location: app/admin/queues/page.tsx

The admin dashboard fetches data server-side but doesn't auto-refresh. Users must manually refresh the page to see job status updates.

Recommendation: Add one of:

• export const revalidate = 5 for automatic ISR refresh every 5 seconds
• Client-side polling with setInterval + router.refresh()
• Manual refresh button

---

7. N+1 Query Pattern

Location: lib/queue/client.ts:189-207

return Promise.all(
  jobs.map(async (job) => ({
    status: (await job.getState()) as JobStatus,  // N queries
    // ...
  }))
)

Issue: Each job's state requires a separate Redis call. For 50 jobs, this is 50+ Redis round trips.

Impact: Low-Medium - performance degradation with large result sets

Note: This may be a BullMQ limitation. Consider batching or caching if performance becomes an issue.

---

8. Hardcoded Page Limit

Location: app/admin/queues/page.tsx:32

limit: 50,

Recommendation: Make configurable via query params or environment variable for flexibility.

---

🎯 Code Quality & Best Practices

✅ Excellent Implementation:

• Clean separation of concerns (client, worker, connection, jobs)
• Type-safe job system with mapped types
• Graceful fallback to PostgreSQL polling when Redis unavailable
• Proper use of Server Actions over API routes
• Comprehensive Zod validation
• Centralized logging with appropriate log levels
• Good use of singleton patterns
• Admin UI follows project conventions (Tailwind, dark theme)

✅ Security:

• All Server Actions use requireAdmin()
• Rate limiting implemented ✨
• Audit logging added ✨
• Sensitive data redacted from logs ✨
• Input validation with Zod schemas

✅ Adherence to CLAUDE.md:

• Server Components by default ✓
• Server Actions for mutations ✓
• TypeScript strict mode ✓
• Proper error handling ✓
• Uses existing UI components ✓
• No over-engineering ✓

---

📊 Summary

Overall Assessment: This is a well-architected implementation with strong type safety and proper separation of concerns. The recent fixes addressed critical security concerns effectively.

Risk Level: Medium (primarily due to missing tests and shutdown coordination issue)

Before Merging:

1. 🚨 MUST FIX: Add comprehensive test coverage (unit + E2E)
2. ⚠️ SHOULD FIX: Resolve multiple SIGINT handler race condition
3. ⚠️ SHOULD FIX: Fix job ID collision risk
4. ⚠️ SHOULD FIX: Add null checks for job.id

Nice to Have:

• Add UI refresh mechanism for live updates
• Document lazy connection pattern
• Make page limit configurable

---

🧪 Suggested Test Structure

lib/queue/__tests__/
  ├── client.test.ts          # Queue operations, job scheduling
  ├── worker.test.ts          # Job processing, error handling
  ├── connection.test.ts      # Redis connection management
  └── jobs/
      └── watchlist-sync.test.ts  # Job processor logic

e2e/
  └── admin-queues.spec.ts    # Admin dashboard E2E tests

Use ioredis-mock for unit tests to avoid Redis dependency.

---

Great work on this feature! The improvements since the last review are substantial. Address the remaining issues and this will be production-ready. 🚀

[claude]

Code Review: Redis-based Job Queue with BullMQ

I've completed a comprehensive review of the Redis-based job queue implementation. Here's my detailed feedback:

---

✅ Strengths

Excellent Architecture & Design:

• Clean separation of concerns with dedicated modules for client, worker, connection, types, and jobs
• Type-safe job system with proper TypeScript generics and mapped types
• Graceful fallback to PostgreSQL polling when Redis is unavailable
• Proper singleton patterns for Queue, Worker, and Redis connections
• Well-structured admin UI with reusable components

Security & Best Practices:

• All Server Actions properly use requireAdmin() for authorization checks
• Comprehensive Zod validation for all inputs (getJobsSchema, jobIdSchema, etc.)
• Follows project guidelines: Server Actions over API routes, type safety, proper error handling
• Good use of logger throughout for observability

Robust Error Handling:

• Graceful shutdown handlers for SIGINT/SIGTERM
• Proper cleanup of Redis connections and worker instances
• Retry mechanism with exponential backoff
• Error boundaries and fallback states in UI

---

🐛 Critical Issues

1. Resource Leak in Singleton Pattern ⚠️

Location: lib/queue/connection.ts:49-72

The Redis connection uses lazyConnect: true but never calls .connect(). BullMQ will auto-connect when first accessed, but this can cause race conditions and connection issues.

// Current (line 49-52):
if (!redisConnection) {
  const url = getRedisUrl()
  redisConnection = new Redis(url, getRedisOptions())
  // Event handlers...
}
return redisConnection

Fix:

export async function getRedisConnection(): Promise<Redis> {
  if (!redisConnection) {
    const url = getRedisUrl()
    redisConnection = new Redis(url, getRedisOptions())
    // ... event handlers ...
    await redisConnection.connect() // Add explicit connect
  }
  return redisConnection
}

Impact: Function signature change breaks all call sites. Alternative: Keep sync, but document lazy connection behavior clearly.

---

2. Multiple SIGINT/SIGTERM Handlers ⚠️

Location: lib/instrumentation/node.ts:89-90, 164-165, 266-267

Three separate process.on('SIGINT') handlers are registered:

• Queue worker shutdown (line 89-90)
• Watchlist polling shutdown (line 164-165)
• Discord bot shutdown (line 266-267)

Problem: All handlers execute in parallel during shutdown, creating a race condition. If one fails, others may not complete.

Fix: Use a single coordinated shutdown handler:

// At module level
const shutdownHandlers: Array<() => Promise<void>> = []

function registerShutdownHandler(handler: () => Promise<void>) {
  shutdownHandlers.push(handler)
}

// Single shutdown coordinator
if (process.env.NODE_ENV !== 'test') {
  const shutdown = async () => {
    console.log('Shutting down...')
    for (const handler of shutdownHandlers) {
      try {
        await handler()
      } catch (error) {
        console.error('Shutdown handler error:', error)
      }
    }
    process.exit(0)
  }

  process.on('SIGINT', shutdown)
  process.on('SIGTERM', shutdown)
}

Then register handlers from each subsystem:

registerShutdownHandler(async () => {
  await stopWorker()
  await closeQueue()
  await closeRedisConnection()
})

---

3. Missing Test Coverage 🧪

Issue: No test files found for the queue system (searched for test.*queue patterns).

Required Tests:

• Unit tests for lib/queue/client.ts (job scheduling, queue operations)
• Unit tests for lib/queue/worker.ts (job processing, error handling)
• Integration tests for watchlist sync job handlers
• E2E tests for admin queue dashboard at /admin/queues

Recommendation: Add test files before merging:

lib/queue/__tests__/client.test.ts
lib/queue/__tests__/worker.test.ts
lib/queue/__tests__/connection.test.ts
lib/queue/jobs/__tests__/watchlist-sync.test.ts
e2e/admin-queues.spec.ts

Use Redis Mock for unit tests to avoid external dependencies.

---

⚠️ High Priority Issues

4. Job ID Collision Risk

Location: lib/queue/client.ts:80

const jobId = options?.jobId ?? `${jobType}:${Date.now()}`

Problem: Date.now() has millisecond precision. Multiple jobs added in the same millisecond will collide.

Fix: Use a more robust ID generation:

import { randomUUID } from 'crypto'

const jobId = options?.jobId ?? `${jobType}:${Date.now()}-${randomUUID()}`

Or let BullMQ generate IDs automatically by not setting jobId when not provided:

const job = await q.add(jobType, data, options)

---

5. Server Component Using Client-Only APIs

Location: components/admin/queues/job-table.tsx:1

Component is marked 'use client' but imports look correct. However, the parent page (app/admin/queues/page.tsx) is a Server Component that passes data directly.

Concern: The page fetches data server-side (good!) but doesn't handle revalidation. When jobs complete/fail, the UI won't update unless user refreshes.

Recommendation: Add polling or use router.refresh() periodically:

// In app/admin/queues/page.tsx
export const revalidate = 5 // Revalidate every 5 seconds

Or add a client-side refresh button/auto-refresh in the UI.

---

6. Non-null Assertion on Job ID

Location: lib/queue/client.ts:92, 190, 220

Multiple uses of job.id! without checking if it's defined:

return job.id!  // Line 92
jobId: job.id!  // Line 190, 220

Problem: BullMQ job IDs are string | undefined. If a job is created but not yet persisted, id may be undefined.

Fix: Add proper error handling:

if (!job.id) {
  throw new Error('Job was created but has no ID')
}
return job.id

---

🔧 Medium Priority Issues

7. Hardcoded Admin UI Values

Location: app/admin/queues/page.tsx:32

limit: 50,

Limit is hardcoded to 50. Consider making this configurable via query params or env var.

---

8. Incomplete Error Details in Logs

Location: lib/queue/worker.ts:58

logger.error('No processor found for job type', undefined, {
  jobType: job.name,
  jobId: job.id,
})

Passing undefined as the error. Should pass an Error object or null:

logger.error('No processor found for job type', null, { ... })

---

9. Missing Data Validation in Job Processors

Location: lib/queue/jobs/watchlist-sync.ts:23-26

Job data is destructured without validation:

const { userId, triggeredBy, triggeredByUserId } = job.data

Recommendation: Add runtime validation with Zod to ensure job data integrity:

const payload = WatchlistSyncUserPayloadSchema.parse(job.data)
const { userId, triggeredBy, triggeredByUserId } = payload

---

10. Scheduler Pattern Format Not Documented

Location: actions/admin/queue.ts:118-119

pattern: s.pattern as string,

The pattern field is returned to UI but its format isn't clear. Is it a cron expression? Milliseconds? Document this for frontend developers.

---

💡 Suggestions & Improvements

11. Consider Adding Queue Metrics

For production observability, consider exposing:

• Queue latency (time in queue before processing)
• Job processing duration histogram
• Error rates by job type
• Worker utilization

BullMQ supports metrics via events. Could integrate with Prometheus/Grafana.

---

12. Add Job Progress Updates

The WatchlistSyncUserPayload could benefit from progress reporting:

await job.updateProgress({
  step: 'fetching_watchlist',
  itemsProcessed: 10,
  itemsTotal: 50
})

Then display in admin UI for long-running jobs.

---

13. Environment Variable Validation

Location: lib/queue/worker.ts:22-30

const parsed = parseInt(envConcurrency, 10)
if (!isNaN(parsed) && parsed > 0) {
  return parsed
}

Consider using Zod schema for all env vars at startup to fail fast:

const envSchema = z.object({
  REDIS_URL: z.string().url().optional(),
  QUEUE_WORKER_CONCURRENCY: z.coerce.number().int().min(1).max(100).default(5),
  // ...
})

export const env = envSchema.parse(process.env)

---

14. Missing CORS Configuration for Queue UI

If the admin dashboard will be accessed from different origins, ensure proper CORS headers are set. Currently not addressed.

---

15. Consider Rate Limiting for Manual Triggers

Location: actions/watchlist.ts:174, actions/admin/watchlist.ts

Users can spam manual sync triggers. Consider adding rate limiting:

import { rateLimit } from '@/lib/security/rate-limit'

const limiter = rateLimit({
  interval: 60 * 1000, // 1 minute
  uniqueTokenPerInterval: 500,
})

// In action:
const identifier = session.user.id
if (!await limiter.check(identifier, 5)) { // 5 requests per minute
  return { success: false, error: 'Rate limit exceeded' }
}

---

📝 Documentation & Code Quality

16. Excellent JSDoc Coverage ✅

Great job on comprehensive JSDoc comments throughout lib/queue/client.ts, worker.ts, and other modules. This makes the codebase very maintainable.

17. Add README for Queue System

Consider adding lib/queue/README.md with:

• Architecture overview
• How to add new job types
• Testing strategies
• Deployment considerations (Redis requirements, scaling workers)

---

🔒 Security Review

18. Admin Authorization ✅

All admin endpoints properly check requireAdmin(). Good!

19. Input Validation ✅

Zod schemas used consistently. Well done!

20. Sensitive Data in Logs ⚠️

Location: lib/queue/client.ts:86-90

logger.info('Job added to queue', {
  jobId: job.id,
  jobType,
  data,  // ⚠️ Logs entire payload
})

If job data contains sensitive info (user IDs, tokens), this could leak in logs. Consider:

data: summarizeData(data), // Only log non-sensitive fields

Similar concern in lib/queue/jobs/watchlist-sync.ts:28-34.

---

📊 Performance Considerations

21. N+1 Query in getJobs

Location: lib/queue/client.ts:189-205

return Promise.all(
  jobs.map(async (job) => ({
    status: (await job.getState()) as JobStatus,  // ⚠️ N queries
    // ...
  }))
)

Each job's state is fetched individually. For large result sets (50 jobs), this is 50+ Redis calls.

BullMQ Optimization: Use job.getState() in batches or cache states from q.getJobs() call if possible.

---

22. Missing Index Considerations

The PR doesn't include Prisma schema changes, but ensure these indexes exist for performance:

• WatchlistSyncHistory.userId (for history queries)
• WatchlistSyncSettings.userId (unique, should already exist)

---

🧹 Code Style & Conventions

23. Follows CLAUDE.md Guidelines ✅

• Server Actions over API routes ✓
• TypeScript strict mode ✓
• Zod validation ✓
• Proper use of 'use client' ✓
• Tailwind CSS styling ✓
• Centralized logger ✓

24. Component File Size

components/admin/queues/job-table.tsx is 284 lines. While within guidelines (max ~300), consider extracting:

• JobRow component (lines 146-259)
• JobTablePagination component (lines 264-281)

This would improve testability and reusability.

---

🎯 Summary

Overall Assessment: This is a high-quality implementation with excellent architecture and adherence to project guidelines. The fallback mechanism, type safety, and admin UI are well-executed.

Before Merging:

1. ✅ MUST FIX: Add comprehensive test coverage (unit + E2E)
2. ✅ MUST FIX: Resolve multiple SIGINT handler race condition
3. ⚠️ SHOULD FIX: Job ID collision risk with Date.now()
4. ⚠️ SHOULD FIX: Non-null assertions on job.id
5. 💡 NICE TO HAVE: Add polling/revalidation to admin UI for live updates

Estimated Risk Level: Medium (main concerns are testing and shutdown coordination)

Recommendation: Address critical issues (#1, #2, #3) before merging. Other issues can be follow-up improvements.

---

Great work on this feature! The code quality is solid and the design is extensible. Let me know if you'd like me to implement any of these fixes. 🚀

[claude]

Code Review Summary

I've completed a comprehensive review of the Redis-based job queue implementation.

Key Findings

Strengths:

• ✅ Excellent architecture with clean separation of concerns
• ✅ Security improvements: rate limiting, audit logging, and sensitive data redaction
• ✅ Type-safe job system with proper validation
• ✅ Graceful fallback to PostgreSQL when Redis unavailable
• ✅ Follows all project guidelines (CLAUDE.md)

Critical Issues:

1. 🚨 Missing test coverage - No unit or E2E tests for queue system
2. ⚠️ Multiple SIGINT handlers - Race condition during shutdown (3 separate handlers)
3. ⚠️ Job ID collisions - Using Date.now() can cause conflicts under load
4. ⚠️ Non-null assertions - job.id! could fail if job not yet persisted

Recommendations:

• Add comprehensive test coverage before merging (highest priority)
• Consolidate shutdown handlers into single coordinator
• Use UUID suffix for job IDs or let BullMQ auto-generate
• Add proper null checks instead of assertions

See full review with code references and suggested fixes in the comment above.

Risk Level: Medium - Great implementation, but needs tests and shutdown coordination fix before production deployment.