Milestone 0.5: API, webhooks, checks b64e2537d702

//! Emit TypeScript declarations for every API type, from their JSON

//! Schemas, so the VS Code extension can never drift from the server:

//!

//! `cargo run -p hub-api --bin hub-api-typescript > ts/index.d.ts`

use serde_json::Value;

fn ts_type(schema: &Value) -> String {

    if let Some(r) = schema.get("$ref").and_then(Value::as_str) {

        return r.rsplit('/').next().unwrap_or("unknown").to_string();

    }

    if let Some(e) = schema.get("enum").and_then(Value::as_array) {

        return e.iter().map(|v| serde_json::to_string(v).unwrap_or_default()).collect::<Vec<_>>().join(" | ");

    }

    if let Some(c) = schema.get("const") {

        return serde_json::to_string(c).unwrap_or_default();

    }

    if let Some(variants) = schema.get("oneOf").or_else(|| schema.get("anyOf")).and_then(Value::as_array) {

        return variants.iter().map(|v| ts_type(v)).collect::<Vec<_>>().join(" | ");

    }

    let nullable = schema.get("type").and_then(Value::as_array).is_some_and(|a| a.iter().any(|t| t == "null"));

    let ty = match schema.get("type") {

        Some(Value::String(t)) => t.clone(),

        Some(Value::Array(a)) => a.iter().filter_map(Value::as_str).find(|t| *t != "null").unwrap_or("unknown").to_string(),

        _ => {

            if schema.get("properties").is_some() {

                "object".into()

            } else {

                "unknown".into()

            }

        }

    };

    let base = match ty.as_str() {

        "string" => "string".to_string(),

        "integer" | "number" => "number".to_string(),

        "boolean" => "boolean".to_string(),

        "array" => format!("{}[]", schema.get("items").map(|i| ts_type(i)).unwrap_or_else(|| "unknown".into())),

        "object" => {

            let props = schema.get("properties").and_then(Value::as_object);

            let required: Vec<&str> =

                schema.get("required").and_then(Value::as_array).map(|a| a.iter().filter_map(Value::as_str).collect()).unwrap_or_default();

            match props {

                Some(p) => {

                    let fields: Vec<String> = p

                        .iter()

                        .map(|(k, v)| format!("  {k}{}: {};", if required.contains(&k.as_str()) { "" } else { "?" }, ts_type(v)))

                        .collect();

                    format!("{{\n{}\n}}", fields.join("\n"))

                }

                None => "Record<string, unknown>".into(),

            }

        }

        _ => "unknown".into(),

    };

    if nullable {

        format!("{base} | null")

    } else {

        base

    }

}

fn main() {

    println!("// Generated by `cargo run -p hub-api --bin hub-api-typescript`; do not edit.");

    println!("// Types of the Oxygenhub JSON API (hub spec §12.1).\n");

    let mut printed = std::collections::BTreeSet::new();

    for (name, schema) in hub_api::schemas() {

        let v = serde_json::to_value(&schema).unwrap();

        let defs = v.get("$defs").and_then(Value::as_object).cloned().unwrap_or_default();

        for (dname, dschema) in &defs {

            if printed.insert(dname.clone()) {

                println!("export type {dname} = {};\n", ts_type(dschema));

            }

        }

        if printed.insert(name.to_string()) {

            println!("export type {name} = {};\n", ts_type(&v));

        }

    }

}

// Generated by `cargo run -p hub-api --bin hub-api-typescript`; do not edit.

// Types of the Oxygenhub JSON API (hub spec §12.1).

export type Visibility = "public" | "internal" | "private";

export type Repository = {

  clone_url: string;

  created_ms: number;

  default_bookmark: string;

  description: string;

  name: string;

  owner: string;

  updated_ms: number;

  visibility: Visibility;

};

export type Bookmark = {

  changes: number;

  default: boolean;

  heads: string[];

  name: string;

};

export type Tag = {

  heads: string[];

  message: string;

  name: string;

  signer?: string | null;

};

export type Author = {

  account?: string | null;

  email: string;

  key: string;

  name: string;

};

export type Signature = {

  status: "unsigned";

} | {

  key: string;

  revoked: boolean;

  sealed: boolean;

  status: "valid";

} | {

  reason: string;

  status: "invalid";

};

export type ChangeSummary = {

  author: Author;

  deps: string[];

  hash: string;

  head: boolean;

  message: string;

  signature: Signature;

  when_ms: number;

};

export type Problem = {

  detail: string;

  status: number;

  title: string;

};

export type Approval = {

  account: string;

  key?: string | null;

  sealed: boolean;

};

export type ProposalSource = {

  bookmark: string;

  kind: "bookmark";

} | {

  bookmark: string;

  kind: "fork";

  name: string;

  owner: string;

} | {

  hashes: string[];

  kind: "changes";

};

export type ProposalState = "draft" | "open" | "landed" | "closed";

export type Proposal = {

  approvals: Approval[];

  author: string;

  blockers: string[];

  body: string;

  changes: ChangeSummary[];

  conflicted: boolean;

  created_ms: number;

  landable: boolean;

  number: number;

  projected_heads: string[];

  projected_state?: string | null;

  repository_url: string;

  source: ProposalSource;

  source_url?: string | null;

  state: ProposalState;

  target: string;

  title: string;

  updated_ms: number;

  url: string;

};

export type Label = {

  colour: string;

  description: string;

  name: string;

};

