feat: implement get logs route + log body payload to clickhouse

Author: kioqq

Diff for: Cargo.lock

@@ -35,6 +35,8 @@ thiserror = "2.0.11"
 confik = "0.12.0"
 
 ## json
+serde = { version = "1.0.217", features = ["derive"] }
+serde_json = "1.0.137"
 sonic-rs = "0.3.17"
 
 ## another

Diff for: Cargo.toml

@@ -17,7 +17,7 @@ impl ClickhouseLogger {
     }
 
     pub async fn log(&self, log: WebServerLog) -> Result<()> {
-        println!("Logging to Clickhouse: {:?}", log);
+        tracing::debug!("Logging to Clickhouse: {:?}", log);
 
         let connection = self
             .clickhouse_pool
@@ -31,7 +31,7 @@ impl ClickhouseLogger {
             .map_err(|e| eyre!("Failed to write log to ClickHouse: {}", e));
 
         if let Err(e) = result {
-            println!("Error: {:?}", e);
+            tracing::error!("Error: {:?}", e);
         }
 
         Ok(())

Diff for: src/clickhouse/logger.rs

@@ -97,7 +97,7 @@ impl MigrationManager {
                 let migration = self.migrations.get(&version).unwrap();
 
                 // Log migration start
-                println!("Running migration v{}: {}", version, migration.name());
+                tracing::info!("Running migration v{}: {}", version, migration.name());
 
                 // Run migration
                 migration.apply(pool).await?;
@@ -116,7 +116,7 @@ impl MigrationManager {
             }
         }
 
-        println!("All migrations applied successfully!");
+        tracing::info!("All migrations applied successfully!");
         Ok(())
     }
 

Diff for: src/clickhouse/migrations/mod.rs

@@ -1,7 +1,8 @@
 use chrono::Utc;
 use klickhouse::Row;
+use serde::{Deserialize, Serialize};
 
-#[derive(Row, Debug)]
+#[derive(Row, Debug, Serialize, Deserialize)]
 pub struct WebServerLog {
     pub timestamp: chrono::DateTime<Utc>,
     pub level: String,

Diff for: src/clickhouse/models/web_server_log.rs

@@ -3,6 +3,7 @@ use klickhouse::*;
 
 use super::config::ClickhouseConfig;
 use super::migrations::MigrationManager;
+use super::queries::LogsQueries;
 
 // Queries
 // use super::queries::trades_queries::TradesQueries;
@@ -51,7 +52,7 @@ impl ClickhousePool {
             .map_err(|e| eyre!("Failed to check Clickhouse connection: {}", e))?;
 
         if result.0 == 1 {
-            println!("Clickhouse connection is getting ok");
+            tracing::info!("Clickhouse connection is getting ok");
         } else {
             return Err(eyre!("Failed to check Clickhouse connection"));
         }
@@ -67,10 +68,11 @@ impl ClickhousePool {
         Ok(())
     }
 
-    // pub async fn get_trades_queries(&self) -> Result<TradesQueries> {
-    //     let connection = self.get_connection().await?;
-    //     Ok(TradesQueries::new(connection))
-    // }
+    // Queries
+    pub async fn get_logs_queries(&self) -> Result<LogsQueries> {
+        let connection = self.get_connection().await?;
+        Ok(LogsQueries::new(connection))
+    }
 }
 
 #[cfg(test)]
@@ -88,7 +90,6 @@ mod tests {
         pool.check_pool().await.unwrap();
 
         let version = MigrationManager::get_current_version(&pool).await.unwrap();
-        println!("Current version: {}", version);
         assert!(version > 0);
     }
 }

Diff for: src/clickhouse/pool.rs

@@ -0,0 +1,33 @@
+use eyre::Result;
+
+use crate::clickhouse::models::WebServerLog;
+use crate::clickhouse::pool::ClickhouseConnection;
+
+pub struct LogsQueries<'a> {
+    conn: ClickhouseConnection<'a>,
+}
+
+impl<'a> LogsQueries<'a> {
+    pub fn new(conn: ClickhouseConnection<'a>) -> Self {
+        Self { conn }
+    }
+
+    pub async fn get_logs(&self, limit: usize, offset: usize) -> Result<Vec<WebServerLog>> {
+        let query = format!(
+            "SELECT * FROM web_server_logs ORDER BY timestamp DESC LIMIT {} OFFSET {}",
+            limit, offset
+        );
+        let logs = match self.conn.query_collect::<WebServerLog>(query).await {
+            Ok(logs) => {
+                tracing::debug!("Fetched {} logs", logs.len());
+                logs
+            }
+            Err(e) => {
+                tracing::error!("Error fetching logs: {}", e);
+                return Err(e.into());
+            }
+        };
+
+        Ok(logs)
+    }
+}

