feat: Discord RPC + HTTP and Crypto utils

packages/core/examples/discord.rs

@@ -0,0 +1,25 @@
+use onelauncher_core::{api::proxy::proxy_empty::ProxyEmpty, error::LauncherResult, initialize_core, store::State};
+
+#[tokio::main]
+async fn main() -> LauncherResult<()> {
+	initialize_core(ProxyEmpty::new()).await?;
+
+	let state = State::get().await?;
+	let Some(rpc) = &state.rpc else {
+		println!("RPC is not initialized");
+		return Ok(());
+	};
+
+	rpc.set_message("Hello, Discord!").await;
+	println!("Set Discord message");
+
+	tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
+	rpc.set_message("This user is testing Discord RPC!").await;
+	println!("Set Discord message");
+
+	tokio::time::sleep(tokio::time::Duration::from_secs(5)).await;
+	rpc.clear_activity().await;
+	println!("Cleared Discord message");
+
+	Ok(())
+}

packages/core/src/constants.rs

@@ -63,16 +63,12 @@ pub const SETTINGS_FILE: &str = "settings.json";
 ///
 /// [`MinecraftAuth`]: crate::store::MinecraftAuth
 pub const AUTH_FILE: &str = "authentication.json";
-/// The public `processor.json` file used to store the global [`Processor`] state.
-///
-/// [`Processor`]: crate::store::Processor
-pub const PROCESSOR_FILE: &str = "processor.json";
 
 /// The current [`Settings`] format version, bumped for breaking changes.
 /// If updated, a config file migration logic **NEEDS** to be implemented.
 ///
 /// [`Settings`]: crate::store::Settings
-pub const CURRENT_SETTINGS_FORMAT_VERSION: u32 = 1;
+pub const CURRENT_SETTINGS_FORMAT_VERSION: u32 = 2;
 
 // =========== Hacky Mojang-spec OS constants ===========
 #[cfg(target_os = "windows")]

packages/core/src/error.rs