export type Issue = {

  assignees: string[];

  author: string;

  body: string;

  closed_ms?: number | null;

  comments: number;

  created_ms: number;

  labels: Label[];

  locked: boolean;

  milestone?: string | null;

  number: number;

  pinned: boolean;

  reason: string;

  state: string;

  title: string;

  updated_ms: number;

  url: string;

};

export type Milestone = {

  closed: number;

  description: string;

  due_ms?: number | null;

  open: number;

  state: string;

  title: string;

};

export type Atom = {

  change: string;

  index: number;

};

export type Comment = {

  anchor?: Atom | unknown;

  author: string;

  body: string;

  created_ms: number;

  id: number;

  path?: string | null;

  resolved: boolean;

};

export type Review = {

  author: string;

  body: string;

  created_ms: number;

  id: number;

  sealed: boolean;

  verdict: string;

};

export type TreeEntry = {

  conflicted: boolean;

  exec: boolean;

  kind: string;

  name: string;

  path: string;

  size?: number | null;

};

export type BlameLine = {

  author: string;

  change: string;

  index: number;

  line: number;

  text: string;

  when_ms: number;

};

export type FileChange = {

  added: number;

  from?: string | null;

  kind: string;

  path: string;

  removed: number;

};

export type ChangeDetail = {

  body: string;

  change: ChangeSummary;

  dependents: string[];

  files: FileChange[];

  in_bookmarks: string[];

  in_tags: string[];

};

export type History = {

  changes: ChangeSummary[];

  heads: string[];

  offset: number;

  reference: string;

  total: number;

};

export type Conflict = {

  bookmark: string;

  contributors: string[];

  kind: string;

  path: string;

};

export type Push = {

  at_ms: number;

  changes: number;

  heads: number;

  new_state: string;

  ok: boolean;

  old_state: string;

  pusher: string;

  reason: string;

  reference: string;

};

export type Policy = {

  expires_ms: number;

  root_threshold: number;

  roots: string[];

  rules: Rule[];

  serial: number;

};

export type Rule = {

  allow_conflicts: boolean;

  keys: string[];

  pattern: string;

  require: string;

  threshold: number;

};

export type Trust = {

  hub_key: string;

  invalid: string[];

  policy?: Policy | unknown;

  sealed: number;

  signed: number;

  unsigned: number;

};

export type Hook = {

  active: boolean;

  created_ms: number;

  events: string;

  id: number;

  url: string;

};

export type NewHook = {

  events?: string;

  secret: string;

  url: string;

};

export type CheckStatus = "queued" | "running" | "passed" | "failed" | "cancelled" | "skipped";

export type Check = {

  finished_ms?: number | null;

  name: string;

  reporter: string;

  seal?: string | null;

  started_ms: number;

  state: string;

  status: CheckStatus;

  summary: string;

  url: string;

};

export type NewCheck = {

  name: string;

  seal?: string | null;

  state: string;

  status: CheckStatus;

  summary?: string;

  url?: string;

};

export type User = {

  created_ms: number;

  display_name: string;

  is_admin: boolean;

  keys: string[];

  username: string;

};

export type Organisation = {

  description: string;

  display_name: string;

  members: string[];

  name: string;

};

export type NewRepository = {

  description?: string;

  name: string;

  owner: string;

  visibility?: Visibility;

};

export type NewIssue = {

  body?: string;

  title: string;

};

export type NewComment = {

  body: string;

};

export type NewProposal = {

  body?: string;

  draft?: boolean;

  source: string;

  target: string;

  title: string;

};

export type NewReview = {

  body?: string;

  verdict: string;

};

export type IssueStateChange = {

  reason?: string;

  state: string;

};

export type WebhookEnvelope = {

  actor: string;

  at_ms: number;

  data: unknown;

  delivery: string;

  event: string;

  repository?: Repository | unknown;

};

export type ProposalEvent = {

  check?: Check | unknown;

  kind: string;

  number: number;

};

-- Webhooks with a delivery log, and checks on state hashes (hub spec

-- §12.2, §13.1).

CREATE TABLE hooks (

    id         INTEGER PRIMARY KEY,

    repo_id    INTEGER REFERENCES repositories(id) ON DELETE CASCADE,

    org_id     INTEGER REFERENCES organisations(id) ON DELETE CASCADE,

    url        TEXT NOT NULL,

    secret     TEXT NOT NULL DEFAULT '',

    events     TEXT NOT NULL DEFAULT '*',

    active     INTEGER NOT NULL DEFAULT 1,

    created_ms INTEGER NOT NULL,

    CHECK ((repo_id IS NULL) <> (org_id IS NULL))

);

CREATE TABLE deliveries (

    id          INTEGER PRIMARY KEY,

    hook_id     INTEGER NOT NULL REFERENCES hooks(id) ON DELETE CASCADE,

    delivery    TEXT NOT NULL,

    event       TEXT NOT NULL,

    payload     TEXT NOT NULL,

    created_ms  INTEGER NOT NULL,

    attempts    INTEGER NOT NULL DEFAULT 0,

    next_ms     INTEGER NOT NULL,

    status      INTEGER,

    response    TEXT NOT NULL DEFAULT '',

    delivered_ms INTEGER

);

CREATE INDEX deliveries_due ON deliveries (next_ms) WHERE delivered_ms IS NULL;

CREATE INDEX deliveries_by_hook ON deliveries (hook_id, created_ms DESC);

