adam-acs/app/services/cwmp/task_queue.ts

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, (props: any) => 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<string, CWMPTask> = new Map()
  private deviceQueues: Map<string, string[]> = new Map()
  private taskCallbacks: Map<string, (result: TaskResult) => Promise<string>> = 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<string>
  ): { 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
    }

    // 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<string>): 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<string> {
    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 ''
  }

  clearPendingTasksForDevice(deviceId: DeviceId): void {
    const deviceKey = JSON.stringify(deviceId)
    const queue = this.deviceQueues.get(deviceKey) || []
    const keep: string[] = []
    for (const taskId of queue) {
      const t = this.tasks.get(taskId)
      if (t && t.status === TaskStatus.SENT) {
        keep.push(taskId)
        continue
      }
      this.tasks.delete(taskId)
      this.taskCallbacks.delete(taskId)
    }
    this.deviceQueues.set(deviceKey, keep)
  }

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