feat: add basic admin authed api (#2058)

.cursor/rules/protobuff.mdc

@@ -0,0 +1,6 @@
+---
+description: Best practices to edit and build .proto files
+globs: *.proto
+---
+
+DO NOT MAKE BREAKING CHANGES

.cursor/rules/rust.mdc

@@ -0,0 +1,104 @@
+---
+description: Rust Langauge best practices
+globs: *.rs
+---
+# Architectural Rules
+
+## General Principles
+
+- **Keep it Simple (KISS)**: Strive for simplicity in code design.
+- **Don't Repeat Yourself (DRY)**: Eliminate duplication to enhance maintainability.
+- **You Ain't Gonna Need It (YAGNI)**: Avoid adding functionality until it's necessary.
+- **Readability Matters**: Write code as if the next person to read it is a beginner.
+
+## Code Structure
+
+- Use meaningful and descriptive variable, function, and class names.
+- Organize files and directories logically.
+- Follow a consistent indentation and formatting style (use linters and formatters).
+- Separate concerns: Keep logic modular and avoid monolithic functions.
+
+## Performance Considerations
+
+- Optimize only when necessary—write clear code first.
+- Use efficient data structures and algorithms.
+- Avoid unnecessary computations and redundant API calls.
+- Be mindful of memory usage and garbage collection.
+
+## Security Best Practices
+
+- Never hardcode sensitive data (use environment variables or secrets management tools).
+- Sanitize inputs to prevent injection attacks.
+- Follow the principle of least privilege when managing permissions.
+- Keep dependencies updated to mitigate vulnerabilities.
+
+
+
+## Error Handling
+1. Error types should be located near their unit of fallibility
+   - No global `Error` type or `errors.rs` module
+   - Each module/component defines its own error types
+   - Error types live close to the functions that generate them
+
+2. Use thiserror for error definitions
+   - Derive Error trait using #[derive(thiserror::Error)]
+   - Use #[error()] attribute for human-readable messages
+   - Use #[from] attribute to implement From trait where appropriate
+   - Mark error types as #[non_exhaustive]
+
+3. Structure errors in layers
+   - Use struct for context (e.g. file paths, line numbers)
+   - Use enum for error variants
+   - Implement source() to chain underlying errors
+   - Example:
+     ```rust
+     #[derive(Debug, thiserror::Error)]
+     #[error("error reading `{path}`")] 
+     pub struct FileError {
+         pub path: PathBuf,
+         #[source]
+         pub kind: FileErrorKind
+     }
+     ```
+
+4. Error matching and handling
+   - Make errors inspectable with specific variants
+   - Provide enough context for actionable error messages
+   - Use meaningful variant names (e.g. ReadFile vs Io)
+   - Document error conditions and handling
+
+5. Error stability and privacy
+   - Consider making error fields private when needed
+   - Don't expose internal error types in public API
+   - Use opaque error types for stable interfaces
+   - Version error types appropriately
+
+6. Do NOT use anyhow::Result
+   - If you see anyhow::Result being used, refactor using this::error
+
+## Constants
+- Use `const` for all static values in Rust unless interior mutability or runtime evaluation is required.
+- Prefer placing constants in a `constants.rs` file.
+- The `constants.rs` file should be located at the deepest level in the module tree but at the highest level where all usages of the constants exist.
+- Ensure constants are appropriately scoped to avoid unnecessary exposure to unrelated modules.
+- Use `pub(crate)` or `pub(super)` instead of `pub` when limiting visibility to the necessary scope.
+- Group related constants together for better maintainability and readability.
+- Use descriptive names in uppercase with underscores (e.g., `MAX_RETRY_COUNT`).
+- When working with enums or complex types, prefer `static` with `lazy_static!` or `once_cell::sync::Lazy` for initialization.
+- Avoid redefining constants in multiple places; ensure they are sourced from `constants.rs` where needed.
+
+## Documentation
+1. All public APIs must be documented
+2. Architecture decisions should be documented
+3. Side effects should be clearly documented
+4. Breaking changes must be documented
+
+## Observability
+
+- Implement logging, metrics, and tracing to gain insights into system behavior.
+- Use structured logging for better searchability and debugging.
+- Leverage distributed tracing to diagnose performance bottlenecks in microservices.
+
+## Linters
+* Always run `cargo clippy` to make sure your changes are right. In Clippy we trust.
+* Don't go around the linters by adding exceptions, try to actually use the variables if you find deadcode or delete the code if it is actually not useful.

apps/framework-cli/src/cli/local_webserver.rs

@@ -2,6 +2,7 @@ use super::display::Message;
 use super::display::MessageType;
 use super::routines::auth::validate_auth_token;
 use super::settings::Settings;
+use crate::infrastructure::redis::redis_client::RedisClient;
 use crate::metrics::MetricEvent;
 
 use crate::cli::display::with_spinner;
@@ -46,6 +47,7 @@ use serde::Serialize;
 use serde::{Deserialize, Deserializer};
 use serde_json::Deserializer as JsonDeserializer;
 use tokio::spawn;
+use tokio::sync::Mutex;
 
 use crate::framework::data_model::model::DataModel;
 use crate::utilities::validate_passthrough::{DataModelArrayVisitor, DataModelVisitor};
@@ -219,6 +221,8 @@ struct RouteService {
     is_prod: bool,
     metrics: Arc<Metrics>,
     http_client: Arc<Client>,
+    project: Arc<Project>,
+    redis_client: Arc<Mutex<RedisClient>>,
 }
 #[derive(Clone)]
 struct ManagementService<I: InfraMapProvider + Clone> {
@@ -249,6 +253,8 @@ impl Service<Request<Incoming>> for RouteService {
                 req,
                 route_table: self.route_table,
             },
+            self.project.clone(),
+            self.redis_client.clone(),
         ))
     }
 }