CREATE TABLE checks (

    id          INTEGER PRIMARY KEY,

    repo_id     INTEGER NOT NULL REFERENCES repositories(id) ON DELETE CASCADE,

    state       TEXT NOT NULL,

    name        TEXT NOT NULL,

    status      TEXT NOT NULL CHECK (status IN ('queued', 'running', 'passed', 'failed', 'cancelled', 'skipped')),

    summary     TEXT NOT NULL DEFAULT '',

    url         TEXT NOT NULL DEFAULT '',

    reporter_id INTEGER REFERENCES accounts(id) ON DELETE SET NULL,

    reporter    TEXT NOT NULL DEFAULT '',

    seal        TEXT NOT NULL DEFAULT '',

    started_ms  INTEGER NOT NULL,

    finished_ms INTEGER,

    UNIQUE (repo_id, state, name)

);

CREATE INDEX checks_by_state ON checks (repo_id, state);

ALTER TABLE repositories ADD COLUMN required_checks TEXT NOT NULL DEFAULT '';

ALTER TABLE tokens ADD COLUMN scopes_note TEXT NOT NULL DEFAULT '';

//! Webhooks, their delivery log, and checks (hub spec §12.2, §13.1).

use super::{now_ms, Pool};

use sqlx::FromRow;

#[derive(Clone, Debug, FromRow)]

pub struct Hook {

    pub id: i64,

    pub repo_id: Option<i64>,

    pub org_id: Option<i64>,

    pub url: String,

    pub secret: String,

    pub events: String,

    pub active: bool,

    pub created_ms: i64,

}

impl Hook {

    /// Whether the hook wants `event` (`*`, or a comma list of prefixes such as `push,proposal`).

    pub fn wants(&self, event: &str) -> bool {

        self.active && self.events.split(',').map(str::trim).any(|e| e == "*" || event == e || event.starts_with(&format!("{e}.")))

    }

}

#[derive(Clone, Debug, FromRow)]

pub struct Delivery {

    pub id: i64,

    pub hook_id: i64,

    pub delivery: String,

    pub event: String,

    pub payload: String,

    pub created_ms: i64,

    pub attempts: i64,

    pub next_ms: i64,

    pub status: Option<i64>,

    pub response: String,

    pub delivered_ms: Option<i64>,

}

