1use anyhow::anyhow;
2use axum::{routing::get, Router};
3use collab::{db, env, executor::Executor, AppState, Config, MigrateConfig, Result};
4use db::Database;
5use std::{
6 env::args,
7 net::{SocketAddr, TcpListener},
8 path::Path,
9};
10use tracing_log::LogTracer;
11use tracing_subscriber::{filter::EnvFilter, fmt::format::JsonFields, Layer};
12use util::ResultExt;
13
14const VERSION: &'static str = env!("CARGO_PKG_VERSION");
15
16#[tokio::main]
17async fn main() -> Result<()> {
18 if let Err(error) = env::load_dotenv() {
19 eprintln!(
20 "error loading .env.toml (this is expected in production): {}",
21 error
22 );
23 }
24
25 match args().skip(1).next().as_deref() {
26 Some("version") => {
27 println!("collab v{VERSION}");
28 }
29 Some("migrate") => {
30 let config = envy::from_env::<MigrateConfig>().expect("error loading config");
31 let mut db_options = db::ConnectOptions::new(config.database_url.clone());
32 db_options.max_connections(5);
33 let db = Database::new(db_options).await?;
34
35 let migrations_path = config
36 .migrations_path
37 .as_deref()
38 .unwrap_or_else(|| Path::new(concat!(env!("CARGO_MANIFEST_DIR"), "/migrations")));
39
40 let migrations = db.migrate(&migrations_path, false).await?;
41 for (migration, duration) in migrations {
42 println!(
43 "Ran {} {} {:?}",
44 migration.version, migration.description, duration
45 );
46 }
47
48 return Ok(());
49 }
50 Some("serve") => {
51 let config = envy::from_env::<Config>().expect("error loading config");
52 init_tracing(&config);
53
54 let state = AppState::new(config).await?;
55
56 let listener = TcpListener::bind(&format!("0.0.0.0:{}", state.config.http_port))
57 .expect("failed to bind TCP listener");
58
59 let rpc_server = collab::rpc::Server::new(state.clone(), Executor::Production);
60 rpc_server.start().await?;
61
62 let app = collab::api::routes(rpc_server.clone(), state.clone())
63 .merge(collab::rpc::routes(rpc_server.clone()))
64 .merge(Router::new().route("/", get(handle_root)));
65
66 axum::Server::from_tcp(listener)?
67 .serve(app.into_make_service_with_connect_info::<SocketAddr>())
68 .with_graceful_shutdown(async move {
69 tokio::signal::ctrl_c()
70 .await
71 .expect("failed to listen for interrupt signal");
72 tracing::info!("Received interrupt signal");
73 rpc_server.teardown();
74 })
75 .await?;
76 }
77 _ => {
78 Err(anyhow!("usage: collab <version | migrate | serve>"))?;
79 }
80 }
81 Ok(())
82}
83
84async fn handle_root() -> String {
85 format!("collab v{VERSION}")
86}
87
88pub fn init_tracing(config: &Config) -> Option<()> {
89 use std::str::FromStr;
90 use tracing_subscriber::layer::SubscriberExt;
91 let rust_log = config.rust_log.clone()?;
92
93 LogTracer::init().log_err()?;
94
95 let subscriber = tracing_subscriber::Registry::default()
96 .with(if config.log_json.unwrap_or(false) {
97 Box::new(
98 tracing_subscriber::fmt::layer()
99 .fmt_fields(JsonFields::default())
100 .event_format(
101 tracing_subscriber::fmt::format()
102 .json()
103 .flatten_event(true)
104 .with_span_list(true),
105 ),
106 ) as Box<dyn Layer<_> + Send + Sync>
107 } else {
108 Box::new(
109 tracing_subscriber::fmt::layer()
110 .event_format(tracing_subscriber::fmt::format().pretty()),
111 )
112 })
113 .with(EnvFilter::from_str(rust_log.as_str()).log_err()?);
114
115 tracing::subscriber::set_global_default(subscriber).unwrap();
116
117 None
118}