@@ -8,6 +8,11 @@ pub enum LauncherError {
 	IngressError(#[from] crate::store::ingress::IngressError),
 	#[error(transparent)]
 	JavaError(#[from] crate::api::java::JavaError),
+	#[error(transparent)]
+	CryptoError(#[from] crate::utils::crypto::CryptoError),
+
+	#[error("failed to connect to discord rpc socket")]
+	DiscordError,
 
 	#[error(transparent)]
 	SerdeError(#[from] serde_json::Error),
@@ -15,6 +20,10 @@ pub enum LauncherError {
 	AnyhowError(#[from] anyhow::Error),
 	#[error(transparent)]
 	DbError(#[from] sea_orm::DbErr),
+	#[error(transparent)]
+	ReqwestError(#[from] reqwest::Error),
+	#[error("couldn't acquire semaphore: {0}")]
+	SemaphoreError(#[from] tokio::sync::AcquireError),
 
 	#[cfg(feature = "tauri")]
 	#[error(transparent)]

packages/core/src/store/discord.rs

@@ -0,0 +1,95 @@
+use std::sync::{atomic::AtomicBool, Arc};
+
+use discord_rich_presence::{activity::{Activity, Assets, Button}, DiscordIpc, DiscordIpcClient};
+use tokio::sync::RwLock;
+
+use crate::{constants, error::{LauncherError, LauncherResult}};
+
+pub struct DiscordRPC {
+	client: Arc<RwLock<DiscordIpcClient>>,
+	connected: Arc<AtomicBool>,
+}
+
+impl DiscordRPC {
+	pub fn initialize() -> LauncherResult<Self> {
+		let mut client = DiscordIpcClient::new(constants::DISCORD_RPC_CLIENT_ID)
+			.map_err(|_| LauncherError::DiscordError)?;
+
+		let connected = client.connect().is_ok();
+
+		Ok(Self {
+			client: Arc::new(RwLock::new(client)),
+			connected: Arc::new(AtomicBool::new(connected)),
+		})
+	}
+
+	pub async fn set_message(&self, msg: &str) -> bool {
+		self.set_activity(
+			Activity::new()
+				.state(msg)
+				.buttons(vec![Button::new("Website", "https://polyfrost.org/")])
+				.assets(
+					Assets::new()
+						.large_image("onelauncher_logo_512")
+						.large_text("OneLauncher"),
+				)
+		).await
+	}
+
+	pub async fn set_activity(&self, activity: Activity<'_>) -> bool {
+		if !self.is_connected() {
+			return false;
+		}
+
+		let mut client = self.client.write().await;
+
+		if client.set_activity(activity).is_err() {
+			return false;
+		}
+
+		true
+	}
+
+	pub async fn clear_activity(&self) -> bool {
+		if !self.is_connected() {
+			return false;
+		}
+
+		let mut client = self.client.write().await;
+
+		if client.clear_activity().is_err() {
+			return false;
+		}
+
+		true
+	}
+
+	pub async fn reconnect(&self) -> bool {
+		let mut client = self.client.write().await;
+
+		let connected = client.reconnect().is_ok();
+
+		self.set_connected(connected);
+		connected
+	}
+
+	pub async fn close(&self) -> bool {
+		let mut client = self.client.write().await;
+
+		if client.close().is_err() {
+			return false;
+		}
+
+		self.set_connected(false);
+		true
+	}
+
+	#[must_use]
+	pub fn is_connected(&self) -> bool {
+		self.connected.load(std::sync::atomic::Ordering::Relaxed)
+	}
+
+	fn set_connected(&self, connected: bool) {
+		self.connected.store(connected, std::sync::atomic::Ordering::Relaxed);
+	}
+}

packages/core/src/store/mod.rs

@@ -1,4 +1,5 @@
 pub(super) mod db;
+pub mod discord;
 pub mod ingress;
 pub mod proxy;
 pub mod java;

packages/core/src/store/settings.rs

@@ -1,7 +1,7 @@
 use onelauncher_entity::setting_profiles;
 use serde::{Deserialize, Serialize};
 
-use crate::{send_warning, utils::io, LauncherResult};
+use crate::{constants, send_warning, utils::io, LauncherResult};
 
 use super::Dirs;
 
@@ -13,6 +13,7 @@ pub struct Settings {
 	pub allow_parallel_running_clusters: bool,
 	pub enable_gamemode: bool,
 	pub discord_enabled: bool,
+	pub max_concurrent_requests: usize,
 	pub settings_version: u32,
 }
 
@@ -23,7 +24,8 @@ impl Default for Settings {
 			allow_parallel_running_clusters: false,
 			discord_enabled: false,
 			enable_gamemode: false,
-			settings_version: 2,
+			max_concurrent_requests: 25,
+			settings_version: constants::CURRENT_SETTINGS_FORMAT_VERSION,
 		}
 	}
 }

packages/core/src/store/state.rs

@@ -3,17 +3,21 @@ use std::sync::Arc;
 use sea_orm::DatabaseConnection;
 use tokio::sync::{OnceCell, RwLock};
 
-use crate::LauncherResult;
+use crate::{utils::http::{self, FetchSemaphore}, LauncherResult};
 
-use super::{ingress::IngressProcessor, Settings};
+use super::{discord::DiscordRPC, ingress::IngressProcessor, Settings};
 
 /// The static [`OnceCell<RwLock<State>>`] for storing the global runtime launcher state.
 static LAUNCHER_STATE: OnceCell<RwLock<State>> = OnceCell::const_new();
 
 pub struct State {
 	pub ingress_processor: IngressProcessor,
-	pub settings: Settings,
+	pub settings: RwLock<Settings>,
 	pub db: DatabaseConnection,
+	pub rpc: Option<DiscordRPC>,
+
+	pub client: Arc<reqwest::Client>,
+	pub(crate) fetch_semaphore: FetchSemaphore,
 }
 
 impl State {
@@ -46,11 +50,26 @@ impl State {
 		let ingress_processor = IngressProcessor::new();
 		let settings = Settings::new().await;
 		let db = super::db::create_pool().await?;
+		
+		let rpc = match DiscordRPC::initialize() {
+			Ok(rpc) => Some(rpc),
+			Err(err) => {
+				tracing::warn!("{}", err);
+				None
+			},
+		};
+
+		let reqwest_client = http::create_client()?;
+		let fetch_semaphore = FetchSemaphore::new(settings.max_concurrent_requests);
 
 		Ok(RwLock::new(Self {
 			ingress_processor,
-			settings,
+			settings: RwLock::new(settings),
 			db,
+			rpc,
+
+			client: Arc::new(reqwest_client),
+			fetch_semaphore,
 		}))
 	}
 }

packages/core/src/utils/crypto.rs

@@ -0,0 +1,46 @@
+use std::{fmt::Display, str::FromStr};
+
+#[derive(Debug, thiserror::Error)]
+pub enum CryptoError {
+	#[error("expected {algorithm} hash {expected}, but got {actual}")]
+	InvalidHash {
+		algorithm: HashAlgorithm,
+		expected: String,
+		actual: String,
+	},
+	#[error("invalid hash algorithm")]
+	InvalidAlgorithm
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum HashAlgorithm {
+	Sha1,
+}
+
+impl HashAlgorithm {
+	#[must_use]
+	pub fn hash(&self, data: &[u8]) -> String {
+		match self {
+			Self::Sha1 => sha1_smol::Sha1::from(data).digest().to_string(),
+		}
+	}
+}
+
+impl Display for HashAlgorithm {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		match self {
+			Self::Sha1 => write!(f, "sha1"),
+		}
+	}
+}
+
+impl FromStr for HashAlgorithm {
+	type Err = CryptoError;
+
+	fn from_str(s: &str) -> Result<Self, Self::Err> {
+		match s.to_lowercase().as_str() {
+			"sha1" => Ok(Self::Sha1),
+			_ => Err(CryptoError::InvalidAlgorithm),
+		}
+	}
+}

packages/core/src/utils/http.rs

@@ -0,0 +1,191 @@
+#![allow(clippy::implicit_hasher)]
+
+use std::{collections::HashMap, path::Path, sync::Arc};
+
+use bytes::Bytes;
+use reqwest::Method;
+use serde::Deserialize;
+use tokio::sync::Semaphore;
+use tokio_stream::StreamExt;
+
+use crate::{api::ingress::send_ingress, constants, error::LauncherResult, store::{ingress::IngressId, State}};
+
+use super::{crypto::{CryptoError, HashAlgorithm}, io};
+
+#[derive(Debug)]
+pub struct FetchSemaphore(pub Arc<Semaphore>);
+
+impl FetchSemaphore {
+	#[must_use]
+	pub fn new(limit: usize) -> Self {
+		Self(Arc::new(Semaphore::new(limit)))
+	}
+}
+
+pub fn create_client() -> LauncherResult<reqwest::Client> {
+	Ok(reqwest::ClientBuilder::new()
+		.tcp_keepalive(Some(std::time::Duration::from_secs(15)))
+		.use_rustls_tls()
+		.default_headers({
+			let mut headers = reqwest::header::HeaderMap::new();
+			let header = reqwest::header::HeaderValue::from_str(&format!(
+				"{}/{} (https://polyfrost.org)",
+				constants::NAME,
+				constants::VERSION,
+			))
+			.expect("failed to build reqwest headers!");
+			headers.insert(reqwest::header::USER_AGENT, header);
+			headers
+		})
+		.build()?)
+}
+
+#[tracing::instrument(skip(body, headers, ingress))]
+pub async fn fetch_advanced(
+	method: Method,
+	url: &str,
+	body: Option<serde_json::Value>,
+	hash: Option<(HashAlgorithm, &str)>,
+	headers: Option<HashMap<&str, &str>>,
+	ingress: Option<(&IngressId, f64)>,
+) -> LauncherResult<Bytes> {
+	let state = State::get().await?;
+	let _permit = state.fetch_semaphore.0.acquire().await?;
+	let client = &state.client;
+
+	for attempt in 0..constants::FETCH_ATTEMPTS {
+		let mut req = client.request(method.clone(), url);
+
+		if let Some(body) = body.clone() {
+			req = req.json(&body);
+		}
+
+		if let Some(headers) = &headers {
+			for (key, value) in headers {
+				req = req.header(*key, *value);
+			}
+		}
+
+		let res = req.send().await;
+
+		match res {
+			Err(_) if attempt <= 3 => {},
+			Err(err) => return Err(err.into()),
+			Ok(res) => {
+				let bytes = if let Some((ingress_id, total)) = ingress {
+					let length = res.content_length();
+					if let Some(total_size) = length {
+						let mut stream = res.bytes_stream();
+						let mut bytes = Vec::new();
+
+						while let Some(item) = stream.next().await {
+							let chunk = item.or(Err(anyhow::anyhow!("no value for fetch bytes")))?;
+							bytes.append(&mut chunk.to_vec());
+
+							send_ingress(
+								ingress_id,
+								(chunk.len() as f64 / total_size as f64) * total,
+							)
+							.await?;
+						}
+
+						Ok(Bytes::from(bytes))
+					} else {
+						res.bytes().await
+					}
+				} else {
+					res.bytes().await
+				};
+
+				if let Ok(bytes) = bytes {
+					if let Some((ref algorithm, expected_hash)) = hash {
+						let expected_hash = expected_hash.to_string();
+						let calculated_hash = algorithm.hash(&bytes);
+
+						if *calculated_hash != expected_hash {
+							if attempt <= 3 {
+								continue;
+							}
+
+							return Err(CryptoError::InvalidHash {
+								algorithm: algorithm.clone(),
+								expected: expected_hash,
+								actual: calculated_hash
+							}.into());
+						}
+					}
+
+					tracing::debug!("finished downloading {url}");
+					return Ok(bytes);
+				} else if let Err(err) = bytes {
+					return Err(err.into());
+				}
+			},
+		}
+	}
+
+	unreachable!()
+}
+
+pub async fn fetch(method: Method, url: &str) -> LauncherResult<Bytes> {
+	fetch_advanced(method, url, None, None, None, None).await
+}
+
+pub async fn fetch_json<T: for<'de> Deserialize<'de>>(
+	method: Method,
+	url: &str,
+	body: Option<serde_json::Value>,
+	headers: Option<HashMap<&str, &str>>,
+) -> LauncherResult<T> {
+	fetch_json_advanced(method, url, body, None, headers, None).await
+}
+
+pub async fn fetch_json_advanced<T: for<'de> Deserialize<'de>>(
+	method: Method,
+	url: &str,
+	body: Option<serde_json::Value>,
+	hash: Option<(HashAlgorithm, &str)>,
+	headers: Option<HashMap<&str, &str>>,
+	ingress: Option<(&IngressId, f64)>,
+) -> LauncherResult<T> {
+	let bytes = fetch_advanced(method, url, body, hash, headers, ingress).await?;
+	Ok(serde_json::from_slice(&bytes)?)
+}
+
+pub async fn download_advanced(
+	method: Method,
+	url: &str,
+	path: impl AsRef<Path>,
+	body: Option<serde_json::Value>,
+	hash: Option<(HashAlgorithm, &str)>,
+	headers: Option<HashMap<&str, &str>>,
+	ingress: Option<(&IngressId, f64)>,
+) -> LauncherResult<()> {
+	let bytes = fetch_advanced(method, url, body, hash, headers, ingress.map(|(id, total)| (id, total / 2.0))).await?;
+	let path = path.as_ref();
+
+	if let Some((id, total)) = ingress {
+		send_ingress(id, total / 4.0).await?;
+	}
+
+	if let Some(parent) = path.parent() {
+		io::create_dir_all(parent).await?;
+	}
+
+	io::write(path, &bytes).await?;
+
+	if let Some((id, total)) = ingress {
+		send_ingress(id, total / 4.0).await?;
+	}
+
+	Ok(())
+}
+
+pub async fn download(
+	method: Method,
+	url: &str,
+	path: impl AsRef<Path>,
+	hash: Option<(HashAlgorithm, &str)>
+) -> LauncherResult<()> {
+	download_advanced(method, url, path, None, hash, None, None).await
+}

packages/core/src/utils/mod.rs

@@ -1 +1,3 @@
+pub mod crypto;
+pub mod http;
 pub mod io;