feat: add basic admin authed api

add lisitng olap operation to the implementation

basic reality checker working

base implementation of reality checker

added ability to read from redis

seems to be mostly working

working ish implementation

some fixes

fixes

fix arnings

fix clippy

fix tests

fix comment

remove some breaking changes

revert back

fix

revert some code changes

.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(