feat(app-delete): resume cleanup with per-app jobs

packages/dal/redis/bullmq/binding.ts

@@ -1,7 +1,7 @@
 import { getRedisRuntime } from '../runtime';
 import { getRedisBullMQRuntime } from './context';
 import type { QueueNames } from './names';
-import type { Processor, Queue, QueueOptions, Worker, WorkerOptions } from './types';
+import type { FlowProducer, Processor, Queue, QueueOptions, Worker, WorkerOptions } from './types';
 
 const defaultWorkerOpts: Omit<WorkerOptions, 'connection'> = {
   removeOnComplete: {
@@ -59,6 +59,11 @@ export class BullMQBinding {
       ...opts
     });
   }
+
+  /** Returns the Runtime-owned FlowProducer for atomically creating dependent job trees. */
+  getFlowProducer(): FlowProducer {
+    return this.getRuntime().getFlowProducer();
+  }
 }
 
 /** 进程级 BullMQ 绑定。 */

packages/dal/redis/bullmq/flow-producer-manager.ts

@@ -0,0 +1,61 @@
+import { FlowProducer } from 'bullmq';
+import type { RedisRuntime } from '../runtime/connection';
+import type { RedisRuntimeLogger } from '../types';
+import { closeWithTimeout, forceDisconnect } from './close';
+
+/** Manages the shared FlowProducer lifecycle for a BullMQ Runtime. */
+export class BullMQFlowProducerManager {
+  private flowProducer: FlowProducer | undefined;
+
+  constructor(
+    private readonly options: {
+      redisRuntime: RedisRuntime;
+      logger: RedisRuntimeLogger;
+      closeTimeoutMs: number;
+    }
+  ) {}
+
+  /** Creates one FlowProducer and reuses it for all flows in this Runtime. */
+  getFlowProducer() {
+    if (this.flowProducer) return this.flowProducer;
+
+    const connection = this.options.redisRuntime.createQueueConnection();
+    try {
+      const flowProducer = new FlowProducer({ connection });
+      const errorHandler = (error: Error) => {
+        this.options.logger.error('BullMQ flow producer error', { error });
+      };
+      flowProducer.on('error', errorHandler);
+      this.flowProducer = flowProducer;
+      return flowProducer;
+    } catch (error) {
+      void this.options.redisRuntime.releaseConnection(connection).catch((releaseError) => {
+        this.options.logger.warn(
+          'Failed to release Redis connection after FlowProducer creation error',
+          { error: releaseError }
+        );
+      });
+      throw error;
+    }
+  }
+
+  async close() {
+    const flowProducer = this.flowProducer;
+    this.flowProducer = undefined;
+    if (!flowProducer) return;
+
+    await closeWithTimeout({
+      operation: () => flowProducer.close(),
+      resource: 'BullMQ flow producer',
+      timeoutMs: this.options.closeTimeoutMs
+    }).catch((error) => {
+      this.options.logger.warn('BullMQ flow producer close failed', { error });
+      forceDisconnect({
+        name: 'flow-producer',
+        resource: 'flow producer connection',
+        disconnect: () => flowProducer.disconnect(),
+        logger: this.options.logger
+      });
+    });
+  }
+}

packages/dal/redis/bullmq/index.ts

@@ -2,14 +2,18 @@ export { bullMQ, BullMQBinding } from './binding';
 export { getConfiguredRedisBullMQRuntime, getRedisBullMQRuntime } from './context';
 export { QueueNames } from './names';
 export { RedisBullMQRuntime } from './runtime';
