lingxia_observability/
lib.rs

1use lingxia_provider::{BoxFuture, ProviderError};
2use serde::Serialize;
3use std::collections::{HashSet, VecDeque};
4use std::io;
5use std::sync::Mutex;
6use std::time::{SystemTime, UNIX_EPOCH};
7use tokio::sync::broadcast;
8
9/// Default live subscriber capacity for the in-memory log pipeline.
10pub const DEFAULT_LOG_LIVE_CAPACITY: usize = 1024;
11/// Default recent history capacity retained in memory.
12pub const DEFAULT_LOG_HISTORY_CAPACITY: usize = 2048;
13/// Recommended recent replay window for devtools log viewers.
14pub const DEFAULT_DEVTOOLS_RECENT_LIMIT: usize = 500;
15
16/// Log levels that match Android/iOS common levels.
17#[derive(Debug, Clone, Copy, Serialize)]
18#[serde(rename_all = "snake_case")]
19pub enum LogLevel {
20    Verbose,
21    Debug,
22    Info,
23    Warn,
24    Error,
25}
26
27#[derive(Debug, Clone, Copy, Serialize)]
28#[serde(rename_all = "snake_case")]
29pub enum LogTag {
30    Native,
31    WebViewConsole,
32    LxAppServiceConsole,
33}
34
35impl LogTag {
36    pub fn as_str(self) -> &'static str {
37        match self {
38            Self::Native => "Native",
39            Self::WebViewConsole => "JSView",
40            Self::LxAppServiceConsole => "JSService",
41        }
42    }
43}
44
45/// Structured log message forwarded to system loggers and network sinks.
46#[derive(Debug, Clone, Serialize)]
47pub struct LogMessage {
48    pub timestamp_ms: u64,
49    pub tag: LogTag,
50    pub level: LogLevel,
51    pub appid: Option<String>,
52    pub path: Option<String>,
53    pub target: Option<String>,
54    pub message: String,
55}
56
57impl Default for LogMessage {
58    fn default() -> Self {
59        Self {
60            timestamp_ms: 0,
61            tag: LogTag::Native,
62            level: LogLevel::Info,
63            appid: None,
64            path: None,
65            target: None,
66            message: String::new(),
67        }
68    }
69}
70
71impl LogMessage {
72    pub fn new(tag: LogTag, message: impl Into<String>) -> Self {
73        Self {
74            timestamp_ms: now_timestamp_ms(),
75            tag,
76            level: LogLevel::Info,
77            appid: None,
78            path: None,
79            target: None,
80            message: message.into(),
81        }
82    }
83
84    pub fn with_level(mut self, level: LogLevel) -> Self {
85        self.level = level;
86        self
87    }
88
89    pub fn with_appid(mut self, appid: impl Into<String>) -> Self {
90        self.appid = normalize_optional_string(Some(appid.into()));
91        self
92    }
93
94    pub fn with_path(mut self, path: impl Into<String>) -> Self {
95        self.path = normalize_optional_string(Some(path.into()));
96        self
97    }
98
99    pub fn with_target(mut self, target: impl Into<String>) -> Self {
100        self.target = normalize_optional_string(Some(target.into()));
101        self
102    }
103}
104
105/// Compressed in-memory log archive payload.
106#[derive(Debug, Clone)]
107pub struct CollectedLogArchive {
108    pub file_name: String,
109    pub content_type: &'static str,
110    pub encoding: &'static str,
111    pub entry_count: usize,
112    pub lxapp_ids: Vec<String>,
113    pub bytes: Vec<u8>,
114}
115
116/// Metadata returned after a collected log archive has been uploaded.
117#[derive(Debug, Clone)]
118pub struct CollectedLogArchiveInfo {
119    pub file_name: String,
120    pub content_type: &'static str,
121    pub encoding: &'static str,
122    pub entry_count: usize,
123    pub lxapp_ids: Vec<String>,
124}
125
126impl CollectedLogArchive {
127    pub fn from_entries(entries: &[LogMessage]) -> io::Result<Self> {
128        let mut lxapp_ids = Vec::new();
129        let mut seen_lxapp_ids = HashSet::new();
130        let mut jsonl = Vec::new();
131        for entry in entries {
132            if let Some(appid) = entry.appid.as_deref()
133                && seen_lxapp_ids.insert(appid.to_string())
134            {
135                lxapp_ids.push(appid.to_string());
136            }
137            serde_json::to_writer(&mut jsonl, entry)
138                .map_err(|err| io::Error::new(io::ErrorKind::InvalidData, err))?;
139            jsonl.push(b'\n');
140        }
141
142        let bytes = zstd::stream::encode_all(io::Cursor::new(jsonl), 3)?;
143        Ok(Self {
144            file_name: format!("lingxia-logs-{}.jsonl.zst", now_timestamp_ms()),
145            content_type: "application/zstd",
146            encoding: "jsonl+zstd",
147            entry_count: entries.len(),
148            lxapp_ids,
149            bytes,
150        })
151    }
152
153    pub fn info(&self) -> CollectedLogArchiveInfo {
154        CollectedLogArchiveInfo {
155            file_name: self.file_name.clone(),
156            content_type: self.content_type,
157            encoding: self.encoding,
158            entry_count: self.entry_count,
159            lxapp_ids: self.lxapp_ids.clone(),
160        }
161    }
162}
163
164/// Combined recent replay plus live log receiver for devtools and diagnostics.
165pub struct AttachedLogStream {
166    pub recent: Vec<LogMessage>,
167    pub receiver: broadcast::Receiver<LogMessage>,
168}
169
170impl AttachedLogStream {
171    /// Borrow the stitched replay window returned when the stream was attached.
172    pub fn recent(&self) -> &[LogMessage] {
173        &self.recent
174    }
175
176    /// Consume the stream and return `(recent, receiver)` for custom integrations.
177    pub fn into_parts(self) -> (Vec<LogMessage>, broadcast::Receiver<LogMessage>) {
178        (self.recent, self.receiver)
179    }
180
181    /// Receive the next live log item.
182    pub async fn recv(&mut self) -> Result<LogMessage, broadcast::error::RecvError> {
183        self.receiver.recv().await
184    }
185
186    /// Try to receive the next live log item without awaiting.
187    pub fn try_recv(&mut self) -> Result<LogMessage, broadcast::error::TryRecvError> {
188        self.receiver.try_recv()
189    }
190}
191
192#[derive(Debug, Clone, Copy, PartialEq, Eq)]
193pub struct LogBufferConfig {
194    pub live_capacity: usize,
195    pub history_capacity: usize,
196}
197
198impl Default for LogBufferConfig {
199    fn default() -> Self {
200        Self {
201            live_capacity: DEFAULT_LOG_LIVE_CAPACITY,
202            history_capacity: DEFAULT_LOG_HISTORY_CAPACITY,
203        }
204    }
205}
206
207/// Reusable in-memory log pipeline shared by SDK runtimes and tooling bridges.
208pub struct LogBuffer {
209    sender: broadcast::Sender<LogMessage>,
210    history: Mutex<VecDeque<LogMessage>>,
211    config: LogBufferConfig,
212}
213
214impl LogBuffer {
215    pub fn new(config: LogBufferConfig) -> Self {
216        let (sender, _) = broadcast::channel(config.live_capacity.max(1));
217        Self {
218            sender,
219            history: Mutex::new(VecDeque::with_capacity(config.history_capacity.max(1))),
220            config,
221        }
222    }
223
224    pub fn subscribe(&self) -> broadcast::Receiver<LogMessage> {
225        self.sender.subscribe()
226    }
227
228    pub fn attach(&self, recent_limit: usize) -> AttachedLogStream {
229        let history = self
230            .history
231            .lock()
232            .unwrap_or_else(|poisoned| poisoned.into_inner());
233        let receiver = self.sender.subscribe();
234        let recent_limit = clamp_recent_limit(recent_limit, history.len());
235        let recent = history
236            .iter()
237            .skip(history.len().saturating_sub(recent_limit))
238            .cloned()
239            .collect();
240        AttachedLogStream { recent, receiver }
241    }
242
243    pub fn snapshot_recent(&self, limit: usize) -> Vec<LogMessage> {
244        let history = self
245            .history
246            .lock()
247            .unwrap_or_else(|poisoned| poisoned.into_inner());
248        let limit = clamp_recent_limit(limit, history.len());
249        history
250            .iter()
251            .skip(history.len().saturating_sub(limit))
252            .cloned()
253            .collect()
254    }
255
256    pub fn collect_archive(&self, limit: usize) -> io::Result<CollectedLogArchive> {
257        let entries = self.snapshot_recent(limit);
258        CollectedLogArchive::from_entries(&entries)
259    }
260
261    pub fn push(&self, message: LogMessage) {
262        let entry = message.clone();
263        {
264            let mut history = self
265                .history
266                .lock()
267                .unwrap_or_else(|poisoned| poisoned.into_inner());
268            if history.len() >= self.config.history_capacity.max(1) {
269                history.pop_front();
270            }
271            history.push_back(entry);
272        }
273
274        let _ = self.sender.send(message);
275    }
276}
277
278fn clamp_recent_limit(requested: usize, available: usize) -> usize {
279    if requested == 0 {
280        available
281    } else {
282        requested.min(available)
283    }
284}
285
286pub fn normalize_optional_string(value: Option<String>) -> Option<String> {
287    value.and_then(|value| {
288        let trimmed = value.trim();
289        if trimmed.is_empty() {
290            None
291        } else {
292            Some(trimmed.to_string())
293        }
294    })
295}
296
297pub fn now_timestamp_ms() -> u64 {
298    SystemTime::now()
299        .duration_since(UNIX_EPOCH)
300        .unwrap_or_default()
301        .as_millis() as u64
302}
303
304/// Realtime plus diagnostic log upload contract.
305///
306/// # Re-entrancy
307///
308/// `on_log` is called synchronously inside the log dispatch path.
309/// Implementations **must not** emit lingxia log events (e.g. via `info!()` or `tracing`)
310/// from within `on_log`, as this would re-enter the log pipeline.  The SDK guards against
311/// same-thread re-entrancy, but cross-thread re-entrancy is not detected and may cause
312/// unbounded recursion on multi-threaded runtimes.
313pub trait LogProvider: Send + Sync + 'static {
314    /// Realtime log hook.
315    ///
316    /// Called synchronously for every structured log event that enters the SDK log pipeline.
317    /// Implementations are expected to enqueue quickly and avoid blocking I/O.
318    /// **Must not** emit lingxia log events — see trait-level re-entrancy note.
319    fn on_log(&self, _message: &LogMessage) {}
320
321    /// Upload a collected compressed log archive for diagnostics.
322    fn upload_collected_logs<'a>(
323        &'a self,
324        _archive: CollectedLogArchive,
325    ) -> BoxFuture<'a, Result<(), ProviderError>> {
326        Box::pin(async { Ok(()) })
327    }
328}
lingxia_observability/lib.rs

lingxia_observability/
lib.rs
