refactor for consistency, hexagonal architecture clean-up

Cargo.toml

@@ -26,8 +26,28 @@ license = "MIT OR Apache-2.0"
 repository = "https://github.com/tensaur/photon"
 publish = false
 
+[workspace.lints.rust]
+unreachable_pub = "warn"
+
+[workspace.lints.clippy]
+all = { level = "warn", priority = -1 }
+pedantic = { level = "warn", priority = -1 }
+module_name_repetitions = "allow"
+must_use_candidate = "allow"
+return_self_not_must_use = "allow"
+missing_errors_doc = "allow"
+missing_panics_doc = "allow"
+cast_possible_truncation = "allow"
+cast_precision_loss = "allow"
+cast_sign_loss = "allow"
+cast_possible_wrap = "allow"
+too_many_arguments = "allow"
+too_many_lines = "allow"
+type_complexity = "allow"
+
 [workspace.dependencies]
-tokio = { version = "1", features = ["full"] }
+tokio = { version = "1", features = ["rt", "sync", "time", "macros", "signal"] }
+tokio-util = { version = "0.7", features = ["rt"] }
 thiserror = "2"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
@@ -65,7 +85,6 @@ wasm-bindgen-futures = "0.4"
 tracing-web = "0.1"
 httparse = "1"
 async-trait = "0.1"
-dyn-clone = "1"
 clickhouse = { version = "0.13", features = ["uuid", "inserter"] }
 criterion = { version = "0.5", features = ["html_reports"] }
 

crates/batch/Cargo.toml

@@ -27,3 +27,6 @@ criterion.workspace = true
 [[bench]]
 name = "service"
 harness = false
+
+[lints]
+workspace = true

crates/batch/benches/service.rs

@@ -6,7 +6,7 @@ use lasso::ThreadedRodeo;
 use photon_batch::domain::service::{BatchService, Service};
 use photon_batch::domain::types::BatchStats;
 use photon_core::types::id::RunId;
-use photon_core::types::metric::{MetricKey, RawPoint};
+use photon_core::types::metric::{MetricKey, RawPoint, Step};
 use photon_core::types::sequence::SequenceNumber;
 use photon_protocol::codec::CodecKind;
 use photon_protocol::compressor::CompressorKind;
@@ -35,7 +35,7 @@ fn make_raw_points(interner: &Arc<ThreadedRodeo>, n: usize) -> Vec<RawPoint> {
         .map(|i| RawPoint {
             key: keys[i % num_keys],
             value: 1.0 / (1.0 + i as f64 * 0.001),
-            step: i as u64,
+            step: Step::new(i as u64),
             timestamp_ns: i as u64 * 1_000_000,
         })
         .collect()

crates/batch/src/domain/assembler.rs

@@ -14,7 +14,7 @@ pub(crate) struct BatchAssembler {
 }
 
 impl BatchAssembler {
-    pub fn new(interner: Arc<ThreadedRodeo>) -> Self {
+    pub(crate) fn new(interner: Arc<ThreadedRodeo>) -> Self {
         Self {
             interner,
             key_map: HashMap::new(),
@@ -23,7 +23,7 @@ impl BatchAssembler {
         }
     }
 
-    pub fn assemble(&mut self, run_id: RunId, pending: &[RawPoint]) -> MetricBatch {
+    pub(crate) fn assemble(&mut self, run_id: RunId, pending: &[RawPoint]) -> MetricBatch {
         self.key_map.clear();
         self.keys.clear();
         self.points.clear();
@@ -52,7 +52,7 @@ impl BatchAssembler {
         }
     }
 
-    pub fn reclaim(&mut self, batch: MetricBatch) {
+    pub(crate) fn reclaim(&mut self, batch: MetricBatch) {
         self.keys = batch.keys;
         self.points = batch.points;
     }

crates/batch/src/inbound/thread.rs

@@ -15,6 +15,7 @@ use photon_wal::WalAppender;
 use crate::domain::service::{BatchError, BatchService, Service};
 use crate::domain::types::BatchStats;
 
+#[allow(clippy::needless_pass_by_value)]
 pub fn run_batch_thread<K, C, A>(
     run_id: RunId,
     interner: Arc<ThreadedRodeo>,
@@ -39,32 +40,29 @@ where
     loop {
         select! {
             recv(rx) -> msg => {
-                match msg {
-                    Ok(point) => {
-                        pending.push(point);
+                if let Ok(point) = msg {
+                    pending.push(point);
 
-                        while pending.len() < config.max_points {
-                            match rx.try_recv() {
-                                Ok(p) => pending.push(p),
-                                Err(_) => break,
-                            }
-                        }
-
-                        if pending.len() >= config.max_points {
-                            let wire = service.batch(&pending, &mut stats)?;
-                            let _ = batch_tx.send(wire);
-                            pending.clear();
+                    while pending.len() < config.max_points {
+                        match rx.try_recv() {
+                            Ok(p) => pending.push(p),
+                            Err(_) => break,
                         }
                     }
-                    Err(_) => {
-                        if !pending.is_empty() {
-                            let wire = service.batch(&pending, &mut stats)?;
-                            let _ = batch_tx.send(wire);
-                            pending.clear();
-                        }
 
-                        return Ok(stats);
+                    if pending.len() >= config.max_points {
+                        let wire = service.batch(&pending, &mut stats)?;
+                        let _ = batch_tx.send(wire);
+                        pending.clear();
                     }
+                } else {
+                    if !pending.is_empty() {
+                        let wire = service.batch(&pending, &mut stats)?;
+                        let _ = batch_tx.send(wire);
+                        pending.clear();
+                    }
+
+                    return Ok(stats);
                 }
             }
 

crates/core/Cargo.toml

@@ -16,3 +16,6 @@ uuid.workspace = true
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 uuid = { workspace = true, features = ["js"] }
+
+[lints]
+workspace = true

crates/core/src/domain/mod.rs

@@ -2,3 +2,22 @@ pub mod experiment;
 pub mod project;
 pub mod run;
 pub mod tenant;
+
+use crate::types::id::{ExperimentId, ProjectId, RunId};
+
+/// Marker trait for domain entities with a typed identifier.
+pub trait Entity: Send + 'static {
+    type Id: Send + Sync;
+}
+
+impl Entity for run::Run {
+    type Id = RunId;
+}
+
+impl Entity for project::Project {
+    type Id = ProjectId;
+}
+
+impl Entity for experiment::Experiment {
+    type Id = ExperimentId;
+}