surface mqtt info in openapi

Author: pelikhan

local.env

@@ -6,4 +6,4 @@ DEVS_STORAGE_CONNECTION_STRING="DefaultEndpointsProtocol=http;AccountName=devsto
 DEVS_PASSWORDS_SECRET="DEVS_LOCAL_USER_PASSWORD"
 DEVS_LOCAL_USER_PASSWORD="devstoreaccount1:Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="
 DEVS_CONNECTION_STRING="AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;ApiRoot=$WEBSITE_PROTOCOL://$WEBSITE_HOSTNAME"
-DEVS_MQTT_CONNETION_STRING="mqtt://public.mqtthq.com:1883"
+DEVS_MQTT_SERVER="mqtt://public.mqtthq.com:1883"

src/apidevices.ts

@@ -31,6 +31,7 @@ import {
     stringifyMeta,
     updateDevice,
 } from "./storage"
+import { mqttTopicPrefix } from "./mqtt"
 
 const CONNECTED_TIMEOUT = 2 * 60 * 1000
 export const MAX_WSSK_SIZE = 230 // for to-device JSON and binary messages
@@ -45,6 +46,7 @@ function externalDevice(info: DeviceInfo) {
         conn,
         scriptId: info.scriptId,
         scriptVersion: info.scriptVersion,
+        mqttTopic: mqttTopicPrefix(info),
         deployedHash: info.deployedHash,
         lastAct: info.lastAct ? new Date(info.lastAct).toISOString() : "",
         meta: tryParseJSON(info.metaJSON),

src/apigateway.ts

@@ -0,0 +1,10 @@
+import { FastifyInstance } from "fastify"
+import { mqttServer } from "./mqtt"
+
+export async function initGatewayRoutes(server: FastifyInstance) {
+    server.get("/info", async req => {
+        return {
+            mqttServer: mqttServer(),
+        }
+    })
+}

src/index.ts

@@ -18,6 +18,7 @@ import { setup as storageSetup, defaultPartition } from "./storage"
 import { setup as envSetup } from "./env"
 import { setup as storageQueueSetup } from "./azure/storagequeue"
 import { setup as mqttSetup } from "./mqtt"
+import { initGatewayRoutes } from "./apigateway"
 
 async function initAuth(server: FastifyInstance) {
     console.log(`starting gateway...`)
@@ -162,6 +163,7 @@ async function main() {
 
             await initHubRoutes(server)
             await initScriptRoutes(server)
+            await initGatewayRoutes(server)
 
             server.all("*", async req => {
                 throwStatus(404, "no such API")

src/mqtt.ts

@@ -1,25 +1,36 @@
-import { connect } from "mqtt" // import connect from mqtt
+import { MqttClient, connect } from "mqtt" // import connect from mqtt
 import { serverTelemetry } from "./azure/appinsights"
 import { registerMessageSink } from "./messages"
 import { getSecret } from "./secrets"
 import { defaultPartition, getDevice } from "./storage"
 import { sendJSON } from "./apidevices"
 import { DeviceId } from "./schema"
 
+let client: MqttClient
+let serverUrl: string
+
+export function mqttTopicPrefix(deviceid: DeviceId) {
+    return client ? `devs/${deviceid.rowKey}` : undefined
+}
+
+export function mqttServer() {
+    return serverUrl
+}
+
 export async function setup() {
-    const connStr = await getSecret(
+    serverUrl = await getSecret(
         "mqttConnectionString",
-        "DEVS_MQTT_CONNECTION_STRING_SECRET",
-        "DEVS_MQTT_CONNETION_STRING"
+        "DEVS_MQTT_SERVER_SECRET",
+        "DEVS_MQTT_SERVER"
     )
-    if (!connStr) {
+    if (!serverUrl) {
         console.log("no MQTT connection string secret, skipping registration")
         return
     }
 
-    console.log("registering MQTT broker")
+    console.log(`MQTT server: ${serverUrl}`)
     const telemetry = serverTelemetry()
-    const client = connect(connStr) // create a client
+    client = connect(serverUrl) // create a client
 
     // device to mqtt
     registerMessageSink({
@@ -28,7 +39,7 @@ export async function setup() {
         ingest: async (topic, message, device) => {
             if (!client.connected) return
 
-            const mqTopic = `devs/from/${device.dev.rowKey}/${topic}`
+            const mqTopic = `${mqttTopicPrefix(device.dev)}/from/${topic}`
             client.publish(
                 mqTopic,
                 Buffer.from(JSON.stringify(message), "utf-8"),
@@ -46,18 +57,18 @@ export async function setup() {
     // dispatch messages to devices
     client.on("connect", () => {
         // devs/deviceid/json/topic
-        client.subscribe("devs/to/+/#", err => {
+        client.subscribe("devs/+/to/#", err => {
             if (err) {
                 console.error(err)
                 telemetry?.trackException({ exception: err })
             } else {
-                console.log(`mqtt: subscribe to 'devs/to/+/#'`)
+                console.log(`mqtt: subscribe to 'devs/+/to/#'`)
             }
         })
     })
     client.on("message", async function (topic, message) {
         const { deviceId, msgTopic } =
-            /^devs\/to\/(?<deviceId>.+?)\/(?<msgTopic>.+)$/.exec(topic)
+            /^devs\/(?<deviceId>.+?)\/to\/(?<msgTopic>.+)$/.exec(topic)
                 ?.groups || {}
         if (!deviceId || !msgTopic) return // unknown topic
         const did: DeviceId = {

src/swagger/openapi.ts

@@ -55,6 +55,16 @@ export function generateOpenApiSpec() {
 
     const methodParam = sString("Method name", "Must match the value on device")
 
+    const gatewaySchema = define(
+        "GatewaySchema",
+        sObj({
+            mqttServer: sString(
+                "MQTT server",
+                "URL(:PORT) of the MQTT server, if any"
+            ),
+        })
+    )
+
     const eventSchema = define(
         "EventSchema",
         sObj({
@@ -91,25 +101,17 @@ export function generateOpenApiSpec() {
                 "Last Connected",
                 "When was device last connected"
             ),
+            mqttTopic: sString(
+                "MQTT Topic prefix",
+                "Topic prefix to filter MQTT messages from/to this device"
+            ),
             ...allNonOptional(devProps),
             deployedHash: sString(
                 "SHA256 Hash of Currently Deployed Script",
                 "This property persists when the device is disconnected, so may not be up to date"
             ),
         })
     )
-    const serviceSchema = sString(
-        "Service Name",
-        "Type of sensor (eg. 'temperature')"
-    )
-    const sensorIdSchema = sString(
-        "Sensor ID",
-        "Unique identifier of the sensor"
-    )
-    const serviceIdxSchema = sNumber(
-        "Service Index",
-        "Typically 0; used when there is more than one instance of service in the sensor"
-    )
 
     post(
         "/hooks",
@@ -130,6 +132,19 @@ export function generateOpenApiSpec() {
         )
     )
 
+    get(
+        "/info",
+        action(
+            "Information",
+            "Information about the Gateway",
+            "Various information about the gateway server",
+            [],
+            {
+                "200": response("Information", gatewaySchema),
+            }
+        )
+    )
+
     get(
         "/devices",
         action(