feat(app): add binary app crates (apihost, githook, gitpro, livechat, mails)

Added app/apihost/src/main.rs +106 −0 View app/apihost/src/main.rs

@@ -0,0 +1,106 @@

1use std::net::{IpAddr, SocketAddr};

2use std::str::FromStr;

3

4use axum::serve;

5use config::ConfigLoader;

6use tokio::net::TcpListener;

7use tracing::info;

8use tracing_subscriber::EnvFilter;

9use webapi::{AppServices, router};

10

11const DEFAULT_BIND: &str = "0.0.0.0";

12const DEFAULT_PORT: u16 = 8080;

13

14fn init_tracing() {

15 let filter = EnvFilter::try_from_default_env()

16 .unwrap_or_else(|_| EnvFilter::new("info,webapi=debug,apihost=debug"));

17 tracing_subscriber::fmt()

18 .with_env_filter(filter)

19 .with_target(true)

20 .init();

21}

22

23#[tokio::main]

24async fn main() {

25 if let Err(e) = run().await {

26 eprintln!("apihost: fatal: {e}");

27 std::process::exit(1);

28 }

29}

30

31async fn run() -> Result<(), Box<dyn std::error::Error>> {

32 init_tracing();

33

34 let mut loader = ConfigLoader::new();

35 let mut bind = IpAddr::from_str(DEFAULT_BIND)?;

36 let mut port = DEFAULT_PORT;

37

38 let mut args = std::env::args().skip(1);

39 while let Some(arg) = args.next() {

40 match arg.as_str() {

41 "--config" => {

42 let path = args.next().ok_or("--config requires a path argument")?;

43 loader = loader.config_file(path);

44 }

45 "--bind" => {

46 let addr = args.next().ok_or("--bind requires an address argument")?;

47 bind = IpAddr::from_str(&addr)

48 .map_err(|_| format!("invalid --bind address: {addr}"))?;

49 }

50 "--port" => {

51 let value = args.next().ok_or("--port requires a port argument")?;

52 port = value

53 .parse::<u16>()

54 .map_err(|_| format!("invalid --port value: {value}"))?;

55 }

56 other => {

57 return Err(format!(

58 "unknown argument: {other} (usage: apihost [--config <path>] [--bind <addr>] [--port <port>])"

59 )

60 .into());

61 }

62 }

63 }

64

65 let app = loader.load()?;

66 let addr = SocketAddr::new(bind, port);

67

68 let services = AppServices::build(&app).await?;

69 let listener = TcpListener::bind(addr).await?;

70 info!(%addr, "webapi listening");

71

72 serve(listener, router(services))

73 .with_graceful_shutdown(shutdown_signal())

74 .await?;

75 info!("apihost shut down cleanly");

76 Ok(())

77}

78

79async fn shutdown_signal() {

80 let ctrl_c = async {

81 if let Err(e) = tokio::signal::ctrl_c().await {

82 tracing::warn!(error = %e, "failed to install ctrl-c handler");

83 }

84 };

85

86 #[cfg(unix)]

87 let term = async {

88 match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {

89 Ok(mut s) => {

90 s.recv().await;

91 }

92 Err(e) => {

93 tracing::warn!(error = %e, "failed to install SIGTERM handler");

94 std::future::pending::<()>().await;

95 }

96 }

97 };

98

99 #[cfg(not(unix))]

100 let term = std::future::pending::<()>();

101

102 tokio::select! {

103 _ = ctrl_c => tracing::info!("ctrl-c received, beginning shutdown"),

104 _ = term => tracing::info!("SIGTERM received, beginning shutdown"),

105 }

106}

Added app/apihost/Cargo.toml +17 −0 View app/apihost/Cargo.toml

@@ -0,0 +1,17 @@

1[package]

2name = "apihost"

3version.workspace = true

4edition.workspace = true

5authors.workspace = true

6categories.workspace = true

7keywords.workspace = true

8repository.workspace = true

9homepage.workspace = true

10

11[dependencies]

12webapi.workspace = true

13config.workspace = true

14axum.workspace = true

