AkurAI-Build / src/server/webhooks_api.rs

Latest change b9bf8c65e7cff19e76ae78f73b05d66f6a1b4729 - OLI-274: commits and pull requests link to the Paperclip issues they mention by Óli

//! Outbound webhook routes (OLI-272) and the push event the post-receive hook posts.

use axum::{
    Json,
    extract::{Path, Query, State},
    http::{HeaderMap, StatusCode},
};
use serde::Deserialize;

use super::access_api::{Identity, identify, role_on};
use super::{ApiError, AppState, Envelope, authorize, ok, unix_time};
use crate::access::Role;
use crate::webhooks::{self, Delivery, NewWebhook, Webhook};

fn require_admin(headers: &HeaderMap, state: &AppState, repository: &str) -> Result<(), ApiError> {
    let identity = identify(headers, state)?;
    match role_on(&identity, state, repository)? {
        Some(role) if role.allows(Role::Admin) => Ok(()),
        _ => Err(ApiError::forbidden(format!(
            "admin access to {repository} is required"
        ))),
    }
}

#[derive(Deserialize)]
pub(super) struct WebhookRequest {
    url: String,
    secret: String,
    events: Vec<String>,
}

#[derive(Deserialize)]
pub(super) struct ActiveRequest {
    active: bool,
}

#[derive(Deserialize)]
pub(super) struct DeliveriesQuery {
    #[serde(default)]
    limit: Option<usize>,
}

#[derive(Deserialize)]
pub(super) struct PushEvent {
    reference: String,
    before: String,
    after: String,
    #[serde(default)]
    pusher: String,
}

pub(super) async fn list_webhooks(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(repository): Path<String>,
) -> Result<Json<Envelope<Vec<Webhook>>>, ApiError> {
    require_admin(&headers, &state, &repository)?;
    let database = state.database.clone();
    let list = tokio::task::spawn_blocking(move || webhooks::list(&database, &repository))
        .await
        .map_err(ApiError::internal)?
        .map_err(ApiError::bad)?;
    Ok(ok(list))
}

pub(super) async fn create_webhook(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(repository): Path<String>,
    Json(request): Json<WebhookRequest>,
) -> Result<(StatusCode, Json<Envelope<Webhook>>), ApiError> {
    require_admin(&headers, &state, &repository)?;
    let database = state.database.clone();
    let created = tokio::task::spawn_blocking(move || {
        webhooks::create(
            &database,
            &repository,
            &NewWebhook {
                url: request.url.trim(),
                secret: &request.secret,
                events: &request.events,
            },
        )
    })
    .await
    .map_err(ApiError::internal)?
    .map_err(ApiError::bad)?;
    Ok((StatusCode::CREATED, ok(created)))
}

pub(super) async fn set_webhook_active(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path((repository, id)): Path<(String, i64)>,
    Json(request): Json<ActiveRequest>,
) -> Result<Json<Envelope<bool>>, ApiError> {
    require_admin(&headers, &state, &repository)?;
    let database = state.database.clone();
    let changed = tokio::task::spawn_blocking(move || {
        webhooks::set_active(&database, &repository, id, request.active)
    })
    .await
    .map_err(ApiError::internal)?
    .map_err(ApiError::bad)?;
    Ok(ok(changed))
}

pub(super) async fn delete_webhook(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path((repository, id)): Path<(String, i64)>,
) -> Result<Json<Envelope<bool>>, ApiError> {
    require_admin(&headers, &state, &repository)?;
    let database = state.database.clone();
    let removed = tokio::task::spawn_blocking(move || webhooks::remove(&database, &repository, id))
        .await
        .map_err(ApiError::internal)?
        .map_err(ApiError::bad)?;
    Ok(ok(removed))
}

pub(super) async fn list_deliveries(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path((repository, id)): Path<(String, i64)>,
    Query(query): Query<DeliveriesQuery>,
) -> Result<Json<Envelope<Vec<Delivery>>>, ApiError> {
    require_admin(&headers, &state, &repository)?;
    let database = state.database.clone();
    let limit = query.limit.unwrap_or(50).clamp(1, 500);
    let list = tokio::task::spawn_blocking(move || {
        webhooks::deliveries(&database, &repository, id, limit)
    })
    .await
    .map_err(ApiError::internal)?
    .map_err(ApiError::bad)?;
    Ok(ok(list))
}

