support local ztm and pgSql

.gitignore

@@ -16,3 +16,6 @@ Cargo.lock
 # Mac
 .DS_Store
 .VSCodeCounter# See https://help.github.com/articles/ignoring-files/ for more about ignoring files.
+
+# zmt agent data file
+ztm_agent.db

gateway/Cargo.toml

@@ -15,6 +15,7 @@ jupiter = { workspace = true }
 callisto = { workspace = true }
 ceres = { workspace = true }
 gemini = { workspace = true }
+vault = { workspace = true }
 venus = { workspace = true }
 mercury = { workspace = true }
 anyhow = { workspace = true }

gateway/src/https_server.rs

@@ -14,7 +14,7 @@ use axum::Router;
 use axum_server::tls_rustls::RustlsConfig;
 use clap::Args;
 use common::enums::ZtmType;
-use gemini::ztm::run_ztm_client;
+use gemini::ztm::{run_ztm_client, LocalZTM};
 use regex::Regex;
 use tower::ServiceBuilder;
 use tower_http::cors::{Any, CorsLayer};
@@ -196,7 +196,10 @@ async fn post_method_router(
         return lfs::lfs_delete_lock(state, &lfs_config, uri.path(), req).await;
     } else if Regex::new(r"/objects/batch$").unwrap().is_match(uri.path()) {
         return lfs::lfs_process_batch(state, &lfs_config, req).await;
-    } else if Regex::new(r"objects/chunkids$").unwrap().is_match(uri.path()) {
+    } else if Regex::new(r"objects/chunkids$")
+        .unwrap()
+        .is_match(uri.path())
+    {
         return lfs::lfs_fetch_chunk_ids(state, &lfs_config, req).await;
     }
     // Routing git services.