15tokio.workspace = true

16tracing.workspace = true

17tracing-subscriber.workspace = true

No newline at end of file

Added app/githook/src/main.rs +109 −0 View app/githook/src/main.rs

@@ -0,0 +1,109 @@

1use std::sync::Arc;

2

3use cache::{CacheBuilder, RedisCache};

4use config::ConfigLoader;

5use db::conn::Db;

6use gitsync::{NoopWebhookTx, SyncWorker};

7use tracing::{info, warn};

8

9fn init_tracing() {

10 let filter = tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| {

11 tracing_subscriber::EnvFilter::new("info,githook=debug,gitsync=debug")

12 });

13 tracing_subscriber::fmt()

14 .with_env_filter(filter)

15 .with_target(true)

16 .init();

17}

18

19#[tokio::main]

20async fn main() {

21 if let Err(e) = run().await {

22 eprintln!("githook: fatal: {e}");

23 std::process::exit(1);

24 }

25}

26

27async fn run() -> Result<(), Box<dyn std::error::Error>> {

28 init_tracing();

29

30 let mut loader = ConfigLoader::new();

31 let mut args = std::env::args().skip(1);

32 while let Some(arg) = args.next() {

33 match arg.as_str() {

34 "--config" => {

35 let path = args.next().ok_or("--config requires a path argument")?;

36 loader = loader.config_file(path);

37 }

38 other => {

39 return Err(

40 format!("unknown argument: {other} (usage: githook [--config <path>])").into(),

41 );

42 }

43 }

44 }

45

46 let app = loader.load()?;

47

48 let db = Db::from_config(&app.db).await?;

49 let cache = CacheBuilder::from_config(app.app_cache.clone())

50 .build()

51 .await?

52 .ok_or("app cache disabled at startup")?;

53 let kv = if app.cache.urls.is_empty() {

54 None

55 } else {

56 Some(Arc::new(RedisCache::new(&app.cache).await?))

57 };

58 let queue = queue::QueueBuilder::from_config(app.queue.clone())

59 .build()

60 .await?

61 .map(Arc::new);

62 let storage = storage::repo::from_config(&app.storage)?;

63

64 let worker = SyncWorker::new(

65 db,

66 Arc::new(app.repo.clone()),

67 kv.ok_or("gitsync requires redis kv (cache.urls) to be configured")?,

68 Arc::new(cache),

69 storage,

70 queue,

71 Arc::new(NoopWebhookTx),

72 app.git_sync.clone(),

73 );

74 let running = worker.run().await?;

75

76 wait_for_shutdown().await;

77 running.shutdown().await?;

78 info!("githook shut down cleanly");

79 Ok(())

80}

81

82async fn wait_for_shutdown() {

83 let ctrl_c = async {

84 if let Err(e) = tokio::signal::ctrl_c().await {

85 warn!(error = %e, "failed to install ctrl-c handler");

86 }

87 };

88

89 #[cfg(unix)]

90 let term = async {

91 match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {

92 Ok(mut s) => {

93 s.recv().await;

94 }

95 Err(e) => {

96 warn!(error = %e, "failed to install SIGTERM handler");

97 std::future::pending::<()>().await;

98 }

99 }

100 };

101

102 #[cfg(not(unix))]

103 let term = std::future::pending::<()>();

104

105 tokio::select! {

106 _ = ctrl_c => info!("ctrl-c received, beginning shutdown"),

107 _ = term => info!("SIGTERM received, beginning shutdown"),

108 }

109}

No newline at end of file

Added app/githook/Cargo.toml +20 −0 View app/githook/Cargo.toml

@@ -0,0 +1,20 @@

1[package]

2name = "githook"

3version.workspace = true

4edition.workspace = true

5authors.workspace = true

6categories.workspace = true

7keywords.workspace = true

8repository.workspace = true

9homepage.workspace = true

10

11[dependencies]

12cache.workspace = true

13config.workspace = true

14db.workspace = true

15gitsync.workspace = true

16queue.workspace = true

17storage.workspace = true

18tokio.workspace = true

