docker-compose.yml

@@ -121,24 +121,7 @@ services:
       - anh_network
 
   # ---------------------------------------------------------------------------
-  # GATEWAY (Punto de entrada unificado para subdominios)
+  # FRONTENDS (Entradas Principales - SaaS Architecture - Manejado por Traefik)
-  # ---------------------------------------------------------------------------
-  nginx-gateway:
-    image: nginx:alpine
-    container_name: anh_nginx_gateway
-    ports:
-      - "80:80"
-    volumes:
-      - ./infrastructure/nginx/gateway.conf:/etc/nginx/conf.d/default.conf
-    depends_on:
-      - landing
-      - frontend
-      - frontend-pwa
-    networks:
-      - anh_network
-
-  # ---------------------------------------------------------------------------
-  # FRONTENDS (Entradas Principales - SaaS Architecture)
   # ---------------------------------------------------------------------------
 
   # Landing Page: omnioil.app / www.omnioil.app
@@ -147,8 +130,10 @@ services:
     build:
       context: .
       dockerfile: services/landing/Dockerfile
-    ports:
+    labels:
-      - "8080:80"
+      - "traefik.enable=true"
+      - "traefik.http.routers.landing.rule=Host(`${DOMAIN:-omnioil.app}`, `www.${DOMAIN:-omnioil.app}`)"
+      - "traefik.http.services.landing.loadbalancer.server.port=80"
     networks:
       - anh_network
 
@@ -158,8 +143,10 @@ services:
     build:
       context: .
       dockerfile: services/frontend-pwa/Dockerfile
-    ports:
+    labels:
-      - "8081:80"
+      - "traefik.enable=true"
+      - "traefik.http.routers.pwa.rule=Host(`campo.${DOMAIN:-omnioil.app}`)"
+      - "traefik.http.services.pwa.loadbalancer.server.port=80"
     depends_on:
       - backend-api
     networks:
@@ -170,11 +157,13 @@ services:
     container_name: anh_frontend
     build:
       context: ./services/frontend-admin
-      dockerfile: services/frontend/Dockerfile
+      dockerfile: Dockerfile
       args:
         - VITE_API_BASE=${VITE_API_BASE:-/api}
-    ports:
+    labels:
-      - "8082:80"
+      - "traefik.enable=true"
+      - "traefik.http.routers.admin.rule=Host(`admin.${DOMAIN:-omnioil.app}`)"
+      - "traefik.http.services.admin.loadbalancer.server.port=80"
     depends_on:
       - backend-api
     networks:

infrastructure/postgres/init.sql

@@ -59,6 +59,9 @@ CREATE TABLE IF NOT EXISTS edge_projects (
     description TEXT,
     metadata JSONB DEFAULT '{}'::jsonb,
     is_active BOOLEAN DEFAULT true,
+    installer_status VARCHAR(50) DEFAULT 'NOT_STARTED'
+        CHECK (installer_status IN ('NOT_STARTED', 'QUEUED', 'BUILDING', 'COMPLETED', 'FAILED')),
+    installer_url TEXT,
     created_at TIMESTAMPTZ DEFAULT NOW(),
     updated_at TIMESTAMPTZ DEFAULT NOW()
 );

services/backend-api/Cargo.toml

@@ -22,4 +22,5 @@ chrono = { version = "0.4.44", features = ["serde"] }
 axum-extra = { version = "0.12.5", features = ["typed-header"] }
 uuid = { version = "1.23.0", features = ["v4", "serde"] }
 anyhow = "1.0.102"
+rumqttc = "0.25.1"
 

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

