feat(api): implement complete REST API with auth, CRUD, InfluxDB proxy, and user management

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

LORAWEB_NEU/backend/api/src/alarms.rs

@@ -0,0 +1,99 @@
+use axum::{
+    extract::State,
+    http::StatusCode,
+    routing::get,
+    Json, Router,
+};
+use serde::Serialize;
+use crate::AppState;
+
+#[derive(Debug, Serialize)]
+struct AlarmSummary {
+    active: i64,
+    warning: i64,
+    alarm: i64,
+    unknown: i64,
+    alarms: Vec<AlarmEntry>,
+}
+
+#[derive(Debug, Serialize, sqlx::FromRow)]
+struct AlarmEntry {
+    topic_id: String,
+    topic: String,
+    alarm_status: String,
+    last_seen: Option<String>,
+    sensor_name: Option<String>,
+}
+
+#[derive(Debug, sqlx::FromRow)]
+struct StatusCount {
+    alarm_status: String,
+    count: i64,
+}
+
+type ApiError = (StatusCode, Json<serde_json::Value>);
+
+fn internal_err(msg: &str) -> ApiError {
+    (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": msg })))
+}
+
+pub fn routes() -> Router<AppState> {
+    Router::new().route("/alarms", get(get_alarms))
+}
+
+async fn get_alarms(
+    State(state): State<AppState>,
+) -> Result<Json<AlarmSummary>, ApiError> {
+    // Get counts per alarm status
+    let counts = sqlx::query_as::<_, StatusCount>(
+        "SELECT alarm_status, COUNT(*) as count FROM topics GROUP BY alarm_status"
+    )
+    .fetch_all(&state.pool)
+    .await
+    .map_err(|_| internal_err("Database error"))?;
+
+    let mut active: i64 = 0;
+    let mut warning: i64 = 0;
+    let mut alarm: i64 = 0;
+    let mut unknown: i64 = 0;
+
+    for row in &counts {
+        match row.alarm_status.as_str() {
+            "ok" => active += row.count,
+            "warning" => {
+                active += row.count;
+                warning += row.count;
+            }
+            "alarm" => {
+                active += row.count;
+                alarm += row.count;
+            }
+            _ => unknown += row.count,
+        }
+    }
+
+    // Get alarm entries (warning + alarm status) with sensor name via LEFT JOIN
+    let alarms = sqlx::query_as::<_, AlarmEntry>(
+        r#"SELECT
+            t.id as topic_id,
+            t.topic,
+            t.alarm_status,
+            t.last_seen,
+            s.name as sensor_name
+           FROM topics t
+           LEFT JOIN sensors s ON s.topic_id = t.id
+           WHERE t.alarm_status IN ('warning', 'alarm')
+           ORDER BY t.alarm_status DESC, t.last_seen ASC"#
+    )
+    .fetch_all(&state.pool)
+    .await
+    .map_err(|_| internal_err("Database error"))?;
+
+    Ok(Json(AlarmSummary {
+        active,
+        warning,
+        alarm,
+        unknown,
+        alarms,
+    }))
+}

LORAWEB_NEU/backend/api/src/auth.rs

@@ -0,0 +1,147 @@
+use axum::{
+    extract::State,
+    http::{Request, StatusCode},
+    middleware::Next,
+    response::{IntoResponse, Response},
+    routing::post,
+    Json, Router,
+};
+use argon2::{Argon2, PasswordHash, PasswordVerifier};
+use jsonwebtoken::{decode, encode, DecodingKey, EncodingKey, Header, Validation};
+use serde::{Deserialize, Serialize};
+use chrono::{Utc, Duration};
+use crate::AppState;
+
+#[derive(Debug, Serialize, Deserialize, Clone)]
+pub struct Claims {
+    pub sub: i64,
+    pub role: String,
+    pub exp: i64,
+    #[serde(default)]
+    pub purpose: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct LoginRequest {
+    username: Option<String>,
+    password: Option<String>,
+    totp_token: Option<String>,
+    totp_code: Option<String>,
+}
+
+#[derive(Serialize)]
+struct LoginResponse {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    token: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    totp_required: Option<bool>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    totp_token: Option<String>,
+}
+
+pub fn routes() -> Router<AppState> {
+    Router::new().route("/auth/login", post(login))
+}
+
+async fn login(
+    State(state): State<AppState>,
+    Json(req): Json<LoginRequest>,
+) -> Result<Json<LoginResponse>, (StatusCode, Json<serde_json::Value>)> {
+    let err = |msg: &str| (StatusCode::UNAUTHORIZED, Json(serde_json::json!({ "error": msg })));
+
+    // TOTP step 2
+    if let (Some(totp_token), Some(totp_code)) = (&req.totp_token, &req.totp_code) {
+        let claims = decode_jwt(&state.jwt_secret, totp_token).map_err(|_| err("Invalid TOTP token"))?;
+        if claims.purpose.as_deref() != Some("totp") {
+            return Err(err("Invalid token purpose"));
+        }
+
+        let user: Option<(i64, String, Option<String>)> = sqlx::query_as(
+            "SELECT id, role, totp_secret FROM users WHERE id = ?"
+        ).bind(claims.sub).fetch_optional(&state.pool).await.map_err(|_| err("Database error"))?;
+
+        let (user_id, role, totp_secret) = user.ok_or_else(|| err("User not found"))?;
+        let secret = totp_secret.ok_or_else(|| err("TOTP not configured"))?;
+
+        let secret_bytes = totp_rs::Secret::Encoded(secret)
+            .to_bytes().map_err(|_| err("TOTP secret decode failed"))?;
+        let totp = totp_rs::TOTP::new(
+            totp_rs::Algorithm::SHA1, 6, 1, 30, secret_bytes,
+        ).map_err(|_| err("TOTP init failed"))?;
+
+        if !totp.check_current(totp_code).map_err(|_| err("TOTP check failed"))? {
+            return Err(err("Invalid TOTP code"));
+        }
+
+        let token = create_jwt(&state.jwt_secret, user_id, &role, None)?;
+        return Ok(Json(LoginResponse { token: Some(token), totp_required: None, totp_token: None }));
+    }
+
+    // Password step 1
+    let username = req.username.as_deref().ok_or_else(|| err("Username required"))?;
+    let password = req.password.as_deref().ok_or_else(|| err("Password required"))?;
+
+    let user: Option<(i64, String, String, i32)> = sqlx::query_as(
+        "SELECT id, password_hash, role, totp_enabled FROM users WHERE username = ?"
+    ).bind(username).fetch_optional(&state.pool).await.map_err(|_| err("Database error"))?;
+
+    let (user_id, hash, role, totp_enabled) = user.ok_or_else(|| err("Invalid credentials"))?;
+
+    let parsed_hash = PasswordHash::new(&hash).map_err(|_| err("Invalid credentials"))?;
+    Argon2::default().verify_password(password.as_bytes(), &parsed_hash).map_err(|_| err("Invalid credentials"))?;
+
+    if totp_enabled != 0 {
+        let totp_token = create_jwt(&state.jwt_secret, user_id, &role, Some("totp"))?;
+        return Ok(Json(LoginResponse { token: None, totp_required: Some(true), totp_token: Some(totp_token) }));
+    }
+
+    let token = create_jwt(&state.jwt_secret, user_id, &role, None)?;
+    Ok(Json(LoginResponse { token: Some(token), totp_required: None, totp_token: None }))
+}
+
+fn create_jwt(secret: &str, user_id: i64, role: &str, purpose: Option<&str>) -> Result<String, (StatusCode, Json<serde_json::Value>)> {
+    let exp = if purpose.is_some() { Utc::now() + Duration::seconds(60) } else { Utc::now() + Duration::hours(24) };
+    let claims = Claims { sub: user_id, role: role.to_string(), exp: exp.timestamp(), purpose: purpose.map(String::from) };
+    encode(&Header::default(), &claims, &EncodingKey::from_secret(secret.as_bytes()))
+        .map_err(|_| (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "Failed to create token" }))))
+}
+
+fn decode_jwt(secret: &str, token: &str) -> Result<Claims, jsonwebtoken::errors::Error> {
+    let data = decode::<Claims>(token, &DecodingKey::from_secret(secret.as_bytes()), &Validation::default())?;
+    Ok(data.claims)
+}
+
+pub async fn require_auth(
+    State(state): State<AppState>,
+    mut req: Request<axum::body::Body>,
+    next: Next,
+) -> Response {
+    let auth_header = req.headers().get("authorization")
+        .and_then(|v| v.to_str().ok())
+        .and_then(|v| v.strip_prefix("Bearer "));
+
+    let token = match auth_header {
+        Some(t) => t,
+        None => return (StatusCode::UNAUTHORIZED, Json(serde_json::json!({ "error": "Missing token" }))).into_response(),
+    };
+
+    let claims = match decode_jwt(&state.jwt_secret, token) {
+        Ok(c) => c,
+        Err(_) => return (StatusCode::UNAUTHORIZED, Json(serde_json::json!({ "error": "Invalid token" }))).into_response(),
+    };
+
+    if claims.purpose.is_some() {
+        return (StatusCode::UNAUTHORIZED, Json(serde_json::json!({ "error": "Invalid token type" }))).into_response();
+    }
+
+    req.extensions_mut().insert(claims);
+    next.run(req).await
+}
+
+pub async fn require_admin(req: Request<axum::body::Body>, next: Next) -> Response {
+    let claims = req.extensions().get::<Claims>();
+    match claims {
+        Some(c) if c.role == "admin" => next.run(req).await,
+        _ => (StatusCode::FORBIDDEN, Json(serde_json::json!({ "error": "Admin required" }))).into_response(),
+    }
+}