pub async fn hooks_for_repo(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<Hook>> {

    sqlx::query_as("SELECT * FROM hooks WHERE repo_id = ? ORDER BY id").bind(repo_id).fetch_all(pool).await

}

pub async fn hooks_for_org(pool: &Pool, org_id: i64) -> sqlx::Result<Vec<Hook>> {

    sqlx::query_as("SELECT * FROM hooks WHERE org_id = ? ORDER BY id").bind(org_id).fetch_all(pool).await

}

/// Every hook that should hear about an event in a repository: its own

/// and its organisation's.

pub async fn hooks_hearing(pool: &Pool, repo_id: i64, owner: &str) -> sqlx::Result<Vec<Hook>> {

    let mut out = hooks_for_repo(pool, repo_id).await?;

    if let Some(org) = super::accounts::org_by_name(pool, owner).await? {

        out.extend(hooks_for_org(pool, org.id).await?);

    }

    Ok(out)

}

pub async fn hook(pool: &Pool, id: i64) -> sqlx::Result<Option<Hook>> {

    sqlx::query_as("SELECT * FROM hooks WHERE id = ?").bind(id).fetch_optional(pool).await

}

pub async fn create_hook(

    pool: &Pool,

    repo_id: Option<i64>,

    org_id: Option<i64>,

    url: &str,

    secret: &str,

    events: &str,

) -> sqlx::Result<Hook> {

    sqlx::query_as("INSERT INTO hooks (repo_id, org_id, url, secret, events, created_ms) VALUES (?, ?, ?, ?, ?, ?) RETURNING *")

        .bind(repo_id)

        .bind(org_id)

        .bind(url)

        .bind(secret)

        .bind(events)

        .bind(now_ms())

        .fetch_one(pool)

        .await

}

pub async fn set_hook_active(pool: &Pool, id: i64, active: bool) -> sqlx::Result<()> {

    sqlx::query("UPDATE hooks SET active = ? WHERE id = ?").bind(active).bind(id).execute(pool).await?;

    Ok(())

}

pub async fn delete_hook(pool: &Pool, id: i64) -> sqlx::Result<()> {

    sqlx::query("DELETE FROM hooks WHERE id = ?").bind(id).execute(pool).await?;

    Ok(())

}

pub async fn queue_delivery(pool: &Pool, hook_id: i64, delivery: &str, event: &str, payload: &str) -> sqlx::Result<i64> {

    let now = now_ms();

    sqlx::query_scalar(

        "INSERT INTO deliveries (hook_id, delivery, event, payload, created_ms, next_ms) VALUES (?, ?, ?, ?, ?, ?) RETURNING id",

    )

    .bind(hook_id)

    .bind(delivery)

    .bind(event)

    .bind(payload)

    .bind(now)

    .bind(now)

    .fetch_one(pool)

    .await

}

pub async fn due_deliveries(pool: &Pool, limit: i64) -> sqlx::Result<Vec<Delivery>> {

    sqlx::query_as("SELECT * FROM deliveries WHERE delivered_ms IS NULL AND next_ms <= ? ORDER BY next_ms LIMIT ?")

        .bind(now_ms())

        .bind(limit)

        .fetch_all(pool)

        .await

}

pub async fn deliveries_for_hook(pool: &Pool, hook_id: i64, limit: i64) -> sqlx::Result<Vec<Delivery>> {

    sqlx::query_as("SELECT * FROM deliveries WHERE hook_id = ? ORDER BY created_ms DESC, id DESC LIMIT ?")

        .bind(hook_id)

        .bind(limit)

        .fetch_all(pool)

        .await

}

pub async fn delivery(pool: &Pool, id: i64) -> sqlx::Result<Option<Delivery>> {

    sqlx::query_as("SELECT * FROM deliveries WHERE id = ?").bind(id).fetch_optional(pool).await

}

/// Record an attempt: done on a 2xx, else retried with backoff up to a day.

pub async fn record_attempt(pool: &Pool, id: i64, attempts: i64, status: Option<i64>, response: &str) -> sqlx::Result<()> {

    let ok = status.is_some_and(|s| (200..300).contains(&s));

    let now = now_ms();

    let delivered: Option<i64> = if ok { Some(now) } else { None };

    // 1 min, 5 min, 30 min, 2 h, 6 h, then give up after a day's worth.

    let backoff_ms = [60_000i64, 300_000, 1_800_000, 7_200_000, 21_600_000];

    let next = if ok || attempts as usize >= backoff_ms.len() { i64::MAX / 4 } else { now + backoff_ms[attempts as usize] };

    sqlx::query("UPDATE deliveries SET attempts = ?, status = ?, response = ?, delivered_ms = ?, next_ms = ? WHERE id = ?")

        .bind(attempts + 1)

        .bind(status)

        .bind(&response.chars().take(2000).collect::<String>())

        .bind(delivered)

        .bind(next)

        .bind(id)

        .execute(pool)

        .await?;

    Ok(())

}

/// Requeue a delivery for an immediate retry.

pub async fn redeliver(pool: &Pool, id: i64) -> sqlx::Result<()> {

    sqlx::query("UPDATE deliveries SET delivered_ms = NULL, next_ms = ? WHERE id = ?").bind(now_ms()).bind(id).execute(pool).await?;

    Ok(())

}

pub async fn prune_deliveries(pool: &Pool, older_than_ms: i64) -> sqlx::Result<()> {

    sqlx::query("DELETE FROM deliveries WHERE created_ms < ?").bind(older_than_ms).execute(pool).await?;

    Ok(())

}

// ------------------------------------------------------------------ checks

#[derive(Clone, Debug, FromRow)]

pub struct Check {

    pub id: i64,

    pub repo_id: i64,

    pub state: String,

    pub name: String,

    pub status: String,

    pub summary: String,

    pub url: String,

    pub reporter_id: Option<i64>,

    pub reporter: String,

    pub seal: String,

    pub started_ms: i64,

    pub finished_ms: Option<i64>,

}

impl Check {

    pub fn done(&self) -> bool {

        matches!(self.status.as_str(), "passed" | "failed" | "cancelled" | "skipped")

    }

}

pub async fn checks_for_state(pool: &Pool, repo_id: i64, state: &str) -> sqlx::Result<Vec<Check>> {

    sqlx::query_as("SELECT * FROM checks WHERE repo_id = ? AND state = ? ORDER BY name").bind(repo_id).bind(state).fetch_all(pool).await

}

/// Create or update a check by `(state, name)`.

pub async fn upsert_check(

    pool: &Pool,

    repo_id: i64,

    state: &str,

    name: &str,

    status: &str,

    summary: &str,

    url: &str,

    reporter: Option<(i64, &str)>,

    seal: &str,

) -> sqlx::Result<Check> {

    let now = now_ms();

    let finished: Option<i64> = if matches!(status, "passed" | "failed" | "cancelled" | "skipped") { Some(now) } else { None };

    sqlx::query_as(

        "INSERT INTO checks (repo_id, state, name, status, summary, url, reporter_id, reporter, seal, started_ms, finished_ms) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) \

         ON CONFLICT (repo_id, state, name) DO UPDATE SET status = excluded.status, summary = excluded.summary, url = excluded.url, reporter_id = excluded.reporter_id, reporter = excluded.reporter, seal = CASE WHEN excluded.seal = '' THEN checks.seal ELSE excluded.seal END, finished_ms = excluded.finished_ms RETURNING *",

    )

    .bind(repo_id)

    .bind(state)

    .bind(name)

    .bind(status)

    .bind(summary)

    .bind(url)

    .bind(reporter.map(|r| r.0))

    .bind(reporter.map(|r| r.1).unwrap_or(""))

    .bind(seal)

    .bind(now)

    .bind(finished)

    .fetch_one(pool)

    .await

}

pub async fn required_checks(pool: &Pool, repo_id: i64) -> sqlx::Result<Vec<String>> {

    let s: String = sqlx::query_scalar("SELECT required_checks FROM repositories WHERE id = ?").bind(repo_id).fetch_one(pool).await?;

    Ok(s.split(',').map(str::trim).filter(|x| !x.is_empty()).map(String::from).collect())

}

pub async fn set_required_checks(pool: &Pool, repo_id: i64, names: &[String]) -> sqlx::Result<()> {

    sqlx::query("UPDATE repositories SET required_checks = ? WHERE id = ?").bind(names.join(",")).bind(repo_id).execute(pool).await?;

    Ok(())

}

//! In-process event bus: nudges the webhook worker and streams proposal

//! updates to open pages over server-sent events (hub spec §13.1).

use hub_api::ProposalEvent;

use tokio::sync::broadcast;

pub struct Events {

    pub deliveries: broadcast::Sender<()>,

    pub proposals: broadcast::Sender<(i64, ProposalEvent)>,

}

impl Default for Events {

    fn default() -> Self {

        Events { deliveries: broadcast::channel(16).0, proposals: broadcast::channel(256).0 }

    }

}

impl Events {

    pub fn notify_deliveries(&self) {

        let _ = self.deliveries.send(());

    }

    /// Announce something about a proposal of a repository.

    pub fn proposal(&self, repo_id: i64, event: ProposalEvent) {

        let _ = self.proposals.send((repo_id, event));

    }

}

//! Webhooks (hub spec §12.2): queueing an event for every hook that wants

//! it, signing and delivering with retries, and the payload envelope.

use crate::db::{self, hooks as hdb};

use crate::Shared;

use hub_api::WebhookEnvelope;

/// Queue `event` for the hooks of a repository (and its organisation).

pub async fn emit(app: &Shared, repo: &db::repos::RepoRow, event: &str, actor: &str, data: serde_json::Value) {

    let hooks = match hdb::hooks_hearing(&app.db, repo.id, &repo.owner).await {

        Ok(h) => h,

        Err(e) => {

            tracing::warn!("hooks: {e}");

            return;

        }

    };

    if hooks.is_empty() {

        return;

    }

    let repository = hub_api::Repository {

        owner: repo.owner.clone(),

        name: repo.name.clone(),

        description: repo.description.clone(),

        default_bookmark: repo.default_bookmark.clone(),

        visibility: repo.visibility(),

        created_ms: repo.created_ms,

        updated_ms: repo.updated_ms,

        clone_url: format!("{}/{}", app.config.base_url(), repo.slug()),

    };

    for h in hooks.into_iter().filter(|h| h.wants(event)) {

        let delivery = crate::auth::random_hex(16);

        let envelope = WebhookEnvelope {

            event: event.into(),

            delivery: delivery.clone(),

            repository: Some(repository.clone()),

            actor: actor.into(),

            at_ms: db::now_ms(),

            data: data.clone(),

        };

        let payload = serde_json::to_string(&envelope).unwrap_or_default();

        if let Err(e) = hdb::queue_delivery(&app.db, h.id, &delivery, event, &payload).await {

            tracing::warn!("queueing delivery: {e}");

        }

    }

    app.events.notify_deliveries();

}

/// `X-Oxyhub-Signature` for a body: keyed BLAKE3 with the hook's secret.

pub fn signature(secret: &str, body: &[u8]) -> String {

    let key = blake3::hash(secret.as_bytes());

    format!("blake3-keyed={}", hex::encode(blake3::keyed_hash(key.as_bytes(), body).as_bytes()))

}

/// Deliver everything that is due; returns how many were attempted.

pub async fn deliver_due(app: &Shared) -> usize {

    let due = match hdb::due_deliveries(&app.db, 50).await {

        Ok(d) => d,

        Err(_) => return 0,

    };

    let client = reqwest::Client::builder()

        .timeout(std::time::Duration::from_secs(20))

        .user_agent(format!("oxygenhub/{}", env!("CARGO_PKG_VERSION")))

        .build();

    let Ok(client) = client else { return 0 };

    let mut n = 0;

    for d in due {

        let Ok(Some(hook)) = hdb::hook(&app.db, d.hook_id).await else { continue };

        n += 1;

        let (status, response) = match client

            .post(&hook.url)

            .header("content-type", "application/json")

            .header("x-oxyhub-event", &d.event)

            .header("x-oxyhub-delivery", &d.delivery)

            .header("x-oxyhub-signature", signature(&hook.secret, d.payload.as_bytes()))

            .body(d.payload.clone())

            .send()

            .await

        {

            Ok(r) => {

                let status = r.status().as_u16() as i64;

                let text = r.text().await.unwrap_or_default();

                (Some(status), text)

            }

            Err(e) => (None, e.to_string()),

        };

        hdb::record_attempt(&app.db, d.id, d.attempts, status, &response).await.ok();

    }

    n

}

/// The background worker: delivers when nudged and every minute anyway.

pub async fn worker(app: Shared) {

    let mut rx = app.events.deliveries.subscribe();

    loop {

        deliver_due(&app).await;

        let cutoff = db::now_ms() - app.config.retention.webhook_log_days as i64 * 86_400_000;

        hdb::prune_deliveries(&app.db, cutoff).await.ok();

        tokio::select! {

            _ = rx.recv() => {}

            _ = tokio::time::sleep(std::time::Duration::from_secs(60)) => {}

        }

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn signatures_are_keyed() {

        let a = signature("s1", b"body");

        assert!(a.starts_with("blake3-keyed="));

        assert_ne!(a, signature("s2", b"body"));

        assert_ne!(a, signature("s1", b"other"));

    }

}

//! Milestone 0.5 end to end: the JSON API with scoped tokens, webhooks

//! delivered to a local receiver with signatures and retries, checks on a

//! projected state that gate landing, and the proposal event stream.

use axum::body::Body;

use axum::http::{header, Request, StatusCode};

use axum::routing::post;

use axum::Router;

use http_body_util::BodyExt;

use oxy_core::repo::Repo;

use oxygenhub::config::Config;

use oxygenhub::db;

use oxygenhub::repos::RepoName;

use oxygenhub::Shared;

use std::path::Path;

use std::sync::{Arc, Mutex};

use tower::ServiceExt;

fn hermetic() {

    static ONCE: std::sync::Once = std::sync::Once::new();

    static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();

    ONCE.call_once(|| {

        let d = tempfile::tempdir().unwrap();

        std::env::set_var("OXY_HOME", d.path());

        DIR.set(d).ok();

    });

}

fn write(root: &Path, rel: &str, content: &[u8]) {

    let p = root.join(rel);

    std::fs::create_dir_all(p.parent().unwrap()).unwrap();

    std::fs::write(p, content).unwrap();

}

struct Hub {

    app: Shared,

    base: String,

    _dir: tempfile::TempDir,

}

async fn hub() -> Hub {

    hermetic();

    let dir = tempfile::tempdir().unwrap();

    let mut config = Config::load(dir.path()).unwrap();

    config.auth.push_tokens = vec!["site".into()];

    config.registration.mode = "open".into();

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();

    let port = listener.local_addr().unwrap().port();

    config.server.base_url = format!("http://127.0.0.1:{port}");

    let app = oxygenhub::build(config).await.unwrap();

    let router = oxygenhub::router(app.clone());

    tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });

    Hub { app, base: format!("http://127.0.0.1:{port}"), _dir: dir }

}