pub(super) async fn redeliver(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path((repository, id, delivery)): Path<(String, i64, i64)>,
) -> Result<Json<Envelope<bool>>, ApiError> {
    require_admin(&headers, &state, &repository)?;
    let database = state.database.clone();
    let now = unix_time()? as i64;
    let queued = tokio::task::spawn_blocking(move || {
        webhooks::redeliver(&database, &repository, id, delivery, now)
    })
    .await
    .map_err(ApiError::internal)?
    .map_err(ApiError::bad)?;
    if !queued {
        return Err(ApiError::not_found("delivery not found"));
    }
    Ok(ok(true))
}

/// `POST /api/repos/{repository}/events/push`: the post-receive hook reporting a
/// ref update (controller token only; the hook has it from the forge).
pub(super) async fn push_event(
    State(state): State<AppState>,
    headers: HeaderMap,
    Path(repository): Path<String>,
    Json(event): Json<PushEvent>,
) -> Result<Json<Envelope<usize>>, ApiError> {
    authorize(&headers, &state)?;
    if !matches!(identify(&headers, &state)?, Identity::Master) {
        return Err(ApiError::forbidden("only the forge reports pushes"));
    }
    let database = state.database.clone();
    let now = unix_time()? as i64;
    let payload = serde_json::json!({
        "ref": event.reference,
        "before": event.before,
        "after": event.after,
        "pusher": event.pusher,
        "deleted": event.after.chars().all(|c| c == '0'),
        "created": event.before.chars().all(|c| c == '0'),
    });
    let repository_for_links = repository.clone();
    let queued = tokio::task::spawn_blocking(move || {
        webhooks::emit(&database, &repository, "push", &payload, now)
    })
    .await
    .map_err(ApiError::internal)?
    .map_err(ApiError::bad)?;
    // Issue links (OLI-274): announce the pushed commits on the issues they mention.
    let hosted = state
        .hosted_root
        .join("hosted")
        .join(format!("{repository_for_links}.git"));
    if hosted.is_dir() {
        let database = state.database.clone();
        let tracker = state.tracker.clone();
        let origin = state.public_origin.to_string();
        let (before, after, reference, pusher) =
            (event.before, event.after, event.reference, event.pusher);
        let linked = tokio::task::spawn_blocking(move || -> anyhow::Result<usize> {
            let commits = crate::linking::pushed_commits(&hosted, &before, &after)?;
            crate::linking::announce_push(
                &database,
                tracker.as_ref(),
                &repository_for_links,
                &commits,
                &reference,
                &pusher,
                &origin,
            )
        })
        .await;
        match linked {
            Ok(Ok(n)) if n > 0 => tracing::info!(posted = n, "issue links announced"),
            Ok(Ok(_)) => {}
            Ok(Err(error)) => tracing::warn!(%error, "issue linking failed"),
            Err(error) => tracing::warn!(%error, "issue linking task failed"),
        }
    }
    Ok(ok(queued.len()))
}

/// Announce a pull request action on the issues it mentions; never fails the caller.
pub(crate) fn link_pull(
    state: &AppState,
    repository: &str,
    number: i64,
    action: &str,
    title: &str,
    body: &str,
) {
    let database = state.database.clone();
    let tracker = state.tracker.clone();
    let origin = state.public_origin.to_string();
    let (repository, action, title, body) = (
        repository.to_string(),
        action.to_string(),
        title.to_string(),
        body.to_string(),
    );
    tokio::task::spawn_blocking(move || {
        if let Err(error) = crate::linking::announce_pull(
            &database,
            tracker.as_ref(),
            &repository,
            number,
            &action,
            &title,
            &body,
            &origin,
        ) {
            tracing::warn!(%error, "issue linking for pull request failed");
        }
    });
}