LORAWEB_NEU/backend/api/src/db.rs

@@ -0,0 +1,40 @@
+use sqlx::SqlitePool;
+use argon2::{Argon2, PasswordHasher, password_hash::SaltString};
+use rand::rngs::OsRng;
+use chrono::Utc;
+
+pub async fn init_db(pool: &SqlitePool) -> Result<(), Box<dyn std::error::Error>> {
+    sqlx::query("PRAGMA journal_mode=WAL").execute(pool).await?;
+    sqlx::query("PRAGMA busy_timeout=5000").execute(pool).await?;
+
+    let migration = include_str!("../../../database/migrations/001_initial.sql");
+    for statement in migration.split(';') {
+        let stmt = statement.trim();
+        if !stmt.is_empty() {
+            sqlx::query(stmt).execute(pool).await?;
+        }
+    }
+
+    let count: (i64,) = sqlx::query_as("SELECT COUNT(*) FROM users")
+        .fetch_one(pool).await?;
+
+    if count.0 == 0 {
+        let password = std::env::var("ADMIN_PASSWORD").unwrap_or_else(|_| "admin123".into());
+        let salt = SaltString::generate(&mut OsRng);
+        let hash = Argon2::default()
+            .hash_password(password.as_bytes(), &salt)?
+            .to_string();
+        let now = Utc::now().to_rfc3339();
+
+        sqlx::query(
+            "INSERT INTO users (username, password_hash, role, totp_enabled, created_at) VALUES ('admin', ?, 'admin', 0, ?)"
+        )
+        .bind(&hash)
+        .bind(&now)
+        .execute(pool)
+        .await?;
+        tracing::info!("Default admin user created");
+    }
+
+    Ok(())
+}

