Refactor initialization into modular submodules

Split initialization.rs into multiple focused modules under src/initialization, improving maintainability and clarity. Updated main.rs to use new initialization API and removed direct Rocket server construction. Cargo.toml dependency section was reorganized for clarity, grouping dependencies by category.

Refactor server initialization and modularize API

Split main.rs into modular components for CORS, endpoints, logging, initialization, and server setup. Added new modules: api_models, cors, endpoints, initialization, logging, and server. Refactored cluster management and peer discovery logic, improved documentation, and updated dependency versions in Cargo.toml. Replaced scale..rs with scale.rs and updated control API module structure. Enhanced SharedState struct documentation and usage.

Refactor apps API into modular route files

Split the monolithic apps.rs file into multiple focused modules for CRUD, control, listing, release, and instance management. Updated mod.rs and main API routing to use the new modular structure for improved maintainability and clarity.

Refactor alerts API into modular route files

Split the monolithic alerts.rs file into multiple focused modules: actions, app_alerts, auto_resolve, bulk, create, escalation, get, list, org_alerts, search, types, and update. This improves maintainability and clarity by grouping related API endpoints and types into separate files. Also applied similar modularization to audit_log.rs. No functional changes to endpoint logic.

Refactor cost API module and update alerts imports

Removed the monolithic cost.rs API module and split its functionality into multiple files for allocation tags, analysis, budgets, metrics, pricing, projections, resource types, and types. Updated all alert API modules to use the new import path for db::queries, reflecting the deeper module nesting.

Refactor API schemas into modular route files

Split monolithic API route files into modular submodules for builds, deployments, instances, metrics, notifications, and permissions. Each resource now has dedicated files for CRUD operations and route registration, improving maintainability and clarity of the codebase.

Author: tristanpoland

Cargo.toml

@@ -3,45 +3,50 @@ name = "omni-orchestrator"
 version = "0.1.4"
 edition = "2021"
 
+
 [dependencies]
-rocket = { version = "0.5.1", features = ["json", "uuid"] }
-uuid = { version = "1.17.0", features = ["v4", "serde"] }
-tokio = { version = "1.45.1", features = ["full"] }
-serde = { version = "1.0.219", features = ["derive", "rc"] }
-serde_json5 = "0.2.1"
-serde_json = "1.0.140"
+# Core async/runtime
+tokio = { version = "1.45", features = ["full"] }
 async-trait = "0.1.88"
-reqwest = { version = "0.12.18", features = ["json", "native-tls-vendored", "rustls-tls"] }
-log = "0.4.27"
-env_logger = "0.11.8"
-thiserror = "2.0.12"
-parking_lot = "0.12.4"
-chrono = { version = "0.4.41", features = ["serde"] }
 futures = "0.3.31"
-colored = "3.0.0"   
-anyhow = "1.0.98"
-lazy_static = "1.5.0"
+
+# Web framework & API
+rocket = { version = "0.5.1", features = ["json", "uuid"] }
 rocket-multipart-form-data = "0.10.7"
+reqwest = { version = "0.12", features = ["json", "native-tls-vendored", "rustls-tls"] }
+
+# Serialization
+serde = { version = "1.0", features = ["derive", "rc"] }
+serde_json = "1.0"
+serde_json5 = "0.2.1"
+
+# Database
+mysql = "26.0.0"
+sqlx = { version = "0.8.6", features = ["runtime-tokio-rustls", "mysql", "chrono", "uuid"] }
+clickhouse = { version = "0.13.3"}
+libomni = { path = "../LibOmni" }
+lighthouse = { git = "https://github.com/OmniCloudOrg/Lighthouse", features = ["metrics-persistence", "predictive-scaling"], rev = "de72b5365bdc1a28e89ffbf06b3c92957ad3403d"}
+
+# Crypto & Security
 sha2 = "0.10.9"
 rand = "0.9.1"
 hex = "0.4.3"