19tracing.workspace = true

20tracing-subscriber.workspace = true

No newline at end of file

Added app/gitpro/src/main.rs +141 −0 View app/gitpro/src/main.rs

@@ -0,0 +1,141 @@

1use std::sync::Arc;

2

3use cache::RedisCache;

4use config::ConfigLoader;

5use db::conn::Db;

6use gitauth::GitAuthService;

7use jwt::{JwtState, SigningKeyCache};

8use tokio::net::TcpListener;

9use tracing::{info, warn};

10

11fn init_tracing() {

12 let filter = tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| {

13 tracing_subscriber::EnvFilter::new("info,gitpro=debug,gitsmart=debug,gitssh=debug")

14 });

15 tracing_subscriber::fmt()

16 .with_env_filter(filter)

17 .with_target(true)

18 .init();

19}

20

21#[tokio::main]

22async fn main() {

23 if let Err(e) = run().await {

24 eprintln!("gitpro: fatal: {e}");

25 std::process::exit(1);

26 }

27}

28

29async fn run() -> Result<(), Box<dyn std::error::Error>> {

30 init_tracing();

31

32 let mut loader = ConfigLoader::new();

33 let mut args = std::env::args().skip(1);

34 while let Some(arg) = args.next() {

35 match arg.as_str() {

36 "--config" => {

37 let path = args.next().ok_or("--config requires a path argument")?;

38 loader = loader.config_file(path);

39 }

40 other => {

41 return Err(

42 format!("unknown argument: {other} (usage: gitpro [--config <path>])").into(),

43 );

44 }

45 }

46 }

47

48 let app = loader.load()?;

49

50 let db = Db::from_config(&app.db).await?;

51 let kv = if app.cache.urls.is_empty() {

52 None

53 } else {

54 Some(Arc::new(RedisCache::new(&app.cache).await?))

55 };

56 let queue = queue::QueueBuilder::from_config(app.queue.clone())

57 .build()

58 .await?

59 .map(Arc::new);

60 let storage = storage::repo::from_config(&app.storage)?;

61

62 let jwt = JwtState::new(

63 db.writer(),

64 app.jwt.clone(),

65 Arc::new(tokio::sync::RwLock::new(SigningKeyCache::new())),

66 );

67 let auth = GitAuthService::new(db, kv.clone(), jwt).with_config(app.git_auth.clone());

68

69 let smart = gitsmart::SmartService::new(auth.clone(), app.git_smart.clone(), storage)

70 .await?

71 .with_kv(kv.clone())

72 .with_queue(queue.clone())

73 .with_sync_settings(app.git_sync.clone());

74 let http_listener = TcpListener::bind((

75 app.git_smart.bind_addr.as_str(),

76 app.git_smart.port,

77 ))

78 .await?;

79 info!(

80 addr = %app.git_smart.bind_addr,

81 port = app.git_smart.port,

82 "gitsmart http server listening"

83 );

84 let router = gitsmart::handler::router(Arc::new(smart));

85

86 let ssh = gitssh::GitSshService::new(auth, app.git_ssh.clone())

87 .with_kv(kv)

88 .with_queue(queue)

89 .with_sync_settings(app.git_sync);

90 let ssh_listener = TcpListener::bind((app.git_ssh.bind_addr.as_str(), app.git_ssh.port)).await?;

91 info!(

92 addr = %app.git_ssh.bind_addr,

93 port = app.git_ssh.port,

94 "gitssh server listening"

95 );

96

97 let http_task = tokio::spawn(async move {

98 axum::serve(http_listener, router)

99 .with_graceful_shutdown(shutdown_signal())

100 .await

101 });

102 let ssh_task = tokio::spawn(async move {

103 gitssh::server::serve_on_listener(ssh, ssh_listener).await

104 });

105

106 tokio::select! {

107 http = http_task => http??,

108 ssh = ssh_task => ssh??,

109 }

110 info!("gitpro shut down cleanly");

111 Ok(())

112}

113

