diff --git a/aries/Cargo.toml b/aries/Cargo.toml index 9adf07b86..4faf14c40 100644 --- a/aries/Cargo.toml +++ b/aries/Cargo.toml @@ -17,8 +17,11 @@ clap = { workspace = true, features = ["derive"] } tracing = { workspace = true } tracing-subscriber = { workspace = true } tracing-appender = { workspace = true } +vault = { workspace = true } axum = { workspace = true } serde_json = { workspace = true } +serde = { workspace = true } +uuid = { workspace = true } tower = { workspace = true } tower-http = { workspace = true, features = [ "cors", diff --git a/aries/src/main.rs b/aries/src/main.rs index e521bc524..6b139d0e4 100644 --- a/aries/src/main.rs +++ b/aries/src/main.rs @@ -1,11 +1,18 @@ use clap::Parser; use common::config::{Config, LogConfig}; -use gemini::ztm::hub::LocalZTMHub; +use gemini::ztm::{ + agent::{run_ztm_client, LocalZTMAgent, ZTMAgent}, + hub::LocalZTMHub, +}; use service::{ ca_server::run_ca_server, relay_server::{run_relay_server, RelayOptions}, }; -use std::{env, thread, time}; +use std::{ + env, + thread::{self}, + time::{self}, +}; use tracing_subscriber::fmt::writer::MakeWriterExt; pub mod service; @@ -35,11 +42,28 @@ async fn main() { let option = RelayOptions::parse(); tracing::info!("{:?}", option); + if option.only_agent { + let (peer_id, _) = vault::init(); + let ztm_agent: LocalZTMAgent = LocalZTMAgent { + agent_port: option.ztm_agent_port, + }; + ztm_agent.clone().start_ztm_agent(); + thread::sleep(time::Duration::from_secs(3)); + run_ztm_client( + "http://34.84.172.121/relay".to_string(), + config.clone(), + peer_id, + ztm_agent, + 8001, + ) + .await + } + //Start a sub thread to ca server let config_clone = config.clone(); let ca_port = option.ca_port; tokio::spawn(async move { run_ca_server(config_clone, ca_port).await }); - thread::sleep(time::Duration::from_secs(5)); + thread::sleep(time::Duration::from_secs(3)); //Start a sub thread to run ztm-hub let ca = format!("127.0.0.1:{ca_port}"); @@ -49,7 +73,23 @@ async fn main() { name: vec!["relay".to_string()], }; ztm_hub.clone().start_ztm_hub(); - thread::sleep(time::Duration::from_secs(5)); + thread::sleep(time::Duration::from_secs(3)); + + //Start a sub thread to run ztm-agent + let ztm_agent = LocalZTMAgent { + agent_port: option.ztm_agent_port, + }; + thread::sleep(time::Duration::from_secs(3)); + + match ztm_agent.get_ztm_endpoints().await { + Ok(ztm_ep_list) => { + tracing::info!("ztm agent connect success"); + tracing::info!("{} online endpoints", ztm_ep_list.len()); + } + Err(_) => { + tracing::error!("ztm agent connect failed"); + } + } //Start relay server run_relay_server(config, option).await; diff --git a/aries/src/service/api/mod.rs b/aries/src/service/api/mod.rs new file mode 100644 index 000000000..13e3d62e6 --- /dev/null +++ b/aries/src/service/api/mod.rs @@ -0,0 +1,4 @@ +pub mod nostr_router; + +#[cfg(test)] +mod tests {} diff --git a/aries/src/service/api/nostr_router.rs b/aries/src/service/api/nostr_router.rs new file mode 100644 index 000000000..4ead97572 --- /dev/null +++ b/aries/src/service/api/nostr_router.rs @@ -0,0 +1,271 @@ +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, +}; + +use axum::{ + extract::{Query, State}, + http::StatusCode, + routing::{get, post}, + Json, Router, +}; + +use callisto::{ztm_nostr_event, ztm_nostr_req}; +use gemini::nostr::{ + client_message::{ClientMessage, SubscriptionId}, + event::NostrEvent, + relay_message::RelayMessage, + tag::TagKind, +}; +use jupiter::storage::ztm_storage::ZTMStorage; +use serde_json::Value; +use tokio::task; +use uuid::Uuid; + +use crate::service::{relay_server::AppState, Req}; + +pub fn routers() -> Router { + Router::new() + .route("/nostr", post(recieve)) + .route("/nostr/test/event_list", get(event_list)) + .route("/nostr/test/req_list", get(req_list)) +} + +async fn recieve( + state: State, + body: String, +) -> Result, (StatusCode, String)> { + // ["EVENT", ], used to publish events. + // ["REQ", , , , ...], used to request events and subscribe to new updates. + // ["CLOSE", ], used to stop previous subscriptions. + tracing::info!("relay nostr recieve:{}", body); + let client_msg: ClientMessage = match serde_json::from_str(&body) { + Ok(client_msg) => client_msg, + Err(e) => { + return Err((StatusCode::BAD_REQUEST, e.to_string())); + } + }; + match client_msg { + ClientMessage::Event(nostr_event) => { + //event message + match nostr_event.verify() { + Ok(_) => {} + Err(e) => { + return Err((StatusCode::UNAUTHORIZED, e.to_string())); + } + } + let ztm_nostr_event: ztm_nostr_event::Model = match nostr_event.clone().try_into() { + Ok(n) => n, + Err(_) => { + return Err((StatusCode::BAD_REQUEST, "Invalid paras".to_string())); + } + }; + //save + let storage = state.context.services.ztm_storage.clone(); + if storage + .get_nostr_event_by_id(&ztm_nostr_event.id) + .await + .unwrap() + .is_some() + { + return Err((StatusCode::BAD_REQUEST, "Duplicate submission".to_string())); + } + storage.insert_nostr_event(ztm_nostr_event).await.unwrap(); + + //Event is forwarded to subscribed nodes + let nostr_event_clone = nostr_event.clone(); + let storage_clone = storage.clone(); + let ztm_agent_port = state.relay_option.clone().ztm_agent_port; + task::spawn(async move { + transfer_event_to_subscribed_nodes(storage_clone, nostr_event_clone, ztm_agent_port) + .await + }); + + let res = RelayMessage::new_ok(nostr_event.id, true, "ok".to_string()); + let value = serde_json::to_value(res).unwrap(); + Ok(Json(value)) + } + ClientMessage::Req { + subscription_id, + filters, + } => { + //subscribe message + //save + let filters_json = serde_json::to_string(&filters).unwrap(); + let ztm_nostr_req = ztm_nostr_req::Model { + subscription_id: subscription_id.to_string(), + filters: filters_json.clone(), + id: Uuid::new_v4().to_string(), + }; + let storage = state.context.services.ztm_storage.clone(); + let req_list: Vec = storage + .get_all_nostr_req_by_subscription_id(&subscription_id.to_string()) + .await + .unwrap() + .iter() + .map(|x| x.clone().into()) + .collect(); + match req_list.iter().find(|&x| x.filters_json() == filters_json) { + Some(_) => {} + None => { + storage.insert_nostr_req(ztm_nostr_req).await.unwrap(); + } + } + let value = serde_json::to_value("ok").unwrap(); + Ok(Json(value)) + } + } +} + +async fn transfer_event_to_subscribed_nodes( + storage: Arc, + nostr_event: NostrEvent, + ztm_agent_port: u16, +) { + // only support p2p_uri subscription + let mut uri = String::new(); + for tag in nostr_event.clone().tags { + if let gemini::nostr::tag::Tag::Generic(TagKind::URI, t) = tag { + if !t.is_empty() { + uri = t.first().unwrap().to_string(); + } + } + } + if uri.is_empty() { + return; + } + let req_list: Vec = storage + .get_all_nostr_req() + .await + .unwrap() + .iter() + .map(|x| x.clone().into()) + .collect(); + let mut subscription_id_set: HashSet = HashSet::new(); + for req in req_list { + for filter in req.clone().filters { + if let Some(uri_vec) = filter.generic_tags.get(&TagKind::URI.to_string()) { + if uri_vec.is_empty() { + continue; + } + let req_uri = uri_vec.first().unwrap(); + if *req_uri == uri { + subscription_id_set.insert(req.subscription_id.clone()); + } + } + } + } + + for subscription_id in subscription_id_set { + //send event + let msg = RelayMessage::new_event( + SubscriptionId::new(subscription_id.clone()), + nostr_event.clone(), + ) + .as_json(); + match gemini::ztm::send_post_request_to_peer_by_tunnel( + ztm_agent_port, + subscription_id.clone(), + "api/v1/mega/nostr".to_string(), + msg, + ) + .await + { + Ok(_) => { + tracing::info!("send event msg to {} successfully", subscription_id) + } + Err(e) => { + tracing::error!("send event msg to {} failed:{}", subscription_id, e) + } + }; + } +} + +// async fn search_event_by_filters( +// storage: Arc, +// filters: Vec, +// ) -> Vec { +// // only support repo_uri subscribe +// // todo support all filter +// // todo Optimizing the code +// let mut list = vec![]; + +// let mut uri_event_map: HashMap> = HashMap::new(); + +// let event_list = storage.get_all_nostr_event().await.unwrap(); +// let event_list: Vec = event_list +// .iter() +// .map(|x| x.clone().try_into().unwrap()) +// .collect(); + +// for event in event_list { +// for tag in event.clone().tags { +// match tag { +// gemini::nostr::tag::Tag::Generic(TagKind::URI, v) => { +// if !v.is_empty() { +// let event_tag_uri = v.first().unwrap().to_string(); +// match uri_event_map.get(&event_tag_uri) { +// Some(vec) => { +// let mut vec = vec.clone(); +// vec.push(event.clone()); +// uri_event_map.insert(event_tag_uri, vec); +// } +// None => { +// let vec = vec![event.clone()]; +// uri_event_map.insert(event_tag_uri, vec); +// } +// } +// } +// } +// _ => {} +// } +// } +// } + +// for f in filters { +// // tracing::info!("filter:{:?}", f); +// match f.generic_tags.get(&TagKind::URI.to_string()) { +// Some(uri_vec) => { +// if uri_vec.is_empty() { +// continue; +// } +// let uri = uri_vec.first().unwrap(); +// if let Some(vec) = uri_event_map.get(uri) { +// list.extend(vec.clone()); +// } +// } +// None => {} +// } +// } +// list +// } + +pub async fn event_list( + Query(_query): Query>, + state: State, +) -> Result>, (StatusCode, String)> { + let storage = state.context.services.ztm_storage.clone(); + let event_list: Vec = storage + .get_all_nostr_event() + .await + .unwrap() + .into_iter() + .map(|x| x.try_into().unwrap()) + .collect(); + Ok(Json(event_list)) +} + +pub async fn req_list( + Query(_query): Query>, + state: State, +) -> Result>, (StatusCode, String)> { + let storage = state.context.services.ztm_storage.clone(); + let req_list: Vec = storage + .get_all_nostr_req() + .await + .unwrap() + .into_iter() + .map(|x| x.into()) + .collect(); + Ok(Json(req_list)) +} diff --git a/aries/src/service/mod.rs b/aries/src/service/mod.rs index eda55b33b..a33daed54 100644 --- a/aries/src/service/mod.rs +++ b/aries/src/service/mod.rs @@ -1,5 +1,32 @@ +use callisto::ztm_nostr_req; +use gemini::nostr::client_message::Filter; +use serde::{Deserialize, Serialize}; + +pub mod api; pub mod ca_server; pub mod relay_server; +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct Req { + pub subscription_id: String, + pub filters: Vec, +} + +impl From for Req { + fn from(n: ztm_nostr_req::Model) -> Self { + let filters: Vec = serde_json::from_str(&n.filters).unwrap(); + Req { + subscription_id: n.subscription_id, + filters, + } + } +} + +impl Req { + fn filters_json(&self) -> String { + serde_json::to_string(&self.filters).unwrap() + } +} + #[cfg(test)] mod tests {} diff --git a/aries/src/service/relay_server.rs b/aries/src/service/relay_server.rs index 31fd6c3fc..b92356065 100644 --- a/aries/src/service/relay_server.rs +++ b/aries/src/service/relay_server.rs @@ -1,17 +1,18 @@ +use std::collections::HashMap; use std::net::SocketAddr; use std::str::FromStr; use std::time::{Duration, SystemTime}; -use axum::extract::{Path, Query, State}; +use axum::extract::{Query, State}; use axum::http::StatusCode; use axum::response::IntoResponse; use axum::routing::{get, post}; use axum::{Json, Router}; use callisto::{ztm_node, ztm_repo_info}; use clap::Parser; -use serde_json::Value; use common::config::Config; use gemini::ztm::hub::{LocalHub, ZTMUserPermit, ZTMCA}; +use gemini::ztm::send_get_request_to_peer_by_tunnel; use gemini::{Node, RelayGetParams, RelayResultRes, RepoInfo}; use jupiter::context::Context; use tower::ServiceBuilder; @@ -19,6 +20,8 @@ use tower_http::cors::{Any, CorsLayer}; use tower_http::decompression::RequestDecompressionLayer; use tower_http::trace::TraceLayer; +use super::api; + #[derive(Clone, Debug, Parser)] pub struct RelayOptions { #[arg(long, default_value_t = String::from("127.0.0.1"))] @@ -30,40 +33,29 @@ pub struct RelayOptions { #[arg(long, default_value_t = 8001)] pub relay_port: u16, + #[arg(long, default_value_t = 7777)] + pub ztm_agent_port: u16, + #[arg(long, default_value_t = 8888)] pub ztm_hub_port: u16, #[arg(long, default_value_t = 9999)] pub ca_port: u16, + + #[arg(long, default_value_t = false)] + pub only_agent: bool, } #[derive(Clone)] pub struct AppState { pub context: Context, - pub host: String, - pub hub_host: String, - pub relay_port: u16, - pub hub_port: u16, - pub ca_port: u16, + pub relay_option: RelayOptions, } pub async fn run_relay_server(config: Config, option: RelayOptions) { - let host = option.host.clone(); - let relay_port = option.relay_port; - let hub_host = option.hub_host; - let hub_port = option.ztm_hub_port; - let ca_port = option.ca_port; - let app = app( - config.clone(), - host.clone(), - hub_host, - relay_port, - hub_port, - ca_port, - ) - .await; + let app = app(config.clone(), option.clone()).await; - let server_url = format!("{}:{}", host, relay_port); + let server_url = format!("{}:{}", option.host, option.relay_port); tracing::info!("start relay server: {server_url}"); let addr = SocketAddr::from_str(&server_url).unwrap(); let listener = tokio::net::TcpListener::bind(addr).await.unwrap(); @@ -72,21 +64,10 @@ pub async fn run_relay_server(config: Config, option: RelayOptions) { .unwrap(); } -pub async fn app( - config: Config, - host: String, - hub_host: String, - relay_port: u16, - hub_port: u16, - ca_port: u16, -) -> Router { +pub async fn app(config: Config, relay_option: RelayOptions) -> Router { let state = AppState { - host, - hub_host, - hub_port, - relay_port, - ca_port, context: Context::new(config.clone()).await, + relay_option, }; let context = Context::new(config.clone()).await; @@ -106,9 +87,11 @@ pub fn routers() -> Router { .route("/node_list", get(node_list)) .route("/repo_provide", post(repo_provide)) .route("/repo_list", get(repo_list)) - .route("/github/webhook/:peer_id", post(github_webhook)); + .route("/test/send", get(send_message)); - Router::new().merge(router) + Router::new() + .merge(router) + .merge(api::nostr_router::routers()) } async fn hello() -> Result { @@ -119,15 +102,16 @@ async fn certificate( Query(query): Query, state: State, ) -> Result, (StatusCode, String)> { + let option = state.relay_option.clone(); if query.name.is_none() { return Err((StatusCode::BAD_REQUEST, "not enough paras".to_string())); } let name = query.name.unwrap(); let ztm: LocalHub = LocalHub { - hub_host: state.hub_host.clone(), - hub_port: state.hub_port, - ca_port: state.ca_port, + hub_host: option.hub_host, + hub_port: option.ztm_hub_port, + ca_port: option.ca_port, }; let permit = match ztm.create_ztm_certificate(name.clone()).await { Ok(p) => p, @@ -222,13 +206,35 @@ pub async fn repo_list( Ok(Json(repo_info_list_result)) } -/// Forwards the GitHub webhook to the corresponding peer. -async fn github_webhook( - Path(peer_id): Path, - Json(payload): Json, -) -> Result { - tracing::info!("GitHub Webhook Event: {:?} \n {:?}", peer_id, payload); - unimplemented!(); +async fn send_message( + Query(query): Query>, + state: State, +) -> Result, (StatusCode, String)> { + let ztm_agent_port = state.relay_option.ztm_agent_port; + let peer_id = match query.get("peer_id") { + Some(i) => i.to_string(), + None => { + return Err(( + StatusCode::BAD_REQUEST, + String::from("peer_id not provide\n"), + )); + } + }; + let path = match query.get("path") { + Some(i) => i.to_string(), + None => { + return Err((StatusCode::BAD_REQUEST, String::from("path not provide\n"))); + } + }; + let result = match send_get_request_to_peer_by_tunnel(ztm_agent_port, peer_id, path).await { + Ok(s) => s, + Err(e) => { + tracing::error!(e); + return Err((StatusCode::INTERNAL_SERVER_ERROR, e)); + } + }; + + Ok(Json(result)) } async fn loop_running(context: Context) { @@ -236,6 +242,7 @@ async fn loop_running(context: Context) { loop { check_nodes_online(context.clone()).await; + // ping_self(context.clone()).await; interval.tick().await; } } @@ -261,5 +268,26 @@ async fn check_nodes_online(context: Context) { } } +// async fn ping_self(context: Context) { +// let storage = context.services.ztm_storage.clone(); +// let nodelist: Vec = +// storage.get_all_node().await.unwrap().into_iter().collect(); +// for mut node in nodelist { +// //check online +// let from_timestamp = Duration::from_millis(node.last_online_time as u64); +// let now = SystemTime::now(); +// let elapsed = match now.duration_since(SystemTime::UNIX_EPOCH) { +// Ok(dur) => dur, +// Err(_) => { +// continue; +// } +// }; +// if elapsed.as_secs() > from_timestamp.as_secs() + 60 { +// node.online = false; +// storage.update_node(node.clone()).await.unwrap(); +// } +// } +// } + #[cfg(test)] mod tests {} diff --git a/gateway/Cargo.toml b/gateway/Cargo.toml index eab6f3a3b..f8e30346a 100644 --- a/gateway/Cargo.toml +++ b/gateway/Cargo.toml @@ -21,6 +21,7 @@ axum = { workspace = true } axum-server = { version = "0.7", features = ["tls-rustls"] } tower = { workspace = true } tracing = { workspace = true } +serde = { workspace = true } serde_json = { workspace = true } clap = { workspace = true, features = ["derive"] } tower-http = { workspace = true, features = [ diff --git a/gateway/src/api/mod.rs b/gateway/src/api/mod.rs index df07ecd11..8d23ba70d 100644 --- a/gateway/src/api/mod.rs +++ b/gateway/src/api/mod.rs @@ -2,6 +2,7 @@ use common::model::ZtmOptions; use mono::api::MonoApiServiceState; pub mod github_router; +pub mod nostr_router; pub mod ztm_router; #[derive(Clone)] diff --git a/gateway/src/api/nostr_router.rs b/gateway/src/api/nostr_router.rs new file mode 100644 index 000000000..f7fda8743 --- /dev/null +++ b/gateway/src/api/nostr_router.rs @@ -0,0 +1,165 @@ +use std::collections::HashMap; + +use axum::{ + extract::{Query, State}, + http::StatusCode, + routing::{get, post}, + Json, Router, +}; + +use common::model::CommonResult; +use gemini::{ + nostr::{event::NostrEvent, relay_message::RelayMessage, GitEvent}, + util::repo_path_to_identifier, +}; +use serde::{Deserialize, Serialize}; + +use crate::api::MegaApiServiceState; + +pub fn routers() -> Router { + Router::new() + .route("/nostr", post(recieve)) + .route("/nostr/send_event", post(send)) + .route("/nostr/event_list", get(event_list)) +} + +async fn recieve( + state: State, + body: String, +) -> Result>, (StatusCode, String)> { + // ["EVENT", , ], used to send events requested by clients. + // ["OK", , , ], used to indicate acceptance or denial of an EVENT message. + // ["EOSE", ], used to indicate the end of stored events and the beginning of events newly received in real-time. + // ["CLOSED", , ], used to indicate that a subscription was ended on the server side. + // ["NOTICE", ], used to send human-readable error messages or other things to clients. + tracing::info!("nostr recieve:{}", body); + let relay_msg: RelayMessage = match serde_json::from_str(&body) { + Ok(r) => r, + Err(e) => { + return Err((StatusCode::BAD_REQUEST, e.to_string())); + } + }; + if let RelayMessage::Event { + subscription_id, + event, + } = relay_msg + { + if subscription_id.to_string() != vault::init().0 { + return Err((StatusCode::BAD_REQUEST, String::from("bad subscription id"))); + } + //save event to database + if let Ok(ztm_nostr_event) = event.try_into() { + let _ = state + .inner + .context + .services + .ztm_storage + .insert_nostr_event(ztm_nostr_event) + .await; + }; + } + + Ok(Json(CommonResult::success(None))) +} + +#[derive(Deserialize, Serialize, Debug, Clone)] +pub struct GitEventReq { + pub path: String, + pub action: String, + pub title: String, + pub content: String, +} + +impl GitEventReq { + pub fn to_git_event(&self, identifier: String, commit: String) -> GitEvent { + GitEvent { + peer: vault::get_peerid(), + uri: identifier, + action: self.action.clone(), + r#ref: "".to_string(), + commit, + issue: "".to_string(), + mr: "".to_string(), + title: self.title.clone(), + content: self.content.clone(), + } + } +} + +async fn send( + state: State, + body: String, +) -> Result>, (StatusCode, String)> { + tracing::info!("git event recieve:{}", body); + let git_event_req: GitEventReq = match serde_json::from_str(&body) { + Ok(r) => r, + Err(e) => { + return Err((StatusCode::BAD_REQUEST, e.to_string())); + } + }; + + let git_db_storage = state.inner.context.services.git_db_storage.clone(); + let git_model = git_db_storage + .find_git_repo_exact_match(&git_event_req.path) + .await + .unwrap(); + + let git_model = match git_model.clone() { + Some(r) => r, + None => { + return Err(( + StatusCode::INTERNAL_SERVER_ERROR, + "Repo not found".to_string(), + )) + } + }; + let http_port = state.port; + let identifier = repo_path_to_identifier(http_port, git_model.clone().repo_path); + + let git_ref = git_db_storage + .get_default_ref(git_model.id) + .await + .unwrap() + .unwrap(); + + let bootstrap_node = match state.ztm.bootstrap_node.clone() { + Some(b) => b, + None => { + return Err(( + StatusCode::INTERNAL_SERVER_ERROR, + "bootstrap node is not set".to_string(), + )); + } + }; + + let git_event = git_event_req.to_git_event(identifier, git_ref.ref_git_id); + + match git_event.sent_to_relay(bootstrap_node.clone()).await { + Ok(_) => { + tracing::info!( + "send event to relay({}) successfully\n{:?}", + bootstrap_node, + git_event + ); + } + Err(e) => { + return Err((StatusCode::INTERNAL_SERVER_ERROR, e.to_string())); + } + } + Ok(Json(CommonResult::success(None))) +} + +pub async fn event_list( + Query(_query): Query>, + state: State, +) -> Result>, (StatusCode, String)> { + let storage = state.inner.context.services.ztm_storage.clone(); + let event_list: Vec = storage + .get_all_nostr_event() + .await + .unwrap() + .into_iter() + .map(|x| x.try_into().unwrap()) + .collect(); + Ok(Json(event_list)) +} diff --git a/gateway/src/api/ztm_router.rs b/gateway/src/api/ztm_router.rs index 50a09b9bb..28d491f49 100644 --- a/gateway/src/api/ztm_router.rs +++ b/gateway/src/api/ztm_router.rs @@ -8,13 +8,16 @@ use axum::{ }; use common::model::CommonResult; +use gemini::nostr::subscribe_git_event; +use vault::get_peerid; use crate::api::MegaApiServiceState; pub fn routers() -> Router { Router::new() .route("/ztm/repo_provide", get(repo_provide)) - .route("/ztm/repo_fork", get(repo_fork)) + .route("/ztm/repo_fork", get(repo_folk)) + .route("/ztm/hello", get(hello)) } async fn repo_provide( @@ -50,7 +53,7 @@ async fn repo_provide( Ok(Json(res)) } -async fn repo_fork( +async fn repo_folk( Query(query): Query>, state: State, ) -> Result>, (StatusCode, String)> { @@ -76,9 +79,9 @@ async fn repo_fork( } }; - let res = gemini::http::handler::repo_fork( + let res = gemini::http::handler::repo_folk( state.ztm.ztm_agent_port, - identifier.to_string(), + identifier.clone().to_string(), local_port, ) .await; @@ -86,5 +89,20 @@ async fn repo_fork( Ok(data) => CommonResult::success(Some(data)), Err(err) => CommonResult::failed(&err.to_string()), }; + + //nostr subscribe to Events + if let Some(bootstrap_node) = state.ztm.bootstrap_node.clone() { + let _ = subscribe_git_event(identifier.to_string(), get_peerid(), bootstrap_node).await; + } + Ok(Json(res)) } + +async fn hello( + Query(_query): Query>, + _state: State, +) -> Result>, (StatusCode, String)> { + let (peer_id, _) = vault::init(); + let msg = format!("hello from {peer_id}"); + Ok(Json(CommonResult::success(Some(msg)))) +} diff --git a/gateway/src/https_server.rs b/gateway/src/https_server.rs index 2d510c1f5..37f6b2871 100644 --- a/gateway/src/https_server.rs +++ b/gateway/src/https_server.rs @@ -22,7 +22,7 @@ use mono::server::https_server::{ get_method_router, post_method_router, put_method_router, AppState, }; -use crate::api::{github_router, ztm_router, MegaApiServiceState}; +use crate::api::{github_router, nostr_router, ztm_router, MegaApiServiceState}; #[derive(Args, Clone, Debug)] pub struct HttpOptions { @@ -63,7 +63,7 @@ pub async fn https_server(config: Config, options: HttpsOptions) { ztm, } = options.clone(); - check_run_with_ztm(config.clone(), options.ztm.clone()); + check_run_with_ztm(config.clone(), options.ztm.clone(), https_port); let app = app( config, @@ -92,7 +92,7 @@ pub async fn http_server(config: Config, options: HttpOptions) { ztm, } = options.clone(); - check_run_with_ztm(config.clone(), options.ztm.clone()); + check_run_with_ztm(config.clone(), options.ztm.clone(), http_port); let app = app( config, @@ -145,6 +145,7 @@ pub async fn app( pub fn mega_routers() -> Router { Router::new() .merge(ztm_router::routers()) + .merge(nostr_router::routers()) .merge(github_router::routers()) } @@ -178,7 +179,7 @@ pub async fn app( .with_state(state) } -pub fn check_run_with_ztm(config: Config, ztm: ZtmOptions) { +pub fn check_run_with_ztm(config: Config, ztm: ZtmOptions, http_port: u16) { //Mega server join a ztm mesh match ztm.bootstrap_node { Some(bootstrap_node) => { @@ -193,11 +194,11 @@ pub fn check_run_with_ztm(config: Config, ztm: ZtmOptions) { ztm_agent.clone().start_ztm_agent(); thread::sleep(time::Duration::from_secs(3)); tokio::spawn(async move { - run_ztm_client(bootstrap_node, config, peer_id, ztm_agent).await + run_ztm_client(bootstrap_node, config, peer_id, ztm_agent, http_port).await }); } None => { - tracing::info!("The bootstrap node is not set, prepare to start mega sever locally"); + tracing::info!("The bootstrap node is not set, prepare to start mega server locally"); } }; } diff --git a/gemini/Cargo.toml b/gemini/Cargo.toml index 460d0a573..73da500be 100644 --- a/gemini/Cargo.toml +++ b/gemini/Cargo.toml @@ -23,4 +23,5 @@ jupiter = { workspace = true } vault = { workspace = true } callisto = { workspace = true } ceres = { workspace = true } -neptune = { workspace = true } \ No newline at end of file +neptune = { workspace = true } +secp256k1 = { version = "0.29.0", features = ["serde", "rand","hashes"] } \ No newline at end of file diff --git a/gemini/src/http/handler.rs b/gemini/src/http/handler.rs index f71b3be8a..b92ea88dd 100644 --- a/gemini/src/http/handler.rs +++ b/gemini/src/http/handler.rs @@ -1,9 +1,9 @@ use jupiter::context::Context; -use ceres::protocol::repo::Repo; use crate::{ - ztm::agent::{share_repo, LocalZTMAgent, ZTMAgent}, - RepoInfo, ZTM_APP_PROVIDER, + util::{get_short_peer_id, repo_path_to_identifier}, + ztm::{agent::share_repo, create_tunnel}, + RepoInfo, }; pub async fn repo_provide( @@ -29,11 +29,10 @@ pub async fn repo_provide( } Err(_) => return Err(String::from("Repo not found")), }; - let repo: Repo = git_model.clone().into(); let git_ref = context .services .git_db_storage - .get_default_ref(repo.repo_id) + .get_default_ref(git_model.id) .await .unwrap() .unwrap(); @@ -41,12 +40,12 @@ pub async fn repo_provide( let name = git_model.repo_name; let repo_path = git_model.repo_path; let (peer_id, _) = vault::init(); - let identifier = format!("p2p://{}/{port}{repo_path}.git", peer_id.clone()); + let identifier = repo_path_to_identifier(port, repo_path); let update_time = git_model.created_at.and_utc().timestamp(); let repo_info = RepoInfo { name, identifier, - origin: peer_id.clone(), + origin: peer_id, update_time, commit: git_ref.ref_git_id, peer_online: true, @@ -55,7 +54,7 @@ pub async fn repo_provide( Ok("success".to_string()) } -pub async fn repo_fork( +pub async fn repo_folk( ztm_agent_port: u16, identifier: String, local_port: u16, @@ -73,63 +72,25 @@ pub async fn repo_fork( Err(e) => return Err(e), }; - let agent: LocalZTMAgent = LocalZTMAgent { - agent_port: ztm_agent_port, - }; - let local_ep = match agent.get_ztm_local_endpoint().await { - Ok(ep) => ep, - Err(e) => return Err(e), - }; - - let remote_ep = match agent.get_ztm_remote_endpoint(remote_peer_id.clone()).await { - Ok(ep) => ep, - Err(e) => return Err(e), - }; - let (peer_id, _) = vault::init(); let bound_name = format!( "{}_{}", get_short_peer_id(peer_id), - get_short_peer_id(remote_peer_id) + get_short_peer_id(remote_peer_id.clone()) ); - //creata inbound - match agent - .create_ztm_app_tunnel_inbound( - local_ep.id, - ZTM_APP_PROVIDER.to_string(), - "tunnel".to_string(), - bound_name.clone(), - local_port, - ) - .await + match create_tunnel( + ztm_agent_port, + remote_peer_id, + local_port, + remote_port, + bound_name, + ) + .await { Ok(_) => (), - Err(s) => { - tracing::error!("create app inbound, {s}"); - return Err(s); - } + Err(e) => return Err(e), } - tracing::info!("create app inbound successfully"); - //creata outbound - match agent - .create_ztm_app_tunnel_outbound( - remote_ep.id, - ZTM_APP_PROVIDER.to_string(), - "tunnel".to_string(), - bound_name, - remote_port, - ) - .await - { - Ok(msg) => { - tracing::info!("create app outbound successfully,{}", msg); - } - Err(s) => { - tracing::error!("create app outbound, {s}"); - return Err(s); - } - } let msg = format!("git clone http://localhost:{local_port}/{git_path}"); Ok(msg) } @@ -165,12 +126,5 @@ pub fn get_git_path_from_identifier(identifier: String) -> Result String { - if peer_id.len() <= 7 { - return peer_id; - } - peer_id[0..7].to_string() -} - #[cfg(test)] mod tests {} diff --git a/gemini/src/lib.rs b/gemini/src/lib.rs index 826179cd5..a035aad1c 100644 --- a/gemini/src/lib.rs +++ b/gemini/src/lib.rs @@ -6,6 +6,8 @@ use serde::{Deserialize, Serialize}; pub mod ca; pub mod http; +pub mod nostr; +pub mod util; pub mod ztm; const ZTM_APP_PROVIDER: &str = "mega"; @@ -17,6 +19,7 @@ pub struct RelayGetParams { pub name: Option, pub agent_name: Option, pub service_name: Option, + pub service_port: Option, } #[derive(Serialize, Deserialize, Debug)] @@ -48,6 +51,7 @@ pub struct Node { pub mega_type: String, pub online: bool, pub last_online_time: i64, + pub service_port: i32, } #[derive(Debug)] @@ -63,6 +67,7 @@ impl TryFrom for Node { || paras.hub.is_none() || paras.agent_name.is_none() || paras.service_name.is_none() + || paras.service_port.is_none() { return Err(ConversionError::InvalidParas); } @@ -75,6 +80,7 @@ impl TryFrom for Node { mega_type: MegaType::Agent.to_string(), online: true, last_online_time: now, + service_port: paras.service_port.unwrap(), }) } } @@ -87,6 +93,7 @@ impl TryFrom for ztm_node::Model { || paras.hub.is_none() || paras.agent_name.is_none() || paras.service_name.is_none() + || paras.service_port.is_none() { return Err(ConversionError::InvalidParas); } @@ -99,6 +106,7 @@ impl TryFrom for ztm_node::Model { r#type: MegaType::Agent.to_string(), online: true, last_online_time: now, + service_port: paras.service_port.unwrap(), }) } } @@ -115,6 +123,7 @@ impl TryFrom for ztm_node::Model { r#type: n.mega_type, online: n.online, last_online_time: n.last_online_time, + service_port: n.service_port, }) } } @@ -129,6 +138,7 @@ impl From for Node { mega_type: n.r#type, online: n.online, last_online_time: n.last_online_time, + service_port: n.service_port, } } } diff --git a/gemini/src/nostr/client_message.rs b/gemini/src/nostr/client_message.rs new file mode 100644 index 000000000..84f88023b --- /dev/null +++ b/gemini/src/nostr/client_message.rs @@ -0,0 +1,429 @@ +use secp256k1::hashes::{sha256, Hash}; +use secp256k1::rand::rngs::OsRng; +use secp256k1::rand::RngCore; +use secp256k1::XOnlyPublicKey; +use serde::de::{MapAccess, Visitor}; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use std::collections::HashMap; +use std::fmt; + +use crate::nostr::event::{EventId, NostrEvent}; +use crate::nostr::kind::NostrKind; +use crate::nostr::tag::TagKind; +use crate::nostr::MessageHandleError; +use serde::ser::SerializeMap; +use serde_json::{json, Value}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum ClientMessage { + /// Event + Event(NostrEvent), + /// Req + Req { + /// Subscription ID + subscription_id: SubscriptionId, + /// Filters + filters: Vec, + }, +} + +impl Serialize for ClientMessage { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + let json_value: Value = self.as_value(); + json_value.serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for ClientMessage { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let json_value = Value::deserialize(deserializer)?; + ClientMessage::from_value(json_value).map_err(serde::de::Error::custom) + } +} + +impl ClientMessage { + /// Create new `EVENT` message + pub fn new_event(event: NostrEvent) -> Self { + Self::Event(event) + } + + /// Create new `REQ` message + pub fn new_req(subscription_id: SubscriptionId, filters: Vec) -> Self { + Self::Req { + subscription_id, + filters, + } + } + + /// Check if is an `EVENT` message + pub fn is_event(&self) -> bool { + matches!(self, ClientMessage::Event(_)) + } + + /// Check if is an `REQ` message + pub fn is_req(&self) -> bool { + matches!(self, ClientMessage::Req { .. }) + } + + /// Serialize as [`Value`] + pub fn as_value(&self) -> Value { + match self { + Self::Event(event) => json!(["EVENT", event]), + Self::Req { + subscription_id, + filters, + } => { + let mut json = json!(["REQ", subscription_id]); + let mut filters = json!(filters); + + if let Some(json) = json.as_array_mut() { + if let Some(filters) = filters.as_array_mut() { + json.append(filters); + } + } + + json + } + } + } + + /// Deserialize from [`Value`] + /// + /// **This method NOT verify the event signature!** + pub fn from_value(msg: Value) -> Result { + let v = msg + .as_array() + .ok_or(MessageHandleError::InvalidMessageFormat)?; + + if v.is_empty() { + return Err(MessageHandleError::InvalidMessageFormat); + } + + let v_len: usize = v.len(); + + // Event + // ["EVENT", ] + if v[0] == "EVENT" { + if v_len != 2 { + return Err(MessageHandleError::InvalidMessageFormat); + } + let event = NostrEvent::from_value(v[1].clone())?; + return Ok(Self::new_event(event)); + } + + // Req + // ["REQ", , , ...] + if v[0] == "REQ" { + if v_len == 2 { + let subscription_id: SubscriptionId = serde_json::from_value(v[1].clone())?; + return Ok(Self::new_req(subscription_id, Vec::new())); + } else if v_len >= 3 { + let subscription_id: SubscriptionId = serde_json::from_value(v[1].clone())?; + let filters: Vec = serde_json::from_value(Value::Array(v[2..].to_vec()))?; + return Ok(Self::new_req(subscription_id, filters)); + } else { + return Err(MessageHandleError::InvalidMessageFormat); + } + } + + Err(MessageHandleError::InvalidMessageFormat) + } + + pub fn as_json(&self) -> String { + json!(self).to_string() + } +} + +/// Subscription ID +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub struct SubscriptionId(String); + +impl SubscriptionId { + /// Create new [`SubscriptionId`] + pub fn new(id: S) -> Self + where + S: Into, + { + Self(id.into()) + } + + /// Generate new random [`SubscriptionId`] + pub fn generate() -> Self { + let mut rng = OsRng; + Self::generate_with_rng(&mut rng) + } + /// Generate new random [`SubscriptionId`] + pub fn generate_with_rng(rng: &mut R) -> Self + where + R: RngCore, + { + let mut os_random = [0u8; 32]; + rng.fill_bytes(&mut os_random); + let hash = sha256::Hash::hash(&os_random).to_string(); + Self::new(&hash[..32]) + } +} + +impl fmt::Display for SubscriptionId { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0) + } +} + +impl Serialize for SubscriptionId { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_str(&self.to_string()) + } +} + +impl<'de> Deserialize<'de> for SubscriptionId { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let value = Value::deserialize(deserializer)?; + let id: String = serde_json::from_value(value).map_err(serde::de::Error::custom)?; + Ok(Self::new(id)) + } +} + +/// Subscription filters +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct Filter { + /// List of [`EventId`] + #[serde(skip_serializing_if = "Vec::is_empty")] + #[serde(default)] + pub ids: Vec, + /// List of [`XOnlyPublicKey`] + #[serde(skip_serializing_if = "Vec::is_empty")] + #[serde(default)] + pub authors: Vec, + /// List of a kind numbers + #[serde(skip_serializing_if = "Vec::is_empty")] + #[serde(default)] + pub kinds: Vec, + /// It's a string describing a query in a human-readable form, i.e. "best nostr apps" + /// + /// + #[serde(skip_serializing_if = "Option::is_none")] + #[serde(default)] + pub search: Option, + /// An integer unix timestamp, events must be newer than this to pass + #[serde(skip_serializing_if = "Option::is_none")] + #[serde(default)] + pub since: Option, + /// An integer unix timestamp, events must be older than this to pass + #[serde(skip_serializing_if = "Option::is_none")] + #[serde(default)] + pub until: Option, + /// Maximum number of events to be returned in the initial query + #[serde(skip_serializing_if = "Option::is_none")] + #[serde(default)] + pub limit: Option, + /// Generic tag queries (NIP12) + #[serde( + flatten, + serialize_with = "serialize_generic_tags", + deserialize_with = "deserialize_generic_tags" + )] + #[serde(default)] + pub generic_tags: HashMap>, +} + +impl Filter { + pub fn new() -> Self { + Self::default() + } + + pub fn author(mut self, author: XOnlyPublicKey) -> Self { + self.authors.push(author); + self + } + pub fn kind(mut self, kind: NostrKind) -> Self { + self.kinds.push(kind); + self + } + + pub fn pubkey(self, pubkey: XOnlyPublicKey) -> Self { + self.custom_tag(TagKind::P.to_string(), vec![pubkey.to_string()]) + } + + pub fn get_pubkey(self) -> Vec { + if let Some(pks) = self.generic_tags.get("#p") { + return pks.clone(); + } + Vec::new() + } + + pub fn peer_id(self, peer_id: String) -> Self { + self.custom_tag(String::from("peer"), vec![peer_id]) + } + + pub fn repo_uri(self, repo_uri: String) -> Self { + self.custom_tag(TagKind::URI.into(), vec![repo_uri]) + } + + pub fn custom_tag(mut self, tag: String, values: Vec) -> Self + where + S: Into, + { + let values: Vec = values.into_iter().map(|value| value.into()).collect(); + self.generic_tags + .entry(tag) + .and_modify(|list| { + for value in values.clone().into_iter() { + list.push(value); + } + }) + .or_insert(values); + self + } +} + +fn serialize_generic_tags( + generic_tags: &HashMap>, + serializer: S, +) -> Result +where + S: Serializer, +{ + let mut map = serializer.serialize_map(Some(generic_tags.len()))?; + for (tag, values) in generic_tags.iter() { + map.serialize_entry(&format!("#{tag}"), values)?; + } + map.end() +} + +fn deserialize_generic_tags<'de, D>( + deserializer: D, +) -> Result>, D::Error> +where + D: Deserializer<'de>, +{ + struct GenericTagsVisitor; + + impl<'de> Visitor<'de> for GenericTagsVisitor { + type Value = HashMap>; + + fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result { + formatter.write_str("map in which the keys are \"#X\" for some character X") + } + + fn visit_map(self, mut map: M) -> Result + where + M: MapAccess<'de>, + { + let mut generic_tags = HashMap::new(); + while let Some(key) = map.next_key::()? { + let mut chars = key.chars().clone(); + if let (Some('#'), Some(ch), None) = (chars.next(), chars.next(), chars.next()) { + let tag: String = String::from(ch); + let values = map.next_value()?; + generic_tags.insert(tag, values); + } else { + let (_a, b) = key.split_at(1); + let values = map.next_value()?; + generic_tags.insert(b.to_string(), values); + } + } + Ok(generic_tags) + } + } + + deserializer.deserialize_map(GenericTagsVisitor) +} + +#[cfg(test)] +mod tests { + use core::str::FromStr; + + use secp256k1::Secp256k1; + + use crate::nostr::tag::Tag; + + use super::*; + + #[test] + fn test_client_message_req() { + let pk = XOnlyPublicKey::from_str( + "379e863e8357163b5bce5d2688dc4f1dcc2d505222fb8d74db600f30535dfdfe", + ) + .unwrap(); + let filters = vec![ + Filter::new().kind(NostrKind::Mega), + Filter::new().pubkey(pk), + ]; + + let client_req = ClientMessage::new_req(SubscriptionId::new("test"), filters); + assert_eq!( + client_req.as_json(), + r##"["REQ","test",{"kinds":[111]},{"#p":["379e863e8357163b5bce5d2688dc4f1dcc2d505222fb8d74db600f30535dfdfe"]}]"## + ); + } + + #[test] + fn test_client_message_from_value() { + let req = json!(["REQ","8aff61479fb8e406a0c29dee13981db6",{"#P":["16Uiu2HAmJMy5xuyCnsKcGESmEar15Ca7c5abG8TadqrqfeHub2tS"]}]); + + let msg = ClientMessage::from_value(req.clone()).unwrap(); + + assert_eq!(msg.as_value(), req); + } + + #[test] + fn test_client_message_event() { + let sk = "6b911fd37cdf5c81d4c0adb1ab7fa822ed253ab0ad9aa18d77257c88b29b718e"; + let secp = Secp256k1::new(); + let keypair = secp256k1::Keypair::from_seckey_str(&secp, sk).unwrap(); + let tag = Tag::Generic(TagKind::P, Vec::new()); + let tags: Vec = vec![tag]; + let content = "123".to_string(); + let event = NostrEvent::new_with_timestamp(keypair, 1111, NostrKind::Mega, tags, content); + let client_msg = ClientMessage::new_event(event); + let json = r#"[ + "EVENT", + { + "content": "123", + "created_at": 1111, + "id": "bc3cc3a1499ea0c618df6dc5a300e8ad05f4d3a9a96e40e8a237cec11cc78667", + "kind": 111, + "pubkey": "385c3a6ec0b9d57a4330dbd6284989be5bd00e41c535f9ca39b6ae7c521b81cd", + "sig": "1285a0697eb44bf41f7a1bb063646ac47de8f6664335b791fe4982b1c9cdae1f56d0f1dd0564aeb053fd3cab16c4a0508cef70ac0bec9d79db3f2a2c28426f20", + "tags": [ + [ + "p" + ] + ] + } + ]"#; + let value = serde_json::from_str::(json).unwrap(); + assert_eq!(client_msg.as_json(), value.to_string()); + } + + #[test] + fn test_client_message_git_req() { + let filters = vec![Filter::new().repo_uri( + "p2p://yfeunFhgJGD83pcB4nXjif9eePeLEmQXP17XjQjFXN4c/8000/third-part/test.git" + .to_string(), + )]; + + let client_req = ClientMessage::new_req( + SubscriptionId::new("yfeunFhgJGD83pcB4nXjif9eePeLEmQXP17XjQjFXN4c"), + filters, + ); + // println!("{}", client_req.as_json()); + assert_eq!( + client_req.as_json(), + r##"["REQ","yfeunFhgJGD83pcB4nXjif9eePeLEmQXP17XjQjFXN4c",{"#uri":["p2p://yfeunFhgJGD83pcB4nXjif9eePeLEmQXP17XjQjFXN4c/8000/third-part/test.git"]}]"## + ); + } +} diff --git a/gemini/src/nostr/event.rs b/gemini/src/nostr/event.rs new file mode 100644 index 000000000..4f38d0a1c --- /dev/null +++ b/gemini/src/nostr/event.rs @@ -0,0 +1,426 @@ +use crate::nostr::kind::NostrKind; +use crate::nostr::tag::Tag; +use crate::util::get_utc_timestamp; +use callisto::ztm_nostr_event; +use secp256k1::hashes::hex::HexToArrayError; +use secp256k1::hashes::{sha256, Hash}; +use secp256k1::schnorr::Signature; +use secp256k1::{rand, Message, Secp256k1}; +use secp256k1::{Keypair, XOnlyPublicKey}; +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use std::fmt; +use std::str::FromStr; +use vault::init; + +use super::GitEvent; + +/// [`NostrEvent`] error +#[derive(Debug)] +pub enum Error { + /// Invalid signature + InvalidSignature, + /// Invalid event id + InvalidId, + /// Error serializing or deserializing JSON data + Json(serde_json::Error), + /// Secp256k1 error + Secp256k1(secp256k1::Error), + /// Hex decoding error + Hex(HexToArrayError), +} +#[derive(Debug)] +pub enum ConversionError { + InvalidParas, + InvalidPubkey, + InvalidSignature, + JsonError, +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidSignature => write!(f, "Invalid signature"), + Self::InvalidId => write!(f, "Invalid event id"), + Self::Json(e) => write!(f, "Json: {e}"), + Self::Secp256k1(e) => write!(f, "Secp256k1: {e}"), + Self::Hex(e) => write!(f, "Hex: {e}"), + } + } +} + +impl From for Error { + fn from(e: serde_json::Error) -> Self { + Self::Json(e) + } +} + +impl From for Error { + fn from(e: secp256k1::Error) -> Self { + Self::Secp256k1(e) + } +} + +impl From for Error { + fn from(e: HexToArrayError) -> Self { + Self::Hex(e) + } +} + +/// nostr event. +#[derive(Serialize, Deserialize, PartialEq, Eq, Debug, Clone)] +pub struct NostrEvent { + pub id: EventId, + pub pubkey: XOnlyPublicKey, + pub created_at: i64, + pub kind: NostrKind, + pub tags: Vec, + pub content: String, + pub sig: Signature, +} +impl TryFrom for ztm_nostr_event::Model { + type Error = ConversionError; + + fn try_from(n: NostrEvent) -> Result { + let tags_string = match serde_json::to_string(&n.tags) { + Ok(s) => s, + Err(_) => { + return Err(ConversionError::JsonError); + } + }; + Ok(ztm_nostr_event::Model { + id: n.id.0, + pubkey: n.pubkey.to_string(), + created_at: n.created_at, + kind: n.kind.as_u32() as i32, + tags: tags_string, + content: n.content, + sig: n.sig.to_string(), + }) + } +} + +impl TryFrom for NostrEvent { + type Error = ConversionError; + + fn try_from(n: ztm_nostr_event::Model) -> Result { + let pk = match XOnlyPublicKey::from_str(&n.pubkey) { + Ok(pk) => pk, + Err(_) => { + return Err(ConversionError::InvalidPubkey); + } + }; + let tags: Vec = match serde_json::from_str(&n.tags) { + Ok(tags) => tags, + Err(_) => { + return Err(ConversionError::JsonError); + } + }; + let sig = match Signature::from_str(&n.sig) { + Ok(sig) => sig, + Err(_) => { + return Err(ConversionError::InvalidSignature); + } + }; + Ok(NostrEvent { + id: EventId(n.id), + pubkey: pk, + created_at: n.created_at, + kind: NostrKind::from(n.kind as u64), + tags, + content: n.content, + sig, + }) + } +} + +impl NostrEvent { + pub fn new(tags: Vec, content: String) -> Self { + let created_at = get_utc_timestamp(); + + let (_, sk) = init(); + let secp = Secp256k1::new(); + let keypair = secp256k1::Keypair::from_seckey_str(&secp, &sk).unwrap(); + + Self::new_with_timestamp(keypair, created_at, NostrKind::Mega, tags, content) + } + + pub fn new_with_keypair( + keypair: Keypair, + kind: NostrKind, + tags: Vec, + content: String, + ) -> Self { + let created_at = get_utc_timestamp(); + Self::new_with_timestamp(keypair, created_at, kind, tags, content) + } + + pub fn new_with_timestamp( + keypair: Keypair, + created_at: i64, + kind: NostrKind, + tags: Vec, + content: String, + ) -> Self { + let (pubkey, _) = XOnlyPublicKey::from_keypair(&keypair); + let json: Value = json!([0, pubkey, created_at, kind, tags, content]); + let event_str: String = json.to_string(); + let id = EventId(sha256::Hash::hash(event_str.as_bytes()).to_string()); + NostrEvent { + id: id.clone(), + pubkey, + created_at, + kind, + tags, + content, + sig: sign_without_rng(id.inner(), &keypair), + } + } + + pub fn new_git_event(keypair: Keypair, git_event: GitEvent) -> Self { + let tags: Vec = git_event.to_tags(); + NostrEvent::new_with_keypair(keypair, NostrKind::Mega, tags, git_event.content) + } + pub fn new_git_event_with_timestamp( + keypair: Keypair, + created_at: i64, + git_event: GitEvent, + ) -> Self { + let tags: Vec = git_event.to_tags(); + NostrEvent::new_with_timestamp( + keypair, + created_at, + NostrKind::Mega, + tags, + git_event.content, + ) + } + + pub fn from_json(json: T) -> Result + where + T: AsRef<[u8]>, + { + Ok(serde_json::from_slice(json.as_ref())?) + } + pub fn as_json(&self) -> String { + json!(self).to_string() + } + + pub fn from_value(value: Value) -> Result { + Ok(serde_json::from_value(value)?) + } + + /// Verify [`EventId`] and [`Signature`] + pub fn verify(&self) -> Result<(), Error> { + // Verify ID + self.verify_id()?; + + // Verify signature + self.verify_signature() + } + + /// Verify if the [`EventId`] it's composed correctly + pub fn verify_id(&self) -> Result<(), Error> { + let id: EventId = EventId::new( + self.pubkey, + self.created_at, + self.kind, + self.tags.clone(), + self.content.clone(), + ); + if id == self.id { + Ok(()) + } else { + Err(Error::InvalidId) + } + } + + /// Verify event [`Signature`] + pub fn verify_signature(&self) -> Result<(), Error> { + let secp = Secp256k1::new(); + let hash = sha256::Hash::from_str(self.id.inner().clone().as_str())?; + let message: Message = Message::from_digest(hash.to_byte_array()); + // let message = Message::from_slice(hash.as_ref())?; + secp.verify_schnorr(&self.sig, &message, &self.pubkey) + .map_err(|_| Error::InvalidSignature) + } +} + +pub fn sign_with_rng(id: String, keypair: &Keypair) -> Signature { + let secp = Secp256k1::new(); + let mut rng = rand::thread_rng(); + let hash = sha256::Hash::from_str(id.as_str()).unwrap(); + let message: Message = Message::from_digest(hash.to_byte_array()); + secp.sign_schnorr_with_rng(&message, keypair, &mut rng) +} + +pub fn sign_without_rng(id: String, keypair: &Keypair) -> Signature { + let secp = Secp256k1::new(); + let hash = sha256::Hash::from_str(id.as_str()).unwrap(); + let message: Message = Message::from_digest(hash.to_byte_array()); + secp.sign_schnorr_no_aux_rand(&message, keypair) +} + +/// Event Id +/// +/// 32-bytes lowercase hex-encoded sha256 of the the serialized event data +/// +/// +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +pub struct EventId(String); + +impl EventId { + ///Generate [`EventId`] + pub fn new( + pubkey: XOnlyPublicKey, + created_at: i64, + kind: NostrKind, + tags: Vec, + content: String, + ) -> Self { + let json: Value = json!([0, pubkey, created_at, kind, tags, content]); + let event_str: String = json.to_string(); + let id = sha256::Hash::hash(event_str.as_bytes()).to_string(); + Self(id) + } + + /// Get [`EventId`] as [`String`] + pub fn inner(&self) -> String { + self.0.clone() + } +} + +#[cfg(test)] +mod tests { + + use secp256k1::{ + hashes::{sha256, Hash}, + rand::{self, rngs::OsRng}, + Keypair, Message, Secp256k1, + }; + use serde_json::Value; + + use crate::nostr::{ + event::{sign_without_rng, EventId, NostrEvent}, + kind::NostrKind, + tag::{Tag, TagKind}, + GitEvent, + }; + #[test] + fn test_secp256k1() { + let secp = Secp256k1::new(); + let (secret_key, public_key) = secp.generate_keypair(&mut OsRng); + let digest = sha256::Hash::hash("Hello World!".as_bytes()); + let message: Message = Message::from_digest(digest.to_byte_array()); + + let sig = secp.sign_ecdsa(&message, &secret_key); + assert!(secp.verify_ecdsa(&message, &sig, &public_key).is_ok()); + } + + #[test] + fn test_new_nostr_id() { + let sk = "6b911fd37cdf5c81d4c0adb1ab7fa822ed253ab0ad9aa18d77257c88b29b718e"; + let secp = Secp256k1::new(); + let keypair = secp256k1::Keypair::from_seckey_str(&secp, sk).unwrap(); + let pk = keypair.x_only_public_key().0; + let tag = Tag::Generic(TagKind::P, Vec::new()); + let tags: Vec = vec![tag]; + let event_id = EventId::new(pk, 123, NostrKind::Mega, tags.clone(), "123".to_string()); + let event_id_right = "bc0d7c8a5c00eff8719c68f6df7f7e31b2a118ba5221807fe850ba689854f467"; + assert_eq!(event_id.inner(), event_id_right); + } + + #[test] + fn test_new_nostr_event() { + let sk = "6b911fd37cdf5c81d4c0adb1ab7fa822ed253ab0ad9aa18d77257c88b29b718e"; + let secp = Secp256k1::new(); + let keypair = secp256k1::Keypair::from_seckey_str(&secp, sk).unwrap(); + let tag = Tag::Generic(TagKind::P, Vec::new()); + let tags: Vec = vec![tag]; + let content = "123".to_string(); + let event = NostrEvent::new_with_timestamp(keypair, 1111, NostrKind::Mega, tags, content); + assert_eq!( + event.as_json(), + r#"{"content":"123","created_at":1111,"id":"bc3cc3a1499ea0c618df6dc5a300e8ad05f4d3a9a96e40e8a237cec11cc78667","kind":111,"pubkey":"385c3a6ec0b9d57a4330dbd6284989be5bd00e41c535f9ca39b6ae7c521b81cd","sig":"1285a0697eb44bf41f7a1bb063646ac47de8f6664335b791fe4982b1c9cdae1f56d0f1dd0564aeb053fd3cab16c4a0508cef70ac0bec9d79db3f2a2c28426f20","tags":[["p"]]}"# + ); + } + + #[test] + fn test_verify_nostr_event() { + let secp = Secp256k1::new(); + let (secret_key, _) = secp.generate_keypair(&mut rand::thread_rng()); + let keypair = Keypair::from_secret_key(&secp, &secret_key); + let (xonly, _) = keypair.x_only_public_key(); + + let kind = NostrKind::Mega; + let tag = Tag::Generic(TagKind::P, Vec::from(["123".into(), "234".into()])); + let tags: Vec = vec![tag]; + let event_id = EventId::new(xonly, 123, kind, tags.clone(), "123".to_string()); + + let signature = sign_without_rng(event_id.inner(), &keypair); + let event = NostrEvent { + id: event_id.clone(), + pubkey: xonly, + created_at: 123, + kind, + tags: tags.clone(), + content: "123".into(), + sig: signature, + }; + let result = event.verify(); + assert!(result.is_ok()); + } + + #[test] + fn test_nostr_event_git() { + let sk = "6b911fd37cdf5c81d4c0adb1ab7fa822ed253ab0ad9aa18d77257c88b29b718e"; + let secp = Secp256k1::new(); + let keypair = secp256k1::Keypair::from_seckey_str(&secp, sk).unwrap(); + + let git_event = GitEvent { + peer: "yfeunFhgJGD83pcB4nXjif9eePeLEmQXP17XjQjFXN4c".to_string(), + uri: "p2p://yfeunFhgJGD83pcB4nXjif9eePeLEmQXP17XjQjFXN4c/8000/third-part/test.git" + .to_string(), + action: "repo_update".to_string(), + r#ref: "".to_string(), + commit: "93e1191cdaa97cb6182fb906f4dea4300db3c734".to_string(), + issue: "".to_string(), + mr: "".to_string(), + title: "hahaha".to_string(), + content: "hello".to_string(), + }; + let event = NostrEvent::new_git_event_with_timestamp(keypair, 123, git_event); + let json = r#" { + "content": "hello", + "created_at": 123, + "id": "829043ebaf5e8a5060b2f8bd0dc0fff2520607c4864c8dd99c7805c547f295ba", + "kind": 111, + "pubkey": "385c3a6ec0b9d57a4330dbd6284989be5bd00e41c535f9ca39b6ae7c521b81cd", + "sig": "c1cad051321e89579e49e83907bd12021e45fac2dd5d63a4396d145f593b9ce339e8ecad771209059335c491661338ea9c58b17949c955d4be1d6d1ac82742f2", + "tags": [ + [ + "peer", + "yfeunFhgJGD83pcB4nXjif9eePeLEmQXP17XjQjFXN4c" + ], + [ + "uri", + "p2p://yfeunFhgJGD83pcB4nXjif9eePeLEmQXP17XjQjFXN4c/8000/third-part/test.git" + ], + [ + "action", + "repo_update" + ], + [ + "commit", + "93e1191cdaa97cb6182fb906f4dea4300db3c734" + ], + [ + "title", + "hahaha" + ] + ] + } "#; + let value = serde_json::from_str::(json).unwrap(); + assert_eq!(event.as_json(), value.to_string()); + } +} diff --git a/gemini/src/nostr/kind.rs b/gemini/src/nostr/kind.rs new file mode 100644 index 000000000..ac492722a --- /dev/null +++ b/gemini/src/nostr/kind.rs @@ -0,0 +1,101 @@ +use serde::de::{Error, Visitor}; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use std::fmt; +use std::hash::{Hash, Hasher}; + +/// Event [`NostrKind`] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum NostrKind { + /// Metadata (NIP01 and NIP05) + Metadata, + /// Short Text Note (NIP01) + TextNote, + /// for mega p2p use + Mega, + /// Custom + Custom(u64), +} + +impl NostrKind { + /// Get [`NostrKind`] as `u32` + pub fn as_u32(&self) -> u32 { + self.as_u64() as u32 + } + + /// Get [`NostrKind`] as `u64` + pub fn as_u64(&self) -> u64 { + (*self).into() + } +} + +impl fmt::Display for NostrKind { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.as_u64()) + } +} + +impl From for NostrKind { + fn from(u: u64) -> Self { + match u { + 0 => Self::Metadata, + 1 => Self::TextNote, + 111 => Self::Mega, + x => Self::Custom(x), + } + } +} + +impl From for u64 { + fn from(e: NostrKind) -> u64 { + match e { + NostrKind::Metadata => 0, + NostrKind::TextNote => 1, + NostrKind::Mega => 111, + NostrKind::Custom(u) => u, + } + } +} + +impl Hash for NostrKind { + fn hash(&self, state: &mut H) + where + H: Hasher, + { + self.as_u64().hash(state); + } +} + +impl Serialize for NostrKind { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + serializer.serialize_u64(From::from(*self)) + } +} + +impl<'de> Deserialize<'de> for NostrKind { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + deserializer.deserialize_u64(KindVisitor) + } +} + +struct KindVisitor; + +impl Visitor<'_> for KindVisitor { + type Value = NostrKind; + + fn expecting(&self, f: &mut fmt::Formatter) -> fmt::Result { + write!(f, "an unsigned number") + } + + fn visit_u64(self, v: u64) -> Result + where + E: Error, + { + Ok(From::::from(v)) + } +} diff --git a/gemini/src/nostr/mod.rs b/gemini/src/nostr/mod.rs new file mode 100644 index 000000000..28bfe0a2f --- /dev/null +++ b/gemini/src/nostr/mod.rs @@ -0,0 +1,199 @@ +use client_message::{ClientMessage, Filter, SubscriptionId}; +use event::NostrEvent; +use reqwest::{header::CONTENT_TYPE, Client}; +use serde::{Deserialize, Serialize}; +use std::fmt; +use tag::{Tag, TagKind}; + +use crate::util::handle_response; + +pub mod client_message; +pub mod event; +pub mod kind; +pub mod relay_message; +pub mod tag; + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +pub struct GitEvent { + pub peer: String, + pub uri: String, + pub action: String, + pub r#ref: String, + pub commit: String, + pub issue: String, + pub mr: String, + pub title: String, + pub content: String, +} + +impl GitEvent { + pub async fn sent_to_relay(&self, bootstrap_node: String) -> Result<(), String> { + let keypair = vault::get_keypair(); + let event = NostrEvent::new_git_event(keypair, self.clone()); + let client_message = ClientMessage::new_event(event); + + //send to relay + let client = Client::new(); + let url = format!("{bootstrap_node}/api/v1/nostr"); + let request_result = client + .post(url.clone()) + .header(CONTENT_TYPE, "application/json") + .body(client_message.as_json()) + .send() + .await; + + match handle_response(request_result).await { + Ok(s) => { + tracing::info!("post response from url {}:\n{}", url, s.clone()); + Ok(()) + } + Err(e) => { + tracing::error!("post response from url {} failed:\n{}", url, e); + Err(e) + } + } + } + + pub fn to_tags(&self) -> Vec { + let mut tags: Vec = Vec::new(); + if !self.peer.is_empty() { + let tag = Tag::Generic(TagKind::Peer, vec![self.peer.clone()]); + tags.push(tag); + } + if !self.uri.is_empty() { + let tag = Tag::Generic(TagKind::URI, vec![self.uri.clone()]); + tags.push(tag); + } + if !self.action.is_empty() { + let tag = Tag::Generic(TagKind::Action, vec![self.action.clone()]); + tags.push(tag); + } + if !self.r#ref.is_empty() { + let tag = Tag::Generic(TagKind::Ref, vec![self.r#ref.clone()]); + tags.push(tag); + } + if !self.commit.is_empty() { + let tag = Tag::Generic(TagKind::Commit, vec![self.commit.clone()]); + tags.push(tag); + } + if !self.issue.is_empty() { + let tag = Tag::Generic(TagKind::Issue, vec![self.issue.clone()]); + tags.push(tag); + } + if !self.mr.is_empty() { + let tag = Tag::Generic(TagKind::MR, vec![self.mr.clone()]); + tags.push(tag); + } + if !self.title.is_empty() { + let tag = Tag::Generic(TagKind::Title, vec![self.title.clone()]); + tags.push(tag); + } + + tags + } + + pub fn from_tags(tags: Vec) -> Self { + let mut git_event = Self { + peer: "".to_string(), + uri: "".to_string(), + action: "".to_string(), + r#ref: "".to_string(), + commit: "".to_string(), + issue: "".to_string(), + mr: "".to_string(), + title: "".to_string(), + content: "".to_string(), + }; + for x in tags { + let vec = x.as_vec(); + if vec.len() > 1 { + let kind = TagKind::from(vec[0].clone()); + let content = vec[1].clone(); + match kind { + TagKind::Peer => git_event.peer = content, + TagKind::URI => git_event.uri = content, + TagKind::Action => git_event.action = content, + TagKind::Ref => git_event.r#ref = content, + TagKind::Commit => git_event.commit = content, + TagKind::Issue => git_event.issue = content, + TagKind::MR => git_event.mr = content, + TagKind::Title => git_event.title = content, + _ => {} + } + } + } + git_event + } +} + +/// Messages error +#[derive(Debug)] +pub enum MessageHandleError { + /// Invalid message format + InvalidMessageFormat, + /// Impossible to deserialize message + Json(serde_json::Error), + /// Empty message + EmptyMsg, + /// Event error + Event(event::Error), +} + +impl std::error::Error for MessageHandleError {} + +impl fmt::Display for MessageHandleError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::InvalidMessageFormat => write!(f, "Message has an invalid format"), + Self::Json(e) => write!(f, "Json deserialization failed: {e}"), + Self::EmptyMsg => write!(f, "Received empty message"), + Self::Event(e) => write!(f, "Event: {e}"), + } + } +} + +impl From for MessageHandleError { + fn from(e: serde_json::Error) -> Self { + Self::Json(e) + } +} + +impl From for MessageHandleError { + fn from(e: event::Error) -> Self { + Self::Event(e) + } +} + +pub async fn subscribe_git_event( + uri: String, + subscription_id: String, + bootstrap_node: String, +) -> Result<(), String> { + let filters = vec![Filter::new().repo_uri(uri)]; + + let client_req = ClientMessage::new_req(SubscriptionId::new(subscription_id), filters); + + //send to relay + let client = Client::new(); + let url = format!("{bootstrap_node}/api/v1/nostr"); + let request_result = client + .post(url.clone()) + .header(CONTENT_TYPE, "application/json") + .body(client_req.as_json()) + .send() + .await; + + match handle_response(request_result).await { + Ok(_s) => { + tracing::info!("subscribe git_event successfully: {}", url); + Ok(()) + } + Err(e) => { + tracing::error!("subscribe git_event successfully failed:\n{}", e); + Err(e) + } + } +} + +#[cfg(test)] +mod tests {} diff --git a/gemini/src/nostr/relay_message.rs b/gemini/src/nostr/relay_message.rs new file mode 100644 index 000000000..2401af876 --- /dev/null +++ b/gemini/src/nostr/relay_message.rs @@ -0,0 +1,239 @@ +use crate::nostr::client_message::SubscriptionId; +use crate::nostr::event::{EventId, NostrEvent}; +use crate::nostr::MessageHandleError; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde_json::{json, Value}; + +/// Messages sent by relays, received by clients +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum RelayMessage { + /// `["EVENT", , ]` (NIP01) + Event { + /// Subscription ID + subscription_id: SubscriptionId, + /// Event + event: NostrEvent, + }, + /// `["OK", , , ]` (NIP01) + Ok { + /// Event ID + event_id: EventId, + /// Status + status: bool, + /// Message + message: String, + }, + /// `["EOSE", ]` (NIP01) + EndOfStoredEvents(SubscriptionId), + /// ["NOTICE", \] (NIP01) + Notice { + /// Message + message: String, + }, +} + +impl Serialize for RelayMessage { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + let json_value: Value = self.as_value(); + json_value.serialize(serializer) + } +} + +impl<'de> Deserialize<'de> for RelayMessage { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + let json_value = Value::deserialize(deserializer)?; + RelayMessage::from_value(json_value).map_err(serde::de::Error::custom) + } +} + +impl RelayMessage { + /// Create new `EVENT` message + pub fn new_event(subscription_id: SubscriptionId, event: NostrEvent) -> Self { + Self::Event { + subscription_id, + event, + } + } + + pub fn new_event_box(subscription_id: SubscriptionId, event: NostrEvent) -> Self { + Self::Event { + subscription_id, + event, + } + } + + /// Create new `NOTICE` message + pub fn new_notice(message: S) -> Self + where + S: Into, + { + Self::Notice { + message: message.into(), + } + } + + /// Create new `EOSE` message + pub fn new_eose(subscription_id: SubscriptionId) -> Self { + Self::EndOfStoredEvents(subscription_id) + } + + /// Create new `OK` message + pub fn new_ok(event_id: EventId, status: bool, message: S) -> Self + where + S: Into, + { + Self::Ok { + event_id, + status, + message: message.into(), + } + } + + fn as_value(&self) -> Value { + match self { + Self::Event { + event, + subscription_id, + } => json!(["EVENT", subscription_id, event]), + Self::Notice { message } => json!(["NOTICE", message]), + Self::EndOfStoredEvents(subscription_id) => { + json!(["EOSE", subscription_id]) + } + Self::Ok { + event_id, + status, + message, + } => json!(["OK", event_id, status, message]), + } + } + + pub fn from_value(msg: Value) -> Result { + let v = msg + .as_array() + .ok_or(MessageHandleError::InvalidMessageFormat)?; + + if v.is_empty() { + return Err(MessageHandleError::InvalidMessageFormat); + } + + let v_len: usize = v.len(); + + // Notice + // Relay response format: ["NOTICE", ] + if v[0] == "NOTICE" { + if v_len != 2 { + return Err(MessageHandleError::InvalidMessageFormat); + } + return Ok(Self::Notice { + message: serde_json::from_value(v[1].clone())?, + }); + } + + // Event + // Relay response format: ["EVENT", , ] + if v[0] == "EVENT" { + if v_len != 3 { + return Err(MessageHandleError::InvalidMessageFormat); + } + + return Ok(Self::Event { + subscription_id: serde_json::from_value(v[1].clone())?, + event: NostrEvent::from_value(v[2].clone())?, + }); + } + + // EOSE (NIP-15) + // Relay response format: ["EOSE", ] + if v[0] == "EOSE" { + if v_len != 2 { + return Err(MessageHandleError::InvalidMessageFormat); + } + let subscription_id: String = serde_json::from_value(v[1].clone())?; + return Ok(Self::EndOfStoredEvents(SubscriptionId::new( + subscription_id, + ))); + } + + // OK (NIP-20) + // Relay response format: ["OK", , , ] + if v[0] == "OK" { + if v_len != 4 { + return Err(MessageHandleError::InvalidMessageFormat); + } + + return Ok(Self::Ok { + event_id: serde_json::from_value(v[1].clone())?, + status: serde_json::from_value(v[2].clone())?, + message: serde_json::from_value(v[3].clone())?, + }); + } + + Err(MessageHandleError::InvalidMessageFormat) + } + + pub fn from_json(json: T) -> Result + where + T: AsRef<[u8]>, + { + let msg: &[u8] = json.as_ref(); + + if msg.is_empty() { + return Err(MessageHandleError::EmptyMsg); + } + + let value: Value = serde_json::from_slice(msg)?; + Self::from_value(value) + } + + pub fn as_json(&self) -> String { + json!(self).to_string() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::nostr::kind::NostrKind; + use crate::nostr::tag::{Tag, TagKind}; + use secp256k1::{Keypair, Secp256k1, SecretKey}; + use std::str::FromStr; + + #[test] + fn test_handle_valid_notice() { + let valid_notice_msg = r#"["NOTICE","Invalid event format!"]"#; + let handled_valid_notice_msg = + RelayMessage::new_notice(String::from("Invalid event format!")); + + assert_eq!( + RelayMessage::from_json(valid_notice_msg).unwrap(), + handled_valid_notice_msg + ); + } + + #[test] + fn test_handle_valid_event() { + let valid_event_msg = r#"["EVENT", "random_string", {"content":"123","created_at":1111,"id":"bc3cc3a1499ea0c618df6dc5a300e8ad05f4d3a9a96e40e8a237cec11cc78667","kind":111,"pubkey":"385c3a6ec0b9d57a4330dbd6284989be5bd00e41c535f9ca39b6ae7c521b81cd","sig":"1285a0697eb44bf41f7a1bb063646ac47de8f6664335b791fe4982b1c9cdae1f56d0f1dd0564aeb053fd3cab16c4a0508cef70ac0bec9d79db3f2a2c28426f20","tags":[["p"]]}]"#; + + let secp = Secp256k1::new(); + let secret_key = + SecretKey::from_str("6b911fd37cdf5c81d4c0adb1ab7fa822ed253ab0ad9aa18d77257c88b29b718e") + .unwrap(); + let key_pair = Keypair::from_secret_key(&secp, &secret_key); + let tag = Tag::Generic(TagKind::P, Vec::new()); + let tags: Vec = vec![tag]; + let content = "123".to_string(); + let handled_event = + NostrEvent::new_with_timestamp(key_pair, 1111, NostrKind::Mega, tags, content); + + assert_eq!( + RelayMessage::from_json(valid_event_msg).unwrap(), + RelayMessage::new_event(SubscriptionId::new("random_string"), handled_event) + ); + } +} diff --git a/gemini/src/nostr/tag.rs b/gemini/src/nostr/tag.rs new file mode 100644 index 000000000..39f23a0c3 --- /dev/null +++ b/gemini/src/nostr/tag.rs @@ -0,0 +1,166 @@ +use serde::ser::SerializeSeq; +use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use std::fmt; + +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)] +pub enum Tag { + Generic(TagKind, Vec), +} + +/// Tag kind +#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)] +pub enum TagKind { + /// Public key + P, + Peer, + URI, + Action, + Ref, + Commit, + Issue, + MR, + Title, + /// Custom tag kind + Custom(String), +} + +/// [`Tag`] error +#[derive(Debug)] +pub enum Error { + /// Impossible to parse [`Tag`] + TagParseError, + /// Impossible to find tag kind + KindNotFound, +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::TagParseError => write!(f, "Impossible to parse tag"), + Self::KindNotFound => write!(f, "Impossible to find tag kind"), + } + } +} + +impl Tag { + /// Parse [`Tag`] from string vector + pub fn parse(data: Vec) -> Result + where + S: AsRef, + { + Tag::try_from(data) + } + + /// Get [`Tag`] as string vector + pub fn as_vec(&self) -> Vec { + self.clone().into() + } +} + +impl TryFrom> for Tag +where + S: AsRef, +{ + type Error = Error; + + fn try_from(tag: Vec) -> Result { + let tag_kind: TagKind = match tag.first() { + Some(kind) => TagKind::from(kind), + None => return Err(Error::KindNotFound), + }; + { + Ok(Self::Generic( + tag_kind, + tag[1..].iter().map(|s| s.as_ref().to_owned()).collect(), + )) + } + } +} + +impl From for Vec { + fn from(data: Tag) -> Self { + match data { + Tag::Generic(kind, data) => [vec![kind.to_string()], data].concat(), + } + } +} + +impl fmt::Display for TagKind { + fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { + + match self { + Self::P => write!(f, "p"), + Self::Peer => write!(f, "peer"), + Self::URI => write!(f, "uri"), + Self::Action => write!(f, "action"), + Self::Ref => write!(f, "ref"), + Self::Commit => write!(f, "commit"), + Self::Issue => write!(f, "issue"), + Self::MR => write!(f, "mr"), + Self::Title => write!(f, "title"), + Self::Custom(tag) => write!(f, "{tag}"), + } + } +} + +impl From for TagKind +where + S: AsRef, +{ + fn from(tag: S) -> Self { + match tag.as_ref() { + "p" => Self::P, + "peer" => Self::Peer, + "uri" => Self::URI, + "action" => Self::Action, + "ref" => Self::Ref, + "commit" => Self::Commit, + "issue" => Self::Issue, + "mr" => Self::MR, + "title" => Self::Title, + t => Self::Custom(t.to_owned()), + } + } +} + +impl From for String { + fn from(e: TagKind) -> String { + match e { + TagKind::P => "p".to_string(), + TagKind::Peer => "peer".to_string(), + TagKind::URI => "uri".to_string(), + TagKind::Action => "action".to_string(), + TagKind::Ref => "ref".to_string(), + TagKind::Commit => "commit".to_string(), + TagKind::Issue => "issue".to_string(), + TagKind::MR => "mr".to_string(), + TagKind::Title => "title".to_string(), + TagKind::Custom(tag) => tag.to_string(), + } + } +} + +impl Serialize for Tag { + fn serialize(&self, serializer: S) -> Result + where + S: Serializer, + { + let data: Vec = self.as_vec(); + let mut seq = serializer.serialize_seq(Some(data.len()))?; + for element in data.into_iter() { + seq.serialize_element(&element)?; + } + seq.end() + } +} + +impl<'de> Deserialize<'de> for Tag { + fn deserialize(deserializer: D) -> Result + where + D: Deserializer<'de>, + { + type Data = Vec; + let vec: Vec = Data::deserialize(deserializer)?; + Self::try_from(vec).map_err(serde::de::Error::custom) + } +} diff --git a/gemini/src/util.rs b/gemini/src/util.rs new file mode 100644 index 000000000..31ec5beb0 --- /dev/null +++ b/gemini/src/util.rs @@ -0,0 +1,49 @@ +use std::{ + net::TcpListener, + time::{SystemTime, UNIX_EPOCH}, +}; + +pub fn get_short_peer_id(peer_id: String) -> String { + if peer_id.len() <= 7 { + return peer_id; + } + peer_id[0..7].to_string() +} + +pub fn get_available_port() -> Result { + // Bind to port 0 to let the OS assign an available port + match TcpListener::bind("127.0.0.1:0") { + Ok(listener) => { + let port = listener.local_addr().unwrap().port(); + Ok(port) + } + Err(e) => Err(format!("Failed to bind to a port: {}", e)), + } +} + +pub fn get_utc_timestamp() -> i64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_millis() as i64 +} + +pub async fn handle_response( + request_result: Result, +) -> Result { + match request_result { + Ok(res) => { + if res.status().is_success() { + Ok(res.text().await.unwrap()) + } else { + Err(res.text().await.unwrap()) + } + } + Err(e) => Err(e.to_string()), + } +} + +pub fn repo_path_to_identifier(http_port: u16, repo_path: String) -> String { + let (peer_id, _) = vault::init(); + format!("p2p://{}/{http_port}{repo_path}.git", peer_id.clone()) +} diff --git a/gemini/src/ztm/agent.rs b/gemini/src/ztm/agent.rs index 73458a351..14710f928 100644 --- a/gemini/src/ztm/agent.rs +++ b/gemini/src/ztm/agent.rs @@ -12,7 +12,7 @@ use serde::{Deserialize, Serialize}; use crate::{RepoInfo, ZTM_APP_PROVIDER}; -use super::{handle_ztm_response, hub::ZTMUserPermit}; +use super::{handle_response, hub::ZTMUserPermit}; #[derive(Deserialize, Serialize, Debug, Clone)] pub struct ZTMMesh { @@ -35,7 +35,6 @@ pub struct Agent { #[derive(Deserialize, Serialize, Debug, Clone)] pub struct ZTMEndPoint { pub id: String, - pub name: String, pub username: String, pub online: bool, #[serde(rename = "isLocal")] @@ -139,7 +138,7 @@ impl ZTMAgent for LocalZTMAgent { .body(permit_string) .send() .await; - let response_text = match handle_ztm_response(request_result).await { + let response_text = match handle_response(request_result).await { Ok(s) => s, Err(s) => { return Err(s); @@ -161,7 +160,7 @@ impl ZTMAgent for LocalZTMAgent { let agent_address = format!("http://127.0.0.1:{agent_port}"); let url = format!("{agent_address}/api/meshes/{MESH_NAME}/endpoints"); let request_result = reqwest::get(url).await; - let eps: String = match handle_ztm_response(request_result).await { + let eps: String = match handle_response(request_result).await { Ok(s) => s, Err(s) => { return Err(s); @@ -194,11 +193,11 @@ impl ZTMAgent for LocalZTMAgent { match self.get_ztm_endpoints().await { Ok(ep_list) => { for ele in ep_list { - if ele.online && ele.name == peer_id { + if ele.online && ele.username == peer_id { return Ok(ele); } } - return Err("Can not find remote ztm endpoint".to_string()); + return Err(format!("Ztm agent({peer_id}) is not online.")); } Err(e) => Err(e), } @@ -229,7 +228,7 @@ impl ZTMAgent for LocalZTMAgent { .body(req_string) .send() .await; - let response_text = match handle_ztm_response(request_result).await { + let response_text = match handle_response(request_result).await { Ok(s) => s, Err(s) => { return Err(s); @@ -264,7 +263,7 @@ impl ZTMAgent for LocalZTMAgent { .body(req_string) .send() .await; - let response_text = match handle_ztm_response(request_result).await { + let response_text = match handle_response(request_result).await { Ok(s) => s, Err(s) => { return Err(s); @@ -294,7 +293,7 @@ impl ZTMAgent for LocalZTMAgent { .body(req) .send() .await; - let response_text = match handle_ztm_response(request_result).await { + let response_text = match handle_response(request_result).await { Ok(s) => s, Err(s) => { return Err(s); @@ -317,6 +316,7 @@ impl ZTMAgent for LocalZTMAgent { let url = format!( "{agent_address}/api/meshes/{MESH_NAME}/apps/{provider}/{app_name}/api/endpoints/{ep_id}/inbound/tcp/{bound_name}" ); + tracing::info!("create_ztm_app_tunnel_inbound url: {}", url); let client = Client::new(); let req = format!(r#"{{"listens": [{{"ip":"127.0.0.1","port":{port}}}]}}"#); // let req = r#"{"listens":[{"ip":"127.0.0.1","port":8081}]}"#; @@ -326,7 +326,7 @@ impl ZTMAgent for LocalZTMAgent { .body(req) .send() .await; - let response_text = match handle_ztm_response(request_result).await { + let response_text = match handle_response(request_result).await { Ok(s) => s, Err(s) => { return Err(s); @@ -358,7 +358,7 @@ impl ZTMAgent for LocalZTMAgent { .body(req) .send() .await; - let response_text = match handle_ztm_response(request_result).await { + let response_text = match handle_response(request_result).await { Ok(s) => s, Err(s) => { return Err(s); @@ -367,53 +367,18 @@ impl ZTMAgent for LocalZTMAgent { Ok(response_text) } } + pub async fn run_ztm_client( bootstrap_node: String, _config: Config, peer_id: String, agent: LocalZTMAgent, + http_port: u16, ) { let name = peer_id.clone(); - // let _context = Context::new(config.clone()).await; - - // let local_permit_option = read_secret(peer_id.as_str()).unwrap(); - // let permit: ZTMUserPermit = match local_permit_option { - // Some(res) => { - // let p = ZTMUserPermit::from_json_map(res.data.unwrap()); - // tracing::info!("read ztm permit file from vault:{:?}", p); - // p - // } - // None => { - // //generate permit from bootstrap_node - // // 1. to get permit json from bootstrap_node - // // GET {bootstrap_node}/api/v1/certificate?name={name} - // let url = format!("{bootstrap_node}/api/v1/certificate?name={name}"); - // let request_result = reqwest::get(url).await; - // let response_text = match handle_ztm_response(request_result).await { - // Ok(s) => s, - // Err(s) => { - // tracing::error!( - // "GET {bootstrap_node}/api/v1/certificate?name={name} failed,{s}" - // ); - // return; - // } - // }; - // let permit: ZTMUserPermit = match serde_json::from_slice(response_text.as_bytes()) { - // Ok(p) => p, - // Err(e) => { - // tracing::error!("{}", e); - // return; - // } - // }; - // //save to vault - // write_secret(peer_id.clone().as_str(), Some(permit.to_json_map().clone())).unwrap(); - // permit - // } - // }; - let url = format!("{bootstrap_node}/api/v1/certificate?name={name}"); let request_result = reqwest::get(url).await; - let response_text = match handle_ztm_response(request_result).await { + let response_text = match handle_response(request_result).await { Ok(s) => s, Err(s) => { tracing::error!("GET {bootstrap_node}/api/v1/certificate?name={name} failed,{s}"); @@ -457,6 +422,23 @@ pub async fn run_ztm_client( } tracing::info!("start tunnel app successfully"); + //test send msg + // sleep(Duration::from_secs(5)); + // match send_get_request_to_peer_by_tunnel( + // agent.agent_port, + // "nX7NRgitx7wUwAiJXxAVcec4iAoV8YwbzUn8FqwfFR4J".to_string(), + // "api/v1/ztm/repo_provide".to_string(), + // ) + // .await + // { + // Ok(s) => { + // tracing::info!("send_get_request_to_peer_by_tunnel successfully:{}", s); + // } + // Err(e) => { + // tracing::error!(e); + // } + // }; + // ping relay let peer_id_clone = peer_id.clone(); let bootstrap_node_clone = bootstrap_node.clone(); @@ -467,21 +449,23 @@ pub async fn run_ztm_client( url.clone(), peer_id_clone.clone(), permit.bootstraps.first().unwrap().to_string(), + http_port, ) .await; interval.tick().await; } } -pub async fn ping(url: String, peer_id: String, hub: String) { +pub async fn ping(url: String, peer_id: String, hub: String, service_port: u16) { let mut params = HashMap::new(); params.insert("peer_id", peer_id.clone()); params.insert("hub", hub); params.insert("agent_name", peer_id.clone()); params.insert("service_name", peer_id.clone()); + params.insert("service_port", service_port.to_string()); let client = reqwest::Client::new(); let response = client.get(url.clone()).query(¶ms).send().await; - let response_text = match handle_ztm_response(response).await { + let response_text = match handle_response(response).await { Ok(s) => s, Err(s) => { tracing::error!("GET {url} failed,{s}"); @@ -501,7 +485,7 @@ pub async fn share_repo(url: String, repo_info: RepoInfo) { .body(req_string) .send() .await; - let response_text = match handle_ztm_response(response).await { + let response_text = match handle_response(response).await { Ok(s) => s, Err(s) => { tracing::error!("POST {} failed,{}", url.clone(), s.clone()); diff --git a/gemini/src/ztm/hub.rs b/gemini/src/ztm/hub.rs index c131a5dd4..8d29630e0 100644 --- a/gemini/src/ztm/hub.rs +++ b/gemini/src/ztm/hub.rs @@ -3,7 +3,7 @@ use axum::async_trait; use reqwest::Client; use serde_json::Value; -use crate::ztm::handle_ztm_response; +use crate::util::handle_response; #[derive(Deserialize, Serialize, Debug, Clone)] pub struct CertAgent { @@ -68,7 +68,7 @@ impl ZTMCA for LocalHub { //1. GET {ca}/api/certificates/ca -> ca certificate let url = format!("{ca_address}/api/certificates/ca"); let request_result = reqwest::get(url).await; - let ca_certificate = match handle_ztm_response(request_result).await { + let ca_certificate = match handle_response(request_result).await { Ok(s) => s, Err(s) => { return Err(s); @@ -79,7 +79,7 @@ impl ZTMCA for LocalHub { let url = format!("{ca_address}/api/certificates/{name}"); let client = Client::new(); let request_result = client.post(url).send().await; - let user_key = match handle_ztm_response(request_result).await { + let user_key = match handle_response(request_result).await { Ok(s) => s, Err(s) => { return Err(s); @@ -89,7 +89,7 @@ impl ZTMCA for LocalHub { //3. GET {ca}/api/certificates/{username} -> user certificate let url = format!("{ca_address}/api/certificates/{name}"); let request_result = reqwest::get(url).await; - let user_certificate = match handle_ztm_response(request_result).await { + let user_certificate = match handle_response(request_result).await { Ok(s) => s, Err(s) => { return Err(s); diff --git a/gemini/src/ztm/mod.rs b/gemini/src/ztm/mod.rs index d99f6db9d..616405e8b 100755 --- a/gemini/src/ztm/mod.rs +++ b/gemini/src/ztm/mod.rs @@ -1,17 +1,208 @@ +use agent::{LocalZTMAgent, ZTMAgent}; +use reqwest::{header::CONTENT_TYPE, Client}; + +use crate::{ + util::{get_available_port, get_short_peer_id, handle_response}, + ZTM_APP_PROVIDER, +}; + pub mod agent; pub mod hub; -pub async fn handle_ztm_response( - request_result: Result, +pub async fn create_tunnel( + ztm_agent_port: u16, + remote_peer_id: String, + local_port: u16, + remote_port: u16, + bound_name: String, +) -> Result<(), String> { + let agent: LocalZTMAgent = LocalZTMAgent { + agent_port: ztm_agent_port, + }; + let local_ep = match agent.get_ztm_local_endpoint().await { + Ok(ep) => ep, + Err(e) => return Err(e), + }; + let remote_ep = match agent.get_ztm_remote_endpoint(remote_peer_id.clone()).await { + Ok(ep) => ep, + Err(e) => return Err(e), + }; + + tracing::info!("create_tunnel remote_ep:{:?}", remote_ep); + + //creata inbound + match agent + .create_ztm_app_tunnel_inbound( + local_ep.id, + ZTM_APP_PROVIDER.to_string(), + "tunnel".to_string(), + bound_name.clone(), + local_port, + ) + .await + { + Ok(_) => (), + Err(s) => { + tracing::error!("create app inbound failed, {s}"); + return Err(s); + } + } + tracing::info!("create app inbound successfully"); + + //creata outbound + match agent + .create_ztm_app_tunnel_outbound( + remote_ep.id, + ZTM_APP_PROVIDER.to_string(), + "tunnel".to_string(), + bound_name, + remote_port, + ) + .await + { + Ok(msg) => { + tracing::info!("create app outbound successfully,{}", msg); + } + Err(s) => { + tracing::error!("create app outbound, {s}"); + return Err(s); + } + } + Ok(()) +} + +pub async fn send_get_request_to_peer_by_tunnel( + ztm_agent_port: u16, + remote_peer_id: String, + path: String, +) -> Result { + //get a random port + let local_port = match get_available_port() { + Ok(port) => port, + Err(e) => { + return Err(e); + } + }; + let (peer_id, _) = vault::init(); + let bound_name = format!( + "get_{}_{}", + get_short_peer_id(peer_id), + get_short_peer_id(remote_peer_id.clone()) + ); + let remote_port = 8000; + match create_tunnel( + ztm_agent_port, + remote_peer_id.clone(), + local_port, + remote_port, + bound_name.clone(), + ) + .await + { + Ok(_) => { + tracing::info!( + "create ztm tunnel successfully: \nbound_name:{}\nlocal port:{}\nremote peer:{}\nremote port:{}", + bound_name, + local_port, + remote_peer_id, + remote_port, + ); + } + Err(e) => { + tracing::error!( + "create ztm tunnel failed: \nbound_name:{}\nlocal port:{}\nremote peer:{}\nremote port:{}\nerror:{}", + bound_name, + local_port, + remote_peer_id, + remote_port, + e.clone(), + ); + return Err(e); + } + } + let url = format!("http://127.0.0.1:{local_port}/{path}"); + tracing::info!("send request to:\n{}", url); + let request_result = reqwest::get(url.clone()).await; + match handle_response(request_result).await { + Ok(s) => { + tracing::info!("get response from url {}:\n{}", url, s.clone()); + Ok(s) + } + Err(e) => { + tracing::error!("get response from url {} failed:\n{}", url, e); + Err(e) + } + } +} + +pub async fn send_post_request_to_peer_by_tunnel( + ztm_agent_port: u16, + remote_peer_id: String, + path: String, + body: String, ) -> Result { - match request_result { - Ok(res) => { - if res.status().is_success() { - Ok(res.text().await.unwrap()) - } else { - Err(res.text().await.unwrap()) - } - } - Err(e) => Err(e.to_string()), + //get a random port + let local_port = match get_available_port() { + Ok(port) => port, + Err(e) => { + return Err(e); + } + }; + let (peer_id, _) = vault::init(); + let bound_name = format!( + "post_{}_{}", + get_short_peer_id(peer_id), + get_short_peer_id(remote_peer_id.clone()) + ); + let remote_port = 8000; + match create_tunnel( + ztm_agent_port, + remote_peer_id.clone(), + local_port, + remote_port, + bound_name.clone(), + ) + .await + { + Ok(_) => { + tracing::info!( + "create ztm tunnel successfully: \nbound_name:{}\nlocal port:{}\nremote peer:{}\nremote port:{}", + bound_name, + local_port, + remote_peer_id, + remote_port, + ); + } + Err(e) => { + tracing::error!( + "create ztm tunnel failed: \nbound_name:{}\nlocal port:{}\nremote peer:{}\nremote port:{}\nerror:{}", + bound_name, + local_port, + remote_peer_id, + remote_port, + e.clone(), + ); + return Err(e); + } + } + let url = format!("http://127.0.0.1:{local_port}/{path}"); + + let client = Client::new(); + + let request_result = client + .post(url.clone()) + .header(CONTENT_TYPE, "application/json") + .body(body) + .send() + .await; + match handle_response(request_result).await { + Ok(s) => { + tracing::info!("post response from url {}:\n{}", url, s.clone()); + Ok(s) + } + Err(e) => { + tracing::error!("post response from url {} failed:\n{}", url, e); + Err(e) + } } } diff --git a/jupiter/callisto/src/lib.rs b/jupiter/callisto/src/lib.rs index 0451807a1..310f38140 100644 --- a/jupiter/callisto/src/lib.rs +++ b/jupiter/callisto/src/lib.rs @@ -23,7 +23,9 @@ pub mod mega_mr_conv; pub mod mega_refs; pub mod mega_tag; pub mod mega_tree; +pub mod mq_storage; pub mod raw_blob; pub mod ztm_node; +pub mod ztm_nostr_event; +pub mod ztm_nostr_req; pub mod ztm_repo_info; -pub mod mq_storage; diff --git a/jupiter/callisto/src/ztm_node.rs b/jupiter/callisto/src/ztm_node.rs index 6e29ca519..ffee50c0f 100644 --- a/jupiter/callisto/src/ztm_node.rs +++ b/jupiter/callisto/src/ztm_node.rs @@ -10,6 +10,7 @@ pub struct Model { pub hub: String, pub agent_name: String, pub service_name: String, + pub service_port: i32, pub r#type: String, pub online: bool, pub last_online_time: i64, diff --git a/jupiter/callisto/src/ztm_nostr_event.rs b/jupiter/callisto/src/ztm_nostr_event.rs new file mode 100644 index 000000000..9038d47e5 --- /dev/null +++ b/jupiter/callisto/src/ztm_nostr_event.rs @@ -0,0 +1,21 @@ +//! `SeaORM` Entity. Generated by sea-orm-codegen 0.11.3 + +use sea_orm::entity::prelude::*; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)] +#[sea_orm(table_name = "ztm_nostr_event")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub id: String, + pub pubkey: String, + pub created_at: i64, + pub kind: i32, + pub tags: String, + pub content: String, + pub sig: String, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/jupiter/callisto/src/ztm_nostr_req.rs b/jupiter/callisto/src/ztm_nostr_req.rs new file mode 100644 index 000000000..104e97844 --- /dev/null +++ b/jupiter/callisto/src/ztm_nostr_req.rs @@ -0,0 +1,17 @@ +//! `SeaORM` Entity. Generated by sea-orm-codegen 0.11.3 + +use sea_orm::entity::prelude::*; + +#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Eq)] +#[sea_orm(table_name = "ztm_nostr_req")] +pub struct Model { + #[sea_orm(primary_key, auto_increment = false)] + pub id: String, + pub subscription_id: String, + pub filters: String, +} + +#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)] +pub enum Relation {} + +impl ActiveModelBehavior for ActiveModel {} diff --git a/jupiter/src/storage/ztm_storage.rs b/jupiter/src/storage/ztm_storage.rs index c804301d8..a1b0366eb 100644 --- a/jupiter/src/storage/ztm_storage.rs +++ b/jupiter/src/storage/ztm_storage.rs @@ -1,9 +1,9 @@ use std::sync::Arc; -use callisto::{ztm_node, ztm_repo_info}; -use sea_orm::{DatabaseConnection, EntityTrait, InsertResult, IntoActiveModel, Set}; - +use callisto::{ztm_node, ztm_nostr_event, ztm_nostr_req, ztm_repo_info}; use common::errors::MegaError; +use sea_orm::InsertResult; +use sea_orm::{ColumnTrait, DatabaseConnection, EntityTrait, IntoActiveModel, QueryFilter, Set}; #[derive(Clone)] pub struct ZTMStorage { @@ -163,4 +163,73 @@ impl ZTMStorage { } } } + + pub async fn insert_nostr_event( + &self, + nostr_event: ztm_nostr_event::Model, + ) -> Result, MegaError> { + Ok( + ztm_nostr_event::Entity::insert(nostr_event.into_active_model()) + .exec(self.get_connection()) + .await + .unwrap(), + ) + } + + pub async fn get_nostr_event_by_id( + &self, + event_id: &str, + ) -> Result, MegaError> { + let result = ztm_nostr_event::Entity::find_by_id(event_id) + .one(self.get_connection()) + .await + .unwrap(); + Ok(result) + } + + pub async fn get_all_nostr_event(&self) -> Result, MegaError> { + Ok(ztm_nostr_event::Entity::find() + .all(self.get_connection()) + .await + .unwrap()) + } + + pub async fn get_all_nostr_event_by_pubkey( + &self, + pubkey: &str, + ) -> Result, MegaError> { + Ok(ztm_nostr_event::Entity::find() + .filter(ztm_nostr_event::Column::Pubkey.eq(pubkey)) + .all(self.get_connection()) + .await + .unwrap()) + } + + pub async fn insert_nostr_req( + &self, + nostr_req: ztm_nostr_req::Model, + ) -> Result, MegaError> { + Ok(ztm_nostr_req::Entity::insert(nostr_req.into_active_model()) + .exec(self.get_connection()) + .await + .unwrap()) + } + + pub async fn get_all_nostr_req_by_subscription_id( + &self, + subscription_id: &str, + ) -> Result, MegaError> { + Ok(ztm_nostr_req::Entity::find() + .filter(ztm_nostr_req::Column::SubscriptionId.eq(subscription_id)) + .all(self.get_connection()) + .await + .unwrap()) + } + + pub async fn get_all_nostr_req(&self) -> Result, MegaError> { + Ok(ztm_nostr_req::Entity::find() + .all(self.get_connection()) + .await + .unwrap()) + } } diff --git a/sql/postgres/pg_20240205__init.sql b/sql/postgres/pg_20240205__init.sql index 591646e67..2fe2ce0c3 100644 --- a/sql/postgres/pg_20240205__init.sql +++ b/sql/postgres/pg_20240205__init.sql @@ -226,7 +226,6 @@ CREATE TABLE IF NOT EXISTS "lfs_split_relations" ( PRIMARY KEY ("ori_oid", "sub_oid", "offset") ); - CREATE TABLE IF NOT EXISTS "ztm_node" ( "peer_id" VARCHAR(64) PRIMARY KEY, "hub" VARCHAR(64), @@ -234,7 +233,8 @@ CREATE TABLE IF NOT EXISTS "ztm_node" ( "service_name" VARCHAR(64), "type" VARCHAR(64), "online" BOOLEAN NOT NULL, - "last_online_time" BIGINT NOT NULL + "last_online_time" BIGINT NOT NULL, + "service_port" INT NOT NULL ); CREATE TABLE IF NOT EXISTS "ztm_repo_info" ( @@ -245,6 +245,22 @@ CREATE TABLE IF NOT EXISTS "ztm_repo_info" ( "commit" VARCHAR(64) ); +CREATE TABLE IF NOT EXISTS "ztm_nostr_event" ( + "id" VARCHAR(128) PRIMARY KEY, + "pubkey" VARCHAR(128), + "created_at" BIGINT NOT NULL, + "kind" INT, + "tags" TEXT, + "content" TEXT, + "sig" VARCHAR(256) +); + +CREATE TABLE IF NOT EXISTS "ztm_nostr_req" ( + "id" VARCHAR(128) PRIMARY KEY, + "subscription_id" VARCHAR(128), + "filters" TEXT +); + CREATE TABLE IF NOT EXISTS "mq_storage" ( "id" BIGINT PRIMARY KEY, "category" VARCHAR(64), diff --git a/sql/sqlite/sqlite_20240711_init.sql b/sql/sqlite/sqlite_20240711_init.sql index fa8df4d75..998a6fcff 100644 --- a/sql/sqlite/sqlite_20240711_init.sql +++ b/sql/sqlite/sqlite_20240711_init.sql @@ -226,15 +226,15 @@ CREATE TABLE IF NOT EXISTS "lfs_split_relations" ( PRIMARY KEY ("ori_oid", "sub_oid", "offset") ); - CREATE TABLE IF NOT EXISTS "ztm_node" ( - "peer_id" TEXT PRIMARY KEY, - "hub" TEXT, - "agent_name" TEXT, - "service_name" TEXT, - "type" TEXT, - "online" INTEGER NOT NULL, - "last_online_time" INTEGER NOT NULL + "peer_id" VARCHAR(64) PRIMARY KEY, + "hub" VARCHAR(64), + "agent_name" VARCHAR(64), + "service_name" VARCHAR(64), + "type" VARCHAR(64), + "online" BOOLEAN NOT NULL, + "last_online_time" BIGINT NOT NULL, + "service_port" INT NOT NULL ); CREATE TABLE IF NOT EXISTS "ztm_repo_info" ( @@ -245,6 +245,22 @@ CREATE TABLE IF NOT EXISTS "ztm_repo_info" ( "commit" TEXT ); +CREATE TABLE IF NOT EXISTS "ztm_nostr_event" ( + "id" VARCHAR(128) PRIMARY KEY, + "pubkey" VARCHAR(128), + "created_at" BIGINT NOT NULL, + "kind" INT, + "tags" TEXT, + "content" TEXT, + "sig" VARCHAR(256) +); + +CREATE TABLE IF NOT EXISTS "ztm_nostr_req" ( + "id" VARCHAR(128) PRIMARY KEY, + "subscription_id" VARCHAR(128), + "filters" TEXT +); + CREATE TABLE IF NOT EXISTS "mq_storage" ( "id" INTEGER PRIMARY KEY, "category" TEXT, diff --git a/vault/src/lib.rs b/vault/src/lib.rs index a3ab9f87d..4f2f38296 100644 --- a/vault/src/lib.rs +++ b/vault/src/lib.rs @@ -1,8 +1,10 @@ +use secp256k1::{Keypair, Secp256k1}; + use crate::vault::{read_secret, write_secret}; +pub mod nostr; pub mod pki; pub mod vault; -pub mod nostr; /// Initialize the Nostr ID if it's not found. /// - return: `(Nostr ID, secret_key)` @@ -31,6 +33,17 @@ pub fn init() -> (String, String) { ) } +pub fn get_peerid() -> String { + let (id, _sk) = init(); + id +} + +pub fn get_keypair() -> Keypair { + let (_, sk) = init(); + let secp = Secp256k1::new(); + secp256k1::Keypair::from_seckey_str(&secp, &sk).unwrap() +} + #[cfg(test)] mod tests { use super::*; @@ -41,4 +54,4 @@ mod tests { println!("Nostr ID: {:?}", id.0); println!("Secret Key: {:?}", id.1); // private key } -} \ No newline at end of file +}