Diff for: src/clickhouse/queries/logs_query.rs

@@ -1 +1,3 @@
+pub mod logs_query;
 
+pub use logs_query::LogsQueries;

Diff for: src/clickhouse/queries/mod.rs

@@ -1,4 +1,5 @@
 pub mod cli;
 pub mod clickhouse;
 pub mod config;
+pub mod services;
 pub mod web;

Diff for: src/lib.rs

@@ -0,0 +1,20 @@
+use crate::clickhouse::{models::WebServerLog, ClickhousePool};
+use eyre::{eyre, Result};
+
+pub struct GetWebLogsService {}
+
+impl GetWebLogsService {
+    pub async fn get_logs(
+        clickhouse_pool: &ClickhousePool,
+        limit: usize,
+        offset: usize,
+    ) -> Result<Vec<WebServerLog>> {
+        // setup default values
+        let logs_queries = clickhouse_pool.get_logs_queries().await?;
+
+        match logs_queries.get_logs(limit, offset).await {
+            Ok(logs) => Ok(logs),
+            Err(e) => Err(eyre!("Error getting logs: {}", e)),
+        }
+    }
+}

Diff for: src/services/get_web_logs_service.rs

@@ -0,0 +1,3 @@
+pub mod get_web_logs_service;
+
+pub use get_web_logs_service::GetWebLogsService;

Diff for: src/services/mod.rs

@@ -41,6 +41,10 @@ impl AppState {
         self.clickhouse_pool.check_pool().await
     }
 
+    pub fn clickhouse_pool(&self) -> &ClickhousePool {
+        &self.clickhouse_pool
+    }
+
     pub fn config(&self) -> &AppConfig {
         &self.config
     }

Diff for: src/web/app_state.rs

@@ -4,7 +4,7 @@ use std::panic;
 
 pub fn setup_global_panic_handler(app_state: Data<AppState>) {
     panic::set_hook(Box::new(move |info| {
-        println!("Global panic handler called");
+        tracing::error!("Global panic handler called");
         let app_state = app_state.clone();
 
         let location = info
@@ -32,7 +32,7 @@ pub fn setup_global_panic_handler(app_state: Data<AppState>) {
 
             // Write log to Clickhouse
             if let Err(e) = app_state.ch_logger().log(log).await {
-                eprintln!("Failed to log panic to ClickHouse: {:?}", e);
+                tracing::error!("Failed to log panic to ClickHouse: {:?}", e);
             }
         });
     }));

Diff for: src/web/global_panic_handler.rs

@@ -0,0 +1,28 @@
+use crate::web::app_state::AppState;
+use actix_web::{get, web, HttpResponse, Responder};
+use serde::Deserialize;
+
+use crate::services::GetWebLogsService;
+
+#[derive(Deserialize)]
+pub struct WebLogsQuery {
+    pub limit: Option<usize>,  // Limit of logs to return, default 10
+    pub offset: Option<usize>, // Offset of logs to return, default 0
+}
+
+#[get("/logs")]
+pub async fn get_logs(
+    app_state: web::Data<AppState>,
+    query: web::Query<WebLogsQuery>,
+) -> impl Responder {
+    let clickhouse_pool = app_state.clickhouse_pool();
+
+    // setup default values
+    let limit = query.limit.unwrap_or(10);
+    let offset = query.offset.unwrap_or(0);
+
+    match GetWebLogsService::get_logs(clickhouse_pool, limit, offset).await {
+        Ok(logs) => HttpResponse::Ok().json(logs),
+        Err(e) => HttpResponse::InternalServerError().json(e.to_string()),
+    }
+}

Diff for: src/web/handlers/get_logs_handler.rs

@@ -1,3 +1,7 @@
+pub mod get_logs_handler;
+
+pub use get_logs_handler::get_logs;
+
 use crate::web::app_state::AppState;
 use actix_web::{get, web, HttpResponse, Responder};
 use tracing_actix_web::RequestId;

Diff for: src/web/handlers/mod.rs

