Unify charge point command channel status

apps/csms/src/ocpp/handler.ts

@@ -1,11 +1,18 @@
 import type { WSContext } from 'hono/ws'
+import dayjs from 'dayjs'
+import { eq } from 'drizzle-orm'
 import { isSupportedOCPP } from '@/constants.js'
+import { useDrizzle } from '@/lib/db.js'
+import { chargePoint } from '@/db/schema.js'
 import {
   OCPP_MESSAGE_TYPE,
   type OcppCall,
+  type OcppCallErrorMessage,
+  type OcppCallResultMessage,
   type OcppErrorCode,
   type OcppMessage,
   type OcppConnectionContext,
+  type CommandChannelStatus,
   type AuthorizeRequest,
   type AuthorizeResponse,
   type BootNotificationRequest,
@@ -24,9 +31,26 @@ import {
 
 /**
  * Global registry of active OCPP WebSocket connections.
- * Key = chargePointIdentifier, Value = WSContext
+ * Key = chargePointIdentifier, Value = connection entry
  */
-export const ocppConnections = new Map<string, WSContext>()
+export type OcppConnectionEntry = {
+  ws: WSContext
+  sessionId: string
+  openedAt: Date
+  lastMessageAt: Date
+}
+
+type PendingCall =
+  | {
+      chargePointIdentifier: string
+      action: string
+      resolve: (payload: unknown) => void
+      reject: (error: Error) => void
+      timeout: ReturnType<typeof setTimeout>
+    }
+
+export const ocppConnections = new Map<string, OcppConnectionEntry>()
+const pendingCalls = new Map<string, PendingCall>()
 import { handleAuthorize } from './actions/authorize.ts'
 import { handleBootNotification } from './actions/boot-notification.ts'
 import { handleHeartbeat } from './actions/heartbeat.ts'
@@ -92,6 +116,80 @@ function sendCallError(
   )
 }
 
+async function updateTransportState(
+  chargePointIdentifier: string,
+  values: Partial<typeof chargePoint.$inferInsert>,
+): Promise<void> {
+  const db = useDrizzle()
+  await db
+    .update(chargePoint)
+    .set({
+      ...values,
+      updatedAt: dayjs().toDate(),
+    })
+    .where(eq(chargePoint.chargePointIdentifier, chargePointIdentifier))
+}
+
+function getCommandChannelStatus(chargePointIdentifier: string): CommandChannelStatus {
+  return ocppConnections.has(chargePointIdentifier) ? 'online' : 'unavailable'
+}
+
+export async function sendOcppCall<TPayload extends Record<string, unknown>, TResult = unknown>(
+  chargePointIdentifier: string,
+  action: string,
+  payload: TPayload,
+  timeoutOrOptions: number | { timeoutMs?: number; uniqueId?: string } = 15000,
+): Promise<TResult> {
+  const entry = ocppConnections.get(chargePointIdentifier)
+  if (!entry) {
+    await updateTransportState(chargePointIdentifier, { transportStatus: 'unavailable' })
+    throw new Error('TransportUnavailable')
+  }
+
+  const timeoutMs =
+    typeof timeoutOrOptions === 'number' ? timeoutOrOptions : (timeoutOrOptions.timeoutMs ?? 15000)
+  const uniqueId =
+    typeof timeoutOrOptions === 'number' ? crypto.randomUUID() : (timeoutOrOptions.uniqueId ?? crypto.randomUUID())
+
+  const resultPromise = new Promise<TResult>((resolve, reject) => {
+    const timeout = setTimeout(async () => {
+      pendingCalls.delete(uniqueId)
+      await updateTransportState(chargePointIdentifier, {
+        transportStatus: getCommandChannelStatus(chargePointIdentifier),
+        lastCommandStatus: 'Timeout',
+        lastCommandAt: dayjs().toDate(),
+      })
+      reject(new Error('CommandTimeout'))
+    }, timeoutMs)
+
+    pendingCalls.set(uniqueId, {
+      chargePointIdentifier,
+      action,
+      resolve: (response) => resolve(response as TResult),
+      reject,
+      timeout,
+    })
+  })
+
+  try {
+    entry.ws.send(JSON.stringify([OCPP_MESSAGE_TYPE.CALL, uniqueId, action, payload]))
+  } catch (error) {
+    const pending = pendingCalls.get(uniqueId)
+    if (pending) {
+      clearTimeout(pending.timeout)
+      pendingCalls.delete(uniqueId)
+    }
+    await updateTransportState(chargePointIdentifier, {
+      transportStatus: 'unavailable',
+      lastCommandStatus: 'Error',
+      lastCommandAt: dayjs().toDate(),
+    })
+    throw error instanceof Error ? error : new Error('CommandSendFailed')
+  }
+
+  return resultPromise
+}
+
 /**
  * Factory that produces a hono-ws event handler object for a single
  * OCPP WebSocket connection.
@@ -104,15 +202,26 @@ export function createOcppHandler(chargePointIdentifier: string, remoteAddr?: st
     chargePointIdentifier,
     isRegistered: false,
   }
+  const sessionId = crypto.randomUUID()
 
   return {
-    onOpen(_evt: Event, ws: WSContext) {
+    async onOpen(_evt: Event, ws: WSContext) {
       const subProtocol = ws.protocol ?? 'unknown'
       if (!isSupportedOCPP(subProtocol)) {
         ws.close(1002, 'Unsupported subprotocol')
         return
       }
-      ocppConnections.set(chargePointIdentifier, ws)
+      ocppConnections.set(chargePointIdentifier, {
+        ws,
+        sessionId,
+        openedAt: new Date(),
+        lastMessageAt: new Date(),
+      })
+      await updateTransportState(chargePointIdentifier, {
+        transportStatus: 'online',
+        connectionSessionId: sessionId,
+        lastWsConnectedAt: dayjs().toDate(),
+      })
       console.log(
         `[OCPP] ${chargePointIdentifier} connected` +
           (remoteAddr ? ` from ${remoteAddr}` : ''),
@@ -122,6 +231,11 @@ export function createOcppHandler(chargePointIdentifier: string, remoteAddr?: st
     async onMessage(evt: MessageEvent, ws: WSContext) {
       let uniqueId = '(unknown)'
       try {
+        const current = ocppConnections.get(chargePointIdentifier)
+        if (current) {
+          current.lastMessageAt = new Date()
+        }
+
         const raw = evt.data
         if (typeof raw !== 'string') return
 
@@ -141,7 +255,36 @@ export function createOcppHandler(chargePointIdentifier: string, remoteAddr?: st
         const [messageType, msgUniqueId] = message
         uniqueId = String(msgUniqueId)
 
-        // CSMS only handles CALL messages from the charge point
+        if (messageType === OCPP_MESSAGE_TYPE.CALLRESULT) {
+          const [, responseUniqueId, payload] = message as OcppCallResultMessage
+          const pending = pendingCalls.get(responseUniqueId)
+          if (!pending) return
+          clearTimeout(pending.timeout)
+          pendingCalls.delete(responseUniqueId)
+          await updateTransportState(pending.chargePointIdentifier, {
+            transportStatus: getCommandChannelStatus(pending.chargePointIdentifier),
+            lastCommandStatus: 'Accepted',
+            lastCommandAt: dayjs().toDate(),
+          })
+          pending.resolve(payload)
+          return
+        }
+
+        if (messageType === OCPP_MESSAGE_TYPE.CALLERROR) {
+          const [, responseUniqueId, errorCode, errorDescription] = message as OcppCallErrorMessage
+          const pending = pendingCalls.get(responseUniqueId)
+          if (!pending) return
+          clearTimeout(pending.timeout)
+          pendingCalls.delete(responseUniqueId)
+          await updateTransportState(pending.chargePointIdentifier, {
+            transportStatus: getCommandChannelStatus(pending.chargePointIdentifier),
+            lastCommandStatus: errorCode === 'InternalError' ? 'Error' : 'Rejected',
+            lastCommandAt: dayjs().toDate(),
+          })
+          pending.reject(new Error(`${errorCode}:${errorDescription}`))
+          return
+        }
+
         if (messageType !== OCPP_MESSAGE_TYPE.CALL) return
 
         const [, , action, payload] = message as OcppCall
@@ -174,8 +317,15 @@ export function createOcppHandler(chargePointIdentifier: string, remoteAddr?: st
       }
     },
 
-    onClose(evt: CloseEvent, _ws: WSContext) {
-      ocppConnections.delete(chargePointIdentifier)
+    async onClose(evt: CloseEvent, _ws: WSContext) {
+      const current = ocppConnections.get(chargePointIdentifier)
+      if (current?.sessionId === sessionId) {
+        ocppConnections.delete(chargePointIdentifier)
+        await updateTransportState(chargePointIdentifier, {
+          transportStatus: 'offline',
+          lastWsDisconnectedAt: dayjs().toDate(),
+        })
+      }
       console.log(`[OCPP] ${chargePointIdentifier} disconnected (code=${evt.code})`)
     },
   }