refactor(server): mail service (#10934)

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

@@ -93,10 +93,6 @@ test('should register job handler', async t => {
 
   t.is(handler!.name, 'JobHandlers.handleJob');
   t.is(typeof handler!.fn, 'function');
-
-  const result = await handler!.fn({ name: 'test' });
-
-  t.is(result, 'test');
 });
 // #endregion
 

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

@@ -63,3 +63,7 @@ If you want to introduce new job queue, please modify the Queue enum first in ${
 export function getJobHandlerMetadata(target: any): JobName[] {
   return sliceMetadata<JobName>(JOB_METADATA, target);
 }
+
+export enum JOB_SIGNAL {
+  RETRY = 'retry',
+}

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

@@ -13,7 +13,7 @@ import { metrics, wrapCallMetric } from '../../metrics';
 import { QueueRedis } from '../../redis';
 import { Runtime } from '../../runtime';
 import { genRequestId } from '../../utils';
-import { namespace, Queue, QUEUES } from './def';
+import { JOB_SIGNAL, namespace, Queue, QUEUES } from './def';
 import { JobHandlerScanner } from './scanner';
 
 @Injectable()
@@ -69,8 +69,12 @@ export class JobExecutor
           try {
             this.logger.debug(`Job started: ${signature}`);
             const result = await handler.fn(payload);
+
+            if (result === JOB_SIGNAL.RETRY) {
+              throw new Error(`Manually job retry`);
+            }
+
             this.logger.debug(`Job finished: ${signature}`);
-            return result;
           } catch (e) {
             this.logger.error(`Job failed: ${signature}`, e);
             throw e;
@@ -88,7 +92,7 @@ export class JobExecutor
     const activeJobs = metrics.queue.counter('active_jobs');
     activeJobs.add(1, { queue: ns });
     try {
-      return await fn();
+      await fn();
     } finally {
       activeJobs.add(-1, { queue: ns });
     }

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

@@ -42,4 +42,4 @@ export class JobModule {
 }
 
 export { JobQueue };
-export { OnJob } from './def';
+export { JOB_SIGNAL, OnJob } from './def';

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

@@ -1,11 +1,11 @@
 import { Injectable, OnModuleInit } from '@nestjs/common';
 
 import { ModuleScanner } from '../../nestjs';
-import { getJobHandlerMetadata } from './def';
+import { getJobHandlerMetadata, JOB_SIGNAL } from './def';
 
 interface JobHandler {
   name: string;
-  fn: (payload: any) => any;
+  fn: (payload: any) => Promise<JOB_SIGNAL | undefined>;
 }
 
 @Injectable()