114async fn shutdown_signal() {

115 let ctrl_c = async {

116 if let Err(e) = tokio::signal::ctrl_c().await {

117 warn!(error = %e, "failed to install ctrl-c handler");

118 }

119 };

120

121 #[cfg(unix)]

122 let term = async {

123 match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {

124 Ok(mut s) => {

125 s.recv().await;

126 }

127 Err(e) => {

128 warn!(error = %e, "failed to install SIGTERM handler");

129 std::future::pending::<()>().await;

130 }

131 }

132 };

133

134 #[cfg(not(unix))]

135 let term = std::future::pending::<()>();

136

137 tokio::select! {

138 _ = ctrl_c => info!("ctrl-c received, shutting down"),

139 _ = term => info!("SIGTERM received, shutting down"),

140 }

141}

No newline at end of file

Added app/gitpro/Cargo.toml +24 −0 View app/gitpro/Cargo.toml

@@ -0,0 +1,24 @@

1[package]

2name = "gitpro"

3version.workspace = true

4edition.workspace = true

5authors.workspace = true

6categories.workspace = true

7keywords.workspace = true

8repository.workspace = true

9homepage.workspace = true

10

11[dependencies]

12axum.workspace = true

13cache.workspace = true

14config.workspace = true

15db.workspace = true

16gitauth.workspace = true

17gitsmart.workspace = true

18gitssh.workspace = true

19jwt.workspace = true

20queue.workspace = true

21storage.workspace = true

22tokio.workspace = true

23tracing.workspace = true

24tracing-subscriber.workspace = true

No newline at end of file

Added app/livechat/src/main.rs +211 −0 View app/livechat/src/main.rs

@@ -0,0 +1,211 @@

1use std::net::{IpAddr, SocketAddr};

2use std::str::FromStr;

3use std::sync::Arc;

4

5use axum::{Json, Router, routing::get, serve};

6use cache::redis::RedisCache;

7use chatlayer::{AuthBridge, ChatProducer, ChatService, MediaService};

8use config::ConfigLoader;

9use db::conn::Db;

10use glayer::{ConnectionHub, GatewayState};

11use jwt::JwtState;

12use jwt::keys::SigningKeyCache;

13use live::{LiveService, live_router};

14use queue::QueueBuilder;

15use serde::Serialize;

16use storage::repo::from_config as storage_from_config;

17use storagelayer::blob::ObjectStoreChatMedia;

18use storagelayer::mapper::SeaChatStorage;

19use tokio::net::TcpListener;

20use tokio::sync::RwLock;

21use tracing::info;

22use tracing_subscriber::EnvFilter;

23

24const DEFAULT_BIND: &str = "0.0.0.0";

25const DEFAULT_PORT: u16 = 8080;

26

27#[derive(Serialize)]

28struct HealthStatus {

29 status: &'static str,

30 service: &'static str,

31}

32

33async fn liveness() -> Json<HealthStatus> {

34 Json(HealthStatus {

35 status: "ok",

36 service: "untitled-chat",

37 })

38}

39

40async fn readiness() -> Json<HealthStatus> {

41 Json(HealthStatus {

42 status: "ok",

43 service: "untitled-chat",

44 })

45}

46

47fn init_tracing() {

48 let filter = EnvFilter::try_from_default_env().unwrap_or_else(|_| {

49 EnvFilter::new("info,chatlayer=debug,glayer=debug,live=debug,livechat=debug")

50 });

51 tracing_subscriber::fmt()

52 .with_env_filter(filter)

53 .with_target(true)

54 .init();

55}

56

57#[tokio::main]

58async fn main() {

59 if let Err(e) = run().await {

60 eprintln!("livechat: fatal: {e}");

61 std::process::exit(1);

62 }

63}

64

