feat: initialize full microservices architecture including backend-api, edge-agent, infrastructure backup, and supporting modules

.env.example.docker

@@ -38,7 +38,7 @@ AWS_REGION=us-east-1
 # --- AGENTE EDGE (OTA updates) ---
 # Versión del agente que se distribuye. Actualizar cuando se publique nueva versión.
 # Ejemplo: "0.2.1" — el backend usará este valor en GET /api/edge/update-info
-AGENT_LATEST_VERSION=0.3.0
+AGENT_LATEST_VERSION=0.3.6
 # URL base del backend (usada internamente para construir download_url en update-info)
 API_BASE_URL=https://api.omnioil.com
 
@@ -98,9 +98,12 @@ REGISTRY_DATA=registry_data                               # Volumen para las im
 INSTALLER_INCOMING=./.docker/installer_incoming           # Carpeta host donde se inyectan los .exe
 INSTALLER_OUTPUT=./.docker/installer_output               # Carpeta host donde se generan los .msi
 
-# --- AGENTE EDGE (OTA updates) ---
+# --- RESPALDOS EN LA NUBE (Cloudflare R2 / S3) ---
-AGENT_LATEST_VERSION=0.2.0                               # Versión del agente distribuida vía OTA
+R2_ENDPOINT=https://<tu_account_id>.r2.cloudflarestorage.com # URL del endpoint de Cloudflare R2
-API_BASE_URL=http://localhost:8000                        # URL base del backend (para download_url en update-info)
+R2_BUCKET=omnioil-backups                                 # Nombre del bucket en R2
+R2_ACCESS_KEY_ID=                                         # ID de clave de acceso de Cloudflare R2
+R2_SECRET_ACCESS_KEY=                                     # Clave de acceso secreta de Cloudflare R2
+
 
 # Monitoring
 GRAFANA_USER=admin

.env.example.local

@@ -2,7 +2,7 @@
 # OMNIOIL SCADA - CONFIGURACIÓN PARA AMBIENTE LOCAL (DOCUMENTADA)
 # =============================================================================
 
-AGENT_LATEST_VERSION=0.3.2
+AGENT_LATEST_VERSION=0.3.6
 # --- ORQUESTACIÓN DE COMPOSE ---
 CONTAINER_PREFIX=anh
 COMPOSE_PATH_SEPARATOR=;                                 # Separador de archivos compose para Windows
@@ -95,8 +95,17 @@ REGISTRY_DATA=registry_data                               # Volumen para las im
 INSTALLER_INCOMING=./.docker/installer_incoming           # Carpeta host donde se inyectan los .exe
 INSTALLER_OUTPUT=./.docker/installer_output               # Carpeta host donde se generan los .msi
 
+# --- RESPALDOS EN LA NUBE (Cloudflare R2 / S3) ---
+R2_ENDPOINT=https://<tu_account_id>.r2.cloudflarestorage.com # URL del endpoint de Cloudflare R2
+R2_BUCKET=omnioil-backups                                 # Nombre del bucket en R2
+R2_ACCESS_KEY_ID=                                         # ID de clave de acceso de Cloudflare R2
+R2_SECRET_ACCESS_KEY=                                     # Clave de acceso secreta de Cloudflare R2
+BACKUP_SCHEDULE=0 3 * * *                                 # Programación cron para el backup (ej: diario a las 3 AM)
+BACKUP_RETENTION_DAYS=7                                   # Días de retención de los respaldos en R2
+
+
 # --- AGENTE EDGE (OTA updates) ---
-AGENT_LATEST_VERSION=0.3.0                               # Versión del agente que se distribuye vía OTA.
+AGENT_LATEST_VERSION=0.3.6                               # Versión del agente que se distribuye vía OTA.
                                                          # Actualizar cuando se publique nueva versión en MinIO.
                                                          # El endpoint GET /api/edge/update-info usa este valor.
 API_BASE_URL=http://localhost:8000                       # URL base del backend (para construir download_url en update-info)

Cargo.lock