struct Res {

    status: StatusCode,

    body: String,

    cookie: Option<String>,

}

async fn call(

    app: &Shared,

    method: &str,

    path: &str,

    bearer: Option<&str>,

    cookie: Option<&str>,

    json: Option<&str>,

    form: Option<&str>,

) -> Res {

    let mut req = Request::builder().method(method).uri(path);

    if let Some(b) = bearer {

        req = req.header(header::AUTHORIZATION, format!("Bearer {b}"));

    }

    if let Some(c) = cookie {

        req = req.header(header::COOKIE, format!("oxyhub={c}"));

    }

    let body = if let Some(j) = json {

        req = req.header(header::CONTENT_TYPE, "application/json");

        Body::from(j.to_string())

    } else if let Some(f) = form {

        req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");

        Body::from(f.to_string())

    } else {

        Body::empty()

    };

    let res = oxygenhub::router(app.clone()).oneshot(req.body(body).unwrap()).await.unwrap();

    let status = res.status();

    let cookie = res

        .headers()

        .get(header::SET_COOKIE)

        .and_then(|v| v.to_str().ok())

        .and_then(|v| v.split(';').next())

        .and_then(|kv| kv.split_once('='))

        .map(|(_, v)| v.to_string());

    let bytes = res.into_body().collect().await.unwrap().to_bytes();

    Res { status, body: String::from_utf8_lossy(&bytes).into_owned(), cookie }

}

