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