Rework topics + syncOp trait

Author: the2pizza

dev/dev.sql

@@ -224,3 +224,10 @@ ALTER COLUMN days TYPE SMALLINT USING days::SMALLINT;
 
 ALTER TABLE subscriptions
   ALTER COLUMN referred_by TYPE VARCHAR(13);
+
+  ===
+
+  ALTER TABLE connections
+  ALTER COLUMN created_at TYPE timestamptz USING created_at AT TIME ZONE 'UTC',
+  ALTER COLUMN modified_at TYPE timestamptz USING modified_at AT TIME ZONE 'UTC',
+  ALTER COLUMN expired_at TYPE timestamptz USING expired_at AT TIME ZONE 'UTC';

dev/h2.yaml

@@ -0,0 +1,17 @@
+listen: :443
+
+acme:
+    domains:
+        - api.example.org
+    email: admin@example.com
+
+auth:
+    type: http
+    http:
+        url: https://api.example.org/auth
+
+masquerade:
+    type: proxy
+    proxy:
+        url: https://www.microsoft.com/
+        rewriteHost: true

src/bin/agent/core/http.rs

@@ -19,7 +19,8 @@ use super::Agent;
 pub struct ConnTypeParam {
     pub proto: Tag,
     pub last_update: Option<u64>,
-    pub env: String,
+    pub env: Option<String>,
+    pub topic: uuid::Uuid,
 }
 
 #[derive(Clone, Debug, Deserialize, Serialize)]
@@ -69,12 +70,14 @@ where
                 .clone()
         };
 
+        let id = node.uuid;
         let env = node.env;
 
         let conn_type_param = ConnTypeParam {
             proto,
             last_update,
-            env,
+            env: Some(env),
+            topic: id,
         };
 
         let mut endpoint_url = Url::parse(&endpoint)?;
@@ -95,13 +98,18 @@ where
         let status = res.status();
         let body = res.text().await?;
         if status.is_success() {
-            log::debug!("Connections Request Accepted: {:?}", status);
+            log::debug!("Connections Request Accepted for {}: {} ", proto, status);
             Ok(())
         } else {
-            log::error!("Connections Request failed: {} - {}", status, body);
+            log::error!(
+                "Connections Request failed for {}: {} - {}",
+                proto,
+                status,
+                body
+            );
             Err(PonyError::Custom(format!(
-                "Connections Request failed: {} - {}",
-                status, body
+                "Connections Request failed for {}: {} - {}",
+                proto, status, body
             )))
         }
     }

src/bin/agent/core/service.rs

