Merge pull request 'add mapper. add v2 handler. add tag description short and full versions. delete topic[1-5].' (#5) from edge5modbus into main

Reviewed-on: #5

app/.dockerignore

@@ -1,4 +1,2 @@
 node_modules
 dist
-.env
-.env.*

app/Dockerfile

@@ -6,9 +6,8 @@ COPY package.json package-lock.json ./
 RUN npm ci
 
 COPY tsconfig.json ./
+COPY public ./public
+# Runtime mappers read JSON files from src/mappings, so they must be copied into the image.
 COPY src ./src
 
-RUN npm run build && npm prune --omit=dev
+CMD ["npm", "run", "dev"]
-
-ENV NODE_ENV=production
-CMD ["node", "dist/index.js"]

app/package-lock.json

@@ -6,13 +6,11 @@
     "": {
       "name": "mqtt-worker",
       "dependencies": {
-        "mqtt": "^5.15.1",
-        "ws": "^8.21.0"
-      },
-      "devDependencies": {
         "@types/node": "^25.9.1",
+        "mqtt": "^5.15.1",
         "tsx": "^4.22.4",
-        "typescript": "^6.0.3"
+        "typescript": "^6.0.3",
+        "ws": "^8.21.0"
       }
     },
     "node_modules/@babel/runtime": {
@@ -31,7 +29,6 @@
       "cpu": [
         "ppc64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -48,7 +45,6 @@
       "cpu": [
         "arm"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -65,7 +61,6 @@
       "cpu": [
         "arm64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -82,7 +77,6 @@
       "cpu": [
         "x64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -99,7 +93,6 @@
       "cpu": [
         "arm64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -116,7 +109,6 @@
       "cpu": [
         "x64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -133,7 +125,6 @@
       "cpu": [
         "arm64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -150,7 +141,6 @@
       "cpu": [
         "x64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -167,7 +157,6 @@
       "cpu": [
         "arm"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -184,7 +173,6 @@
       "cpu": [
         "arm64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -201,7 +189,6 @@
       "cpu": [
         "ia32"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -218,7 +205,6 @@
       "cpu": [
         "loong64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -235,7 +221,6 @@
       "cpu": [
         "mips64el"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -252,7 +237,6 @@
       "cpu": [
         "ppc64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -269,7 +253,6 @@
       "cpu": [
         "riscv64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -286,7 +269,6 @@
       "cpu": [
         "s390x"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -303,7 +285,6 @@
       "cpu": [
         "x64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -320,7 +301,6 @@
       "cpu": [
         "arm64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -337,7 +317,6 @@
       "cpu": [
         "x64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -354,7 +333,6 @@
       "cpu": [
         "arm64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -371,7 +349,6 @@
       "cpu": [
         "x64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -388,7 +365,6 @@
       "cpu": [
         "arm64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -405,7 +381,6 @@
       "cpu": [
         "x64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -422,7 +397,6 @@
       "cpu": [
         "arm64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -439,7 +413,6 @@
       "cpu": [
         "ia32"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -456,7 +429,6 @@
       "cpu": [
         "x64"
       ],
-      "dev": true,
       "license": "MIT",
       "optional": true,
       "os": [
@@ -635,7 +607,6 @@
       "version": "0.28.0",
       "resolved": "https://registry.npmjs.org/esbuild/-/esbuild-0.28.0.tgz",
       "integrity": "sha512-sNR9MHpXSUV/XB4zmsFKN+QgVG82Cc7+/aaxJ8Adi8hyOac+EXptIp45QBPaVyX3N70664wRbTcLTOemCAnyqw==",
-      "dev": true,
       "hasInstallScript": true,
       "license": "MIT",
       "bin": {
@@ -708,7 +679,6 @@
       "version": "2.3.3",
       "resolved": "https://registry.npmjs.org/fsevents/-/fsevents-2.3.3.tgz",
       "integrity": "sha512-5xoDfX+fL7faATnagmWPpbFtwh/R77WmMMqqHGS65C3vvB0YHrgF+B1YmZ3441tMj5n63k0212XNoJwzlhffQw==",
-      "dev": true,
       "hasInstallScript": true,
       "license": "MIT",
       "optional": true,
@@ -953,7 +923,6 @@
       "version": "4.22.4",
       "resolved": "https://registry.npmjs.org/tsx/-/tsx-4.22.4.tgz",
       "integrity": "sha512-X8EX+XV4QR5xCsrgxaED954zTDfY8KqlDtskKEL0cHhyS/P8b4IFOvGDQpsC9Q1XnLq915wEfwwY/zzskCtmhg==",
-      "dev": true,
       "license": "MIT",
       "dependencies": {
         "esbuild": "~0.28.0"
@@ -978,7 +947,6 @@
       "version": "6.0.3",
       "resolved": "https://registry.npmjs.org/typescript/-/typescript-6.0.3.tgz",
       "integrity": "sha512-y2TvuxSZPDyQakkFRPZHKFm+KKVqIisdg9/CZwm9ftvKXLP8NRWj38/ODjNbr43SsoXqNuAisEf1GdCxqWcdBw==",
-      "dev": true,
       "license": "Apache-2.0",
       "bin": {
         "tsc": "bin/tsc",

app/package.json

@@ -8,12 +8,10 @@
     "start": "node --env-file=.env dist/index.js"
   },
   "dependencies": {
-    "mqtt": "^5.15.1",
-    "ws": "^8.21.0"
-  },
-  "devDependencies": {
     "@types/node": "^25.9.1",
+    "mqtt": "^5.15.1",
     "tsx": "^4.22.4",
-    "typescript": "^6.0.3"
+    "typescript": "^6.0.3",
+    "ws": "^8.21.0"
   }
 }

app/src/cloud-ingest.ts

@@ -1,4 +1,7 @@
-export const cloudIngestUrl = process.env.CLOUD_INGEST_URL as string
+const DEFAULT_CLOUD_INGEST_URL = 'http://localhost:3101/ingest'
+
+export const cloudIngestUrl =
+  process.env.CLOUD_INGEST_URL ?? DEFAULT_CLOUD_INGEST_URL
 
 export function postCloudIngest(body: unknown): Promise<Response> {
   return fetch(cloudIngestUrl, {

app/src/handlers/demo-modbus.ts

@@ -1,63 +0,0 @@
-import type { TopicHandler } from '../types.js'
-
-export const TOPIC = 'data/demo/modbus/v1'
-
-const EDGE = 'demo'
-const REGISTER_COUNT = 125
-const DEMO_CLOUD_INGEST_URL = process.env.DEMO_CLOUD_INGEST_URL
-
-type DemoModbusMetric = {
-  edge: string
-  timestamp: string
-  tag: string
-  value: number
-}
-
-function parseValues(payload: Buffer): number[] {
-  const values = JSON.parse(payload.toString('utf8')) as number[]
-  assertRegisterCount(values)
-  return values
-}
-
-function assertRegisterCount(values: number[]): void {
-  if (values.length !== REGISTER_COUNT) {
-    throw new Error(`Expected ${REGISTER_COUNT} modbus values, got ${values.length}`)
-  }
-}
-
-function toMetrics(values: number[], timestamp: string): DemoModbusMetric[] {
-  return values.map((value, index) => ({
-    edge: EDGE,
-    timestamp,
-    tag: `${EDGE}.modbus.${index + 1}`,
-    value,
-  }))
-}
-
-async function postMetrics(metrics: DemoModbusMetric[]): Promise<void> {
-  for (const metric of metrics) {
-    const response = await fetch(DEMO_CLOUD_INGEST_URL as string, {
-      method: 'POST',
-      headers: {
-        'content-type': 'application/json',
-        'x-api-key': process.env.CLOUD_INGEST_API_KEY as string,
-      },
-      body: JSON.stringify(metric),
-    })
-
-    if (!response.ok) {
-      const text = await response.text()
-      throw new Error(text)
-    }
-  }
-
-  console.log('demo.modbus.posted', { count: metrics.length })
-}
-
-export const handleDemoModbus: TopicHandler = async (topic, payload) => {
-  console.log('demo.modbus.received', { topic, bytes: payload.length })
-
-  const values = parseValues(payload)
-  const metrics = toMetrics(values, new Date().toISOString())
-  await postMetrics(metrics)
-}

app/src/handlers/demo-plc.ts

@@ -1,10 +1,10 @@
+import { postCloudIngest } from '../cloud-ingest.js'
 import { parseJson } from '../helpers/parse.js'
 import type { TopicHandler } from '../types.js'
 
 export const TOPIC = 'data/demo/plc/v1'
 
 const EDGE = 'demo'
-const DEMO_CLOUD_INGEST_URL = process.env.DEMO_CLOUD_INGEST_URL
 
 type DemoPlcPayload = {
   tag: string
@@ -13,12 +13,12 @@ type DemoPlcPayload = {
 
 type DemoPlcMetric = {
   edge: string
-  timestamp: string
+  timestamp: number
   tag: string
   value: number
 }
 
-function toMetric(payload: DemoPlcPayload, timestamp: string): DemoPlcMetric {
+function toMetric(payload: DemoPlcPayload, timestamp: number): DemoPlcMetric {
   return {
     edge: EDGE,
     timestamp,
@@ -28,14 +28,7 @@ function toMetric(payload: DemoPlcPayload, timestamp: string): DemoPlcMetric {
 }
 
 async function postMetric(metric: DemoPlcMetric): Promise<void> {
-  const response = await fetch(DEMO_CLOUD_INGEST_URL as string, {
+  const response = await postCloudIngest([metric])
-    method: 'POST',
-    headers: {
-      'content-type': 'application/json',
-      'x-api-key': process.env.CLOUD_INGEST_API_KEY as string,
-    },
-    body: JSON.stringify(metric),
-  })
 
   if (!response.ok) {
     const text = await response.text()
@@ -49,6 +42,6 @@ export const handleDemoPlc: TopicHandler = async (topic, payload) => {
   console.log('demo.plc.received', { topic, bytes: payload.length })
 
   const value = parseJson<DemoPlcPayload>(payload)
-  const metric = toMetric(value, new Date().toISOString())
+  const metric = toMetric(value, Date.now())
   await postMetric(metric)
 }

app/src/handlers/edge5-modbus-v2.ts

@@ -4,56 +4,42 @@ import type { TopicHandler } from '../types.js'
 
 export const TOPIC = 'data/edge5/modbus/v2'
 
-const REGISTER_COUNT = 125
 const mapper = createMetricMapper({
-  filePath: process.env.EDGE5_MODBUS_V2_MAPPING_FILE as string,
+  filePath:
+    process.env.EDGE5_MODBUS_V2_MAPPING_FILE ??
+    new URL('../mappings/edge5-modbus.json', import.meta.url),
 })
+const REGISTER_COUNT = mapper.tagCount
 
 function parseValues(payload: Buffer): number[] {
-  const values = JSON.parse(payload.toString('utf8')) as number[]
+  const value = JSON.parse(payload.toString('utf8')) as unknown
-  assertRegisterCount(values)
+
-  return values
+  if (
+    Array.isArray(value) &&
+    value.length === REGISTER_COUNT &&
+    value.every((item) => typeof item === 'number' && Number.isFinite(item))
+  ) {
+    return value
   }
 
-function assertRegisterCount(values: number[]): void {
+  throw new Error(`Expected JSON number[${REGISTER_COUNT}]`)
-  if (values.length !== REGISTER_COUNT) {
-    throw new Error(`Expected ${REGISTER_COUNT} modbus values, got ${values.length}`)
-  }
 }
 
 async function postMetrics(metrics: MappedMetric[]): Promise<void> {
-  const errors: Error[] = []
+  const response = await postCloudIngest(metrics)
-
-  for (const metric of metrics) {
-    try {
-      const response = await postCloudIngest(metric)
 
   if (!response.ok) {
     const text = await response.text()
-        throw new Error(text || `${response.status} ${response.statusText}`)
+    throw new Error(text)
-      }
-    } catch (error) {
-      const message = error instanceof Error ? error.message : String(error)
-      errors.push(new Error(`${metric.tag}: ${message}`))
-      console.error('edge5.modbus.v2.metric.failed', metric.tag, message)
-    }
   }
 
-  console.log('edge5.modbus.v2.posted', {
+  console.log('edge5.modbus.v2.posted', { count: metrics.length })
-    total: metrics.length,
-    succeeded: metrics.length - errors.length,
-    failed: errors.length,
-  })
-
-  if (errors.length) {
-    throw new Error(`Failed to post ${errors.length}/${metrics.length} edge5 modbus v2 metrics`)
-  }
 }
 
 export const handleEdge5ModbusV2: TopicHandler = async (topic, payload) => {
   console.log('edge5.modbus.v2.received', { topic, bytes: payload.length })
 
   const values = parseValues(payload)
-  const metrics = mapper.mapValues(values, new Date().toISOString())
+  const metrics = mapper.mapValues(values, Date.now())
   await postMetrics(metrics)
 }

app/src/handlers/edge5-modbus.ts

@@ -8,24 +8,32 @@ const REGISTER_COUNT = 125
 
 type Edge5ModbusMetric = {
   edge: string
-  timestamp: string
+  timestamp: number
   tag: string
   value: number
 }
 
 function parseValues(payload: Buffer): number[] {
-  const values = JSON.parse(payload.toString('utf8')) as number[]
+  let value: unknown
-  assertRegisterCount(values)
+
-  return values
+  try {
+    value = JSON.parse(payload.toString('utf8')) as unknown
+  } catch {
+    throw new Error(`Expected JSON number[${REGISTER_COUNT}]`)
   }
 
-function assertRegisterCount(values: number[]): void {
+  if (
-  if (values.length !== REGISTER_COUNT) {
+    Array.isArray(value) &&
-    throw new Error(`Expected ${REGISTER_COUNT} modbus values, got ${values.length}`)
+    value.length === REGISTER_COUNT &&
-  }
+    value.every((item) => typeof item === 'number' && Number.isFinite(item))
+  ) {
+    return value
   }
 
-function toMetrics(values: number[], timestamp: string): Edge5ModbusMetric[] {
+  throw new Error(`Expected JSON number[${REGISTER_COUNT}]`)
+}
+
+function toMetrics(values: number[], timestamp: number): Edge5ModbusMetric[] {
   return values.map((value, index) => ({
     edge: EDGE,
     timestamp,
@@ -35,13 +43,16 @@ function toMetrics(values: number[], timestamp: string): Edge5ModbusMetric[] {
 }
 
 async function postMetrics(metrics: Edge5ModbusMetric[]): Promise<void> {
-  for (const metric of metrics) {
+  const response = await postCloudIngest(metrics)
-    const response = await postCloudIngest(metric)
 
   if (!response.ok) {
     const text = await response.text()
-      throw new Error(text)
+    console.error('edge5.modbus.post_error', {
-    }
+      status: response.status,
+      statusText: response.statusText,
+      body: text.slice(0, 500),
+    })
+    return
   }
 
   console.log('edge5.modbus.posted', { count: metrics.length })
@@ -50,7 +61,15 @@ async function postMetrics(metrics: Edge5ModbusMetric[]): Promise<void> {
 export const handleEdge5Modbus: TopicHandler = async (topic, payload) => {
   console.log('edge5.modbus.received', { topic, bytes: payload.length })
 
-  const values = parseValues(payload)
+  let values: number[]
-  const metrics = toMetrics(values, new Date().toISOString())
+
+  try {
+    values = parseValues(payload)
+  } catch (err) {
+    console.warn('edge5.modbus.invalid_payload', { topic, err })
+    return
+  }
+
+  const metrics = toMetrics(values, Date.now())
   await postMetrics(metrics)
 }

app/src/handlers/index.ts

@@ -1,5 +1,4 @@
 import { register } from '../registry.js'
-import { handleDemoModbus, TOPIC as demoModbus } from './demo-modbus.js'
 import { handleDemoPlc, TOPIC as demoPlc } from './demo-plc.js'
 import { handleEdge5Modbus, TOPIC as edge5Modbus } from './edge5-modbus.js'
 import {
@@ -8,7 +7,6 @@ import {
 } from './edge5-modbus-v2.js'
 import { handleEdge5Video, TOPIC as edge5Video } from './edge5-video.js'
 
-register(demoModbus, handleDemoModbus)
 register(demoPlc, handleDemoPlc)
 register(edge5Modbus, handleEdge5Modbus)
 register(edge5ModbusV2, handleEdge5ModbusV2)

app/src/helpers/parse.ts

@@ -7,6 +7,14 @@ export function parseJson<T = unknown>(payload: Buffer): T {
   return JSON.parse(text) as T
 }
 
+export function parseJsonArray(payload: Buffer): unknown[] {
+  const value = parseJson(payload)
+  if (!Array.isArray(value)) {
+    throw new Error('Expected JSON array')
+  }
+  return value
+}
+
 export function decodeBase64(payload: Buffer): Buffer {
   return Buffer.from(payload.toString('utf8'), 'base64')
 }

app/src/helpers/validate.ts

@@ -0,0 +1,39 @@
+const EXPECTED_CHUNK_LENGTH = 125
+
+export function isNonEmptyString(value: unknown): value is string {
+  return typeof value === 'string' && value.length > 0
+}
+
+export function isFiniteNumber(value: unknown): value is number {
+  return typeof value === 'number' && Number.isFinite(value)
+}
+
+export function isArrayOfLength(
+  value: unknown,
+  length: number,
+): value is unknown[] {
+  return Array.isArray(value) && value.length === length
+}
+
+export function isArrayOfLength125(value: unknown): value is unknown[] {
+  return isArrayOfLength(value, EXPECTED_CHUNK_LENGTH)
+}
+
+export function isArrayOfChunks125(
+  value: unknown,
+): value is unknown[][] {
+  if (!Array.isArray(value)) {
+    return false
+  }
+  return value.every((chunk) => isArrayOfLength125(chunk))
+}
+
+export function assertValid<T>(
+  value: unknown,
+  guard: (v: unknown) => v is T,
+  label: string,
+): asserts value is T {
+  if (!guard(value)) {
+    throw new Error(`Invalid payload: ${label}`)
+  }
+}

app/src/http/server.ts

@@ -3,7 +3,7 @@ import { createServer } from 'node:http'
 import { dirname, join } from 'node:path'
 import { fileURLToPath } from 'node:url'
 
-const httpPort = Number(process.env.HTTP_PORT)
+const httpPort = Number(process.env.HTTP_PORT ?? 80)
 const publicDir = join(dirname(fileURLToPath(import.meta.url)), '../../public')
 const indexPath = join(publicDir, 'index.html')
 

app/src/mapping.ts

@@ -2,7 +2,7 @@ import { readFileSync } from 'node:fs'
 
 export type MappedMetric = {
   edge: string
-  timestamp: string
+  timestamp: number
   tag: string
   value: number
 }
@@ -14,7 +14,8 @@ type MappingConfig = {
 
 export type MetricMapper = {
   edge: string
-  mapValues: (values: number[], timestamp: string) => MappedMetric[]
+  tagCount: number
+  mapValues: (values: number[], timestamp: number) => MappedMetric[]
 }
 
 type MetricMapperOptions = {
@@ -33,6 +34,7 @@ export function createMetricMapper(options: MetricMapperOptions): MetricMapper {
 
   return {
     edge: config.edge,
+    tagCount: config.tags.length,
     mapValues: (values, timestamp) =>
       values.map((value, index) => ({
         edge: config.edge,

app/src/mqtt.ts

@@ -3,7 +3,7 @@ import mqtt from 'mqtt'
 import { getRegisteredTopics } from './registry.js'
 import { routeMessage } from './router.js'
 
-const mqttUrl = process.env.MQTT_URL as string
+const mqttUrl = process.env.MQTT_URL ?? 'mqtt://localhost:1883'
 
 export const mqttClient = mqtt.connect(mqttUrl)
 

app/src/stream/edge-chunk-relay.ts

@@ -1,7 +1,7 @@
 import type { IncomingMessage } from 'http'
 import { WebSocket, WebSocketServer } from 'ws'
 
-const wsPort = Number(process.env.WS_PORT)
+const wsPort = Number(process.env.WS_PORT ?? 9090)
 
 const wss = new WebSocketServer({ port: wsPort, perMessageDeflate: false })
 const clients = new Set<WebSocket>()

app/tsconfig.json

@@ -7,7 +7,8 @@
     "esModuleInterop": true,
     "skipLibCheck": true,
     "outDir": "dist",
-    "rootDir": "src"
+    "rootDir": "src",
+    "noEmit": true
   },
   "include": ["src/**/*.ts"]
 }

docker-compose.yml

@@ -13,7 +13,7 @@ services:
     depends_on:
       - mosquitto
     ports:
-      - "80:80"
+      # - "80:80"
       - "9090:9090"
     environment:
       MQTT_URL: mqtt://mosquitto:1883
@@ -21,7 +21,6 @@ services:
       WS_PORT: "9090"
       CLOUD_INGEST_URL: ${CLOUD_INGEST_URL:-https://demo.backend.drill.greact.ru/ingest}
       CLOUD_INGEST_API_KEY: ${CLOUD_INGEST_API_KEY:-}
-      DEMO_CLOUD_INGEST_URL: ${DEMO_CLOUD_INGEST_URL:-}
       EDGE5_MODBUS_V2_MAPPING_FILE: ${EDGE5_MODBUS_V2_MAPPING_FILE:-/app/src/mappings/edge5-modbus.json}
 
 volumes: