chore: add appflowy indexer service start to cargo run xtask (#586)

Cargo.lock

@@ -778,6 +778,7 @@ dependencies = [
  "dashmap",
  "database",
  "database-entity",
+ "dotenvy",
  "env_logger",
  "futures",
  "humantime",

services/appflowy-indexer/Cargo.toml

@@ -31,7 +31,7 @@ log = "0.4"
 axum = "0.7"
 thiserror = "1.0"
 clap = { version = "4.5", features = ["derive", "env"] }
-tracing-subscriber = { version = "0.3", features = ["tracing-log", "env-filter"] }
+tracing-subscriber = { version = "0.3", features = ["tracing-log", "env-filter", "json"] }
 humantime = "2.1"
 dashmap = "5.5"
 uuid = { version = "1.8", features = ["v4"] }
@@ -39,6 +39,7 @@ env_logger = "0.11.3"
 async-stream = "0.3"
 async-trait = "0.1"
 rand = "0.8.5"
+dotenvy = "0.15.0"
 
 [dev-dependencies]
 env_logger = "0.11"

services/appflowy-indexer/dev.env

@@ -1,2 +1,2 @@
+APPFLOWY_INDEXER_DATABASE_URL=postgres://postgres:password@localhost:5432/postgres
 APPFLOWY_INDEXER_REDIS_URL=redis://localhost:6379
-APPFLOWY_INDEXER_AI_URL=http://localhost:5001

services/appflowy-indexer/src/main.rs

@@ -1,8 +1,8 @@
 use clap::Parser;
-use std::sync::Arc;
+use std::sync::{Arc, Once};
+use tracing::subscriber::set_global_default;
 use tracing_subscriber::layer::SubscriberExt;
-use tracing_subscriber::util::SubscriberInitExt;
+use tracing_subscriber::EnvFilter;
-use tracing_subscriber::Layer;
 
 use collab_stream::client::CollabRedisStream;
 
@@ -38,13 +38,22 @@ pub struct Config {
 
   #[clap(long, env = "APPFLOWY_INDEXER_INGEST_INTERVAL", default_value = "30s")]
   pub ingest_interval: humantime::Duration,
+
+  #[clap(
+    long,
+    env = "APPFLOWY_INDEXER_ENVIRONMENT",
+    default_value = "local",
+    value_enum
+  )]
+  pub app_env: Environment,
 }
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn std::error::Error>> {
-  setup_tracing();
+  dotenvy::dotenv().ok();
 
   let config = Config::parse();
+  init_subscriber(&config.app_env);
   run_server(config).await
 }
 
@@ -64,18 +73,41 @@ async fn run_server(config: Config) -> Result<(), Box<dyn std::error::Error>> {
   Ok(())
 }
 
-fn setup_tracing() {
+fn init_subscriber(app_env: &Environment) {
-  if std::env::var("RUST_LOG").is_err() {
+  static START: Once = Once::new();
-    std::env::set_var("RUST_LOG", "info");
+  START.call_once(|| {
-  }
+    let level = std::env::var("RUST_LOG").unwrap_or("info".to_string());
+    let mut filters = vec![];
+    filters.push(format!("appflowy_history={}", level));
+    let env_filter = EnvFilter::new(filters.join(","));
 
-  let registry = tracing_subscriber::registry();
+    let builder = tracing_subscriber::fmt()
+      .with_target(true)
+      .with_max_level(tracing::Level::TRACE)
+      .with_thread_ids(false)
+      .with_file(false);
 
-  registry
+    match app_env {
-    .with(
+      Environment::Local => {
-      tracing_subscriber::fmt::layer()
+        let subscriber = builder
-        .with_ansi(false)
+          .with_ansi(true)
-        .with_filter(tracing_subscriber::EnvFilter::from_default_env()),
+          .with_target(false)
-    )
+          .with_file(false)
-    .init();
+          .pretty()
+          .finish()
+          .with(env_filter);
+        set_global_default(subscriber).unwrap();
+      },
+      Environment::Production => {
+        let subscriber = builder.json().finish().with(env_filter);
+        set_global_default(subscriber).unwrap();
+      },
+    }
+  });
+}
+
+#[derive(Clone, Debug, clap::ValueEnum)]
+pub enum Environment {
+  Local,
+  Production,
 }

xtask/src/main.rs

@@ -5,6 +5,7 @@ use tokio::select;
 /// Using 'cargo run --package xtask' to run servers in parallel.
 /// 1. AppFlowy Cloud
 /// 2. AppFlowy History
+/// 3. AppFlowy Indexer
 ///
 /// Before running this command, make sure the other dependencies servers are running. For example,
 /// Redis, Postgres, etc.
@@ -12,9 +13,11 @@ use tokio::select;
 async fn main() -> Result<()> {
   let appflowy_cloud_bin_name = "appflowy_cloud";
   let appflowy_history_bin_name = "appflowy_history";
+  let appflowy_indexer_bin_name = "appflowy_indexer";
 
   kill_existing_process(appflowy_cloud_bin_name).await?;
   kill_existing_process(appflowy_history_bin_name).await?;
+  kill_existing_process(appflowy_indexer_bin_name).await?;
 
   let mut appflowy_cloud_cmd = Command::new("cargo")
     .args(["run", "--features", "history"])
@@ -30,6 +33,15 @@ async fn main() -> Result<()> {
     .spawn()
     .context("Failed to start AppFlowy-History process")?;
 
+  let mut appflowy_indexer_cmd = Command::new("cargo")
+    .args([
+      "run",
+      "--manifest-path",
+      "./services/appflowy-indexer/Cargo.toml",
+    ])
+    .spawn()
+    .context("Failed to start AppFlowy-Indexer process")?;
+
   select! {
       status = appflowy_cloud_cmd.wait() => {
           handle_process_exit(status?, appflowy_cloud_bin_name)?;
@@ -37,6 +49,9 @@ async fn main() -> Result<()> {
       status = appflowy_history_cmd.wait() => {
           handle_process_exit(status?, appflowy_history_bin_name)?;
       },
+      status = appflowy_indexer_cmd.wait() => {
+          handle_process_exit(status?, appflowy_indexer_bin_name)?;
+      },
   }
 
   Ok(())