@@ -257,16 +260,17 @@ pub fn check_run_with_ztm(config: Config, common: CommonOptions) {
     match ztm_type {
         ZtmType::Agent => {
             //Mega server join a ztm mesh
-            let config = config.ztm;
             let bootstrap_node = match common.bootstrap_node {
                 Some(n) => n,
                 None => {
                     tracing::error!("bootstrap node is not provide");
                     return;
                 }
             };
-            let peer_id = "123".to_string();
-            tokio::spawn(async move { run_ztm_client(bootstrap_node, config, peer_id).await });
+            let (peer_id, _) = vault::init();
+            let ztm: LocalZTM = LocalZTM { agent_port: 7778 };
+            ztm.clone().start_ztm_agent();
+            tokio::spawn(async move { run_ztm_client(bootstrap_node, config, peer_id, ztm).await });
         }
         ZtmType::Relay => {
             //Start a sub thread to run relay server

gateway/src/relay_server.rs

@@ -2,13 +2,14 @@ use std::net::SocketAddr;
 use std::str::FromStr;
 
 use axum::body::Body;
-use axum::extract::{Query, State};
+use axum::extract::{FromRequest, Query, State};
 use axum::http::{Request, Response, StatusCode, Uri};
 use axum::routing::get;
-use axum::Router;
+use axum::{Json, Router};
+use callisto::{ztm_node, ztm_repo_info};
 use common::config::{Config, ZTMConfig};
-use gemini::ztm::{RemoteZTM, ZTM};
-use gemini::RelayGetParams;
+use gemini::ztm::{RemoteZTM, ZTMAgent, ZTMCA};
+use gemini::{Node, RelayGetParams, RelayResultRes, RepoInfo};
 use jupiter::context::Context;
 use regex::Regex;
 use tower::ServiceBuilder;
@@ -63,10 +64,7 @@ pub async fn app(config: Config, host: String, port: u16) -> Router {
     // add TraceLayer for log record
     // add CorsLayer to add cors header
     Router::new()
-        .nest(
-            "/api/v1",
-            api_router::routers().with_state(api_state),
-        )
+        .nest("/api/v1", api_router::routers().with_state(api_state))
         .route(
             "/*path",
             get(get_method_router).post(post_method_router),
@@ -88,6 +86,12 @@ async fn get_method_router(
         return hello_relay(params).await;
     } else if Regex::new(r"/certificate$").unwrap().is_match(uri.path()) {
         return certificate(ztm_config, params).await;
+    } else if Regex::new(r"/ping$").unwrap().is_match(uri.path()) {
+        return ping(state, params).await;
+    } else if Regex::new(r"/node_list$").unwrap().is_match(uri.path()) {
+        return node_list(state, params).await;
+    } else if Regex::new(r"/repo_list$").unwrap().is_match(uri.path()) {
+        return repo_list(state, params).await;
     }
     Err((
         StatusCode::NOT_FOUND,
@@ -97,10 +101,13 @@ async fn get_method_router(
 
 async fn post_method_router(
     state: State<AppState>,
-    _uri: Uri,
-    _req: Request<Body>,
+    uri: Uri,
+    req: Request<Body>,
 ) -> Result<Response<Body>, (StatusCode, String)> {
     let _ztm_config = state.context.config.ztm.clone();
+    if Regex::new(r"/repo_provide$").unwrap().is_match(uri.path()) {
+        return repo_provide(state, req).await;
+    }
     Err((
         StatusCode::NOT_FOUND,
         String::from("Operation not supported\n"),
@@ -137,6 +144,101 @@ pub async fn certificate(
         .unwrap())
 }
 
+pub async fn ping(
+    state: State<AppState>,
+    params: RelayGetParams,
+) -> Result<Response<Body>, (StatusCode, String)> {
+    let storage = state.context.services.ztm_storage.clone();
+    let node: ztm_node::Model = match params.try_into() {
+        Ok(n) => n,
+        Err(_) => {
+            return Err((
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "invalid paras".to_string(),
+            ));
+        }
+    };
+    match storage.insert_or_update_node(node).await {
+        Ok(_) => {
+            let res = serde_json::to_string(&RelayResultRes { success: true }).unwrap();
+            Ok(Response::builder()
+                .header("Content-Type", "application/json")
+                .body(Body::from(res))
+                .unwrap())
+        }
+        Err(_) => Err((
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "invalid paras".to_string(),
+        )),
+    }
+}
+
+pub async fn node_list(
+    state: State<AppState>,
+    _params: RelayGetParams,
+) -> Result<Response<Body>, (StatusCode, String)> {
+    let storage = state.context.services.ztm_storage.clone();
+    let nodelist: Vec<Node> = storage
+        .get_all_node()
+        .await
+        .unwrap()
+        .into_iter()
+        .map(|x| x.into())
+        .collect();
+    let json_string = serde_json::to_string(&nodelist).unwrap();
+    Ok(Response::builder()
+        .header("Content-Type", "application/json")
+        .body(Body::from(json_string))
+        .unwrap())
+}
+
+pub async fn repo_provide(
+    state: State<AppState>,
+    req: Request<Body>,
+) -> Result<Response<Body>, (StatusCode, String)> {
+    let storage = state.context.services.ztm_storage.clone();
+    let request = Json::from_request(req, &state)
+        .await
+        .unwrap_or_else(|_| Json(RepoInfo::default()));
+    let repo_info: RepoInfo = request.0;
+    if repo_info.identifier.is_empty() {
+        return Err((StatusCode::BAD_REQUEST, "paras invalid".to_string()));
+    }
+    let repo_info_model: ztm_repo_info::Model = repo_info.into();
+    match storage.insert_or_update_repo_info(repo_info_model).await {
+        Ok(_) => {
+            let res = serde_json::to_string(&RelayResultRes { success: true }).unwrap();
+            Ok(Response::builder()
+                .header("Content-Type", "application/json")
+                .body(Body::from(res))
+                .unwrap())
+        }
+        Err(_) => Err((
+            StatusCode::INTERNAL_SERVER_ERROR,
+            "invalid paras".to_string(),
+        )),
+    }
+}
+
+pub async fn repo_list(
+    state: State<AppState>,
+    _params: RelayGetParams,
+) -> Result<Response<Body>, (StatusCode, String)> {
+    let storage = state.context.services.ztm_storage.clone();
+    let repo_info_list: Vec<RepoInfo> = storage
+        .get_all_repo_info()
+        .await
+        .unwrap()
+        .into_iter()
+        .map(|x| x.into())
+        .collect();
+    let json_string = serde_json::to_string(&repo_info_list).unwrap();
+    Ok(Response::builder()
+        .header("Content-Type", "application/json")
+        .body(Body::from(json_string))
+        .unwrap())
+}
+
 pub async fn relay_connect_ztm(config: ZTMConfig, relay_port: u16) -> Result<String, String> {
     // 1. generate a permit for relay
     let name = "relay".to_string();

gemini/Cargo.toml

@@ -16,4 +16,9 @@ axum = { workspace = true }
 serde = { workspace = true }
 serde_json = { workspace = true }
 reqwest = { workspace = true }
-tracing = { workspace = true }
+tracing = { workspace = true }
+tokio = { workspace = true, features = ["net"] }
+chrono = { workspace = true }
+jupiter = { workspace = true }
+callisto = { workspace = true }
+rust-ztm = { git = "https://github.com/flomesh-io/rust-ztm.git" }

gemini/src/lib.rs

@@ -1,9 +1,148 @@
-use serde::Deserialize;
+use std::fmt;
+
+use callisto::{ztm_node, ztm_repo_info};
+use chrono::Utc;
+use serde::{Deserialize, Serialize};
 
 pub mod http;
 pub mod ztm;
 
 #[derive(Deserialize, Debug)]
 pub struct RelayGetParams {
+    pub peer_id: Option<String>,
+    pub hub: Option<String>,
     pub name: Option<String>,
+    pub agent_name: Option<String>,
+    pub service_name: Option<String>,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct RelayResultRes {
+    pub success: bool,
+}
+
+#[derive(Debug)]
+pub enum MegaType {
+    Agent,
+    Relay,
+}
+
+impl fmt::Display for MegaType {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        match self {
+            MegaType::Agent => write!(f, "Agent"),
+            MegaType::Relay => write!(f, "Relay"),
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
+pub struct Node {
+    pub peer_id: String,
+    pub hub: String,
+    pub agent_name: String,
+    pub service_name: String,
+    pub mega_type: String,
+    pub online: bool,
+    pub last_online_time: i64,
+}
+
+#[derive(Debug)]
+pub enum ConversionError {
+    InvalidParas,
+}
+
+impl TryFrom<RelayGetParams> for Node {
+    type Error = ConversionError;
+
+    fn try_from(paras: RelayGetParams) -> Result<Self, Self::Error> {
+        if paras.peer_id.is_none()
+            || paras.hub.is_none()
+            || paras.agent_name.is_none()
+            || paras.service_name.is_none()
+        {
+            return Err(ConversionError::InvalidParas);
+        }
+        let now = Utc::now().timestamp_millis();
+        Ok(Node {
+            peer_id: paras.peer_id.unwrap(),
+            hub: paras.hub.unwrap(),
+            agent_name: paras.agent_name.unwrap(),
+            service_name: paras.service_name.unwrap(),
+            mega_type: MegaType::Agent.to_string(),
+            online: true,
+            last_online_time: now,
+        })
+    }
+}
+
+impl TryFrom<RelayGetParams> for ztm_node::Model {
+    type Error = ConversionError;
+
+    fn try_from(paras: RelayGetParams) -> Result<Self, Self::Error> {
+        if paras.peer_id.is_none()
+            || paras.hub.is_none()
+            || paras.agent_name.is_none()
+            || paras.service_name.is_none()
+        {
+            return Err(ConversionError::InvalidParas);
+        }
+        let now = Utc::now().timestamp_millis();
+        Ok(ztm_node::Model {
+            peer_id: paras.peer_id.unwrap(),
+            hub: paras.hub.unwrap(),
+            agent_name: paras.agent_name.unwrap(),
+            service_name: paras.service_name.unwrap(),
+            r#type: MegaType::Agent.to_string(),
+            online: true,
+            last_online_time: now,
+        })
+    }
+}
+
+impl From<ztm_node::Model> for Node {
+    fn from(n: ztm_node::Model) -> Self {
+        Node {
+            peer_id: n.peer_id,
+            hub: n.hub,
+            agent_name: n.agent_name,
+            service_name: n.service_name,
+            mega_type: n.r#type,
+            online: n.online,
+            last_online_time: n.last_online_time,
+        }
+    }
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone, Default)]
+pub struct RepoInfo {
+    pub name: String,
+    pub identifier: String,
+    pub origin: String,
+    pub update_time: i64,
+    pub commit: String,
+}
+
+impl From<RepoInfo> for ztm_repo_info::Model {
+    fn from(r: RepoInfo) -> Self {
+        ztm_repo_info::Model {
+            identifier: r.identifier,
+            name: r.name,
+            origin: r.origin,
+            update_time: r.update_time,
+            commit: r.commit,
+        }
+    }
+}
+
+impl From<ztm_repo_info::Model> for RepoInfo {
+    fn from(r: ztm_repo_info::Model) -> Self {
+        RepoInfo {
+            identifier: r.identifier,
+            name: r.name,
+            origin: r.origin,
+            update_time: r.update_time,
+            commit: r.commit,
+        }
+    }
 }