feat(daemon): add HTTP ingest listener with ChirpStack JSON parsing and InfluxDB writer

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

LORAWEB_NEU/backend/daemon/src/http.rs

@@ -0,0 +1,119 @@
+// backend/daemon/src/http.rs
+use axum::{
+    extract::{Query, State},
+    http::StatusCode,
+    response::IntoResponse,
+    routing::post,
+    Json, Router,
+};
+use serde::Deserialize;
+use serde_json::Value;
+use sqlx::SqlitePool;
+use std::sync::Arc;
+use crate::influx::InfluxWriter;
+
+pub struct IngestState {
+    pub pool: SqlitePool,
+    pub influx: InfluxWriter,
+}
+
+#[derive(Deserialize)]
+pub struct EventQuery {
+    pub event: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct ChirpStackUplink {
+    time: Option<String>,
+    #[serde(rename = "deviceInfo")]
+    device_info: Option<DeviceInfo>,
+    dr: Option<i64>,
+    #[serde(rename = "fCnt")]
+    f_cnt: Option<i64>,
+    #[serde(rename = "fPort")]
+    f_port: Option<i64>,
+    #[serde(rename = "rxInfo")]
+    rx_info: Option<Vec<RxInfo>>,
+    object: Option<serde_json::Map<String, Value>>,
+}
+
+#[derive(Deserialize)]
+struct DeviceInfo {
+    #[serde(rename = "devEui")]
+    dev_eui: String,
+    #[serde(rename = "deviceName")]
+    device_name: Option<String>,
+    #[serde(rename = "applicationName")]
+    application_name: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct RxInfo {
+    rssi: Option<f64>,
+    snr: Option<f64>,
+}
+
+pub fn router(state: Arc<IngestState>) -> Router {
+    Router::new()
+        .route("/", post(handle_event))
+        .with_state(state)
+}
+
+async fn handle_event(
+    State(state): State<Arc<IngestState>>,
+    Query(query): Query<EventQuery>,
+    Json(body): Json<Value>,
+) -> impl IntoResponse {
+    let event = query.event.unwrap_or_default();
+
+    if event != "up" {
+        tracing::debug!(event = %event, "Non-uplink event received, ignoring");
+        return StatusCode::OK;
+    }
+
+    let uplink: ChirpStackUplink = match serde_json::from_value(body) {
+        Ok(u) => u,
+        Err(e) => {
+            tracing::warn!("Failed to parse uplink JSON: {}", e);
+            return StatusCode::BAD_REQUEST;
+        }
+    };
+
+    let device_info = match uplink.device_info {
+        Some(di) => di,
+        None => {
+            tracing::warn!("Uplink missing deviceInfo");
+            return StatusCode::BAD_REQUEST;
+        }
+    };
+
+    let dev_eui = &device_info.dev_eui;
+    let device_name = device_info.device_name.as_deref().unwrap_or("");
+    let app_name = device_info.application_name.as_deref().unwrap_or("");
+    let timestamp = uplink.time.as_deref().unwrap_or("");
+
+    if let Err(e) = crate::db::register_topic(&state.pool, dev_eui).await {
+        tracing::warn!("Failed to register topic {}: {}", dev_eui, e);
+    }
+
+    if !timestamp.is_empty() {
+        if let Err(e) = crate::db::update_last_seen(&state.pool, dev_eui, timestamp).await {
+            tracing::warn!("Failed to update last_seen for {}: {}", dev_eui, e);
+        }
+    }
+
+    if let Some(object) = &uplink.object {
+        let rx = uplink.rx_info.as_ref().and_then(|r| r.first());
+        let rssi = rx.and_then(|r| r.rssi);
+        let snr = rx.and_then(|r| r.snr);
+
+        if let Err(e) = state.influx.write_uplink(
+            dev_eui, device_name, app_name, timestamp,
+            object, rssi, snr, uplink.dr, uplink.f_cnt, uplink.f_port,
+        ).await {
+            tracing::warn!("InfluxDB write failed for {}: {}", dev_eui, e);
+        }
+    }
+
+    StatusCode::OK
+}

LORAWEB_NEU/backend/daemon/src/influx.rs

@@ -0,0 +1,104 @@
+// backend/daemon/src/influx.rs
+use serde_json::Value;
+use chrono::DateTime;
+
+pub struct InfluxWriter {
+    client: reqwest::Client,
+    url: String,
+    token: String,
+    org: String,
+    bucket: String,
+}
+
+impl InfluxWriter {
+    pub fn new(url: &str, token: &str, org: &str, bucket: &str) -> Self {
+        Self {
+            client: reqwest::Client::new(),
+            url: url.to_string(),
+            token: token.to_string(),
+            org: org.to_string(),
+            bucket: bucket.to_string(),
+        }
+    }
+
+    pub async fn write_uplink(
+        &self,
+        dev_eui: &str,
+        device_name: &str,
+        application_name: &str,
+        timestamp: &str,
+        object: &serde_json::Map<String, Value>,
+        rssi: Option<f64>,
+        snr: Option<f64>,
+        dr: Option<i64>,
+        f_cnt: Option<i64>,
+        f_port: Option<i64>,
+    ) -> Result<(), Box<dyn std::error::Error>> {
+        let ts_nanos = DateTime::parse_from_rfc3339(timestamp)?
+            .timestamp_nanos_opt()
+            .unwrap_or(0);
+
+        let tags = format!(
+            "sensor_data,dev_eui={},device_name={},application_name={}",
+            escape_tag(dev_eui),
+            escape_tag(device_name),
+            escape_tag(application_name),
+        );
+
+        let mut fields = Vec::new();
+
+        for (key, val) in object {
+            match val {
+                Value::Number(n) => {
+                    if let Some(f) = n.as_f64() {
+                        fields.push(format!("{}={}", escape_tag(key), f));
+                    }
+                }
+                Value::Bool(b) => {
+                    fields.push(format!("{}={}", escape_tag(key), b));
+                }
+                Value::String(s) => {
+                    fields.push(format!("{}=\"{}\"", escape_tag(key), escape_field_value(s)));
+                }
+                _ => {}
+            }
+        }
+
+        if let Some(v) = rssi { fields.push(format!("rssi={}", v)); }
+        if let Some(v) = snr { fields.push(format!("snr={}", v)); }
+        if let Some(v) = dr { fields.push(format!("dr={}i", v)); }
+        if let Some(v) = f_cnt { fields.push(format!("f_cnt={}i", v)); }
+        if let Some(v) = f_port { fields.push(format!("f_port={}i", v)); }
+
+        if fields.is_empty() {
+            return Ok(());
+        }
+
+        let line = format!("{} {} {}", tags, fields.join(","), ts_nanos);
+
+        let resp = self.client
+            .post(format!("{}/api/v2/write?org={}&bucket={}&precision=ns",
+                self.url, self.org, self.bucket))
+            .header("Authorization", format!("Token {}", self.token))
+            .header("Content-Type", "text/plain")
+            .body(line)
+            .send()
+            .await?;
+
+        if !resp.status().is_success() {
+            let status = resp.status();
+            let body = resp.text().await.unwrap_or_default();
+            return Err(format!("InfluxDB write failed ({}): {}", status, body).into());
+        }
+
+        Ok(())
+    }
+}
+
+fn escape_tag(s: &str) -> String {
+    s.replace(',', "\\,").replace('=', "\\=").replace(' ', "\\ ")
+}
+
+fn escape_field_value(s: &str) -> String {
+    s.replace('\\', "\\\\").replace('"', "\\\"")
+}