webhook_bridge/src/main.rs

#![forbid(unsafe_code)]
use std::time::Duration;

use axum::http::StatusCode;
use axum::middleware;
use axum::routing::get;
use axum::routing::post;
use axum::Json;
use axum::Router;
use kube::Client;
use serde::Serialize;
use tokio::signal;
use tower_http::timeout::TimeoutLayer;
use tower_http::trace::TraceLayer;
use tracing_subscriber::layer::SubscriberExt;
use tracing_subscriber::util::SubscriberInitExt;

use self::gitea_client::GiteaClient;
use self::hook_push::HookPush;
use self::webhook::handle_push;
use self::webhook::hook;
use self::webhook::verify_signature;

mod crd_pipeline_run;
mod discovery;
mod gitea_client;
mod hook_push;
mod kubernetes;
mod remote_config;
mod webhook;

const EXAMPLE_WEBHOOK_PAYLOAD: &str = include_str!("../example_tag_webhook_payload.json");

#[tokio::main]
#[allow(clippy::needless_return)]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    tracing_subscriber::registry()
        .with(
            tracing_subscriber::EnvFilter::try_from_default_env().unwrap_or_else(|_| {
                "webhook_bridge=info,tower_http=debug,axum::rejection=trace".into()
            }),
        )
        .with(tracing_subscriber::fmt::layer())
        .init();

    launch_server().await
}

async fn launch_server() -> Result<(), Box<dyn std::error::Error>> {
    let kubernetes_client: Client = Client::try_default()
        .await
        .expect("Set KUBECONFIG to a valid kubernetes config.");

    let gitea_api_root = std::env::var("WEBHOOK_BRIDGE_API_ROOT")?;
    let gitea_api_token = std::env::var("WEBHOOK_BRIDGE_OAUTH_TOKEN")?;
    let gitea = GiteaClient::new(gitea_api_root, gitea_api_token);

    let app = Router::new()
        .route("/hook", post(hook))
        .layer(middleware::from_fn(verify_signature))
        .route("/health", get(health))
        .layer((
            TraceLayer::new_for_http(),
            // Add a timeout layer so graceful shutdown can't wait forever.
            TimeoutLayer::new(Duration::from_secs(600)),
        ))
        .with_state(AppState {
            kubernetes_client,
            gitea,
        });

    let listener = tokio::net::TcpListener::bind("0.0.0.0:9988").await?;
    tracing::info!("listening on {}", listener.local_addr().unwrap());
    axum::serve(listener, app)
        .with_graceful_shutdown(shutdown_signal())
        .await?;
    Ok(())
}

async fn local_trigger() -> Result<(), Box<dyn std::error::Error>> {
    let kubernetes_client: Client = Client::try_default()
        .await
        .expect("Set KUBECONFIG to a valid kubernetes config.");

    let gitea_api_root = std::env::var("WEBHOOK_BRIDGE_API_ROOT")?;
    let gitea_api_token = std::env::var("WEBHOOK_BRIDGE_OAUTH_TOKEN")?;
    let gitea = GiteaClient::new(gitea_api_root, gitea_api_token);

    let webhook_payload: HookPush = serde_json::from_str(EXAMPLE_WEBHOOK_PAYLOAD)?;

    handle_push(gitea, kubernetes_client, webhook_payload).await?;

    Ok(())
}

#[derive(Clone)]
struct AppState {
    kubernetes_client: Client,
    gitea: GiteaClient,
}

async fn shutdown_signal() {
    let ctrl_c = async {
        signal::ctrl_c()
            .await
            .expect("failed to install Ctrl+C handler");
    };

    #[cfg(unix)]
    let terminate = async {
        signal::unix::signal(signal::unix::SignalKind::terminate())
            .expect("failed to install signal handler")
            .recv()
            .await;
    };

    #[cfg(not(unix))]
    let terminate = std::future::pending::<()>();

    tokio::select! {
        _ = ctrl_c => {},
        _ = terminate => {},
    }
}

async fn health() -> (StatusCode, Json<HealthResponse>) {
    (StatusCode::OK, Json(HealthResponse { ok: true }))
}

#[derive(Serialize)]
struct HealthResponse {
    ok: bool,
}