mqtt-ingest draft

2026-06-04 14:40:18 +03:00
commit e7b44539ae
27 changed files with 1492 additions and 0 deletions
--- a/.gitignore
+++ b/.gitignore
@@ -0,0 +1,28 @@
+# dependencies
+node_modules/
+
+# build
+dist/
+
+# env & secrets
+.env
+.env.*
+
+# logs
+*.log
+mosquitto/log/*
+!mosquitto/log/.gitkeep
+
+# mosquitto runtime persistence (keep directory)
+mosquitto/data/*
+!mosquitto/data/.gitkeep
+
+# OS
+.DS_Store
+Thumbs.db
+
+# IDE
+.idea/
+.vscode/
+*.swp
+*.swo
--- a/app/.dockerignore
+++ b/app/.dockerignore
@@ -0,0 +1,2 @@
+node_modules
+dist
--- a/app/Dockerfile
+++ b/app/Dockerfile
@@ -0,0 +1,12 @@
+FROM node:22-alpine
+
+WORKDIR /app
+
+COPY package.json package-lock.json ./
+RUN npm ci
+
+COPY tsconfig.json ./
+COPY public ./public
+COPY src ./src
+
+CMD ["npm", "run", "dev"]
--- a/app/package-lock.json
+++ b/app/package-lock.json
--- a/app/package.json
+++ b/app/package.json
@@ -0,0 +1,18 @@
+{
+  "name": "mqtt-worker",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "dev": "tsx watch src/index.ts",
+    "start": "tsx src/index.ts",
+    "build": "tsc",
+    "typecheck": "tsc --noEmit"
+  },
+  "dependencies": {
+    "@types/node": "^25.9.1",
+    "mqtt": "^5.15.1",
+    "tsx": "^4.22.4",
+    "typescript": "^6.0.3",
+    "ws": "^8.21.0"
+  }
+}
--- a/app/public/index.html
+++ b/app/public/index.html
@@ -0,0 +1,21 @@
+<!DOCTYPE html>
+<html>
+<head>
+  <meta charset="utf-8" />
+  <title>edge5 video relay</title>
+  <script src="https://cdn.jsdelivr.net/npm/mpegts.js/dist/mpegts.js"></script>
+</head>
+<body>
+  <video id="v" controls autoplay muted style="width:100%"></video>
+  <script>
+    const wsUrl = `ws://${location.hostname}:9090`
+    const player = mpegts.createPlayer(
+      { type: 'mpegts', isLive: true, url: wsUrl },
+      { liveBufferLatencyChasing: true, liveBufferLatencyMaxLatency: 1.5 }
+    )
+    player.attachMediaElement(document.getElementById('v'))
+    player.load()
+    player.play()
+  </script>
+</body>
+</html>
--- a/app/src/handlers/edge5-video.ts
+++ b/app/src/handlers/edge5-video.ts
@@ -0,0 +1,13 @@
+import { onEdgeChunk } from '../stream/edge-chunk-relay.js'
+import type { TopicHandler } from '../types.js'
+
+export const TOPIC = 'data/edge5/video/v1'
+
+export const handleEdge5Video: TopicHandler = async (topic, payload) => {
+  console.log('edge5.video.received', {
+    topic,
+    bytes: payload.length,
+    payload,
+  })
+  onEdgeChunk(payload)
+}
--- a/app/src/handlers/index.ts
+++ b/app/src/handlers/index.ts
@@ -0,0 +1,14 @@
+import { register } from '../registry.js'
+import { handleTopic1, TOPIC as topic1 } from './topic1.js'
+import { handleTopic2, TOPIC as topic2 } from './topic2.js'
+import { handleTopic3, TOPIC as topic3 } from './topic3.js'
+import { handleTopic4, TOPIC as topic4 } from './topic4.js'
+import { handleTopic5, TOPIC as topic5 } from './topic5.js'
+import { handleEdge5Video, TOPIC as edge5Video } from './edge5-video.js'
+
+register(topic1, handleTopic1)
+register(topic2, handleTopic2)
+register(topic3, handleTopic3)
+register(topic4, handleTopic4)
+register(topic5, handleTopic5)
+register(edge5Video, handleEdge5Video)
--- a/app/src/handlers/topic1.ts
+++ b/app/src/handlers/topic1.ts
@@ -0,0 +1,16 @@
+import { parseJson } from '../helpers/parse.js'
+import { isArrayOfChunks125 } from '../helpers/validate.js'
+import type { TopicHandler } from '../types.js'
+
+export const TOPIC = 'poc/topic1'
+
+export const handleTopic1: TopicHandler = async (topic, payload) => {
+  const raw = parseJson(payload)
+
+  if (!isArrayOfChunks125(raw)) {
+    console.warn('topic1.invalid_payload', { topic })
+    return
+  }
+
+  console.log('topic1.received', { topic, chunks: raw.length })
+}
--- a/app/src/handlers/topic2.ts
+++ b/app/src/handlers/topic2.ts
@@ -0,0 +1,16 @@
+import { parseJson } from '../helpers/parse.js'
+import { isArrayOfLength125 } from '../helpers/validate.js'
+import type { TopicHandler } from '../types.js'
+
+export const TOPIC = 'poc/topic2'
+
+export const handleTopic2: TopicHandler = async (topic, payload) => {
+  const raw = parseJson(payload)
+
+  if (!isArrayOfLength125(raw)) {
+    console.warn('topic2.invalid_payload', { topic })
+    return
+  }
+
+  console.log('topic2.received', { topic, length: raw.length })
+}
--- a/app/src/handlers/topic3.ts
+++ b/app/src/handlers/topic3.ts
@@ -0,0 +1,16 @@
+import { parseUtf8 } from '../helpers/parse.js'
+import { isNonEmptyString } from '../helpers/validate.js'
+import type { TopicHandler } from '../types.js'
+
+export const TOPIC = 'poc/topic3'
+
+export const handleTopic3: TopicHandler = async (topic, payload) => {
+  const text = parseUtf8(payload)
+
+  if (!isNonEmptyString(text)) {
+    console.warn('topic3.invalid_payload', { topic })
+    return
+  }
+
+  console.log('topic3.received', { topic, text })
+}
--- a/app/src/handlers/topic4.ts
+++ b/app/src/handlers/topic4.ts
@@ -0,0 +1,29 @@
+import { parseJson } from '../helpers/parse.js'
+import { isFiniteNumber } from '../helpers/validate.js'
+import type { TopicHandler } from '../types.js'
+
+export const TOPIC = 'poc/topic4'
+
+type Topic4Payload = {
+  value: number
+}
+
+function isTopic4Payload(value: unknown): value is Topic4Payload {
+  return (
+    typeof value === 'object' &&
+    value !== null &&
+    'value' in value &&
+    isFiniteNumber((value as Topic4Payload).value)
+  )
+}
+
+export const handleTopic4: TopicHandler = async (topic, payload) => {
+  const raw = parseJson(payload)
+
+  if (!isTopic4Payload(raw)) {
+    console.warn('topic4.invalid_payload', { topic })
+    return
+  }
+
+  console.log('topic4.received', { topic, value: raw.value })
+}
--- a/app/src/handlers/topic5.ts
+++ b/app/src/handlers/topic5.ts
@@ -0,0 +1,10 @@
+import { parseJsonArray } from '../helpers/parse.js'
+import type { TopicHandler } from '../types.js'
+
+export const TOPIC = 'poc/topic5'
+
+export const handleTopic5: TopicHandler = async (topic, payload) => {
+  const items = parseJsonArray(payload)
+
+  console.log('topic5.received', { topic, count: items.length })
+}
--- a/app/src/helpers/parse.ts
+++ b/app/src/helpers/parse.ts
@@ -0,0 +1,20 @@
+export function parseUtf8(payload: Buffer): string {
+  return payload.toString('utf8')
+}
+
+export function parseJson<T = unknown>(payload: Buffer): T {
+  const text = parseUtf8(payload)
+  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')
+}
--- a/app/src/helpers/validate.ts
+++ b/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}`)
+  }
+}
--- a/app/src/http/server.ts
+++ b/app/src/http/server.ts
@@ -0,0 +1,28 @@
+import { readFile } from 'node:fs/promises'
+import { createServer } from 'node:http'
+import { dirname, join } from 'node:path'
+import { fileURLToPath } from 'node:url'
+
+const httpPort = Number(process.env.HTTP_PORT ?? 80)
+const publicDir = join(dirname(fileURLToPath(import.meta.url)), '../../public')
+const indexPath = join(publicDir, 'index.html')
+
+createServer(async (req, res) => {
+  const path = req.url?.split('?')[0]
+
+  if (path === '/' || path === '/index.html') {
+    try {
+      const html = await readFile(indexPath, 'utf8')
+      res.writeHead(200, { 'Content-Type': 'text/html; charset=utf-8' })
+      res.end(html)
+    } catch (err) {
+      console.error('http.read_index_error', err)
+      res.writeHead(500).end('Internal Server Error')
+    }
+    return
+  }
+
+  res.writeHead(404).end('Not Found')
+}).listen(httpPort, () => {
+  console.log('http.server_started', { port: httpPort })
+})
--- a/app/src/index.ts
+++ b/app/src/index.ts
@@ -0,0 +1,6 @@
+import './handlers/index.js'
+import './http/server.js'
+import './mqtt.js'
+import './stream/edge-chunk-relay.js'
+
+console.log('mqtt-worker.started')
--- a/app/src/mqtt.ts
+++ b/app/src/mqtt.ts
@@ -0,0 +1,38 @@
+import mqtt from 'mqtt'
+
+import { getRegisteredTopics } from './registry.js'
+import { routeMessage } from './router.js'
+
+const mqttUrl = process.env.MQTT_URL ?? 'mqtt://localhost:1883'
+
+export const mqttClient = mqtt.connect(mqttUrl)
+
+mqttClient.on('connect', () => {
+  console.log('mqtt.connected', { url: mqttUrl })
+
+  const topics = getRegisteredTopics()
+  if (topics.length === 0) {
+    console.warn('mqtt.no_topics_registered')
+    return
+  }
+
+  mqttClient.subscribe(topics, (err) => {
+    if (err) {
+      console.error('mqtt.subscribe_error', err)
+      return
+    }
+    console.log('mqtt.subscribed', { topics })
+  })
+})
+
+mqttClient.on('message', (topic, payload) => {
+  void routeMessage(topic, payload)
+})
+
+mqttClient.on('error', (err) => {
+  console.error('mqtt.error', err)
+})
+
+mqttClient.on('reconnect', () => {
+  console.log('mqtt.reconnect')
+})
--- a/app/src/registry.ts
+++ b/app/src/registry.ts
@@ -0,0 +1,18 @@
+import type { TopicHandler } from './types.js'
+
+const handlers = new Map<string, TopicHandler>()
+
+export function register(topic: string, handler: TopicHandler): void {
+  if (handlers.has(topic)) {
+    throw new Error(`Handler already registered for topic: ${topic}`)
+  }
+  handlers.set(topic, handler)
+}
+
+export function getHandler(topic: string): TopicHandler | undefined {
+  return handlers.get(topic)
+}
+
+export function getRegisteredTopics(): string[] {
+  return [...handlers.keys()]
+}
--- a/app/src/router.ts
+++ b/app/src/router.ts
@@ -0,0 +1,19 @@
+import { getHandler } from './registry.js'
+
+export async function routeMessage(
+  topic: string,
+  payload: Buffer,
+): Promise<void> {
+  const handler = getHandler(topic)
+
+  if (!handler) {
+    console.warn('mqtt.unhandled_topic', { topic })
+    return
+  }
+
+  try {
+    await handler(topic, payload)
+  } catch (err) {
+    console.error('mqtt.handler_error', { topic, err })
+  }
+}
--- a/app/src/stream/edge-chunk-relay.ts
+++ b/app/src/stream/edge-chunk-relay.ts
@@ -0,0 +1,32 @@
+import type { IncomingMessage } from 'http'
+import { WebSocket, WebSocketServer } from 'ws'
+
+const wsPort = Number(process.env.WS_PORT ?? 9090)
+
+const wss = new WebSocketServer({ port: wsPort, perMessageDeflate: false })
+const clients = new Set<WebSocket>()
+
+wss.on('connection', (ws: WebSocket, req: IncomingMessage) => {
+  clients.add(ws)
+  console.log('ws.client_connected', {
+    clients: clients.size,
+    ip: req.socket.remoteAddress,
+  })
+
+  ws.on('close', () => {
+    clients.delete(ws)
+    console.log('ws.client_closed', { clients: clients.size })
+  })
+})
+
+wss.on('listening', () => {
+  console.log('ws.server_started', { port: wsPort })
+})
+
+export function onEdgeChunk(chunk: Buffer): void {
+  for (const client of clients) {
+    if (client.readyState === WebSocket.OPEN) {
+      client.send(chunk)
+    }
+  }
+}
--- a/app/src/types.ts
+++ b/app/src/types.ts
@@ -0,0 +1,9 @@
+export type TopicHandler = (
+  topic: string,
+  payload: Buffer,
+) => Promise<void>
+
+export type TopicSubscription =
+  | string
+  | string[]
+  | Record<string, { qos?: 0 | 1 | 2 }>
--- a/app/tsconfig.json
+++ b/app/tsconfig.json
@@ -0,0 +1,14 @@
+{
+  "compilerOptions": {
+    "target": "ES2022",
+    "module": "NodeNext",
+    "moduleResolution": "NodeNext",
+    "strict": true,
+    "esModuleInterop": true,
+    "skipLibCheck": true,
+    "outDir": "dist",
+    "rootDir": "src",
+    "noEmit": true
+  },
+  "include": ["src/**/*.ts"]
+}
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -0,0 +1,27 @@
+services:
+  mosquitto:
+    image: eclipse-mosquitto:2
+    container_name: mosquitto
+    ports:
+      - "1883:1883"
+    volumes:
+      - ./mosquitto/config:/mosquitto/config
+      - ./mosquitto/data:/mosquitto/data
+      - ./mosquitto/log:/mosquitto/log
+
+  app:
+    build: ./app
+    container_name: mqtt-worker
+    depends_on:
+      - mosquitto
+    ports:
+      - "80:80"
+      - "9090:9090"
+    environment:
+      MQTT_URL: mqtt://mosquitto:1883
+      HTTP_PORT: "80"
+      WS_PORT: "9090"
+    volumes:
+      - ./app:/app
+      # Linux node_modules из образа; без этого bind ./app затирает их win32-бинарниками с хоста
+      - /app/node_modules
--- a/mosquitto/config/mosquitto.conf
+++ b/mosquitto/config/mosquitto.conf
@@ -0,0 +1,7 @@
+listener 1883
+allow_anonymous true
+
+persistence true
+persistence_location /mosquitto/data/
+
+log_dest file /mosquitto/log/mosquitto.log
--- a/mosquitto/data/.gitkeep
+++ b/mosquitto/data/.gitkeep
--- a/mosquitto/log/.gitkeep
+++ b/mosquitto/log/.gitkeep