LORAWEB_NEU/backend/api/src/main.rs

@@ -1,7 +1,82 @@
-#[tokio::main]
-async fn main() {
-    tracing_subscriber::fmt()
-        .with_env_filter(tracing_subscriber::EnvFilter::from_default_env())
-        .init();
-    tracing::info!("loraweb-api starting...");
+mod db;
+mod auth;
+mod sensors;
+mod topics;
+mod alarms;
+mod users;
+
+use axum::Router;
+use sqlx::sqlite::SqlitePoolOptions;
+use tower_http::cors::{CorsLayer, Any};
+
+#[derive(Clone)]
+pub struct AppState {
+    pub pool: sqlx::SqlitePool,
+    pub jwt_secret: String,
+    pub influx_url: String,
+    pub influx_token: String,
+    pub influx_org: String,
+    pub influx_bucket: String,
+}
+
+#[tokio::main]
+async fn main() -> anyhow::Result<()> {
+    tracing_subscriber::fmt()
+        .with_env_filter(
+            tracing_subscriber::EnvFilter::from_default_env()
+                .add_directive("loraweb_api=info".parse()?)
+        )
+        .init();
+
+    let db_path = std::env::var("DATABASE_PATH").unwrap_or_else(|_| "/data/loraweb.db".into());
+    let port = std::env::var("PORT").unwrap_or_else(|_| "3001".into());
+
+    let pool = SqlitePoolOptions::new()
+        .max_connections(10)
+        .connect(&format!("sqlite:{}?mode=rwc", db_path))
+        .await?;
+    db::init_db(&pool).await?;
+
+    let state = AppState {
+        pool,
+        jwt_secret: std::env::var("JWT_SECRET").unwrap_or_else(|_| "insecure-dev-secret".into()),
+        influx_url: std::env::var("INFLUX_URL").unwrap_or_else(|_| "http://influxdb:8086".into()),
+        influx_token: std::env::var("INFLUX_TOKEN").unwrap_or_default(),
+        influx_org: std::env::var("INFLUX_ORG").unwrap_or_else(|_| "loraweb".into()),
+        influx_bucket: std::env::var("INFLUX_BUCKET").unwrap_or_else(|_| "sensors".into()),
+    };
+
+    let cors = CorsLayer::new().allow_origin(Any).allow_methods(Any).allow_headers(Any);
+
+    let app = Router::new()
+        .nest("/api", api_routes(state))
+        .layer(cors);
+
+    let addr = format!("0.0.0.0:{}", port);
+    tracing::info!("LoRaWEB API listening on {}", addr);
+
+    let listener = tokio::net::TcpListener::bind(&addr).await?;
+    axum::serve(listener, app).await?;
+    Ok(())
+}
+
+fn api_routes(state: AppState) -> Router {
+    let public = Router::new().merge(auth::routes());
+
+    let protected = Router::new()
+        .merge(sensors::routes())
+        .merge(topics::routes())
+        .merge(alarms::routes())
+        .route_layer(axum::middleware::from_fn_with_state(state.clone(), auth::require_auth));
+
+    let admin = Router::new()
+        .merge(users::routes())
+        .route_layer(axum::middleware::from_fn(auth::require_admin))
+        .route_layer(axum::middleware::from_fn_with_state(state.clone(), auth::require_auth));
+
+    Router::new()
+        .merge(public)
+        .merge(protected)
+        .merge(admin)
+        .with_state(state)
 }

LORAWEB_NEU/backend/api/src/sensors.rs

@@ -0,0 +1,377 @@
+use axum::{
+    extract::{Path, Query, State},
+    http::StatusCode,
+    routing::get,
+    Json, Router,
+};
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use chrono::Utc;
+use uuid::Uuid;
+use crate::AppState;
+
+#[derive(Debug, Serialize, sqlx::FromRow)]
+struct Sensor {
+    id: String,
+    name: String,
+    description: String,
+    location: String,
+    sensor_type: String,
+    active: i32,
+    topic_id: Option<String>,
+    display_config: String,
+    created_at: String,
+}
+
+#[derive(Deserialize)]
+struct CreateSensorRequest {
+    name: String,
+    description: Option<String>,
+    location: Option<String>,
+    sensor_type: Option<String>,
+    active: Option<i32>,
+    topic_id: Option<String>,
+    display_config: Option<Value>,
+}
+
+#[derive(Deserialize)]
+struct UpdateSensorRequest {
+    name: Option<String>,
+    description: Option<String>,
+    location: Option<String>,
+    sensor_type: Option<String>,
+    active: Option<i32>,
+    topic_id: Option<String>,
+    display_config: Option<Value>,
+}
+
+#[derive(Deserialize)]
+struct DataQuery {
+    start: Option<String>,
+    stop: Option<String>,
+    window: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct DisplayConfigRequest {
+    config: Value,
+}
+
+type ApiError = (StatusCode, Json<serde_json::Value>);
+
+fn internal_err(msg: &str) -> ApiError {
+    (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": msg })))
+}
+
+fn not_found(msg: &str) -> ApiError {
+    (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": msg })))
+}
+
+fn bad_request(msg: &str) -> ApiError {
+    (StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": msg })))
+}
+
+pub fn routes() -> Router<AppState> {
+    Router::new()
+        .route("/sensors", get(list_sensors).post(create_sensor))
+        .route("/sensors/{id}", get(get_sensor).put(update_sensor).delete(delete_sensor))
+        .route("/sensors/{id}/data", get(get_sensor_data))
+        .route("/sensors/{id}/display-config", get(get_display_config).put(save_display_config))
+}
+
+async fn list_sensors(
+    State(state): State<AppState>,
+) -> Result<Json<Vec<Sensor>>, ApiError> {
+    let sensors = sqlx::query_as::<_, Sensor>("SELECT * FROM sensors ORDER BY name")
+        .fetch_all(&state.pool)
+        .await
+        .map_err(|_| internal_err("Database error"))?;
+    Ok(Json(sensors))
+}
+
+async fn create_sensor(
+    State(state): State<AppState>,
+    Json(req): Json<CreateSensorRequest>,
+) -> Result<(StatusCode, Json<Sensor>), ApiError> {
+    let id = Uuid::new_v4().to_string();
+    let now = Utc::now().to_rfc3339();
+    let description = req.description.unwrap_or_default();
+    let location = req.location.unwrap_or_default();
+    let sensor_type = req.sensor_type.unwrap_or_default();
+    let active = req.active.unwrap_or(1);
+    let display_config = req.display_config
+        .map(|v| serde_json::to_string(&v).unwrap_or_else(|_| "{}".into()))
+        .unwrap_or_else(|| "{}".into());
+
+    sqlx::query(
+        "INSERT INTO sensors (id, name, description, location, sensor_type, active, topic_id, display_config, created_at)
+         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
+    )
+    .bind(&id)
+    .bind(&req.name)
+    .bind(&description)
+    .bind(&location)
+    .bind(&sensor_type)
+    .bind(active)
+    .bind(&req.topic_id)
+    .bind(&display_config)
+    .bind(&now)
+    .execute(&state.pool)
+    .await
+    .map_err(|_| internal_err("Failed to create sensor"))?;
+
+    let sensor = sqlx::query_as::<_, Sensor>("SELECT * FROM sensors WHERE id = ?")
+        .bind(&id)
+        .fetch_one(&state.pool)
+        .await
+        .map_err(|_| internal_err("Failed to fetch created sensor"))?;
+
+    Ok((StatusCode::CREATED, Json(sensor)))
+}
+
+async fn get_sensor(
+    State(state): State<AppState>,
+    Path(id): Path<String>,
+) -> Result<Json<Sensor>, ApiError> {
+    let sensor = sqlx::query_as::<_, Sensor>("SELECT * FROM sensors WHERE id = ?")
+        .bind(&id)
+        .fetch_optional(&state.pool)
+        .await
+        .map_err(|_| internal_err("Database error"))?
+        .ok_or_else(|| not_found("Sensor not found"))?;
+    Ok(Json(sensor))
+}
+
+async fn update_sensor(
+    State(state): State<AppState>,
+    Path(id): Path<String>,
+    Json(req): Json<UpdateSensorRequest>,
+) -> Result<Json<Sensor>, ApiError> {
+    let existing = sqlx::query_as::<_, Sensor>("SELECT * FROM sensors WHERE id = ?")
+        .bind(&id)
+        .fetch_optional(&state.pool)
+        .await
+        .map_err(|_| internal_err("Database error"))?
+        .ok_or_else(|| not_found("Sensor not found"))?;
+
+    let name = req.name.unwrap_or(existing.name);
+    let description = req.description.unwrap_or(existing.description);
+    let location = req.location.unwrap_or(existing.location);
+    let sensor_type = req.sensor_type.unwrap_or(existing.sensor_type);
+    let active = req.active.unwrap_or(existing.active);
+    let topic_id = if req.topic_id.is_some() { req.topic_id } else { existing.topic_id };
+    let display_config = req.display_config
+        .map(|v| serde_json::to_string(&v).unwrap_or_else(|_| "{}".into()))
+        .unwrap_or(existing.display_config);
+
+    sqlx::query(
+        "UPDATE sensors SET name=?, description=?, location=?, sensor_type=?, active=?, topic_id=?, display_config=? WHERE id=?"
+    )
+    .bind(&name)
+    .bind(&description)
+    .bind(&location)
+    .bind(&sensor_type)
+    .bind(active)
+    .bind(&topic_id)
+    .bind(&display_config)
+    .bind(&id)
+    .execute(&state.pool)
+    .await
+    .map_err(|_| internal_err("Failed to update sensor"))?;
+
+    let sensor = sqlx::query_as::<_, Sensor>("SELECT * FROM sensors WHERE id = ?")
+        .bind(&id)
+        .fetch_one(&state.pool)
+        .await
+        .map_err(|_| internal_err("Failed to fetch updated sensor"))?;
+
+    Ok(Json(sensor))
+}
+
+async fn delete_sensor(
+    State(state): State<AppState>,
+    Path(id): Path<String>,
+) -> Result<StatusCode, ApiError> {
+    let sensor = sqlx::query_as::<_, Sensor>("SELECT * FROM sensors WHERE id = ?")
+        .bind(&id)
+        .fetch_optional(&state.pool)
+        .await
+        .map_err(|_| internal_err("Database error"))?
+        .ok_or_else(|| not_found("Sensor not found"))?;
+
+    let mut tx = state.pool.begin().await.map_err(|_| internal_err("Transaction error"))?;
+
+    if let Some(topic_id) = &sensor.topic_id {
+        sqlx::query("UPDATE topics SET approved=0 WHERE id=?")
+            .bind(topic_id)
+            .execute(&mut *tx)
+            .await
+            .map_err(|_| internal_err("Failed to reset topic"))?;
+    }
+
+    sqlx::query("DELETE FROM sensors WHERE id=?")
+        .bind(&id)
+        .execute(&mut *tx)
+        .await
+        .map_err(|_| internal_err("Failed to delete sensor"))?;
+
+    tx.commit().await.map_err(|_| internal_err("Transaction commit failed"))?;
+
+    Ok(StatusCode::NO_CONTENT)
+}
+
+async fn get_sensor_data(
+    State(state): State<AppState>,
+    Path(id): Path<String>,
+    Query(params): Query<DataQuery>,
+) -> Result<Json<Vec<serde_json::Value>>, ApiError> {
+    let row: Option<(Option<String>, Option<String>)> = sqlx::query_as(
+        "SELECT s.topic_id, t.topic FROM sensors s LEFT JOIN topics t ON s.topic_id = t.id WHERE s.id = ?"
+    )
+    .bind(&id)
+    .fetch_optional(&state.pool)
+    .await
+    .map_err(|_| internal_err("Database error"))?;
+
+    let (topic_id, topic_name) = row.ok_or_else(|| not_found("Sensor not found"))?;
+
+    let _topic_id = topic_id.ok_or_else(|| bad_request("Sensor has no linked topic"))?;
+    let dev_eui = topic_name.ok_or_else(|| bad_request("Topic has no name"))?;
+
+    let start = params.start.as_deref().unwrap_or("-1h");
+    let stop = params.stop.as_deref().unwrap_or("now()");
+    let window = params.window.as_deref().unwrap_or("5m");
+
+    let flux_query = format!(
+        r#"from(bucket: "{bucket}")
+  |> range(start: {start}, stop: {stop})
+  |> filter(fn: (r) => r["dev_eui"] == "{dev_eui}" or r["topic"] == "{dev_eui}")
+  |> aggregateWindow(every: {window}, fn: mean, createEmpty: false)
+  |> yield(name: "mean")"#,
+        bucket = state.influx_bucket,
+        start = start,
+        stop = stop,
+        dev_eui = dev_eui,
+        window = window,
+    );
+
+    let client = reqwest::Client::new();
+    let response = client
+        .post(format!("{}/api/v2/query", state.influx_url))
+        .header("Authorization", format!("Token {}", state.influx_token))
+        .header("Content-Type", "application/vnd.flux")
+        .query(&[("org", &state.influx_org)])
+        .body(flux_query)
+        .send()
+        .await
+        .map_err(|_| internal_err("InfluxDB request failed"))?;
+
+    if !response.status().is_success() {
+        let status = response.status().as_u16();
+        let body = response.text().await.unwrap_or_default();
+        tracing::warn!("InfluxDB error {}: {}", status, body);
+        return Err((
+            StatusCode::BAD_GATEWAY,
+            Json(serde_json::json!({ "error": "InfluxDB query failed", "detail": body })),
+        ));
+    }
+
+    let csv_text = response.text().await.map_err(|_| internal_err("Failed to read InfluxDB response"))?;
+    let data = parse_influx_csv(&csv_text);
+
+    Ok(Json(data))
+}
+
+fn parse_influx_csv(csv: &str) -> Vec<serde_json::Value> {
+    let mut results = Vec::new();
+    let mut headers: Vec<String> = Vec::new();
+
+    for line in csv.lines() {
+        if line.starts_with('#') || line.trim().is_empty() {
+            continue;
+        }
+
+        let cols: Vec<&str> = line.split(',').collect();
+
+        if headers.is_empty() {
+            headers = cols.iter().map(|s| s.trim().to_string()).collect();
+            continue;
+        }
+
+        let mut obj = serde_json::Map::new();
+        for (i, header) in headers.iter().enumerate() {
+            if header.is_empty() {
+                continue;
+            }
+            if header == "result" || header == "table" {
+                continue;
+            }
+            // Keep _time, _value, _field, _measurement; skip other _-prefixed
+            if header.starts_with('_')
+                && header != "_time"
+                && header != "_value"
+                && header != "_field"
+                && header != "_measurement"
+            {
+                continue;
+            }
+
+            let value = cols.get(i).map(|s| s.trim()).unwrap_or("");
+
+            if let Ok(n) = value.parse::<f64>() {
+                obj.insert(header.clone(), serde_json::json!(n));
+            } else if value == "true" {
+                obj.insert(header.clone(), serde_json::json!(true));
+            } else if value == "false" {
+                obj.insert(header.clone(), serde_json::json!(false));
+            } else {
+                obj.insert(header.clone(), serde_json::json!(value));
+            }
+        }
+
+        if !obj.is_empty() {
+            results.push(serde_json::Value::Object(obj));
+        }
+    }
+
+    results
+}
+
+async fn get_display_config(
+    State(state): State<AppState>,
+    Path(id): Path<String>,
+) -> Result<Json<Value>, ApiError> {
+    let row: Option<(String,)> = sqlx::query_as("SELECT display_config FROM sensors WHERE id = ?")
+        .bind(&id)
+        .fetch_optional(&state.pool)
+        .await
+        .map_err(|_| internal_err("Database error"))?;
+
+    let (config_str,) = row.ok_or_else(|| not_found("Sensor not found"))?;
+    let config: Value = serde_json::from_str(&config_str).unwrap_or(serde_json::json!({}));
+
+    Ok(Json(config))
+}
+
+async fn save_display_config(
+    State(state): State<AppState>,
+    Path(id): Path<String>,
+    Json(req): Json<DisplayConfigRequest>,
+) -> Result<Json<Value>, ApiError> {
+    let config_str = serde_json::to_string(&req.config)
+        .map_err(|_| bad_request("Invalid config JSON"))?;
+
+    let rows_affected = sqlx::query("UPDATE sensors SET display_config=? WHERE id=?")
+        .bind(&config_str)
+        .bind(&id)
+        .execute(&state.pool)
+        .await
+        .map_err(|_| internal_err("Failed to save display config"))?
+        .rows_affected();
+
+    if rows_affected == 0 {
+        return Err(not_found("Sensor not found"));
+    }
+
+    Ok(Json(req.config))
+}

LORAWEB_NEU/backend/api/src/topics.rs

@@ -0,0 +1,127 @@
+use axum::{
+    extract::{Path, State},
+    http::StatusCode,
+    routing::{get, put},
+    Json, Router,
+};
+use serde::{Deserialize, Serialize};
+use crate::AppState;
+
+#[derive(Debug, Serialize, sqlx::FromRow)]
+struct Topic {
+    id: String,
+    topic: String,
+    approved: i32,
+    last_seen: Option<String>,
+    alarm_status: String,
+    alarm_threshold_hours: f64,
+    created_at: String,
+}
+
+#[derive(Deserialize)]
+struct UpdateTopicRequest {
+    approved: Option<bool>,
+    sensor_id: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct ThresholdRequest {
+    threshold_minutes: f64,
+}
+
+type ApiError = (StatusCode, Json<serde_json::Value>);
+
+fn internal_err(msg: &str) -> ApiError {
+    (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": msg })))
+}
+
+fn not_found(msg: &str) -> ApiError {
+    (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": msg })))
+}
+
+pub fn routes() -> Router<AppState> {
+    Router::new()
+        .route("/topics", get(list_topics))
+        .route("/topics/{id}", put(update_topic))
+        .route("/topics/{id}/threshold", put(set_threshold))
+}
+
+async fn list_topics(
+    State(state): State<AppState>,
+) -> Result<Json<Vec<Topic>>, ApiError> {
+    let topics = sqlx::query_as::<_, Topic>("SELECT * FROM topics ORDER BY created_at DESC")
+        .fetch_all(&state.pool)
+        .await
+        .map_err(|_| internal_err("Database error"))?;
+    Ok(Json(topics))
+}
+
+async fn update_topic(
+    State(state): State<AppState>,
+    Path(id): Path<String>,
+    Json(req): Json<UpdateTopicRequest>,
+) -> Result<Json<Topic>, ApiError> {
+    // Verify topic exists
+    let existing = sqlx::query_as::<_, Topic>("SELECT * FROM topics WHERE id = ?")
+        .bind(&id)
+        .fetch_optional(&state.pool)
+        .await
+        .map_err(|_| internal_err("Database error"))?
+        .ok_or_else(|| not_found("Topic not found"))?;
+
+    let approved = if let Some(sensor_id) = &req.sensor_id {
+        // Link sensor to this topic and auto-approve
+        sqlx::query("UPDATE sensors SET topic_id=? WHERE id=?")
+            .bind(&id)
+            .bind(sensor_id)
+            .execute(&state.pool)
+            .await
+            .map_err(|_| internal_err("Failed to link sensor"))?;
+        1 // auto-approve when sensor is linked
+    } else {
+        req.approved.map(|a| if a { 1 } else { 0 }).unwrap_or(existing.approved)
+    };
+
+    sqlx::query("UPDATE topics SET approved=? WHERE id=?")
+        .bind(approved)
+        .bind(&id)
+        .execute(&state.pool)
+        .await
+        .map_err(|_| internal_err("Failed to update topic"))?;
+
+    let topic = sqlx::query_as::<_, Topic>("SELECT * FROM topics WHERE id = ?")
+        .bind(&id)
+        .fetch_one(&state.pool)
+        .await
+        .map_err(|_| internal_err("Failed to fetch updated topic"))?;
+
+    Ok(Json(topic))
+}
+
+async fn set_threshold(
+    State(state): State<AppState>,
+    Path(id): Path<String>,
+    Json(req): Json<ThresholdRequest>,
+) -> Result<Json<Topic>, ApiError> {
+    let threshold_hours = req.threshold_minutes / 60.0;
+
+    let rows_affected = sqlx::query("UPDATE topics SET alarm_threshold_hours=? WHERE id=?")
+        .bind(threshold_hours)
+        .bind(&id)
+        .execute(&state.pool)
+        .await
+        .map_err(|_| internal_err("Failed to update threshold"))?
+        .rows_affected();
+
+    if rows_affected == 0 {
+        return Err(not_found("Topic not found"));
+    }
+
+    let topic = sqlx::query_as::<_, Topic>("SELECT * FROM topics WHERE id = ?")
+        .bind(&id)
+        .fetch_one(&state.pool)
+        .await
+        .map_err(|_| internal_err("Failed to fetch updated topic"))?;
+
+    Ok(Json(topic))
+}

LORAWEB_NEU/backend/api/src/users.rs

@@ -0,0 +1,265 @@
+use axum::{
+    extract::{Path, State},
+    http::StatusCode,
+    routing::{delete, get, post, put},
+    Json, Router,
+};
+use argon2::{Argon2, PasswordHasher, password_hash::SaltString};
+use rand::rngs::OsRng;
+use serde::{Deserialize, Serialize};
+use chrono::Utc;
+use crate::AppState;
+
+#[derive(Debug, Serialize, sqlx::FromRow)]
+struct User {
+    id: i64,
+    username: String,
+    role: String,
+    totp_enabled: i32,
+    created_at: String,
+}
+
+#[derive(Deserialize)]
+struct CreateUserRequest {
+    username: String,
+    password: String,
+    role: Option<String>,
+}
+
+#[derive(Deserialize)]
+struct UpdateUserRequest {
+    username: Option<String>,
+    password: Option<String>,
+    role: Option<String>,
+}
+
+#[derive(Serialize)]
+struct TotpSetupResponse {
+    secret: String,
+    uri: String,
+}
+
+type ApiError = (StatusCode, Json<serde_json::Value>);
+
+fn internal_err(msg: &str) -> ApiError {
+    (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": msg })))
+}
+
+fn not_found(msg: &str) -> ApiError {
+    (StatusCode::NOT_FOUND, Json(serde_json::json!({ "error": msg })))
+}
+
+fn bad_request(msg: &str) -> ApiError {
+    (StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": msg })))
+}
+
+fn conflict(msg: &str) -> ApiError {
+    (StatusCode::CONFLICT, Json(serde_json::json!({ "error": msg })))
+}
+
+pub fn routes() -> Router<AppState> {
+    Router::new()
+        .route("/users", get(list_users).post(create_user))
+        .route("/users/{id}", put(update_user).delete(delete_user))
+        .route("/users/{id}/totp", post(enable_totp).delete(disable_totp))
+}
+
+async fn list_users(
+    State(state): State<AppState>,
+) -> Result<Json<Vec<User>>, ApiError> {
+    let users = sqlx::query_as::<_, User>(
+        "SELECT id, username, role, totp_enabled, created_at FROM users ORDER BY created_at ASC"
+    )
+    .fetch_all(&state.pool)
+    .await
+    .map_err(|_| internal_err("Database error"))?;
+    Ok(Json(users))
+}
+
+async fn create_user(
+    State(state): State<AppState>,
+    Json(req): Json<CreateUserRequest>,
+) -> Result<(StatusCode, Json<User>), ApiError> {
+    if req.username.trim().is_empty() {
+        return Err(bad_request("Username is required"));
+    }
+    if req.password.is_empty() {
+        return Err(bad_request("Password is required"));
+    }
+
+    let role = req.role.unwrap_or_else(|| "user".into());
+    let salt = SaltString::generate(&mut OsRng);
+    let hash = Argon2::default()
+        .hash_password(req.password.as_bytes(), &salt)
+        .map_err(|_| internal_err("Failed to hash password"))?
+        .to_string();
+    let now = Utc::now().to_rfc3339();
+
+    let result = sqlx::query(
+        "INSERT INTO users (username, password_hash, role, totp_enabled, created_at) VALUES (?, ?, ?, 0, ?)"
+    )
+    .bind(&req.username)
+    .bind(&hash)
+    .bind(&role)
+    .bind(&now)
+    .execute(&state.pool)
+    .await;
+
+    let insert_result = match result {
+        Ok(r) => r,
+        Err(e) => {
+            let msg = e.to_string();
+            if msg.contains("UNIQUE") || msg.contains("unique") {
+                return Err(conflict("Username already exists"));
+            }
+            return Err(internal_err("Failed to create user"));
+        }
+    };
+
+    let user_id = insert_result.last_insert_rowid();
+    let user = sqlx::query_as::<_, User>(
+        "SELECT id, username, role, totp_enabled, created_at FROM users WHERE id = ?"
+    )
+    .bind(user_id)
+    .fetch_one(&state.pool)
+    .await
+    .map_err(|_| internal_err("Failed to fetch created user"))?;
+
+    Ok((StatusCode::CREATED, Json(user)))
+}
+
+async fn update_user(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+    Json(req): Json<UpdateUserRequest>,
+) -> Result<Json<User>, ApiError> {
+    let existing = sqlx::query_as::<_, User>(
+        "SELECT id, username, role, totp_enabled, created_at FROM users WHERE id = ?"
+    )
+    .bind(id)
+    .fetch_optional(&state.pool)
+    .await
+    .map_err(|_| internal_err("Database error"))?
+    .ok_or_else(|| not_found("User not found"))?;
+
+    let username = req.username.unwrap_or(existing.username);
+    let role = req.role.unwrap_or(existing.role);
+
+    if let Some(password) = req.password {
+        if password.is_empty() {
+            return Err(bad_request("Password cannot be empty"));
+        }
+        let salt = SaltString::generate(&mut OsRng);
+        let hash = Argon2::default()
+            .hash_password(password.as_bytes(), &salt)
+            .map_err(|_| internal_err("Failed to hash password"))?
+            .to_string();
+
+        sqlx::query("UPDATE users SET username=?, role=?, password_hash=? WHERE id=?")
+            .bind(&username)
+            .bind(&role)
+            .bind(&hash)
+            .bind(id)
+            .execute(&state.pool)
+            .await
+            .map_err(|_| internal_err("Failed to update user"))?;
+    } else {
+        sqlx::query("UPDATE users SET username=?, role=? WHERE id=?")
+            .bind(&username)
+            .bind(&role)
+            .bind(id)
+            .execute(&state.pool)
+            .await
+            .map_err(|_| internal_err("Failed to update user"))?;
+    }
+
+    let user = sqlx::query_as::<_, User>(
+        "SELECT id, username, role, totp_enabled, created_at FROM users WHERE id = ?"
+    )
+    .bind(id)
+    .fetch_one(&state.pool)
+    .await
+    .map_err(|_| internal_err("Failed to fetch updated user"))?;
+
+    Ok(Json(user))
+}
+
+async fn delete_user(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+) -> Result<StatusCode, ApiError> {
+    let rows_affected = sqlx::query("DELETE FROM users WHERE id=?")
+        .bind(id)
+        .execute(&state.pool)
+        .await
+        .map_err(|_| internal_err("Database error"))?
+        .rows_affected();
+
+    if rows_affected == 0 {
+        return Err(not_found("User not found"));
+    }
+
+    Ok(StatusCode::NO_CONTENT)
+}
+
+async fn enable_totp(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+) -> Result<Json<TotpSetupResponse>, ApiError> {
+    // Verify user exists and get username for the TOTP URI
+    let row: Option<(String,)> = sqlx::query_as("SELECT username FROM users WHERE id = ?")
+        .bind(id)
+        .fetch_optional(&state.pool)
+        .await
+        .map_err(|_| internal_err("Database error"))?;
+
+    let (username,) = row.ok_or_else(|| not_found("User not found"))?;
+
+    let secret = totp_rs::Secret::generate_secret();
+    let secret_encoded = secret.to_encoded().to_string();
+
+    let secret_bytes = totp_rs::Secret::Encoded(secret_encoded.clone())
+        .to_bytes()
+        .map_err(|_| internal_err("Failed to decode TOTP secret"))?;
+
+    let totp = totp_rs::TOTP::new(
+        totp_rs::Algorithm::SHA1,
+        6,
+        1,
+        30,
+        secret_bytes,
+    )
+    .map_err(|_| internal_err("Failed to create TOTP"))?;
+
+    let uri = totp.get_url(&username, "LoRaWEB");
+
+    sqlx::query("UPDATE users SET totp_secret=?, totp_enabled=1 WHERE id=?")
+        .bind(&secret_encoded)
+        .bind(id)
+        .execute(&state.pool)
+        .await
+        .map_err(|_| internal_err("Failed to save TOTP secret"))?;
+
+    Ok(Json(TotpSetupResponse {
+        secret: secret_encoded,
+        uri,
+    }))
+}
+
+async fn disable_totp(
+    State(state): State<AppState>,
+    Path(id): Path<i64>,
+) -> Result<StatusCode, ApiError> {
+    let rows_affected = sqlx::query("UPDATE users SET totp_secret=NULL, totp_enabled=0 WHERE id=?")
+        .bind(id)
+        .execute(&state.pool)
+        .await
+        .map_err(|_| internal_err("Database error"))?
+        .rows_affected();
+
+    if rows_affected == 0 {
+        return Err(not_found("User not found"));
+    }
+
+    Ok(StatusCode::NO_CONTENT)
+}