@@ -100,20 +100,20 @@ pub async fn run(settings: AgentSettings) -> Result<()> {
             Ok(cfg) => {
                 if let Err(e) = cfg.validate() {
                     log::error!("Hysteria2 config validation failed: {}", e);
-                    None
+                    panic!("Hysteria2 config: {}", e);
                 } else {
                     match H2Settings::try_from(cfg) {
                         Ok(settings) => Some(settings),
                         Err(e) => {
                             log::error!("Hysteria2 validation error: {}", e);
-                            None
+                            panic!("Hysteria2 config: {}", e);
                         }
                     }
                 }
             }
             Err(e) => {
                 log::error!("Failed to load Hysteria2 config: {}", e);
-                None
+                panic!("Hysteria2 config: {}", e);
             }
         }
     } else {
@@ -228,7 +228,8 @@ pub async fn run(settings: AgentSettings) -> Result<()> {
                         let tags: Vec<_> = node
                             .inbounds
                             .keys()
-                            .filter(|k| !matches!(k, Tag::Hysteria2))
+                            .filter(|k| !matches!(k, Tag::Hysteria2)) // Hysteria2 uses external auth provider
+                            .filter(|k| !matches!(k, Tag::Mtproto)) // Mtproto doesn't support auth provider
                             .collect();
 
                         for tag in tags {
@@ -322,7 +323,7 @@ pub async fn run(settings: AgentSettings) -> Result<()> {
         tasks.push(xray_stats_task);
     }
 
-    if settings.agent.stat_enabled {
+    if settings.agent.stat_enabled && settings.wg.enabled {
         log::info!("Running WG Stat Task");
         let wg_stats_task = tokio::spawn({
             let agent = Arc::new(agent.clone());
@@ -343,7 +344,7 @@ pub async fn run(settings: AgentSettings) -> Result<()> {
         tasks.push(wg_stats_task);
     }
 
-    if settings.xray.enabled {
+    if settings.xray.enabled && settings.agent.local {
         if let Some(xray_handler_client) = xray_handler_client {
             let conn_task = tokio::spawn({
                 let conn_id = uuid::Uuid::new_v4();
@@ -406,7 +407,7 @@ pub async fn run(settings: AgentSettings) -> Result<()> {
         }
     }
 
-    if settings.wg.enabled {
+    if settings.wg.enabled && settings.agent.local {
         let conn_task = tokio::spawn({
             let agent = agent.clone();
             let settings = settings.clone();

src/bin/api/core/http/filters.rs

@@ -1,6 +1,5 @@
 use warp::Filter;
 
-use pony::zmq::publisher::Publisher as ZmqPublisher;
 use pony::Connection;
 use pony::ConnectionApiOp;
 use pony::ConnectionBaseOp;
@@ -29,13 +28,6 @@ pub fn with_ch(
     warp::any().map(move || ch.clone())
 }
 
-/// Provides zmq publisher filter
-pub fn publisher(
-    publisher: ZmqPublisher,
-) -> impl Filter<Extract = (ZmqPublisher,), Error = std::convert::Infallible> + Clone {
-    warp::any().map(move || publisher.clone())
-}
-
 pub fn with_param_string(
     param: String,
 ) -> impl Filter<Extract = (String,), Error = std::convert::Infallible> + Clone {

src/bin/api/core/http/handlers/connection.rs

@@ -11,7 +11,6 @@ use pony::http::MyRejection;
 use pony::http::ResponseMessage;
 use pony::memory::tag::ProtoTag;
 use pony::utils;
-use pony::zmq::publisher::Publisher as ZmqPublisher;
 use pony::Connection;
 use pony::ConnectionApiOp;
 use pony::ConnectionBaseOp;
@@ -34,8 +33,7 @@ use super::super::request::ConnCreateRequest;
 /// Handler get connection
 // GET /connections
 pub async fn get_connections_handler<N, C, S>(
-    conn_req: ConnTypeParam,
-    publisher: ZmqPublisher,
+    req: ConnTypeParam,
     memory: MemSync<N, C, S>,
 ) -> Result<impl warp::Reply, warp::Rejection>
 where
@@ -58,55 +56,58 @@ where
         + std::convert::From<pony::Subscription>,
 {
     let mem = memory.memory.read().await;
-    let proto = conn_req.proto;
-    let env = conn_req.env;
-
-    let last_update = conn_req.last_update;
+    let proto = req.proto;
+    let topic = req.topic;
+    let last_update = req.last_update;
+    let env = req.env;
 
     let connections_to_send: Vec<_> = mem
         .connections
         .iter()
         .filter(|(_, conn)| {
-            if conn.get_deleted() {
-                return false;
-            }
-
-            if conn.get_proto().proto() != proto {
-                return false;
-            }
-
-            if let Some(ts) = last_update {
-                conn.get_modified_at().and_utc().timestamp() as u64 >= ts
-            } else {
-                true
-            }
+            !conn.get_deleted()
+                && conn.get_proto().proto() == proto
+                && env.as_ref().is_none_or(|e| conn.get_env() == *e)
+                && last_update.is_none_or(|ts| conn.get_modified_at().timestamp() as u64 >= ts)
         })
         .collect();
 
-    log::debug!(
-        "Sending {} {:?} connections to auth",
-        connections_to_send.len(),
-        proto
-    );
-
-    let messages: Vec<_> = connections_to_send
-        .iter()
-        .map(|(conn_id, conn)| conn.as_create_message(conn_id))
-        .collect();
-
-    let bytes = to_bytes::<_, 1024>(&messages).map_err(|e| {
-        log::error!("Serialization error: {}", e);
-        warp::reject::custom(MyRejection(Box::new(e)))
-    })?;
-
-    publisher
-        .send_binary(&env, bytes.as_ref())
-        .await
-        .map_err(|e| {
-            log::error!("Publish error: {}", e);
+    if !connections_to_send.is_empty() {
+        log::debug!(
+            "Sending {} {:?} connections for env {:?} to topic {} ",
+            connections_to_send.len(),
+            proto,
+            env,
+            topic
+        );
+
+        let messages: Vec<_> = connections_to_send
+            .iter()
+            .map(|(conn_id, conn)| conn.as_create_message(conn_id))
+            .collect();
+
+        let bytes = to_bytes::<_, 1024>(&messages).map_err(|e| {
+            log::error!("Serialization error: {}", e);
             warp::reject::custom(MyRejection(Box::new(e)))
         })?;
 
+        memory
+            .publisher
+            .send_binary(&topic, bytes.as_ref())
+            .await
+            .map_err(|e| {
+                log::error!("Publish error: {}", e);
+                warp::reject::custom(MyRejection(Box::new(e)))
+            })?;
+    } else {
+        log::debug!(
+            "No message {} to send for env {:?} to topic {}",
+            proto,
+            env,
+            topic
+        );
+    }
+
     let resp = ResponseMessage::<Option<IdResponse>> {
         status: StatusCode::OK.as_u16(),
         message: "Ok".to_string(),
@@ -122,7 +123,6 @@ where
 // POST /connection
 pub async fn create_connection_handler<N, C, S>(
     conn_req: ConnCreateRequest,
-    publisher: ZmqPublisher,
     memory: MemSync<N, C, S>,
 ) -> Result<impl warp::Reply, warp::Rejection>
 where
@@ -342,7 +342,7 @@ where
                 conn.get_env()
             };
 
-            let _ = publisher.send_binary(&topic, bytes.as_ref()).await;
+            let _ = memory.publisher.send_binary(&topic, bytes.as_ref()).await;
 
             Ok(http::success_response(
                 format!("Connection {} has been created", id),
@@ -376,7 +376,7 @@ where
 // DELETE /connection?id=
 pub async fn delete_connection_handler<N, C, S>(
     conn_param: ConnQueryParam,
-    publisher: ZmqPublisher,
+
     memory: MemSync<N, C, S>,
 ) -> Result<impl warp::Reply, warp::Rejection>
 where
@@ -413,36 +413,17 @@ where
 
     if conn.get_deleted() {
         return Ok(http::not_found(&format!(
-            "Connection {} is deleted",
+            "Connection {} already is deleted",
             conn_id
         )));
     }
 
-    match SyncOp::delete_connection(&memory, &conn_id).await {
-        Ok(StorageOperationStatus::Ok(id)) => {
-            let msg = conn.as_delete_message(&conn_id);
-
-            let bytes = match rkyv::to_bytes::<_, 1024>(&msg) {
-                Ok(b) => b,
-                Err(e) => {
-                    return Ok(http::internal_error(&format!("Serialization error: {}", e)));
-                }
-            };
-
-            if let Some(node_id) = conn.get_wireguard_node_id() {
-                let _ = publisher
-                    .send_binary(&node_id.to_string(), bytes.as_ref())
-                    .await;
-            } else {
-                let _ = publisher.send_binary(&conn.get_env(), bytes.as_ref()).await;
-            }
-
-            Ok(http::success_response(
-                format!("Connection {} has been deleted", id),
-                Some(id),
-                http::Instance::Connection(conn.clone().into()),
-            ))
-        }
+    match SyncOp::delete_connection(&memory, &conn_id, &conn).await {
+        Ok(StorageOperationStatus::Ok(id)) => Ok(http::success_response(
+            format!("Connection {} has been deleted", id),
+            Some(id),
+            http::Instance::Connection(conn.clone().into()),
+        )),
 
         Ok(StorageOperationStatus::NotFound(id)) => {
             Ok(http::not_found(&format!("Connection {} not found", id)))