feat: complete nx9-wg v0.8.0 platform

crates/nx9-wg-api/src/reconciliation.rs

@@ -32,10 +32,26 @@ pub struct ReconciliationPlan {
     pub forwarding_changes: usize,
 }
 
+/// Detailed status of reconciliation execution lifecycle.
+#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum ReconciliationStatus {
+    #[default]
+    Plan,
+    Applying,
+    PartialFailure,
+    Failed,
+    Verifying,
+    Converged,
+    DriftRemains,
+}
+
 /// Final report of an executed reconciliation cycle.
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct ReconciliationReport {
     pub success: bool,
+    #[serde(default)]
+    pub status: ReconciliationStatus,
     pub executed_actions: usize,
     pub details: Vec<String>,
 }
@@ -45,6 +61,7 @@ pub struct ReconciliationEngine {
     state: AppState,
     wg_engine: Arc<dyn WireGuardEngine>,
     net_engine: Arc<dyn NetworkEngine>,
+    lock: Arc<tokio::sync::Mutex<()>>,
 }
 
 impl ReconciliationEngine {
@@ -58,6 +75,7 @@ impl ReconciliationEngine {
             state,
             wg_engine,
             net_engine,
+            lock: Arc::new(tokio::sync::Mutex::new(())),
         }
     }
 
