ordinary_utils/
events.rs

1// Copyright (C) 2026 Ordinary Labs, LLC.
2//
3// SPDX-License-Identifier: AGPL-3.0-only
4
5#[cfg(tracing_unstable)]
6use crate::json::JsonValuable;
7use async_compression::tokio::write::ZlibDecoder;
8use axum::Router;
9use axum::http::StatusCode;
10use axum::routing::post;
11use bytes::Bytes;
12use serde::{Deserialize, Serialize};
13use std::collections::BTreeMap;
14use tokio::io::AsyncWriteExt;
15#[cfg(tracing_unstable)]
16use valuable::{Mappable, Valuable, Value, Visit};
17
18#[derive(Serialize, Deserialize, Debug)]
19pub struct ClientEvent {
20    /// timestamp
21    pub ts: String,
22    /// timezone
23    pub tz: String,
24    /// log level
25    #[serde(skip_serializing)]
26    pub lvl: String,
27    /// version
28    pub v: String,
29    /// correlation id
30    pub cid: String,
31    /// memory id
32    pub mid: String,
33    /// path
34    pub p: String,
35    #[serde(skip_serializing_if = "Option::is_none")]
36    #[serde(default)]
37    /// query
38    pub q: Option<String>,
39    #[serde(skip_serializing_if = "Option::is_none")]
40    #[serde(default)]
41    /// fragment
42    pub f: Option<String>,
43    #[serde(skip_serializing_if = "Option::is_none")]
44    #[serde(default)]
45    /// flags
46    pub flg: Option<BTreeMap<String, String>>,
47    #[serde(skip_serializing_if = "Option::is_none")]
48    #[serde(default)]
49    /// fields
50    pub fld: Option<BTreeMap<String, serde_json::Value>>,
51    #[serde(skip_serializing)]
52    /// message
53    pub msg: String,
54}
55
56#[cfg(tracing_unstable)]
57impl Mappable for ClientEvent {
58    fn size_hint(&self) -> (usize, Option<usize>) {
59        (6, Some(9))
60    }
61}
62
63#[cfg(tracing_unstable)]
64impl Valuable for ClientEvent {
65    fn as_value(&self) -> Value<'_> {
66        Value::Mappable(self)
67    }
68
69    fn visit(&self, visit: &mut dyn Visit) {
70        visit.visit_entry("ts".as_value(), self.ts.as_value());
71        visit.visit_entry("tz".as_value(), self.tz.as_value());
72
73        visit.visit_entry("v".as_value(), self.v.as_value());
74        visit.visit_entry("mid".as_value(), self.mid.as_value());
75        visit.visit_entry("cid".as_value(), self.cid.as_value());
76
77        visit.visit_entry("p".as_value(), self.p.as_value());
78        if let Some(q) = &self.q {
79            visit.visit_entry("q".as_value(), q.as_value());
80        }
81        if let Some(f) = &self.f {
82            visit.visit_entry("f".as_value(), f.as_value());
83        }
84        if let Some(flg) = &self.flg {
85            visit.visit_entry("flg".as_value(), flg.as_value());
86        }
87        if let Some(fld) = &self.fld {
88            for (fld_nm, fld_v) in fld {
89                if !fld_v.is_null() {
90                    visit.visit_entry(fld_nm.as_value(), JsonValuable(fld_v.to_owned()).as_value());
91                }
92            }
93        }
94    }
95}
96
97#[must_use]
98pub fn setup_routes<S>(client_events: Option<&bool>) -> Option<Router<S>>
99where
100    S: Clone + Send + Sync + 'static,
101{
102    if client_events == Some(&true) {
103        let router = Router::new().route(
104            "/.ordinary/v1/events",
105            post(|body: Bytes| async move {
106                let mut decoder = ZlibDecoder::new(Vec::new());
107                if let Err(err) = decoder.write_all(body.as_ref()).await {
108                    tracing::error!(%err);
109                    return StatusCode::INTERNAL_SERVER_ERROR;
110                }
111                if let Err(err) = decoder.shutdown().await {
112                    tracing::error!(%err);
113                    return StatusCode::INTERNAL_SERVER_ERROR;
114                }
115
116                let mut events_slice = decoder.into_inner();
117
118                let events: Vec<ClientEvent> = match simd_json::from_slice(&mut events_slice) {
119                    Ok(parsed) => parsed,
120                    Err(err) => {
121                        tracing::error!(%err);
122                        return StatusCode::INTERNAL_SERVER_ERROR;
123                    }
124                };
125
126                let client_span = tracing::info_span!("client");
127
128                client_span.in_scope(|| {
129                    for event in events {
130                        #[cfg(tracing_unstable)]
131                        {
132                            let level = event.lvl.as_str();
133                            let message = event.msg.as_str();
134
135                            match level {
136                                "TRACE" => tracing::trace!(
137                                    _f = tracing::field::valuable(&event),
138                                    "{message}"
139                                ),
140                                "DEBUG" => tracing::debug!(
141                                    _f = tracing::field::valuable(&event),
142                                    "{message}"
143                                ),
144                                "INFO" => {
145                                    tracing::info!(
146                                        _f = tracing::field::valuable(&event),
147                                        "{message}"
148                                    );
149                                }
150                                "WARN" => {
151                                    tracing::warn!(
152                                        _f = tracing::field::valuable(&event),
153                                        "{message}"
154                                    );
155                                }
156                                _ => tracing::error!(
157                                    _f = tracing::field::valuable(&event),
158                                    "{message}"
159                                ),
160                            }
161                        }
162
163                        #[cfg(not(tracing_unstable))]
164                        {
165                            match event.lvl.as_ref() {
166                                "TRACE" => tracing::trace!(evt = debug(&event)),
167                                "DEBUG" => tracing::debug!(evt = debug(&event)),
168                                "INFO" => tracing::info!(evt = debug(&event)),
169                                "WARN" => tracing::warn!(evt = debug(&event)),
170                                _ => tracing::error!(evt = debug(&event)),
171                            }
172                        }
173                    }
174                });
175
176                StatusCode::OK
177            }),
178        );
179        return Some(router);
180    }
181
182    None
183}
ordinary_utils/events.rs

ordinary_utils/
events.rs
