From de6e92c6f58886ce43604ae651959053528c2741 Mon Sep 17 00:00:00 2001 From: Christian Mueller Date: Thu, 19 Mar 2026 22:54:27 +0100 Subject: [PATCH] feat(daemon): add HTTP ingest listener with ChirpStack JSON parsing and InfluxDB writer Co-Authored-By: Claude Sonnet 4.6 --- LORAWEB_NEU/backend/daemon/src/http.rs | 119 +++++++++++++++++++++++ LORAWEB_NEU/backend/daemon/src/influx.rs | 104 ++++++++++++++++++++ 2 files changed, 223 insertions(+) create mode 100644 LORAWEB_NEU/backend/daemon/src/http.rs create mode 100644 LORAWEB_NEU/backend/daemon/src/influx.rs diff --git a/LORAWEB_NEU/backend/daemon/src/http.rs b/LORAWEB_NEU/backend/daemon/src/http.rs new file mode 100644 index 0000000..718c10a --- /dev/null +++ b/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, +} + +#[derive(Deserialize)] +struct ChirpStackUplink { + time: Option, + #[serde(rename = "deviceInfo")] + device_info: Option, + dr: Option, + #[serde(rename = "fCnt")] + f_cnt: Option, + #[serde(rename = "fPort")] + f_port: Option, + #[serde(rename = "rxInfo")] + rx_info: Option>, + object: Option>, +} + +#[derive(Deserialize)] +struct DeviceInfo { + #[serde(rename = "devEui")] + dev_eui: String, + #[serde(rename = "deviceName")] + device_name: Option, + #[serde(rename = "applicationName")] + application_name: Option, +} + +#[derive(Deserialize)] +struct RxInfo { + rssi: Option, + snr: Option, +} + +pub fn router(state: Arc) -> Router { + Router::new() + .route("/", post(handle_event)) + .with_state(state) +} + +async fn handle_event( + State(state): State>, + Query(query): Query, + Json(body): Json, +) -> 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 +} diff --git a/LORAWEB_NEU/backend/daemon/src/influx.rs b/LORAWEB_NEU/backend/daemon/src/influx.rs new file mode 100644 index 0000000..6547c88 --- /dev/null +++ b/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, + rssi: Option, + snr: Option, + dr: Option, + f_cnt: Option, + f_port: Option, + ) -> Result<(), Box> { + 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('"', "\\\"") +}