@@ -1,6 +1,8 @@
 use actix_web::{
+    body::{BoxBody, MessageBody},
     dev::{Service, ServiceRequest, ServiceResponse, Transform},
-    Error, HttpMessage,
+    web::Bytes,
+    Error, HttpMessage, HttpResponse,
 };
 use futures::future::{ok, Ready};
 use std::{
@@ -20,10 +22,9 @@ pub struct ClickhouseLogger;
 impl<S, B> Transform<S, ServiceRequest> for ClickhouseLogger
 where
     S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
-    S::Future: 'static,
-    B: 'static,
+    B: MessageBody + 'static,
 {
-    type Response = ServiceResponse<B>;
+    type Response = ServiceResponse<BoxBody>;
     type Error = Error;
     type Transform = ClickhouseLoggerMiddleware<S>;
     type InitError = ();
@@ -43,10 +44,9 @@ pub struct ClickhouseLoggerMiddleware<S> {
 impl<S, B> Service<ServiceRequest> for ClickhouseLoggerMiddleware<S>
 where
     S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
-    S::Future: 'static,
-    B: 'static,
+    B: MessageBody + 'static,
 {
-    type Response = ServiceResponse<B>;
+    type Response = ServiceResponse<BoxBody>;
     type Error = Error;
     type Future = Pin<Box<dyn std::future::Future<Output = Result<Self::Response, Self::Error>>>>;
 
@@ -75,33 +75,54 @@ where
                     match result {
                         Ok(res) => {
                             // Успешный запрос
+                            let (req_head, http_resp) = res.into_parts();
+
                             let duration = start_time.elapsed();
-                            let request_id = res
-                                .request()
+                            let request_id = req_head
                                 .extensions()
                                 .get::<RequestId>()
                                 .map(|id| id.to_string())
                                 .unwrap_or_default();
 
-                            let uri = res.request().uri().to_string();
-                            let method = res.request().method().to_string();
-                            let status_code = res.status().as_u16() as i32;
+                            let uri = req_head.uri().to_string();
+                            let method = req_head.method().to_string();
+                            let status_code = http_resp.status();
+
+                            let body_bytes_result =
+                                actix_web::body::to_bytes(http_resp.into_body()).await;
+                            let message = match body_bytes_result {
+                                Ok(ref bytes) => format!(
+                                    "Request processed. Body: {}",
+                                    String::from_utf8_lossy(bytes)
+                                ),
+                                Err(_) => "Request processed. Failed to read body.".to_string(),
+                            };
 
                             let log = WebServerLog {
                                 timestamp: chrono::Utc::now(),
                                 level: "INFO".to_string(),
-                                message: "Request processed".to_string(),
+                                message,
                                 module: "web_server".to_string(),
                                 request_id,
                                 uri,
                                 method,
-                                status_code,
+                                status_code: status_code.as_u16() as i32,
                                 response_time: duration.as_secs_f64(),
                             };
 
                             let _ = app_state.ch_logger().log(log).await;
 
-                            Ok(res)
+                            let new_body = match body_bytes_result {
+                                Ok(ref b) => b.clone(),
+                                Err(_) => Bytes::new(),
+                            };
+
+                            let new_http_response = HttpResponse::build(status_code).body(new_body);
+                            let new_srv_response =
+                                ServiceResponse::new(req_head, new_http_response)
+                                    .map_into_boxed_body();
+
+                            Ok(new_srv_response)
                         }
                         Err(e) => {
                             // Ошибка от следующего middleware/handler

Diff for: src/web/middleware/clickhouse_logger.rs

@@ -1,5 +1,5 @@
 use actix_web::middleware::NormalizePath;
-use actix_web::{web, App, HttpServer};
+use actix_web::{web, App, HttpResponse, HttpServer};
 use eyre::Result;
 use tracing_actix_web::TracingLogger;
 
@@ -10,7 +10,8 @@ use crate::web::middleware::ClickhouseLogger;
 
 use crate::web::app_state::AppState;
 use crate::web::global_panic_handler::setup_global_panic_handler;
-use crate::web::handlers::{fail_endpoint, health, index};
+// handlers
+use crate::web::handlers::{fail_endpoint, get_logs, health, index};
 
 pub async fn run_serve(config: AppConfig) -> Result<actix_web::dev::Server> {
     let app_state = AppState::build(config).await?;
@@ -20,20 +21,25 @@ pub async fn run_serve(config: AppConfig) -> Result<actix_web::dev::Server> {
     let port = app_state.config().http_port;
     let addr = format!("0.0.0.0:{}", port);
 
-    let server = HttpServer::new(move || {
-        App::new()
-            .app_data(data_app_state.clone())
-            .wrap(TracingLogger::default())
-            .wrap(NormalizePath::new(
-                actix_web::middleware::TrailingSlash::Trim,
-            ))
-            .wrap(ClickhouseLogger)
-            .service(index)
-            .service(health)
-            .service(fail_endpoint)
-    })
-    .bind(addr)?
-    .run();
+    let server =
+        HttpServer::new(move || {
+            App::new()
+                .app_data(data_app_state.clone())
+                .wrap(TracingLogger::default())
+                .wrap(NormalizePath::new(
+                    actix_web::middleware::TrailingSlash::Trim,
+                ))
+                .wrap(ClickhouseLogger)
+                .service(index)
+                .service(health)
+                .service(fail_endpoint)
+                .service(get_logs)
+                .default_service(web::route().to(|| async {
+                    HttpResponse::NotFound().body("Oops! This route does not exist.")
+                }))
+        })
+        .bind(addr)?
+        .run();
 
     Ok(server)
 }