async fn api(app: &Shared, method: &str, path: &str, bearer: Option<&str>, json: Option<&str>) -> (StatusCode, serde_json::Value) {

    let r = call(app, method, path, bearer, None, json, None).await;

    let v = serde_json::from_str(&r.body).unwrap_or(serde_json::Value::String(r.body.clone()));

    (r.status, v)

}

fn csrf_of(page: &str) -> String {

    let i = page.find("name=\"_csrf\" value=\"").expect("csrf field") + "name=\"_csrf\" value=\"".len();

    page[i..].split('"').next().unwrap().to_string()

}

/// A webhook receiver that records what it gets and fails the first time.

struct Receiver {

    url: String,

    seen: Arc<Mutex<Vec<(String, String, String)>>>,

    fail_first: Arc<Mutex<bool>>,

}

async fn receiver() -> Receiver {

    let seen: Arc<Mutex<Vec<(String, String, String)>>> = Arc::new(Mutex::new(vec![]));

    let fail_first = Arc::new(Mutex::new(true));

    let (s2, f2) = (seen.clone(), fail_first.clone());

    let router = Router::new().route(

        "/hook",

        post(move |headers: header::HeaderMap, body: String| {

            let (s2, f2) = (s2.clone(), f2.clone());

            async move {

                let event = headers.get("x-oxyhub-event").and_then(|v| v.to_str().ok()).unwrap_or("").to_string();

                let sig = headers.get("x-oxyhub-signature").and_then(|v| v.to_str().ok()).unwrap_or("").to_string();

                let mut f = f2.lock().unwrap();

                if *f {

                    *f = false;

                    return (StatusCode::INTERNAL_SERVER_ERROR, "flaky");

                }

                s2.lock().unwrap().push((event, sig, body));

                (StatusCode::OK, "ok")

            }

        }),

    );

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();

    let port = listener.local_addr().unwrap().port();

    tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });

    Receiver { url: format!("http://127.0.0.1:{port}/hook"), seen, fail_first }

}

#[tokio::test(flavor = "multi_thread")]