@@ -287,14 +293,88 @@ fn options_route() -> Result<Response<Full<Bytes>>, hyper::http::Error> {
     Ok(response)
 }
 
-fn health_route() -> Result<Response<Full<Bytes>>, hyper::http::Error> {
+async fn health_route() -> Result<Response<Full<Bytes>>, hyper::http::Error> {
     let response = Response::builder()
         .status(StatusCode::OK)
         .body(Full::new(Bytes::from("Success")))
         .unwrap();
     Ok(response)
 }
 
+async fn admin_reality_check_route(
+    req: Request<hyper::body::Incoming>,
+    admin_api_key: &Option<String>,
+    project: &Project,
+    redis_client: &Arc<Mutex<RedisClient>>,
+) -> Result<Response<Full<Bytes>>, hyper::http::Error> {
+    let auth_header = req.headers().get(hyper::header::AUTHORIZATION);
+    let bearer_token = auth_header
+        .and_then(|header_value| header_value.to_str().ok())
+        .and_then(|header_str| header_str.strip_prefix("Bearer "));
+
+    // Check API key authentication
+    if let Some(key) = admin_api_key.as_ref() {
+        if !validate_token(bearer_token, key).await {
+            return Response::builder()
+                .status(StatusCode::UNAUTHORIZED)
+                .body(Full::new(Bytes::from(
+                    "Unauthorized: Invalid or missing token",
+                )));
+        }
+    } else {
+        return Response::builder()
+            .status(StatusCode::UNAUTHORIZED)
+            .body(Full::new(Bytes::from(
+                "Unauthorized: Admin API key not configured",
+            )));
+    }
+
+    // Create OLAP client and reality checker
+    let olap_client =
+        crate::infrastructure::olap::clickhouse::create_client(project.clickhouse_config.clone());
+    let reality_checker =
+        crate::framework::core::infra_reality_checker::InfraRealityChecker::new(olap_client);
+
+    // Load infrastructure map from Redis
+    let redis_guard = redis_client.lock().await;
+    let infra_map = match InfrastructureMap::get_from_redis(&redis_guard).await {
+        Ok(map) => map,
+        Err(e) => {
+            return Response::builder()
+                .status(StatusCode::INTERNAL_SERVER_ERROR)
+                .body(Full::new(Bytes::from(format!(
+                    "Failed to get infrastructure map: {}",
+                    e
+                ))))
+        }
+    };
+
+    // Perform reality check
+    match reality_checker.check_reality(project, &infra_map).await {
+        Ok(discrepancies) => {
+            let response = serde_json::json!({
+                "status": "success",
+                "discrepancies": {
+                    "unmapped_tables": discrepancies.unmapped_tables,
+                    "missing_tables": discrepancies.missing_tables,
+                    "mismatched_tables": discrepancies.mismatched_tables,
+                }
+            });
+
+            Ok(Response::builder()
+                .status(StatusCode::OK)
+                .header("Content-Type", "application/json")
+                .body(Full::new(Bytes::from(response.to_string())))?)
+        }
+        Err(e) => Ok(Response::builder()
+            .status(StatusCode::INTERNAL_SERVER_ERROR)
+            .body(Full::new(Bytes::from(format!(
+                "{{\"status\": \"error\", \"message\": \"{}\"}}",
+                e
+            ))))?),
+    }
+}
+
 async fn log_route(req: Request<Incoming>) -> Response<Full<Bytes>> {
     let body = to_reader(req).await;
     let parsed: Result<CliMessage, serde_json::Error> = serde_json::from_reader(body);
@@ -710,6 +790,8 @@ async fn router(
     metrics: Arc<Metrics>,
     http_client: Arc<Client>,
     request: RouterRequest,
+    project: Arc<Project>,
+    redis_client: Arc<Mutex<RedisClient>>,
 ) -> Result<Response<Full<Bytes>>, hyper::http::Error> {
     let now = Instant::now();
 
@@ -772,8 +854,16 @@ async fn router(
                 }
             }
         }
-        (&hyper::Method::GET, ["health"]) => health_route(),
-
+        (&hyper::Method::GET, ["health"]) => health_route().await,
+        (&hyper::Method::GET, ["admin", "reality-check"]) => {
+            admin_reality_check_route(
+                req,
+                &project.authentication.admin_api_key,
+                &project,
+                &redis_client,
+            )
+            .await
+        }
         (&hyper::Method::OPTIONS, _) => options_route(),
         _ => route_not_found_response(),
     };