@@ -1888,7 +1888,7 @@ dependencies = [
 
 [[package]]
 name = "edge-agent"
-version = "0.3.5"
+version = "0.3.6"
 dependencies = [
  "aes-gcm",
  "anyhow",

docker-compose.infra.yml

@@ -68,6 +68,32 @@ services:
     networks:
       - anh_network
 
+  backup:
+    restart: always
+    build:
+      context: ./infrastructure/backup
+      dockerfile: Dockerfile
+    depends_on:
+      timescaledb:
+        condition: service_healthy
+      minio:
+        condition: service_started
+    environment:
+      - DB_HOST=timescaledb
+      - DB_NAME=${DB_NAME}
+      - DB_USER=${DB_USER}
+      - DB_PASSWORD=${DB_PASSWORD}
+      - R2_ENDPOINT=${R2_ENDPOINT}
+      - R2_BUCKET=${R2_BUCKET}
+      - R2_ACCESS_KEY_ID=${R2_ACCESS_KEY_ID}
+      - R2_SECRET_ACCESS_KEY=${R2_SECRET_ACCESS_KEY}
+      - BACKUP_SCHEDULE=${BACKUP_SCHEDULE:-0 3 * * *}
+      - BACKUP_RETENTION_DAYS=${BACKUP_RETENTION_DAYS:-7}
+    volumes:
+      - ${MINIO_DATA:-minio_data}:/minio_data:ro
+    networks:
+      - anh_network
+
 networks:
   anh_network:
     name: ${CONTAINER_PREFIX:-anh}_network

docs/DESPLIEGUE.md

@@ -105,6 +105,12 @@ SMTP_PORT=587
 SMTP_USER=soporte@omnioil.app
 SMTP_PASS=<gestionar-en-secrets-manager>
 SMTP_FROM=OmniOil <soporte@omnioil.app>
+
+# Respaldos de Base de Datos en Cloudflare R2 (S3 compatible):
+R2_ENDPOINT=https://<tu_account_id>.r2.cloudflarestorage.com
+R2_BUCKET=omnioil-backups
+R2_ACCESS_KEY_ID=<tu_r2_access_key_id>
+R2_SECRET_ACCESS_KEY=<tu_r2_secret_access_key>
 ```
 
 `INVITATION_BASE_URL` debe apuntar al Admin App público porque los clientes activan su cuenta en `https://app.omnioil.app/activate?token=...`, no en la Console interna. Las variables `SMTP_*` son requeridas para acuses de recibo, rechazos, aprobación/activación y reutilizan la configuración SMTP del backend; `SMTP_PASS` debe cargarse desde el gestor de secretos de Dokploy o equivalente, nunca desde el repositorio. Los únicos correos autorizados para aprobar, rechazar, marcar en revisión o reenviar invitaciones son `w.farfan@omnioil.app` y `h.ortegon@omnioil.app`; esta allowlist se aplica en backend.
@@ -205,6 +211,17 @@ volumes:
 
 > ✅ En producción, dejar vacías las variables `TIMESCALEDB_DATA`, `REDIS_DATA`, etc. en el `.env` para activar los Named Volumes.
 
+### Respaldos Automatizados (Sidecar de Base de Datos)
+
+En producción, el stack incluye un contenedor sidecar (`db-backup`) basado en `ghcr.io/solectrus/postgres-s3-backup:16`. Este contenedor se encarga de:
+1. Conectarse de manera segura a la base de datos `timescaledb` una vez que esté lista (`service_healthy`).
+2. Realizar un dump lógico consistente con `pg_dump`.
+3. Comprimir y subir el archivo `.sql.gz` de forma automática a **Cloudflare R2** diariamente (`SCHEDULE=@daily`).
+4. Mantener únicamente los últimos 7 días de copias de seguridad en la nube (`BACKUP_KEEP_DAYS=7`) para optimizar el almacenamiento gratuito de 10 GB.
+
+Para activar este flujo, es mandatorio configurar en Dokploy las variables `R2_ENDPOINT`, `R2_BUCKET`, `R2_ACCESS_KEY_ID` y `R2_SECRET_ACCESS_KEY`.
+
+
 ---
 
 ## 5. Monitoreo con Prometheus y Grafana

infrastructure/backup/Dockerfile

@@ -0,0 +1,14 @@
+FROM alpine:3.19
+
+# Instalar cliente de postgresql16 (compatible con el de timescaledb), rclone, ca-certificates, bash, tar y gzip
+RUN apk add --no-cache postgresql16-client rclone ca-certificates bash tar gzip
+
+# Copiar el script de backup y el entrypoint
+COPY backup.sh /usr/local/bin/backup.sh
+COPY entrypoint.sh /usr/local/bin/entrypoint.sh
+
+# Dar permisos de ejecución a los scripts
+RUN chmod +x /usr/local/bin/backup.sh /usr/local/bin/entrypoint.sh
+
+# Iniciar mediante el entrypoint dinámico
+ENTRYPOINT ["/usr/local/bin/entrypoint.sh"]

infrastructure/backup/backup.sh

@@ -0,0 +1,55 @@
+#!/bin/bash
+set -e
+
+# Validar variables obligatorias de R2
+if [ -z "$R2_ENDPOINT" ] || [ -z "$R2_BUCKET" ] || [ -z "$R2_ACCESS_KEY_ID" ] || [ -z "$R2_SECRET_ACCESS_KEY" ]; then
+  echo "Error: Faltan variables obligatorias de Cloudflare R2."
+  exit 1
+fi
+
+# Validar variables obligatorias de la base de datos
+if [ -z "$DB_HOST" ] || [ -z "$DB_NAME" ] || [ -z "$DB_USER" ] || [ -z "$DB_PASSWORD" ]; then
+  echo "Error: Faltan variables obligatorias de PostgreSQL."
+  exit 1
+fi
+
+# Configurar rclone dinámicamente mediante variables de entorno
+export RCLONE_CONFIG_R2_TYPE=s3
+export RCLONE_CONFIG_R2_PROVIDER=Cloudflare
+export RCLONE_CONFIG_R2_ACCESS_KEY_ID="$R2_ACCESS_KEY_ID"
+export RCLONE_CONFIG_R2_SECRET_ACCESS_KEY="$R2_SECRET_ACCESS_KEY"
+export RCLONE_CONFIG_R2_ENDPOINT="$R2_ENDPOINT"
+
+TIMESTAMP=$(date +"%Y-%m-%dT%H-%M-%S")
+DB_BACKUP_FILE="/tmp/db-backup-$TIMESTAMP.sql.gz"
+MINIO_BACKUP_FILE="/tmp/minio-backup-$TIMESTAMP.tar.gz"
+
+echo "=== Iniciando Backup a las $(date) ==="
+
+# 1. Copia de seguridad de la base de datos (PostgreSQL)
+echo "Ejecutando pg_dump para la base de datos '$DB_NAME' en '$DB_HOST'..."
+PGPASSWORD="$DB_PASSWORD" pg_dump -h "$DB_HOST" -U "$DB_USER" -d "$DB_NAME" | gzip > "$DB_BACKUP_FILE"
+echo "Subiendo base de datos a R2 (ruta: database/db-backup-$TIMESTAMP.sql.gz)..."
+rclone copyto "$DB_BACKUP_FILE" "r2:$R2_BUCKET/database/db-backup-$TIMESTAMP.sql.gz"
+rm -f "$DB_BACKUP_FILE"
+echo "[OK] Backup de base de datos completado."
+
+# 2. Copia de seguridad de MinIO (Archivos del volumen)
+if [ -d "/minio_data" ]; then
+  echo "Comprimiendo directorio de datos de MinIO (/minio_data)..."
+  tar -czf "$MINIO_BACKUP_FILE" -C /minio_data .
+  echo "Subiendo MinIO a R2 (ruta: minio/minio-backup-$TIMESTAMP.tar.gz)..."
+  rclone copyto "$MINIO_BACKUP_FILE" "r2:$R2_BUCKET/minio/minio-backup-$TIMESTAMP.tar.gz"
+  rm -f "$MINIO_BACKUP_FILE"
+  echo "[OK] Backup de MinIO completado."
+else
+  echo "[WARNING] El directorio /minio_data no existe o no está montado."
+fi
+
+# 3. Retención de días configurada (Limpiar copias antiguas)
+RETENTION_DAYS="${BACKUP_RETENTION_DAYS:-7}"
+echo "Aplicando política de retención de $RETENTION_DAYS días..."
+rclone delete --min-age "${RETENTION_DAYS}d" "r2:$R2_BUCKET/database/"
+rclone delete --min-age "${RETENTION_DAYS}d" "r2:$R2_BUCKET/minio/"
+
+echo "=== Backup finalizado exitosamente a las $(date) ==="

infrastructure/backup/entrypoint.sh

@@ -0,0 +1,8 @@
+#!/bin/bash
+# Configurar la programación del cron dinámicamente desde la variable de entorno
+SCHEDULE="${BACKUP_SCHEDULE:-0 3 * * *}"
+echo "Configurando cron con la programación: $SCHEDULE"
+echo "$SCHEDULE /usr/local/bin/backup.sh > /proc/1/fd/1 2>&1" > /etc/crontabs/root
+
+# Ejecutar el demonio crond de Alpine en primer plano
+exec crond -f -l 2

scratch/test_r2.py

@@ -0,0 +1,80 @@
+import os
+import sys
+
+try:
+    import boto3
+    from botocore.client import Config
+    from botocore.exceptions import ClientError
+except ImportError:
+    print("Error: La biblioteca 'boto3' no está instalada.")
+    print("Por favor, ejecute: pip install boto3")
+    sys.exit(1)
+
+# --- CONFIGURACIÓN ---
+# Puedes establecer estas variables de entorno en la VPS, o pegarlas directamente aquí para probar.
+R2_ENDPOINT = os.getenv("R2_ENDPOINT", "https://<TU_ACCOUNT_ID>.r2.cloudflarestorage.com")
+R2_ACCESS_KEY_ID = os.getenv("R2_ACCESS_KEY_ID", "TU_ACCESS_KEY_ID")
+R2_SECRET_ACCESS_KEY = os.getenv("R2_SECRET_ACCESS_KEY", "TU_SECRET_ACCESS_KEY")
+R2_BUCKET = os.getenv("R2_BUCKET", "omnioil-backups")
+
+print("=" * 60)
+print("PROBADOR DE CONECTIVIDAD CLOUDFLARE R2")
+print("=" * 60)
+print(f"Endpoint: {R2_ENDPOINT}")
+print(f"Bucket:   {R2_BUCKET}")
+print(f"Access Key ID: {R2_ACCESS_KEY_ID[:8]}... (truncado)")
+print("=" * 60)
+
+# Inicializar cliente de S3 compatible con R2
+s3 = boto3.client(
+    service_name='s3',
+    endpoint_url=R2_ENDPOINT,
+    aws_access_key_id=R2_ACCESS_KEY_ID,
+    aws_secret_access_key=R2_SECRET_ACCESS_KEY,
+    config=Config(signature_version='s3v4'),
+)
+
+# 1. Intentar listar objetos
+try:
+    print("\n1. Probando conexión y listado de objetos...")
+    response = s3.list_objects_v2(Bucket=R2_BUCKET)
+    print("   [OK] ¡Conexión exitosa al bucket!")
+    
+    if 'Contents' in response:
+        print("   Objetos en el bucket:")
+        for obj in response['Contents']:
+            print(f"    - {obj['Key']} ({obj['Size']} bytes)")
+    else:
+        print("   (El bucket está vacío actualmente)")
+except ClientError as e:
+    error_code = e.response['Error']['Code']
+    print(f"   [ERROR] No se pudo listar el bucket. Código de error: {error_code}")
+    print(f"   Detalle: {e}")
+    if error_code in ['AccessDenied', 'Forbidden', 'InvalidAccessKeyId']:
+        print("\n   >>> ATENCIÓN: Si configuraste restricción de IP, es muy probable que esta IP esté bloqueada o que las credenciales sean incorrectas.")
+    sys.exit(1)
+
+# 2. Intentar subir un archivo de prueba
+test_filename = "test_connection_file.txt"
+try:
+    print(f"\n2. Intentando subir archivo de prueba '{test_filename}'...")
+    s3.put_object(
+        Bucket=R2_BUCKET,
+        Key=test_filename,
+        Body=b"Prueba de conexion exitosa desde la VPS a Cloudflare R2",
+        ContentType="text/plain"
+    )
+    print("   [OK] ¡Archivo subido correctamente!")
+except Exception as e:
+    print(f"   [ERROR] Falló la subida del archivo de prueba: {e}")
+    sys.exit(1)
+
+# 3. Intentar eliminar el archivo de prueba
+try:
+    print(f"\n3. Intentando limpiar y eliminar '{test_filename}'...")
+    s3.delete_object(Bucket=R2_BUCKET, Key=test_filename)
+    print("   [OK] ¡Archivo de prueba eliminado con éxito!")
+    print("\n=== PRUEBA DE CONECTIVIDAD Y PERMISOS DE R2 COMPLETA CON ÉXITO ===")
+except Exception as e:
+    print(f"   [ERROR] No se pudo eliminar el archivo de prueba: {e}")
+    sys.exit(1)

services/backend-api/src/auth.rs

@@ -391,8 +391,7 @@ pub fn role_satisfies_platform_admin_claims(role: &str) -> bool {
 pub fn require_access_request_reviewer(claims: &Claims) -> Result<(), crate::errors::AppError> {
     let email = claims.email.trim().to_ascii_lowercase();
 
-    let reviewers = std::env::var("ACCESS_REQUEST_REVIEWER_EMAILS")
+    let reviewers = std::env::var("ACCESS_REQUEST_REVIEWER_EMAILS").unwrap_or_default();
-        .unwrap_or_default();
     let reviewer_list: Vec<String> = reviewers
         .split(',')
         .map(|s| s.trim().to_ascii_lowercase())

services/backend-api/src/db.rs

@@ -45,10 +45,7 @@ pub async fn establish_connection() -> Result<DbPool, sqlx::Error> {
     let max_retries = 10;
 
     loop {
-        match default_pool_options()
+        match default_pool_options().connect(&database_url).await {
-            .connect(&database_url)
-            .await
-        {
             Ok(pool) => return Ok(pool),
             Err(e) => {
                 retries += 1;

services/backend-api/src/handlers/update_info.rs

@@ -36,7 +36,12 @@ type HmacSha256 = Hmac<Sha256>;
 
 /// Verifica que la URL firmada sea válida y no haya expirado.
 /// Si AGENT_DOWNLOAD_SECRET no está configurado, opera en modo compatibilidad (sin firma).
-fn verify_signed_url(message: &str, timestamp: &str, signature: &str, max_age_secs: u64) -> Result<(), Response> {
+fn verify_signed_url(
+    message: &str,
+    timestamp: &str,
+    signature: &str,
+    max_age_secs: u64,
+) -> Result<(), Response> {
     let secret = match std::env::var("AGENT_DOWNLOAD_SECRET") {
         Ok(s) => s,
         Err(_) => {
@@ -51,27 +56,38 @@ fn verify_signed_url(message: &str, timestamp: &str, signature: &str, max_age_se
         .as_secs();
 
     let ts: u64 = timestamp.parse().map_err(|_| {
-        (StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": "Invalid timestamp" }))).into_response()
+        (
+            StatusCode::BAD_REQUEST,
+            Json(serde_json::json!({ "error": "Invalid timestamp" })),
+        )
+            .into_response()
     })?;
 
     if now.saturating_sub(ts) > max_age_secs {
-        return Err(
+        return Err((
-            (StatusCode::BAD_REQUEST, Json(serde_json::json!({ "error": "Signature expired" }))).into_response()
+            StatusCode::BAD_REQUEST,
-        );
+            Json(serde_json::json!({ "error": "Signature expired" })),
+        )
+            .into_response());
     }
 
-    let mut mac = HmacSha256::new_from_slice(secret.as_bytes())
+    let mut mac = HmacSha256::new_from_slice(secret.as_bytes()).map_err(|_| {
-        .map_err(|_| {
+        (
-            (StatusCode::INTERNAL_SERVER_ERROR, Json(serde_json::json!({ "error": "HMAC error" }))).into_response()
+            StatusCode::INTERNAL_SERVER_ERROR,
-        })?;
+            Json(serde_json::json!({ "error": "HMAC error" })),
+        )
+            .into_response()
+    })?;
     mac.update(message.as_bytes());
     let expected = hex::encode(mac.finalize().into_bytes());
 
     // Comparación en tiempo constante
     if expected.as_bytes() != signature.as_bytes() {
-        return Err(
+        return Err((
-            (StatusCode::FORBIDDEN, Json(serde_json::json!({ "error": "Invalid signature" }))).into_response()
+            StatusCode::FORBIDDEN,
-        );
+            Json(serde_json::json!({ "error": "Invalid signature" })),
+        )
+            .into_response());
     }
     Ok(())
 }
@@ -99,8 +115,7 @@ fn generate_signed_url(api_base: &str, version: &str, asset_type: &str) -> Strin
     let ts = now.to_string();
     let message = format!("{}:{}", version, ts);
 
-    let mut mac = HmacSha256::new_from_slice(secret.as_bytes())
+    let mut mac = HmacSha256::new_from_slice(secret.as_bytes()).expect("HMAC key init");
-        .expect("HMAC key init");
     mac.update(message.as_bytes());
     let sig = hex::encode(mac.finalize().into_bytes());
 
@@ -179,7 +194,8 @@ pub async fn get_update_info(
         return (
             StatusCode::BAD_REQUEST,
             Json(serde_json::json!({ "error": "Missing signature (ts and sig params required)" })),
-        ).into_response();
+        )
+            .into_response();
     }
 
     // 1. Parsear versión actual
@@ -273,7 +289,8 @@ pub async fn download_agent(
         return (
             StatusCode::BAD_REQUEST,
             Json(serde_json::json!({ "error": "Missing signature (ts and sig params required)" })),
-        ).into_response();
+        )
+            .into_response();
     }
 
     let version = version_clean;

services/backend-api/src/lib.rs

@@ -11,12 +11,12 @@ pub mod invitation_tokens;
 pub mod metrics;
 pub mod mqtt_loop;
 pub mod notifier;
-pub mod ws;
 pub mod passwords;
 pub mod rate_limiter;
 pub mod smtp_mailer;
 pub mod validation;
 pub mod watchdog;
+pub mod ws;
 
 use axum::extract::FromRef;
 use serde::{Deserialize, Serialize};

services/backend-api/src/main.rs

@@ -3,8 +3,8 @@ use axum::{
     routing::{get, patch, post, put},
 };
 use backend_api::{
-    AppState, TelemetryEvent, api_version, crypto, db, handlers, metrics, mqtt_loop,
+    AppState, TelemetryEvent, api_version, crypto, db, handlers, metrics, mqtt_loop, rate_limiter,
-    rate_limiter, watchdog, ws,
+    watchdog, ws,
 };
 use dotenvy::dotenv;
 use std::net::SocketAddr;
@@ -130,13 +130,7 @@ async fn main() {
     let mqtt_pool = pool.clone();
     let mqtt_tx = tx.clone();
     tokio::spawn(async move {
-        mqtt_loop::run_event_loop(
+        mqtt_loop::run_event_loop(eventloop, mqtt_client_for_loop, mqtt_pool, mqtt_tx).await;
-            eventloop,
-            mqtt_client_for_loop,
-            mqtt_pool,
-            mqtt_tx,
-        )
-        .await;
     });
 
     // Suscribirse a todos los topics de omnioil y de provisioning

services/backend-api/src/mqtt_loop.rs

@@ -1,6 +1,4 @@
-use crate::{
+use crate::{TelemetryEvent, metrics, notifier};
-    TelemetryEvent, metrics, notifier,
-};
 use rumqttc::{AsyncClient, Event, Packet};
 use sqlx::PgPool;
 use std::sync::Arc;
@@ -183,13 +181,14 @@ pub async fn run_event_loop(
                             let payload_bytes = publish.payload.to_vec();
 
                             tokio::spawn(async move {
-                                if let Err(e) = crate::handlers::provisioning::handle_mqtt_provisioning(
+                                if let Err(e) =
-                                    &pool_ref,
+                                    crate::handlers::provisioning::handle_mqtt_provisioning(
-                                    &client_ref,
+                                        &pool_ref,
-                                    &project_id_str,
+                                        &client_ref,
-                                    &payload_bytes,
+                                        &project_id_str,
-                                )
+                                        &payload_bytes,
-                                .await
+                                    )
+                                    .await
                                 {
                                     tracing::error!(
                                         "❌ Error en provisioning MQTT para proyecto {}: {}",

services/backend-api/src/ws.rs

@@ -1,11 +1,12 @@
+use crate::{AppState, auth};
 use axum::{
-    extract::{Query, State, ws::{Message, WebSocket, WebSocketUpgrade}},
+    Json, Router,
+    extract::{
+        Query, State,
+        ws::{Message, WebSocket, WebSocketUpgrade},
+    },
     response::IntoResponse,
     routing::{get, post},
-    Json, Router,
-};
-use crate::{
-    auth, AppState,
 };
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};

services/build-orchestrator/src/main.rs

@@ -143,7 +143,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 .unwrap();
             (
                 axum::http::StatusCode::OK,
-                [(axum::http::header::CONTENT_TYPE, "text/plain; charset=utf-8")],
+                [(
+                    axum::http::header::CONTENT_TYPE,
+                    "text/plain; charset=utf-8",
+                )],
                 buffer,
             )
         }),

services/build-orchestrator/src/metrics.rs

@@ -1,4 +1,4 @@
-use prometheus::{register_counter, Counter};
+use prometheus::{Counter, register_counter};
 use std::sync::LazyLock;
 
 pub static BUILDS_TRIGGERED: LazyLock<Counter> = LazyLock::new(|| {

services/edge-agent/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "edge-agent"
-version = "0.3.5"
+version = "0.3.6"
 edition = "2024"
 
 [dependencies]

services/edge-agent/src/bin/edge-tray.rs

@@ -24,10 +24,19 @@ mod app {
 
     const PIPE_NAME: &str = r"\\.\pipe\omnioil-edge-agent";
 
+    // ─── Colores branding OmniOil ──────────────────────────────────────────────
+    const BRAND_GOLD: [u8; 3] = [212, 160, 23]; // #D4A017 — dorado llama
+    const BRAND_RED: [u8; 3] = [231, 76, 60]; // #E74C3C — rojo desconectado
+    const BRAND_BG: [u8; 3] = [26, 26, 46]; // #1A1A2E — fondo oscuro
+
     struct EdgeTrayApp {
         tray_icon: Option<TrayIcon>,
         quit_id: MenuId,
-        test_id: MenuId,
+        test_connectivity_id: MenuId,
+        restart_service_id: MenuId,
+        open_logs_id: MenuId,
+        open_install_id: MenuId,
+        about_id: MenuId,
         status_label: MenuItem,
         command_tx: tokio::sync::mpsc::Sender<AgentCommand>,
         status_rx: tokio::sync::mpsc::Receiver<IpcMessage>,
@@ -36,6 +45,7 @@ mod app {
         animation_tick: usize,
         last_tick_time: std::time::Instant,
         connected_mqtt: bool,
+        exe_dir: std::path::PathBuf,
     }
 
     impl ApplicationHandler for EdgeTrayApp {
@@ -125,7 +135,10 @@ mod app {
                         }
                     }
                     IpcMessage::Notification { title, message } => {
-                        println!("[NOTIFICATION] {}: {}", title, message);
+                        self.status_label
+                            .set_text(format!("{}: {}", title, message));
+                        // Restaurar al estado después de 5 segundos
+                        // (el próximo StatusUpdate lo restaurará automáticamente)
                     }
                     _ => {}
                 }
@@ -148,10 +161,28 @@ mod app {
                 if event.id == self.quit_id {
                     self.tray_icon.take();
                     event_loop.exit();
-                } else if event.id == self.test_id {
+                } else if event.id == self.test_connectivity_id {
-                    let _ = self.command_tx.try_send(AgentCommand::Notify(
+                    let _ = self.command_tx.try_send(AgentCommand::CheckConnectivity);
-                        "¡Hola desde el Tray Icon!".to_string(),
+                } else if event.id == self.restart_service_id {
-                    ));
+                    let _ = self.command_tx.try_send(AgentCommand::RestartService);
+                } else if event.id == self.open_logs_id {
+                    let logs_dir = self.exe_dir.join("logs");
+                    let _ = std::process::Command::new("explorer.exe")
+                        .arg(&logs_dir)
+                        .spawn();
+                } else if event.id == self.open_install_id {
+                    let _ = std::process::Command::new("explorer.exe")
+                        .arg(&self.exe_dir)
+                        .spawn();
+                } else if event.id == self.about_id {
+                    let version = env!("CARGO_PKG_VERSION");
+                    let status = if self.connected_mqtt {
+                        "MQTT: Conectado"
+                    } else {
+                        "MQTT: Desconectado"
+                    };
+                    self.status_label
+                        .set_text(format!("OmniOil Edge Agent v{} — {}", version, status));
                 }
             }
         }
@@ -161,20 +192,41 @@ mod app {
         let event_loop = EventLoop::new().unwrap();
         event_loop.set_control_flow(ControlFlow::Wait);
 
+        // Obtener directorio del ejecutable para acciones locales
+        let exe_dir = std::env::current_exe()
+            .unwrap()
+            .parent()
+            .unwrap()
+            .to_path_buf();
+
         // Setup Tray Menu
         let tray_menu = Menu::new();
         let status_label = MenuItem::new("Estado: Conectando...", false, None);
-        let test_i = MenuItem::new("Probar Notificación", true, None);
+        let test_connectivity_i = MenuItem::new("Probar Conexión", true, None);
+        let restart_service_i = MenuItem::new("Reiniciar Servicio", true, None);
+        let open_logs_i = MenuItem::new("Abrir Logs", true, None);
+        let open_install_i = MenuItem::new("Abrir Carpeta Instalación", true, None);
+        let about_i = MenuItem::new("Acerca de...", true, None);
         let quit_i = MenuItem::new("Salir", true, None);
 
         let quit_id = quit_i.id().clone();
-        let test_id = test_i.id().clone();
+        let test_connectivity_id = test_connectivity_i.id().clone();
+        let restart_service_id = restart_service_i.id().clone();
+        let open_logs_id = open_logs_i.id().clone();
+        let open_install_id = open_install_i.id().clone();
+        let about_id = about_i.id().clone();
 
         tray_menu
             .append_items(&[
                 &status_label,
-                &test_i,
                 &PredefinedMenuItem::separator(),
+                &test_connectivity_i,
+                &restart_service_i,
+                &PredefinedMenuItem::separator(),
+                &open_logs_i,
+                &open_install_i,
+                &PredefinedMenuItem::separator(),
+                &about_i,
                 &quit_i,
             ])
             .unwrap();
@@ -253,7 +305,11 @@ mod app {
         let mut app = EdgeTrayApp {
             tray_icon: Some(tray_icon),
             quit_id,
-            test_id,
+            test_connectivity_id,
+            restart_service_id,
+            open_logs_id,
+            open_install_id,
+            about_id,
             status_label,
             command_tx,
             status_rx,
@@ -261,6 +317,7 @@ mod app {
             animation_tick: 0,
             last_tick_time: std::time::Instant::now(),
             connected_mqtt: false,
+            exe_dir,
         };
 
         event_loop.run_app(&mut app).unwrap();
@@ -278,7 +335,7 @@ mod app {
             }
         }
 
-        // Generar icono de estado elegante de 16x16 en memoria
+        // Generar icono de estado elegante de 16x16 en memoria (branding OmniOil)
         let width = 16;
         let height = 16;
         let mut rgba = vec![0u8; (width * height * 4) as usize];
@@ -290,23 +347,23 @@ mod app {
                 let is_dot = x >= 11 && x <= 14 && y >= 11 && y <= 14;
                 if is_dot {
                     if connected {
-                        // Punto Verde (Conectado)
+                        // Punto Dorado OmniOil (Conectado)
-                        rgba[idx] = 46;
+                        rgba[idx] = BRAND_GOLD[0];
-                        rgba[idx + 1] = 204;
+                        rgba[idx + 1] = BRAND_GOLD[1];
-                        rgba[idx + 2] = 113;
+                        rgba[idx + 2] = BRAND_GOLD[2];
                         rgba[idx + 3] = 255;
                     } else {
                         // Punto Rojo (Desconectado)
-                        rgba[idx] = 231;
+                        rgba[idx] = BRAND_RED[0];
-                        rgba[idx + 1] = 76;
+                        rgba[idx + 1] = BRAND_RED[1];
-                        rgba[idx + 2] = 60;
+                        rgba[idx + 2] = BRAND_RED[2];
                         rgba[idx + 3] = 255;
                     }
                 } else {
-                    // Fondo gris slate de tema oscuro moderno
+                    // Fondo oscuro consistente con branding
-                    rgba[idx] = 44;
+                    rgba[idx] = BRAND_BG[0];
-                    rgba[idx + 1] = 62;
+                    rgba[idx + 1] = BRAND_BG[1];
-                    rgba[idx + 2] = 80;
+                    rgba[idx + 2] = BRAND_BG[2];
                     rgba[idx + 3] = 255;
                 }
             }
@@ -336,16 +393,16 @@ mod app {
                 let dist_sq = dx * dx + dy * dy;
 
                 if dist_sq <= 1 {
-                    // Punto naranja brillante de animación
+                    // Punto dorado brillante de animación (branding OmniOil)
-                    rgba[idx] = 255;
+                    rgba[idx] = BRAND_GOLD[0];
-                    rgba[idx + 1] = 165;
+                    rgba[idx + 1] = BRAND_GOLD[1];
-                    rgba[idx + 2] = 0;
+                    rgba[idx + 2] = BRAND_GOLD[2];
                     rgba[idx + 3] = 255;
                 } else {
-                    // Fondo azul oscuro moderno
+                    // Fondo oscuro consistente
-                    rgba[idx] = 0;
+                    rgba[idx] = BRAND_BG[0];
-                    rgba[idx + 1] = 50;
+                    rgba[idx + 1] = BRAND_BG[1];
-                    rgba[idx + 2] = 100;
+                    rgba[idx + 2] = BRAND_BG[2];
                     rgba[idx + 3] = 255;
                 }
             }

services/edge-agent/src/data_collector.rs

@@ -433,10 +433,7 @@ async fn run_device_collector(spec: DeviceTaskSpec) {
                             let _ = storage.save(&topic, &payload_str).await;
                         } else {
                             let _ = reporter
-                                .info(
+                                .info("MQTT", &format!("✅ Data sent for {}", device.name))
-                                    "MQTT",
-                                    &format!("✅ Data sent for {}", device.name),
-                                )
                                 .await;
                         }
                     }
@@ -444,17 +441,13 @@ async fn run_device_collector(spec: DeviceTaskSpec) {
                     let _ = reporter
                         .warn(
                             "COLLECTOR",
-                            &format!(
+                            &format!("⚠️ No values returned for device {}", device.name),
-                                "⚠️ No values returned for device {}",
-                                device.name
-                            ),
                         )
                         .await;
                 }
             }
             Err(e) => {
-                let msg =
+                let msg = format!("❌ Error reading from device {}: {}", device.name, e);
-                    format!("❌ Error reading from device {}: {}", device.name, e);
                 tracing::error!("{}", msg);
                 let _ = reporter.error("COLLECTOR", &msg).await;
                 let _ = driver.disconnect().await;

services/edge-agent/src/deadband.rs

@@ -53,11 +53,7 @@ impl DeadbandFilter {
             }
             Some(entry) => {
                 let change_pct = if entry.last_value == 0.0 {
-                    if new_val == 0.0 {
+                    if new_val == 0.0 { 0.0 } else { 100.0 }
-                        0.0
-                    } else {
-                        100.0
-                    }
                 } else {
                     ((new_val - entry.last_value) / entry.last_value.abs() * 100.0).abs()
                 };

services/edge-agent/src/ipc_protocol.rs

@@ -27,4 +27,10 @@ pub enum IpcMessage {
 pub enum AgentCommand {
     Restart,
     Notify(String),
+    /// Detiene y reinicia el servicio Windows OmniOilEdgeAgent via SCM.
+    RestartService,
+    /// Envía un mensaje de prueba al servidor vía MQTT (log reporter) para
+    /// verificar conectividad end-to-end. La latencia es ~RTT MQTT (100-500ms
+    /// local, 1-3s internet).
+    CheckConnectivity,
 }

services/edge-agent/src/ipc_server.rs

@@ -8,7 +8,10 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
 #[cfg(windows)]
 use tokio::net::windows::named_pipe::ServerOptions;
 
-pub async fn start_ipc_server(receiver: broadcast::Receiver<IpcMessage>) -> anyhow::Result<()> {
+pub async fn start_ipc_server(
+    receiver: broadcast::Receiver<IpcMessage>,
+    tray_cmd_tx: tokio::sync::mpsc::Sender<AgentCommand>,
+) -> anyhow::Result<()> {
     let pipe_name = r"\\.\pipe\omnioil-edge-agent";
 
     tracing::info!("📡 IPC Server (Named Pipes) listening on {}...", pipe_name);
@@ -48,8 +51,9 @@ pub async fn start_ipc_server(receiver: broadcast::Receiver<IpcMessage>) -> anyh
             }
 
             let receiver_clone = receiver.resubscribe();
+            let cmd_tx = tray_cmd_tx.clone();
             tokio::spawn(async move {
-                if let Err(e) = handle_client(server, receiver_clone).await {
+                if let Err(e) = handle_client(server, receiver_clone, cmd_tx).await {
                     tracing::warn!("⚠️ IPC client disconnected: {}", e);
                 }
             });
@@ -59,6 +63,7 @@ pub async fn start_ipc_server(receiver: broadcast::Receiver<IpcMessage>) -> anyh
         #[cfg(not(windows))]
         {
             let _ = &receiver;
+            let _ = &tray_cmd_tx;
             tokio::time::sleep(std::time::Duration::from_secs(3600)).await;
         }
     }
@@ -75,8 +80,8 @@ fn create_pipe_server(
     use std::ptr::addr_of_mut;
     use windows_sys::Win32::Foundation::{FALSE, TRUE};
     use windows_sys::Win32::Security::{
-        InitializeSecurityDescriptor, SetSecurityDescriptorDacl, SECURITY_ATTRIBUTES,
+        InitializeSecurityDescriptor, SECURITY_ATTRIBUTES, SECURITY_DESCRIPTOR,
-        SECURITY_DESCRIPTOR,
+        SetSecurityDescriptorDacl,
     };
     // SECURITY_DESCRIPTOR_REVISION siempre vale 1 según el Windows SDK; la
     // constante fue eliminada en versiones nuevas de windows-sys.
@@ -123,6 +128,7 @@ fn create_pipe_server(
 async fn handle_client(
     server: tokio::net::windows::named_pipe::NamedPipeServer,
     mut receiver: broadcast::Receiver<IpcMessage>,
+    tray_cmd_tx: tokio::sync::mpsc::Sender<AgentCommand>,
 ) -> anyhow::Result<()> {
     let (read_half, mut write_half) = tokio::io::split(server);
     let mut reader = BufReader::new(read_half);
@@ -147,13 +153,21 @@ async fn handle_client(
                     Ok(0) | Err(_) => break,
                     Ok(_) => {
                         if let Ok(IpcMessage::Command(cmd)) = serde_json::from_str(&line) {
-                            match cmd {
+                            match &cmd {
                                 AgentCommand::Restart => tracing::warn!("🔄 Restart requested!"),
                                 AgentCommand::Notify(msg) => {
-                                    let resp = IpcMessage::Notification { title: "Agent".into(), message: msg };
+                                    let resp = IpcMessage::Notification { title: "Agent".into(), message: msg.clone() };
                                     let _ = write_half.write_all((serde_json::to_string(&resp)? + "\n").as_bytes()).await;
                                 }
+                                AgentCommand::RestartService => {
+                                    tracing::info!("🔄 RestartService requested from tray icon");
+                                }
+                                AgentCommand::CheckConnectivity => {
+                                    tracing::info!("📡 CheckConnectivity requested from tray icon");
+                                }
                             }
+                            // Reenviar comandos que requieren acción del agente principal
+                            let _ = tray_cmd_tx.send(cmd).await;
                         }
                         line.clear();
                     }

services/edge-agent/src/main.rs

@@ -107,19 +107,19 @@ fn main() -> Result<()> {
                         let opts = rumqttc::MqttOptions::new("crash-reporter", mqtt_host, port);
                         let (client, mut eventloop) = rumqttc::AsyncClient::new(opts, 4);
                         // Dar 2s para conectar y enviar
-                        let _ = tokio::time::timeout(
+                        let _ = tokio::time::timeout(std::time::Duration::from_secs(2), async {
-                            std::time::Duration::from_secs(2),
+                            let _ = eventloop.poll().await; // ConnAck
-                            async {
+                            let _ = client
-                                let _ = eventloop.poll().await; // ConnAck
+                                .publish(
-                                let _ = client.publish(
                                     &topic,
                                     rumqttc::QoS::AtMostOnce,
                                     false,
                                     serde_json::to_vec(&log_payload).unwrap_or_default(),
-                                ).await;
+                                )
-                                let _ = eventloop.poll().await; // flush
+                                .await;
-                            }
+                            let _ = eventloop.poll().await; // flush
-                        ).await;
+                        })
+                        .await;
                     }
                 });
             }
@@ -300,8 +300,11 @@ pub async fn run_agent(
     let (ipc_tx, _) = tokio::sync::broadcast::channel::<ipc_protocol::IpcMessage>(32);
     let ipc_rx = ipc_tx.subscribe();
 
+    // Canal para comandos desde el tray icon hacia el agente principal
+    let (tray_cmd_tx, mut tray_cmd_rx) = mpsc::channel::<ipc_protocol::AgentCommand>(16);
+
     tokio::spawn(async move {
-        let _ = ipc_server::start_ipc_server(ipc_rx).await;
+        let _ = ipc_server::start_ipc_server(ipc_rx, tray_cmd_tx).await;
     });
 
     // ─── Limpieza de logs locales ───────────────────────────────────────────────
@@ -556,6 +559,42 @@ pub async fn run_agent(
                     }
                 }
             }
+            Some(tray_cmd) = tray_cmd_rx.recv() => {
+                match tray_cmd {
+                    ipc_protocol::AgentCommand::RestartService => {
+                        log_reporter.info("TRAY", "Restart service requested from tray icon").await;
+                        #[cfg(windows)]
+                        {
+                            // Detener el servicio — esto terminará el proceso actual.
+                            // El Service Control Manager lo reiniciará automáticamente
+                            // gracias a la política de recuperación configurada.
+                            let _ = std::process::Command::new("sc")
+                                .arg("stop")
+                                .arg("OmniOilEdgeAgent")
+                                .output();
+                        }
+                        #[cfg(not(windows))]
+                        {
+                            tracing::warn!("RestartService not supported on non-Windows");
+                        }
+                    }
+                    ipc_protocol::AgentCommand::CheckConnectivity => {
+                        log_reporter.info("CONNECTIVITY_TEST", "Test message from tray icon — connectivity OK").await;
+                        let _ = ipc_tx.send(ipc_protocol::IpcMessage::Notification {
+                            title: "Conectividad".to_string(),
+                            message: "Conexión verificada — el agente está comunicando con el servidor.".to_string(),
+                        });
+                    }
+                    ipc_protocol::AgentCommand::Notify(msg) => {
+                        log_reporter.info("TRAY", &format!("Test notification: {}", msg)).await;
+                        let _ = ipc_tx.send(ipc_protocol::IpcMessage::Notification {
+                            title: "Agent".to_string(),
+                            message: msg,
+                        });
+                    }
+                    _ => {}
+                }
+            }
             _ = tokio::time::sleep(std::time::Duration::from_secs(15)) => {
                 // Despierta periódicamente el bucle para actualizar el heartbeat
                 // del watchdog y demostrar que el runtime de Tokio sigue activo.

services/edge-agent/src/persistence_manager.rs

@@ -53,8 +53,8 @@ impl StoreAndForward {
         // Salt aleatorio por dispositivo — persistir junto al DB
         let salt_path = db_path.with_extension("salt");
         let salt: [u8; 16] = if salt_path.exists() {
-            let existing = std::fs::read(&salt_path)
+            let existing =
-                .context("Failed to read existing salt file")?;
+                std::fs::read(&salt_path).context("Failed to read existing salt file")?;
             existing.try_into().unwrap_or_else(|_| {
                 tracing::warn!("Salt file corrupt, generating new salt");
                 let new: [u8; 16] = rand::random();
@@ -63,8 +63,7 @@ impl StoreAndForward {
             })
         } else {
             let new: [u8; 16] = rand::random();
-            std::fs::write(&salt_path, &new)
+            std::fs::write(&salt_path, &new).context("Failed to write salt file")?;
-                .context("Failed to write salt file")?;
             new
         };
 

services/edge-agent/src/updater.rs

@@ -25,11 +25,11 @@
 //!   credenciales almacenadas en el agente. La URL es temporal (TTL 30 min).
 //! - El agente nunca expone ningún puerto HTTP.
 
+use crate::ipc_protocol::IpcMessage;
 use anyhow::{Context, Result, bail};
 use semver::Version;
 use sha2::{Digest, Sha256};
 use tracing::{info, warn};
-use crate::ipc_protocol::IpcMessage;
 
 /// Aplica una actualización OTA a partir de los datos recibidos vía MQTT.
 ///
@@ -91,7 +91,8 @@ pub async fn apply_ota_from_command(
         .build()
         .context("Error al construir cliente HTTP para descarga OTA")?;
 
-    let content = match download_asset(&client, download_url, target_version, ipc_tx.as_ref()).await {
+    let content = match download_asset(&client, download_url, target_version, ipc_tx.as_ref()).await
+    {
         Ok(c) => c,
         Err(e) => {
             if let Some(ref tx) = ipc_tx {

services/edge-agent/wix/main.wxs

@@ -1,7 +1,7 @@
 <?xml version='1.0' encoding='windows-1252'?>
 <Wix xmlns='http://schemas.microsoft.com/wix/2006/wi'>
     <?ifndef Version?>
-        <?define Version = "0.3.0" ?>
+        <?define Version = "0.3.6" ?>
     <?endif?>
     <Product Name='OmniOil Edge Agent' 
              Id='*' 

services/events-relay/src/main.rs

@@ -22,8 +22,7 @@ async fn main() -> Result<()> {
         .json()
         .flatten_event(false)
         .with_env_filter(
-            tracing_subscriber::EnvFilter::try_from_default_env()
+            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
-                .unwrap_or_else(|_| "info".into()),
         )
         .init();
 
@@ -39,7 +38,10 @@ async fn main() -> Result<()> {
                 .unwrap();
             (
                 axum::http::StatusCode::OK,
-                [(axum::http::header::CONTENT_TYPE, "text/plain; charset=utf-8")],
+                [(
+                    axum::http::header::CONTENT_TYPE,
+                    "text/plain; charset=utf-8",
+                )],
                 buffer,
             )
         }),

services/events-relay/src/metrics.rs

@@ -1,4 +1,4 @@
-use prometheus::{register_counter, register_gauge, Counter, Gauge};
+use prometheus::{Counter, Gauge, register_counter, register_gauge};
 use std::sync::LazyLock;
 
 pub static OUTBOX_EVENTS_PROCESSED: LazyLock<Counter> = LazyLock::new(|| {

services/json-generator/src/main.rs

@@ -39,8 +39,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         .json()
         .flatten_event(false)
         .with_env_filter(
-            tracing_subscriber::EnvFilter::try_from_default_env()
+            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
-                .unwrap_or_else(|_| "info".into()),
         )
         .init();
 
@@ -56,7 +55,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
                 .unwrap();
             (
                 axum::http::StatusCode::OK,
-                [(axum::http::header::CONTENT_TYPE, "text/plain; charset=utf-8")],
+                [(
+                    axum::http::header::CONTENT_TYPE,
+                    "text/plain; charset=utf-8",
+                )],
                 buffer,
             )
         }),

services/json-generator/src/metrics.rs

@@ -1,4 +1,4 @@
-use prometheus::{register_counter, Counter};
+use prometheus::{Counter, register_counter};
 use std::sync::LazyLock;
 
 pub static REPORTS_GENERATED: LazyLock<Counter> = LazyLock::new(|| {

services/telemetry-ingestor/src/main.rs

@@ -28,8 +28,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .json()
         .flatten_event(false)
         .with_env_filter(
-            tracing_subscriber::EnvFilter::try_from_default_env()
+            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| "info".into()),
-                .unwrap_or_else(|_| "info".into()),
         )
         .init();
 
@@ -45,7 +44,10 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                 .unwrap();
             (
                 axum::http::StatusCode::OK,
-                [(axum::http::header::CONTENT_TYPE, "text/plain; charset=utf-8")],
+                [(
+                    axum::http::header::CONTENT_TYPE,
+                    "text/plain; charset=utf-8",
+                )],
                 buffer,
             )
         }),
@@ -317,7 +319,9 @@ async fn process_message(
                 return Ok(());
             }
 
-            let asset_id = match resolve_device(pool, cache, project_id, &payload.device_name).await? {
+            let asset_id = match resolve_device(pool, cache, project_id, &payload.device_name)
+                .await?
+            {
                 device_cache::DeviceResolution::Resolved(asset_id) => asset_id,
                 device_cache::DeviceResolution::Unresolved => {
                     warn!(

services/telemetry-ingestor/src/metrics.rs

@@ -1,5 +1,5 @@
 use prometheus::{
-    register_counter, register_counter_vec, register_gauge, Counter, CounterVec, Gauge,
+    Counter, CounterVec, Gauge, register_counter, register_counter_vec, register_gauge,
 };
 use std::sync::LazyLock;