src/charging-station/ocpp/OCPPRequestService.ts

   1 import _Ajv, { type ValidateFunction } from 'ajv'
   2 import _ajvFormats from 'ajv-formats'
   3
   4 import type { ChargingStation } from '../../charging-station/index.js'
   5 import { OCPPError } from '../../exception/index.js'
   6 import { PerformanceStatistics } from '../../performance/index.js'
   7 import {
   8   ChargingStationEvents,
   9   type ErrorCallback,
  10   type ErrorResponse,
  11   ErrorType,
  12   type IncomingRequestCommand,
  13   type JsonType,
  14   MessageType,
  15   type OCPPVersion,
  16   type OutgoingRequest,
  17   RequestCommand,
  18   type RequestParams,
  19   type Response,
  20   type ResponseCallback,
  21   type ResponseType
  22 } from '../../types/index.js'
  23 import {
  24   clone,
  25   formatDurationMilliSeconds,
  26   handleSendMessageError,
  27   logger
  28 } from '../../utils/index.js'
  29 import { OCPPConstants } from './OCPPConstants.js'
  30 import type { OCPPResponseService } from './OCPPResponseService.js'
  31 import {
  32   ajvErrorsToErrorType,
  33   convertDateToISOString,
  34   getMessageTypeString
  35 } from './OCPPServiceUtils.js'
  36 type Ajv = _Ajv.default
  37 // eslint-disable-next-line @typescript-eslint/no-redeclare
  38 const Ajv = _Ajv.default
  39 const ajvFormats = _ajvFormats.default
  40
  41 const moduleName = 'OCPPRequestService'
  42
  43 const defaultRequestParams: RequestParams = {
  44   skipBufferingOnError: false,
  45   triggerMessage: false,
  46   throwError: false
  47 }
  48
  49 export abstract class OCPPRequestService {
  50   private static instance: OCPPRequestService | null = null
  51   private readonly version: OCPPVersion
  52   private readonly ocppResponseService: OCPPResponseService
  53   protected readonly ajv: Ajv
  54   protected abstract payloadValidateFunctions: Map<RequestCommand, ValidateFunction<JsonType>>
  55
  56   protected constructor (version: OCPPVersion, ocppResponseService: OCPPResponseService) {
  57     this.version = version
  58     this.ajv = new Ajv({
  59       keywords: ['javaType'],
  60       multipleOfPrecision: 2
  61     })
  62     ajvFormats(this.ajv)
  63     this.ocppResponseService = ocppResponseService
  64     this.requestHandler = this.requestHandler.bind(this)
  65     this.sendMessage = this.sendMessage.bind(this)
  66     this.sendResponse = this.sendResponse.bind(this)
  67     this.sendError = this.sendError.bind(this)
  68     this.internalSendMessage = this.internalSendMessage.bind(this)
  69     this.buildMessageToSend = this.buildMessageToSend.bind(this)
  70     this.validateRequestPayload = this.validateRequestPayload.bind(this)
  71     this.validateIncomingRequestResponsePayload =
  72       this.validateIncomingRequestResponsePayload.bind(this)
  73   }
  74
  75   public static getInstance<T extends OCPPRequestService>(
  76     this: new (ocppResponseService: OCPPResponseService) => T,
  77     ocppResponseService: OCPPResponseService
  78   ): T {
  79     if (OCPPRequestService.instance === null) {
  80       OCPPRequestService.instance = new this(ocppResponseService)
  81     }
  82     return OCPPRequestService.instance as T
  83   }
  84
  85   public async sendResponse (
  86     chargingStation: ChargingStation,
  87     messageId: string,
  88     messagePayload: JsonType,
  89     commandName: IncomingRequestCommand
  90   ): Promise<ResponseType> {
  91     try {
  92       // Send response message
  93       return await this.internalSendMessage(
  94         chargingStation,
  95         messageId,
  96         messagePayload,
  97         MessageType.CALL_RESULT_MESSAGE,
  98         commandName
  99       )
 100     } catch (error) {
 101       handleSendMessageError(
 102         chargingStation,
 103         commandName,
 104         MessageType.CALL_RESULT_MESSAGE,
 105         error as Error,
 106         {
 107           throwError: true
 108         }
 109       )
 110       return null
 111     }
 112   }
 113
 114   public async sendError (
 115     chargingStation: ChargingStation,
 116     messageId: string,
 117     ocppError: OCPPError,
 118     commandName: RequestCommand | IncomingRequestCommand
 119   ): Promise<ResponseType> {
 120     try {
 121       // Send error message
 122       return await this.internalSendMessage(
 123         chargingStation,
 124         messageId,
 125         ocppError,
 126         MessageType.CALL_ERROR_MESSAGE,
 127         commandName
 128       )
 129     } catch (error) {
 130       handleSendMessageError(
 131         chargingStation,
 132         commandName,
 133         MessageType.CALL_ERROR_MESSAGE,
 134         error as Error
 135       )
 136       return null
 137     }
 138   }
 139
 140   protected async sendMessage (
 141     chargingStation: ChargingStation,
 142     messageId: string,
 143     messagePayload: JsonType,
 144     commandName: RequestCommand,
 145     params?: RequestParams
 146   ): Promise<ResponseType> {
 147     params = {
 148       ...defaultRequestParams,
 149       ...params
 150     }
 151     try {
 152       return await this.internalSendMessage(
 153         chargingStation,
 154         messageId,
 155         messagePayload,
 156         MessageType.CALL_MESSAGE,
 157         commandName,
 158         params
 159       )
 160     } catch (error) {
 161       handleSendMessageError(
 162         chargingStation,
 163         commandName,
 164         MessageType.CALL_MESSAGE,
 165         error as Error,
 166         {
 167           throwError: params.throwError
 168         }
 169       )
 170       return null
 171     }
 172   }
 173
 174   private validateRequestPayload<T extends JsonType>(
 175     chargingStation: ChargingStation,
 176     commandName: RequestCommand | IncomingRequestCommand,
 177     payload: T
 178   ): boolean {
 179     if (chargingStation.stationInfo?.ocppStrictCompliance === false) {
 180       return true
 181     }
 182     if (!this.payloadValidateFunctions.has(commandName as RequestCommand)) {
 183       logger.warn(
 184         `${chargingStation.logPrefix()} ${moduleName}.validateRequestPayload: No JSON schema found for command '${commandName}' PDU validation`
 185       )
 186       return true
 187     }
 188     const validate = this.payloadValidateFunctions.get(commandName as RequestCommand)
 189     payload = clone<T>(payload)
 190     convertDateToISOString<T>(payload)
 191     if (validate?.(payload) === true) {
 192       return true
 193     }
 194     logger.error(
 195       `${chargingStation.logPrefix()} ${moduleName}.validateRequestPayload: Command '${commandName}' request PDU is invalid: %j`,
 196       validate?.errors
 197     )
 198     // OCPPError usage here is debatable: it's an error in the OCPP stack but not targeted to sendError().
 199     throw new OCPPError(
 200       ajvErrorsToErrorType(validate?.errors),
 201       'Request PDU is invalid',
 202       commandName,
 203       JSON.stringify(validate?.errors, undefined, 2)
 204     )
 205   }
 206
 207   private validateIncomingRequestResponsePayload<T extends JsonType>(
 208     chargingStation: ChargingStation,
 209     commandName: RequestCommand | IncomingRequestCommand,
 210     payload: T
 211   ): boolean {
 212     if (chargingStation.stationInfo?.ocppStrictCompliance === false) {
 213       return true
 214     }
 215     if (
 216       !this.ocppResponseService.incomingRequestResponsePayloadValidateFunctions.has(
 217         commandName as IncomingRequestCommand
 218       )
 219     ) {
 220       logger.warn(
 221         `${chargingStation.logPrefix()} ${moduleName}.validateIncomingRequestResponsePayload: No JSON schema validation function found for command '${commandName}' PDU validation`
 222       )
 223       return true
 224     }
 225     const validate = this.ocppResponseService.incomingRequestResponsePayloadValidateFunctions.get(
 226       commandName as IncomingRequestCommand
 227     )
 228     payload = clone<T>(payload)
 229     convertDateToISOString<T>(payload)
 230     if (validate?.(payload) === true) {
 231       return true
 232     }
 233     logger.error(
 234       `${chargingStation.logPrefix()} ${moduleName}.validateIncomingRequestResponsePayload: Command '${commandName}' response PDU is invalid: %j`,
 235       validate?.errors
 236     )
 237     // OCPPError usage here is debatable: it's an error in the OCPP stack but not targeted to sendError().
 238     throw new OCPPError(
 239       ajvErrorsToErrorType(validate?.errors),
 240       'Response PDU is invalid',
 241       commandName,
 242       JSON.stringify(validate?.errors, undefined, 2)
 243     )
 244   }
 245
 246   private async internalSendMessage (
 247     chargingStation: ChargingStation,
 248     messageId: string,
 249     messagePayload: JsonType | OCPPError,
 250     messageType: MessageType,
 251     commandName: RequestCommand | IncomingRequestCommand,
 252     params?: RequestParams
 253   ): Promise<ResponseType> {
 254     params = {
 255       ...defaultRequestParams,
 256       ...params
 257     }
 258     if (
 259       (chargingStation.inUnknownState() && commandName === RequestCommand.BOOT_NOTIFICATION) ||
 260       (chargingStation.stationInfo?.ocppStrictCompliance === false &&
 261         chargingStation.inUnknownState()) ||
 262       chargingStation.inAcceptedState() ||
 263       (chargingStation.inPendingState() &&
 264         (params.triggerMessage === true || messageType === MessageType.CALL_RESULT_MESSAGE))
 265     ) {
 266       // eslint-disable-next-line @typescript-eslint/no-this-alias
 267       const self = this
 268       // Send a message through wsConnection
 269       return await new Promise<ResponseType>((resolve, reject: (reason?: unknown) => void) => {
 270         /**
 271          * Function that will receive the request's response
 272          *
 273          * @param payload -
 274          * @param requestPayload -
 275          */
 276         const responseCallback = (payload: JsonType, requestPayload: JsonType): void => {
 277           if (chargingStation.stationInfo?.enableStatistics === true) {
 278             chargingStation.performanceStatistics?.addRequestStatistic(
 279               commandName,
 280               MessageType.CALL_RESULT_MESSAGE
 281             )
 282           }
 283           // Handle the request's response
 284           self.ocppResponseService
 285             .responseHandler(
 286               chargingStation,
 287               commandName as RequestCommand,
 288               payload,
 289               requestPayload
 290             )
 291             .then(() => {
 292               resolve(payload)
 293             })
 294             .catch(reject)
 295             .finally(() => {
 296               chargingStation.requests.delete(messageId)
 297               chargingStation.emit(ChargingStationEvents.updated)
 298             })
 299         }
 300
 301         /**
 302          * Function that will receive the request's error response
 303          *
 304          * @param ocppError -
 305          * @param requestStatistic -
 306          */
 307         const errorCallback = (ocppError: OCPPError, requestStatistic = true): void => {
 308           if (requestStatistic && chargingStation.stationInfo?.enableStatistics === true) {
 309             chargingStation.performanceStatistics?.addRequestStatistic(
 310               commandName,
 311               MessageType.CALL_ERROR_MESSAGE
 312             )
 313           }
 314           logger.error(
 315             `${chargingStation.logPrefix()} Error occurred at ${getMessageTypeString(
 316               messageType
 317             )} command ${commandName} with PDU %j:`,
 318             messagePayload,
 319             ocppError
 320           )
 321           chargingStation.requests.delete(messageId)
 322           chargingStation.emit(ChargingStationEvents.updated)
 323           reject(ocppError)
 324         }
 325
 326         const handleSendError = (ocppError: OCPPError): void => {
 327           if (params.skipBufferingOnError === false) {
 328             // Buffer
 329             chargingStation.bufferMessage(messageToSend)
 330             if (messageType === MessageType.CALL_MESSAGE) {
 331               this.setCachedRequest(
 332                 chargingStation,
 333                 messageId,
 334                 messagePayload as JsonType,
 335                 commandName,
 336                 responseCallback,
 337                 errorCallback
 338               )
 339             }
 340           } else if (
 341             params.skipBufferingOnError === true &&
 342             messageType === MessageType.CALL_MESSAGE
 343           ) {
 344             // Remove request from the cache
 345             chargingStation.requests.delete(messageId)
 346           }
 347           reject(ocppError)
 348         }
 349
 350         if (chargingStation.stationInfo?.enableStatistics === true) {
 351           chargingStation.performanceStatistics?.addRequestStatistic(commandName, messageType)
 352         }
 353         const messageToSend = this.buildMessageToSend(
 354           chargingStation,
 355           messageId,
 356           messagePayload,
 357           messageType,
 358           commandName
 359         )
 360         // Check if wsConnection opened
 361         if (chargingStation.isWebSocketConnectionOpened()) {
 362           const beginId = PerformanceStatistics.beginMeasure(commandName)
 363           const sendTimeout = setTimeout(() => {
 364             handleSendError(
 365               new OCPPError(
 366                 ErrorType.GENERIC_ERROR,
 367                 `Timeout ${formatDurationMilliSeconds(
 368                   OCPPConstants.OCPP_WEBSOCKET_TIMEOUT
 369                 )} reached for ${
 370                   params.skipBufferingOnError === false ? '' : 'non '
 371                 }buffered message id '${messageId}' with content '${messageToSend}'`,
 372                 commandName,
 373                 (messagePayload as OCPPError).details
 374               )
 375             )
 376           }, OCPPConstants.OCPP_WEBSOCKET_TIMEOUT)
 377           chargingStation.wsConnection?.send(messageToSend, (error?: Error) => {
 378             PerformanceStatistics.endMeasure(commandName, beginId)
 379             clearTimeout(sendTimeout)
 380             if (error == null) {
 381               logger.debug(
 382                 `${chargingStation.logPrefix()} >> Command '${commandName}' sent ${getMessageTypeString(
 383                   messageType
 384                 )} payload: ${messageToSend}`
 385               )
 386               if (messageType === MessageType.CALL_MESSAGE) {
 387                 this.setCachedRequest(
 388                   chargingStation,
 389                   messageId,
 390                   messagePayload as JsonType,
 391                   commandName,
 392                   responseCallback,
 393                   errorCallback
 394                 )
 395               } else {
 396                 // Resolve response
 397                 resolve(messagePayload)
 398               }
 399             } else {
 400               handleSendError(
 401                 new OCPPError(
 402                   ErrorType.GENERIC_ERROR,
 403                   `WebSocket errored for ${
 404                     params.skipBufferingOnError === false ? '' : 'non '
 405                   }buffered message id '${messageId}' with content '${messageToSend}'`,
 406                   commandName,
 407                   {
 408                     name: error.name,
 409                     message: error.message,
 410                     stack: error.stack
 411                   }
 412                 )
 413               )
 414             }
 415           })
 416         } else {
 417           handleSendError(
 418             new OCPPError(
 419               ErrorType.GENERIC_ERROR,
 420               `WebSocket closed for ${
 421                 params.skipBufferingOnError === false ? '' : 'non '
 422               }buffered message id '${messageId}' with content '${messageToSend}'`,
 423               commandName,
 424               (messagePayload as OCPPError).details
 425             )
 426           )
 427         }
 428       })
 429     }
 430     throw new OCPPError(
 431       ErrorType.SECURITY_ERROR,
 432       `Cannot send command ${commandName} PDU when the charging station is in ${chargingStation.bootNotificationResponse?.status} state on the central server`,
 433       commandName
 434     )
 435   }
 436
 437   private buildMessageToSend (
 438     chargingStation: ChargingStation,
 439     messageId: string,
 440     messagePayload: JsonType | OCPPError,
 441     messageType: MessageType,
 442     commandName: RequestCommand | IncomingRequestCommand
 443   ): string {
 444     let messageToSend: string
 445     // Type of message
 446     switch (messageType) {
 447       // Request
 448       case MessageType.CALL_MESSAGE:
 449         // Build request
 450         this.validateRequestPayload(chargingStation, commandName, messagePayload as JsonType)
 451         messageToSend = JSON.stringify([
 452           messageType,
 453           messageId,
 454           commandName as RequestCommand,
 455           messagePayload as JsonType
 456         ] satisfies OutgoingRequest)
 457         break
 458       // Response
 459       case MessageType.CALL_RESULT_MESSAGE:
 460         // Build response
 461         this.validateIncomingRequestResponsePayload(
 462           chargingStation,
 463           commandName,
 464           messagePayload as JsonType
 465         )
 466         messageToSend = JSON.stringify([
 467           messageType,
 468           messageId,
 469           messagePayload as JsonType
 470         ] satisfies Response)
 471         break
 472       // Error Message
 473       case MessageType.CALL_ERROR_MESSAGE:
 474         // Build Error Message
 475         messageToSend = JSON.stringify([
 476           messageType,
 477           messageId,
 478           (messagePayload as OCPPError).code,
 479           (messagePayload as OCPPError).message,
 480           (messagePayload as OCPPError).details ?? {
 481             command: (messagePayload as OCPPError).command
 482           }
 483         ] satisfies ErrorResponse)
 484         break
 485     }
 486     return messageToSend
 487   }
 488
 489   private setCachedRequest (
 490     chargingStation: ChargingStation,
 491     messageId: string,
 492     messagePayload: JsonType,
 493     commandName: RequestCommand | IncomingRequestCommand,
 494     responseCallback: ResponseCallback,
 495     errorCallback: ErrorCallback
 496   ): void {
 497     chargingStation.requests.set(messageId, [
 498       responseCallback,
 499       errorCallback,
 500       commandName,
 501       messagePayload
 502     ])
 503   }
 504
 505   public abstract requestHandler<ReqType extends JsonType, ResType extends JsonType>(
 506     chargingStation: ChargingStation,
 507     commandName: RequestCommand,
 508     commandParams?: ReqType,
 509     params?: RequestParams
 510   ): Promise<ResType>
 511 }