webhook_bridge/src/main.rs

#![forbid(unsafe_code)]
use std::borrow::Borrow;
use std::sync::Arc;
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::api::PostParams;
use kube::Api;
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::crd_pipeline_run::PipelineRun;
use self::discovery::discover_matching_push_triggers;
use self::discovery::discover_webhook_bridge_config;
use self::gitea_client::GiteaClient;
use self::hook_push::HookPush;
use self::hook_push::PipelineParamters;
use self::kubernetes::run_pipelines;
use self::remote_config::RemoteConfig;
use self::webhook::hook;
use self::webhook::verify_signature;
use kube::CustomResourceExt;

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

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

#[tokio::main]
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();

    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)?;
    let repo_tree = gitea
        .get_tree(
            "talexander",
            "webhook_bridge",
            webhook_payload.get_pull_base_sha()?,
        )
        .await?;
    let remote_config = discover_webhook_bridge_config(&gitea, &repo_tree).await?;
    let pipelines = discover_matching_push_triggers(
        &gitea,
        &repo_tree,
        &webhook_payload.ref_field,
        &remote_config,
    )
    .await?;

    run_pipelines(webhook_payload, pipelines, kubernetes_client).await?;

    // 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(())
}

#[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,
}