/// Fire-and-forget emit for handlers that must not fail on webhook trouble.
pub(crate) fn emit_quietly(
    database: &crate::db::Database,
    repository: &str,
    event: &str,
    payload: serde_json::Value,
) {
    let now = unix_time().map(|t| t as i64).unwrap_or(0);
    if let Err(error) = webhooks::emit(database, repository, event, &payload, now) {
        tracing::warn!(%repository, event, %error, "webhook emit failed");
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use axum::{
        Router,
        body::{Body, to_bytes},
        http::{Request, header},
        routing::{delete, get, post},
    };
    use std::process::Command;
    use tower::ServiceExt;

    fn app() -> (Router, AppState) {
        app_with_hosted().1
    }

    /// Also returns the hosted demo.git (one commit mentioning OLI-5) and the recorder.
    fn app_with_hosted() -> (
        (
            tempfile::TempDir,
            std::sync::Arc<crate::linking::testing::RecordingTracker>,
            String,
        ),
        (Router, AppState),
    ) {
        let root = tempfile::tempdir().expect("root");
        let work = root.path().join("work");
        std::fs::create_dir_all(&work).expect("work");
        let git = |args: &[&str]| {
            let status = Command::new("git")
                .arg("-C")
                .arg(&work)
                .args(args)
                .status()
                .expect("git");
            assert!(status.success(), "git {args:?}");
        };
        git(&["init", "-q", "-b", "main"]);
        git(&["config", "user.name", "Ada"]);
        git(&["config", "user.email", "ada@example.invalid"]);
        git(&[
            "commit",
            "-q",
            "--allow-empty",
            "-m",
            "OLI-5: hook up linking",
        ]);
        let sha = crate::git_process::output(&work, ["rev-parse", "HEAD"])
            .expect("sha")
            .trim()
            .to_string();
        let hosted = root.path().join("hosted");
        std::fs::create_dir_all(&hosted).expect("hosted");
        git(&[
            "clone",
            "-q",
            "--bare",
            ".",
            hosted.join("demo.git").to_str().expect("utf8"),
        ]);
        let mut state = crate::server::tests::test_state("api-token-123", "webhook-secret");
        state.hosted_root = root.path().to_owned();
        let recorder = std::sync::Arc::new(crate::linking::testing::RecordingTracker::default());
        state.tracker = recorder.clone();
        state
            .database
            .add_repository("demo", "https://example.com/demo.git", "main")
            .expect("repo");
        let router = Router::new()
            .route(
                "/api/repos/{repository}/webhooks",
                get(list_webhooks).post(create_webhook),
            )
            .route(
                "/api/repos/{repository}/webhooks/{id}",
                delete(delete_webhook).patch(set_webhook_active),
            )
            .route(
                "/api/repos/{repository}/webhooks/{id}/deliveries",
                get(list_deliveries),
            )
            .route(
                "/api/repos/{repository}/webhooks/{id}/deliveries/{delivery}/redeliver",
                post(redeliver),
            )
            .route("/api/repos/{repository}/events/push", post(push_event))
            .with_state(state.clone());
        ((root, recorder, sha), (router, state))
    }

    #[tokio::test]
    async fn a_push_event_links_the_pushed_commits_to_the_issues_they_mention() {
        let ((_root, recorder, sha), (app, _state)) = app_with_hosted();
        let (status, _) = send(
            &app,
            "POST",
            "/api/repos/demo/events/push",
            Some("api-token-123"),
            serde_json::json!({"reference": "refs/heads/main", "before": "0".repeat(40), "after": sha, "pusher": "ada@example.is"}),
        )
        .await;
        assert_eq!(status, StatusCode::OK);
        let comments = recorder.comments.lock().expect("lock").clone();
        assert_eq!(comments.len(), 1, "{comments:?}");
        assert_eq!(comments[0].0, "OLI-5");
        assert!(
            comments[0].1.contains(&sha[..12])
                && comments[0].1.contains("ada@example.is")
                && comments[0]
                    .1
                    .contains("https://build.example/repos/demo/commit/")
        );
        send(&app, "POST", "/api/repos/demo/events/push", Some("api-token-123"), serde_json::json!({"reference": "refs/heads/main", "before": "0".repeat(40), "after": sha})).await;
        assert_eq!(
            recorder.comments.lock().expect("lock").len(),
            1,
            "the same push announces nothing new"
        );
    }

    async fn send(
        app: &Router,
        method: &str,
        uri: &str,
        token: Option<&str>,
        body: serde_json::Value,
    ) -> (StatusCode, serde_json::Value) {
        let mut request = Request::builder()
            .method(method)
            .uri(uri)
            .header(header::CONTENT_TYPE, "application/json");
        if let Some(token) = token {
            request = request.header(header::AUTHORIZATION, format!("Bearer {token}"));
        }
        let response = app
            .clone()
            .oneshot(request.body(Body::from(body.to_string())).expect("request"))
            .await
            .expect("response");
        let status = response.status();
        let bytes = to_bytes(response.into_body(), usize::MAX)
            .await
            .expect("body");
        (
            status,
            serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null),
        )
    }

    #[tokio::test]
    async fn webhooks_are_managed_over_http_and_a_push_event_queues_a_signed_delivery() {
        let (app, state) = app();
        let master = Some("api-token-123");
        let (status, _) = send(
            &app,
            "GET",
            "/api/repos/demo/webhooks",
            None,
            serde_json::json!(null),
        )
        .await;
        assert_eq!(status, StatusCode::UNAUTHORIZED);
        let (status, created) = send(
            &app,
            "POST",
            "/api/repos/demo/webhooks",
            master,
            serde_json::json!({"url": "https://hooks.example.com/in", "secret": "s3cret-s3cret", "events": ["push", "run"]}),
        )
        .await;
        assert_eq!(status, StatusCode::CREATED, "{created}");
        let id = created["data"]["id"].as_i64().expect("id");
        assert!(
            created["data"].get("secret").is_none(),
            "secrets never leave the forge"
        );
        let (status, _) = send(&app, "POST", "/api/repos/demo/webhooks", master, serde_json::json!({"url": "http://evil/x", "secret": "s3cret-s3cret", "events": ["push"]})).await;
        assert_eq!(status, StatusCode::BAD_REQUEST);

        let (status, queued) = send(
            &app,
            "POST",
            "/api/repos/demo/events/push",
            master,
            serde_json::json!({"reference": "refs/heads/main", "before": "a".repeat(40), "after": "b".repeat(40), "pusher": "ada@example.is"}),
        )
        .await;
        assert_eq!(status, StatusCode::OK, "{queued}");
        assert_eq!(queued["data"], 1);
        let (_, deliveries) = send(
            &app,
            "GET",
            &format!("/api/repos/demo/webhooks/{id}/deliveries"),
            master,
            serde_json::json!(null),
        )
        .await;
        let delivery = &deliveries["data"][0];
        assert_eq!(delivery["status"], "pending");
        let payload: serde_json::Value =
            serde_json::from_str(delivery["payload"].as_str().expect("payload")).expect("json");
        assert_eq!(payload["data"]["pusher"], "ada@example.is");
        assert_eq!(payload["data"]["created"], false);

        let delivery_id = delivery["id"].as_i64().expect("id");
        let (status, _) = send(
            &app,
            "POST",
            &format!("/api/repos/demo/webhooks/{id}/deliveries/{delivery_id}/redeliver"),
            master,
            serde_json::json!(null),
        )
        .await;
        assert_eq!(status, StatusCode::OK);
        let (status, _) = send(
            &app,
            "POST",
            &format!("/api/repos/demo/webhooks/{id}/deliveries/999/redeliver"),
            master,
            serde_json::json!(null),
        )
        .await;
        assert_eq!(status, StatusCode::NOT_FOUND);

        let (_, off) = send(
            &app,
            "PATCH",
            &format!("/api/repos/demo/webhooks/{id}"),
            master,
            serde_json::json!({"active": false}),
        )
        .await;
        assert_eq!(off["data"], true);
        let (_, queued) = send(&app, "POST", "/api/repos/demo/events/push", master, serde_json::json!({"reference": "refs/heads/main", "before": "a".repeat(40), "after": "c".repeat(40)})).await;
        assert_eq!(queued["data"], 0, "inactive hooks get nothing");
        let (_, removed) = send(
            &app,
            "DELETE",
            &format!("/api/repos/demo/webhooks/{id}"),
            master,
            serde_json::json!(null),
        )
        .await;
        assert_eq!(removed["data"], true);
        let _ = state;
    }
}