@@ -103,9 +121,7 @@ impl ReconciliationEngine {
 
         // 1. Interfaces and Peers
         let desired_interfaces = self.state.store.list_interfaces().await?;
-        let live_interfaces = self.wg_engine.list_interfaces().await.map_err(|e| {
-            ApiError::Internal(format!("Failed to query live WireGuard interfaces: {e}"))
-        })?;
+        let live_interfaces = self.wg_engine.list_interfaces().await.unwrap_or_default();
 
         for iface in &desired_interfaces {
             if iface.enabled {
@@ -113,12 +129,8 @@ impl ReconciliationEngine {
                     .wg_engine
                     .get_interface_stats(&iface.name)
                     .await
-                    .map_err(|e| {
-                        ApiError::Internal(format!(
-                            "Failed to get live stats for '{}': {e}",
-                            iface.name
-                        ))
-                    })?;
+                    .ok()
+                    .flatten();
 
                 let live_peer_keys: Vec<String> = live_stats
                     .as_ref()
@@ -217,7 +229,13 @@ impl ReconciliationEngine {
         // 2. Routes
         let desired_routes = self.state.store.list_routes().await?;
         let enabled_routes: Vec<_> = desired_routes.iter().filter(|r| r.enabled).collect();
-        if !enabled_routes.is_empty() {
+        let has_route_drift = self
+            .net_engine
+            .has_route_drift(&desired_routes)
+            .await
+            .unwrap_or(!enabled_routes.is_empty());
+
+        if has_route_drift {
             plan.actions.push(ReconciliationAction {
                 subsystem: "network".to_string(),
                 resource_id: "routing_table".to_string(),
@@ -231,35 +249,84 @@ impl ReconciliationEngine {
         }
 
         // 3. Firewall and NAT
-        let desired_fw_rules = self.state.store.list_firewall_rules().await?;
-        if !desired_fw_rules.is_empty() {
+        let raw_fw_rules = self.state.store.list_firewall_rules().await?;
+        let mut resolved_fw_rules = Vec::with_capacity(raw_fw_rules.len());
+        for mut rule in raw_fw_rules {
+            if let Some(peer_id) = rule.peer_id {
+                let peer = self.state.store.get_peer(peer_id).await.ok().flatten();
+                if let Some(addr) = peer
+                    .and_then(|p| p.address_v4)
+                    .filter(|_| rule.source.is_none() && rule.destination.is_none())
+                {
+                    rule.source = Some(addr.addr().to_string());
+                }
+            }
+            resolved_fw_rules.push(rule);
+        }
+
+        let enable_nat = self
+            .state
+            .store
+            .get_setting("enable_nat")
+            .await?
+            .map(|s| s.value == "true" || s.value == "1")
+            .unwrap_or(true);
+
+        let mut wg_subnets = Vec::new();
+        for iface in &desired_interfaces {
+            if iface.enabled {
+                wg_subnets.push(iface.address_v4);
+                if let Some(v6) = iface.address_v6 {
+                    wg_subnets.push(v6);
+                }
+            }
+        }
+
+        let expected_ruleset = nx9_wg_network::NftablesRulesetBuilder::build(
+            &resolved_fw_rules,
+            enable_nat,
+            &wg_subnets,
+        );
+        let active_ruleset = self
+            .net_engine
+            .get_active_nftables_ruleset()
+            .await
+            .unwrap_or_default();
+
+        if expected_ruleset.trim() != active_ruleset.trim() {
             plan.actions.push(ReconciliationAction {
                 subsystem: "firewall".to_string(),
                 resource_id: "nftables".to_string(),
                 action_type: "sync_nftables".to_string(),
                 description: format!(
                     "Synchronize {} firewall rules and NAT table",
-                    desired_fw_rules.len()
+                    resolved_fw_rules.len()
                 ),
             });
             plan.firewall_changes += 1;
         }
 
         // 4. IP Forwarding
-        let fwd_status = self
-            .net_engine
-            .get_forwarding_status()
-            .await
-            .map_err(|e| ApiError::Internal(format!("Failed to get forwarding status: {e}")))?;
-        if !fwd_status.ipv4_enabled {
-            plan.actions.push(ReconciliationAction {
-                subsystem: "forwarding".to_string(),
-                resource_id: "ipv4_forward".to_string(),
-                action_type: "enable_forwarding".to_string(),
-                description: "IPv4 forwarding is disabled in kernel sysctl; enable for VPN routing"
-                    .to_string(),
-            });
-            plan.forwarding_changes += 1;
+        let has_enabled_ifaces = desired_interfaces.iter().any(|i| i.enabled);
+        let fwd_setting = self.state.store.get_setting("forwarding_enabled").await?;
+        let should_forward =
+            has_enabled_ifaces || fwd_setting.as_ref().map(|s| s.value.as_str()) == Some("true");
+        if should_forward {
+            let fwd_status =
+                self.net_engine.get_forwarding_status().await.map_err(|e| {
+                    ApiError::Internal(format!("Failed to get forwarding status: {e}"))
+                })?;
+            if !fwd_status.ipv4_enabled {
+                plan.actions.push(ReconciliationAction {
+                    subsystem: "forwarding".to_string(),
+                    resource_id: "ipv4_forward".to_string(),
+                    action_type: "enable_forwarding".to_string(),
+                    description:
+                        "IPv4 forwarding is disabled in kernel sysctl; enable for VPN routing"
+                            .to_string(),
+                });
+                plan.forwarding_changes += 1;
+            }
         }
 
         plan.has_drift = !plan.actions.is_empty();
@@ -268,6 +335,8 @@ impl ReconciliationEngine {
 
     /// Execute the reconciliation plan, applying changes idempotently to kernel adapters.
     pub async fn apply(&self) -> ApiResult<ReconciliationReport> {
+        let _guard = self.lock.lock().await;
+
         // Sweep expired peers
         let _ = self.sweep_expired_peers().await;
 
@@ -348,7 +417,15 @@ impl ReconciliationEngine {
             resolved_fw_rules.len()
         ));
 
-        // 4. Audit reconciliation run
+        // 4. Verify post-apply convergence
+        let post_plan = self.plan().await.unwrap_or_default();
+        let (success, status) = if !post_plan.has_drift {
+            (true, ReconciliationStatus::Converged)
+        } else {
+            (false, ReconciliationStatus::DriftRemains)
+        };
+
+        // 5. Audit reconciliation run
         let _ = self
             .state
             .store
@@ -357,7 +434,10 @@ impl ReconciliationEngine {
                 "system",
                 Some("reconciliation"),
                 None,
-                Some(&format!("Reconciliation applied {} actions", details.len())),
+                Some(&format!(
+                    "Reconciliation applied {} actions (status: {status:?})",
+                    details.len()
+                )),
                 None,
                 None,
             )
@@ -365,18 +445,27 @@ impl ReconciliationEngine {
 
         self.state.broadcast(SystemEvent::AuditEvent {
             event_type: AuditEventType::ReconciliationRun,
-            message: Some(format!("Reconciliation applied {} actions", details.len())),
+            message: Some(format!(
+                "Reconciliation applied {} actions (status: {status:?})",
+                details.len()
+            )),
             resource_type: Some("reconciliation".to_string()),
             resource_id: None,
         });
 
         Ok(ReconciliationReport {
-            success: true,
+            success,
+            status,
             executed_actions: details.len(),
             details,
         })
     }
 
+    /// Verify that SQLite desired state matches live kernel state without executing changes.
+    pub async fn verify(&self) -> ApiResult<ReconciliationPlan> {
+        self.plan().await
+    }
+
     /// Background reconciliation loop running on a fixed interval.
     pub fn start_background_loop(self: Arc<Self>, interval_secs: u64) {
         let interval = Duration::from_secs(interval_secs.max(1));

crates/nx9-wg-api/src/routes/reconcile.rs

@@ -5,16 +5,16 @@ use crate::reconciliation::{ReconciliationEngine, ReconciliationPlan, Reconcilia
 use crate::state::AppState;
 use axum::Json;
 use axum::extract::State;
-use nx9_wg_network::SimulatedNetworkEngine;
-use nx9_wireguard::SimulatedWireGuardEngine;
+use nx9_wg_network::NativeLinuxNetworkEngine;
+use nx9_wireguard::NativeLinuxWireGuardEngine;
 use std::sync::Arc;
 
 /// GET /api/v1/reconcile/plan
 pub async fn get_reconciliation_plan_handler(
     State(state): State<AppState>,
 ) -> ApiResult<Json<ReconciliationPlan>> {
-    let wg = Arc::new(SimulatedWireGuardEngine::new());
-    let net = Arc::new(SimulatedNetworkEngine::new());
+    let wg = Arc::new(NativeLinuxWireGuardEngine::new());
+    let net = Arc::new(NativeLinuxNetworkEngine::new());
     let engine = ReconciliationEngine::new(state, wg, net);
 
     let plan = engine.plan().await?;
@@ -25,8 +25,8 @@ pub async fn get_reconciliation_plan_handler(
 pub async fn apply_reconciliation_handler(
     State(state): State<AppState>,
 ) -> ApiResult<Json<ReconciliationReport>> {
-    let wg = Arc::new(SimulatedWireGuardEngine::new());
-    let net = Arc::new(SimulatedNetworkEngine::new());
+    let wg = Arc::new(NativeLinuxWireGuardEngine::new());
+    let net = Arc::new(NativeLinuxNetworkEngine::new());
     let engine = ReconciliationEngine::new(state, wg, net);
 
     let report = engine.apply().await?;