fix(gateway): disable Redis consumers on local tier (#689)

apps/gateway/src/queue/queue.service.spec.ts

@@ -0,0 +1,35 @@
+import { describe, expect, it, vi } from 'vitest';
+import type { MosaicConfig } from '@mosaicstack/config';
+import { QueueService } from './queue.service.js';
+
+const localConfig = {
+  queue: { type: 'local' },
+} as MosaicConfig;
+
+describe('QueueService local tier', () => {
+  it('disables BullMQ and treats queue operations as local no-ops', async () => {
+    const service = new QueueService(null, localConfig);
+
+    expect(service.isEnabled()).toBe(false);
+    expect(service.getQueue('mosaic-test')).toBeNull();
+    expect(service.registerWorker('mosaic-test', vi.fn())).toBeNull();
+
+    await expect(
+      service.addRepeatableJob('mosaic-test', 'local-noop', {}, '* * * * *'),
+    ).resolves.toBeUndefined();
+    await expect(service.getHealthStatus()).resolves.toEqual({ queues: {}, healthy: true });
+    await expect(service.listJobs()).resolves.toEqual([]);
+    await expect(service.retryJob('mosaic-test__1')).resolves.toEqual({
+      ok: false,
+      message: 'BullMQ is disabled on local tier.',
+    });
+    await expect(service.pauseQueue('mosaic-test')).resolves.toEqual({
+      ok: false,
+      message: 'BullMQ is disabled on local tier.',
+    });
+    await expect(service.resumeQueue('mosaic-test')).resolves.toEqual({
+      ok: false,
+      message: 'BullMQ is disabled on local tier.',
+    });
+  });
+});

apps/gateway/src/queue/queue.service.ts

@@ -8,7 +8,9 @@ import {
 } from '@nestjs/common';
 import { Queue, Worker, type Job, type ConnectionOptions } from 'bullmq';
 import type { LogService } from '@mosaicstack/log';
+import type { MosaicConfig } from '@mosaicstack/config';
 import { LOG_SERVICE } from '../log/log.tokens.js';
+import { MOSAIC_CONFIG } from '../config/config.module.js';
 import type { JobDto, JobStatus } from './queue-admin.dto.js';
 
 // ---------------------------------------------------------------------------
@@ -108,21 +110,42 @@ export class QueueService implements OnModuleInit, OnModuleDestroy {
   private readonly connection: ConnectionOptions;
   private readonly queues = new Map<string, Queue<MosaicJobData>>();
   private readonly workers = new Map<string, Worker<MosaicJobData>>();
+  /** False on Local tier — BullMQ/Redis operations become no-ops. */
+  private readonly enabled: boolean;
 
   constructor(
     @Optional()
     @Inject(LOG_SERVICE)
     private readonly logService: LogService | null,
+    @Optional()
+    @Inject(MOSAIC_CONFIG)
+    private readonly mosaicConfig: MosaicConfig | null,
   ) {
-    this.connection = getConnection();
+    this.enabled = this.mosaicConfig?.queue?.type !== 'local';
+    this.connection = this.enabled
+      ? getConnection()
+      : ({ host: '127.0.0.1', port: 6380 } as ConnectionOptions);
+  }
+
+  /** Returns true when BullMQ/Redis is active (Standalone and Federated tiers). */
+  isEnabled(): boolean {
+    return this.enabled;
   }
 
   onModuleInit(): void {
-    this.logger.log('QueueService initialised (BullMQ)');
+    if (this.enabled) {
+      this.logger.log('QueueService initialised (BullMQ)');
+    } else {
+      this.logger.log(
+        'QueueService: BullMQ disabled for local tier — no Redis connections will be opened',
+      );
+    }
   }
 
   async onModuleDestroy(): Promise<void> {
-    await this.closeAll();
+    if (this.enabled) {
+      await this.closeAll();
+    }
   }
 
   // -------------------------------------------------------------------------
@@ -131,8 +154,10 @@ export class QueueService implements OnModuleInit, OnModuleDestroy {
 
   /**
    * Get or create a BullMQ Queue for the given queue name.
+   * Returns null on Local tier where BullMQ is disabled.
    */
-  getQueue<T extends MosaicJobData = MosaicJobData>(name: string): Queue<T> {
+  getQueue<T extends MosaicJobData = MosaicJobData>(name: string): Queue<T> | null {
+    if (!this.enabled) return null;
     let queue = this.queues.get(name) as Queue<T> | undefined;
     if (!queue) {
       queue = new Queue<T>(name, { connection: this.connection });
@@ -144,6 +169,7 @@ export class QueueService implements OnModuleInit, OnModuleDestroy {
   /**
    * Add a BullMQ repeatable job (cron-style).
    * Uses `jobId` as a deterministic key so duplicate registrations are idempotent.
+   * No-op on Local tier.
    */
   async addRepeatableJob<T extends MosaicJobData>(
     queueName: string,
@@ -151,7 +177,13 @@ export class QueueService implements OnModuleInit, OnModuleDestroy {
     data: T,
     cronExpression: string,
   ): Promise<void> {
-    const queue = this.getQueue<T>(queueName);
+    if (!this.enabled) {
+      this.logger.debug(
+        `Skipping repeatable job "${jobName}" on "${queueName}" (local tier — BullMQ disabled)`,
+      );
+      return;
+    }
+    const queue = this.getQueue<T>(queueName)!;
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     await (queue as Queue<any>).add(jobName, data, {
       repeat: { pattern: cronExpression },
@@ -165,8 +197,18 @@ export class QueueService implements OnModuleInit, OnModuleDestroy {
   /**
    * Register a Worker for the given queue name with error handling and
    * exponential backoff.
+   * Returns null on Local tier where BullMQ is disabled.
    */
-  registerWorker<T extends MosaicJobData>(queueName: string, handler: JobHandler<T>): Worker<T> {
+  registerWorker<T extends MosaicJobData>(
+    queueName: string,
+    handler: JobHandler<T>,
+  ): Worker<T> | null {
+    if (!this.enabled) {
+      this.logger.debug(
+        `Skipping worker registration for "${queueName}" (local tier — BullMQ disabled)`,
+      );
+      return null;
+    }
     const worker = new Worker<T>(
       queueName,
       async (job) => {
@@ -223,8 +265,12 @@ export class QueueService implements OnModuleInit, OnModuleDestroy {
 
   /**
    * Return queue health statistics for all managed queues.
+   * Returns an empty healthy result on Local tier.
    */
   async getHealthStatus(): Promise<QueueHealthStatus> {
+    if (!this.enabled) {
+      return { queues: {}, healthy: true };
+    }
     const queues: QueueHealthStatus['queues'] = {};
     let healthy = true;
 
@@ -255,8 +301,10 @@ export class QueueService implements OnModuleInit, OnModuleDestroy {
   /**
    * List jobs across all managed queues, optionally filtered by status.
    * BullMQ jobs are fetched by state type from each queue.
+   * Returns empty array on Local tier.
    */
   async listJobs(status?: JobStatus): Promise<JobDto[]> {
+    if (!this.enabled) return [];
     const jobs: JobDto[] = [];
     const states: JobStatus[] = status
       ? [status]
@@ -283,8 +331,10 @@ export class QueueService implements OnModuleInit, OnModuleDestroy {
    * Retry a specific failed job by its BullMQ job ID (format: "queueName:id").
    * The caller passes "<queueName>__<jobId>" as the composite ID because BullMQ
    * job IDs are not globally unique — they are scoped to their queue.
+   * Returns an error on Local tier.
    */
   async retryJob(compositeId: string): Promise<{ ok: boolean; message: string }> {
+    if (!this.enabled) return { ok: false, message: 'BullMQ is disabled on local tier.' };
     const sep = compositeId.lastIndexOf('__');
     if (sep === -1) {
       return { ok: false, message: 'Invalid job id format. Expected "<queue>__<jobId>".' };
@@ -316,6 +366,7 @@ export class QueueService implements OnModuleInit, OnModuleDestroy {
    * Pause a queue by name.
    */
   async pauseQueue(name: string): Promise<{ ok: boolean; message: string }> {
+    if (!this.enabled) return { ok: false, message: 'BullMQ is disabled on local tier.' };
     const queue = this.queues.get(name);
     if (!queue) return { ok: false, message: `Queue "${name}" not found.` };
     await queue.pause();
@@ -327,6 +378,7 @@ export class QueueService implements OnModuleInit, OnModuleDestroy {
    * Resume a paused queue by name.
    */
   async resumeQueue(name: string): Promise<{ ok: boolean; message: string }> {
+    if (!this.enabled) return { ok: false, message: 'BullMQ is disabled on local tier.' };
     const queue = this.queues.get(name);
     if (!queue) return { ok: false, message: `Queue "${name}" not found.` };
     await queue.resume();