async fn api_webhooks_and_checks() {

    let h = hub().await;

    let app = &h.app;

    // Accounts and tokens through the pages.

    let r =

        call(app, "POST", "/register", None, None, None, Some("username=andy&email=a%40example.com&password=correct-horse-battery")).await;

    let andy = r.cookie.unwrap();

    let r =

        call(app, "POST", "/register", None, None, None, Some("username=bob&email=b%40example.com&password=correct-horse-battery")).await;

    let bob = r.cookie.unwrap();

    let mint = |cookie: String, scopes: &'static str| {

        let app = app.clone();

        async move {

            let page = call(&app, "GET", "/settings/tokens", None, Some(&cookie), None, None).await;

            let csrf = csrf_of(&page.body);

            let r = call(&app, "POST", "/settings/tokens", None, Some(&cookie), None, Some(&format!("_csrf={csrf}&name=t&{scopes}"))).await;

            r.body.split("oxyhub_").nth(1).map(|s| format!("oxyhub_{}", s.split('<').next().unwrap().trim())).unwrap()

        }

    };

    let andy_admin = mint(andy.clone(), "scopes=repo%3Aadmin").await;

    let andy_ro = mint(andy.clone(), "scopes=repo%3Aread").await;

    let bob_rw = mint(bob.clone(), "scopes=repo%3Aread&scopes=repo%3Awrite").await;

    // Create a repository through the API; the read-only token cannot.

    let (st, _) = api(app, "POST", "/api/v1/repos", Some(&andy_ro), Some(r#"{"owner":"andy","name":"x"}"#)).await;

    assert_eq!(st, StatusCode::FORBIDDEN);

    let (st, v) = api(

        app,

        "POST",

        "/api/v1/repos",

        Some(&andy_admin),

        Some(r#"{"owner":"andy","name":"demo","description":"d","visibility":"public"}"#),

    )

    .await;

    assert_eq!(st, StatusCode::CREATED, "{v}");

    assert_eq!(v["clone_url"], format!("{}/andy/demo", h.base));

    let (st, _) = api(app, "POST", "/api/v1/repos", Some(&andy_admin), Some(r#"{"owner":"andy","name":"demo"}"#)).await;

    assert_eq!(st, StatusCode::CONFLICT);

    let (st, _) = api(app, "GET", "/api/v1/user", Some(&andy_ro), None).await;

    assert_eq!(st, StatusCode::FORBIDDEN, "user:read is a scope of its own");

    let (st, v) = api(app, "GET", "/api/v1/users/andy", None, None).await;

    assert_eq!(st, StatusCode::OK);

    assert_eq!(v["username"], "andy");

    // A webhook on the repository; push, then see the delivery with a valid signature after a retry.

    let rx = receiver().await;

    let (st, v) = api(

        app,

        "POST",

        "/api/v1/repos/andy/demo/hooks",

        Some(&andy_admin),

        Some(&format!(r#"{{"url":"{}","secret":"s3","events":"push,proposal,check"}}"#, rx.url)),

    )

    .await;

    assert_eq!(st, StatusCode::CREATED, "{v}");

    let work = tempfile::tempdir().unwrap();

    let mut client = Repo::init(work.path()).unwrap();

    client.config.set("remote.origin.url", &format!("{}/andy/demo", h.base)).unwrap();

    client.config.set("remote.origin.token", "site").unwrap();

    client.save_config().unwrap();

    write(work.path(), "a.txt", b"a\n");

    client.snapshot("t").unwrap();

    let base = client.commit("base", &[]).unwrap();

    assert!(client.push("origin", false).unwrap().iter().all(|r| r.ok));

    // The worker retries with backoff; drive it directly for the retry.

    tokio::time::sleep(std::time::Duration::from_millis(400)).await;

    assert!(rx.seen.lock().unwrap().is_empty(), "the first attempt failed");

    sqlx::query("UPDATE deliveries SET next_ms = 0").execute(&app.db).await.unwrap();

    oxygenhub::hooks::deliver_due(app).await;

    let seen = rx.seen.lock().unwrap().clone();

    assert_eq!(seen.len(), 1, "{seen:?}");

    assert_eq!(seen[0].0, "push");

    assert_eq!(seen[0].1, oxygenhub::hooks::signature("s3", seen[0].2.as_bytes()));

    let payload: serde_json::Value = serde_json::from_str(&seen[0].2).unwrap();

    assert_eq!(payload["repository"]["name"], "demo");

    assert_eq!(payload["data"]["updates"][0]["reference"], "main");

    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/hooks", Some(&andy_admin), None).await;

    assert_eq!(v.as_array().unwrap().len(), 1);

    // History, tree, file, blame, bookmarks, trust, pushes through the API.

    let (st, v) = api(app, "GET", "/api/v1/repos/andy/demo/history/main", None, None).await;

    assert_eq!(st, StatusCode::OK);

    assert_eq!(v["changes"][0]["hash"], base.to_hex());

    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/tree/main", None, None).await;

    assert_eq!(v[0]["name"], "a.txt");

    let r = call(app, "GET", "/api/v1/repos/andy/demo/file/main/a.txt", None, None, None, None).await;

    assert_eq!(r.body, "a\n");

    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/blame/main/a.txt", None, None).await;

    assert_eq!(v[0]["change"], base.to_hex());

    let (_, v) = api(app, "GET", &format!("/api/v1/repos/andy/demo/changes/{}", base.to_hex()), None, None).await;

    assert_eq!(v["files"][0]["path"], "a.txt");

    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/trust", None, None).await;

    assert!(v["policy"].is_null());

    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/pushes", None, None).await;

    assert_eq!(v[0]["reference"], "main");

    // Issues through the API: Bob opens one, comments, and closes it.

    let (st, v) = api(app, "POST", "/api/v1/repos/andy/demo/issues", Some(&bob_rw), Some(r#"{"title":"Bug","body":"it broke"}"#)).await;

    assert_eq!(st, StatusCode::CREATED, "{v}");

    assert_eq!(v["number"], 1);

    let (st, _) = api(app, "POST", "/api/v1/repos/andy/demo/issues/1/comments", Some(&bob_rw), Some(r#"{"body":"more"}"#)).await;

    assert_eq!(st, StatusCode::CREATED);

    let (st, v) = api(app, "POST", "/api/v1/repos/andy/demo/issues/1", Some(&bob_rw), Some(r#"{"state":"closed"}"#)).await;

    assert_eq!(st, StatusCode::OK, "{v}");

    assert_eq!(v["state"], "closed");

    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/issues?state=all", None, None).await;

    assert_eq!(v[0]["comments"], 1);

    let (_, v) = api(app, "GET", "/api/v1/search?q=broke", None, None).await;

    assert_eq!(v[0]["number"], 1);

    // A proposal through the API, then a required check gates its landing.

    client.bookmark_create("feat", Some(&[base])).unwrap();

    client.bookmark_switch("feat", false).unwrap();

    write(work.path(), "b.txt", b"b\n");

    client.snapshot("t").unwrap();

    let feat = client.commit("feat", &[]).unwrap();

    assert!(client.push_bookmark("origin", "feat", false).unwrap().iter().all(|r| r.ok));

    let (st, v) =

        api(app, "POST", "/api/v1/repos/andy/demo/proposals", Some(&bob_rw), Some(r#"{"title":"Feat","target":"main","source":"feat"}"#))

            .await;

    assert_eq!(st, StatusCode::CREATED, "{v}");

    assert_eq!(v["number"], 2);

    let state = v["projected_state"].as_str().unwrap().to_string();

    assert_eq!(v["landable"], true);

    // Require the "build" check.

    let settings = call(app, "GET", "/andy/demo/settings", None, Some(&andy), None, None).await;

    let csrf = csrf_of(&settings.body);

    assert_eq!(

        call(app, "POST", "/andy/demo/settings/checks", None, Some(&andy), None, Some(&format!("_csrf={csrf}&required_checks=build")))

            .await

            .status,

        StatusCode::OK

    );

    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/proposals/2", None, None).await;

    assert_eq!(v["landable"], false);

    assert!(v["blockers"][0].as_str().unwrap().contains("build"), "{v}");

    // A checker reports on the projected state: first running, then passed; the page streams it.

    let (st, v) = api(

        app,

        "POST",

        "/api/v1/repos/andy/demo/checks",

        Some(&andy_ro),

        Some(&format!(r#"{{"state":"{state}","name":"build","status":"running"}}"#)),

    )

    .await;

    assert_eq!(st, StatusCode::FORBIDDEN, "{v}");

    let mut events = app.events.proposals.subscribe();

    let (st, v) = api(

        app,

        "POST",

        "/api/v1/repos/andy/demo/checks",

        Some(&bob_rw),

        Some(&format!(r#"{{"state":"{state}","name":"build","status":"passed","summary":"42 tests","url":"https://ci.example/1"}}"#)),

    )

    .await;

    assert_eq!(st, StatusCode::CREATED, "{v}");

    let (_, ev) = events.recv().await.unwrap();

    assert_eq!(ev.kind, "check");

    assert_eq!(ev.check.unwrap().name, "build");

    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/proposals/2/checks", None, None).await;

    assert_eq!(v[0]["status"], "passed");

    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/proposals/2", None, None).await;

    assert_eq!(v["landable"], true, "{v}");

    let page = call(app, "GET", "/andy/demo/proposal/2", None, Some(&andy), None, None).await;

    assert!(page.body.contains("42 tests"), "{}", page.body);

    // The check and the proposal reached the webhook too.

    sqlx::query("UPDATE deliveries SET next_ms = 0").execute(&app.db).await.unwrap();

    oxygenhub::hooks::deliver_due(app).await;

    let kinds: Vec<String> = rx.seen.lock().unwrap().iter().map(|s| s.0.clone()).collect();

    assert!(kinds.contains(&"proposal.opened".to_string()) && kinds.contains(&"check".to_string()), "{kinds:?}");

    let _ = feat;

    let _ = rx.fail_first;

}

{% extends "base.html" %}

{% import "repo_tabs.html" as m %}

{% block title %}Deliveries · {{ repo.owner }}/{{ repo.name }}{% endblock %}

{% block tabs %}{% call m::tabs(repo, "settings") %}{% endblock %}

{% block mainclass %} class="wide"{% endblock %}

{% block content %}

<div class="head"><h1>Deliveries <small>{{ hook.url }} · events {{ hook.events }}</small></h1></div>

<p><a href="{{ repo.url }}/settings">← settings</a></p>

{% if rows.is_empty() %}<p class="empty">Nothing delivered yet.</p>{% else %}

<div class="table"><table>

<tr><th>When</th><th>Event</th><th>Delivery</th><th>Attempts</th><th>Status</th><th>Response</th><th></th></tr>

{% for d in rows %}

<tr><td><time title="{{ d.created_ms|when_i }}">{{ d.created_ms|ago_i }}</time></td><td>{{ d.event }}</td><td><code>{{ d.delivery }}</code></td><td>{{ d.attempts }}</td>

<td>{% if let Some(s) = d.status %}{% if d.delivered_ms.is_some() %}<span class="sig sig-valid">✔</span>{% else %}<span class="sig sig-invalid">✘</span>{% endif %} {{ s }}{% else if d.attempts > 0 %}<span class="sig sig-invalid">✘</span> no response{% else %}<span class="mute">queued</span>{% endif %}</td>

<td class="small mute">{{ d.response|truncate(80) }}</td>

<td><form class="tiny" method="post" action="{{ repo.url }}/settings/hooks/{{ hook.id }}/redeliver/{{ d.id }}"><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button>redeliver</button></form></td></tr>

{% endfor %}

</table></div>

{% endif %}

{% endblock %}

name = "dyn-clone"

version = "1.0.20"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555"

[[package]]

name = "futures"

version = "0.3.34"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3"

dependencies = [

 "futures-channel",

 "futures-core",

 "futures-executor",

 "futures-io",

 "futures-sink",

 "futures-task",

 "futures-util",

]

[[package]]

name = "futures-macro"

version = "0.3.34"

source = "registry+https://github.com/rust-lang/crates.io-index"

checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44"

dependencies = [

 "proc-macro2",

 "quote",

 "syn 3.0.6",

]

[[package]]

 "futures-channel",

 "futures-macro",

 "schemars",

 "serde_json",

 "futures",

 "tokio-stream",