@@ -1043,6 +1133,12 @@ impl Webserver {
             is_prod: project.is_production,
             http_client,
             metrics: metrics.clone(),
+            project: project.clone(),
+            redis_client: Arc::new(Mutex::new(
+                RedisClient::new(project.name(), project.redis_config.clone())
+                    .await
+                    .unwrap(),
+            )),
         };
 
         let management_service = ManagementService {

apps/framework-cli/src/framework/consumption/model.rs

@@ -23,6 +23,14 @@ impl ConsumptionQueryParam {
             special_fields: Default::default(),
         }
     }
+
+    pub fn from_proto(proto: ProtoConsumptionQueryParam) -> Self {
+        ConsumptionQueryParam {
+            name: proto.name,
+            data_type: ColumnType::from_proto(proto.data_type.unwrap()),
+            required: proto.required,
+        }
+    }
 }
 
 #[derive(Debug, Clone, Default)]

apps/framework-cli/src/framework/controller.rs

@@ -38,10 +38,15 @@ pub struct RouteMeta {
 pub struct InitialDataLoad {
     pub table: Table,
     pub topic: String,
-
     pub status: InitialDataLoadStatus,
 }
 
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
+pub enum InitialDataLoadStatus {
+    InProgress(i64),
+    Completed,
+}
+
 impl InitialDataLoad {
     pub(crate) fn expanded_display(&self) -> String {
         format!(
@@ -54,19 +59,25 @@ impl InitialDataLoad {
         ProtoInitialDataLoad {
             table: MessageField::some(self.table.to_proto()),
             topic: self.topic.clone(),
-            progress: match self.status {
-                InitialDataLoadStatus::InProgress(i) => Some(i as u64),
+            progress: match &self.status {
+                InitialDataLoadStatus::InProgress(i) => Some(*i as u64),
                 InitialDataLoadStatus::Completed => None,
             },
             special_fields: Default::default(),
         }
     }
-}
 
-#[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
-pub enum InitialDataLoadStatus {
-    InProgress(i64),
-    Completed,
+    pub fn from_proto(proto: ProtoInitialDataLoad) -> Self {
+        InitialDataLoad {
+            table: Table::from_proto(proto.table.unwrap()),
+            topic: proto.topic,
+            status: match proto.progress {
+                Some(i) if i >= 100 => InitialDataLoadStatus::Completed,
+                Some(i) => InitialDataLoadStatus::InProgress(i as i64),
+                None => InitialDataLoadStatus::Completed,
+            },
+        }
+    }
 }
 
 pub async fn initial_data_load(