194c44c
use std::net::SocketAddr;
use std::path::PathBuf;
use std::sync::Arc;
use anyhow::Context;
use cache::CacheClient;
use config::scoped::{ScopedConfig, SynchostConfig};
use git::sync::{RepoWebhookTx, SyncQueue, SyncWorker, WebhookTx};
use import::ImportWorker;
use search::{SearchService, SearchWorker};
use tokio::net::TcpListener;
use tokio_util::sync::CancellationToken;
use tracing::info;
#[tokio::main]
async fn main() -> anyhow::Result<()> {
export::telemetry::init_tracing();
let cfg = SynchostConfig::load().context("loading configuration")?;
info!("connecting to database / redis / cache");
let db = Arc::new(
export::AppDatabase::connect(cfg.database.clone())
.await
.context("connecting to postgres")?,
);
let redis = export::AppRedis::connect(cfg.redis.clone())
.await
.context("connecting to redis")?;
let cache = Arc::new(CacheClient::new(redis.clone()));
let repo_root = PathBuf::from(cfg.git.repo_root.clone());
info!("dependencies ready");
if let Err(error) = export::telemetry::init_meter_provider() {
tracing::warn!(%error, "failed to init meter provider; metrics disabled");
}
export::metrics::spawn_queue_depth_reporter(
redis.clone(),
&[
("git.sync", cfg.git.sync.queue_topic.clone()),
("git.sync.dlq", cfg.git.sync.dlq_topic.clone()),
],
);
let cancel = CancellationToken::new();
let search = SearchService::from_parts(db.clone(), cfg.search.clone())
.map_err(|e| anyhow::anyhow!("building search service: {e}"))?;
if search.enabled() {
let workers = cfg.search.workers.max(1);
for index in 0..workers {
let worker = SearchWorker::new(
db.clone(),
search.clone(),
PathBuf::from(cfg.git.repo_root.clone()),
cancel.clone(),
);
tokio::spawn(worker.run());
info!(index, "search index worker started");
}
}
if cfg.import.enabled {
let worker = ImportWorker::new(
db.clone(),
redis.clone(),
cache.clone(),
PathBuf::from(cfg.git.repo_root.clone()),
cfg.git.sync.clone(),
cfg.import.clone(),
cancel.clone(),
);
tokio::spawn(worker.run());
info!("import worker started");
}
if cfg.health.enabled() {
let health_addr: SocketAddr =
format!("{}:{}", cfg.health.bind_addr(), cfg.health.synchost_port())
.parse()
.context("parsing health bind address")?;
let health_listener = TcpListener::bind(health_addr)
.await
.with_context(|| format!("binding health {health_addr}"))?;
info!(%health_addr, "synchost health listening");
let health_router = export::health::router_with(Some(db.clone()), redis.clone(), None);
let health_cancel = cancel.clone();
tokio::spawn(async move {
let _ = axum::serve(health_listener, health_router)
.with_graceful_shutdown(async move { health_cancel.cancelled().await })
.await;
});
}
let queue = SyncQueue::new(redis.clone());
let webhook: Arc<dyn WebhookTx> = Arc::new(RepoWebhookTx::new(db.clone()));
let settings = cfg.git.sync.clone();
let worker = SyncWorker::new(
db,
redis,
cache,
repo_root,
queue,
webhook,
settings.clone(),
cancel.clone(),
);
let running = worker
.run()
.await
.map_err(|e| anyhow::anyhow!("starting sync worker: {e}"))?;
info!(
concurrency = settings.concurrency.max(1),
topic = %settings.queue_topic,
"synchost running"
);
export::telemetry::shutdown_signal().await;
info!("synchost shutting down");
cancel.cancel();
running
.shutdown()
.await
.map_err(|e| anyhow::anyhow!("worker shutdown: {e}"))?;
info!("synchost shut down");
Ok(())
}