Webapp

src/engine.rs

@@ -0,0 +1,129 @@
+use crate::data_value::DataValue;
+use crate::dto::{LogEntryDto, LogsParams, StatsDto, TimelineBucketDto, TimelineParams};
+use crate::index_key::IndexKey;
+use crate::timestamp::Timestamp;
+use std::collections::BTreeMap;
+use tokio::sync::{mpsc, oneshot};
+
+pub struct IngestedLine {
+    pub sequence: u64,
+    pub timestamp: Timestamp,
+    pub text: String,
+}
+
+pub enum EngineQuery {
+    Logs(LogsParams),
+    Timeline(TimelineParams),
+    Stats,
+}
+
+pub enum EngineMessage {
+    Ingest(IngestedLine),
+    Query {
+        query: EngineQuery,
+        respond_to: oneshot::Sender<EngineResponse>,
+    },
+}
+
+pub enum EngineResponse {
+    Logs(Vec<LogEntryDto>),
+    Timeline(Vec<TimelineBucketDto>),
+    Stats(StatsDto),
+}
+
+pub type EngineHandle = mpsc::Sender<EngineMessage>;
+
+/// Sole owner of the primary index. Runs single-threaded so the map never needs a
+/// lock, and does its own response serialization/DTO construction so callers never
+/// have to clone the index to answer a query.
+pub async fn run(mut rx: mpsc::Receiver<EngineMessage>) {
+    let mut index: BTreeMap<IndexKey, DataValue> = BTreeMap::new();
+
+    while let Some(message) = rx.recv().await {
+        match message {
+            EngineMessage::Ingest(line) => {
+                index.insert(
+                    IndexKey::IngestedLog {
+                        sequence: line.sequence,
+                        timestamp: line.timestamp,
+                    },
+                    DataValue::LogLiteral { text: line.text },
+                );
+            }
+            EngineMessage::Query { query, respond_to } => {
+                let response = handle_query(&index, query);
+                let _ = respond_to.send(response);
+            }
+        }
+    }
+}
+
+fn logs_range(index: &BTreeMap<IndexKey, DataValue>) -> impl Iterator<Item = (u64, Timestamp, &str)> {
+    index.range(IndexKey::Start..IndexKey::LogParamSplit).filter_map(|(key, value)| {
+        match (key, value) {
+            (IndexKey::IngestedLog { sequence, timestamp }, DataValue::LogLiteral { text }) => {
+                Some((*sequence, *timestamp, text.as_str()))
+            }
+            _ => None,
+        }
+    })
+}
+
+fn handle_query(index: &BTreeMap<IndexKey, DataValue>, query: EngineQuery) -> EngineResponse {
+    match query {
+        EngineQuery::Logs(params) => {
+            let limit = params.limit.unwrap_or(500).min(10_000);
+            let entries: Vec<LogEntryDto> = logs_range(index)
+                .filter(|(sequence, timestamp, _)| {
+                    params.since_sequence.is_none_or(|since| *sequence > since)
+                        && params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
+                        && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)
+                })
+                .take(limit)
+                .map(|(sequence, timestamp, text)| LogEntryDto {
+                    sequence,
+                    timestamp_ms: timestamp.as_millis(),
+                    text: text.to_string(),
+                })
+                .collect();
+            EngineResponse::Logs(entries)
+        }
+        EngineQuery::Timeline(params) => {
+            let bucket_ms = params.bucket_ms.unwrap_or(1_000).max(1);
+            let mut buckets: BTreeMap<i64, u64> = BTreeMap::new();
+            for (_, timestamp, _) in logs_range(index).filter(|(_, timestamp, _)| {
+                params.start_ms.is_none_or(|start| timestamp.as_millis() >= start)
+                    && params.end_ms.is_none_or(|end| timestamp.as_millis() <= end)
+            }) {
+                let bucket_start = (timestamp.as_millis() / bucket_ms) * bucket_ms;
+                *buckets.entry(bucket_start).or_insert(0) += 1;
+            }
+            let out = buckets
+                .into_iter()
+                .map(|(start_ms, count)| TimelineBucketDto { start_ms, count })
+                .collect();
+            EngineResponse::Timeline(out)
+        }
+        EngineQuery::Stats => {
+            let mut log_count = 0u64;
+            let mut first_sequence = None;
+            let mut last_sequence = None;
+            let mut first_timestamp_ms = None;
+            let mut last_timestamp_ms = None;
+            for (sequence, timestamp, _) in logs_range(index) {
+                log_count += 1;
+                first_sequence.get_or_insert(sequence);
+                last_sequence = Some(sequence);
+                first_timestamp_ms.get_or_insert(timestamp.as_millis());
+                last_timestamp_ms = Some(timestamp.as_millis());
+            }
+            EngineResponse::Stats(StatsDto {
+                log_count,
+                first_sequence,
+                last_sequence,
+                first_timestamp_ms,
+                last_timestamp_ms,
+            })
+        }
+    }
+}