-mysql = "26.0.0"
+jsonwebtoken = "9.3.1"
+
+# Utilities
+anyhow = "1.0.98"
+thiserror = "2.0.12"
+colored = "3.0.0"
+lazy_static = "1.5.0"
 once_cell = "1.21.3"
-sqlx = { version = "0.8.6", features = ["runtime-tokio-rustls", "mysql", "chrono", "uuid"] }
+dotenv = { version = "0.15.0" }
+parking_lot = "0.12.4"
+chrono = { version = "0.4.41", features = ["serde"] }
+uuid = { version = "1.17.0", features = ["v4", "serde"] }
 tracing = { version = "0.1.41", features = ["log"] }
+env_logger = "0.11.8"
+log = "0.4.27"
 hadris-iso = "0.0.2"
 ssh2 = { version = "0.9.5" }
-libomni = { path = "../LibOmni" }
-dotenv = { version = "0.15.0"}
 paxakos = "0.13.0"
-jsonwebtoken = "9.3.1"
 rust_decimal = "1.37.1"
-clickhouse = { version = "0.13.3"}
-
-[profile.dev]
-codegen-units = 32
-incremental = true
-debug-assertions = false
-overflow-checks = false
-opt-level = 0
-lto = "thin"

src/api_models.rs

@@ -0,0 +1,20 @@
+//! API models for the OmniOrchestrator
+//! 
+//! These are the general API models used by the OmniOrchestrator
+//! service outside the primary platform routes located in
+//! /scr/schema/VERSION/api
+
+use serde::{Deserialize, Serialize};
+use crate::cluster::NodeInfo;
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct ClusterStatusMessage {
+    pub node_roles: String,
+    pub cluster_nodes: Vec<NodeInfo>,
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct ApiResponse {
+    pub status: String,
+    pub message: ClusterStatusMessage,
+}

src/cluster.rs

@@ -1,10 +1,14 @@
+use anyhow::{anyhow, Result};
 use colored::Colorize;
 use log::debug;
+use reqwest::Client;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::RwLock;
 
+use crate::config::ServerConfig;
 use crate::state::SharedState;
 
 /// Represents a node in the OmniOrchestrator cluster.
@@ -239,4 +243,70 @@ impl ClusterManager {
         let nodes = self.nodes.read().await;
         nodes.contains_key(&node_uid)
     }
+
+    pub async fn discover_peers(&self, config: &ServerConfig, my_port: u16) -> Result<()> {
+        let client = Client::new();
+        log::info!("{}", "Starting peer discovery...".cyan());
+
+        for instance in &config.instances {
+            let string = format!("{:#?}", instance);
+            log::info!("{}", format!("Discovered: {}", string).blue().bold());
+            if instance.port == my_port {
+                log::debug!("Skipping self-connection at port {}", my_port);
+                continue;
+            }
+
+            let node_address: Arc<str> = format!("{}:{}", instance.address, instance.port).into();
+            let node_uri = format!("{}", node_address);
+
+            match self.connect_to_peer(&client, &node_uri.clone()).await {
+                Ok(_) => log::info!(
+                    "{}",
+                    format!("Successfully connected to peer: {}", node_uri).green()
+                ),
+                Err(e) => {
+                    log::warn!(
+                        "{}",
+                        format!("Failed to connect to peer: {} {}", node_uri, e).yellow()
+                    );
+                    self.remove_node(node_uri.into()).await;
+                }
+            }
+        }
+
+        log::info!("{}", "Peer discovery completed".cyan());
+        Ok(())
+    }
+
+    async fn connect_to_peer(&self, client: &Client, node_address: &str) -> Result<()> {
+        let health_url = format!("{}/health", node_address);
+        log::debug!("Checking health at: {}", health_url);
+
+        let response = client
+            .get(&health_url)
+            .timeout(Duration::from_secs(5))
+            .send()
+            .await?;
+
+        if response.status().is_success() {
+            let port = node_address
+                .split(':')
+                .next_back()
+                .unwrap_or("80")
+                .parse::<u16>()
+                .unwrap_or(80);
+
+            let node_info = NodeInfo {
+                id: node_address.into(),
+                address: node_address.into(),
+                port,
+            };
+
+            self.register_node(node_info).await;
+            log::debug!("Node registered: {}", node_address);
+            Ok(())
+        } else {
+            Err(anyhow!("Node health check failed"))
+        }
+    }
 }

