Refactor MQTT client setup; streamline connection logic and enhance message handling

Author: Bagus-DevLab

app/main.py

@@ -8,7 +8,7 @@
 # Import Module Kita
 from app.core.database import engine, get_db, Base
 from app.models.device import Device
-from app.mqtt.client import connect_mqtt
+from app.mqtt.client import start_mqtt
 from app.api.v1.devices import router as device_router
 from app.core.limiter import limiter
 

app/mqtt/callbacks.py

@@ -1,35 +1,47 @@
-import paho.mqtt.client as mqtt
-from app.core.config import settings # Asumsi config ada di sini
-from app.mqtt.callbacks import on_message # Import logika langkah 2
+from sqlalchemy.orm import Session
+from app.core.database import SessionLocal
+from app.models.device import SensorLog
 
-# Global variable buat client
-mqtt_client = mqtt.Client(client_id="Backend_Listener_Worker", protocol=mqtt.MQTTv311)
-
-def start_mqtt():
-    # 1. Setup Auth (Sesuaikan dengan kredensial EMQX kamu)
-    # Lebih baik ambil dari .env/config, tapi hardcode dulu buat test gapapa
-    MQTT_BROKER = "k2519aa6.ala.asia-southeast1.emqxsl.com"
-    MQTT_PORT = 8883
-    MQTT_USER = "PCB01"
-    MQTT_PASS = "5ywnMzsVX4Ss9vH"
-
-    mqtt_client.username_pw_set(MQTT_USER, MQTT_PASS)
-    mqtt_client.tls_set() # Wajib SSL untuk port 8883
-
-    # 2. Hubungkan logic Callback
-    mqtt_client.on_message = on_message
+# --- FUNGSI SIMPAN KE DB ---
+def save_sensor_data(device_id: str, topic_type: str, payload: str):
+    db: Session = SessionLocal()
+    try:
+        new_log = SensorLog(
+            device_id=device_id,
+            topic_type=topic_type,
+            value=payload
+        )
+        db.add(new_log)
+        db.commit()
+        print(f"💾 Saved DB: [{device_id}] {topic_type} -> {payload}")
+    except Exception as e:
+        print(f"❌ Gagal Simpan DB: {e}")
+        db.rollback()
+    finally:
+        db.close()
 
-    # 3. Konek
+# --- CALLBACK SAAT PESAN MASUK ---
+def on_message(client, userdata, msg):
     try:
-        mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)
+        topic = msg.topic
+        payload = msg.payload.decode()
 
-        # 4. Subscribe Dynamic Topic
-        # Tanda '+' adalah wildcard. 
-        # Artinya kita dengar semua alat di topik status
-        mqtt_client.subscribe("alat/+/status/#")
+        # LOGIKA MEMBEDAH TOPIK DINAMIS
+        # Format: alat/{DEVICE_ID}/status/{TIPE}
+        # Contoh: alat/441D64BE2208/status/suhu
 
-        mqtt_client.loop_start() # Jalankan di background thread
-        print("✅ MQTT Listener Berjalan! Mendengarkan: alat/+/status/#")
+        parts = topic.split("/")
 
+        # Validasi: Harus ada 4 bagian, depannya 'alat', tengahnya 'status'
+        if len(parts) == 4 and parts[0] == "alat" and parts[2] == "status":
+            device_id = parts[1]   # Ambil ID (441D...)
+            sensor_type = parts[3] # Ambil Tipe (suhu/ldr/lampu)
+            
+            # Simpan ke Database
+            save_sensor_data(device_id, sensor_type, payload)
+            
+        else:
+            print(f"⚠️ Topik tidak sesuai format: {topic}")
+
     except Exception as e:
-        print(f"❌ Gagal Konek MQTT: {e}")
+        print(f"Error processing MQTT: {e}")

app/mqtt/client.py

@@ -1,52 +1,40 @@
 import paho.mqtt.client as mqtt
-# Jika kamu belum punya app.core.config, kita hardcode dulu credentials-nya
-# from app.core.config import settings 
 from app.mqtt.callbacks import on_message 
 
-# Variabel Global Client
+# Global Client
 mqtt_client = mqtt.Client(client_id="Backend_Listener_Worker", protocol=mqtt.MQTTv311)
 
 def start_mqtt():
-    """
-    Fungsi ini dipanggil oleh main.py saat server start.
-    Tugasnya konek ke Broker dan mulai loop listener.
-    """
-    # 1. Setup Auth (Sesuaikan dengan kredensial EMQX kamu)
+    # 1. SETUP KREDENSIAL (Hardcode dulu biar aman dari error import)
     MQTT_BROKER = "k2519aa6.ala.asia-southeast1.emqxsl.com"
     MQTT_PORT = 8883
     MQTT_USER = "PCB01"
     MQTT_PASS = "5ywnMzsVX4Ss9vH"
 
-    print(f"🔌 Menghubungkan ke MQTT Broker: {MQTT_BROKER}...")
+    print(f"🔌 Menghubungkan ke EMQX: {MQTT_BROKER}...")
 
-    # Set Username & Password
+    # 2. CONFIG CLIENT
     mqtt_client.username_pw_set(MQTT_USER, MQTT_PASS)
-    
-    # Wajib SSL untuk port 8883
-    mqtt_client.tls_set() 
+    mqtt_client.tls_set() # WAJIB SSL KARENA PORT 8883
+    mqtt_client.on_message = on_message # Sambungkan ke otak logic tadi
 
-    # 2. Hubungkan logic Callback (PENTING!)
-    mqtt_client.on_message = on_message
-
-    # 3. Konek & Subscribe
+    # 3. CONNECT & SUBSCRIBE
     try:
         mqtt_client.connect(MQTT_BROKER, MQTT_PORT, 60)
 
-        # Subscribe ke Topik Dinamis
+        # SUBSCRIBE WILDCARD
+        # "+" artinya ID alat apa saja
+        # "#" artinya status apa saja (suhu, ldr, dll)
         mqtt_client.subscribe("alat/+/status/#")
 
-        # Jalankan di background thread
-        mqtt_client.loop_start() 
-        
+        mqtt_client.loop_start() # Jalan di background
         print("✅ MQTT Listener Berjalan! Mendengarkan: alat/+/status/#")
 
     except Exception as e:
         print(f"❌ Gagal Konek MQTT: {e}")
 
-# Fungsi helper untuk publish (dipakai nanti buat control)
+# Fungsi helper untuk control (dipakai nanti)
 def publish(topic: str, payload: str):
     if mqtt_client.is_connected():
         mqtt_client.publish(topic, payload)
-        print(f"📡 Published: {topic} -> {payload}")
-    else:
-        print("⚠️ MQTT belum terhubung, gagal publish.")
+        print(f"📡 Backend Kirim: {topic} -> {payload}")