@@ -25,6 +25,7 @@ pub async fn list_projects(
 /// POST /api/edge/projects — Crear un proyecto.
 pub async fn create_project(
     State(pool): State<PgPool>,
+    State(mqtt): State<rumqttc::AsyncClient>,
     Json(req): Json<CreateProjectRequest>,
 ) -> Result<(StatusCode, Json<EdgeProject>), (StatusCode, String)> {
     let project = sqlx::query_as::<_, EdgeProject>(
@@ -40,29 +41,24 @@ pub async fn create_project(
     .await
     .map_err(|e| (StatusCode::INTERNAL_SERVER_ERROR, e.to_string()))?;
 
-    // ─── Dispatch Domain Event via Postgres NOTIFY (Internal Messaging) ──────────────────────────────
+    // ─── Dispatch Domain Event via MQTT (Internal Messaging) ──────────────────────────────
-    let project_id_clone = project.id;
-    let name_clone = project.name.clone();
-    let client_clone = project.client.clone();
-
     let event = shared_lib::mqtt_messages::ProjectCreatedEvent {
-        project_id: project_id_clone,
+        project_id: project.id,
-        name: name_clone,
+        name: project.name.clone(),
-        client: client_clone,
+        client: project.client.clone(),
         timestamp: chrono::Utc::now(),
     };
 
     if let Ok(payload) = serde_json::to_string(&event) {
-        tracing::info!(
+        tracing::info!("📡 Publishing Domain Event via MQTT: project_created ({})", project.id);
-            "📡 Publishing Domain Event (Internal DB Pub/Sub): project_created ({})",
+        let mqtt_clone = mqtt.clone();
-            project_id_clone
-        );
-        let notify_query = format!(
-            "NOTIFY omnioil_domain_events, '{}'",
-            payload.replace("'", "''")
-        );
         tokio::spawn(async move {
-            let _ = sqlx::query(&notify_query).execute(&pool).await;
+            if let Err(e) = mqtt_clone
+                .publish("omnioil/events/project_created", rumqttc::QoS::AtLeastOnce, false, payload)
+                .await
+            {
+                tracing::error!("Failed to publish ProjectCreatedEvent to MQTT: {:?}", e);
+            }
         });
     }
 

services/backend-api/src/main.rs

@@ -10,6 +10,26 @@ use dotenv::dotenv;
 use std::net::SocketAddr;
 use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt};
 
+use axum::extract::FromRef;
+
+#[derive(Clone)]
+pub struct AppState {
+    pub pool: sqlx::PgPool,
+    pub mqtt_client: rumqttc::AsyncClient,
+}
+
+impl FromRef<AppState> for sqlx::PgPool {
+    fn from_ref(state: &AppState) -> Self {
+        state.pool.clone()
+    }
+}
+
+impl FromRef<AppState> for rumqttc::AsyncClient {
+    fn from_ref(state: &AppState) -> Self {
+        state.mqtt_client.clone()
+    }
+}
+
 #[tokio::main]
 async fn main() {
     dotenv().ok();
@@ -25,12 +45,36 @@ async fn main() {
 
     tracing::info!("Starting Backend API...");
 
+    // Conectar al MQTT broker
+    let mqtt_host = std::env::var("MQTT_HOST").unwrap_or_else(|_| "localhost".to_string());
+    let mqtt_port: u16 = std::env::var("MQTT_PORT")
+        .unwrap_or_else(|_| "1883".to_string())
+        .parse()
+        .unwrap_or(1883);
+
+    let mut mqtt_options = rumqttc::MqttOptions::new("omnioil-backend-api", &mqtt_host, mqtt_port);
+    mqtt_options.set_keep_alive(std::time::Duration::from_secs(30));
+
+    let (mqtt_client, mut eventloop) = rumqttc::AsyncClient::new(mqtt_options, 10);
+
+    // Spawn MQTT event loop para mantener la conexión viva
+    tokio::spawn(async move {
+        loop {
+            if let Err(e) = eventloop.poll().await {
+                tracing::error!("MQTT connection error: {:?}", e);
+                tokio::time::sleep(std::time::Duration::from_secs(5)).await;
+            }
+        }
+    });
+
     // Conectar a DB
     let pool = db::establish_connection()
         .await
         .expect("Failed to connect to database");
     tracing::info!("Connected to database.");
 
+    let state = AppState { pool, mqtt_client };
+
     // Rutas Edge System
     let edge_routes = Router::new()
         // Projects
@@ -126,7 +170,7 @@ async fn main() {
         .nest("/api/edge", edge_routes)
         // Estado compartido y Middleware
         .layer(cors)
-        .with_state(pool);
+        .with_state(state);
 
     let addr = SocketAddr::from(([0, 0, 0, 0], 8000));
     tracing::info!("listening on {}", addr);

services/build-orchestrator/src/main.rs

@@ -57,31 +57,23 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 
     let (mqtt_client, mut eventloop) = rumqttc::AsyncClient::new(mqtt_options, 10);
 
-    // SPAWN Domain Events listener (Internal Messaging)
+    // Subscribe to Domain Events
+    mqtt_client
+        .subscribe("omnioil/events/project_created", rumqttc::QoS::AtLeastOnce)
+        .await?;
+
+    // Spawn MQTT event loop (Handles Domain Events AND agent communication)
     let db_pool = pool.clone();
-    let database_url_listener = database_url.clone();
     tokio::spawn(async move {
-        // Obtenemos una conexión individual para el Listener
+        loop {
-        let mut listener = match sqlx::postgres::PgListener::connect(&database_url_listener).await {
+            match eventloop.poll().await {
-            Ok(l) => l,
+                Ok(item) => {
-            Err(e) => {
+                    if let rumqttc::Event::Incoming(rumqttc::Packet::Publish(publish)) = item {
-                tracing::error!("Failed to connect PgListener: {}", e);
+                        if publish.topic == "omnioil/events/project_created" {
-                return;
+                            let payload = String::from_utf8_lossy(&publish.payload);
-            }
+                            tracing::info!("📥 Recibido Domain Event (MQTT): {}", payload);
-        };
                             
-        if let Err(e) = listener.listen("omnioil_domain_events").await {
+                            if let Ok(event) = serde_json::from_str::<shared_lib::mqtt_messages::ProjectCreatedEvent>(&payload) {
-            tracing::error!("Failed to listen to omnioil_domain_events: {}", e);
-            return;
-        }
-
-        tracing::info!("👂 Escuchando dominio interno en omnioil_domain_events...");
-        while let Ok(notification) = listener.recv().await {
-            let payload = notification.payload();
-            tracing::info!("📥 Recibido Domain Event interno: {}", payload);
-            
-            // Intentar parsear como ProjectCreatedEvent
-            if let Ok(event) = serde_json::from_str::<shared_lib::mqtt_messages::ProjectCreatedEvent>(payload) {
                                 let pool_for_worker = db_pool.clone();
                                 tokio::spawn(async move {
                                     if let Err(e) = installer_worker::handle_project_created(pool_for_worker, event).await {
@@ -90,13 +82,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
                                 });
                             }
                         }
-    });
+                    }
-
+                }
-    // Spawn MQTT event loop (usado para agentes de borde, no backend-api)
-    tokio::spawn(async move {
-        loop {
-            match eventloop.poll().await {
-                Ok(_) => {}
                 Err(e) => {
                     tracing::error!("MQTT event loop error: {}", e);
                     tokio::time::sleep(std::time::Duration::from_secs(5)).await;
@@ -104,6 +91,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
             }
         }
     });
+
+    tracing::info!("Connected to MQTT broker at {}:{}", mqtt_host, mqtt_port);
     tracing::info!("Connected to MQTT broker at {}:{}", mqtt_host, mqtt_port);
 
     let registry_url =

services/flow-generator/src/flow_builder.rs

@@ -2,7 +2,7 @@
 //! a partir de la configuración de un proyecto.
 
 use crate::nodes::{alarm, common, modbus, mqtt_out, opcua};
-use serde_json::{json, Value};
+use serde_json::{Value, json};
 use shared_lib::edge_models::{DeviceConfig, DeviceProtocol, ProjectConfig};
 
 /// Configuración del broker MQTT al que Node-RED publicará telemetría.
@@ -422,6 +422,8 @@ mod tests {
             description: Some("Test project".to_string()),
             metadata: serde_json::json!({}),
             is_active: true,
+            installer_status: Some("NOT_STARTED".to_string()),
+            installer_url: None,
             created_at: Utc::now(),
             updated_at: Utc::now(),
         };
@@ -519,9 +521,11 @@ mod tests {
         assert_eq!(modbus_reads.len(), 2, "Expected 2 modbus-read nodes");
 
         // Should require node-red-contrib-modbus
-        assert!(result
+        assert!(
+            result
                 .npm_packages
-            .contains(&"node-red-contrib-modbus".to_string()));
+                .contains(&"node-red-contrib-modbus".to_string())
+        );
     }
 
     #[test]

services/frontend-admin/.dockerignore

@@ -0,0 +1,8 @@
+node_modules
+dist
+build.log
+.DS_Store
+.git
+.env
+.env.local
+! .env.example

services/frontend-admin/Dockerfile

@@ -12,15 +12,14 @@ RUN corepack enable && corepack prepare pnpm@latest --activate
 WORKDIR /app
 
 # Copiar archivos de definición de dependencias
-# Priorizamos pnpm-lock.yaml si existe, pero copiamos todos por compatibilidad
+COPY package.json ./
-COPY services/frontend/package.json ./
+COPY pnpm-lock.yaml* ./ 
-COPY services/frontend/pnpm-lock.yaml* ./ 
 
 # Instalar dependencias usando PNPM (más rápido que NPM)
 RUN pnpm install --frozen-lockfile || pnpm install
 
 # Copiar el resto del código fuente del frontend
-COPY services/frontend/ .
+COPY . .
 
 # Generar el build de producción
 RUN pnpm build