feat(apps): assemble mails worker and apihost/githost/synchost binaries

Modified app/apihost/src/main.rs +89 −2 View app/apihost/src/main.rs

@@ -1,3 +1,90 @@

1fn main() {

2 println!("Hello, world!");

1use std::net::SocketAddr;

2

3use anyhow::Context;

4use axum::Router;

5use tokio::net::TcpListener;

6use tracing::{info, warn};

7use utoipa_swagger_ui::SwaggerUi;

8use webapi::{openapi::api_doc, router};

9

10#[tokio::main]

11async fn main() -> anyhow::Result<()> {

12 init_tracing();

13

14 let cfg = config::AppConfig::load().context("loading configuration")?;

15 cfg.validate()

16 .map_err(|e| anyhow::anyhow!(e.to_string()))?;

17 if !cfg.apihost.enabled() {

18 info!("apihost is disabled by config; exiting");

19 return Ok(());

20 }

21

22 info!("connecting to database / redis / storage / email");

23 let state = export::AppState::with_config(cfg.clone())

24 .await

25 .context("building AppState")?;

26 info!("dependencies ready");

27

28 let api_router: Router = router(state)

29 .await

30 .map_err(|err| anyhow::anyhow!("failed to build router: {err}"))?;

31 let doc = api_doc();

32 let api_router = api_router.merge(

33 SwaggerUi::new("/api/docs").url("/api/docs/openapi.json", doc),

34 );

35

36 let bind_addr = cfg.apihost.bind_addr().to_string();

37 let port = cfg.apihost.port();

38 let addr: SocketAddr = format!("{bind_addr}:{port}")

39 .parse()

40 .with_context(|| format!("parsing bind address {bind_addr}:{port}"))?;

41

42 let listener = TcpListener::bind(addr)

43 .await

44 .with_context(|| format!("binding {addr}"))?;

45 info!(%addr, "apihost listening");

46

47 axum::serve(listener, api_router.into_make_service())

48 .with_graceful_shutdown(shutdown_signal())

49 .await

50 .context("axum server error")?;

51

52 info!("apihost shut down");

53 Ok(())

354}

55

56fn init_tracing() {

57 use tracing_subscriber::EnvFilter;

58 let filter = EnvFilter::try_from_default_env()

59 .unwrap_or_else(|_| EnvFilter::new("info"));

60 tracing_subscriber::fmt().with_env_filter(filter).init();

61}

62

63async fn shutdown_signal() {

64 let ctrl_c = async {

65 if let Err(err) = tokio::signal::ctrl_c().await {

66 warn!(error = %err, "failed to install ctrl_c handler");

67 }

68 };

69

70 #[cfg(unix)]

71 let terminate = async {

72 use tokio::signal::unix::{signal, SignalKind};

73 match signal(SignalKind::terminate()) {

74 Ok(mut s) => {

75 s.recv().await;

76 }

77 Err(err) => {

78 warn!(error = %err, "failed to install SIGTERM handler");

79 }

80 }

81 };

82

83 #[cfg(not(unix))]

84 let terminate = std::future::pending::<()>();

85

86 tokio::select! {

87 _ = ctrl_c => info!("received SIGINT"),

88 _ = terminate => info!("received SIGTERM"),

89 }

90}

No newline at end of file

Modified app/apihost/Cargo.toml +12 −0 View app/apihost/Cargo.toml

@@ -2,5 +2,17 @@

22name = "apihost"

33version.workspace = true

44edition.workspace = true

5publish = false

56

67[dependencies]

8webapi = { workspace = true }

9export = { workspace = true }

10session = { workspace = true }

11config = { workspace = true }

12axum = { workspace = true }

13tokio = { workspace = true, features = ["signal", "net", "macros", "rt-multi-thread"] }

14tracing = { workspace = true }

15tracing-subscriber = { workspace = true }

16anyhow = { workspace = true }

17utoipa = { workspace = true }

18utoipa-swagger-ui = { workspace = true }

No newline at end of file

Modified app/githost/src/main.rs +115 −2 View app/githost/src/main.rs

@@ -1,3 +1,116 @@

1fn main() {

2 println!("Hello, world!");

1use std::net::SocketAddr;

2use std::sync::Arc;

3

4use anyhow::Context;

5use tokio::net::TcpListener;

6use tracing::{info, warn};

7

8#[tokio::main]

9async fn main() -> anyhow::Result<()> {

10 init_tracing();

11

12 let cfg = config::AppConfig::load().context("loading configuration")?;

13 cfg.validate()

14 .map_err(|e| anyhow::anyhow!(e.to_string()))?;

15 if !cfg.githost.enabled() {

16 info!("githost is disabled by config; exiting");

17 return Ok(());

18 }

19

20 info!("connecting to database / redis / storage / email");

21 let state = export::AppState::with_config(cfg.clone())

22 .await

23 .context("building AppState")?;

24 info!("dependencies ready");

25

26 let git_svc = git::service::GitService::new(state.clone());

27 let sync_queue = Arc::new(git::sync::queue::SyncQueue::new(state.redis.clone()));

28 let smart_svc = git::smart::service::SmartService::new(git_svc)

29 .with_kv(Some(Arc::new(state.redis.clone())))

30 .with_queue(Some(sync_queue))

31 .with_sync_settings(cfg.git.sync.clone());

32 let ssh_svc = git::ssh::service::GitSshService::new(smart_svc.clone(), cfg.git.ssh.clone());

33 let smart_router = git::smart::handler::router(smart_svc);

34

35 let http_addr = SocketAddr::new(cfg.githost.http_bind_addr(), cfg.githost.http_port());

36 let http_listener = TcpListener::bind(http_addr)

37 .await

38 .with_context(|| format!("binding smart HTTP {http_addr}"))?;

39 info!(%http_addr, "githost smart HTTP listening");

40

41 let ssh_bind = cfg.git.ssh.bind_addr.clone();

42 let ssh_port = cfg.git.ssh.port;

43 let ssh_addr: SocketAddr = format!("{ssh_bind}:{ssh_port}")

44 .parse()

45 .with_context(|| format!("parsing ssh address {ssh_bind}:{ssh_port}"))?;

46 let ssh_listener = TcpListener::bind(ssh_addr)

47 .await

48 .with_context(|| format!("binding SSH {ssh_addr}"))?;

49 info!(%ssh_addr, "githost SSH listening");

50

51 let (http_tx, http_rx) = tokio::sync::oneshot::channel::<()>();

52 let (ssh_tx, ssh_rx) = tokio::sync::oneshot::channel::<()>();

53

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

55 let http_shutdown = async move {

56 let _ = http_rx.await;

57 };

58 axum::serve(http_listener, smart_router.into_make_service())

59 .with_graceful_shutdown(http_shutdown)

60 .await

61 .map_err(anyhow::Error::from)

62 });

63

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

65 tokio::select! {

66 res = git::ssh::serve_on_listener(ssh_svc, ssh_listener) => res.map_err(anyhow::Error::from),

67 _ = ssh_rx => Ok(()),

68 }

69 });