65async fn run() -> Result<(), Box<dyn std::error::Error>> {

66 init_tracing();

67

68 let mut loader = ConfigLoader::new();

69 let mut bind = IpAddr::from_str(DEFAULT_BIND)?;

70 let mut port = DEFAULT_PORT;

71

72 let mut args = std::env::args().skip(1);

73 while let Some(arg) = args.next() {

74 match arg.as_str() {

75 "--config" => {

76 let path = args.next().ok_or("--config requires a path argument")?;

77 loader = loader.config_file(path);

78 }

79 "--bind" => {

80 let addr = args.next().ok_or("--bind requires an address argument")?;

81 bind = IpAddr::from_str(&addr)

82 .map_err(|_| format!("invalid --bind address: {addr}"))?;

83 }

84 "--port" => {

85 let value = args.next().ok_or("--port requires a port argument")?;

86 port = value

87 .parse::<u16>()

88 .map_err(|_| format!("invalid --port value: {value}"))?;

89 }

90 other => {

91 return Err(format!(

92 "unknown argument: {other} (usage: livechat [--config <path>] [--bind <addr>] [--port <port>])"

93 )

94 .into());

95 }

96 }

97 }

98

99 let app = loader.load()?;

100 let addr = SocketAddr::new(bind, port);

101

102 let db = Arc::new(Db::from_config(&app.db).await?);

103 info!("database connected");

104 let kv = Arc::new(RedisCache::new(&app.cache).await?);

105 info!("cache connected");

106

107 let queue = QueueBuilder::from_config(app.queue.clone())

108 .build()

109 .await?

110 .map(Arc::new);

111 info!(enabled = queue.is_some(), "queue ready");

112

113 let storage = storage_from_config(&app.storage)?;

114 info!("storage ready");

115

116 let jwt = JwtState::new(

117 db.writer(),

118 app.jwt.clone(),

119 Arc::new(RwLock::new(SigningKeyCache::new())),

120 );

121

122 let blobs: Arc<dyn storagelayer::blob::ChatMediaStore> =

123 Arc::new(ObjectStoreChatMedia::new(storage.clone()));

124 let store: storagelayer::DynChatStorage = Arc::new(SeaChatStorage::new(db.clone(), blobs));

125 let producer = Arc::new(ChatProducer::new(

126 queue.clone(),

127 app.chat.topic_prefix.clone(),

128 ));

129 let auth = AuthBridge::new(Arc::new(jwt.clone()));

130 let media = Arc::new(MediaService::new(

131 store.clone(),

132 storage.clone(),

133 kv.clone(),

134 app.chat.upload_ttl_seconds,

135 ));

136

137 let chat = Arc::new(ChatService::new(

138 store.clone(),

139 producer.clone(),

140 kv.clone(),

141 auth,

142 media,

143 app.chat.clone(),

144 ));

145 let hub = ConnectionHub::new(app.gateway.max_connections_per_user);

146 let gateway = GatewayState::new(chat.clone(), hub, app.gateway.clone());

147 let live = LiveService::new(store, producer, kv, jwt, app.live.clone());

148

149 let room_worker = chat.spawn_room_worker()?;

150 let user_worker = chat.spawn_user_worker()?;

151 let live_worker = chat.spawn_live_worker()?;

152 info!("chat workers started (room, user, live)");

153

154 let user_view = gateway.spawn_user_view_consumer()?;

155 let typing = gateway.spawn_typing_consumer()?;

156 let live_view = gateway.spawn_live_view_consumer()?;

157 info!("gateway consumers started (user-view, typing, live-view)");

158

159 let routes = Router::new()

160 .route("/health/live", get(liveness))

161 .route("/health/ready", get(readiness))

162 .merge(glayer::router(gateway.clone()))

163 .merge(live_router(live));

164 let listener = TcpListener::bind(addr).await?;

165 info!(%addr, "livechat gateway listening");

166

167 serve(listener, routes)

168 .with_graceful_shutdown(shutdown_signal())

169 .await?;

170 info!("livechat shut down cleanly");

171

172 chat.shutdown().await?;

173 gateway.shutdown_token().cancel();

174 for handle in [room_worker, user_worker, live_worker] {

175 handle.shutdown().await?;

176 }

177 for handle in [user_view, typing, live_view] {

178 handle.shutdown().await?;

179 }

180 info!("chat workers shut down cleanly");

181 Ok(())

182}

183

