collect text messages

prisma/migrations/20240314102230_create_text_messages_table/migration.sql

@@ -0,0 +1,25 @@
+-- CreateTable
+CREATE TABLE `text_messages` (
+    `id` BIGINT NOT NULL AUTO_INCREMENT,
+    `to` BIGINT NOT NULL,
+    `from` BIGINT NOT NULL,
+    `channel` INTEGER NOT NULL,
+    `packet_id` BIGINT NOT NULL,
+    `channel_id` VARCHAR(191) NOT NULL,
+    `gateway_id` BIGINT NULL,
+    `text` VARCHAR(191) NOT NULL,
+    `rx_time` BIGINT NULL,
+    `rx_snr` DECIMAL(65, 30) NULL,
+    `rx_rssi` INTEGER NULL,
+    `hop_limit` INTEGER NULL,
+    `created_at` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+    `updated_at` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+
+    INDEX `text_messages_created_at_idx`(`created_at`),
+    INDEX `text_messages_updated_at_idx`(`updated_at`),
+    INDEX `text_messages_to_idx`(`to`),
+    INDEX `text_messages_from_idx`(`from`),
+    INDEX `text_messages_packet_id_idx`(`packet_id`),
+    INDEX `text_messages_gateway_id_idx`(`gateway_id`),
+    PRIMARY KEY (`id`)
+) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;

prisma/schema.prisma

@@ -100,6 +100,34 @@ model DeviceMetric {
   @@map("device_metrics")
 }
 
+model TextMessage {
+  id         BigInt  @id @default(autoincrement())
+  to         BigInt
+  from       BigInt
+  channel    Int
+  packet_id  BigInt
+  channel_id String
+  gateway_id BigInt?
+
+  text String
+
+  rx_time   BigInt?
+  rx_snr    Decimal?
+  rx_rssi   Int?
+  hop_limit Int?
+
+  created_at DateTime @default(now())
+  updated_at DateTime @default(now()) @updatedAt
+
+  @@index(created_at)
+  @@index(updated_at)
+  @@index(to)
+  @@index(from)
+  @@index(packet_id)
+  @@index(gateway_id)
+  @@map("text_messages")
+}
+
 model TraceRoute {
   id      BigInt @id @default(autoincrement())
   node_id BigInt

src/mqtt.js

@@ -99,7 +99,39 @@ client.on("message", async (topic, message) => {
         const logUnknownPacketTypes = false;
         const portnum = envelope.packet?.decoded?.portnum;
 
-        if(portnum === 3) {
+        if(portnum === 1) {
+
+            if(logKnownPacketTypes) {
+                console.log("TEXT_MESSAGE_APP", {
+                    to: envelope.packet.to.toString(16),
+                    from: envelope.packet.from.toString(16),
+                    text: envelope.packet.decoded.payload.toString(),
+                });
+            }
+
+            try {
+                await prisma.textMessage.create({
+                    data: {
+                        to: envelope.packet.to,
+                        from: envelope.packet.from,
+                        channel: envelope.packet.channel,
+                        packet_id: envelope.packet.id,
+                        channel_id: envelope.channelId,
+                        gateway_id: envelope.gatewayId ? BigInt('0x' + envelope.gatewayId.replaceAll("!", "")) : null, // convert hex id "!f96a92f0" to bigint
+                        text: envelope.packet.decoded.payload.toString(),
+                        rx_time: envelope.packet.rxTime,
+                        rx_snr: envelope.packet.rxSnr,
+                        rx_rssi: envelope.packet.rxRssi,
+                        hop_limit: envelope.packet.hopLimit,
+                    },
+                });
+            } catch (e) {
+                console.error(e);
+            }
+
+        }
+
+        else if(portnum === 3) {
 
             const position = Position.decode(envelope.packet.decoded.payload);