70

71 shutdown_signal().await;

72 info!("shutdown signal received; stopping listeners");

73 let _ = http_tx.send(());

74 let _ = ssh_tx.send(());

75

76 http_task.await.context("smart HTTP task panicked")??;

77 let _ = ssh_task.await;

78 info!("githost shut down");

79 Ok(())

80}

81

82fn init_tracing() {

83 use tracing_subscriber::EnvFilter;

84 let filter = EnvFilter::try_from_default_env()

85 .unwrap_or_else(|_| EnvFilter::new("info"));

86 tracing_subscriber::fmt().with_env_filter(filter).init();

387}

88

89async fn shutdown_signal() {

90 let ctrl_c = async {

91 if let Err(err) = tokio::signal::ctrl_c().await {

92 warn!(error = %err, "failed to install ctrl_c handler");

93 }

94 };

95

96 #[cfg(unix)]

97 let terminate = async {

98 use tokio::signal::unix::{SignalKind, signal};

99 match signal(SignalKind::terminate()) {

100 Ok(mut s) => {

101 s.recv().await;

102 }

103 Err(err) => {

104 warn!(error = %err, "failed to install SIGTERM handler");

105 }

106 }

107 };

108

109 #[cfg(not(unix))]

110 let terminate = std::future::pending::<()>();

111

112 tokio::select! {

113 _ = ctrl_c => info!("received SIGINT"),

114 _ = terminate => info!("received SIGTERM"),

115 }

116}

No newline at end of file

Modified app/githost/Cargo.toml +8 −0 View app/githost/Cargo.toml

@@ -4,3 +4,11 @@

44edition.workspace = true

55

66[dependencies]

7git = { workspace = true }

8export = { workspace = true }

9config = { workspace = true }

10axum = { workspace = true }

11tokio = { workspace = true, features = ["signal", "net", "macros", "rt-multi-thread", "sync"] }

12tracing = { workspace = true }

13tracing-subscriber = { workspace = true }

14anyhow = { workspace = true }

No newline at end of file

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

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

1use std::time::Duration;

2

3use email::runtime::{run_until_shutdown, WorkerOptions};

4

5#[tokio::main]

6async fn main() -> anyhow::Result<()> {

7 tracing_subscriber::fmt()

8 .with_env_filter(

9 tracing_subscriber::EnvFilter::try_from_default_env()

10 .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),

11 )

12 .init();

13

14 let cfg = config::AppConfig::load()?;

15 cfg.validate().map_err(|e| anyhow::anyhow!(e.to_string()))?;

16

17 let redis_url = cfg.redis.url().to_string();

18