+export { BullMQFlowProducerManager } from './flow-producer-manager';
 export { addOrRequeueFailedJob } from './job-recovery';
 export { DelayedError, UnrecoverableError } from 'bullmq';
 export * from './services';
 export type {
   BullMQRuntimeState,
   BullMQWorkerLifecycleOptions,
   ConnectionOptions,
+  FlowJob,
+  FlowProducer,
   Job,
+  JobNode,
   JobSchedulerJson,
   Processor,
   Queue,

packages/dal/redis/bullmq/runtime.ts

@@ -1,11 +1,12 @@
-import type { Queue, Processor, QueueOptions, Worker, WorkerOptions } from 'bullmq';
+import type { FlowProducer, Processor, Queue, QueueOptions, Worker, WorkerOptions } from 'bullmq';
 import type { RedisRuntime } from '../runtime/connection';
 import {
   DEFAULT_CLOSE_TIMEOUT_MS,
   DEFAULT_RESTART_DELAY_MS,
   silentBullMQLogger
 } from './constants';
 import { BullMQQueueManager } from './queue-manager';
+import { BullMQFlowProducerManager } from './flow-producer-manager';
 import type {
   BullMQRuntimeState,
   BullMQWorkerLifecycleOptions,
@@ -20,6 +21,7 @@ export class RedisBullMQRuntime {
 
   private readonly logger: RedisRuntimeLogger;
   private readonly queueManager: BullMQQueueManager;
+  private readonly flowProducerManager: BullMQFlowProducerManager;
   private readonly workerManager: BullMQWorkerManager;
   private readonly unregisterBeforeCloseHook: () => void;
   private state: BullMQRuntimeState = 'running';
@@ -44,6 +46,11 @@ export class RedisBullMQRuntime {
       logger,
       closeTimeoutMs
     });
+    this.flowProducerManager = new BullMQFlowProducerManager({
+      redisRuntime,
+      logger,
+      closeTimeoutMs
+    });
     this.workerManager = new BullMQWorkerManager({
       redisRuntime,
       logger,
@@ -83,6 +90,12 @@ export class RedisBullMQRuntime {
     return this.workerManager.getWorker<DataType, ReturnType>(name, processor, opts);
   }
 
+  /** Returns the Runtime-shared FlowProducer, which owns a dedicated queue connection. */
+  getFlowProducer(): FlowProducer {
+    this.assertRunning();
+    return this.flowProducerManager.getFlowProducer();
+  }
+
   close() {
     if (this.closePromise) return this.closePromise;
 
@@ -101,6 +114,13 @@ export class RedisBullMQRuntime {
         }
 
         // Worker 关闭失败也不能跳过 Queue，否则队列连接会被 Redis Runtime 强制回收。
+        try {
+          await this.flowProducerManager.close();
+        } catch (error) {
+          firstError = error;
+          hasError = true;
+        }
+
         try {
           await this.queueManager.close();
         } catch (error) {

packages/dal/redis/bullmq/services/appDelete.ts

@@ -1,11 +1,14 @@
 import { bullMQ, type BullMQBinding } from '../binding';
 import { addOrRequeueFailedJob } from '../job-recovery';
 import { QueueNames } from '../names';
-import type { Processor, Queue, Worker } from '../types';
+import type { FlowJob, JobNode, Processor, Queue, Worker } from '../types';
 
 export type AppDeleteJobData = {
   teamId: string;
   appId: string;
+  /** Distinguishes task roots, internal Flow steps, and pre-Flow jobs. */
+  jobType?: 'task' | 'step' | 'root' | 'app';
+  taskId?: string;
 };
 
 const appDeleteQueueOptions = {
@@ -20,6 +23,22 @@ const appDeleteQueueOptions = {
   }
 };
 
+const appDeleteFlowJobOptions = appDeleteQueueOptions.defaultJobOptions;
+
+/** Apply the App deletion queue retry and retention policy to every node in a Flow tree. */
+const applyFlowJobOptions = (flow: FlowJob): FlowJob => ({
+  ...flow,
+  opts: {
+    ...appDeleteFlowJobOptions,
+    ...flow.opts
+  },
+  ...(flow.children
+    ? {
+        children: flow.children.map((child) => applyFlowJobOptions(child))
+      }
+    : {})
+});
+
 /** App 删除队列的业务合同和生命周期入口。 */
 export class AppDeleteMQService {
   constructor(private readonly binding: BullMQBinding = bullMQ) {}
@@ -40,19 +59,31 @@ export class AppDeleteMQService {
     });
   }
 
-  /** 投递幂等的 App 删除任务，并延迟一秒让请求先完成。 */
+  /** Add a root deletion job and delay it by one second for request completion. */
   addJob(data: AppDeleteJobData) {
     const jobId = `${String(data.teamId)}-${String(data.appId)}`;
     return addOrRequeueFailedJob({
       queue: this.getQueue(),
       name: 'delete_app',
-      data,
+      data: { ...data, jobType: 'root' },
       opts: {
         jobId,
         delay: 1000
       }
     });
   }
+
+  /** Returns the FlowProducer used to atomically submit deletion task Flows. */
+  getFlowProducer() {
+    return this.binding.getFlowProducer();
+  }
+
+  /** Atomically submits task Flows; callers define step order through the dependency chain. */
+  addFlows(flows: FlowJob[]): Promise<JobNode[]> {
+    if (flows.length === 0) return Promise.resolve([]);
+
+    return this.getFlowProducer().addBulk(flows.map((flow) => applyFlowJobOptions(flow)));
+  }
 }
 
 export const appDeleteMQService = new AppDeleteMQService();

packages/dal/redis/bullmq/types.ts

@@ -1,6 +1,9 @@
 import type {
   ConnectionOptions,
+  FlowJob,
+  FlowProducer,
   Job,
+  JobNode,
   JobSchedulerJson,
   Processor,
   Queue,
@@ -42,7 +45,10 @@ export type WorkerListenerSnapshot = {
 
 export type {
   ConnectionOptions,
+  FlowJob,
+  FlowProducer,
   Job,
+  JobNode,
   JobSchedulerJson,
   Processor,
   Queue,

packages/dal/test/redis/bullmq-binding.test.ts

@@ -4,19 +4,22 @@ vi.unmock('@fastgpt/dal/redis/bullmq');
 
 const bullMQMocks = vi.hoisted(() => {
   const queue = { id: 'queue' };
+  const flowProducer = { id: 'flow-producer' };
   const worker = { id: 'worker' };
   const redisRuntime = {
     id: 'redis-runtime',
     getLogger: vi.fn(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() }))
   };
   const runtime = {
     getQueue: vi.fn(() => queue),
+    getFlowProducer: vi.fn(() => flowProducer),
     getWorker: vi.fn(() => worker),
     getLogger: vi.fn(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() }))
   };
 
   return {
     queue,
+    flowProducer,
     worker,
     redisRuntime,
     runtime,
@@ -38,6 +41,7 @@ describe('DAL BullMQ binding', () => {
   beforeEach(async () => {
     vi.resetModules();
     bullMQMocks.runtime.getQueue.mockClear();
+    bullMQMocks.runtime.getFlowProducer.mockClear();
     bullMQMocks.runtime.getWorker.mockClear();
     bullMQMocks.getRedisBullMQRuntime.mockClear();
     bullMQ = await import('@fastgpt/dal/redis/bullmq');
@@ -78,4 +82,9 @@ describe('DAL BullMQ binding', () => {
       })
     );
   });
+
+  it('delegates FlowProducer access to the DAL BullMQ Runtime', () => {
+    expect(bullMQ.bullMQ.getFlowProducer()).toBe(bullMQMocks.flowProducer);
+    expect(bullMQMocks.runtime.getFlowProducer).toHaveBeenCalledTimes(1);
+  });
 });

packages/dal/test/redis/bullmq-services.test.ts

@@ -25,11 +25,13 @@ describe('BullMQ business services', () => {
   it('allows queue binding injection while keeping queue contracts in the service class', async () => {
     const queue = {
       add: vi.fn().mockResolvedValue({ id: 'job-1' }),
+      addBulk: vi.fn().mockResolvedValue([{ id: 'job-1' }]),
       getJob: vi.fn().mockResolvedValue(null)
     };
     const binding = {
       getQueue: vi.fn(() => queue),
       getWorker: vi.fn(),
+      getFlowProducer: vi.fn(),
       getLogger: vi.fn(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() }))
     } as unknown as BullMQBinding;
     const service = new AppDeleteMQService(binding);
@@ -47,13 +49,67 @@ describe('BullMQ business services', () => {
         removeOnFail: { age: 30 * 24 * 60 * 60 }
       }
     });
-    expect(queue.add).toHaveBeenCalledWith('delete_app', data, {
-      jobId: 'team-1-app-1',
-      delay: 1000
-    });
+    expect(queue.add).toHaveBeenCalledWith(
+      'delete_app',
+      { ...data, jobType: 'root' },
+      {
+        jobId: 'team-1-app-1',
+        delay: 1000
+      }
+    );
     expect(queue.getJob).toHaveBeenCalledWith('team-1-app-1');
   });
 
+  it('adds one task Flow with retry options on every node', async () => {
+    const flowProducer = {
+      addBulk: vi.fn().mockResolvedValue([{ job: { id: 'task-1' } }])
+    };
+    const binding = {
+      getQueue: vi.fn(),
+      getWorker: vi.fn(),
+      getFlowProducer: vi.fn(() => flowProducer),
+      getLogger: vi.fn(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() }))
+    } as unknown as BullMQBinding;
+    const service = new AppDeleteMQService(binding);
+
+    await expect(
+      service.addFlows([
+        {
+          name: 'delete_app_task',
+          queueName: 'appDelete',
+          data: { jobType: 'task' },
+          opts: { jobId: 'task-1' },
+          children: [
+            {
+              name: 'delete_app_step',
+              queueName: 'appDelete',
+              data: { jobType: 'step' },
+              opts: { jobId: 'task-1:step:app-1' }
+            }
+          ]
+        }
+      ])
+    ).resolves.toEqual([{ job: { id: 'task-1' } }]);
+
+    expect(flowProducer.addBulk).toHaveBeenCalledWith([
+      expect.objectContaining({
+        opts: expect.objectContaining({
+          jobId: 'task-1',
+          attempts: 10,
+          removeOnComplete: true
+        }),
+        children: [
+          expect.objectContaining({
+            opts: expect.objectContaining({
+              jobId: 'task-1:step:app-1',
+              attempts: 10
+            })
+          })
+        ]
+      })
+    ]);
+  });
+
   it('uses failed-job recovery for Dataset deletion jobs', async () => {
     const queue = {
       add: vi.fn().mockResolvedValue({ id: 'job-2' }),