重构 TgGroup 相关功能，添加加入群组的接口和 DTO，更新任务执行器以支持新的 tgUser 管理逻辑，优化 Telegram 客户端的连接管理，移除不再使用的 TgClientService，提升代码一致性和可维护性。

src/app.ts

@@ -90,9 +90,9 @@ export const createApp = async () => {
   app.register(fishRoutes, { prefix: '/api/fish' })
   app.register(fishFriendsRoutes, { prefix: '/api/fish-friends' })
   app.register(messagesRoutes, { prefix: '/api/messages' })
-  app.register(tgMsgSendRoutes, { prefix: '/api/msg' })
   app.register(taskRoutes, { prefix: '/api/tasks' })
   app.register(tgUserRoutes, { prefix: '/api/tg-users' })
+  app.register(tgMsgSendRoutes, { prefix: '/api/tg-msg' })
   app.register(tgGroupRoutes, { prefix: '/api/tg-groups' })
   app.register(testRoutes, { prefix: '/api/test' })
   const dataSource = createDataSource(app)

src/controllers/tg-group.controller.ts

@@ -5,7 +5,8 @@ import {
   InviteMembersBody,
   CreateTgGroupRecordBody,
   ListTgGroupQuery,
-  UpdateTgGroupRecordBody
+  UpdateTgGroupRecordBody,
+  JoinTgGroupBody
 } from '../dto/tg-group.dto'
 
 export class TgGroupController {
@@ -56,6 +57,20 @@ export class TgGroupController {
     }
   }
 
+  async joinGroup(request: FastifyRequest<{ Body: JoinTgGroupBody }>, reply: FastifyReply) {
+    const { chatId, members } = request.body
+    try {
+      const result = await this.tgGroupService.joinGroup(chatId, members)
+      return reply.code(result.success ? 200 : 500).send(result)
+    } catch (error) {
+      return reply.code(500).send({
+        success: false,
+        message: '进群任务失败',
+        error: error instanceof Error ? error.message : String(error)
+      })
+    }
+  }
+
   async inviteMembers(request: FastifyRequest<{ Body: InviteMembersBody }>, reply: FastifyReply) {
     const { senderId, chatId, accessHash, members } = request.body
 

src/dto/tg-group.dto.ts

@@ -8,6 +8,11 @@ export interface CreateTgGroupBody {
   senderId?: string
 }
 
+export interface JoinTgGroupBody {
+  chatId: string
+  members: string[]
+}
+
 export interface InviteMembersBody {
   senderId?: string
   chatId?: string

src/executor/task.executor.ts

@@ -6,7 +6,7 @@ import { Task, TaskStatus, TaskType } from '../entities/task.entity'
 import { TaskItem, TaskItemStatus } from '../entities/task-item.entity'
 import { TgUser } from '../entities/tg-user.entity'
 import { TgUserService } from '../services/tg-user.service'
-import { TgClientService } from '../services/tgClient.service'
+import { TgClientManager } from '../services/clients/tg-client.manager'
 import { buildStringSessionByDcIdAndAuthKey } from '../utils/tg.util'
 
 export class TaskExecutor {
@@ -16,11 +16,22 @@ export class TaskExecutor {
   private senderRepo: Repository<TgUser>
   private senderService: TgUserService
 
-  private readonly defaultSenderSendLimit = 5
-  private currentSenderSendLimit = this.defaultSenderSendLimit
-  private senderUsageInBatch: Map<string, number> = new Map()
-  private senderCursor = 0
-  private senderCache: TgUser[] = []
+  // 默认账号使用上限
+  private readonly defaultAccountLimit = 5
+  // 当前账号使用上限
+  private currentAccountLimit = this.defaultAccountLimit
+  // 账号缓存默认拉取上限
+  private readonly defaultAccountCacheTake = 100
+  // 账号缓存最大拉取上限
+  private readonly maxAccountCacheTake = 1000
+  // 当前任务的账号缓存拉取上限
+  private currentAccountCacheTake = this.defaultAccountCacheTake
+  // 账号使用批次
+  private accountUsageInBatch: Map<string, number> = new Map()
+  // 账号游标
+  private accountCursor = 0
+  // 账号缓存
+  private accountCache: TgUser[] = []
 
   constructor(private app: FastifyInstance) {
     const ds = app.dataSource
@@ -38,12 +49,15 @@ export class TaskExecutor {
     try {
       await this.beforeExecute(task)
 
-      // 初始化 sender 配置
-      this.currentSenderSendLimit =
-        task.accountLimit && Number(task.accountLimit) > 0 ? Number(task.accountLimit) : this.defaultSenderSendLimit
-      this.senderUsageInBatch.clear()
-      this.senderCursor = 0
-      await this.refreshSenderCache()
+      // 初始化 tgUser 配置
+      this.currentAccountLimit =
+        task.accountLimit && Number(task.accountLimit) > 0 ? Number(task.accountLimit) : this.defaultAccountLimit
+      this.accountUsageInBatch.clear()
+      this.accountCursor = 0
+      // 计算任务需要拉取的账号池大小
+      const concurrency = Math.min(10, Math.max(1, Number(task.threads ?? 1)))
+      this.currentAccountCacheTake = this.computeAccountCacheTake(task, concurrency)
+      await this.refreshAccountCache()
 
       await this.process(task)
       await this.finalize(task.id)
@@ -53,6 +67,31 @@ export class TaskExecutor {
     }
   }
 
+  /**
+   * 计算本次任务需要拉取多少个 tgUser 账号进入缓存
+   * - 优先支持 task.payload.accountPoolLimit/accountCacheTake 覆盖
+   * - 否则使用 ceil(total / accountLimit)，并至少不小于 threads（减少并发下账号共享）
+   * - 最终受 maxAccountCacheTake 硬上限保护
+   */
+  private computeAccountCacheTake(task: Task, concurrency: number): number {
+    const maxTake = this.maxAccountCacheTake
+
+    const overrideRaw = task.payload?.accountPoolLimit ?? task.payload?.accountCacheTake
+    if (overrideRaw !== undefined && overrideRaw !== null) {
+      const override = Number(overrideRaw)
+      if (!Number.isNaN(override) && override > 0) {
+        return Math.min(maxTake, Math.max(1, Math.floor(override)))
+      }
+    }
+
+    const total = Number(task.total ?? 0)
+    const perAccount = Math.max(1, Number(this.currentAccountLimit || 1))
+    const needByTotal = total > 0 ? Math.ceil(total / perAccount) : 0
+
+    const desired = Math.max(concurrency, needByTotal || this.defaultAccountCacheTake)
+    return Math.min(maxTake, desired)
+  }
+
   /**
    * 执行前校验 & 标记
    */
@@ -88,9 +127,9 @@ export class TaskExecutor {
    * 单 worker 循环
    */
   private async workerLoop(taskId: number): Promise<void> {
-    let sender: TgUser | null = null
-    const workerTgClient = new TgClientService()
-    let senderSentInRound = 0
+    let tgUser: TgUser | null = null
+    const workerTgClient = new TgClientManager()
+    let accountUsageInRound = 0
     let inviteGroupEntity: any | null = null
 
     try {
@@ -114,42 +153,33 @@ export class TaskExecutor {
           return
         }
 
-        // Sender 轮换逻辑
-        if (!sender || senderSentInRound >= this.currentSenderSendLimit) {
+        // tgUser 轮换逻辑
+        if (!tgUser || accountUsageInRound >= this.currentAccountLimit) {
           await workerTgClient.disconnect()
-          sender = await this.pickSender()
-          const sessionString = await this.ensureSessionString(sender)
+          tgUser = await this.pickAccount()
+          const sessionString = await this.ensureSessionString(tgUser)
 
           try {
             await workerTgClient.connect(sessionString)
           } catch (error) {
             const msg = error instanceof Error ? error.message : String(error)
             if (this.isSessionRevokedMessage(msg)) {
-              await this.handleSessionRevoked(sender)
-              sender = null
+              await this.handleSessionRevoked(tgUser)
+              tgUser = null
               continue
             }
             throw error
           }
 
-          senderSentInRound = 0
+          accountUsageInRound = 0
           inviteGroupEntity = null
 
-          // 获取当前账号信息并延迟
-          const me = await workerTgClient
-            .getClient()
-            ?.getMe()
-            .catch(() => null)
+          // 延迟
           const delaySeconds = this.getRandomDelaySeconds()
-          const displayName = `${me?.firstName ?? ''} ${me?.lastName ?? ''}`.trim() || me?.username || ''
-          this.app.log.info(
-            `当前登录账号: id: ${me?.id ?? sender.id}, name: ${
-              displayName || sender.id
-            }，延迟 ${delaySeconds}s 后开始发送`
-          )
+          this.app.log.info(`延迟 ${delaySeconds}s 后开始处理任务`)
           await this.sleep(delaySeconds * 1000)
 
-          // 邀请任务：每次换号后，确保已加入目标群并拿到群实体
+          // 群拉人任务：每次换号后，确保已加入目标群并拿到群实体
           if (task.type === TaskType.INVITE_TO_GROUP) {
             const inviteLink = String(task.payload?.inviteLink ?? '').trim()
             if (!inviteLink) {
@@ -160,20 +190,20 @@ export class TaskExecutor {
         }
 
         try {
-          await this.processTaskItem(task, taskItem, sender, workerTgClient, inviteGroupEntity)
+          await this.processTaskItem(task, taskItem, tgUser, workerTgClient, inviteGroupEntity)
         } catch (error) {
           const msg = error instanceof Error ? error.message : '未知错误'
-          if (sender && this.isSessionRevokedMessage(msg)) {
-            await this.handleSessionRevoked(sender)
+          if (tgUser && this.isSessionRevokedMessage(msg)) {
+            await this.handleSessionRevoked(tgUser)
             await workerTgClient.disconnect()
-            sender = null
-            senderSentInRound = 0
+            tgUser = null
+            accountUsageInRound = 0
           }
         } finally {
-          senderSentInRound++
-          if (sender) {
-            const used = (this.senderUsageInBatch.get(sender.id) ?? 0) + 1
-            this.senderUsageInBatch.set(sender.id, used)
+          accountUsageInRound++
+          if (tgUser) {
+            const used = (this.accountUsageInBatch.get(tgUser.id) ?? 0) + 1
+            this.accountUsageInBatch.set(tgUser.id, used)
           }
         }
 
@@ -233,7 +263,7 @@ export class TaskExecutor {
     task: Task,
     item: TaskItem,
     sender: TgUser,
-    workerTgClient: TgClientService,
+    workerTgClient: TgClientManager,
     inviteGroupEntity: any | null
   ): Promise<void> {
     if (task.type === TaskType.INVITE_TO_GROUP) {
@@ -247,7 +277,7 @@ export class TaskExecutor {
     task: Task,
     item: TaskItem,
     sender: TgUser,
-    workerTgClient: TgClientService
+    workerTgClient: TgClientManager
   ): Promise<void> {
     const message = String(task.payload?.message ?? '').trim()
     if (!message) {
@@ -314,7 +344,7 @@ export class TaskExecutor {
     task: Task,
     item: TaskItem,
     sender: TgUser,
-    workerTgClient: TgClientService,
+    workerTgClient: TgClientManager,
     inviteGroupEntity: any | null
   ): Promise<void> {
     try {
@@ -471,74 +501,76 @@ export class TaskExecutor {
   }
 
   /**
-   * 刷新发送者缓存
+   * 刷新 tgUser 账号缓存
    */
-  private async refreshSenderCache(): Promise<void> {
-    this.senderCache = await this.senderRepo.find({
+  private async refreshAccountCache(): Promise<void> {
+    this.accountCache = await this.senderRepo.find({
       where: { delFlag: false },
-      order: { lastUsageTime: 'ASC', usageCount: 'ASC' }
+      order: { lastUsageTime: 'ASC', usageCount: 'ASC' },
+      take: this.currentAccountCacheTake
     })
-    this.senderCursor = 0
+    this.accountCursor = 0
   }
 
   /**
-   * 选择可用的发送者
+   * 选择可用的tgUser 账号
    */
-  private async pickSender(): Promise<TgUser> {
-    if (this.senderCache.length === 0) {
-      this.senderCache = await this.senderRepo.find({
+  private async pickAccount(): Promise<TgUser> {
+    if (this.accountCache.length === 0) {
+      this.accountCache = await this.senderRepo.find({
         where: { delFlag: false },
-        order: { lastUsageTime: 'ASC', usageCount: 'ASC' }
+        order: { lastUsageTime: 'ASC', usageCount: 'ASC' },
+        take: this.currentAccountCacheTake
       })
-      this.senderCursor = 0
+      this.accountCursor = 0
     }
 
-    if (this.senderCache.length === 0) {
-      throw new Error('暂无可用 sender 账号')
+    if (this.accountCache.length === 0) {
+      throw new Error('暂无可用 tgUser 账号')
     }
 
-    const total = this.senderCache.length
+    const total = this.accountCache.length
     for (let i = 0; i < total; i++) {
-      const index = (this.senderCursor + i) % total
-      const sender = this.senderCache[index]
-      const used = this.senderUsageInBatch.get(sender.id) ?? 0
-      if (used < this.currentSenderSendLimit) {
-        this.senderCursor = (index + 1) % total
-        return sender
+      const index = (this.accountCursor + i) % total
+      const account = this.accountCache[index]
+      const used = this.accountUsageInBatch.get(account.id) ?? 0
+      if (used < this.currentAccountLimit) {
+        this.accountCursor = (index + 1) % total
+        return account
       }
     }
 
-    this.app.log.info('所有 sender 均已达到当前批次上限，重置计数后重新轮询')
-    this.senderUsageInBatch.clear()
-    this.senderCursor = 0
-    return this.senderCache[0]
+    this.app.log.info('所有 tgUser 均已达到当前批次上限，重置计数后重新轮询')
+    this.accountUsageInBatch.clear()
+    this.accountCursor = 0
+    return this.accountCache[0]
   }
 
   /**
    * 确保发送者有有效的会话字符串
    */
-  private async ensureSessionString(sender: TgUser): Promise<string> {
-    if (sender.sessionStr) {
-      return sender.sessionStr
+  private async ensureSessionString(tgUser: TgUser): Promise<string> {
+    if (tgUser.sessionStr) {
+      return tgUser.sessionStr
     }
 
-    if (sender.dcId && sender.authKey) {
-      const session = buildStringSessionByDcIdAndAuthKey(sender.dcId, sender.authKey)
-      await this.senderRepo.update(sender.id, { sessionStr: session })
+    if (tgUser.dcId && tgUser.authKey) {
+      const session = buildStringSessionByDcIdAndAuthKey(tgUser.dcId, tgUser.authKey)
+      await this.senderRepo.update(tgUser.id, { sessionStr: session })
       return session
     }
 
-    throw new Error(`sender=${sender.id} 缺少 session 信息`)
+    throw new Error(`tgUser=${tgUser.id} 缺少 session 信息`)
   }
 
   /**
    * 处理会话被撤销的情况
    */
-  private async handleSessionRevoked(sender: TgUser): Promise<void> {
-    await this.senderRepo.update(sender.id, { delFlag: true })
-    this.senderCache = this.senderCache.filter(s => s.id !== sender.id)
-    this.senderCursor = 0
-    this.app.log.warn(`sender=${sender.id} session 失效，已删除`)
+  private async handleSessionRevoked(tgUser: TgUser): Promise<void> {
+    await this.senderRepo.update(tgUser.id, { delFlag: true })
+    this.accountCache = this.accountCache.filter(a => a.id !== tgUser.id)
+    this.accountCursor = 0
+    this.app.log.warn(`tgUser=${tgUser.id} session 失效，已删除`)
   }
 
   /**

src/routes/tg-group.routes.ts

@@ -5,6 +5,7 @@ import { UserRole } from '../entities/user.entity'
 import {
   CreateTgGroupBody,
   CreateTgGroupRecordBody,
+  JoinTgGroupBody,
   ListTgGroupQuery,
   UpdateTgGroupRecordBody
 } from '../dto/tg-group.dto'
@@ -14,6 +15,8 @@ export default async function tgGroupRoutes(fastify: FastifyInstance) {
 
   fastify.post<{ Body: CreateTgGroupBody }>('/create', controller.createGroup.bind(controller))
 
+  fastify.post<{ Body: JoinTgGroupBody }>('/join', controller.joinGroup.bind(controller))
+
   fastify.put<{ Body: UpdateTgGroupRecordBody }>(
     '/',
     { onRequest: [hasRole(UserRole.ADMIN)] },

src/services/tgClient.service.ts → src/services/clients/tg-client.manager.ts

@@ -1,9 +1,9 @@
 import { Api, TelegramClient } from 'telegram'
 import { StringSession } from 'telegram/sessions'
-import { createApp } from '../app'
+import { createApp } from '../../app'
 import bigInt from 'big-integer'
 
-export class TgClientService {
+export class TgClientManager {
   private app: any
   private client: TelegramClient | null = null
   private apiId: number
@@ -64,7 +64,7 @@ export class TgClientService {
    * @throws 当连接数达到上限或连接失败时抛出错误
    */
   async connect(sessionString: string): Promise<void> {
-    if (TgClientService.activeClientsCount >= TgClientService.maxActiveClients) {
+    if (TgClientManager.activeClientsCount >= TgClientManager.maxActiveClients) {
       throw new Error('TelegramClient 并发连接数已达上限')
     }
 
@@ -100,7 +100,7 @@ export class TgClientService {
         }
 
         this.client = client
-        TgClientService.activeClientsCount++
+        TgClientManager.activeClientsCount++
         this.logActiveClientCount()
         return
       } catch (error) {
@@ -170,7 +170,7 @@ export class TgClientService {
   ): Promise<{ chatId: string; accessHash: string }> {
     this.ensureConnected()
 
-    const waitTime = Math.floor(Math.random() * 21) + 20
+    const waitTime = Math.floor(Math.random() * 21) + 5
     this.app.log.info(`连接成功后等待 ${waitTime} 秒，避免新 Session 被限制`)
     await new Promise(resolve => setTimeout(resolve, waitTime * 1000))
 
@@ -200,6 +200,21 @@ export class TgClientService {
 
     this.app.log.info(`群组创建成功，ID: ${chatId}, accessHash: ${accessHash}`)
 
+    // 创建超级群后补充权限：允许群成员添加/邀请用户（inviteUsers）
+    if (groupType === 'megagroup') {
+      const inputChannel = await this.getInputChannel(String(chatId), String(accessHash))
+      const full = await this.client!.invoke(new Api.channels.GetFullChannel({ channel: inputChannel }))
+      const current = (full as any)?.chats?.[0]?.defaultBannedRights ?? {}
+      const { inviteUsers: _ignoreInviteUsers, ...rest } = current as any
+      await this.client!.invoke(
+        new Api.messages.EditChatDefaultBannedRights({
+          peer: inputChannel,
+          bannedRights: new Api.ChatBannedRights({ ...rest, inviteUsers: false, untilDate: 0 } as any)
+        })
+      )
+      this.app.log.info('群组权限已更新：允许群成员添加用户')
+    }
+
     return {
       chatId: String(chatId),
       accessHash: String(accessHash)
@@ -221,12 +236,16 @@ export class TgClientService {
 
   /**
    * 通过邀请链接加入群组
-   * @param inviteLink - Telegram 邀请链接
+   * @param inviteLink
    * @returns 返回加入的群组信息，包含 chatId 和 title
    * @throws 当邀请链接无效、过期或加入失败时抛出错误
    */
   async joinGroupByInviteLink(inviteLink: string): Promise<{ chatId: string; title: string }> {
     this.ensureConnected()
+
+    const waitTime = Math.floor(Math.random() * 21) + 5
+    this.app.log.info(`连接成功后等待 ${waitTime} 秒，避免新 Session 被限制`)
+    await new Promise(resolve => setTimeout(resolve, waitTime * 1000))
     this.app.log.info(`正在通过邀请链接加入群组: ${inviteLink}`)
 
     try {
@@ -236,48 +255,11 @@ export class TgClientService {
         throw new Error('无效的邀请链接格式')
       }
 
-      // 检查邀请链接信息（如果已在群内，直接返回 chat 信息）
-      const check = await this.client!.invoke(
-        new Api.messages.CheckChatInvite({
-          hash
-        })
-      )
-
-      const checkAny = check as any
-      // ChatInviteAlready: 已经是成员，会直接带 chat
-      if (checkAny?.className === 'ChatInviteAlready' && checkAny?.chat) {
-        const chat = checkAny.chat
-        const chatId = chat.id?.toString() || chat.id
-        const title = chat.title || '未知群组'
-        this.app.log.info(`已是群组成员: ${title} (ID: ${chatId})`)
-        return { chatId: String(chatId), title }
-      }
-
-      // 使用邀请链接加入群组
-      const result = await this.client!.invoke(
-        new Api.messages.ImportChatInvite({
-          hash
-        })
-      )
+      const result = await this.client!.invoke(new Api.messages.ImportChatInvite({ hash }))
 
       const updates = result as any
       let chat = updates.chats?.[0]
 
-      if (!chat) {
-        // 如果是 Updates 类型，尝试从 updates 中获取
-        if (updates.updates) {
-          for (const update of updates.updates) {
-            if (update.className === 'UpdateNewMessage' && update.message?.peerId) {
-              const chatId = update.message.peerId.channelId || update.message.peerId.chatId
-              if (chatId && updates.chats) {
-                chat = updates.chats.find((c: any) => c.id?.toString() === chatId?.toString())
-                break
-              }
-            }
-          }
-        }
-      }
-
       if (!chat) {
         throw new Error('加入群组失败，未返回群组信息')
       }
@@ -285,8 +267,6 @@ export class TgClientService {
       const chatId = chat.id?.toString() || chat.id
       const title = chat.title || '未知群组'
 
-      this.app.log.info(`成功加入群组: ${title} (ID: ${chatId})`)
-
       return {
         chatId: String(chatId),
         title
@@ -298,39 +278,18 @@ export class TgClientService {
         errorMessage = '邀请链接已过期'
       } else if (errorMessage.includes('INVITE_HASH_INVALID')) {
         errorMessage = '邀请链接无效'
-      } else if (errorMessage.includes('USER_ALREADY_PARTICIPANT')) {
-        errorMessage = '您已经是该群组的成员'
       } else if (errorMessage.includes('CHANNELS_TOO_MUCH')) {
         errorMessage = '加入的频道/群组数量已达上限'
       } else if (errorMessage.includes('INVITE_REQUEST_SENT')) {
         errorMessage = '已发送加入请求，等待管理员批准'
+      } else if (errorMessage.includes('USER_ALREADY_PARTICIPANT')) {
+        errorMessage = '您已经是该群组的成员'
       }
 
       throw new Error(`通过邀请链接加入群组失败: ${errorMessage}`)
     }
   }
 
-  /**
-   * 通过邀请链接获取群组实体（会自动加入或复用已加入状态）
-   * - 用于“拉人进群”任务：需要拿到 chat/channel 实体，进而拿到 accessHash
-   */
-  async resolveGroupEntityByInviteLink(inviteLink: string): Promise<any> {
-    this.ensureConnected()
-    const joined = await this.joinGroupByInviteLink(inviteLink)
-
-    // 通过 chatId 获取实体；部分场景需要 -100 前缀重试（频道/超级群）
-    try {
-      return await this.client!.getEntity(joined.chatId)
-    } catch (err) {
-      const chatIdStr = String(joined.chatId)
-      if (!chatIdStr.startsWith('-100')) {
-        const prefixed = `-100${chatIdStr}`
-        return await this.client!.getEntity(prefixed)
-      }
-      throw err
-    }
-  }
-
   /**
    * 退出指定的频道或群组
    * @param target - 群组标识，可以是 chatId、username 或 InputChannel 对象
@@ -394,6 +353,27 @@ export class TgClientService {
     }
   }
 
+  /**
+   * 通过邀请链接获取群组实体
+   * - 用于“拉人进群”任务：需要拿到 chat/channel 实体，进而拿到 accessHash
+   */
+  async resolveGroupEntityByInviteLink(inviteLink: string): Promise<any> {
+    this.ensureConnected()
+    const joined = await this.joinGroupByInviteLink(inviteLink)
+
+    // 通过 chatId 获取实体；部分场景需要 -100 前缀重试（频道/超级群）
+    try {
+      return await this.client!.getEntity(joined.chatId)
+    } catch (err) {
+      const chatIdStr = String(joined.chatId)
+      if (!chatIdStr.startsWith('-100')) {
+        const prefixed = `-100${chatIdStr}`
+        return await this.client!.getEntity(prefixed)
+      }
+      throw err
+    }
+  }
+
   /**
    * 获取群组的邀请链接
    * @param inputChannel - InputChannel 对象
@@ -652,7 +632,7 @@ export class TgClientService {
     }
 
     if (decrementCount) {
-      TgClientService.activeClientsCount = Math.max(0, TgClientService.activeClientsCount - 1)
+      TgClientManager.activeClientsCount = Math.max(0, TgClientManager.activeClientsCount - 1)
       this.logActiveClientCount()
     }
   }
@@ -845,6 +825,6 @@ export class TgClientService {
    * @private
    */
   private logActiveClientCount(): void {
-    this.app?.log?.info?.(`当前活跃 Telegram 客户端数量: ${TgClientService.activeClientsCount}`)
+    this.app?.log?.info?.(`当前活跃 Telegram 客户端数量: ${TgClientManager.activeClientsCount}`)
   }
 }

src/services/test.service.ts

@@ -7,7 +7,7 @@ import { Task, TaskStatus } from '../entities/task.entity'
 import { TaskItem, TaskItemStatus } from '../entities/task-item.entity'
 import { TgUser } from '../entities/tg-user.entity'
 import { TgUserService } from './tg-user.service'
-import { TgClientService } from './tgClient.service'
+import { TgClientManager } from './clients/tg-client.manager'
 import { TgGroupService } from './tg-group.service'
 import { buildStringSession, buildStringSessionByDcIdAndAuthKey } from '../utils/tg.util'
 
@@ -17,7 +17,7 @@ export class TestService {
   private readonly taskItemRepository: Repository<TaskItem>
   private readonly senderRepository: Repository<TgUser>
   private readonly senderService: TgUserService
-  private readonly tgClientService: TgClientService
+  private readonly tgClientService: TgClientManager
   private readonly tgGroupService: TgGroupService
   private senderSendLimit = 5
   private senderUsageInBatch: Map<string, number> = new Map()
@@ -30,7 +30,7 @@ export class TestService {
     this.taskItemRepository = app.dataSource.getRepository(TaskItem)
     this.senderRepository = app.dataSource.getRepository(TgUser)
     this.senderService = new TgUserService(app)
-    this.tgClientService = new TgClientService()
+    this.tgClientService = new TgClientManager()
     this.tgGroupService = new TgGroupService(app)
   }
 
@@ -113,7 +113,7 @@ export class TestService {
         const sender = await pickSender()
         const sessionString = await this.ensureSessionStringForSender(sender)
 
-        const senderTgClient = new TgClientService()
+        const senderTgClient = new TgClientManager()
         const connectWithTimeout = async () => {
           const timeoutMs = 10_000
           return Promise.race([

src/services/tg-group.service.ts

@@ -2,24 +2,18 @@ import { FastifyInstance } from 'fastify'
 import { Repository } from 'typeorm'
 import { TgUser } from '../entities/tg-user.entity'
 import { TgGroup, TgGroupType } from '../entities/tg-group.entity'
-import { TgClientService } from './tgClient.service'
+import { TgClientManager } from './clients/tg-client.manager'
 import { buildStringSessionByDcIdAndAuthKey } from '../utils/tg.util'
 import { TelegramClient } from 'telegram'
 import { PaginationResponse } from '../dto/common.dto'
-import {
-  CreateTgGroupRecordBody,
-  ListTgGroupQuery,
-  UpdateTgGroupRecordBody
-} from '../dto/tg-group.dto'
+import { CreateTgGroupRecordBody, ListTgGroupQuery, UpdateTgGroupRecordBody } from '../dto/tg-group.dto'
 
 export class TgGroupService {
   private readonly app: FastifyInstance
-  private readonly tgClientService: TgClientService
   private readonly tgGroupRepository: Repository<TgGroup>
   private readonly senderRepository: Repository<TgUser>
   constructor(app: FastifyInstance) {
     this.app = app
-    this.tgClientService = new TgClientService()
     this.tgGroupRepository = app.dataSource.getRepository(TgGroup)
     this.senderRepository = app.dataSource.getRepository(TgUser)
   }
@@ -39,7 +33,7 @@ export class TgGroupService {
     senderId?: string
   ): Promise<any> {
     let sender: TgUser | null = null
-    let client: TelegramClient | null = null
+    let tgClient: TgClientManager | null = null
 
     try {
       if (senderId) {
@@ -87,20 +81,21 @@ export class TgGroupService {
         }
       }
 
-      await this.tgClientService.connect(sender.sessionStr)
+      tgClient = new TgClientManager()
+      await tgClient.connect(sender.sessionStr)
 
-      const groupInfo = await this.tgClientService.createChannelGroup(
+      const groupInfo = await tgClient.createChannelGroup(
         groupName,
         groupDescription || '',
         groupType as 'megagroup' | 'channel'
       )
 
-      const inputChannel = await this.tgClientService.getInputChannel(groupInfo.chatId, groupInfo.accessHash)
-      const inviteLink = await this.tgClientService.getInviteLink(inputChannel)
-      const publicLink = await this.tgClientService.getPublicLink(inputChannel, groupName)
+      const inputChannel = await tgClient.getInputChannel(groupInfo.chatId, groupInfo.accessHash)
+      const inviteLink = await tgClient.getInviteLink(inputChannel)
+      const publicLink = await tgClient.getPublicLink(inputChannel, groupName)
 
       if (initMsg && initMsg.trim().length > 0) {
-        await this.tgClientService.sendMessageToChannelGroup(inputChannel, initMsg.trim())
+        await tgClient.sendMessageToChannelGroup(inputChannel, initMsg.trim())
       }
 
       await this.upsertGroup({
@@ -140,7 +135,83 @@ export class TgGroupService {
         message: `创建群组失败: ${errorMessage}`
       }
     } finally {
-      await this.tgClientService.disconnect()
+      if (tgClient) {
+        await tgClient.disconnect().catch(() => {})
+      }
+    }
+  }
+
+  async joinGroup(chatId: string, members: string[]): Promise<any> {
+    const group = await this.findByChatId(chatId)
+    if (!group) {
+      return {
+        success: false,
+        message: `群组不存在: ${chatId}`
+      }
+    }
+
+    // 通过 inviteLink 加入群组
+    const inviteLink = group.inviteLink
+    if (!inviteLink) {
+      return {
+        success: false,
+        message: `群组 ${chatId} 没有邀请链接`
+      }
+    }
+    // 使用 members 中的用户 id 加入群组
+    const results: Array<{ memberId: string; success: boolean; message: string; data?: any }> = []
+    for (const member of members) {
+      const user = await this.senderRepository.findOne({ where: { id: member } })
+      if (!user) {
+        this.app.log.error(`用户不存在: ${member}`)
+        results.push({ memberId: member, success: false, message: '用户不存在' })
+        continue
+      }
+      if (!user.sessionStr) {
+        this.app.log.error(`用户 ${member} 缺少 session`)
+        results.push({ memberId: String(user.id), success: false, message: '缺少 session' })
+        continue
+      }
+      let joinClient: TgClientManager | null = null
+      try {
+        joinClient = new TgClientManager()
+        await joinClient.connect(user.sessionStr)
+        const joined = await joinClient.joinGroupByInviteLink(inviteLink)
+        results.push({
+          memberId: String(user.id),
+          success: true,
+          message: '加入成功',
+          data: joined
+        })
+        await new Promise(resolve => setTimeout(resolve, 2000))
+      } catch (error) {
+        const msg = error instanceof Error ? error.message : String(error)
+        this.app.log.error(`账户 ${user.id} 加入群组失败: ${msg}`)
+        results.push({
+          memberId: String(user.id),
+          success: false,
+          message: msg
+        })
+      } finally {
+        if (joinClient) {
+          await joinClient.disconnect().catch(() => {})
+        }
+      }
+    }
+
+    const successCount = results.filter(r => r.success).length
+    const failCount = results.length - successCount
+    return {
+      success: failCount === 0,
+      message: `进群任务完成: 成功 ${successCount}，失败 ${failCount}`,
+      data: {
+        chatId,
+        inviteLink,
+        total: results.length,
+        successCount,
+        failCount,
+        results
+      }
     }
   }
 
@@ -149,7 +220,7 @@ export class TgGroupService {
   async upsertGroup(payload: CreateTgGroupRecordBody): Promise<TgGroup> {
     const existing = await this.tgGroupRepository.findOne({ where: { chatId: payload.chatId } })
     const normalizedType = this.normalizeGroupType(payload.groupType)
-    
+
     if (existing) {
       // 更新现有记录
       Object.assign(existing, {

src/services/tg-msg-send.service.ts

@@ -1,14 +1,14 @@
 import { FastifyInstance } from 'fastify'
 import { SendMessageResult } from '../dto/tg-msg-send.dto'
-import { TgClientService } from './tgClient.service'
+import { TgClientManager } from './clients/tg-client.manager'
 
 export class TgMsgSendService {
   private app: FastifyInstance
-  private tgClientService: TgClientService
+  private tgClientService: TgClientManager
 
   constructor(app: FastifyInstance) {
     this.app = app
-    this.tgClientService = new TgClientService()
+    this.tgClientService = new TgClientManager()
   }
 
   /**