[sync] fix(computed): add failure diagnostics for outbox workers (#2836)

Synced from teableio/teable-ee@3d8d8f8

Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Aries X <caoxing9@gmail.com>
Co-authored-by: Boris <boris2code@outlook.com>
Co-authored-by: Jun Lu <hammond@teable.io>
Co-authored-by: nichenqin <nichenqin@hotmail.com>

Author: 6 people

.github/workflows/issue-id-check.yml

@@ -153,7 +153,7 @@ describe('SelectQueryPostgres FROMNOW/TONOW', () => {
 });
 
 describe('SelectQueryPostgres workday', () => {
-  it('uses interval multiplication for dynamic day-count expressions', () => {
+  it('generates CTE-based workday SQL that skips weekends and holidays', () => {
     const query = new SelectQueryPostgres();
     query.setContext({ timeZone: 'Asia/Shanghai' } as unknown as never);
     query.setCallMetadata([
@@ -162,7 +162,9 @@ describe('SelectQueryPostgres workday', () => {
     ] as unknown as never);
 
     const sql = query.workday('"t"."Date"', '"t"."Number"');
-    expect(sql).toContain(`INTERVAL '1 day' * ("t"."Number")::double precision`);
-    expect(sql).not.toContain(" days'");
+    expect(sql).toContain('WITH params AS');
+    expect(sql).toContain('generate_series');
+    expect(sql).toContain('EXTRACT(DOW FROM c.candidate_date)');
+    expect(sql).toContain(`("t"."Number")::double precision`);
   });
 });

apps/nestjs-backend/src/db-provider/select-query/postgres/select-query.postgres.spec.ts

@@ -2,8 +2,8 @@ import type { OnModuleInit } from '@nestjs/common';
 import { Injectable, Logger } from '@nestjs/common';
 import { Reflector, DiscoveryService } from '@nestjs/core';
 import type { InstanceWrapper } from '@nestjs/core/injector/instance-wrapper';
-import type { Job } from 'bullmq';
 import { localQueueEventEmitter } from './event-emitter';
+import type { ILocalJob } from './local-queue.provider';
 
 export const PROCESSOR_METADATA = 'bullmq:processor_metadata';
 
@@ -43,7 +43,7 @@ export class FallbackQueueService implements OnModuleInit {
         instance.constructor || metatype
       );
       localQueueEventEmitter.removeAllListeners(`handle-listener-${queueName}`);
-      localQueueEventEmitter.on(`handle-listener-${queueName}`, (job: Job<unknown>) => {
+      localQueueEventEmitter.on(`handle-listener-${queueName}`, (job: ILocalJob) => {
         if (job.queueName !== queueName) {
           return;
         }
@@ -55,7 +55,7 @@ export class FallbackQueueService implements OnModuleInit {
   private async handleListener(
     // eslint-disable-next-line @typescript-eslint/no-explicit-any
     wrapper: InstanceWrapper,
-    job: Job<unknown>
+    job: ILocalJob
   ) {
     const { instance } = wrapper;
     const methodName = 'process';
@@ -64,8 +64,13 @@ export class FallbackQueueService implements OnModuleInit {
       return;
     }
     try {
-      await instance[methodName].call(instance, job);
+      job.state = 'active';
+      const result = await instance[methodName].call(instance, job);
+      job.state = 'completed';
+      job.returnvalue = result;
     } catch (error) {
+      job.state = 'failed';
+      job.failedReason = error instanceof Error ? error.message : String(error);
       this.logger.error(`Error processing job ${job.name}:`, error);
     }
   }

apps/nestjs-backend/src/event-emitter/event-job/fallback/fallback-queue.service.ts

@@ -4,24 +4,66 @@ import { getRandomString } from '@teable/core';
 import type { JobsOptions } from 'bullmq';
 import { localQueueEventEmitter } from './event-emitter';
 
+export interface ILocalJob {
+  id: string;
+  name: string;
+  data: unknown;
+  opts?: JobsOptions;
+  queueName: string;
+  progress: number | object;
+  returnvalue: unknown;
+  failedReason?: string;
+  state: string;
+  getState: () => Promise<string>;
+  updateProgress: (progress: number | object) => Promise<void>;
+}
+
 export const createLocalQueueProvider = (queueName: string): Provider => ({
   provide: getQueueToken(queueName),
   useFactory: async () => {
+    const jobs = new Map<string, ILocalJob>();
+
+    const createJob = (id: string, name: string, data: unknown, opts?: JobsOptions): ILocalJob => {
+      const job: ILocalJob = {
+        id,
+        name,
+        data,
+        opts,
+        queueName,
+        progress: 0,
+        returnvalue: undefined,
+        failedReason: undefined,
+        state: 'waiting',
+        getState: async () => job.state,
+        updateProgress: async (p: number | object) => {
+          job.progress = p;
+        },
+      };
+      return job;
+    };
+
     return {
       add: (name: string, data: unknown, opts?: JobsOptions) => {
-        localQueueEventEmitter.emit(`handle-listener-${queueName}`, {
-          id: getRandomString(10),
-          name,
-          data,
-          opts,
-          queueName,
-        });
+        const id = opts?.jobId ?? getRandomString(10);
+        const job = createJob(id, name, data, opts);
+        jobs.set(id, job);
+        localQueueEventEmitter.emit(`handle-listener-${queueName}`, job);
+        return job;
       },
-      addBulk: (jobs: JobsOptions[]) => {
-        jobs.forEach((job) => {
+      addBulk: (bulkJobs: JobsOptions[]) => {
+        bulkJobs.forEach((job) => {
           localQueueEventEmitter.emit(`handle-listener-${queueName}`, job);
         });
       },
+      getJob: async (jobId: string) => {
+        return jobs.get(jobId) ?? null;
+      },
+      getJobs: async () => {
+        return Array.from(jobs.values());
+      },
+      getJobCountByTypes: async () => {
+        return jobs.size;
+      },
     };
   },
 });