.

package.json

@@ -46,6 +46,7 @@
     "@nestjs/websockets": "^10.3.8",
     "@types/ws": "^8.5.10",
     "ali-oss": "^6.17.1",
+    "async-lock": "^1.4.1",
     "axios": "^1.3.6",
     "bcrypt": "^5.1.0",
     "big.js": "^6.2.1",
@@ -94,6 +95,7 @@
     "@nestjs/cli": "^9.2.0",
     "@nestjs/schematics": "^9.0.4",
     "@nestjs/testing": "^9.3.3",
+    "@types/async-lock": "^1.4.2",
     "@types/express": "^4.17.17",
     "@types/jest": "^29.4.0",
     "@types/multer": "^1.4.7",

src/balance/balance.service.ts

@@ -82,8 +82,9 @@ export class BalanceService {
         }
     }
 
-    async feeDeduction(userId: number, cost: number, users: Users) {
+    async feeDeduction(userId: number, cost: number) {
         try {
+            const user = await this.usersService.findById(userId)
             // 获取余额
             let balance = await this.balanceRepository.findOne({
                 where: {
@@ -107,8 +108,8 @@ export class BalanceService {
             record.type = BalanceType.CONSUMPTION
             await this.recordRepository.save(record)
 
-            users.balance = balance.currentBalance
-            await this.userRepository.save(users)
+            user.balance = balance.currentBalance
+            await this.userRepository.save(user)
 
             return 'Deduction success!'
         } catch (e) {

src/device/device.service.ts

@@ -3,7 +3,7 @@ import { PageRequest } from '../common/dto/page-request'
 import { Device } from './entities/device.entity'
 import { Pagination, paginate } from 'nestjs-typeorm-paginate'
 import { InjectRepository } from '@nestjs/typeorm'
-import { Repository } from 'typeorm'
+import { Like, Repository } from 'typeorm'
 
 @Injectable()
 export class DeviceService {
@@ -48,13 +48,27 @@ export class DeviceService {
         const device = await this.deviceRepository.findOneBy({ socketId })
         if (device) {
             Object.assign(device, data)
-            console.log(device)
             await this.deviceRepository.save(device)
         }
     }
 
-    async findAvailableDevice() {
-        return await this.deviceRepository.findOneBy({ online: true, busy: false, canSend: true })
+    async findAvailableDevice(matchDevice?: string) {
+        const where: any = { online: true, busy: false, canSend: true }
+        if (matchDevice) {
+            where.name = Like(`%${matchDevice}%`)
+        }
+        return await this.deviceRepository.findOneBy(where)
+    }
+
+    async findAvailableDevices(num = 100, matchDevice?: string) {
+        const where: any = { online: true, busy: false, canSend: true }
+        if (matchDevice) {
+            where.name = Like(`%${matchDevice}%`)
+        }
+        return await this.deviceRepository.find({
+            where,
+            take: num
+        })
     }
 
     async setBusy(id: string, busy: boolean) {

src/task/entities/task.entity.ts

@@ -62,4 +62,10 @@ export class Task {
 
     @Column({ default: true })
     checkConnection: boolean
+    
+    @Column({ default: false })
+    paid: boolean
+
+    @Column({nullable: true})
+    matchDevice?: string
 }

src/task/task.service.ts

@@ -19,72 +19,48 @@ import { BalanceService } from '../balance/balance.service'
 import Decimal from 'decimal.js'
 import { Role } from '../model/role.enum'
 import { Phone } from '../phone-list/entities/phone.entity'
-import { ta } from 'date-fns/locale'
-
+import { Interval } from '@nestjs/schedule'
+import * as AsyncLock from 'async-lock'
+import { UsersService } from 'src/users/users.service'
 @Injectable()
 export class TaskService implements OnModuleInit {
+    private lock = new AsyncLock()
+    private TAG = 'TaskService'
     constructor(
         @InjectRepository(Task)
         private taskRepository: Repository<Task>,
         @InjectRepository(TaskItem)
         private taskItemRepository: Repository<TaskItem>,
-        @InjectRepository(Users)
-        private userRepository: Repository<Users>,
         @InjectRepository(Phone)
         private phoneRepository: Repository<Phone>,
-        @InjectRepository(Balance)
-        private balanceRepository: Repository<Balance>,
         @Inject(forwardRef(() => EventsGateway))
         private readonly eventsGateway: EventsGateway,
         private readonly phoneListService: PhoneListService,
         private readonly deviceService: DeviceService,
         private readonly sysConfigService: SysConfigService,
-        private readonly balanceService: BalanceService
+        private readonly balanceService: BalanceService,
+        private readonly userService: UsersService
     ) {
     }
 
     onModuleInit() {
-        this.taskRepository.update({ status: TaskStatus.PENDING }, { status: TaskStatus.IDLE })
-        this.taskItemRepository.update({ status: TaskStatus.PENDING }, { status: TaskItemStatus.IDLE })
+        this.lock.acquire('dispatchTask', async () => {
+            await this.taskItemRepository.update(
+                {
+                    status: TaskItemStatus.PENDING
+                },
+                {
+                    status: TaskItemStatus.IDLE
+                }
+            )
+            await setTimeout(10000)
+        })
     }
 
     private taskControllers: { [key: number]: AbortController } = {}
 
     async findAllTask(req: PageRequest<Task>): Promise<Pagination<Task>> {
-        const result = await paginate<Task>(this.taskRepository, req.page, req.search)
-
-        const taskItems = result.items
-        if (taskItems.length > 0) {
-            for (const task of taskItems) {
-                if (
-                    task.status === TaskStatus.PENDING ||
-                    (task.status === TaskStatus.COMPLETED &&
-                        moment(task.createdAt).isSameOrAfter(moment().subtract(2, 'hours')))
-                ) {
-                    const id = task.id
-                    const successCount = await this.taskItemRepository.countBy({
-                        taskId: id,
-                        status: 'success'
-                    })
-                    const totalCount = await this.taskItemRepository.countBy({
-                        taskId: id
-                    })
-                    if (totalCount === 0) {
-                        throw new Error('No tasks found for the given taskId.')
-                    }
-                    // 计算成功率
-                    const successRate = ((successCount / totalCount) * 100).toFixed(1) + '%'
-                    task.successRate = String(successRate)
-                    const sendCount = await this.taskItemRepository.countBy({
-                        taskId: id,
-                        status: In(['success', 'fail'])
-                    })
-                    task.sent = sendCount
-                    await this.taskRepository.save(task)
-                }
-            }
-        }
-        return result
+        return await paginate<Task>(this.taskRepository, req.page, req.search)
     }
 
     async findAllTaskItem(req: PageRequest<TaskItem>): Promise<Pagination<TaskItem>> {
@@ -101,32 +77,35 @@ export class TaskService implements OnModuleInit {
             dynamicMessageList = task.dynamicMessage.split(',')
         }
         task = await this.taskRepository.save(task)
-        await this.taskItemRepository.save(
-            phones.map((phone) => {
-                const taskItem = new TaskItem()
-                taskItem.taskId = task.id
-                taskItem.number = phone.number
-                if (dynamicMessageList !== null && task.message.includes('[#random#]')) {
-                    taskItem.message = task.message.replace(
-                        '[#random#]',
-                        dynamicMessageList[Math.floor(Math.random() * dynamicMessageList.length)]
-                    )
-                } else {
-                    taskItem.message = task.message
-                }
-                taskItem.status = TaskStatus.IDLE
-                return taskItem
-            })
+
+        await this.phoneRepository.manager.insert(
+            TaskItem,
+            this.taskItemRepository.create(
+                phones.map((phone) => {
+                    const taskItem = new TaskItem()
+                    taskItem.taskId = task.id
+                    taskItem.number = phone.number
+                    if (dynamicMessageList !== null && task.message.includes('[#random#]')) {
+                        taskItem.message = task.message.replace(
+                            '[#random#]',
+                            dynamicMessageList[Math.floor(Math.random() * dynamicMessageList.length)]
+                        )
+                    } else {
+                        taskItem.message = task.message
+                    }
+                    taskItem.status = TaskStatus.IDLE
+                    return taskItem
+                })
+            )
         )
+
         return task
     }
 
     async balanceVerification(id: number) {
         const task = await this.taskRepository.findOneBy({ id })
         // 获取用户信息
-        const user = await this.userRepository.findOneBy({
-            id: task.userId
-        })
+        const user = await this.userService.findById(task.userId)
         if (user.roles.includes(Role.Admin)) {
             return 0
         }
@@ -144,7 +123,24 @@ export class TaskService implements OnModuleInit {
     }
 
     async startTask(id: number): Promise<void> {
-        const task = await this.taskRepository.findOneBy({ id })
+        const task = await this.taskRepository.findOneOrFail({
+            where: { id }
+        })
+
+        if (task.status !== TaskStatus.IDLE) return
+
+        if (!task.paid) {
+            const user = await this.userService.findById(task.userId)
+            if (!user.roles.includes(Role.Admin)) {
+                // 开始任务前扣费
+                const cost = await this.getCost(task, user)
+                if (cost > (user.balance || 0)) {
+                    throw new Error('Insufficient balance!')
+                }
+                await this.balanceService.feeDeduction(task.userId, cost)
+                await this.taskRepository.update({ id }, { paid: true })
+            }
+        }
 
         // 查询当前是否有任务执行
         const num = await this.taskRepository.count({
@@ -152,81 +148,14 @@ export class TaskService implements OnModuleInit {
                 status: TaskStatus.PENDING
             }
         })
-        // 当前有任务
-        if (num > 0 && task.status === TaskStatus.IDLE) {
-            // 当前任务进入队列
-            task.status = TaskStatus.QUEUED
-            await this.taskRepository.save(task)
-            return
-        }
 
-        // 队列没有任务或队列中下一个待执行任务,启动此任务
-        if (
-            (task && task.status === TaskStatus.IDLE) ||
-            task.status === TaskStatus.PAUSE ||
-            task.status === TaskStatus.QUEUED
-        ) {
-            if (task.status !== TaskStatus.PAUSE) {
-                const user = await this.userRepository.findOneBy({
-                    id: task.userId
-                })
-                if (!user.roles.includes(Role.Admin)) {
-                    // 开始任务前扣费
-                    const cost = await this.getCost(task, user)
-                    if (cost > (user.balance || 0)) {
-                        throw new Error('Insufficient balance!')
-                    }
-                    await this.balanceService.feeDeduction(task.userId, cost, user)
-                }
-            }
-
-            task.status = TaskStatus.PENDING
-            await this.taskRepository.save(task)
-            await this.runTask(task)
-
-            const newTask = await this.taskRepository.findOneBy({ id })
-            if ([TaskStatus.COMPLETED, TaskStatus.PAUSE].includes(newTask.status)) {
-                try {
-                    const successCount = await this.taskItemRepository.countBy({
-                        taskId: id,
-                        status: 'success'
-                    })
-                    const totalCount = await this.taskItemRepository.countBy({
-                        taskId: id
-                    })
-                    if (totalCount === 0) {
-                        throw new Error('No tasks found for the given taskId.')
-                    }
-                    // 计算成功率
-                    const successRate = ((successCount / totalCount) * 100).toFixed(1) + '%'
-                    newTask.successRate = String(successRate)
-                    const sendCount = await this.taskItemRepository.countBy({
-                        taskId: id,
-                        status: In(['success', 'fail'])
-                    })
-                    newTask.sent = sendCount
-                    await this.taskRepository.save(newTask)
-                } catch (e) {
-                    Logger.error('Error startTask ', e, 'RcsService')
-                }
-            }
-        }
+        await this.taskRepository.update({ id }, { status: num > 0 ? TaskStatus.QUEUED : TaskStatus.PENDING })
     }
 
     async pauseTask(id: number): Promise<void> {
         const task = await this.taskRepository.findOneBy({ id })
-        const successCount = await this.taskItemRepository.countBy({
-            taskId: id,
-            status: In(['success', 'fail'])
-        })
-        if (task && task.status === TaskStatus.PENDING) {
-            task.status = TaskStatus.PAUSE
-            task.sent = successCount
-            await this.taskRepository.save(task)
-            const controller = this.taskControllers[task.id]
-            if (controller) {
-                controller.abort()
-            }
+        if (task.status === TaskStatus.PENDING || task.status === TaskStatus.QUEUED) {
+            await this.taskRepository.update({ id }, { status: TaskStatus.IDLE })
         }
     }
 
@@ -311,8 +240,8 @@ export class TaskService implements OnModuleInit {
     async getConfig(name, defValue) {
         try {
             return await this.sysConfigService.getNumber(name, defValue)
-        } catch (error) {
-            Logger.error('Error getting rcs wait time', error, 'RcsService')
+        } catch (e) {
+            Logger.error('Error getting rcs wait time', e.stack, this.TAG)
         }
         return defValue
     }
@@ -335,7 +264,7 @@ export class TaskService implements OnModuleInit {
             let finish = false
             while (!finish) {
                 if (controller.signal.aborted) {
-                    Logger.log('Task aborted', 'RcsService')
+                    Logger.log('Task aborted', this.TAG)
                     return
                 }
 
@@ -382,12 +311,12 @@ export class TaskService implements OnModuleInit {
                 let device = null
                 while (device === null) {
                     if (controller.signal.aborted) {
-                        Logger.log('Task aborted', 'RcsService')
+                        Logger.log('Task aborted', this.TAG)
                         return
                     }
                     device = await this.deviceService.findAvailableDevice()
                     if (device === null) {
-                        Logger.log('No device available, waiting...', 'RcsService')
+                        Logger.log('No device available, waiting...', this.TAG)
                         await setTimeout(2000)
                     }
                 }
@@ -396,7 +325,7 @@ export class TaskService implements OnModuleInit {
                     TaskItemStatus.PENDING
                 )
                 await this.deviceService.setBusy(device.id, true)
-                Logger.log(`Send task to device ${device.id}(${device.model})`, 'RcsService')
+                Logger.log(`Send task to device ${device.id}(${device.model})`, this.TAG)
 
                 Promise.race([
                     this.eventsGateway.sendForResult(
@@ -423,7 +352,7 @@ export class TaskService implements OnModuleInit {
                 ])
                     .then(async (res: any) => {
                         if (res.error) {
-                            Logger.error('Task timeout', 'RcsService')
+                            Logger.error('Task timeout', this.TAG)
                             await this.updateTaskItemStatus(
                                 taskItems.map((i) => i.id),
                                 TaskItemStatus.IDLE
@@ -433,7 +362,7 @@ export class TaskService implements OnModuleInit {
                                 `Task completed: ${res.success.length} success, ${res.fail.length} fail, ${
                                     res.retry?.length || 0
                                 } retry`,
-                                'RcsService'
+                                this.TAG
                             )
                             if (res.success?.length > 0) {
                                 await this.updateTaskItemStatus(res.success, TaskItemStatus.SUCCESS)
@@ -447,7 +376,7 @@ export class TaskService implements OnModuleInit {
                         }
                     })
                     .catch(async (e) => {
-                        Logger.error('Error running task', e.stack, 'RcsService')
+                        Logger.error('Error running task', e.stack, this.TAG)
                         await this.updateTaskItemStatus(
                             taskItems.map((i) => i.id),
                             TaskItemStatus.IDLE
@@ -455,7 +384,7 @@ export class TaskService implements OnModuleInit {
                     })
             }
         } catch (e) {
-            Logger.error('Error running task', e.stack, 'RcsService')
+            Logger.error('Error running task', e.stack, this.TAG)
             task.status = TaskStatus.ERROR
             task.error = e.message
             await this.taskRepository.save(task)
@@ -473,4 +402,136 @@ export class TaskService implements OnModuleInit {
         const cost = rate.mul(num)
         return cost.toNumber()
     }
+
+    @Interval(2000)
+    async scheduleTask() {
+        this.lock
+            .acquire(
+                'dispatchTask',
+                async () => {
+                    const task = await this.taskRepository.findOneBy({
+                        status: TaskStatus.PENDING
+                    })
+                    if (task) {
+                        await this.dispatchTask(task)
+                    } else {
+                        const nextTask = await this.taskRepository.findOne({
+                            where: {
+                                status: TaskStatus.QUEUED
+                            },
+                            order: {
+                                createdAt: 'ASC'
+                            }
+                        })
+                        if (nextTask) {
+                            await this.taskRepository.update({ id: nextTask.id }, { status: TaskStatus.PENDING })
+                        }
+                    }
+                },
+                {
+                    timeout: 1
+                }
+            )
+            .catch((e) => {
+                if (e.message.includes('timed out')) return
+                Logger.error('Error dispatchTask', e.stack, this.TAG)
+            })
+    }
+
+    async dispatchTask(task: Task) {
+        const batchSize = 200
+        const taskItems = await this.taskItemRepository.find({
+            where: {
+                taskId: task.id,
+                status: TaskItemStatus.IDLE
+            },
+            take: batchSize * 5
+        })
+        if (taskItems.length === 0) {
+            return
+        }
+
+        const devices = await this.deviceService.findAvailableDevices(Math.ceil(taskItems.length / 5), task.matchDevice)
+        if (devices.length === 0) {
+            return
+        }
+
+        const taskConfig = {
+            rcsWait: task.rcsWait || (await this.getConfig('rcs_wait', 2000)),
+            rcsInterval: task.rcsInterval || (await this.getConfig('rcs_interval', 3000)),
+            cleanCount: task.cleanCount || (await this.getConfig('clean_count', 20)),
+            requestNumberInterval: task.requestNumberInterval || (await this.getConfig('request_number_interval', 100)),
+            checkConnection: task.checkConnection
+        }
+        Promise.all(
+            devices.map(async (device, i) => {
+                const items = taskItems.slice(i * 5, i * 5 + 5)
+                await this.updateTaskItemStatus(
+                    items.map((i) => i.id),
+                    TaskItemStatus.PENDING
+                )
+                try {
+                    const res: any = await Promise.race([
+                        this.eventsGateway.sendForResult(
+                            {
+                                id: randomUUID(),
+                                action: 'task',
+                                data: {
+                                    config: taskConfig,
+                                    tasks: items,
+                                    taskId: task.id
+                                }
+                            },
+                            device.socketId
+                        ),
+                        setTimeout(60000).then(() => {
+                            return Promise.reject(new Error('timeout waiting for response'))
+                        })
+                    ])
+                    Logger.log(
+                        `Task completed: ${res.success.length} success, ${res.fail.length} fail, ${
+                            res.retry?.length || 0
+                        } retry`,
+                        this.TAG
+                    )
+                    if (res.success?.length > 0) {
+                        await this.updateTaskItemStatus(res.success, TaskItemStatus.SUCCESS)
+                    }
+                    if (res.fail?.length > 0) {
+                        await this.updateTaskItemStatus(res.fail, TaskItemStatus.FAIL)
+                    }
+                    if (res.retry?.length > 0) {
+                        await this.updateTaskItemStatus(res.retry, TaskItemStatus.IDLE)
+                    }
+                } catch (e) {
+                    Logger.error('Error running task', e.stack, this.TAG)
+                    await this.updateTaskItemStatus(
+                        items.map((i) => i.id),
+                        TaskItemStatus.IDLE
+                    )
+                }
+            })
+        ).then(async () => {
+            const counts = await this.taskItemRepository.manager.query(
+                `select status, count(*) as count from task_item where taskId = ${task.id} group by status`
+            )
+            Logger.log('Task counts', JSON.stringify(counts), this.TAG)
+            const successCount = parseInt(counts.find((c) => c.status === 'success')?.count) || 0
+            const failCount = parseInt(counts.find((c) => c.status === 'fail')?.count) || 0
+            const pendingCount = parseInt(counts.find((c) => c.status === 'pending')?.count) || 0
+            const idleCount = parseInt(counts.find((c) => c.status === 'idle')?.count) || 0
+            const finish = pendingCount === 0 && idleCount === 0
+            const data: Partial<Task> = {
+                sent: successCount + failCount,
+                successRate:
+                    successCount + failCount > 0
+                        ? ((successCount / (successCount + failCount)) * 100).toFixed(1) + '%'
+                        : '0%'
+            }
+            if (finish) {
+                data.status = TaskStatus.COMPLETED
+            }
+            await this.taskRepository.update({ id: task.id }, data)
+        })
+    }
 }

yarn.lock

@@ -1245,6 +1245,11 @@
   resolved "https://registry.npmmirror.com/@tsconfig/node16/-/node16-1.0.4.tgz#0b92dcc0cc1c81f6f306a381f28e31b1a56536e9"
   integrity sha512-vxhUy4J8lyeyinH7Azl1pdd43GJhZH/tP2weN8TntQblOY+A0XbT8DJk1/oCPuOOyg/Ja757rG0CgHcWC8OfMA==
 
+"@types/async-lock@^1.4.2":
+  version "1.4.2"
+  resolved "https://registry.npmmirror.com/@types/async-lock/-/async-lock-1.4.2.tgz#c2037ba1d6018de766c2505c3abe3b7b6b244ab4"
+  integrity sha512-HlZ6Dcr205BmNhwkdXqrg2vkFMN2PluI7Lgr8In3B3wE5PiQHhjRqtW/lGdVU9gw+sM0JcIDx2AN+cW8oSWIcw==
+
 "@types/babel__core@^7.1.14":
   version "7.20.5"
   resolved "https://registry.npmmirror.com/@types/babel__core/-/babel__core-7.20.5.tgz#3df15f27ba85319caa07ba08d0721889bb39c017"
@@ -2334,6 +2339,11 @@ assign-symbols@^1.0.0:
   resolved "https://registry.npmmirror.com/assign-symbols/-/assign-symbols-1.0.0.tgz#59667f41fadd4f20ccbc2bb96b8d4f7f78ec0367"
   integrity sha512-Q+JC7Whu8HhmTdBph/Tq59IoRtoy6KAm5zzPv00WdujX82lbAL8K7WVjne7vdCsAmbF4AYaDOPyO3k0kl8qIrw==
 
+async-lock@^1.4.1:
+  version "1.4.1"
+  resolved "https://registry.npmmirror.com/async-lock/-/async-lock-1.4.1.tgz#56b8718915a9b68b10fce2f2a9a3dddf765ef53f"
+  integrity sha512-Az2ZTpuytrtqENulXwO3GGv1Bztugx6TT37NIo7imr/Qo0gsYiGtSdBa2B6fsXhTpVZDNfu1Qn3pk531e3q+nQ==
+
 async@^3.2.4:
   version "3.2.5"
   resolved "https://registry.npmmirror.com/async/-/async-3.2.5.tgz#ebd52a8fdaf7a2289a24df399f8d8485c8a46b66"