src/cors.rs

@@ -0,0 +1,32 @@
+use rocket::{Request, Response, fairing::{Fairing, Info, Kind}, http::Header};
+
+pub struct CORS;
+
+#[rocket::async_trait]
+impl Fairing for CORS {
+    fn info(&self) -> Info {
+        Info {
+            name: "Add comprehensive CORS headers to responses",
+            kind: Kind::Response,
+        }
+    }
+
+    async fn on_response<'r>(&self, _request: &'r Request<'_>, response: &mut Response<'r>) {
+        response.set_header(Header::new("Access-Control-Allow-Origin", "*"));
+        response.set_header(Header::new(
+            "Access-Control-Allow-Methods",
+            "GET, POST, PUT, PATCH, DELETE, OPTIONS, HEAD",
+        ));
+        response.set_header(Header::new(
+            "Access-Control-Allow-Headers",
+            "Authorization, Content-Type, Accept, Origin, X-Requested-With",
+        ));
+        response.set_header(Header::new("Access-Control-Allow-Credentials", "true"));
+        response.set_header(Header::new("Access-Control-Max-Age", "86400"));
+    }
+}
+
+#[options("/<_..>")]
+pub fn cors_preflight() -> &'static str {
+    ""
+}

src/db_manager/manager.rs

@@ -53,8 +53,10 @@ impl DatabaseManager {
             .platform_pool(platform_id, &platform_name)
             .await?;
 
-        // TODO: @Caznix @tristanpoland We need to find a new home for this
-        // Initialize the schema
+        // TODO: Platform schema initialization needs to be relocated
+        // Currently commented out pending architectural decisions about where
+        // platform-specific schema initialization should be handled.
+        // Consider moving to a dedicated platform management service.
         // MigrationManager::initialize_platform_schema(&pool, platform).await?;
 
         Ok(pool)

src/endpoints.rs

@@ -0,0 +1,79 @@
+//! Core API endpoints for the OmniOrchestrator server.
+//!
+//! This module provides essential HTTP endpoints for monitoring and managing
+//! the OmniOrchestrator cluster. These endpoints serve as the primary interface
+//! for external systems to query the health and status of the cluster.
+
+use rocket;
+use std::sync::Arc;
+use tokio::sync::RwLock;
+use crate::cluster::ClusterManager;
+use crate::state::SharedState;
+use crate::api_models::{ApiResponse, ClusterStatusMessage};
+
+/// Health check endpoint that provides basic service availability status.
+///
+/// This endpoint is used by load balancers, monitoring systems, and other external
+/// services to determine if the OmniOrchestrator service is running and responding
+/// to requests. It returns a simple JSON response indicating service availability.
+///
+/// # Returns
+///
+/// A JSON response with status "ok" and basic cluster information.
+#[get("/health")]
+pub async fn health_check() -> rocket::serde::json::Json<ApiResponse> {
+    log::debug!("Health check endpoint called");
+    rocket::serde::json::Json(ApiResponse {
+        status: "ok".to_string(),
+        message: ClusterStatusMessage {
+            node_roles: "unknown".to_string(),
+            cluster_nodes: vec![],
+        },
+    })
+}
+
+/// Provides detailed cluster status information including node roles and membership.
+///
+/// This endpoint returns comprehensive information about the current state of the
+/// OmniOrchestrator cluster, including which node is the leader, cluster membership,
+/// and the role of the current node. This information is crucial for cluster
+/// monitoring and debugging distributed system issues.
+///
+/// # Arguments
+///
+/// * `state` - Shared state containing node role and cluster information
+/// * `cluster` - Cluster manager with information about all known nodes
+///
+/// # Returns
+///
+/// A JSON response containing:
+/// - Overall cluster status
+/// - Current node's role (leader/follower)
+/// - List of all known cluster nodes
+#[get("/cluster/status")]
+pub async fn cluster_status(
+    state: &rocket::State<Arc<RwLock<SharedState>>>,
+    cluster: &rocket::State<Arc<RwLock<ClusterManager>>>,
+) -> rocket::serde::json::Json<ApiResponse> {
+    log::debug!("Cluster status endpoint called");
+    let state = state.read().await;
+    let nodes = cluster.read().await;
+
+    let role = if state.is_leader {
+        "leader".to_string()
+    } else {
+        "follower".to_string()
+    };
+
+    log::info!("{}", format!("Current node role: {}", role));
+
+    let response = ApiResponse {
+        status: "ok".to_string(),
+        message: ClusterStatusMessage {
+            node_roles: role,
+            cluster_nodes: nodes.get_nodes().await,
+        },
+    };
+
+    rocket::serde::json::Json(response)
+}

src/initialization/create_auth_config.rs

@@ -0,0 +1,19 @@
+use libomni::types::db::auth::AuthConfig;
+
+/// Constructs the authentication configuration from environment variables.
+///
+/// - Loads the JWT secret and token expiry hours from environment variables.
+/// - Panics if `JWT_SECRET` is not set or if `TOKEN_EXPIRY_HOURS` is invalid.
+///
+/// # Returns
+/// Returns an `AuthConfig` struct with the loaded values.
+pub fn create_auth_config() -> AuthConfig {
+    AuthConfig {
+        jwt_secret: std::env::var("JWT_SECRET")
+            .expect("Environment variable JWT_SECRET must be set for secure operation."),
+        token_expiry_hours: std::env::var("TOKEN_EXPIRY_HOURS")
+            .unwrap_or_else(|_| "24".to_string())
+            .parse()
+            .expect("Invalid value for TOKEN_EXPIRY_HOURS"),
+    }
+}

src/initialization/launch_server.rs

@@ -0,0 +1,43 @@
+use crate::server::build_rocket;
+// use crate::{CLUSTER_MANAGER}; // removed unused import
+use crate::db_manager::DatabaseManager;
+use crate::state::SharedState;
+// use libomni::types::db::auth::AuthConfig; // removed unused import
+use std::sync::Arc;
+use tokio::sync::RwLock;
+use colored::Colorize;
+
+/// Builds and launches the Rocket server with the provided configuration and dependencies.
+///
+/// # Arguments
+/// * `port` - The port to bind the server to.
+/// * `db_manager` - Shared database manager instance.
+/// * `pool` - Main database pool.
+/// * `cluster_manager` - Shared cluster manager instance.
+/// * `clickhouse_client` - ClickHouse client instance.
+/// * `shared_state_for_server` - Shared state for the server.
+///
+/// # Errors
+/// Returns an error if the Rocket server fails to launch.
+pub async fn launch_server(
+    port: u16,
+    db_manager: Arc<DatabaseManager>,
+    pool: sqlx::MySqlPool,
+    cluster_manager: Arc<RwLock<crate::cluster::ClusterManager>>,
+    clickhouse_client: clickhouse::Client,
+    shared_state_for_server: Arc<RwLock<SharedState>>,
+) -> Result<(), Box<dyn std::error::Error>> {
+    let auth_config = super::create_auth_config();
+    let rocket_with_routes = build_rocket(
+        port,
+        db_manager,
+        pool,
+        cluster_manager,
+        clickhouse_client,
+        shared_state_for_server,
+        auth_config,
+    );
+    log::info!("{}", "🚀 LAUNCHING SERVER...".bright_cyan().bold());
+    rocket_with_routes.launch().await?;
+    Ok(())
+}