Add search channel

api/Cargo.toml

@@ -67,7 +67,7 @@ sendgrid = { version = "0.17", features = ["async"] }
 
 pay_u = { version = '0.1', features = ["single-client"] }
 
-sonic-channel = { version = "0.6.0" }
+sonic-channel = { version = "0.6.0", features = ["ingest"] }
 
 # For rewrite into bus-based app
 messagebus = { version = "0.9.13" }

api/src/actors/mod.rs

@@ -3,4 +3,5 @@ pub mod database;
 pub mod email_manager;
 pub mod order_manager;
 pub mod payment_manager;
+pub mod search_manager;
 pub mod token_manager;

api/src/actors/search_manager.rs

@@ -0,0 +1,49 @@
+use std::sync::{Arc, Mutex};
+
+use sonic_channel::SonicChannel;
+
+use crate::config::SharedAppConfig;
+
+pub struct Channels {
+    search: sonic_channel::SearchChannel,
+    ingest: sonic_channel::IngestChannel,
+}
+
+#[derive(Clone)]
+pub struct SearchManager {
+    channels: Option<Arc<Mutex<Channels>>>,
+    config: SharedAppConfig,
+}
+
+impl SearchManager {
+    pub fn new(config: SharedAppConfig) -> Self {
+        let enabled = config.lock().search().search_active();
+
+        let channels = if enabled {
+            let search = {
+                let l = config.lock();
+                sonic_channel::SearchChannel::start(
+                    l.search().sonic_search_addr(),
+                    l.search().sonic_search_pass(),
+                )
+                .expect("Failed to connect to sonic search channel")
+            };
+            let ingest = {
+                let l = config.lock();
+                sonic_channel::IngestChannel::start(
+                    l.search().sonic_ingest_addr(),
+                    l.search().sonic_ingest_pass(),
+                )
+                .expect("Failed to connect to sonic ingest channel")
+            };
+            Some(Arc::new(Mutex::new(Channels { search, ingest })))
+        } else {
+            None
+        };
+        Self { channels, config }
+    }
+}
+
+impl actix::Actor for SearchManager {
+    type Context = actix::Context<Self>;
+}

api/src/config.rs

@@ -269,12 +269,87 @@ impl DatabaseConfig {
     }
 }
 
+#[derive(Serialize, Deserialize)]
+pub struct SearchConfig {
+    sonic_search_addr: Option<String>,
+    sonic_search_pass: Option<String>,
+    sonic_ingest_addr: Option<String>,
+    sonic_ingest_pass: Option<String>,
+    search_active: bool,
+}
+
+impl Example for SearchConfig {
+    fn example() -> Self {
+        Self {
+            sonic_search_addr: None,
+            sonic_search_pass: None,
+            sonic_ingest_addr: None,
+            sonic_ingest_pass: None,
+            search_active: true,
+        }
+    }
+}
+
+impl Default for SearchConfig {
+    fn default() -> Self {
+        Self {
+            sonic_search_addr: Some(String::from("0.0.0.0:1491")),
+            sonic_search_pass: Some(String::from("SecretPassword")),
+            sonic_ingest_addr: Some(String::from("0.0.0.0:1491")),
+            sonic_ingest_pass: Some(String::from("SecretPassword")),
+            search_active: true,
+        }
+    }
+}
+
+impl SearchConfig {
+    pub fn sonic_search_addr(&self) -> String {
+        self.sonic_search_addr
+            .as_ref()
+            .cloned()
+            .or_else(|| std::env::var("SONIC_SEARCH_ADDR").ok())
+            .expect("Search sonic_search_addr nor SONIC_SEARCH_ADDR env variable was provided")
+    }
+    pub fn sonic_search_pass(&self) -> String {
+        self.sonic_search_pass
+            .as_ref()
+            .cloned()
+            .or_else(|| std::env::var("SONIC_SEARCH_PASS").ok())
+            .expect("Search sonic_search_pass nor SONIC_SEARCH_PASS env variable was provided")
+    }
+    pub fn sonic_ingest_addr(&self) -> String {
+        self.sonic_ingest_addr
+            .as_ref()
+            .cloned()
+            .or_else(|| std::env::var("SONIC_INGEST_ADDR").ok())
+            .expect("Search sonic_ingest_addr nor SONIC_INGEST_ADDR env variable was provided")
+    }
+    pub fn sonic_ingest_pass(&self) -> String {
+        self.sonic_ingest_pass
+            .as_ref()
+            .cloned()
+            .or_else(|| std::env::var("SONIC_INGEST_PASS").ok())
+            .expect("Search sonic_ingest_pass nor SONIC_INGEST_PASS env variable was provided")
+    }
+    pub fn search_active(&self) -> bool {
+        self.search_active
+            || std::env::var("SEARCH_ACTIVE")
+                .ok()
+                .map(|s| {
+                    let s = s.to_lowercase();
+                    s.as_str() == "true" || s.as_str() == "1" || s.as_str() == "enabled"
+                })
+                .unwrap_or_default()
+    }
+}
+
 #[derive(Serialize, Deserialize)]
 pub struct AppConfig {
     payment: PaymentConfig,
     web: WebConfig,
     mail: MailConfig,
     database: DatabaseConfig,
+    search: SearchConfig,
     #[serde(skip)]
     config_path: String,
 }
