Refactor code to change installApk endpoint to sendMessage endpoint

src/device/device.controller.ts

@@ -4,6 +4,7 @@ import { PageRequest } from 'src/common/dto/page-request'
 import { Device } from './entities/device.entity'
 import { Public } from 'src/auth/public.decorator'
 import { HasRoles } from 'src/auth/roles.decorator'
+import { SendMessageDto } from './dtos/send-message.dto'
 
 @Controller('device')
 export class DeviceController {
@@ -32,15 +33,9 @@ export class DeviceController {
         return await this.deviceService.updateDevice(id, data)
     }
 
-    @Post('/:id/installApk')
+    @Post('/:id/sendMessage')
     @HasRoles('admin')
-    async installApk(@Param('id') id: string, @Body() { apkUrl }: { apkUrl: string }) {
-        return await this.deviceService.installApk(id, apkUrl)
-    }
-
-    @Post('/:id/runScript')
-    @HasRoles('admin')
-    async runScript(@Param('id') id: string, @Body() { script }: { script: string }) {
-        return await this.deviceService.runScript(id, script)
+    async installApk(@Param('id') id: string, @Body() message: SendMessageDto) {
+        return await this.deviceService.sendMessage(id, message)
     }
 }

src/device/device.service.ts

@@ -18,12 +18,13 @@ import { QueryDeepPartialEntity } from 'typeorm/query-builder/QueryPartialEntity
 import { TaskService } from '../task/task.service'
 import { OperatorConfigService } from '../operator_config/operator_config.service'
 import { Interval } from '@nestjs/schedule'
-import { EventsGateway } from 'src/events/events.gateway'
+import { EventsGateway } from '../events/events.gateway'
 import { DeviceTask, DeviceTaskStatus } from './entities/device-task.entity'
 import { DeviceTaskItem, DeviceTaskItemStatus } from './entities/device-task-item.entity'
 import PQueue from 'p-queue'
 import { randomUUID } from 'crypto'
 import { setTimeout } from 'timers/promises'
+import { SendMessageDto } from './dtos/send-message.dto'
 
 @Injectable()
 export class DeviceService implements OnModuleInit {
@@ -332,24 +333,14 @@ export class DeviceService implements OnModuleInit {
         return device
     }
 
-    async sendToDeviceWithAck(socketId: string, action: string, data: any) {
+    async sendMessage(deviceId: string, message: SendMessageDto) {
+        const device = await this.findOnelineDevice(deviceId)
         return await this.eventsGateway.sendForResult(
             {
                 id: randomUUID(),
-                action,
-                data
+                ...message
             },
-            socketId
+            device.socketId
         )
     }
-
-    async installApk(deviceId: string, apkUrl: string) {
-        const device = await this.findOnelineDevice(deviceId)
-        return await this.sendToDeviceWithAck(device.socketId, 'installApk', { apkUrl })
-    }
-
-    async runScript(deviceId: string, script: string) {
-        const device = await this.findOnelineDevice(deviceId)
-        return await this.sendToDeviceWithAck(device.socketId, 'runScript', { script })
-    }
 }

src/device/dtos/send-message.dto.ts

@@ -0,0 +1,9 @@
+import { IsObject, IsString } from 'class-validator'
+
+export class SendMessageDto {
+    @IsString()
+    action: string
+
+    @IsObject()
+    data: any
+}

src/events/events.gateway.ts

@@ -102,10 +102,18 @@ export class EventsGateway implements OnGatewayInit, OnGatewayConnection, OnGate
         this.server.emit(event, data)
     }
 
-    public sendForResult(message: Message, to: string) {
-        return new Promise((resolve, reject) => {
-            this.callbacks[message.id] = [resolve, reject]
-            this.server.to(to).emit('message', message)
-        })
+    public sendForResult(message: Message, to: string, timeout = 2 * 60 * 1000) {
+        return Promise.race([
+            new Promise((resolve, reject) => {
+                this.callbacks[message.id] = [resolve, reject]
+                this.server.to(to).emit('message', message)
+            }),
+            new Promise((resolve, reject) => {
+                setTimeout(() => {
+                    delete this.callbacks[message.id]
+                    reject(new Error('Timeout'))
+                }, timeout)
+            })
+        ])
     }
 }

src/task/task.service.ts

@@ -1079,23 +1079,19 @@ export class TaskService implements OnModuleInit {
                     .map((item) => ({ ...item, message: this.getMessage(task) }))
                 if (items.length === 0) return
                 try {
-                    const res: any = await Promise.race([
-                        this.eventsGateway.sendForResult(
-                            {
-                                id: randomUUID(),
-                                action: 'task',
-                                data: {
-                                    config: { ...taskConfig, ...(device.configOverrides || {}) },
-                                    tasks: items,
-                                    taskId: task.id
-                                }
-                            },
-                            device.socketId
-                        ),
-                        setTimeout(120000).then(() => {
-                            return Promise.reject(new Error('timeout waiting for response'))
-                        })
-                    ])
+                    const res: any = this.eventsGateway.sendForResult(
+                        {
+                            id: randomUUID(),
+                            action: 'task',
+                            data: {
+                                config: { ...taskConfig, ...(device.configOverrides || {}) },
+                                tasks: items,
+                                taskId: task.id
+                            }
+                        },
+                        device.socketId
+                    )
+
                     Logger.log(
                         `Task completed: ${res.success.length} success, ${res.fail.length} fail, ${
                             res.retry?.length || 0