rcs-api/src/task/task.service.ts

import { PhoneListService } from './../phone-list/phone-list.service'
import { forwardRef, Inject, Injectable, Logger, OnModuleInit } from '@nestjs/common'
import { InjectRepository } from '@nestjs/typeorm'
import { Task, TaskStatus } from './entities/task.entity'
import { In, Repository } from 'typeorm'
import { TaskItem, TaskItemStatus } from './entities/task-item.entity'
import { PageRequest } from '../common/dto/page-request'
import { paginate, Pagination } from 'nestjs-typeorm-paginate'
import { EventsGateway } from '../events/events.gateway'
import { randomUUID } from 'crypto'
import { setTimeout } from 'timers/promises'
import { DeviceService } from '../device/device.service'
import { SysConfigService } from '../sys-config/sys-config.service'
import { Users } from '../users/entities/users.entity'
import { Balance } from '../balance/entities/balance.entities'
import * as ExcelJS from 'exceljs'
import * as moment from 'moment'
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'

@Injectable()
export class TaskService implements OnModuleInit {
    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
    ) {}

    onModuleInit() {
        this.taskRepository.update({ status: TaskStatus.PENDING }, { status: TaskStatus.IDLE })
    }

    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(12, '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
    }

    async findAllTaskItem(req: PageRequest<TaskItem>): Promise<Pagination<TaskItem>> {
        return await paginate<TaskItem>(this.taskItemRepository, req.page, req.search)
    }

    async createTask(task: Task): Promise<Task> {
        let dynamicMessageList = null
        if (task.dynamicMessage && task.dynamicMessage !== '') {
            dynamicMessageList = task.dynamicMessage.split(',')
        }
        task = await this.taskRepository.save(task)
        const phones = await this.phoneListService.findPhoneByListId(task.listId)
        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
            })
        )
        return task
    }

    async balanceVerification(id: number) {
        const task = await this.taskRepository.findOneBy({ id })
        // 获取用户信息
        const user = await this.userRepository.findOneBy({
            id: task.userId
        })
        if (user.roles.includes(Role.Admin)) {
            return 0
        }
        const cost: number = await this.getCost(task, user)
        // 验证余额
        if (cost > (user.balance || 0)) {
            return -1
        } else {
            return cost
        }
    }

    async delTask(id: number): Promise<void> {
        await this.taskRepository.delete(id)
    }

    async startTask(id: number): Promise<void> {
        const task = await this.taskRepository.findOneBy({ id })

        // 查询当前是否有任务执行
        const num = await this.taskRepository.count({
            where: {
                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')
                }
            }
        }
    }

    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()
            }
        }
    }

    async exportTaskItem(taskId: number) {
        const workbook = new ExcelJS.Workbook()
        const worksheet = workbook.addWorksheet('Sheet1')

        const taskItems = await this.taskItemRepository.find({
            where: {
                taskId: taskId
            }
        })

        // 设置列头
        worksheet.columns = [
            { header: '手机号', key: 'number', width: 30, style: { alignment: { horizontal: 'center' } } },
            { header: '是否有效', key: 'isValid', width: 15, style: { alignment: { horizontal: 'center' } } },
            { header: '发送成功', key: 'status', width: 15, style: { alignment: { horizontal: 'center' } } },
            {
                header: '发送时间',
                key: 'sendAt',
                width: 30,
                style: { alignment: { horizontal: 'center' }, numFmt: 'YYYY-MM-DD HH:mm:ss' }
            }
        ]

        taskItems.forEach((item) => {
            let valid = '有效'
            let status = '发送成功'
            const sendAt: Date = item.sendAt
            const formattedSendAt = moment(sendAt).format('YYYY-MM-DD HH:mm:ss')
            if (item.status === TaskItemStatus.FAIL) {
                valid = '无效'
                status = ''
            }
            worksheet.addRow({
                number: item.number,
                isValid: valid,
                status: status,
                sendAt: formattedSendAt
            })
        })

        return await workbook.xlsx.writeBuffer()
    }

    async getToBeSentNum() {
        let total = 0
        // 任务队列
        const tasks = await this.taskRepository.find({
            where: {
                status: TaskStatus.QUEUED
            },
            order: {
                createdAt: 'ASC'
            }
        })
        if (tasks.length > 0) {
            const list = tasks.map((task) => task.listId)
            // 队列中剩余任务发送数
            const number = await this.phoneRepository.countBy({
                listId: In(list)
            })
            total += number
        }

        // 正在执行的任务
        const current = await this.taskRepository.find({
            where: {
                status: TaskStatus.PENDING
            }
        })
        if (current.length > 0) {
            const number = await this.phoneRepository.countBy({
                listId: current[0].listId
            })
            total += number
        }
        return total
    }

    async getConfig(name, defValue) {
        try {
            return await this.sysConfigService.getNumber(name, defValue)
        } catch (error) {
            Logger.error('Error getting rcs wait time', error, 'RcsService')
        }
        return defValue
    }

    async updateTaskItemStatus(taskIds: number[], status: TaskItemStatus) {
        await this.taskItemRepository.update(
            {
                id: In(taskIds)
            },
            {
                status: status
            }
        )
    }

    async runTask(task: Task) {
        let controller = new AbortController()
        this.taskControllers[task.id] = controller
        try {
            let finish = false
            while (!finish) {
                if (controller.signal.aborted) {
                    Logger.log('Task aborted', 'RcsService')
                    return
                }

                task = await this.taskRepository.findOneBy({ id: task.id })
                let rcsWait = task.rcsWait || (await this.getConfig('rcs_wait', 2000))
                let rcsInterval = task.rcsInterval || (await this.getConfig('rcs_interval', 3000))
                let cleanCount = task.cleanCount || (await this.getConfig('clean_count', 20))
                let requestNumberInterval =
                    task.requestNumberInterval || (await this.getConfig('request_number_interval', 100))

                let taskItems = await this.taskItemRepository.find({
                    where: { taskId: task.id, status: TaskItemStatus.IDLE },
                    take: 5
                })

                if (taskItems.length === 0) {
                    finish = true
                    task.status = TaskStatus.COMPLETED
                    await this.taskRepository.save(task)
                    // 从队列中获取下一个任务
                    const tasks = await this.taskRepository.find({
                        where: {
                            status: TaskStatus.QUEUED
                        },
                        order: {
                            createdAt: 'ASC'
                        }
                    })
                    // 异步执行startTask方法
                    if (tasks.length > 0) {
                        this.startTask(tasks[0].id)
                    }
                    break
                }

                let device = null
                while (device === null) {
                    if (controller.signal.aborted) {
                        Logger.log('Task aborted', 'RcsService')
                        return
                    }
                    device = await this.deviceService.findAvailableDevice()
                    if (device === null) {
                        Logger.log('No device available, waiting...', 'RcsService')
                        await setTimeout(2000)
                    }
                }
                await this.updateTaskItemStatus(
                    taskItems.map((i) => i.id),
                    TaskItemStatus.PENDING
                )
                await this.deviceService.setBusy(device.id, true)
                Logger.log(`Send task to device ${device.id}(${device.model})`, 'RcsService')

                Promise.race([
                    this.eventsGateway.sendForResult(
                        {
                            id: randomUUID(),
                            action: 'task',
                            data: {
                                config: {
                                    rcsWait,
                                    rcsInterval,
                                    cleanCount,
                                    requestNumberInterval,
                                    checkConnection: task.checkConnection
                                },
                                tasks: taskItems,
                                taskId: task.id
                            }
                        },
                        device.socketId
                    ),
                    setTimeout(60000).then(() => {
                        return Promise.resolve({ error: 'Timeout' })
                    })
                ])
                    .then(async (res: any) => {
                        if (res.error) {
                            Logger.error('Task timeout', 'RcsService')
                            await this.updateTaskItemStatus(
                                taskItems.map((i) => i.id),
                                TaskItemStatus.PENDING
                            )
                        } else {
                            Logger.log(
                                `Task completed: ${res.success.length} success, ${res.fail.length} fail, ${
                                    res.retry?.length || 0
                                } retry`,
                                'RcsService'
                            )
                            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.PENDING)
                            }
                        }
                    })
                    .catch(async (e) => {
                        Logger.error('Error running task', e.stack, 'RcsService')
                        await this.updateTaskItemStatus(
                            taskItems.map((i) => i.id),
                            TaskItemStatus.PENDING
                        )
                    })
            }
        } catch (e) {
            Logger.error('Error running task', e.stack, 'RcsService')
            task.status = TaskStatus.ERROR
            task.error = e.message
            await this.taskRepository.save(task)
        }

        this.taskControllers[task.id] = null
        return task
    }

    async getCost(task: Task, user: Users) {
        const number: number = await this.phoneListService.findCountByListId(task.listId)
        // 费用 = 费率*需要发送量
        const rate = new Decimal(String(user.rate))
        const num = new Decimal(String(number))
        const cost = rate.mul(num)
        return cost.toNumber()
    }
}