@@ -286,6 +361,7 @@ impl Example for AppConfig {
             web: WebConfig::example(),
             mail: MailConfig::example(),
             database: DatabaseConfig::example(),
+            search: SearchConfig::example(),
             config_path: "".to_string(),
         }
     }
@@ -323,6 +399,10 @@ impl AppConfig {
     pub fn database_mut(&mut self) -> &mut DatabaseConfig {
         &mut self.database
     }
+
+    pub fn search(&self) -> &SearchConfig {
+        &self.search
+    }
 }
 
 impl Default for AppConfig {
@@ -332,6 +412,7 @@ impl Default for AppConfig {
             web: WebConfig::default(),
             mail: Default::default(),
             database: DatabaseConfig::default(),
+            search: Default::default(),
             config_path: "".to_string(),
         }
     }
@@ -379,6 +460,11 @@ SENDGRID_SECRET - e-mail sending service secret
 SENDGRID_API_KEY - e-mail sending service api key
 SMTP_FROM - e-mail sending service authorized e-mail address used as sender e-mail address
 DATABASE_URL - postgresql address (ex. postgres://postgres@localhost/bazzar)
+SONIC_SEARCH_ADDR - search engine query address
+SONIC_SEARCH_PASS - search engine query password
+SONIC_INGEST_ADDR - search engine push address
+SONIC_INGEST_PASS - search engine push password
+SEARCH_ACTIVE - should use search engine
     "#
     );
 

api/src/main.rs

@@ -16,7 +16,9 @@ use opts::{
 use password_hash::SaltString;
 use validator::{validate_email, validate_length};
 
-use crate::actors::{database, email_manager, order_manager, payment_manager, token_manager};
+use crate::actors::{
+    database, email_manager, order_manager, payment_manager, search_manager, token_manager,
+};
 use crate::email_manager::TestMail;
 use crate::logic::encrypt_password;
 use crate::model::{Email, Login, PassHash, Password, Role};
@@ -58,6 +60,7 @@ async fn server(opts: ServerOpts) -> Result<()> {
         .await
         .expect("Failed to start payment manager")
         .start();
+    let search_manager = search_manager::SearchManager::new(app_config.clone());
     let addr = {
         let l = app_config.lock();
         let w = l.web();
@@ -84,6 +87,7 @@ async fn server(opts: ServerOpts) -> Result<()> {
             .app_data(Data::new(token_manager.clone()))
             .app_data(Data::new(order_manager.clone()))
             .app_data(Data::new(payment_manager.clone()))
+            .app_data(Data::new(search_manager.clone()))
             .configure(routes::configure)
         // .default_service(web::to(HttpResponse::Ok))
     })