cc_log_viewer/
lib.rs

1// ABOUTME: Claude Code log viewer library - Core functionality for viewing and streaming JSONL logs
2// ABOUTME: Exposes types and handlers for real-time WebSocket monitoring and rich tool rendering
3
4use axum::{
5    extract::{
6        ws::{Message, WebSocket},
7        Path, State, WebSocketUpgrade,
8    },
9    http::StatusCode,
10    response::{Html, Json, Response},
11};
12use chrono::{DateTime, Utc};
13use dashmap::DashMap;
14use futures_util::{SinkExt, StreamExt};
15use notify::{Event, EventKind, RecommendedWatcher, RecursiveMode, Watcher};
16use serde::{Deserialize, Serialize};
17use serde_json::Value;
18use std::{fs, path::PathBuf, sync::Arc, time::SystemTime};
19use tokio::sync::broadcast;
20use walkdir::WalkDir;
21
22#[derive(Debug, Clone, Serialize, Deserialize)]
23pub struct LogEntry {
24    #[serde(rename = "type")]
25    pub entry_type: Option<String>,
26    pub summary: Option<String>,
27    #[serde(rename = "parentUuid")]
28    pub parent_uuid: Option<String>,
29    #[serde(rename = "isSidechain")]
30    pub is_sidechain: Option<bool>,
31    #[serde(rename = "userType")]
32    pub user_type: Option<String>,
33    pub cwd: Option<String>,
34    #[serde(rename = "sessionId")]
35    pub session_id: Option<String>,
36    pub version: Option<String>,
37    pub message: Option<Value>,
38    pub uuid: Option<String>,
39    pub timestamp: Option<DateTime<Utc>>,
40    #[serde(rename = "requestId")]
41    pub request_id: Option<String>,
42    #[serde(rename = "leafUuid")]
43    pub leaf_uuid: Option<String>,
44    #[serde(rename = "toolUseResult")]
45    pub tool_use_result: Option<Value>,
46}
47
48#[derive(Debug, Clone, Serialize)]
49pub struct ProjectSummary {
50    pub name: String,
51    pub path: String,
52    pub session_count: usize,
53    pub latest_activity: Option<DateTime<Utc>>,
54}
55
56#[derive(Debug, Clone, Serialize)]
57pub struct SessionSummary {
58    pub id: String,
59    pub summary: String,
60    pub timestamp: DateTime<Utc>,
61    pub message_count: usize,
62    pub project_name: String,
63}
64
65#[derive(Debug, Clone, Serialize)]
66pub struct WatchEvent {
67    #[serde(rename = "type")]
68    pub event_type: String,
69    pub project: String,
70    pub session: Option<String>,
71    pub entry: Option<LogEntry>,
72    pub timestamp: DateTime<Utc>,
73}
74
75#[derive(Debug, Clone)]
76#[allow(dead_code)]
77pub struct SessionState {
78    pub project_name: String,
79    pub session_file: PathBuf,
80    pub last_position: u64,
81    pub last_modified: SystemTime,
82}
83
84#[derive(Debug)]
85#[allow(dead_code)]
86pub struct WatchManager {
87    _watcher: RecommendedWatcher,
88    active_sessions: Arc<DashMap<String, SessionState>>,
89    broadcast_tx: broadcast::Sender<WatchEvent>,
90    projects_dir: PathBuf,
91}
92
93impl WatchManager {
94    pub fn new(projects_dir: PathBuf) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
95        let (broadcast_tx, _) = broadcast::channel(1000);
96        let active_sessions = Arc::new(DashMap::new());
97
98        let tx_clone = broadcast_tx.clone();
99        let sessions_clone = active_sessions.clone();
100        let projects_dir_clone = projects_dir.clone();
101
102        let mut watcher = notify::recommended_watcher(move |res: Result<Event, notify::Error>| {
103            if let Ok(event) = res {
104                if let Err(e) =
105                    Self::handle_fs_event(event, &tx_clone, &sessions_clone, &projects_dir_clone)
106                {
107                    eprintln!("Error handling file system event: {}", e);
108                }
109            }
110        })?;
111
112        watcher.watch(&projects_dir, RecursiveMode::Recursive)?;
113
114        Ok(WatchManager {
115            _watcher: watcher,
116            active_sessions,
117            broadcast_tx,
118            projects_dir,
119        })
120    }
121
122    fn handle_fs_event(
123        event: Event,
124        broadcast_tx: &broadcast::Sender<WatchEvent>,
125        active_sessions: &DashMap<String, SessionState>,
126        _projects_dir: &std::path::Path,
127    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
128        match event.kind {
129            EventKind::Create(_) | EventKind::Modify(_) => {
130                for path in event.paths {
131                    if path.extension().is_some_and(|ext| ext == "jsonl") {
132                        if let Some(project_name) = path
133                            .parent()
134                            .and_then(|p| p.file_name())
135                            .and_then(|n| n.to_str())
136                        {
137                            let session_id = path
138                                .file_stem()
139                                .and_then(|s| s.to_str())
140                                .unwrap_or("unknown")
141                                .to_string();
142
143                            // Read new entries from the file
144                            if let Ok(metadata) = fs::metadata(&path) {
145                                let key = format!("{}:{}", project_name, session_id);
146                                let current_pos =
147                                    if let Some(session_state) = active_sessions.get(&key) {
148                                        session_state.last_position
149                                    } else {
150                                        0
151                                    };
152
153                                if let Ok(entries) = Self::read_new_entries(&path, current_pos) {
154                                    // Update session state
155                                    active_sessions.insert(
156                                        key,
157                                        SessionState {
158                                            project_name: project_name.to_string(),
159                                            session_file: path.clone(),
160                                            last_position: metadata.len(),
161                                            last_modified: metadata
162                                                .modified()
163                                                .unwrap_or(SystemTime::now()),
164                                        },
165                                    );
166
167                                    // Broadcast new entries (limit to prevent spam)
168                                    let max_entries_per_event = 10;
169                                    for entry in entries.into_iter().take(max_entries_per_event) {
170                                        let watch_event = WatchEvent {
171                                            event_type: "log_entry".to_string(),
172                                            project: project_name.to_string(),
173                                            session: Some(session_id.clone()),
174                                            entry: Some(entry),
175                                            timestamp: Utc::now(),
176                                        };
177
178                                        if broadcast_tx.send(watch_event).is_err() {
179                                            // Channel is closed, stop trying to send
180                                            break;
181                                        }
182                                    }
183                                }
184                            }
185                        }
186                    }
187                }
188            }
189            _ => {}
190        }
191        Ok(())
192    }
193
194    fn read_new_entries(
195        path: &PathBuf,
196        from_position: u64,
197    ) -> Result<Vec<LogEntry>, Box<dyn std::error::Error + Send + Sync>> {
198        // Handle potential file access errors gracefully
199        let content = match fs::read_to_string(path) {
200            Ok(content) => content,
201            Err(e) => {
202                eprintln!("Warning: Could not read file {}: {}", path.display(), e);
203                return Ok(Vec::new());
204            }
205        };
206
207        let mut entries = Vec::new();
208        let mut current_pos = 0u64;
209
210        for line in content.lines() {
211            let line_end = current_pos + line.len() as u64 + 1; // +1 for newline
212
213            if current_pos >= from_position {
214                // Only parse lines that look like JSON to avoid errors
215                if line.trim().starts_with('{') && line.trim().ends_with('}') {
216                    if let Ok(entry) = serde_json::from_str::<LogEntry>(line) {
217                        entries.push(entry);
218                    }
219                }
220            }
221
222            current_pos = line_end;
223        }
224
225        Ok(entries)
226    }
227
228    pub fn subscribe(&self) -> broadcast::Receiver<WatchEvent> {
229        self.broadcast_tx.subscribe()
230    }
231}
232
233#[derive(Debug, Clone)]
234pub struct AppState {
235    pub projects_dir: PathBuf,
236    pub cached_projects: Arc<tokio::sync::RwLock<Vec<ProjectSummary>>>,
237    pub watch_manager: Arc<WatchManager>,
238}
239
240impl AppState {
241    pub fn new(projects_dir: PathBuf) -> Result<Self, Box<dyn std::error::Error + Send + Sync>> {
242        let watch_manager = Arc::new(WatchManager::new(projects_dir.clone())?);
243
244        Ok(Self {
245            projects_dir,
246            cached_projects: Arc::new(tokio::sync::RwLock::new(Vec::new())),
247            watch_manager,
248        })
249    }
250
251    async fn refresh_cache(&self) -> Result<(), Box<dyn std::error::Error>> {
252        let mut projects = Vec::new();
253
254        for entry in WalkDir::new(&self.projects_dir).min_depth(1).max_depth(1) {
255            let entry = entry?;
256            if entry.file_type().is_dir() {
257                let project_name = entry.file_name().to_string_lossy().to_string();
258                let project_path = entry.path().to_string_lossy().to_string();
259
260                let mut session_count = 0;
261                let mut latest_activity: Option<DateTime<Utc>> = None;
262
263                for log_entry in WalkDir::new(entry.path()).min_depth(1).max_depth(1) {
264                    let log_entry = log_entry?;
265                    if log_entry.file_type().is_file()
266                        && log_entry
267                            .path()
268                            .extension()
269                            .is_some_and(|ext| ext == "jsonl")
270                    {
271                        session_count += 1;
272
273                        if let Ok(content) = fs::read_to_string(log_entry.path()) {
274                            for line in content.lines().take(5) {
275                                if let Ok(entry) = serde_json::from_str::<LogEntry>(line) {
276                                    if let Some(timestamp) = entry.timestamp {
277                                        match latest_activity {
278                                            None => latest_activity = Some(timestamp),
279                                            Some(latest) if timestamp > latest => {
280                                                latest_activity = Some(timestamp)
281                                            }
282                                            _ => {}
283                                        }
284                                    }
285                                }
286                            }
287                        }
288                    }
289                }
290
291                projects.push(ProjectSummary {
292                    name: project_name,
293                    path: project_path,
294                    session_count,
295                    latest_activity,
296                });
297            }
298        }
299
300        projects.sort_by(|a, b| b.latest_activity.cmp(&a.latest_activity));
301
302        *self.cached_projects.write().await = projects;
303        Ok(())
304    }
305}
306
307pub async fn index() -> Html<&'static str> {
308    Html(include_str!("../static/index.html"))
309}
310
311pub async fn live_activity() -> Html<&'static str> {
312    Html(include_str!("../static/live.html"))
313}
314
315pub async fn get_projects(
316    State(state): State<AppState>,
317) -> Result<Json<Vec<ProjectSummary>>, StatusCode> {
318    if let Err(e) = state.refresh_cache().await {
319        eprintln!("Failed to refresh project cache: {}", e);
320        return Err(StatusCode::INTERNAL_SERVER_ERROR);
321    }
322
323    let projects = state.cached_projects.read().await;
324    Ok(Json(projects.clone()))
325}
326
327pub async fn get_sessions(
328    Path(project_name): Path<String>,
329    State(state): State<AppState>,
330) -> Result<Json<Vec<SessionSummary>>, StatusCode> {
331    let project_path = state.projects_dir.join(&project_name);
332
333    if !project_path.exists() {
334        return Err(StatusCode::NOT_FOUND);
335    }
336
337    let mut sessions = Vec::new();
338
339    for entry in WalkDir::new(&project_path).min_depth(1).max_depth(1) {
340        let entry = entry.map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
341        if entry.file_type().is_file() && entry.path().extension().is_some_and(|ext| ext == "jsonl")
342        {
343            let session_id = entry
344                .path()
345                .file_stem()
346                .unwrap_or_default()
347                .to_string_lossy()
348                .to_string();
349
350            if let Ok(content) = fs::read_to_string(entry.path()) {
351                let mut summary = "Untitled Session".to_string();
352                let mut timestamp = Utc::now();
353                let message_count = content.lines().count();
354
355                for line in content.lines().take(10) {
356                    if let Ok(entry) = serde_json::from_str::<LogEntry>(line) {
357                        if entry.entry_type.as_deref() == Some("summary") {
358                            if let Some(s) = entry.summary {
359                                summary = s;
360                            }
361                        }
362                        if let Some(ts) = entry.timestamp {
363                            timestamp = ts;
364                            break;
365                        }
366                    }
367                }
368
369                sessions.push(SessionSummary {
370                    id: session_id,
371                    summary,
372                    timestamp,
373                    message_count,
374                    project_name: project_name.clone(),
375                });
376            }
377        }
378    }
379
380    sessions.sort_by(|a, b| b.timestamp.cmp(&a.timestamp));
381    Ok(Json(sessions))
382}
383
384pub async fn get_session_logs(
385    Path((project_name, session_id)): Path<(String, String)>,
386    State(state): State<AppState>,
387) -> Result<Json<Vec<LogEntry>>, StatusCode> {
388    let log_path = state
389        .projects_dir
390        .join(&project_name)
391        .join(format!("{}.jsonl", session_id));
392
393    if !log_path.exists() {
394        return Err(StatusCode::NOT_FOUND);
395    }
396
397    let content = fs::read_to_string(&log_path).map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)?;
398
399    let mut entries = Vec::new();
400    for line in content.lines() {
401        if let Ok(entry) = serde_json::from_str::<LogEntry>(line) {
402            entries.push(entry);
403        }
404    }
405
406    Ok(Json(entries))
407}
408
409pub async fn websocket_handler(ws: WebSocketUpgrade, State(state): State<AppState>) -> Response {
410    ws.on_upgrade(|socket| handle_websocket(socket, state))
411}
412
413async fn handle_websocket(socket: WebSocket, state: AppState) {
414    let (mut sender, mut receiver) = socket.split();
415    let mut watch_rx = state.watch_manager.subscribe();
416
417    // Handle incoming messages from client
418    let recv_task = tokio::spawn(async move {
419        while let Some(msg) = receiver.next().await {
420            match msg {
421                Ok(Message::Text(text)) => {
422                    println!("Received WebSocket message: {}", text);
423                    // TODO: Handle client messages for subscription management
424                }
425                Ok(Message::Close(_)) => {
426                    println!("WebSocket connection closed");
427                    break;
428                }
429                Err(e) => {
430                    eprintln!("WebSocket error: {}", e);
431                    break;
432                }
433                _ => {}
434            }
435        }
436    });
437
438    // Handle outgoing messages to client
439    let send_task = tokio::spawn(async move {
440        while let Ok(watch_event) = watch_rx.recv().await {
441            let json_msg = match serde_json::to_string(&watch_event) {
442                Ok(json) => json,
443                Err(e) => {
444                    eprintln!("Failed to serialize watch event: {}", e);
445                    continue;
446                }
447            };
448
449            if sender.send(Message::Text(json_msg)).await.is_err() {
450                break;
451            }
452        }
453    });
454
455    // Wait for either task to complete
456    tokio::select! {
457        _ = recv_task => {},
458        _ = send_task => {},
459    }
460}
cc_log_viewer/lib.rs

cc_log_viewer/
lib.rs
