1use chrono::Utc;
14use futures::stream::{FuturesUnordered, StreamExt};
15use reqwest::Client;
16use serde::{Deserialize, Serialize};
17use std::net::IpAddr;
18use std::sync::Arc;
19use std::time::{Duration, Instant};
20use tokio::sync::{Mutex, RwLock};
21
22use crate::events::{emit, EventSender, Phase, SelectedServer, TestEvent};
23use crate::types::{
24 ConnectionQuality, ServerCapabilities, ServerProvider, SpeedTestResult, TestConfig, TestServer,
25};
26
27const PARALLEL_CONNECTIONS: usize = 50;
28const SERVER_SELECTION_COUNT: usize = 3;
29
30#[derive(Debug, Clone, Serialize, Deserialize)]
31pub struct GeoLocation {
32 pub country: String,
33 pub city: String,
34 pub latitude: f64,
35 pub longitude: f64,
36 pub isp: Option<String>,
37}
38
39#[derive(Debug, Clone)]
40#[allow(dead_code)]
41pub struct ServerPerformance {
42 pub server: TestServer,
43 pub latency_ms: f64,
44 pub jitter_ms: f64,
45 pub packet_loss: f64,
46 pub download_score: f64,
47 pub upload_score: f64,
48 pub overall_score: f64,
49}
50
51pub struct SpeedTest {
52 config: TestConfig,
53 client: Client,
54 events: Option<EventSender>,
55 geo_location: Arc<RwLock<Option<GeoLocation>>>,
56 server_pool: Arc<RwLock<Vec<TestServer>>>,
57}
58
59impl SpeedTest {
60 pub fn new(config: TestConfig) -> Result<Self, Box<dyn std::error::Error>> {
61 Self::with_events(config, None)
62 }
63
64 pub fn with_events(
69 config: TestConfig,
70 events: Option<EventSender>,
71 ) -> Result<Self, Box<dyn std::error::Error>> {
72 let client = Client::builder()
73 .timeout(Duration::from_secs(30))
74 .pool_max_idle_per_host(100)
75 .pool_idle_timeout(Duration::from_secs(120))
76 .tcp_keepalive(Duration::from_secs(10))
77 .http2_keep_alive_interval(Duration::from_secs(10))
78 .http2_adaptive_window(true)
79 .http2_initial_stream_window_size(1024 * 1024) .http2_initial_connection_window_size(2 * 1024 * 1024) .danger_accept_invalid_certs(false)
82 .build()?;
83
84 Ok(Self {
85 config,
86 client,
87 events,
88 geo_location: Arc::new(RwLock::new(None)),
89 server_pool: Arc::new(RwLock::new(Vec::new())),
90 })
91 }
92
93 pub fn set_events(&mut self, events: Option<EventSender>) {
95 self.events = events;
96 }
97
98 pub async fn run_full_test(&self) -> Result<SpeedTestResult, Box<dyn std::error::Error>> {
100 let start = Instant::now();
101
102 emit(&self.events, TestEvent::PhaseStarted(Phase::Locating));
104 let geo = self.detect_location().await?;
105 *self.geo_location.write().await = Some(geo.clone());
106
107 emit(
109 &self.events,
110 TestEvent::PhaseStarted(Phase::BuildingServers),
111 );
112 self.build_server_pool(&geo).await?;
113
114 emit(
116 &self.events,
117 TestEvent::PhaseStarted(Phase::SelectingServers),
118 );
119 let best_servers = self.select_best_servers().await?;
120
121 emit(
122 &self.events,
123 TestEvent::PrimarySelected {
124 name: best_servers[0].name.clone(),
125 location: best_servers[0].location.clone(),
126 distance_km: best_servers[0].distance_km.unwrap_or(0.0),
127 },
128 );
129
130 let ping_ms = self.measure_latency(&best_servers[0]).await?;
132
133 let download_mbps = self.progressive_download_test(&best_servers).await?;
135
136 let upload_mbps = self.progressive_upload_test(&best_servers).await?;
138
139 emit(&self.events, TestEvent::PhaseStarted(Phase::Jitter));
141 let (jitter_ms, packet_loss) = self.measure_jitter_and_loss(&best_servers[0]).await?;
142 emit(
143 &self.events,
144 TestEvent::JitterComplete {
145 jitter_ms,
146 packet_loss_percent: packet_loss,
147 },
148 );
149
150 let quality = ConnectionQuality::from_speed_and_ping(download_mbps, upload_mbps, ping_ms);
151 let test_duration = start.elapsed().as_secs_f64();
152
153 let result = SpeedTestResult {
154 timestamp: Utc::now(),
155 download_mbps,
156 upload_mbps,
157 ping_ms,
158 jitter_ms,
159 packet_loss_percent: packet_loss,
160 server_location: best_servers[0].location.clone(),
161 server_ip: self.resolve_server_ip(&best_servers[0].url).await,
162 client_ip: self.get_client_ip().await,
163 quality,
164 test_duration_seconds: test_duration,
165 isp: geo.isp.clone(),
166 };
167
168 emit(&self.events, TestEvent::Completed(Box::new(result.clone())));
169
170 Ok(result)
171 }
172
173 async fn detect_location(&self) -> Result<GeoLocation, Box<dyn std::error::Error>> {
175 emit(
176 &self.events,
177 TestEvent::Status("Detecting your location...".to_string()),
178 );
179
180 match self.try_ipapi_co().await {
183 Ok(geo) => {
184 self.emit_location(&geo, "ipapi.co");
185 return Ok(geo);
186 }
187 Err(e) => {
188 if std::env::var("NETRUNNER_DEBUG").is_ok() {
190 eprintln!("[TRACE] ipapi.co geolocation failed: {}", e);
191 }
192 }
193 }
194
195 match self.try_ip_api_com().await {
197 Ok(geo) => {
198 self.emit_location(&geo, "ip-api.com");
199 return Ok(geo);
200 }
201 Err(e) => {
202 if std::env::var("NETRUNNER_DEBUG").is_ok() {
204 eprintln!("[TRACE] ip-api.com geolocation failed: {}", e);
205 }
206 }
207 }
208
209 match self.try_ipinfo_io().await {
211 Ok(geo) => {
212 self.emit_location(&geo, "ipinfo.io");
213 return Ok(geo);
214 }
215 Err(e) => {
216 if std::env::var("NETRUNNER_DEBUG").is_ok() {
218 eprintln!("[TRACE] ipinfo.io geolocation failed: {}", e);
219 }
220 }
221 }
222
223 match self.try_freegeoip_app().await {
225 Ok(geo) => {
226 self.emit_location(&geo, "freegeoip.app");
227 return Ok(geo);
228 }
229 Err(e) => {
230 if std::env::var("NETRUNNER_DEBUG").is_ok() {
232 eprintln!("[TRACE] freegeoip.app geolocation failed: {}", e);
233 }
234 }
235 }
236
237 match self.try_ipwhois_app().await {
239 Ok(geo) => {
240 self.emit_location(&geo, "ipwhois.app");
241 return Ok(geo);
242 }
243 Err(e) => {
244 if std::env::var("NETRUNNER_DEBUG").is_ok() {
246 eprintln!("[TRACE] ipwhois.app geolocation failed: {}", e);
247 }
248 }
249 }
250
251 emit(
253 &self.events,
254 TestEvent::Status(
255 "Using default location (USA Central) - all geolocation services failed"
256 .to_string(),
257 ),
258 );
259
260 Ok(GeoLocation {
261 country: "United States".to_string(),
262 city: "Kansas City".to_string(),
263 latitude: 39.0997,
264 longitude: -94.5786,
265 isp: None,
266 })
267 }
268
269 fn emit_location(&self, geo: &GeoLocation, source: &str) {
271 emit(
272 &self.events,
273 TestEvent::LocationDetected {
274 city: geo.city.clone(),
275 country: geo.country.clone(),
276 isp: geo.isp.clone(),
277 source: source.to_string(),
278 },
279 );
280 }
281
282 async fn try_ipapi_co(&self) -> Result<GeoLocation, Box<dyn std::error::Error>> {
283 let response = self
284 .client
285 .get("https://ipapi.co/json/")
286 .timeout(Duration::from_secs(5))
287 .send()
288 .await?;
289
290 if !response.status().is_success() {
291 return Err(format!("HTTP error: {}", response.status()).into());
292 }
293
294 let json: serde_json::Value = response.json().await?;
295
296 if json.get("error").is_some() {
298 return Err(format!(
299 "API error: {}",
300 json["reason"].as_str().unwrap_or("Unknown")
301 )
302 .into());
303 }
304
305 let country = json["country_name"]
306 .as_str()
307 .filter(|s| !s.is_empty() && *s != "Unknown")
308 .ok_or("Invalid country")?
309 .to_string();
310
311 let city = json["city"]
312 .as_str()
313 .filter(|s| !s.is_empty() && *s != "Unknown")
314 .ok_or("Invalid city")?
315 .to_string();
316
317 let latitude = json["latitude"].as_f64().ok_or("Invalid latitude")?;
318 let longitude = json["longitude"].as_f64().ok_or("Invalid longitude")?;
319
320 if latitude == 0.0 && longitude == 0.0 {
321 return Err("Invalid coordinates".into());
322 }
323
324 Ok(GeoLocation {
325 country,
326 city,
327 latitude,
328 longitude,
329 isp: json["org"].as_str().map(String::from),
330 })
331 }
332
333 async fn try_ip_api_com(&self) -> Result<GeoLocation, Box<dyn std::error::Error>> {
334 let response = self
335 .client
336 .get("http://ip-api.com/json/?fields=status,message,country,city,lat,lon,isp")
337 .timeout(Duration::from_secs(5))
338 .send()
339 .await?;
340
341 if !response.status().is_success() {
342 return Err(format!("HTTP error: {}", response.status()).into());
343 }
344
345 let json: serde_json::Value = response.json().await?;
346
347 if json["status"].as_str() != Some("success") {
349 return Err(format!(
350 "API error: {}",
351 json["message"].as_str().unwrap_or("Unknown")
352 )
353 .into());
354 }
355
356 let country = json["country"]
357 .as_str()
358 .filter(|s| !s.is_empty() && *s != "Unknown")
359 .ok_or("Invalid country")?
360 .to_string();
361
362 let city = json["city"]
363 .as_str()
364 .filter(|s| !s.is_empty() && *s != "Unknown")
365 .ok_or("Invalid city")?
366 .to_string();
367
368 let latitude = json["lat"].as_f64().ok_or("Invalid latitude")?;
369 let longitude = json["lon"].as_f64().ok_or("Invalid longitude")?;
370
371 if latitude == 0.0 && longitude == 0.0 {
372 return Err("Invalid coordinates".into());
373 }
374
375 Ok(GeoLocation {
376 country,
377 city,
378 latitude,
379 longitude,
380 isp: json["isp"].as_str().map(String::from),
381 })
382 }
383
384 async fn try_ipinfo_io(&self) -> Result<GeoLocation, Box<dyn std::error::Error>> {
385 let response = self
386 .client
387 .get("https://ipinfo.io/json")
388 .timeout(Duration::from_secs(5))
389 .send()
390 .await?;
391
392 if !response.status().is_success() {
393 return Err(format!("HTTP error: {}", response.status()).into());
394 }
395
396 let json: serde_json::Value = response.json().await?;
397
398 let country = json["country"]
399 .as_str()
400 .filter(|s| !s.is_empty())
401 .ok_or("Invalid country")?
402 .to_string();
403
404 let city = json["city"]
405 .as_str()
406 .filter(|s| !s.is_empty())
407 .ok_or("Invalid city")?
408 .to_string();
409
410 let loc = json["loc"].as_str().ok_or("Invalid location")?;
412 let coords: Vec<&str> = loc.split(',').collect();
413 if coords.len() != 2 {
414 return Err("Invalid coordinates format".into());
415 }
416
417 let latitude: f64 = coords[0].parse().map_err(|_| "Invalid latitude")?;
418 let longitude: f64 = coords[1].parse().map_err(|_| "Invalid longitude")?;
419
420 if latitude == 0.0 && longitude == 0.0 {
421 return Err("Invalid coordinates".into());
422 }
423
424 Ok(GeoLocation {
425 country,
426 city,
427 latitude,
428 longitude,
429 isp: json["org"].as_str().map(String::from),
430 })
431 }
432
433 async fn try_freegeoip_app(&self) -> Result<GeoLocation, Box<dyn std::error::Error>> {
434 let response = self
435 .client
436 .get("https://freegeoip.app/json/")
437 .timeout(Duration::from_secs(5))
438 .send()
439 .await?;
440
441 if !response.status().is_success() {
442 return Err(format!("HTTP error: {}", response.status()).into());
443 }
444
445 let json: serde_json::Value = response.json().await?;
446
447 let country = json["country_name"]
448 .as_str()
449 .filter(|s| !s.is_empty())
450 .ok_or("Invalid country")?
451 .to_string();
452
453 let city = json["city"]
454 .as_str()
455 .filter(|s| !s.is_empty())
456 .ok_or("Invalid city")?
457 .to_string();
458
459 let latitude = json["latitude"].as_f64().ok_or("Invalid latitude")?;
460 let longitude = json["longitude"].as_f64().ok_or("Invalid longitude")?;
461
462 if latitude == 0.0 && longitude == 0.0 {
463 return Err("Invalid coordinates".into());
464 }
465
466 Ok(GeoLocation {
467 country,
468 city,
469 latitude,
470 longitude,
471 isp: None,
472 })
473 }
474
475 async fn try_ipwhois_app(&self) -> Result<GeoLocation, Box<dyn std::error::Error>> {
476 let response = self
477 .client
478 .get("https://ipwho.is/")
479 .timeout(Duration::from_secs(5))
480 .send()
481 .await?;
482
483 if !response.status().is_success() {
484 return Err(format!("HTTP error: {}", response.status()).into());
485 }
486
487 let json: serde_json::Value = response.json().await?;
488
489 if !json["success"].as_bool().unwrap_or(false) {
490 return Err(format!(
491 "API error: {}",
492 json["message"].as_str().unwrap_or("Unknown")
493 )
494 .into());
495 }
496
497 let country = json["country"]
498 .as_str()
499 .filter(|s| !s.is_empty())
500 .ok_or("Invalid country")?
501 .to_string();
502
503 let city = json["city"]
504 .as_str()
505 .filter(|s| !s.is_empty())
506 .ok_or("Invalid city")?
507 .to_string();
508
509 let latitude = json["latitude"].as_f64().ok_or("Invalid latitude")?;
510 let longitude = json["longitude"].as_f64().ok_or("Invalid longitude")?;
511
512 if latitude == 0.0 && longitude == 0.0 {
513 return Err("Invalid coordinates".into());
514 }
515
516 Ok(GeoLocation {
517 country,
518 city,
519 latitude,
520 longitude,
521 isp: json["connection"]["isp"].as_str().map(String::from),
522 })
523 }
524
525 async fn build_server_pool(&self, geo: &GeoLocation) -> Result<(), Box<dyn std::error::Error>> {
527 emit(
528 &self.events,
529 TestEvent::Status("Building server pool...".to_string()),
530 );
531
532 let mut servers = Vec::new();
533
534 servers.extend(self.discover_nearby_servers(geo).await);
536
537 servers.extend(self.get_global_cdn_servers());
539
540 for server in &mut servers {
542 if server.distance_km.is_none() {
543 server.distance_km = Some(self.estimate_distance(geo, server));
544 }
545 }
546
547 servers.sort_by(|a, b| {
549 a.distance_km
550 .unwrap_or(f64::MAX)
551 .partial_cmp(&b.distance_km.unwrap_or(f64::MAX))
552 .unwrap_or(std::cmp::Ordering::Equal)
553 });
554
555 servers.truncate(20);
557
558 let server_count = servers.len();
559 *self.server_pool.write().await = servers;
560
561 emit(
562 &self.events,
563 TestEvent::ServerPoolBuilt {
564 count: server_count,
565 },
566 );
567
568 Ok(())
569 }
570
571 fn get_global_cdn_servers(&self) -> Vec<TestServer> {
572 vec![
574 TestServer {
575 name: "Cloudflare Global".to_string(),
576 url: "https://speed.cloudflare.com".to_string(),
577 location: "Global CDN".to_string(),
578 distance_km: Some(5000.0), latency_ms: None,
580 provider: ServerProvider::Cloudflare,
581 capabilities: ServerCapabilities {
582 supports_download: true,
583 supports_upload: true,
584 supports_latency: true,
585 max_test_size_mb: 2000,
586 geographic_weight: 0.5, },
588 quality_score: None,
589 country_code: None,
590 city: None,
591 is_backup: true,
592 },
593 TestServer {
594 name: "Google Global".to_string(),
595 url: "https://www.google.com".to_string(),
596 location: "Global CDN".to_string(),
597 distance_km: Some(5000.0),
598 latency_ms: None,
599 provider: ServerProvider::Google,
600 capabilities: ServerCapabilities {
601 supports_download: true,
602 supports_upload: false,
603 supports_latency: true,
604 max_test_size_mb: 100,
605 geographic_weight: 0.4,
606 },
607 quality_score: None,
608 country_code: None,
609 city: None,
610 is_backup: true,
611 },
612 ]
613 }
614
615 async fn discover_nearby_servers(&self, geo: &GeoLocation) -> Vec<TestServer> {
617 let mut servers = Vec::new();
618
619 emit(
620 &self.events,
621 TestEvent::Status("Discovering nearby speed test servers...".to_string()),
622 );
623
624 if let Ok(speedtest_servers) = self.fetch_speedtest_net_servers(geo).await {
626 servers.extend(speedtest_servers);
627 }
628
629 servers.extend(self.get_continent_servers(geo));
631
632 servers.extend(self.get_country_servers(geo));
634
635 emit(
636 &self.events,
637 TestEvent::NearbyServersFound {
638 count: servers.len(),
639 },
640 );
641
642 servers
643 }
644
645 async fn fetch_speedtest_net_servers(
647 &self,
648 geo: &GeoLocation,
649 ) -> Result<Vec<TestServer>, Box<dyn std::error::Error>> {
650 let url = "https://www.speedtest.net/api/js/servers?engine=js&limit=10";
652
653 if let Ok(response) = self.client.get(url).send().await {
654 if let Ok(text) = response.text().await {
655 if let Ok(servers) = self.parse_speedtest_servers(&text, geo) {
657 return Ok(servers);
658 }
659 }
660 }
661
662 self.get_open_speedtest_servers(geo).await
664 }
665
666 fn parse_speedtest_servers(
667 &self,
668 json: &str,
669 geo: &GeoLocation,
670 ) -> Result<Vec<TestServer>, Box<dyn std::error::Error>> {
671 let mut servers = Vec::new();
675
676 if let Ok(parsed) = serde_json::from_str::<serde_json::Value>(json) {
678 if let Some(array) = parsed.as_array() {
679 for server in array.iter().take(10) {
680 if let (Some(host), Some(name), Some(country), Some(lat), Some(lon)) = (
681 server.get("host").and_then(|v| v.as_str()),
682 server.get("name").and_then(|v| v.as_str()),
683 server.get("country").and_then(|v| v.as_str()),
684 server.get("lat").and_then(|v| v.as_f64()),
685 server.get("lon").and_then(|v| v.as_f64()),
686 ) {
687 let distance =
688 self.calculate_distance(geo.latitude, geo.longitude, lat, lon);
689
690 servers.push(TestServer {
691 name: format!("{}, {}", name, country),
692 url: format!("https://{}", host),
693 location: format!("{}, {}", name, country),
694 distance_km: Some(distance),
695 latency_ms: None,
696 provider: ServerProvider::Custom(
697 host.split('.').next().unwrap_or("speedtest").to_string(),
698 ),
699 capabilities: ServerCapabilities {
700 supports_download: true,
701 supports_upload: true,
702 supports_latency: true,
703 max_test_size_mb: 1000,
704 geographic_weight: 1.0,
705 },
706 quality_score: None,
707 country_code: Some(country.to_string()),
708 city: Some(name.to_string()),
709 is_backup: false,
710 });
711 }
712 }
713 }
714 }
715
716 if servers.is_empty() {
717 Err("No servers parsed".into())
718 } else {
719 Ok(servers)
720 }
721 }
722
723 async fn get_open_speedtest_servers(
724 &self,
725 geo: &GeoLocation,
726 ) -> Result<Vec<TestServer>, Box<dyn std::error::Error>> {
727 let mut servers = Vec::new();
729
730 let endpoints = vec![
732 (
733 "Cloudflare (Anycast)",
734 "https://speed.cloudflare.com",
735 0.0,
736 0.0,
737 "Global",
738 ),
739 (
740 "LibreSpeed DE-IX",
741 "https://frankfurt.speedtest.wtnet.de",
742 50.1109,
743 8.6821,
744 "Frankfurt, Germany",
745 ),
746 (
747 "LibreSpeed AMS-IX",
748 "https://ams.speedtest.wtnet.de",
749 52.3676,
750 4.9041,
751 "Amsterdam, Netherlands",
752 ),
753 (
754 "LibreSpeed Singapore",
755 "https://sg.speedtest.wtnet.de",
756 1.3521,
757 103.8198,
758 "Singapore",
759 ),
760 (
761 "LibreSpeed New York",
762 "https://nyc.speedtest.wtnet.de",
763 40.7128,
764 -74.0060,
765 "New York, USA",
766 ),
767 (
768 "LibreSpeed Los Angeles",
769 "https://la.speedtest.wtnet.de",
770 34.0522,
771 -118.2437,
772 "Los Angeles, USA",
773 ),
774 (
775 "LibreSpeed Tokyo",
776 "https://tyo.speedtest.wtnet.de",
777 35.6762,
778 139.6503,
779 "Tokyo, Japan",
780 ),
781 (
782 "LibreSpeed London",
783 "https://lon.speedtest.wtnet.de",
784 51.5074,
785 -0.1278,
786 "London, UK",
787 ),
788 (
789 "LibreSpeed Sydney",
790 "https://syd.speedtest.wtnet.de",
791 -33.8688,
792 151.2093,
793 "Sydney, Australia",
794 ),
795 ];
796
797 for (name, url, lat, lon, location) in endpoints {
798 let distance = if lat == 0.0 && lon == 0.0 {
799 999999.0 } else {
801 self.calculate_distance(geo.latitude, geo.longitude, lat, lon)
802 };
803
804 servers.push(TestServer {
805 name: name.to_string(),
806 url: url.to_string(),
807 location: location.to_string(),
808 distance_km: Some(distance),
809 latency_ms: None,
810 provider: ServerProvider::Custom("LibreSpeed".to_string()),
811 capabilities: ServerCapabilities {
812 supports_download: true,
813 supports_upload: true,
814 supports_latency: true,
815 max_test_size_mb: 2000,
816 geographic_weight: 0.9,
817 },
818 quality_score: None,
819 country_code: Some(location.split(", ").last().unwrap_or("").to_string()),
820 city: Some(location.split(", ").next().unwrap_or(location).to_string()),
821 is_backup: false,
822 });
823 }
824
825 Ok(servers)
826 }
827
828 fn get_continent_servers(&self, geo: &GeoLocation) -> Vec<TestServer> {
829 let mut servers = Vec::new();
830
831 let continent = self.determine_continent(geo.latitude, geo.longitude);
833
834 match continent.as_str() {
835 "North America" => {
836 servers.push(self.create_server_with_coords(
837 geo,
838 "US East Coast Hub",
839 "https://ash.speedtest.wtnet.de",
840 "Ashburn, USA",
841 Some("US".to_string()),
842 39.0438,
843 -77.4874,
844 ));
845 servers.push(self.create_server_with_coords(
846 geo,
847 "US West Coast Hub",
848 "https://lax.speedtest.wtnet.de",
849 "Los Angeles, USA",
850 Some("US".to_string()),
851 34.0522,
852 -118.2437,
853 ));
854 }
855 "Europe" => {
856 servers.push(self.create_server_with_coords(
857 geo,
858 "Europe Central Hub",
859 "https://frankfurt.speedtest.wtnet.de",
860 "Frankfurt, Germany",
861 Some("DE".to_string()),
862 50.1109,
863 8.6821,
864 ));
865 servers.push(self.create_server_with_coords(
866 geo,
867 "Europe West Hub",
868 "https://lon.speedtest.wtnet.de",
869 "London, UK",
870 Some("GB".to_string()),
871 51.5074,
872 -0.1278,
873 ));
874 }
875 "Asia" => {
876 servers.push(self.create_server_with_coords(
877 geo,
878 "Asia Pacific Hub",
879 "https://sg.speedtest.wtnet.de",
880 "Singapore",
881 Some("SG".to_string()),
882 1.3521,
883 103.8198,
884 ));
885 servers.push(self.create_server_with_coords(
886 geo,
887 "Asia East Hub",
888 "https://tokyo.speedtest.wtnet.de",
889 "Tokyo, Japan",
890 Some("JP".to_string()),
891 35.6762,
892 139.6503,
893 ));
894 }
895 "South America" => {
896 servers.push(self.create_server_with_coords(
897 geo,
898 "South America Hub",
899 "https://saopaulo.speedtest.wtnet.de",
900 "São Paulo, Brazil",
901 Some("BR".to_string()),
902 -23.5505,
903 -46.6333,
904 ));
905 }
906 "Africa" => {
907 servers.push(self.create_server_with_coords(
908 geo,
909 "Africa Hub",
910 "https://capetown.speedtest.wtnet.de",
911 "Cape Town, South Africa",
912 Some("ZA".to_string()),
913 -33.9249,
914 18.4241,
915 ));
916 }
917 "Oceania" => {
918 servers.push(self.create_server_with_coords(
919 geo,
920 "Oceania Hub",
921 "https://syd.speedtest.wtnet.de",
922 "Sydney, Australia",
923 Some("AU".to_string()),
924 -33.8688,
925 151.2093,
926 ));
927 }
928 _ => {}
929 }
930
931 servers
932 }
933
934 fn determine_continent(&self, lat: f64, lon: f64) -> String {
935 if lat > 15.0 && lon > -130.0 && lon < -50.0 {
937 "North America".to_string()
938 } else if lat < 15.0 && lat > -60.0 && lon > -85.0 && lon < -30.0 {
939 "South America".to_string()
940 } else if lat > 35.0 && lon > -15.0 && lon < 60.0 {
941 "Europe".to_string()
942 } else if lat > -40.0 && lat < 40.0 && lon > -20.0 && lon < 55.0 {
943 "Africa".to_string()
944 } else if lat > -15.0 && lon > 60.0 && lon < 180.0 {
945 "Asia".to_string()
946 } else if lat < -10.0 && lon > 110.0 && lon < 180.0 {
947 "Oceania".to_string()
948 } else {
949 "Unknown".to_string()
950 }
951 }
952
953 fn get_country_servers(&self, geo: &GeoLocation) -> Vec<TestServer> {
954 let mut servers = Vec::new();
955
956 match geo.country.as_str() {
958 "United States" | "US" => {
959 servers.push(self.create_server(
960 "US Central",
961 "https://dal.speedtest.wtnet.de",
962 "Dallas, USA",
963 Some("US".to_string()),
964 ));
965 }
966 "United Kingdom" | "GB" | "UK" => {
967 servers.push(self.create_server(
968 "UK Primary",
969 "https://lon.speedtest.wtnet.de",
970 "London, UK",
971 Some("GB".to_string()),
972 ));
973 }
974 "Germany" | "DE" => {
975 servers.push(self.create_server(
976 "DE Primary",
977 "https://frankfurt.speedtest.wtnet.de",
978 "Frankfurt, Germany",
979 Some("DE".to_string()),
980 ));
981 }
982 "France" | "FR" => {
983 servers.push(self.create_server(
984 "FR Primary",
985 "https://paris.speedtest.wtnet.de",
986 "Paris, France",
987 Some("FR".to_string()),
988 ));
989 }
990 "Japan" | "JP" => {
991 servers.push(self.create_server(
992 "JP Primary",
993 "https://tyo.speedtest.wtnet.de",
994 "Tokyo, Japan",
995 Some("JP".to_string()),
996 ));
997 }
998 "Australia" | "AU" => {
999 servers.push(self.create_server(
1000 "AU Primary",
1001 "https://syd.speedtest.wtnet.de",
1002 "Sydney, Australia",
1003 Some("AU".to_string()),
1004 ));
1005 }
1006 "Canada" | "CA" => {
1007 servers.push(self.create_server(
1008 "CA Primary",
1009 "https://tor.speedtest.wtnet.de",
1010 "Toronto, Canada",
1011 Some("CA".to_string()),
1012 ));
1013 }
1014 _ => {}
1015 }
1016
1017 servers
1018 }
1019
1020 fn calculate_distance(&self, lat1: f64, lon1: f64, lat2: f64, lon2: f64) -> f64 {
1021 let r = 6371.0; let d_lat = (lat2 - lat1).to_radians();
1024 let d_lon = (lon2 - lon1).to_radians();
1025 let lat1 = lat1.to_radians();
1026 let lat2 = lat2.to_radians();
1027
1028 let a = (d_lat / 2.0).sin() * (d_lat / 2.0).sin()
1029 + lat1.cos() * lat2.cos() * (d_lon / 2.0).sin() * (d_lon / 2.0).sin();
1030 let c = 2.0 * a.sqrt().atan2((1.0 - a).sqrt());
1031
1032 r * c
1033 }
1034
1035 #[allow(clippy::too_many_arguments)]
1036 fn create_server_with_coords(
1037 &self,
1038 geo: &GeoLocation,
1039 name: &str,
1040 url: &str,
1041 location: &str,
1042 country_code: Option<String>,
1043 lat: f64,
1044 lon: f64,
1045 ) -> TestServer {
1046 let distance = self.calculate_distance(geo.latitude, geo.longitude, lat, lon);
1047
1048 TestServer {
1049 name: name.to_string(),
1050 url: url.to_string(),
1051 location: location.to_string(),
1052 distance_km: Some(distance),
1053 latency_ms: None,
1054 provider: ServerProvider::Custom("LibreSpeed".to_string()),
1055 capabilities: ServerCapabilities {
1056 supports_download: true,
1057 supports_upload: true,
1058 supports_latency: true,
1059 max_test_size_mb: 2000,
1060 geographic_weight: 1.0,
1061 },
1062 quality_score: None,
1063 country_code,
1064 city: Some(location.split(", ").next().unwrap_or(location).to_string()),
1065 is_backup: false,
1066 }
1067 }
1068
1069 fn create_server(
1070 &self,
1071 name: &str,
1072 url: &str,
1073 location: &str,
1074 country_code: Option<String>,
1075 ) -> TestServer {
1076 TestServer {
1077 name: name.to_string(),
1078 url: url.to_string(),
1079 location: location.to_string(),
1080 distance_km: None,
1081 latency_ms: None,
1082 provider: ServerProvider::Cloudflare,
1083 capabilities: ServerCapabilities {
1084 supports_download: true,
1085 supports_upload: true,
1086 supports_latency: true,
1087 max_test_size_mb: 1000,
1088 geographic_weight: 1.2,
1089 },
1090 quality_score: None,
1091 country_code,
1092 city: Some(location.split(',').next().unwrap_or("").trim().to_string()),
1093 is_backup: false,
1094 }
1095 }
1096
1097 fn determine_region(&self, country: &str) -> String {
1098 match country {
1099 "United States" | "Canada" | "Mexico" => "North America".to_string(),
1100 "United Kingdom" | "Germany" | "France" | "Spain" | "Italy" | "Netherlands"
1101 | "Belgium" | "Switzerland" | "Austria" | "Poland" => "Europe".to_string(),
1102 "Japan" | "China" | "South Korea" | "Singapore" | "Australia" | "New Zealand"
1103 | "India" => "Asia Pacific".to_string(),
1104 "Brazil" | "Argentina" | "Chile" => "South America".to_string(),
1105 _ => "Other".to_string(),
1106 }
1107 }
1108
1109 fn estimate_distance(&self, geo: &GeoLocation, server: &TestServer) -> f64 {
1110 let region = self.determine_region(&geo.country);
1113
1114 if let Some(city) = &server.city {
1115 if city.contains(&geo.city) {
1116 return 10.0; }
1118 }
1119
1120 match (region.as_str(), server.location.as_str()) {
1121 ("North America", loc) if loc.contains("USA") || loc.contains("Canada") => 500.0,
1122 ("Europe", loc) if loc.contains("Europe") || loc.contains("UK") => 300.0,
1123 ("Asia Pacific", loc) if loc.contains("Asia") || loc.contains("Japan") => 400.0,
1124 _ => 5000.0, }
1126 }
1127
1128 async fn select_best_servers(&self) -> Result<Vec<TestServer>, Box<dyn std::error::Error>> {
1130 emit(
1131 &self.events,
1132 TestEvent::Status("Testing server performance...".to_string()),
1133 );
1134
1135 let servers = self.server_pool.read().await.clone();
1136
1137 if servers.is_empty() {
1138 return Err("No servers in pool".into());
1139 }
1140
1141 let mut test_results = Vec::new();
1142
1143 let mut futures = FuturesUnordered::new();
1145
1146 for server in servers.into_iter().take(15) {
1147 let client = self.client.clone();
1148 futures.push(async move { Self::quick_latency_test(&client, &server).await });
1149 }
1150
1151 while let Some(result) = futures.next().await {
1152 if let Ok(mut server) = result {
1153 if let Some(latency) = server.latency_ms {
1154 let distance = server.distance_km.unwrap_or(1000.0);
1155 let geographic_weight = server.capabilities.geographic_weight;
1156
1157 let latency_penalty = latency.max(1.0); let distance_penalty = (distance / 100.0).max(1.0);
1162 server.quality_score =
1163 Some((10000.0 * geographic_weight) / (latency_penalty + distance_penalty));
1164
1165 test_results.push(server);
1166 }
1167 }
1168 }
1169
1170 if test_results.is_empty() {
1171 return Err("No servers responded to latency tests".into());
1172 }
1173
1174 test_results.sort_by(|a, b| {
1176 b.quality_score
1177 .unwrap_or(0.0)
1178 .partial_cmp(&a.quality_score.unwrap_or(0.0))
1179 .unwrap_or(std::cmp::Ordering::Equal)
1180 });
1181
1182 let selected = test_results
1183 .into_iter()
1184 .take(SERVER_SELECTION_COUNT)
1185 .collect::<Vec<_>>();
1186
1187 if !self.config.json_output {
1188 emit(
1189 &self.events,
1190 TestEvent::ServersSelected {
1191 servers: selected
1192 .iter()
1193 .map(|server| SelectedServer {
1194 name: server.name.clone(),
1195 location: server.location.clone(),
1196 latency_ms: server.latency_ms.unwrap_or(0.0),
1197 distance_km: server.distance_km.unwrap_or(0.0),
1198 })
1199 .collect(),
1200 },
1201 );
1202 }
1203
1204 Ok(selected)
1205 }
1206
1207 async fn quick_latency_test(
1208 client: &Client,
1209 server: &TestServer,
1210 ) -> Result<TestServer, Box<dyn std::error::Error>> {
1211 let mut latencies = Vec::new();
1212 let mut server = server.clone();
1213
1214 for _ in 0..3 {
1215 let start = Instant::now();
1216 match client
1217 .head(&server.url)
1218 .timeout(Duration::from_secs(2))
1219 .send()
1220 .await
1221 {
1222 Ok(resp) if resp.status().is_success() || resp.status().is_redirection() => {
1223 latencies.push(start.elapsed().as_millis() as f64);
1224 }
1225 _ => {}
1226 }
1227 }
1228
1229 if !latencies.is_empty() {
1230 server.latency_ms = Some(latencies.iter().sum::<f64>() / latencies.len() as f64);
1231 }
1232
1233 Ok(server)
1234 }
1235
1236 async fn progressive_download_test(
1238 &self,
1239 servers: &[TestServer],
1240 ) -> Result<f64, Box<dyn std::error::Error>> {
1241 emit(&self.events, TestEvent::PhaseStarted(Phase::Download));
1242
1243 let total_bytes = Arc::new(Mutex::new(0usize));
1244 let start = Instant::now();
1245 let test_duration = Duration::from_secs(15);
1246
1247 let mut handles = Vec::new();
1248
1249 for i in 0..PARALLEL_CONNECTIONS {
1251 let server = &servers[i % servers.len()];
1252 let url = format!("{}/__down?bytes=100000000", server.url); let client = self.client.clone();
1254 let total_bytes = Arc::clone(&total_bytes);
1255 let test_start = start;
1256
1257 let handle = tokio::spawn(async move {
1258 let end_time = test_start + test_duration;
1259
1260 while Instant::now() < end_time {
1261 match client.get(&url).send().await {
1262 Ok(response) => {
1263 let mut stream = response.bytes_stream();
1264
1265 while let Some(chunk_result) = stream.next().await {
1266 if Instant::now() >= end_time {
1267 break;
1268 }
1269 if let Ok(chunk) = chunk_result {
1270 let mut total = total_bytes.lock().await;
1271 *total += chunk.len();
1272 }
1273 }
1274 }
1275 Err(_) => {
1276 tokio::time::sleep(Duration::from_millis(100)).await;
1277 }
1278 }
1279
1280 if Instant::now() >= end_time {
1281 break;
1282 }
1283 }
1284 });
1285
1286 handles.push(handle);
1287 }
1288
1289 let total_bytes_monitor = Arc::clone(&total_bytes);
1291 let events = self.events.clone();
1292
1293 let monitor_handle = tokio::spawn(async move {
1294 let mut last_bytes = 0;
1295 let mut last_time = Instant::now();
1296 let mut peak = 0.0f64;
1297 let end_time = start + test_duration;
1298
1299 while Instant::now() < end_time {
1300 tokio::time::sleep(Duration::from_millis(200)).await;
1301
1302 let bytes = *total_bytes_monitor.lock().await;
1303 let time_diff = last_time.elapsed().as_secs_f64();
1304
1305 if time_diff >= 0.2 {
1306 let bytes_diff = bytes.saturating_sub(last_bytes);
1307 let speed = (bytes_diff as f64 * 8.0) / (time_diff * 1_000_000.0);
1308 peak = peak.max(speed);
1309
1310 emit(
1311 &events,
1312 TestEvent::DownloadSample {
1313 mbps: speed,
1314 peak_mbps: peak,
1315 elapsed_secs: start.elapsed().as_secs_f64(),
1316 },
1317 );
1318
1319 last_bytes = bytes;
1320 last_time = Instant::now();
1321 }
1322 }
1323 });
1324
1325 for handle in handles {
1327 let _ = handle.await;
1328 }
1329 let _ = monitor_handle.await;
1330
1331 let elapsed = start.elapsed().as_secs_f64();
1333 let total = *total_bytes.lock().await;
1334
1335 let mbps = if total > 1_000_000 && elapsed > 1.0 {
1336 let bits = total as f64 * 8.0;
1337 bits / (elapsed * 1_000_000.0)
1338 } else {
1339 1.0 };
1341
1342 let mbps = mbps.clamp(1.0, 10_000.0);
1343 emit(&self.events, TestEvent::DownloadComplete { mbps });
1344
1345 Ok(mbps)
1346 }
1347
1348 async fn progressive_upload_test(
1350 &self,
1351 servers: &[TestServer],
1352 ) -> Result<f64, Box<dyn std::error::Error>> {
1353 emit(&self.events, TestEvent::PhaseStarted(Phase::Upload));
1354
1355 let total_bytes = Arc::new(Mutex::new(0usize));
1356 let start = Instant::now();
1357 let test_duration = Duration::from_secs(15);
1358
1359 let chunk_size = 5 * 1024 * 1024;
1361 let test_data = vec![0u8; chunk_size];
1362
1363 let mut handles = Vec::new();
1364
1365 for i in 0..10 {
1367 let server = &servers[i % servers.len()];
1368 let url = format!("{}/__up", server.url);
1369 let client = self.client.clone();
1370 let total_bytes = Arc::clone(&total_bytes);
1371 let data = test_data.clone();
1372 let test_start = start;
1373
1374 let handle = tokio::spawn(async move {
1375 let end_time = test_start + test_duration;
1376
1377 while Instant::now() < end_time {
1378 match client
1379 .post(&url)
1380 .body(data.clone())
1381 .timeout(Duration::from_secs(10))
1382 .send()
1383 .await
1384 {
1385 Ok(_) => {
1386 let mut total = total_bytes.lock().await;
1387 *total += data.len();
1388 }
1389 Err(_) => {
1390 tokio::time::sleep(Duration::from_millis(100)).await;
1391 }
1392 }
1393 }
1394 });
1395
1396 handles.push(handle);
1397 }
1398
1399 let total_bytes_monitor = Arc::clone(&total_bytes);
1401 let events = self.events.clone();
1402
1403 let monitor_handle = tokio::spawn(async move {
1404 let mut last_bytes = 0;
1405 let mut last_time = Instant::now();
1406 let mut peak = 0.0f64;
1407 let end_time = start + test_duration;
1408
1409 while Instant::now() < end_time {
1410 tokio::time::sleep(Duration::from_millis(200)).await;
1411
1412 let bytes = *total_bytes_monitor.lock().await;
1413 let time_diff = last_time.elapsed().as_secs_f64();
1414
1415 if time_diff >= 0.2 {
1416 let bytes_diff = bytes.saturating_sub(last_bytes);
1417 let speed = (bytes_diff as f64 * 8.0) / (time_diff * 1_000_000.0);
1418 peak = peak.max(speed);
1419
1420 emit(
1421 &events,
1422 TestEvent::UploadSample {
1423 mbps: speed,
1424 peak_mbps: peak,
1425 elapsed_secs: start.elapsed().as_secs_f64(),
1426 },
1427 );
1428
1429 last_bytes = bytes;
1430 last_time = Instant::now();
1431 }
1432 }
1433 });
1434
1435 for handle in handles {
1437 let _ = handle.await;
1438 }
1439 let _ = monitor_handle.await;
1440
1441 let elapsed = start.elapsed().as_secs_f64();
1443 let total = *total_bytes.lock().await;
1444
1445 let mbps = if total > 1_000_000 && elapsed > 1.0 {
1446 let bits = total as f64 * 8.0;
1447 bits / (elapsed * 1_000_000.0)
1448 } else {
1449 1.0 };
1451
1452 let mbps = mbps.clamp(1.0, 10_000.0);
1453 emit(&self.events, TestEvent::UploadComplete { mbps });
1454
1455 Ok(mbps)
1456 }
1457
1458 async fn measure_latency(
1459 &self,
1460 server: &TestServer,
1461 ) -> Result<f64, Box<dyn std::error::Error>> {
1462 emit(&self.events, TestEvent::PhaseStarted(Phase::Latency));
1463
1464 let mut latencies = Vec::new();
1465
1466 for _i in 0..10 {
1467 let start = Instant::now();
1468 match self
1469 .client
1470 .head(&server.url)
1471 .timeout(Duration::from_secs(2))
1472 .send()
1473 .await
1474 {
1475 Ok(resp) if resp.status().is_success() || resp.status().is_redirection() => {
1476 let latency = start.elapsed().as_millis() as f64;
1477 latencies.push(latency);
1478
1479 let current_avg = latencies.iter().sum::<f64>() / latencies.len() as f64;
1480 emit(
1481 &self.events,
1482 TestEvent::LatencyProgress {
1483 avg_ms: current_avg,
1484 },
1485 );
1486 }
1487 _ => {}
1488 }
1489
1490 tokio::time::sleep(Duration::from_millis(100)).await;
1491 }
1492
1493 let avg_latency = if !latencies.is_empty() {
1494 latencies.iter().sum::<f64>() / latencies.len() as f64
1495 } else {
1496 50.0
1497 };
1498
1499 emit(
1500 &self.events,
1501 TestEvent::LatencyComplete {
1502 avg_ms: avg_latency,
1503 },
1504 );
1505
1506 Ok(avg_latency)
1507 }
1508
1509 async fn measure_jitter_and_loss(
1510 &self,
1511 server: &TestServer,
1512 ) -> Result<(f64, f64), Box<dyn std::error::Error>> {
1513 let mut latencies = Vec::new();
1514 let mut lost = 0;
1515 let total = 20;
1516
1517 for _ in 0..total {
1518 let start = Instant::now();
1519 match self
1520 .client
1521 .head(&server.url)
1522 .timeout(Duration::from_secs(1))
1523 .send()
1524 .await
1525 {
1526 Ok(resp) if resp.status().is_success() || resp.status().is_redirection() => {
1527 latencies.push(start.elapsed().as_millis() as f64);
1528 }
1529 _ => {
1530 lost += 1;
1531 }
1532 }
1533 tokio::time::sleep(Duration::from_millis(50)).await;
1534 }
1535
1536 let jitter = if latencies.len() > 1 {
1537 let mean = latencies.iter().sum::<f64>() / latencies.len() as f64;
1538 let variance =
1539 latencies.iter().map(|&x| (x - mean).powi(2)).sum::<f64>() / latencies.len() as f64;
1540 variance.sqrt()
1541 } else {
1542 0.0
1543 };
1544
1545 let packet_loss = (lost as f64 / total as f64) * 100.0;
1546
1547 Ok((jitter, packet_loss))
1548 }
1549
1550 async fn get_client_ip(&self) -> Option<IpAddr> {
1551 if let Ok(response) = self
1552 .client
1553 .get("https://api.ipify.org?format=json")
1554 .timeout(Duration::from_secs(3))
1555 .send()
1556 .await
1557 {
1558 if let Ok(json) = response.json::<serde_json::Value>().await {
1559 return json["ip"].as_str().and_then(|s| s.parse::<IpAddr>().ok());
1560 }
1561 }
1562 None
1563 }
1564
1565 async fn resolve_server_ip(&self, url: &str) -> Option<IpAddr> {
1566 if let Ok(parsed) = url.parse::<reqwest::Url>() {
1567 if let Some(host) = parsed.host_str() {
1568 if let Ok(addrs) = tokio::net::lookup_host(format!("{}:443", host)).await {
1569 return addrs.into_iter().next().map(|addr| addr.ip());
1570 }
1571 }
1572 }
1573 None
1574 }
1575}
1576
1577#[cfg(test)]
1578mod tests {
1579 use super::*;
1580
1581 #[test]
1582 fn test_region_determination() {
1583 let _ = rustls::crypto::ring::default_provider().install_default();
1585 let config = TestConfig::default();
1586 let speed_test = SpeedTest::new(config).unwrap();
1587
1588 assert_eq!(
1589 speed_test.determine_region("United States"),
1590 "North America"
1591 );
1592 assert_eq!(speed_test.determine_region("Germany"), "Europe");
1593 assert_eq!(speed_test.determine_region("Japan"), "Asia Pacific");
1594 }
1595}