Connect db to cli

ahnlich/cli/Cargo.toml

@@ -15,8 +15,11 @@ path = "src/lib.rs"
 [dependencies]
 crossterm = "0.28.1"
 clap.workspace = true
+dsl = { path = "../dsl", version = "*" }
 thiserror.workspace = true
 tokio.workspace = true
 ahnlich_client_rs = { path = "../client", version = "*" }
 deadpool.workspace = true
 ahnlich_types = { path = "../types", version = "*" }
+serde_json.workspace = true
+serde.workspace = true

ahnlich/cli/src/connect.rs

@@ -1,11 +1,15 @@
 use super::config::cli::Agent;
 use ahnlich_client_rs::{
     ai::{AIClient, AIConnManager},
-    db::{DbClient, DbConnManager},
+    db::{DbClient, DbConnManager, DbPipeline},
     prelude::{AIServerResponse, ServerResponse},
 };
-use ahnlich_types::ServerType;
+use ahnlich_types::{db::ServerDBQuery, ServerType};
 use deadpool::managed::Pool;
+use dsl::db::parse_db_query;
+
+use crossterm::style::Stylize;
+use serde::Serialize;
 
 #[derive(Debug)]
 pub enum AgentPool {
@@ -68,6 +72,28 @@ impl AgentPool {
 
         Ok(false)
     }
+
+    pub async fn parse_queries(&self, input: &str) -> Result<Vec<String>, String> {
+        match self {
+            AgentPool::AI(_pool) => Err(String::from("Unimplemented")),
+            AgentPool::DB(pool) => {
+                let queries = parse_db_query(input).map_err(|err| err.to_string())?;
+
+                let server_query = ServerDBQuery::from_queries(&queries);
+
+                let conn = pool
+                    .get()
+                    .await
+                    .map_err(|err| format!("Could not get db client connection {err}"))?;
+
+                let pipeline = DbPipeline::new_from_queries_and_conn(server_query, conn);
+
+                let response = pipeline.exec().await.map_err(|err| err.to_string())?;
+
+                Ok(render(response.into_inner()))
+            }
+        }
+    }
 }
 
 impl std::fmt::Display for AgentPool {
@@ -78,3 +104,25 @@ impl std::fmt::Display for AgentPool {
         }
     }
 }
+
+fn render(input: Vec<Result<impl Serialize, String>>) -> Vec<String> {
+    input
+        .into_iter()
+        .map(|val| match val {
+            Ok(success) => format_success(
+                serde_json::to_string_pretty(&success)
+                    .map_err(|err| err.to_string())
+                    .expect("Failed to parse ai success response to json"),
+            ),
+            Err(err) => format_error(err),
+        })
+        .collect()
+}
+
+fn format_success(input: String) -> String {
+    format!("{}", input.green())
+}
+
+fn format_error(input: String) -> String {
+    format!("{}", input.red())
+}

ahnlich/cli/src/term.rs

@@ -7,7 +7,7 @@ use std::io::{self, stdout, Write};
 
 use crate::connect::AgentPool;
 
-const RESERVED_WORDS: [&str; 3] = ["hello", "print", "ping"];
+const RESERVED_WORDS: [&str; 3] = ["ping", "infoserver", "createpredindex"];
 
 pub struct Term {
     client_pool: AgentPool,
@@ -55,14 +55,14 @@ impl Term {
         stdout.execute(Print(">>> "))?;
         stdout.execute(SetForegroundColor(Color::White))?;
         stdout.flush()?;
-        stdout.flush()?;
+        //stdout.flush()?;
         Ok(())
     }
 
-    pub(crate) fn query_output(&self, query: String) -> io::Result<()> {
+    pub(crate) fn print_query(&self, query: &str) -> io::Result<()> {
         self.ahnlich_prompt()?;
         let output = String::from_iter(query.split(' ').map(|ex| {
-            if RESERVED_WORDS.contains(&ex) {
+            if RESERVED_WORDS.contains(&(ex.to_lowercase().as_str())) {
                 format!("{} ", ex.magenta())
             } else {
                 format!("{} ", ex.white())
@@ -80,7 +80,17 @@ impl Term {
             let input = self.read_line()?;
             match input.as_str() {
                 "quit" | "exit()" => break,
-                _ => self.query_output(input)?,
+                command => {
+                    self.print_query(command)?;
+                    let response = self.client_pool.parse_queries(command).await;
+
+                    match response {
+                        Ok(success) => {
+                            println!("{}", success.join("\n\n"))
+                        }
+                        Err(err) => println!("{}", err.red()),
+                    }
+                }
             };
         }
         Ok(())

ahnlich/client/src/ai.rs

@@ -45,6 +45,9 @@ pub struct AIPipeline {
 }
 
 impl AIPipeline {
+    pub fn new_from_queries_and_conn(queries: AIServerQuery, conn: Object<AIConnManager>) -> Self {
+        Self { queries, conn }
+    }
     /// push create store command to pipeline
     pub fn create_store(
         &mut self,
@@ -197,10 +200,10 @@ impl AIClient {
         capacity: usize,
         tracing_id: Option<String>,
     ) -> Result<AIPipeline, AhnlichError> {
-        Ok(AIPipeline {
-            queries: AIServerQuery::with_capacity_and_tracing_id(capacity, tracing_id),
-            conn: self.pool.get().await?,
-        })
+        Ok(AIPipeline::new_from_queries_and_conn(
+            AIServerQuery::with_capacity_and_tracing_id(capacity, tracing_id),
+            self.pool.get().await?,
+        ))
     }
 
     pub async fn create_store(

ahnlich/client/src/db.rs

@@ -45,6 +45,10 @@ pub struct DbPipeline {
 }
 
 impl DbPipeline {
+    pub fn new_from_queries_and_conn(queries: ServerDBQuery, conn: Object<DbConnManager>) -> Self {
+        Self { queries, conn }
+    }
+
     /// push create store command to pipeline
     pub fn create_store(
         &mut self,
@@ -214,10 +218,10 @@ impl DbClient {
         capacity: usize,
         tracing_id: Option<String>,
     ) -> Result<DbPipeline, AhnlichError> {
-        Ok(DbPipeline {
-            queries: ServerDBQuery::with_capacity_and_tracing_id(capacity, tracing_id)?,
-            conn: self.pool.get().await?,
-        })
+        Ok(DbPipeline::new_from_queries_and_conn(
+            ServerDBQuery::with_capacity_and_tracing_id(capacity, tracing_id)?,
+            self.pool.get().await?,
+        ))
     }
 
     pub async fn create_store(

ahnlich/types/src/ai/server.rs

@@ -69,6 +69,10 @@ impl AIServerResult {
     pub fn is_empty(&self) -> bool {
         self.results.is_empty()
     }
+
+    pub fn into_inner(self) -> AIServerResultInner {
+        self.results
+    }
 }
 
 impl BinCodeSerAndDeserResponse for AIServerResult {

ahnlich/types/src/db/server.rs

@@ -95,6 +95,10 @@ impl ServerResult {
     pub fn push(&mut self, entry: Result<ServerResponse, String>) {
         self.results.push(entry)
     }
+
+    pub fn into_inner(self) -> ServerResultInner {
+        self.results
+    }
 }
 
 impl BinCodeSerAndDeserResponse for ServerResult {