19 let opts = WorkerOptions {

20 template_dir: None,

21 shutdown_timeout: Some(Duration::from_secs(5)),

22 };

23 run_until_shutdown(cfg.email.clone(), cfg.email_queue.clone(), redis_url, opts)

24 .await

25 .map_err(|e| anyhow::anyhow!(e.to_string()))?;

26 Ok(())

27}

No newline at end of file

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

@@ -0,0 +1,13 @@

1[package]

2name = "mails"

3version.workspace = true

4edition.workspace = true

5publish = false

6

7[dependencies]

8email = { workspace = true }

9config = { workspace = true }

10tokio = { workspace = true }

11tracing = { workspace = true }

12tracing-subscriber = { workspace = true }

13anyhow = { workspace = true }

No newline at end of file

Modified app/synchost/src/main.rs +106 −2 View app/synchost/src/main.rs

@@ -1,3 +1,107 @@

1fn main() {

2 println!("Hello, world!");

1use std::path::PathBuf;

2use std::sync::Arc;

3

4use anyhow::Context;

5use cache::CacheClient;

6use git::sync::{NoopWebhookTx, SyncQueue, SyncWorker, WebhookTx};

7use tokio_util::sync::CancellationToken;

8use tracing::{info, warn};

9

10#[tokio::main]

11async fn main() -> anyhow::Result<()> {

12 init_tracing();

13

14 let cfg = config::AppConfig::load().context("loading configuration")?;

15 cfg.validate()

16 .map_err(|e| anyhow::anyhow!(e.to_string()))?;

17

18 info!("connecting to database / redis / cache");

19 let db = Arc::new(

20 export::AppDatabase::connect(cfg.database.clone())

21 .await

22 .context("connecting to postgres")?,

23 );

24 let redis = export::AppRedis::connect(cfg.redis.clone())

25 .await

26 .context("connecting to redis")?;

27 let cache = Arc::new(

28 CacheClient::new(&cfg.cache)

29 .await

30 .context("building cache client")?,

31 );

32 let repo_root = PathBuf::from(cfg.git.repo_root.clone());

33 info!("dependencies ready");

34

35 let cancel = CancellationToken::new();

36 let queue = SyncQueue::new(redis.clone());

37 let webhook: Arc<dyn WebhookTx> = Arc::new(NoopWebhookTx);

38 let settings = cfg.git.sync.clone();

39

40 let worker = SyncWorker::new(

41 db,

42 redis,

43 cache,

44 repo_root,

45 queue,

46 webhook,

47 settings.clone(),

48 cancel.clone(),

49 );

50 let running = worker

51 .run()

52 .await

53 .map_err(|e| anyhow::anyhow!("starting sync worker: {e}"))?;

54 info!(

55 concurrency = settings.concurrency.max(1),

56 topic = %settings.queue_topic,

57 "synchost running"

58 );

59

60 wait_for_shutdown().await;

61 info!("synchost shutting down");

62

63 cancel.cancel();

64 running

65 .shutdown()

66 .await

67 .map_err(|e| anyhow::anyhow!("worker shutdown: {e}"))?;

68

69 info!("synchost shut down");

70 Ok(())

371}

72

73fn init_tracing() {

74 use tracing_subscriber::EnvFilter;

75 let filter = EnvFilter::try_from_default_env()

76 .unwrap_or_else(|_| EnvFilter::new("info"));

77 tracing_subscriber::fmt().with_env_filter(filter).init();

78}

79

80async fn wait_for_shutdown() {

81 let ctrl_c = async {

82 if let Err(err) = tokio::signal::ctrl_c().await {

83 warn!(error = %err, "failed to install ctrl_c handler");

84 }

85 };

86

87 #[cfg(unix)]

88 let terminate = async {

89 use tokio::signal::unix::{signal, SignalKind};

90 match signal(SignalKind::terminate()) {

91 Ok(mut s) => {

92 s.recv().await;

93 }

94 Err(err) => {

95 warn!(error = %err, "failed to install SIGTERM handler");

96 }

97 }

98 };

99

100 #[cfg(not(unix))]

101 let terminate = std::future::pending::<()>();

102

103 tokio::select! {

104 _ = ctrl_c => info!("received SIGINT"),

105 _ = terminate => info!("received SIGTERM"),

106 }

107}

No newline at end of file

Modified app/synchost/Cargo.toml +10 −0 View app/synchost/Cargo.toml

@@ -2,5 +2,15 @@

22name = "synchost"

33version.workspace = true

44edition.workspace = true

5publish = false

56

67[dependencies]

8git = { workspace = true }

9export = { workspace = true }

10config = { workspace = true }

11cache = { workspace = true }

12tokio = { workspace = true, features = ["signal", "rt-multi-thread", "macros"] }

13tokio-util = { workspace = true }

14tracing = { workspace = true }

15tracing-subscriber = { workspace = true }

16anyhow = { workspace = true }

No newline at end of file