datafold 0.1.55

A personal database for data sovereignty with AI-powered ingestion
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

//! Logging abstraction for Lambda deployments
//!
//! Re-exports generic logging functionality from `crate::logging::core`.

pub use crate::logging::core::*;

use super::context::LambdaContext;
use crate::ingestion::IngestionError;
use serde_json::Value;

impl LambdaContext {
    /// List logs
    ///
    /// # Arguments
    ///
    /// * `since` - Optional timestamp to filter logs
    /// * `limit` - Optional limit on number of logs (default 1000)
    /// * `user_id` - User ID for node context
    pub async fn list_logs(
        since: Option<i64>,
        limit: Option<usize>,
        user_id: String,
    ) -> Result<Vec<Value>, IngestionError> {
        let processor = {
            let node_mutex = Self::get_node(&user_id).await?;
            let node_guard = node_mutex.lock().await;
            crate::datafold_node::OperationProcessor::new(node_guard.clone())
        };

        let limit = limit.unwrap_or(1000);
        let logs = processor.list_logs(since, Some(limit)).await;

        Ok(logs
            .into_iter()
            .map(|log| {
                serde_json::to_value(log)
                    .unwrap_or(serde_json::json!({"error": "Failed to serialize log"}))
            })
            .collect())
    }

    /// Get log configuration
    pub async fn get_log_config(user_id: String) -> Result<Value, IngestionError> {
        let processor = {
            let node_mutex = Self::get_node(&user_id).await?;
            let node_guard = node_mutex.lock().await;
            crate::datafold_node::OperationProcessor::new(node_guard.clone())
        };

        if let Some(config) = processor.get_log_config().await {
            Ok(serde_json::json!({
                "config": config
            }))
        } else {
            let current_level = log::max_level().to_string();
            Ok(serde_json::json!({
                "message": "Basic logging configuration",
                "current_level": current_level
            }))
        }
    }

    /// Reload log configuration
    pub async fn reload_log_config(user_id: String) -> Result<(), IngestionError> {
        let processor = {
            let node_mutex = Self::get_node(&user_id).await?;
            let node_guard = node_mutex.lock().await;
            crate::datafold_node::OperationProcessor::new(node_guard.clone())
        };

        processor
            .reload_log_config("config/logging.toml")
            .await
            .map_err(|e| {
                IngestionError::InvalidInput(format!("Failed to reload configuration: {}", e))
            })
    }

    /// Get log features
    pub async fn get_log_features(user_id: String) -> Result<Value, IngestionError> {
        let processor = {
            let node_mutex = Self::get_node(&user_id).await?;
            let node_guard = node_mutex.lock().await;
            crate::datafold_node::OperationProcessor::new(node_guard.clone())
        };

        if let Some(features) = processor.get_log_features().await {
            Ok(serde_json::json!({
                "features": features,
                "available_levels": ["TRACE", "DEBUG", "INFO", "WARN", "ERROR"]
            }))
        } else {
            let current_level = log::max_level().to_string();
            Ok(serde_json::json!({
                "features": {
                    "transform": current_level,
                    "network": current_level,
                    "database": current_level,
                    "schema": current_level,
                    "query": current_level,
                    "mutation": current_level,
                    "permissions": current_level,
                    "http_server": current_level,
                    "tcp_server": current_level,
                    "ingestion": current_level
                },
                "available_levels": ["TRACE", "DEBUG", "INFO", "WARN", "ERROR"]
            }))
        }
    }

    /// Update log feature level
    pub async fn update_log_feature_level(
        feature: &str,
        level: &str,
        user_id: String,
    ) -> Result<(), IngestionError> {
        let valid_levels = ["TRACE", "DEBUG", "INFO", "WARN", "ERROR"];
        if !valid_levels.contains(&level) {
            return Err(IngestionError::InvalidInput(format!(
                "Invalid log level: {}",
                level
            )));
        }

        let processor = {
            let node_mutex = Self::get_node(&user_id).await?;
            let node_guard = node_mutex.lock().await;
            crate::datafold_node::OperationProcessor::new(node_guard.clone())
        };

        processor
            .update_log_feature_level(feature, level)
            .await
            .map_err(|e| IngestionError::InvalidInput(format!("Failed to update log level: {}", e)))
    }
}