OXYGENHUB
# a46d15e72101 oxygenhub / crates / hub / src / lib.rs
6.2 KiB · Rust · last changed Jetstream: the swarm's tracker and peer tickets 24 days ago source blame history raw
1
//! OxygenHub: a forge for Oxygen repositories (`docs/spec.md`).
2
//!
3
//! One process: an axum router serving pages, the JSON API and the wire
4
//! protocol; SQLite for hub state; repositories as bare Oxygen stores.
5
6
pub mod api;
7
pub mod apt;
8
pub mod assistants;
9
pub mod auth;
10
pub mod config;
11
pub mod coverage;
12
pub mod db;
13
pub mod events;
14
pub mod flow;
15
pub mod hooks;
16
pub mod images;
17
pub mod junit;
18
pub mod keys;
19
pub mod levels;
20
pub mod live;
21
pub mod mail;
22
pub mod maintenance;
23
pub mod mirror;
24
pub mod notify;
25
pub mod oidc;
26
pub mod protocol;
27
pub mod render;
28
pub mod reports;
29
pub mod repos;
30
pub mod secrets;
31
pub mod signin;
32
pub mod social;
33
pub mod storage;
34
pub mod swarm;
35
pub mod variables;
36
pub mod web;
37
pub mod wiki;
38
39
use crate::config::Config;
40
use crate::repos::Repos;
41
use std::sync::Arc;
42
43
/// Everything a request handler can reach.
44
pub struct App {
45
    pub config: Config,
46
    pub db: db::Pool,
47
    pub repos: Arc<Repos>,
48
    /// Wikis, one Oxygen repository per repository (hub spec §6.11).
49
    pub wikis: Arc<Repos>,
50
    pub server: Arc<oxy_server::Server>,
51
    pub hub_key: keys::HubKey,
52
    pub cookie_key: auth::CookieKey,
53
    /// Seals AI assistants' tokens at rest (hub spec §13.4).
54
    pub assistant_key: assistants::SecretKey,
55
    /// Seals Flow secrets at rest (hub spec §13.2, #107).
56
    pub secret_key: assistants::SecretKey,
57
    /// The OIDC issuer for jobs (hub spec §13.5), made on first use.
58
    pub oidc: tokio::sync::OnceCell<oidc::Issuer>,
59
    /// Which assistants are answering which proposals, and recent calls
60
    /// per repository (hub spec §13.4).
61
    pub assistant_gate: std::sync::Mutex<assistants::Gate>,
62
    pub events: events::Events,
63
    /// Who is resolving which conflict together (hub spec §6.10).
64
    pub rooms: Arc<live::Rooms>,
65
    /// Outgoing mail: hub.toml's `[email]`, overridden by what the admin
66
    /// page saved (#101).
67
    pub mail: std::sync::RwLock<config::Email>,
68
    /// The image swarm's tracker (oxygenhub#144).
69
    pub swarm: swarm::Tracker,
70
    pub started: std::time::Instant,
71
}
72
73
impl App {
74
    /// The rootfs images this hub keeps for runners (oxygenflow#39).
75
    pub fn images(&self) -> images::Store {
76
        images::Store::new(&self.config.data)
77
    }
78
79
    /// The OIDC issuer, its key loaded or made the first time it is asked for.
80
    pub async fn oidc(&self) -> anyhow::Result<&oidc::Issuer> {
81
        let keys = self.config.keys_dir();
82
        self.oidc.get_or_try_init(|| async move { tokio::task::spawn_blocking(move || oidc::Issuer::load_or_create(&keys)).await? }).await
83
    }
84
}
85
86
pub type Shared = Arc<App>;
87
88
/// Open the data directory and build the application state. `init` must
89
/// have run once; `serve` and the tests call this.
90
pub async fn build(config: Config) -> anyhow::Result<Shared> {
91
    std::fs::create_dir_all(config.repos_dir())?;
92
    let db = db::open(&config.db_path()).await?;
93
    let repos = Arc::new(Repos::new(config.repos_dir()));
94
    let wikis = Arc::new(Repos::new(config.data.join("wikis")));
95
    let hub_key = keys::ensure_identity(&config.keys_dir(), &config.site.name, config.base_url())?;
96
    let cookie_key = auth::CookieKey::load_or_create(&config.keys_dir())?;
97
    let assistant_key = assistants::SecretKey::load_or_create(&config.keys_dir())?;
98
    let secret_key = assistants::SecretKey::load_or_create_named(&config.keys_dir(), "secrets.key")?;
99
    let backend = protocol::HubBackend::new(
100
        repos.clone(),
101
        wikis.clone(),
102
        db.clone(),
103
        config.auth.push_tokens.clone(),
104
        tokio::runtime::Handle::current(),
105
    );
106
    let slot = backend.app_slot();
107
    let server = oxy_server::Server::new(backend);
108
    let mail = mail::load(&db, &config.email).await;
109
    let app = Arc::new(App {
110
        mail: std::sync::RwLock::new(mail),
111
        config,
112
        db,
113
        repos,
114
        wikis,
115
        server,
116
        hub_key,
117
        cookie_key,
118
        assistant_key,
119
        secret_key,
120
        oidc: tokio::sync::OnceCell::new(),
121
        assistant_gate: std::sync::Mutex::new(assistants::Gate::default()),
122
        events: events::Events::default(),
123
        rooms: Arc::new(live::Rooms::default()),
124
        swarm: Default::default(),
125
        started: std::time::Instant::now(),
126
    });
127
    *slot.lock().unwrap() = Some(Arc::downgrade(&app));
128
    storage::load(&app).await?;
129
    coverage::load(&app).await?;
130
    junit::load(&app).await?;
131
    reports::load(&app).await?;
132
    tokio::spawn(hooks::worker(app.clone()));
133
    tokio::spawn(flow::reaper(app.clone()));
134
    tokio::spawn(flow::scheduler(app.clone()));
135
    tokio::spawn(flow::placement(app.clone()));
136
    tokio::spawn(maintenance::nightly(app.clone()));
137
    Ok(app)
138
}
139
140
/// The full router: pages, API, protocol and static assets.
141
pub fn router(app: Shared) -> axum::Router {
142
    web::router(app)
143
}
144
145
/// Serve on the configured addresses until the process is told to stop.
146
pub async fn serve(app: Shared) -> anyhow::Result<()> {
147
    let router = router(app.clone());
148
    let http = app.config.bind_addr()?;
149
    let https = app.config.tls_bind_addr()?;
150
    if http.is_none() && https.is_none() {
151
        anyhow::bail!("nothing to bind: set server.bind or server.tls_bind");
152
    }
153
    let mut tasks = tokio::task::JoinSet::new();
154
    if let Some(addr) = http {
155
        let listener = tokio::net::TcpListener::bind(addr).await?;
156
        tracing::info!("http on {addr}");
157
        let r = router.clone();
158
        tasks.spawn(async move {
159
            axum::serve(listener, r.into_make_service_with_connect_info::<std::net::SocketAddr>()).await.map_err(anyhow::Error::from)
160
        });
161
    }
162
    if let Some(addr) = https {
163
        let _ = rustls::crypto::ring::default_provider().install_default();
164
        let tls = axum_server::tls_rustls::RustlsConfig::from_pem_file(app.config.tls_cert(), app.config.tls_key())
165
            .await
166
            .map_err(|e| anyhow::anyhow!("loading TLS files: {e}"))?;
167
        tracing::info!("https on {addr}");
168
        let r = router.clone();
169
        tasks.spawn(async move {
170
            axum_server::bind_rustls(addr, tls)
171
                .serve(r.into_make_service_with_connect_info::<std::net::SocketAddr>())
172
                .await
173
                .map_err(anyhow::Error::from)
174
        });
175
    }
176
    tokio::select! {
177
        r = tasks.join_next() => { r.transpose()?.transpose()?; Ok(()) }
178
        _ = tokio::signal::ctrl_c() => { tracing::info!("stopping"); Ok(()) }
179
    }
180
}