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' export class TaskResult { status!: ResultStatus data: any timestamp!: Date } export enum ResultStatus { Success = 'success', Error = 'error', } export interface CWMPTask { id: string deviceId: DeviceId type: string payload: any status: TaskStatus createdAt: Date completedAt?: Date result?: any 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: string) => { return CWMPMessageBuilder.getParameterNames(undefined, parameterPath, false) }, get_parameter_attributes: (parameterNames: string[]) => { return CWMPMessageBuilder.getParameterAttributes(undefined, parameterNames) }, get_parameter_values: (parameterNames: string[]) => { return CWMPMessageBuilder.getParameterValues(undefined, parameterNames) }, set_parameter_values: (props: { parameterValueList: any[]; parameterKey?: string }) => { const params = (props.parameterValueList || []).map((p: any) => ({ name: p.Name, value: String(p.Value?._ ?? ''), type: p.Value?.$?.Type || 'xsd:string', })) return CWMPMessageBuilder.setParameterValues(undefined, params) }, set_parameter_attributes: (attrs: any[]) => { const mapped = attrs.map((a: any) => ({ name: a.name || a.Name, notificationChange: !!(a.notificationChange ?? a.NotificationChange), notification: a.notification ?? a.Notification ?? 0, accessListChange: !!(a.accessListChange ?? a.AccessListChange), accessList: a.accessList ?? a.AccessList ?? [], })) return CWMPMessageBuilder.setParameterAttributes(undefined, mapped) }, schedule_inform: (props: { delayInSeconds: number; commandKey?: string }) => { return CWMPMessageBuilder.scheduleInform(undefined, props.delayInSeconds || 300, props.commandKey) }, download: (props: any) => { return CWMPMessageBuilder.download(undefined, props) }, upload: (props: any) => { return CWMPMessageBuilder.upload(undefined, props) }, factory_reset: () => { return CWMPMessageBuilder.factoryReset(undefined) }, reboot: (props: any) => { return CWMPMessageBuilder.reboot(undefined, props?.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: string, props: any, 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 } // Track the most recent task created for this device so callers can retrieve its ID this.lastCreatedTaskIdByDevice.set(JSON.stringify(deviceId), cwmpTask.id) // Generate CWMP envelope 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 (e.g. added via addCWMPTask) */ registerCallback(taskId: string, callback: (result: TaskResult) => Promise): void { this.taskCallbacks.set(taskId, callback) } /** * Adds a CWMP task (internal) */ addCWMPTask(deviceId: DeviceId, type: string, payload: any): 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: string; props: any } | 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 // Dequeue queue.shift() if (cwmpTask.status === TaskStatus.PENDING) { cwmpTask.status = TaskStatus.SENT } return { id: taskId, task: cwmpTask.type, props: cwmpTask.payload } } completeTask(taskId: string, result: any): 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: any): 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 '' } /** * Returns the ID of the most recently created task for a device (via addTask). * Used by the state machine manager to correlate responses with in-flight tasks * that were created directly by state handlers (not served from the external queue). */ 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) || [] // Clear ALL tasks (PENDING and SENT) — after session completion, no task // from the previous session should bleed into the next one. Tasks that were // already SENT (created by state-machine states via addTask()) are ghost // entries that would otherwise be served to the next session. 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) } } }