refactor(job-queue-plugin): Use multiple BullMQ queues instead of one

packages/admin-ui/src/lib/core/src/data/definitions/settings-definitions.ts

@@ -703,6 +703,9 @@ export const GET_SERVER_CONFIG = gql`
                         ...CustomFields
                     }
                 }
+                jobQueue {
+                    supportsListAllQueues
+                }
             }
         }
     }

packages/admin-ui/src/lib/system/src/components/job-list/job-list.component.ts

@@ -8,6 +8,7 @@ import {
     GetJobQueueListQuery,
     ItemOf,
     JobState,
+    ServerConfigService,
     SortOrder,
 } from '@vendure/admin-ui/core';
 import { Observable, timer } from 'rxjs';
@@ -28,8 +29,14 @@ export class JobListComponent
     queueFilter = new FormControl('all');
     stateFilter = new FormControl<JobState | string>('');
 
-    constructor(private dataService: DataService, router: Router, route: ActivatedRoute) {
+    constructor(
+        private dataService: DataService,
+        router: Router,
+        route: ActivatedRoute,
+        private serverConfigService: ServerConfigService,
+    ) {
         super(router, route);
+
         super.setQueryFn(
             (...args: any[]) => this.dataService.settings.getAllJobs(...args),
             data => data.jobs,
@@ -64,10 +71,21 @@ export class JobListComponent
             .subscribe(() => {
                 this.refresh();
             });
+
+        const supportsListAllQueues = this.serverConfigService.serverConfig.jobQueue.supportsListAllQueues;
+        const defaultQueues = supportsListAllQueues ? [{ name: 'all', running: true }] : [];
+
         this.queues$ = this.dataService.settings
             .getJobQueues()
             .mapStream(res => res.jobQueues)
-            .pipe(map(queues => [{ name: 'all', running: true }, ...queues]));
+            .pipe(map(queues => [...defaultQueues, ...queues]));
+
+        this.queues$.subscribe(queues => {
+            if (!supportsListAllQueues) {
+                this.queueFilter.setValue(queues.at(0).name);
+                this.refresh();
+            }
+        });
     }
 
     hasResult(job: ItemOf<GetAllJobsQuery, 'jobs'>): boolean {

packages/asset-server-plugin/e2e/graphql/generated-e2e-asset-server-plugin-types.ts

@@ -2104,6 +2104,10 @@ export type JobQueue = {
     running: Scalars['Boolean']['output'];
 };
 
+export type JobQueueConfig = {
+    supportsListAllQueues: Scalars['Boolean']['output'];
+};
+
 export type JobSortParameter = {
     attempts?: InputMaybe<SortOrder>;
     createdAt?: InputMaybe<SortOrder>;
@@ -5411,6 +5415,7 @@ export type ServerConfig = {
      */
     customFieldConfig: CustomFields;
     entityCustomFields: Array<EntityCustomFields>;
+    jobQueue: JobQueueConfig;
     moneyStrategyPrecision: Scalars['Int']['output'];
     orderProcess: Array<OrderProcessState>;
     permissions: Array<PermissionDefinition>;

packages/common/src/generated-types.ts

@@ -2153,6 +2153,11 @@ export type JobQueue = {
   running: Scalars['Boolean']['output'];
 };
 
+export type JobQueueConfig = {
+  __typename?: 'JobQueueConfig';
+  supportsListAllQueues: Scalars['Boolean']['output'];
+};
+
 export type JobSortParameter = {
   attempts?: InputMaybe<SortOrder>;
   createdAt?: InputMaybe<SortOrder>;
@@ -5729,6 +5734,7 @@ export type ServerConfig = {
    */
   customFieldConfig: CustomFields;
   entityCustomFields: Array<EntityCustomFields>;
+  jobQueue: JobQueueConfig;
   moneyStrategyPrecision: Scalars['Int']['output'];
   orderProcess: Array<OrderProcessState>;
   permissions: Array<PermissionDefinition>;

packages/core/e2e/graphql/generated-e2e-admin-types.ts

@@ -2104,6 +2104,10 @@ export type JobQueue = {
     running: Scalars['Boolean']['output'];
 };
 
+export type JobQueueConfig = {
+    supportsListAllQueues: Scalars['Boolean']['output'];
+};
+
 export type JobSortParameter = {
     attempts?: InputMaybe<SortOrder>;
     createdAt?: InputMaybe<SortOrder>;
@@ -5411,6 +5415,7 @@ export type ServerConfig = {
      */
     customFieldConfig: CustomFields;
     entityCustomFields: Array<EntityCustomFields>;
+    jobQueue: JobQueueConfig;
     moneyStrategyPrecision: Scalars['Int']['output'];
     orderProcess: Array<OrderProcessState>;
     permissions: Array<PermissionDefinition>;

packages/core/package.json

@@ -93,7 +93,7 @@
         "@types/progress": "^2.0.7",
         "@types/prompts": "^2.4.9",
         "@types/semver": "^7.5.8",
-        "better-sqlite3": "^9.4.3",
+        "better-sqlite3": "^11.3.0",
         "chokidar": "^3.6.0",
         "fs-extra": "^11.2.0",
         "glob": "^10.3.10",

packages/core/src/api/resolvers/admin/global-settings.resolver.ts

@@ -69,6 +69,9 @@ export class GlobalSettingsResolver {
             permittedAssetTypes: this.configService.assetOptions.permittedFileTypes,
             permissions,
             moneyStrategyPrecision: this.configService.entityOptions.moneyStrategy.precision ?? 2,
+            jobQueue: {
+                supportsListAllQueues: this.configService.jobQueueOptions.supportsListAllQueues ?? true,
+            },
         };
     }
 
@@ -134,8 +137,8 @@ export class GlobalSettingsResolver {
                         c.requiresPermission = Array.isArray(requiresPermission)
                             ? requiresPermission
                             : !!requiresPermission
-                            ? [requiresPermission]
-                            : [];
+                              ? [requiresPermission]
+                              : [];
                         return c;
                     })
                     .map(c => {

packages/core/src/api/schema/admin-api/global-settings.type.graphql

@@ -19,10 +19,15 @@ type PermissionDefinition {
     assignable: Boolean!
 }
 
+type JobQueueConfig {
+    supportsListAllQueues: Boolean!
+}
+
 # Programatically extended by the addGraphQLCustomFields function
 type ServerConfig {
     orderProcess: [OrderProcessState!]!
     permittedAssetTypes: [String!]!
     permissions: [PermissionDefinition!]!
     moneyStrategyPrecision: Int!
+    jobQueue: JobQueueConfig!
 }

packages/core/src/config/default-config.ts

@@ -189,6 +189,7 @@ export const defaultConfig: RuntimeVendureConfig = {
         jobBufferStorageStrategy: new InMemoryJobBufferStorageStrategy(),
         activeQueues: [],
         prefix: '',
+        supportsListAllQueues: true,
     },
     customFields: {
         Address: [],

packages/core/src/config/vendure-config.ts

@@ -942,6 +942,15 @@ export interface JobQueueOptions {
      * @since 1.5.0
      */
     prefix?: string;
+
+    /**
+     * @description
+     * Defines if the job queue strategy supports listing all jobs of all queues.
+     * By default, this is set to true, but some strategies may not support this feature.
+     *
+     * @since 3.1.0
+     */
+    supportsListAllQueues?: boolean;
 }
 
 /**

packages/core/src/job-queue/job-queue.service.ts

@@ -54,7 +54,10 @@ export class JobQueueService implements OnModuleDestroy {
         return this.configService.jobQueueOptions.jobQueueStrategy;
     }
 
-    constructor(private configService: ConfigService, private jobBufferService: JobBufferService) {}
+    constructor(
+        private configService: ConfigService,
+        private jobBufferService: JobBufferService,
+    ) {}
 
     /** @internal */
     onModuleDestroy() {
@@ -154,6 +157,13 @@ export class JobQueueService implements OnModuleDestroy {
         return this.jobBufferService.flush(forBuffers);
     }
 
+    /**
+     * @description Returns the raw objects representing the JobQueues.
+     */
+    getRawJobQueues() {
+        return this.queues;
+    }
+
     /**
      * @description
      * Returns an array of `{ name: string; running: boolean; }` for each

packages/dev-server/dev-config.ts

@@ -130,7 +130,7 @@ function getDbConfig(): DataSourceOptions {
         case 'postgres':
             console.log('Using postgres connection');
             return {
-                synchronize: false,
+                synchronize: process.env.DB_SYNC === 'true',
                 type: 'postgres',
                 host: process.env.DB_HOST || 'localhost',
                 port: Number(process.env.DB_PORT) || 5432,
@@ -142,7 +142,7 @@ function getDbConfig(): DataSourceOptions {
         case 'sqlite':
             console.log('Using sqlite connection');
             return {
-                synchronize: false,
+                synchronize: process.env.DB_SYNC === 'true',
                 type: 'better-sqlite3',
                 database: path.join(__dirname, 'vendure.sqlite'),
             };
@@ -158,7 +158,7 @@ function getDbConfig(): DataSourceOptions {
         default:
             console.log('Using mysql connection');
             return {
-                synchronize: true,
+                synchronize: process.env.DB_SYNC === 'true',
                 type: 'mariadb',
                 host: '127.0.0.1',
                 port: 3306,

packages/dev-server/docker-compose.yml

@@ -96,6 +96,11 @@ services:
       - --import-realm
     volumes:
       - keycloak_data:/opt/keycloak/data
+  redis:
+    image: redis:alpine
+    ports:
+      - "6379:6379"
+
 volumes:
   postgres_data:
     driver: local

packages/elasticsearch-plugin/e2e/graphql/generated-e2e-elasticsearch-plugin-types.ts

@@ -2104,6 +2104,10 @@ export type JobQueue = {
     running: Scalars['Boolean']['output'];
 };
 
+export type JobQueueConfig = {
+    supportsListAllQueues: Scalars['Boolean']['output'];
+};
+
 export type JobSortParameter = {
     attempts?: InputMaybe<SortOrder>;
     createdAt?: InputMaybe<SortOrder>;
@@ -5411,6 +5415,7 @@ export type ServerConfig = {
      */
     customFieldConfig: CustomFields;
     entityCustomFields: Array<EntityCustomFields>;
+    jobQueue: JobQueueConfig;
     moneyStrategyPrecision: Scalars['Int']['output'];
     orderProcess: Array<OrderProcessState>;
     permissions: Array<PermissionDefinition>;