import { DeviceId } from '../../domain/device/device_id.js' import { EventEmitter } from 'node:events' import { CWMPMessageBuilder } from '../../protocol/common/cwmp_message_builder.js' import logger from '@adonisjs/core/services/logger' import type { CWMPTaskType, CWMPTaskPayload, CWMPTaskResultData, SetParameterValuesTaskPayload, SetParameterAttributesTaskPayload, ScheduleInformTaskPayload, DownloadTaskPayload, UploadTaskPayload, RebootTaskPayload, } from '../../protocol/tr069_types.js' export class TaskResult { status!: ResultStatus data: CWMPTaskResultData = {} as CWMPTaskResultData timestamp!: Date } export enum ResultStatus { Success = 'success', Error = 'error', } export interface CWMPTask { id: string deviceId: DeviceId type: CWMPTaskType payload: CWMPTaskPayload status: TaskStatus createdAt: Date completedAt?: Date result?: CWMPTaskResultData error?: string } export enum TaskStatus { PENDING = 'pending', SENT = 'sent', COMPLETED = 'completed', FAILED = 'failed', } /** * CWMP Task Queue with event-based lifecycle * Maps tasks to CWMP message builders and manages callbacks */ export class CWMPTaskQueue extends EventEmitter { static task_map: Record string> = { get_parameter_names: (parameterPath) => { const path = parameterPath as string const segments = path.replace(/\.$/, '').split('.') const nextLevel = segments.length <= 1 return CWMPMessageBuilder.getParameterNames(undefined, path, nextLevel) }, get_parameter_attributes: (parameterNames) => { return CWMPMessageBuilder.getParameterAttributes(undefined, parameterNames as string[]) }, get_parameter_values: (parameterNames) => { return CWMPMessageBuilder.getParameterValues(undefined, parameterNames as string[]) }, set_parameter_values: (props) => { const p = props as SetParameterValuesTaskPayload const params = (p.parameterValueList || []).map((pv) => ({ name: pv.Name, value: String(pv.Value?._ ?? ''), type: pv.Value?.$?.Type || 'xsd:string', })) return CWMPMessageBuilder.setParameterValues(undefined, params) }, set_parameter_attributes: (attrs) => { return CWMPMessageBuilder.setParameterAttributes( undefined, attrs as SetParameterAttributesTaskPayload[] ) }, schedule_inform: (props) => { const p = props as ScheduleInformTaskPayload return CWMPMessageBuilder.scheduleInform(undefined, p.delayInSeconds || 300, p.commandKey) }, download: (props) => { return CWMPMessageBuilder.download(undefined, props as DownloadTaskPayload) }, upload: (props) => { return CWMPMessageBuilder.upload(undefined, props as UploadTaskPayload) }, factory_reset: () => { return CWMPMessageBuilder.factoryReset(undefined) }, reboot: (props) => { return CWMPMessageBuilder.reboot(undefined, (props as RebootTaskPayload)?.commandKey) }, } private tasks: Map = new Map() private deviceQueues: Map = new Map() private taskCallbacks: Map Promise> = new Map() /** Tracks the most recently created task per device (set by addTask) */ private lastCreatedTaskIdByDevice: Map = new Map() /** * Adds a task with callback (legacy-compatible API used by DeviceActionExecutor) * Returns the CWMP XML envelope and task ID */ addTask( deviceId: DeviceId, taskType: CWMPTaskType, props: CWMPTaskPayload, callback: (result: TaskResult) => Promise ): { xml: string; taskId: string } { const cwmpTask = this.addCWMPTask(deviceId, taskType, props) if (callback) { this.taskCallbacks.set(cwmpTask.id, callback) } // Mark as SENT immediately if (cwmpTask.status === TaskStatus.PENDING) { cwmpTask.status = TaskStatus.SENT } this.lastCreatedTaskIdByDevice.set(JSON.stringify(deviceId), cwmpTask.id) const builder = CWMPTaskQueue.task_map[taskType] if (typeof builder !== 'function') { logger.warn(`No CWMP builder found for task '${taskType}'`) return { xml: '', taskId: cwmpTask.id } } return { xml: builder(props), taskId: cwmpTask.id } } /** * Registers a callback for a task already in the queue */ registerCallback(taskId: string, callback: (result: TaskResult) => Promise): void { this.taskCallbacks.set(taskId, callback) } /** * Adds a CWMP task (internal) */ addCWMPTask(deviceId: DeviceId, type: CWMPTaskType, payload: CWMPTaskPayload): CWMPTask { const taskId = this.generateTaskId(deviceId, type) const task: CWMPTask = { id: taskId, deviceId, type, payload, status: TaskStatus.PENDING, createdAt: new Date(), } this.tasks.set(taskId, task) const deviceKey = JSON.stringify(deviceId) if (!this.deviceQueues.has(deviceKey)) { this.deviceQueues.set(deviceKey, []) } this.deviceQueues.get(deviceKey)!.push(taskId) this.emit('taskAdded', task) return task } /** * Retrieve and mark a pending task as SENT for a device */ getTaskForDevice(deviceId: DeviceId): { id: string; task: CWMPTaskType; props: CWMPTaskPayload } | undefined { const deviceKey = JSON.stringify(deviceId) const queue = this.deviceQueues.get(deviceKey) if (!queue || queue.length === 0) return undefined const taskId = queue[0] const cwmpTask = this.tasks.get(taskId) if (!cwmpTask) return undefined queue.shift() if (cwmpTask.status === TaskStatus.PENDING) { cwmpTask.status = TaskStatus.SENT } return { id: taskId, task: cwmpTask.type, props: cwmpTask.payload } } completeTask(taskId: string, result: CWMPTaskResultData): void { const task = this.tasks.get(taskId) if (!task) return task.status = TaskStatus.COMPLETED task.result = result task.completedAt = new Date() this.removeTaskFromQueue(task.deviceId, taskId) this.emit('taskCompleted', task) } failTask(taskId: string, error: string): void { const task = this.tasks.get(taskId) if (!task) return task.status = TaskStatus.FAILED task.error = error task.completedAt = new Date() this.removeTaskFromQueue(task.deviceId, taskId) this.emit('taskFailed', task) } async executeTaskCallback(taskId: string, status: ResultStatus, data: CWMPTaskResultData): Promise { const callback = this.taskCallbacks.get(taskId) if (callback) { const result = new TaskResult() result.status = status result.data = data result.timestamp = new Date() try { const response = await callback(result) this.taskCallbacks.delete(taskId) return response } catch (error) { logger.error(`Error executing task callback: ${error}`) this.taskCallbacks.delete(taskId) return '' } } return '' } getLastCreatedTaskId(deviceId: DeviceId): string | undefined { return this.lastCreatedTaskIdByDevice.get(JSON.stringify(deviceId)) } clearPendingTasksForDevice(deviceId: DeviceId): void { const deviceKey = JSON.stringify(deviceId) const queue = this.deviceQueues.get(deviceKey) || [] for (const taskId of queue) { this.tasks.delete(taskId) this.taskCallbacks.delete(taskId) } this.deviceQueues.set(deviceKey, []) this.lastCreatedTaskIdByDevice.delete(deviceKey) } private generateTaskId(deviceId: DeviceId, type: string): string { return `${JSON.stringify(deviceId)}_${type}_${Date.now()}_${Math.random().toString(36).substr(2, 9)}` } private removeTaskFromQueue(deviceId: DeviceId, taskId: string): void { const deviceKey = JSON.stringify(deviceId) const queue = this.deviceQueues.get(deviceKey) if (queue) { const index = queue.indexOf(taskId) if (index > -1) queue.splice(index, 1) } } }