refactor(server): config system (#11081)

packages/backend/server/src/base/job/queue/__tests__/queue.spec.ts

@@ -52,13 +52,15 @@ class JobHandlers {
 test.before(async () => {
   module = await createTestingModule({
     imports: [
-      ConfigModule.forRoot({
+      ConfigModule.override({
         job: {
           worker: {
-            // NOTE(@forehalo):
-            //   bullmq will hold the connection to check stalled jobs,
-            //   which will keep the test process alive to timeout.
-            stalledInterval: 100,
+            defaultWorkerOptions: {
+              // NOTE(@forehalo):
+              //   bullmq will hold the connection to check stalled jobs,
+              //   which will keep the test process alive to timeout.
+              stalledInterval: 100,
+            },
           },
           queue: {
             defaultJobOptions: { delay: 1000 },
@@ -82,7 +84,7 @@ test.afterEach(async () => {
   // @ts-expect-error private api
   const inner = queue.getQueue('nightly');
   await inner.obliterate({ force: true });
-  inner.resume();
+  await inner.resume();
 });
 
 test.after.always(async () => {
@@ -132,7 +134,7 @@ test('should remove job from queue', async t => {
 // #region executor
 test('should start workers', async t => {
   // @ts-expect-error private api
-  const worker = executor.workers['nightly'];
+  const worker = executor.workers.get('nightly')!;
 
   t.truthy(worker);
   t.true(worker.isRunning());

packages/backend/server/src/base/job/queue/config.ts

@@ -1,61 +1,86 @@
 import { QueueOptions, WorkerOptions } from 'bullmq';
 
-import {
-  defineRuntimeConfig,
-  defineStartupConfig,
-  ModuleConfig,
-} from '../../config';
+import { defineModuleConfig, JSONSchema } from '../../config';
 import { Queue } from './def';
 
-declare module '../../config' {
-  interface AppConfig {
-    job: ModuleConfig<
-      {
-        queue: Omit<QueueOptions, 'connection'>;
-        worker: Omit<WorkerOptions, 'connection'>;
-      },
-      {
-        queues: {
-          [key in Queue]: {
-            concurrency: number;
-          };
-        };
-      }
-    >;
+declare global {
+  interface AppConfigSchema {
+    job: {
+      queue: ConfigItem<Omit<QueueOptions, 'connection' | 'telemetry'>>;
+      worker: ConfigItem<{
+        defaultWorkerOptions: Omit<WorkerOptions, 'connection' | 'telemetry'>;
+      }>;
+      queues: {
+        [key in Queue]: ConfigItem<{
+          concurrency: number;
+        }>;
+      };
+    };
   }
 }
 
-defineStartupConfig('job', {
+const schema: JSONSchema = {
+  type: 'object',
+  properties: {
+    concurrency: { type: 'number' },
+  },
+};
+
+defineModuleConfig('job', {
   queue: {
-    prefix: AFFiNE.node.test ? 'affine_job_test' : 'affine_job',
-    defaultJobOptions: {
-      attempts: 5,
-      // should remove job after it's completed, because we will add a new job with the same job id
-      removeOnComplete: true,
-      removeOnFail: {
-        age: 24 * 3600 /* 1 day */,
-        count: 500,
+    desc: 'The config for job queues',
+    default: {
+      prefix: env.testing ? 'affine_job_test' : 'affine_job',
+      defaultJobOptions: {
+        attempts: 5,
+        // should remove job after it's completed, because we will add a new job with the same job id
+        removeOnComplete: true,
+        removeOnFail: {
+          age: 24 * 3600 /* 1 day */,
+          count: 500,
+        },
       },
     },
+    link: 'https://api.docs.bullmq.io/interfaces/v5.QueueOptions.html',
   },
-  worker: {},
-});
 
-defineRuntimeConfig('job', {
-  'queues.nightly.concurrency': {
-    default: 1,
-    desc: 'Concurrency of worker consuming of nightly checking job queue',
+  worker: {
+    desc: 'The config for job workers',
+    default: {
+      defaultWorkerOptions: {},
+    },
+    link: 'https://api.docs.bullmq.io/interfaces/v5.WorkerOptions.html',
   },
-  'queues.notification.concurrency': {
-    default: 10,
-    desc: 'Concurrency of worker consuming of notification job queue',
+
+  'queues.copilot': {
+    desc: 'The config for copilot job queue',
+    default: {
+      concurrency: 1,
+    },
+    schema,
   },
-  'queues.doc.concurrency': {
-    default: 1,
-    desc: 'Concurrency of worker consuming of doc job queue',
+
+  'queues.doc': {
+    desc: 'The config for doc job queue',
+    default: {
+      concurrency: 1,
+    },
+    schema,
   },
-  'queues.copilot.concurrency': {
-    default: 1,
-    desc: 'Concurrency of worker consuming of copilot job queue',
+
+  'queues.notification': {
+    desc: 'The config for notification job queue',
+    default: {
+      concurrency: 10,
+    },
+    schema,
+  },
+
+  'queues.nightly': {
+    desc: 'The config for nightly job queue',
+    default: {
+      concurrency: 1,
+    },
+    schema,
   },
 });

packages/backend/server/src/base/job/queue/def.ts

@@ -49,7 +49,7 @@ export const OnJob = (job: JobName) => {
   if (!QUEUES.includes(ns as Queue)) {
     throw new Error(
       `Invalid job queue: ${ns}, must be one of [${QUEUES.join(', ')}].
-If you want to introduce new job queue, please modify the Queue enum first in ${join(AFFiNE.projectRoot, 'src/base/job/queue/def.ts')}`
+If you want to introduce new job queue, please modify the Queue enum first in ${join(env.projectRoot, 'src/base/job/queue/def.ts')}`
     );
   }
 

packages/backend/server/src/base/job/queue/executor.ts

@@ -1,49 +1,51 @@
-import {
-  Injectable,
-  Logger,
-  OnApplicationBootstrap,
-  OnApplicationShutdown,
-} from '@nestjs/common';
+import { Injectable, Logger, OnModuleDestroy } from '@nestjs/common';
 import { Worker } from 'bullmq';
-import { difference } from 'lodash-es';
+import { difference, merge } from 'lodash-es';
 import { CLS_ID, ClsServiceManager } from 'nestjs-cls';
 
 import { Config } from '../../config';
+import { OnEvent } from '../../event';
 import { metrics, wrapCallMetric } from '../../metrics';
 import { QueueRedis } from '../../redis';
-import { Runtime } from '../../runtime';
 import { genRequestId } from '../../utils';
 import { JOB_SIGNAL, namespace, Queue, QUEUES } from './def';
 import { JobHandlerScanner } from './scanner';
 
 @Injectable()
-export class JobExecutor
-  implements OnApplicationBootstrap, OnApplicationShutdown
-{
+export class JobExecutor implements OnModuleDestroy {
   private readonly logger = new Logger('job');
-  private readonly workers: Record<string, Worker> = {};
+  private readonly workers: Map<Queue, Worker> = new Map();
 
   constructor(
     private readonly config: Config,
     private readonly redis: QueueRedis,
-    private readonly scanner: JobHandlerScanner,
-    private readonly runtime: Runtime
+    private readonly scanner: JobHandlerScanner
   ) {}
 
-  async onApplicationBootstrap() {
-    const queues = this.config.flavor.graphql
-      ? difference(QUEUES, [Queue.DOC])
-      : [];
+  @OnEvent('config.init')
+  async onConfigInit() {
+    const queues = env.flavors.graphql ? difference(QUEUES, [Queue.DOC]) : [];
 
     // NOTE(@forehalo): only enable doc queue in doc service
-    if (this.config.flavor.doc) {
+    if (env.flavors.doc) {
       queues.push(Queue.DOC);
     }
 
     await this.startWorkers(queues);
   }
 
-  async onApplicationShutdown() {
+  @OnEvent('config.changed')
+  async onConfigChanged({ updates }: Events['config.changed']) {
+    if (updates.job?.queues) {
+      Object.entries(updates.job.queues).forEach(([queue, options]) => {
+        if (options.concurrency) {
+          this.setConcurrency(queue as Queue, options.concurrency);
+        }
+      });
+    }
+  }
+
+  async onModuleDestroy() {
     await this.stopWorkers();
   }
 
@@ -98,38 +100,35 @@ export class JobExecutor
     }
   }
 
-  private async startWorkers(queues: Queue[]) {
-    const configs =
-      (await this.runtime.fetchAll(
-        queues.reduce(
-          (ret, queue) => {
-            ret[`job/queues.${queue}.concurrency`] = true;
-            return ret;
-          },
-          {} as {
-            [key in `job/queues.${Queue}.concurrency`]: true;
-          }
-        )
-        // TODO(@forehalo): fix the override by [payment/service.spec.ts]
-      )) ?? {};
+  setConcurrency(queue: Queue, concurrency: number) {
+    const worker = this.workers.get(queue);
+    if (!worker) {
+      throw new Error(`Worker for [${queue}] not found.`);
+    }
 
+    worker.concurrency = concurrency;
+  }
+
+  private async startWorkers(queues: Queue[]) {
     for (const queue of queues) {
-      const concurrency =
-        (configs[`job/queues.${queue}.concurrency`] as number) ??
-        this.config.job.worker.concurrency ??
-        1;
+      const queueOptions = this.config.job.queues[queue];
+      const concurrency = queueOptions.concurrency ?? 1;
 
       const worker = new Worker(
         queue,
         async job => {
           await this.run(job.name as JobName, job.data);
         },
-        {
-          ...this.config.job.queue,
-          ...this.config.job.worker,
-          connection: this.redis,
-          concurrency,
-        }
+        merge(
+          {},
+          this.config.job.queue,
+          this.config.job.worker.defaultWorkerOptions,
+          queueOptions,
+          {
+            concurrency,
+            connection: this.redis,
+          }
+        )
       );
 
       worker.on('error', error => {
@@ -140,13 +139,13 @@ export class JobExecutor
         `Queue Worker [${queue}] started; concurrency=${concurrency};`
       );
 
-      this.workers[queue] = worker;
+      this.workers.set(queue, worker);
     }
   }
 
   private async stopWorkers() {
     await Promise.all(
-      Object.values(this.workers).map(async worker => {
+      Array.from(this.workers.values()).map(async worker => {
         await worker.close(true);
       })
     );