Unsubscribe

packages/dal/redis/adapter.ts

@@ -755,6 +755,40 @@ export class RedisCacheAdapter {
       }
     });
 
+  /** 局部更新 hash 字段且保留现有 TTL。 */
+  updateHashFields = ({
+    key,
+    fields
+  }: {
+    key: RedisLogicalKey;
+    fields: Record<string, string>;
+  }) => {
+    const operation = 'hash.updateFields';
+    if (
+      !fields ||
+      Object.keys(fields).length === 0 ||
+      Object.values(fields).some((value) => typeof value !== 'string')
+    ) {
+      throw new RedisInvalidArgumentError({
+        operation,
+        message: 'hash fields must contain at least one string value'
+      });
+    }
+
+    return this.operationExecutor.uncertainWrite({
+      operation,
+      execute: async () => {
+        const result = await this.getCommandClient().hmset(toPhysicalRedisKey(key), fields);
+        if (result !== 'OK') {
+          throw new RedisInvalidResponseError({
+            operation,
+            message: 'Redis HMSET returned an unsupported response'
+          });
+        }
+      }
+    });
+  };
+
   /** 在一个事务中写入 hash 并设置 TTL，避免 hash 无过期时间。 */
   setHashWithTtl = ({
     key,

packages/dal/redis/bullmq/index.ts

@@ -2,7 +2,8 @@ export { bullMQ, BullMQBinding } from './binding';
 export { getConfiguredRedisBullMQRuntime, getRedisBullMQRuntime } from './context';
 export { QueueNames } from './names';
 export { RedisBullMQRuntime } from './runtime';
-export { UnrecoverableError } from 'bullmq';
+export { addOrRequeueFailedJob } from './job-recovery';
+export { DelayedError, UnrecoverableError } from 'bullmq';
 export * from './services';
 export type {
   BullMQRuntimeState,

packages/dal/redis/bullmq/job-recovery.ts

@@ -0,0 +1,78 @@
+import { LeaseCache } from '../caches';
+import { bullMQ } from './binding';
+import type { Queue } from './types';
+
+const FAILED_JOB_RECOVERY_LEASE_TTL_MS = 30 * 1000;
+
+/**
+ * 添加稳定 ID 任务；若同 ID 历史任务已失败，则在分布式租约内刷新数据并手动重试。
+ * 其它状态继续复用现有任务，避免并发生产者制造重复工作。
+ */
+export async function addOrRequeueFailedJob<DataType, ReturnType = void>({
+  queue,
+  name,
+  data,
+  opts
+}: {
+  queue: Queue<DataType, ReturnType>;
+  name: Parameters<Queue<DataType, ReturnType>['add']>[0];
+  data: Parameters<Queue<DataType, ReturnType>['add']>[1];
+  opts: NonNullable<Parameters<Queue<DataType, ReturnType>['add']>[2]> & { jobId: string };
+}) {
+  /** unknown 可能来自 retention cleanup；二次读取确认不存在后才允许重建。 */
+  const getJobWithConfirmedState = async () => {
+    const job = await queue.getJob(opts.jobId);
+    if (!job) return;
+
+    const state = await job.getState();
+    if (state !== 'unknown') return { job, state };
+
+    const latestJob = await queue.getJob(opts.jobId);
+    if (!latestJob) return;
+
+    const latestState = await latestJob.getState();
+    if (latestState === 'unknown') {
+      throw new Error(`BullMQ job is in an unknown state: ${queue.name}/${opts.jobId}`);
+    }
+    return { job: latestJob, state: latestState };
+  };
+
+  const existing = await getJobWithConfirmedState();
+  if (existing) {
+    if (existing.state !== 'failed') return existing.job;
+
+    const leaseCache = new LeaseCache({ logger: bullMQ.getLogger() });
+    return leaseCache.withLease({
+      key: `bullmq:failed-job-recovery:${queue.name}:${opts.jobId}`,
+      label: 'bullmq-failed-job-recovery',
+      ttlMs: FAILED_JOB_RECOVERY_LEASE_TTL_MS,
+      fn: async () => {
+        const current = await getJobWithConfirmedState();
+        if (!current) return queue.add(name, data, opts);
+        if (current.state !== 'failed') return current.job;
+        const currentJob = current.job;
+
+        try {
+          await currentJob.updateData(data as DataType);
+        } catch (error) {
+          const latest = await getJobWithConfirmedState();
+          if (!latest) return queue.add(name, data, opts);
+          if (latest.state !== 'failed') return latest.job;
+          throw error;
+        }
+
+        try {
+          await currentJob.retry('failed');
+          return currentJob;
+        } catch (error) {
+          const latest = await getJobWithConfirmedState();
+          if (!latest) return queue.add(name, data, opts);
+          if (latest.state !== 'failed') return latest.job;
+          throw error;
+        }
+      }
+    });
+  }
+
+  return queue.add(name, data, opts);
+}

packages/dal/redis/bullmq/names.ts

@@ -11,6 +11,7 @@ export enum QueueNames {
   appDelete = 'appDelete',
   agentSkillDelete = 'agentSkillDelete',
   teamDelete = 'teamDelete',
+  accountCancellation = 'accountCancellation',
 
   // Publish
   wechatPoll = 'wechatPoll',

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

@@ -1,4 +1,5 @@
 import { bullMQ, type BullMQBinding } from '../binding';
+import { addOrRequeueFailedJob } from '../job-recovery';
 import { QueueNames } from '../names';
 import type { Processor, Queue, Worker } from '../types';
 
@@ -40,9 +41,14 @@ export class TeamDeleteMQService {
 
   /** 投递幂等的 Team 删除任务，并延迟一秒让请求先完成。 */
   addJob(data: TeamDeleteJobData) {
-    return this.getQueue().add('delete_team', data, {
-      jobId: String(data.teamId),
-      delay: 1000
+    return addOrRequeueFailedJob({
+      queue: this.getQueue(),
+      name: 'delete_team',
+      data,
+      opts: {
+        jobId: String(data.teamId),
+        delay: 1000
+      }
     });
   }
 }

packages/dal/redis/caches/session.ts

@@ -123,6 +123,21 @@ export class SessionCache {
     await this.redis.delete(this.getKey(sessionId));
   }
 
+  /** 更新 Session 的团队上下文，不刷新原有过期时间。 */
+  updateTeam = ({
+    sessionId,
+    teamId,
+    tmbId
+  }: {
+    sessionId: string;
+    teamId: string;
+    tmbId: string;
+  }) =>
+    this.redis.updateHashFields({
+      key: this.getKey(sessionId),
+      fields: { teamId, tmbId }
+    });
+
   /** 分页扫描某个用户的全部 typed session，损坏记录会被尽力清理。 */
   async listByUser(userId: string): Promise<SessionRecord[]> {
     const records: SessionRecord[] = [];

packages/dal/test/redis/caches/session.test.ts

@@ -16,7 +16,8 @@ describe('SessionCache', () => {
     deleteMany: vi.fn(),
     getHashAll: vi.fn(),
     iterateByPrefix: vi.fn(),
-    setHashWithTtl: vi.fn()
+    setHashWithTtl: vi.fn(),
+    updateHashFields: vi.fn()
   };
 
   beforeEach(() => {
@@ -33,6 +34,7 @@ describe('SessionCache', () => {
     });
     redis.iterateByPrefix.mockReturnValue(createKeyBatches([]));
     redis.setHashWithTtl.mockResolvedValue(undefined);
+    redis.updateHashFields.mockResolvedValue(undefined);
   });
 
   it('decodes a complete session hash and normalizes isRoot/createdAt', async () => {
@@ -188,6 +190,22 @@ describe('SessionCache', () => {
     ]);
   });
 
+  it('updates only the team context so the existing TTL is preserved', async () => {
+    const cache = new SessionCache({ redis: redis as any, logger });
+
+    await cache.updateTeam({
+      sessionId: 'user-1:token-1',
+      teamId: 'team-2',
+      tmbId: 'tmb-2'
+    });
+
+    expect(redis.updateHashFields).toHaveBeenCalledWith({
+      key: 'session:user-1:token-1',
+      fields: { teamId: 'team-2', tmbId: 'tmb-2' }
+    });
+    expect(redis.setHashWithTtl).not.toHaveBeenCalled();
+  });
+
   it('scans all user pages and returns only valid typed sessions', async () => {
     redis.iterateByPrefix.mockReturnValue(
       createKeyBatches([
@@ -252,6 +270,7 @@ describe('SessionCache adapter integration', () => {
         isRoot: '0',
         createdAt: '1000'
       }),
+      hmset: vi.fn().mockResolvedValue('OK'),
       multi: vi.fn(),
       scan: vi.fn(),
       set: vi.fn()
@@ -282,6 +301,11 @@ describe('SessionCache adapter integration', () => {
         createdAt: 1000
       }
     });
+    await cache.updateTeam({
+      sessionId: 'user-1:token-1',
+      teamId: 'team-2',
+      tmbId: 'tmb-2'
+    });
 
     expect(commandClient.hgetall).toHaveBeenCalledWith('fastgpt:session:user-1:token-1');
     expect(multi.hmset).toHaveBeenCalledWith('fastgpt:session:user-1:token-1', {
@@ -295,5 +319,9 @@ describe('SessionCache adapter integration', () => {
       'fastgpt:session:user-1:token-1',
       SESSION_TTL_SECONDS
     );
+    expect(commandClient.hmset).toHaveBeenCalledWith('fastgpt:session:user-1:token-1', {
+      teamId: 'team-2',
+      tmbId: 'tmb-2'
+    });
   });
 });

packages/global/common/error/code/team.ts

@@ -35,7 +35,8 @@ export enum TeamErrEnum {
   invitationLinkInvalid = 'invitationLinkInvalid',
   youHaveBeenInTheTeam = 'youHaveBeenInTheTeam',
   tooManyInvitations = 'tooManyInvitations',
-  unPermission = 'unPermission'
+  unPermission = 'unPermission',
+  accountCancellationPending = 'accountCancellationPending'
 }
 
 const teamErr = [
@@ -47,6 +48,10 @@ const teamErr = [
     statusText: TeamErrEnum.unPermission,
     message: i18nT('common:error_un_permission')
   },
+  {
+    statusText: TeamErrEnum.accountCancellationPending,
+    message: i18nT('common:code_error.team_error.account_cancellation_pending')
+  },
   {
     statusText: TeamErrEnum.teamOverSize,
     message: i18nT('common:code_error.team_error.over_size')

packages/global/common/error/code/user.ts

@@ -11,6 +11,7 @@ export enum UserErrEnum {
   sendVerificationCodeTooFrequently = 'sendVerificationCodeTooFrequently',
   verifyCodeTooFrequently = 'verifyCodeTooFrequently',
   invalidAccount = 'invalidAccount',
+  accountCancellationPending = 'accountCancellationPending',
   registrationMethodNotSupported = 'registrationMethodNotSupported'
 }
 const errList = [
@@ -49,6 +50,10 @@ const errList = [
     statusText: UserErrEnum.invalidAccount,
     message: i18nT('common:code_error.invalid_account')
   },
+  {
+    statusText: UserErrEnum.accountCancellationPending,
+    message: i18nT('common:code_error.account_cancellation_pending')
+  },
   {
     statusText: UserErrEnum.registrationMethodNotSupported,
     message: i18nT('common:error.registration_method_not_supported'),

packages/global/common/middle/tracks/constants.ts

@@ -13,6 +13,9 @@ export enum TrackEnum {
   teamChatQPM = 'teamChatQPM',
   enterpriseAuthStart = 'enterpriseAuthStart',
   enterpriseAuthBenefitGrant = 'enterpriseAuthBenefitGrant',
+  accountCancellationSubmitSuccess = 'account_cancellation_submit_success',
+  accountCancellationCancelSuccess = 'account_cancellation_cancel_success',
+  accountCancellationFinalizeSuccess = 'account_cancellation_finalize_success',
 
   // Admin cron job tracks
   subscriptionDeleted = 'subscriptionDeleted',

packages/global/common/system/types/index.ts

@@ -1,4 +1,5 @@
 import type { SubPlanType } from '../../../support/wallet/sub/type';
+import type { AccountCancellationVerificationCapabilities } from '../../../support/user/account/cancellation/type';
 import type {
   LLMModelItemType,
   EmbeddingModelItemType,
@@ -75,6 +76,13 @@ export type FastGPTFeConfigsType = {
   show_enterprise_auth?: boolean;
   showWecomConfig?: boolean;
   wecomLoginAutoRedirect?: boolean;
+  accountCancellation?: {
+    enabled?: boolean;
+  };
+  /** 仅暴露注销验证的布尔能力，不包含任何 Provider 密钥。 */
+  accountVerification?: {
+    accountCancellation?: AccountCancellationVerificationCapabilities;
+  };
 
   show_dataset_feishu?: boolean;
   show_dataset_yuque?: boolean;