portail 0.1.0

Unified proxy/gateway: AI Gateway + MCP Gateway + CDN cache
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

use clap::Parser;
use mimalloc::MiMalloc;
use portail::cdn;
use portail::config;
use portail::config::Config;
use portail::events::EventLog;
use portail::AppState;
use portail::mcp;
use std::sync::{Arc, RwLock};
use tracing_subscriber::EnvFilter;

#[global_allocator]
static GLOBAL: MiMalloc = MiMalloc;

const MAX_EVENTS: usize = 2000;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(EnvFilter::from_default_env())
        .json()
        .init();

    let handle = metrics_exporter_prometheus::PrometheusBuilder::new()
        .install_recorder()
        .expect("failed to install prometheus recorder");

    let cli = config::Cli::parse();
    let config = Config::load(&cli)?;
    let listen = config.listen.clone();
    tracing::info!(%listen, "portail starting");

    let event_log = Arc::new(EventLog::new(MAX_EVENTS));

    let cdn_cache = if let Some(cdn_cfg) = config.cdn.as_ref().filter(|c| c.enabled) {
        let cache = cdn::CacheManager::new(cdn_cfg);
        tokio::spawn({
            let c = Arc::clone(&cache);
            async move { cdn::stats_logger(c).await }
        });
        if let Some(ref nats_url) = cdn_cfg.nats_url {
            let nc = async_nats::connect(nats_url).await
                .expect("NATS connection failed");
            let sub = nc.subscribe("index.invalidated.>").await
                .expect("NATS subscribe failed");
            let c = Arc::clone(&cache);
            tokio::spawn(async move { cdn::purge_loop(sub, c).await });
            tracing::info!("NATS invalidation connected");
        }
        Some(cache)
    } else {
        None
    };

    if let Some(mcp_cfg) = config.mcp.as_ref().filter(|c| c.enabled) {
        mcp::start_sidecar(&mcp_cfg.socket_path).await?;
    }

    let state = Arc::new(AppState {
        config: RwLock::new(config),
        event_log: Arc::clone(&event_log),
        cdn_cache: cdn_cache.clone(),
        hooks: Arc::new(portail::hooks::HookStore::new()),
        metrics_handle: handle,
    });

    // ── Sentinel: background watcher ──────────────────────────────────
    tokio::spawn({
        let log = Arc::clone(&event_log);
        let cache = cdn_cache.clone();
        async move { portail::sentinel::run_sentinel(log, cache).await }
    });

    let app = portail::proxy::build_router(Arc::clone(&state));

    let sighup_state = Arc::clone(&state);
    let sighup_cli = cli.clone();
    tokio::spawn(async move {
        use tokio::signal::unix::{signal, SignalKind};
        let mut sig = signal(SignalKind::hangup())
            .expect("failed to install SIGHUP handler");
        loop {
            sig.recv().await;
            match Config::load(&sighup_cli) {
                Ok(new) => {
                    *sighup_state.config.write().unwrap() = new;
                    tracing::info!("config reloaded on SIGHUP");
                }
                Err(e) => tracing::error!(error = %e, "config reload failed"),
            }
        }
    });

    let listener = tokio::net::TcpListener::bind(&listen).await?;
    tracing::info!("listening on {}", listener.local_addr()?);

    axum::serve(listener, app)
        .with_graceful_shutdown(shutdown_signal())
        .await?;

    Ok(())
}

async fn shutdown_signal() {
    use tokio::signal::unix::{signal, SignalKind};
    let mut sigterm = signal(SignalKind::terminate())
        .expect("failed to install SIGTERM handler");
    tokio::select! {
        _ = tokio::signal::ctrl_c() => tracing::info!("SIGINT received"),
        _ = sigterm.recv() => tracing::info!("SIGTERM received"),
    }
}