collect service envelopes

prisma/migrations/20240316052726_create_service_envelopes_table/migration.sql

@@ -0,0 +1,19 @@
+-- CreateTable
+CREATE TABLE `service_envelopes` (
+    `id` BIGINT NOT NULL AUTO_INCREMENT,
+    `mqtt_topic` VARCHAR(191) NOT NULL,
+    `channel_id` VARCHAR(191) NOT NULL,
+    `gateway_id` BIGINT NULL,
+    `to` BIGINT NOT NULL,
+    `from` BIGINT NOT NULL,
+    `protobuf` LONGBLOB NOT NULL,
+    `created_at` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+    `updated_at` DATETIME(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3),
+
+    INDEX `service_envelopes_created_at_idx`(`created_at`),
+    INDEX `service_envelopes_updated_at_idx`(`updated_at`),
+    PRIMARY KEY (`id`)
+) DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci;
+
+-- CreateIndex
+CREATE INDEX `waypoints_to_idx` ON `waypoints`(`to`);

prisma/schema.prisma

@@ -100,6 +100,23 @@ model DeviceMetric {
   @@map("device_metrics")
 }
 
+model ServiceEnvelope {
+  id         BigInt  @id @default(autoincrement())
+  mqtt_topic String
+  channel_id String
+  gateway_id BigInt?
+  to         BigInt
+  from       BigInt
+  protobuf   Bytes
+
+  created_at DateTime @default(now())
+  updated_at DateTime @default(now()) @updatedAt
+
+  @@index(created_at)
+  @@index(updated_at)
+  @@map("service_envelopes")
+}
+
 model TextMessage {
   id         BigInt  @id @default(autoincrement())
   to         BigInt

src/mqtt.js

@@ -86,6 +86,27 @@ client.on("message", async (topic, message) => {
 
         // decode service envelope
         const envelope = ServiceEnvelope.decode(message);
+        if(!envelope.packet){
+            return;
+        }
+
+        // create service envelope in db
+        try {
+            await prisma.serviceEnvelope.create({
+                data: {
+                    mqtt_topic: topic,
+                    channel_id: envelope.channelId,
+                    gateway_id: envelope.gatewayId ? BigInt('0x' + envelope.gatewayId.replaceAll("!", "")) : null, // convert hex id "!f96a92f0" to bigint
+                    to: envelope.packet.to,
+                    from: envelope.packet.from,
+                    protobuf: message,
+                },
+            });
+        } catch (e) {
+            console.error(e, {
+                envelope: envelope.packet,
+            });
+        }
 
         // attempt to decrypt encrypted packets
         const isEncrypted = envelope.packet.encrypted?.length > 0;