use anyhow::{Context, Result};
use std::collections::{BTreeMap, HashMap};
use std::sync::{Arc, Mutex, atomic::Ordering, RwLock};
use super::cdn_provider::::{CdnEndpoint, CdnProviderType, EndpointStatus};
use tokio::time::{TokioDuration, TokioInstant};
use std::time::{Duration, Instant};
use std::sync::atomic::Ordering;
#[derive(Debug)]
pub struct GlobalRouter {
edge_nodes: Arc<RwLock<Vec<EdgeNode>>>,
geo_mapping: Arc<RwLock<HashMap<String, String>, routing_rules: Arc<RwLock<Vec<RoutingRule>>>,
}
#[derive(Debug, Clone)]
struct EdgeNode {
id: String,
region: String,
ip: String,
latitude: f64,
longitude: f64,
latency: f64,
capacity: u64,
current_load: f64,
status: NodeStatus,
}
#[derive(Debug, Clone, Copy, PartialEq)]
enum NodeStatus {
Online,
Degraded,
Offline,
}
#[derive(Debug, Clone)]
struct RoutingRule {
pattern: String,
target_region: String,
priority: u8,
}
impl GlobalRouter {
pub fn new() -> Self {
let default_nodes: _ = vec![
EdgeNode {
id: "us-west-1".to_string(),
region: "us-west".to_string(),
ip: "203.0.113.1".to_string(),
latitude: 37.7749,
longitude: -122.4194,
latency: 25.0,
capacity: 100000,
current_load: 0.35,
status: NodeStatus::Online,
},
EdgeNode {
id: "us-east-1".to_string(),
region: "us-east".to_string(),
ip: "203.0.113.2".to_string(),
latitude: 40.7128,
longitude: -74.0060,
latency: 30.0,
capacity: 100000,
current_load: 0.38,
status: NodeStatus::Online,
},
EdgeNode {
id: "eu-west-1".to_string(),
region: "eu-west".to_string(),
ip: "203.0.113.3".to_string(),
latitude: 51.5074,
longitude: -0.1278,
latency: 35.0,
capacity: 100000,
current_load: 0.42,
status: NodeStatus::Online,
},
EdgeNode {
id: "eu-central-1".to_string(),
region: "eu-central".to_string(),
ip: "203.0.113.4".to_string(),
latitude: 50.1109,
longitude: 8.6821,
latency: 40.0,
capacity: 100000,
current_load: 0.41,
status: NodeStatus::Online,
},
EdgeNode {
id: "ap-southeast-1".to_string(),
region: "ap-southeast".to_string(),
ip: "203.0.113.5".to_string(),
latitude: 1.3521,
longitude: 103.8198,
latency: 45.0,
capacity: 100000,
current_load: 0.35,
status: NodeStatus::Online,
},
EdgeNode {
id: "ap-northeast-1".to_string(),
region: "ap-northeast".to_string(),
ip: "203.0.113.6".to_string(),
latitude: 35.6762,
longitude: 139.6503,
latency: 50.0,
capacity: 100000,
current_load: 0.48,
status: NodeStatus::Online,
},
];
GlobalRouter {
edge_nodes: Arc::new(Mutex::new(default_nodes)))
geo_mapping: Arc::new(Mutex::new(HashMap::new()))
routing_rules: Arc::new(Mutex::new(Vec::new()))
}
}
pub async fn add_edge_node(&self, node: EdgeNode) -> Result<()> {
let mut nodes = self.edge_nodes.write().await;
nodes.push(node);
Ok(())
}
pub async fn get_available_routes(&self) -> Result<Vec<String> {
let nodes: _ = self.edge_nodes.read().await;
let routes: Vec<String> = nodes.iter()
.filter(|node| node.status == NodeStatus::Online)
.map(|node| node.region.clone())
.collect();
Ok(routes)
}
pub async fn ping_region(&self, region: &str) -> Result<Duration> {
let nodes: _ = self.edge_nodes.read().await;
let node: _ = nodes.iter().find(|n| n.region == region);
if let Some(node) = node {
let start: _ = Instant::now();
tokio::time::sleep(Duration::from_millis(node.latency as u64)).await;
Ok(start.elapsed())
} else {
Err(anyhow::anyhow!("Region not found: {}", region))
}
}
pub async fn resolve_anycast(&self, domain: &str) -> Result<Vec<String> {
let nodes: _ = self.edge_nodes.read().await;
let online_nodes: Vec<_> = nodes.iter()
.filter(|node| node.status == NodeStatus::Online)
.collect();
let ips: Vec<String> = online_nodes.iter()
.map(|node| node.ip.clone())
.collect();
Ok(ips)
}
pub async fn resolve_geo_aware(&self, domain: &str, client_ip: &str) -> Result<CdnEndpoint> {
let nodes: _ = self.edge_nodes.read().await;
let region: _ = self.guess_region_from_ip(client_ip).await?;
let best_node: _ = nodes.iter()
.filter(|node| node.region == region && node.status == NodeStatus::Online)
.min_by(|a, b| {
let score_a: _ = a.clone();latency + (a.current_load * 100.0);
let score_b: _ = b.clone();latency + (b.current_load * 100.0);
score_a.partial_cmp(&score_b).unwrap_or(std::cmp::Ordering::Equal)
})
.context("No healthy nodes in region")?;
Ok(CdnEndpoint {
id: best_node.id.clone(),
region: best_node.region.clone(),
endpoint_url: format!("https://{}.{}", best_node.id, domain),
latency: best_node.latency,
status: EndpointStatus::Healthy,
capacity: best_node.capacity,
current_load: best_node.current_load,
provider: CdnProviderType::Cloudflare,
})
}
async fn guess_region_from_ip(&self, client_ip: &str) -> Result<String> {
let geo_mapping: _ = self.geo_mapping.read().await;
if let Some(region) = geo_mapping.get(client_ip) {
return Ok(region.clone());
}
let region: _ = match client_ip.chars().next() {
Some('2') => "eu-west",
Some('1') => "us-east",
Some('3') => "ap-southeast",
_ => "us-west",
};
Ok(region.to_string())
}
pub async fn check_node_health(&self) -> Result<HashMap<String, bool> {
let mut nodes = self.edge_nodes.write().await;
let mut health_status = HashMap::new();
for node in nodes.iter_mut() {
let start: _ = Instant::now();
tokio::time::sleep(Duration::from_millis(10)).await; let elapsed: _ = start.elapsed();
if elapsed.as_millis() < 100 && node.current_load < 0.9 {
if node.status != NodeStatus::Online {
node.status = NodeStatus::Online;
}
health_status.insert(node.id.clone(), true);
} else {
node.status = NodeStatus::Degraded;
health_status.insert(node.id.clone(), false);
}
}
Ok(health_status)
}
pub async fn trigger_automatic_failover(&self, failed_node_id: &str) -> Result<String> {
let mut nodes = self.edge_nodes.write().await;
let (failed_node_region, fallback_id) = {
let nodes_ref: _ = &*nodes;
let failed_node: _ = nodes_ref.iter()
.find(|n| n.id == failed_node_id)
.context("Failed node not found")?;
let fallback: _ = nodes_ref.iter()
.filter(|n| n.id != failed_node_id && n.status == NodeStatus::Online)
.min_by(|a, b| {
if a.region == failed_node.region && b.region != failed_node.region {
std::cmp::Ordering::Less
} else if a.region != failed_node.region && b.region == failed_node.region {
std::cmp::Ordering::Greater
} else {
a.current_load.partial_cmp(&b.current_load).unwrap_or(std::cmp::Ordering::Equal)
}
})
.context("No fallback node available")?;
(failed_node.region.clone(), fallback.id.clone())
};
if let Some(node) = nodes.iter_mut().find(|n| n.id == failed_node_id) {
node.status = NodeStatus::Offline;
}
println!("Automatic failover: {} -> {}", failed_node_id, fallback_id);
Ok(fallback_id)
}
pub async fn get_topology(&self) -> Result<NetworkTopology> {
let nodes: _ = self.edge_nodes.read().await;
let topology: _ = NetworkTopology {
total_nodes: nodes.len(),
online_nodes: nodes.iter().filter(|n| n.status == NodeStatus::Online).count(),
degraded_nodes: nodes.iter().filter(|n| n.status == NodeStatus::Degraded).count(),
offline_nodes: nodes.iter().filter(|n| n.status == NodeStatus::Offline).count(),
regions: nodes.iter().map(|n| n.region.clone()).collect(),
};
Ok(topology)
}
pub async fn update_node_load(&self, node_id: &str, load: f64) -> Result<()> {
let mut nodes = self.edge_nodes.write().await;
if let Some(node) = nodes.iter_mut().find(|n| n.id == node_id) {
node.current_load = load;
}
Ok(())
}
pub async fn route_request(&self, client_latitude: f64, client_longitude: f64) -> Result<RouteResult> {
let nodes: _ = self.edge_nodes.read().await;
let healthy_nodes: Vec<_> = nodes.iter()
.filter(|node| node.status == NodeStatus::Online && node.current_load < 0.95)
.collect();
if healthy_nodes.is_empty() {
return Err(anyhow::anyhow!("No healthy edge nodes available"));
}
let mut scored_nodes: Vec<(f64, &EdgeNode)> = healthy_nodes.iter()
.map(|node| {
let distance_score: _ = self.calculate_geographic_distance(
client_latitude,
client_longitude,
node.latitude,
node.longitude,
);
let latency_score: _ = node.latency;
let load_score: _ = node.current_load * 100.0;
let capacity_score: _ = (1.0 - (node.current_load / node.capacity as f64)) * 50.0;
let total_score =
distance_score * 0.4 +
latency_score * 0.35 +
load_score * 0.20 +
capacity_score * 0.05;
(total_score, *node)
})
.collect();
scored_nodes.sort_by(|a, b| a.0.partial_cmp(&b.0).unwrap_or(std::cmp::Ordering::Equal));
let (best_score, best_node) = &scored_nodes[0];
let expected_latency: _ = best_node.latency + (best_node.current_load * 50.0);
Ok(RouteResult {
selected_node_id: best_node.id.clone(),
selected_region: best_node.region.clone(),
endpoint_url: format!("https://{}.edge.amberjs.ai", best_node.id),
expected_latency_ms: expected_latency,
routing_score: *best_score,
distance_km: self.calculate_geographic_distance(
client_latitude,
client_longitude,
best_node.latitude,
best_node.longitude,
),
alternatives: scored_nodes[1..].iter().take(3).map(|(score, node)| {
AlternativeRoute {
node_id: node.id.clone(),
region: node.region.clone(),
routing_score: *score,
expected_latency: node.latency + (node.current_load * 50.0),
}
}).collect(),
})
}
fn calculate_geographic_distance(&self, lat1: f64, lon1: f64, lat2: f64, lon2: f64) -> f64 {
let r: _ = 6371.0; let dlat: _ = (lat2 - lat1).to_radians();
let dlon: _ = (lon2 - lon1).to_radians();
let a: _ = (dlat / 2.0).sin().powi(2) +
lat1.to_radians().cos() *
lat2.to_radians().cos() *
(dlon / 2.0).sin().powi(2);
let c: _ = 2.0 * a.sqrt().atan2((1.0 - a).sqrt());
r * c
}
pub async fn batch_route_requests(
&self,
clients: &[(f64, f64)],
) -> Result<Vec<RouteResult> {
let mut results = Vec::with_capacity(clients.len());
for (lat, lon) in clients {
let route: _ = self.route_request(*lat, *lon).await?;
results.push(route);
}
Ok(results)
}
pub async fn get_routing_stats(&self) -> Result<RoutingStats> {
let nodes: _ = self.edge_nodes.read().await;
let total_capacity: u64 = nodes.iter().map(|n| n.capacity).sum();
let total_load: f64 = nodes.iter().map(|n| n.current_load).sum();
let avg_load: _ = if nodes.is_empty() { 0.0 } else { total_load / nodes.len() as f64 };
let healthy_nodes: _ = nodes.iter().filter(|n| n.status == NodeStatus::Online).count();
let degraded_nodes: _ = nodes.iter().filter(|n| n.status == NodeStatus::Degraded).count();
Ok(RoutingStats {
total_nodes: nodes.len(),
healthy_nodes,
degraded_nodes,
avg_load,
total_capacity,
capacity_utilization: if total_capacity > 0 {
(total_load / total_capacity as f64) * 100.0
} else {
0.0
},
})
}
}
#[derive(Debug, Clone)]
pub struct NetworkTopology {
pub total_nodes: usize,
pub online_nodes: usize,
pub degraded_nodes: usize,
pub offline_nodes: usize,
pub regions: Vec<String>,
}
#[derive(Debug, Clone)]
pub struct RouteResult {
pub selected_node_id: String,
pub selected_region: String,
pub endpoint_url: String,
pub expected_latency_ms: f64,
pub routing_score: f64,
pub distance_km: f64,
pub alternatives: Vec<AlternativeRoute>,
}
#[derive(Debug, Clone)]
pub struct AlternativeRoute {
pub node_id: String,
pub region: String,
pub routing_score: f64,
pub expected_latency: f64,
}
#[derive(Debug, Clone)]
pub struct RoutingStats {
pub total_nodes: usize,
pub healthy_nodes: usize,
pub degraded_nodes: usize,
pub avg_load: f64,
pub total_capacity: u64,
pub capacity_utilization: f64,
}
#[derive(Debug)]
pub struct AnycastDns {
router: GlobalRouter,
}
impl AnycastDns {
pub fn new() -> Self {
AnycastDns {
router: GlobalRouter::new(),
}
}
pub async fn resolve(&self, domain: &str) -> Result<Vec<String> {
self.router.resolve_anycast(domain).await
}
}
#[derive(Debug)]
pub struct GeoDns {
router: GlobalRouter,
}
impl GeoDns {
pub fn new() -> Self {
GeoDns {
router: GlobalRouter::new(),
}
}
pub async fn resolve_with_region(&self, domain: &str, client_ip: &str) -> Result<CdnEndpoint> {
self.router.resolve_geo_aware(domain, client_ip).await
}
}
#[cfg(test)]
mod tests {
#[tokio::test]
async fn test_global_router_initialization() {
let router: _ = GlobalRouter::new();
let is_initialized: _ = router.get_available_routes().await.is_ok();
assert!(is_initialized);
}
#[tokio::test]
async fn test_add_edge_node() {
let router: _ = GlobalRouter::new();
let node: _ = EdgeNode {
id: "test-node".to_string(),
region: "test-region".to_string(),
ip: "192.0.2.1".to_string(),
latitude: 0.0,
longitude: 0.0,
latency: 100.0,
capacity: 1000,
current_load: 0.0,
status: NodeStatus::Online,
};
let result: _ = router.add_edge_node(node).await;
assert!(result.is_ok());
}
#[tokio::test]
async fn test_anycast_dns() {
let anycast: _ = AnycastDns::new();
let ips: _ = anycast.resolve("amberjs-edge.com").await;
assert!(ips.is_ok());
let result: _ = ips.unwrap();
assert!(!result.is_empty());
}
#[tokio::test]
async fn test_geo_dns() {
let geo_dns: _ = GeoDns::new();
let endpoint: _ = geo_dns.resolve_with_region("amberjs-edge.com", "203.0.113.1").await;
assert!(endpoint.is_ok());
let result: _ = endpoint.unwrap();
assert!(!result.region.is_empty());
}
#[tokio::test]
async fn test_ping_region() {
let router: _ = GlobalRouter::new();
let latency: _ = router.ping_region("us-west").await;
assert!(latency.is_ok());
let duration: _ = latency.unwrap();
assert!(duration.as_millis() > 0);
}
#[tokio::test]
async fn test_node_health_check() {
let router: _ = GlobalRouter::new();
let health: _ = router.check_node_health().await;
assert!(health.is_ok());
let status: _ = health.unwrap();
assert!(!status.is_empty());
}
#[tokio::test]
async fn test_automatic_failover() {
let router: _ = GlobalRouter::new();
router.update_node_load("us-west-1", 0.5).await.unwrap();
let fallback: _ = router.trigger_automatic_failover("nonexistent-node").await;
}
#[tokio::test]
async fn test_network_topology() {
let router: _ = GlobalRouter::new();
let topology: _ = router.get_topology().await;
assert!(topology.is_ok());
let topo: _ = topology.unwrap();
assert!(topo.total_nodes > 0);
}
}