datastore refactor

rnex-prudp/src/socket_addr.rs

@@ -22,7 +22,7 @@ impl PRUDPSockAddr {
     }
 
     pub fn calculate_connection_signature(&self) -> [u8; 16] {
-        let mut hmac = Md5Hmac::new_from_slice(&[0; 16]).expect("?");
+        let mut hmac = Md5Hmac::new_from_slice(&[0; 16]).expect("incorrect slice size");
 
         let data = match self.regular_socket_addr.ip() {
             IpAddr::V4(v) => v.octets().to_vec(),
@@ -33,7 +33,7 @@ impl PRUDPSockAddr {
         hmac.update(&data);
         let result: [u8; 16] = hmac.finalize().into_bytes()[0..16]
             .try_into()
-            .expect("fuck");
+            .expect("incorrect result size");
         result
     }
 }

rnex-rmc/src/networking.rs

@@ -73,7 +73,7 @@ impl RmcSerialize for SocketAddrV6 {
         ))
     }
     fn serialize_write_size(&self) -> Result<u32> {
-        Ok(6)
+        Ok(26)
     }
 }
 /*

rnex-rmc/src/primitives.rs

@@ -124,7 +124,7 @@ impl RmcSerialize for u128 {
     }
     #[inline(always)]
     fn serialize_write_size(&self) -> Result<u32> {
-        Ok(8)
+        Ok(16)
     }
 }
 

rnex-rmc/src/response.rs

@@ -465,6 +465,6 @@ impl From<Error> for ErrorCode {
 
 impl Into<u32> for ErrorCode {
     fn into(self) -> u32 {
-        unsafe { transmute(self) }
+        self as u32
     }
 }

rnex-rmc/src/rmc_struct.rs

@@ -77,7 +77,7 @@ impl<'a, T: Read + ?Sized> SubRead<'a, T> {
 impl<T: Read + ?Sized> Read for SubRead<'_, T> {
     #[inline(always)]
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
-        let max_read = usize::max(self.left_to_read, buf.len());
+        let max_read = usize::min(self.left_to_read, buf.len());
         let read = self.origin.read(&mut buf[..max_read])?;
         self.left_to_read -= read;
         Ok(read)

rnex-rmc/src/station_url.rs

@@ -1,5 +1,6 @@
 use std::fmt::Write;
 use std::io::{self, Read};
+use std::str::FromStr;
 
 use rnex_util::station_url::StationUrl;
 
@@ -12,7 +13,7 @@ impl RmcSerialize for StationUrl {
     fn deserialize(reader: &mut (impl Read + ?Sized)) -> Result<Self> {
         let str = String::deserialize(reader)?;
 
-        Self::try_from(str.as_str()).map_err(|_| StationUrlInvalid)
+        Self::from_str(str.as_str()).map_err(|_| StationUrlInvalid)
     }
     fn serialize(&self, writer: &mut (impl io::Write + ?Sized)) -> Result<()> {
         let str: String = self.into();

rnex-server-nex-modules/rnex-ds/Cargo.toml

@@ -4,6 +4,25 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
+rnex-rmc = { path = "../../rnex-rmc" }
+rnex-util = { path = "../../rnex-util" }
+rnex-base = { path = "../rnex-base" }
+rnex-base-protos = { path = "../../rnex-protocols/base-protos" }
+rnex-ds-protos = { path = "../../rnex-protocols/ds-protos" }
+rnex-server = { path = "../../rnex-server" }
+sqlx = "0.9.0"
+tracing = "0.1.44"
+thiserror = "2.0.18"
+chrono = "0.4.45"
+aws-sdk-s3 = "1.138.0"
+aws-config = "1.9.0"
+sha2 = "0.11.0"
+hmac = "0.13.0"
+base64 = "0.22.1"
+serde_json = "1.0.150"
+hex = "0.4.3"
+urlencoding = "2.1.3"
+futures = "0.3.32"
 
 [lints]
 workspace = true

rnex-server-nex-modules/rnex-ds/src/lib.rs

@@ -0,0 +1,61 @@
+use std::env;
+
+use rnex_server::{ConnectionInitData, RnexManager, RnexModule};
+use sqlx::PgPool;
+use thiserror::Error;
+
+use crate::{datastore::DatastoreUser, s3presigner::S3Presigner};
+
+pub mod datastore;
+pub(crate) mod s3presigner;
+
+struct DatastoreManager {
+    db_pool: PgPool,
+    s3_presigner: S3Presigner,
+}
+struct DatastoreModule;
+impl RnexManager for DatastoreManager {
+    type User = DatastoreUser;
+    type InitData = ConnectionInitData;
+    async fn init_new_user(
+        this: rnex_server::PassthroughInitModule<Self>,
+        mod_holder: &rnex_server::ModuleHolder,
+        remote: &rnex_rmc::RmcConnection,
+        init_data: &Self::InitData,
+        weak_user: rnex_server::WeakPassthroughInitModule<Self::User>,
+    ) -> Self::User {
+        DatastoreUser {
+            dm: this,
+            base: mod_holder
+                .get_ref_init_pt()
+                .expect("datastore module cannot work without base module"),
+        }
+    }
+}
+
+#[derive(Error, Debug)]
+pub enum ModuleInitError {
+    #[error(transparent)]
+    Sqlx(#[from] sqlx::Error),
+    #[error(transparent)]
+    Env(#[from] env::VarError),
+}
+
+impl RnexModule for DatastoreModule {
+    type Manager = DatastoreManager;
+    type InitError = ModuleInitError;
+
+    async fn create_manager(
+        mod_holder: &rnex_server::ModuleHolder,
+    ) -> Result<Self::Manager, Self::InitError> {
+        Ok(DatastoreManager {
+            db_pool: PgPool::connect(&env::var("RNEX_DATASTORE_DATABASE")?).await?,
+            s3_presigner: S3Presigner::new(
+                env::var("RNEX_DATASTORE_S3_ENDPOINT")?
+                    .trim_end_matches('/')
+                    .to_string(),
+                env::var("RNEX_DATASTORE_S3_BUCKET")?,
+            ),
+        })
+    }
+}

rnex-server-nex-modules/rnex-ds/src/main.rs

@@ -1,3 +0,0 @@
-fn main() {
-    println!("Hello, world!");
-}

rnex-server-nex-modules/rnex-ds/src/s3presigner.rs

@@ -1,9 +1,8 @@
-use base64::{engine::general_purpose, Engine as _};
-use hmac::{Hmac, Mac};
-use sha2::{Sha256, Digest};
-use chrono::{Utc, Duration};
+use base64::{Engine, engine::general_purpose::STANDARD};
+use chrono::{Duration, Utc};
+use hmac::{Hmac, KeyInit, Mac};
 use serde_json::json;
-use rnex_core::executables::common::RNEX_DATASTORE_S3_ENDPOINT;
+use sha2::{Digest, Sha256};
 
 pub struct S3Presigner {
     endpoint: String,
@@ -11,9 +10,9 @@ pub struct S3Presigner {
 }
 
 impl S3Presigner {
-    pub async fn new(endpoint: &str, bucket: String) -> Self {
+    pub fn new(endpoint: String, bucket: String) -> Self {
         Self {
-            endpoint: endpoint.trim_end_matches('/').to_string(),
+            endpoint: endpoint,
             bucket,
         }
     }
@@ -23,7 +22,9 @@ impl S3Presigner {
         let region = "us-east-1"; // hardcoded because its the default region for most s3 clones
         let date_short = Utc::now().format("%Y%m%d").to_string();
         let date_full = Utc::now().format("%Y%m%dT%H%M%SZ").to_string();
-        let expiration = (Utc::now() + Duration::minutes(15)).format("%Y-%m-%dT%H:%M:%SZ").to_string();
+        let expiration = (Utc::now() + Duration::minutes(15))
+            .format("%Y-%m-%dT%H:%M:%SZ")
+            .to_string();
 
         let credential = format!("{}/{}/{}/s3/aws4_request", access_key, date_short, region);
 
@@ -38,20 +39,23 @@ impl S3Presigner {
             ]
         });
 
-        let policy_base64 = general_purpose::STANDARD.encode(policy_json.to_string());
+        let policy_base64 = STANDARD.encode(policy_json.to_string());
 
         let signature = self.calculate_signature(&secret_key, &date_short, region, &policy_base64);
 
         let fields = vec![
             ("key".to_string(), key.to_string()),
-            ("X-Amz-Algorithm".to_string(), "AWS4-HMAC-SHA256".to_string()),
+            (
+                "X-Amz-Algorithm".to_string(),
+                "AWS4-HMAC-SHA256".to_string(),
+            ),
             ("X-Amz-Credential".to_string(), credential),
             ("X-Amz-Date".to_string(), date_full),
             ("Policy".to_string(), policy_base64),
             ("X-Amz-Signature".to_string(), signature),
         ];
 
-        let url = format!("https://{}/{}", *RNEX_DATASTORE_S3_ENDPOINT, self.bucket);
+        let url = format!("https://{}/{}", self.endpoint, self.bucket);
         (url, fields)
     }
 
@@ -77,7 +81,7 @@ impl S3Presigner {
 
         let canonical_request = format!(
             "GET\n/{}/{}\n{}\nhost:{}\n\nhost\nUNSIGNED-PAYLOAD",
-            self.bucket, key, query_string, *RNEX_DATASTORE_S3_ENDPOINT
+            self.bucket, key, query_string, self.endpoint
         );
 
         let hashed_request = hex::encode(Sha256::digest(canonical_request.as_bytes()));
@@ -95,7 +99,7 @@ impl S3Presigner {
 
         format!(
             "https://{}/{}/{}?{}&X-Amz-Signature={}",
-            *RNEX_DATASTORE_S3_ENDPOINT, self.bucket, key, query_string, signature
+            self.endpoint, self.bucket, key, query_string, signature
         )
     }
 
@@ -113,4 +117,4 @@ impl S3Presigner {
         mac.update(data.as_bytes());
         mac.finalize().into_bytes().to_vec()
     }
-}
+}

rnex-util/Cargo.toml

@@ -9,6 +9,7 @@ tracing = "0.1.44"
 chrono = "0.4.39"
 bytemuck = { version = "1.25.0", features = ["derive"] }
 md-5 = "0.11.0"
+thiserror = "2.0.18"
 
 [features]
 nx = []

rnex-util/src/station_url.rs

@@ -1,8 +1,10 @@
 use std::{
     fmt::{Debug, Display, Formatter},
     net::IpAddr,
+    str::FromStr,
 };
 
+use thiserror::Error;
 use tracing::error;
 
 use crate::PID;
@@ -86,11 +88,15 @@ impl StationUrl {
     }
 }
 
-impl TryFrom<&str> for StationUrl {
-    type Error = ();
+// todo: add more specific error messages to parsing
+#[derive(Error, Debug)]
+#[error("failed to parse station url")]
+pub struct StationUrlParseError;
 
-    fn try_from(value: &str) -> Result<Self, ()> {
-        let (url_type, options) = value.split_at(value.find(":/").ok_or(())?);
+impl FromStr for StationUrl {
+    type Err = StationUrlParseError;
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        let (url_type, options) = value.split_at(value.find(":/").ok_or(StationUrlParseError)?);
 
         let options = &options[2..];
 
@@ -100,10 +106,10 @@ impl TryFrom<&str> for StationUrl {
             "udp" => UDP,
             "prudp" => PRUDP,
             "prudps" => PRUDPS,
-            _ => return Err(()),
+            _ => return Err(StationUrlParseError),
         };
 
-        let options = Self::read_options(options).ok_or(())?;
+        let options = Self::read_options(options).ok_or(StationUrlParseError)?;
 
         Ok(Self { url_type, options })
     }