From d94bfe471eca7866e8aa8e4969a7edfc97fcc53c Mon Sep 17 00:00:00 2001
From: liamcottle <liam@liamcottle.com>
Date: Sat, 16 Mar 2024 18:32:19 +1300
Subject: [PATCH] collect service envelopes

---
 .../migration.sql                             | 19 +++++++++++++++++
 prisma/schema.prisma                          | 17 +++++++++++++++
 src/mqtt.js                                   | 21 +++++++++++++++++++
 3 files changed, 57 insertions(+)
 create mode 100644 prisma/migrations/20240316052726_create_service_envelopes_table/migration.sql

diff --git a/prisma/migrations/20240316052726_create_service_envelopes_table/migration.sql b/prisma/migrations/20240316052726_create_service_envelopes_table/migration.sql
new file mode 100644
index 0000000..b8c4087
--- /dev/null
+++ b/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`);
diff --git a/prisma/schema.prisma b/prisma/schema.prisma
index a81a180..d925233 100644
--- a/prisma/schema.prisma
+++ b/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
diff --git a/src/mqtt.js b/src/mqtt.js
index 176cb73..9274326 100644
--- a/src/mqtt.js
+++ b/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;