184async fn shutdown_signal() {

185 let ctrl_c = async {

186 if let Err(e) = tokio::signal::ctrl_c().await {

187 tracing::warn!(error = %e, "failed to install ctrl-c handler");

188 }

189 };

190

191 #[cfg(unix)]

192 let term = async {

193 match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) {

194 Ok(mut s) => {

195 s.recv().await;

196 }

197 Err(e) => {

198 tracing::warn!(error = %e, "failed to install SIGTERM handler");

199 std::future::pending::<()>().await;

200 }

201 }

202 };

203

204 #[cfg(not(unix))]

205 let term = std::future::pending::<()>();

206

207 tokio::select! {

208 _ = ctrl_c => tracing::info!("ctrl-c received, beginning shutdown"),

209 _ = term => tracing::info!("SIGTERM received, beginning shutdown"),

210 }

211}

Added app/livechat/Cargo.toml +27 −0 View app/livechat/Cargo.toml

@@ -0,0 +1,27 @@

1[package]

2name = "livechat"

3version.workspace = true

4edition.workspace = true

5authors.workspace = true

6categories.workspace = true

7keywords.workspace = true

8repository.workspace = true

9homepage.workspace = true

10

11[dependencies]

12config.workspace = true

13chatlayer.workspace = true

14glayer.workspace = true

15live.workspace = true

16storagelayer.workspace = true

17db.workspace = true

18cache.workspace = true

19queue.workspace = true

20storage.workspace = true

21jwt.workspace = true

22

23axum.workspace = true

24tokio.workspace = true

25serde.workspace = true

26tracing.workspace = true

27tracing-subscriber.workspace = true

No newline at end of file

Added app/mails/src/main.rs +58 −0 View app/mails/src/main.rs

@@ -0,0 +1,58 @@

1use std::path::PathBuf;

2

3use config::ConfigLoader;

4use email::{WorkerOptions, run_until_shutdown};

5use tracing::info;

6use tracing_subscriber::EnvFilter;

7

8fn init_tracing() {

9 let filter = EnvFilter::try_from_default_env()

10 .unwrap_or_else(|_| EnvFilter::new("info,email=debug,mails=debug"));

11 tracing_subscriber::fmt()

12 .with_env_filter(filter)

13 .with_target(true)

14 .init();

15}

16

17#[tokio::main]

18async fn main() {

19 if let Err(e) = run().await {

20 eprintln!("mails: fatal: {e}");

21 std::process::exit(1);

22 }

23}

24

25async fn run() -> Result<(), Box<dyn std::error::Error>> {

26 init_tracing();

27

28 let mut loader = ConfigLoader::new();

29 let mut args = std::env::args().skip(1);

30 while let Some(arg) = args.next() {

31 match arg.as_str() {

32 "--config" => {

33 let path = args.next().ok_or("--config requires a path argument")?;

34 loader = loader.config_file(path);

35 }

36 other => {

37 return Err(

38 format!("unknown argument: {other} (usage: mails [--config <path>])").into(),

39 );

40 }

41 }

42 }

43

44 let app = loader.load()?;

45 info!(

46 enabled = %app.email.enabled,

47 smtp = %format!("{}:{}", app.email.smtp_host, app.email.smtp_port),

48 from = %app.email.from_address,

49 template_dir = ?app.email.template_dir,

50 "email worker configuration loaded"

51 );

52

53 let opts = WorkerOptions {

54 template_dir: app.email.template_dir.clone().map(PathBuf::from),

55 };

56 run_until_shutdown(app.email, app.queue, opts).await?;

57 Ok(())

58}

Added app/mails/Cargo.toml +16 −0 View app/mails/Cargo.toml

@@ -0,0 +1,16 @@

1[package]

2name = "mails"

3version.workspace = true

4edition.workspace = true

5authors.workspace = true

6categories.workspace = true

7keywords.workspace = true

8repository.workspace = true

9homepage.workspace = true

10

11[dependencies]

12config.workspace = true

13email.workspace = true

14tokio.workspace = true

15tracing.workspace = true

16tracing-subscriber.workspace = true

No newline at end of file