use std::collections::{
BTreeSet,
HashMap,
};
use std::time::Duration;
use async_trait::async_trait;
use futures::future::join_all;
use pipedash_plugin_api::*;
use crate::log_dialect::BuildkiteLogDialect;
use crate::{
client,
config,
mapper,
metadata,
};
pub struct BuildkitePlugin {
metadata: PluginMetadata,
client: Option<client::BuildkiteClient>,
provider_id: Option<i64>,
config: HashMap<String, String>,
status_source: Option<StatusSource>,
}
impl Default for BuildkitePlugin {
fn default() -> Self {
Self::new()
}
}
impl BuildkitePlugin {
pub fn new() -> Self {
Self {
metadata: metadata::create_metadata(),
client: None,
provider_id: None,
config: HashMap::new(),
status_source: None,
}
}
fn client(&self) -> PluginResult<&client::BuildkiteClient> {
self.client
.as_ref()
.ok_or_else(|| PluginError::Internal("Plugin not initialized".to_string()))
}
fn org_and_slug(pipeline_id: &str) -> PluginResult<(&str, &str)> {
match pipeline_id.split("__").collect::<Vec<_>>().as_slice() {
[_, _, org, slug] => Ok((*org, *slug)),
_ => Err(PluginError::InvalidConfig(format!(
"Invalid pipeline ID format: {pipeline_id}"
))),
}
}
}
fn buildkite_status_source() -> PluginResult<StatusSource> {
let url = |value: &str| {
reqwest::Url::parse(value)
.map_err(|e| PluginError::Internal(format!("Invalid Buildkite status URL: {e}")))
};
Ok(StatusSource {
name: "Buildkite".to_string(),
page_url: url("https://www.buildkitestatus.com/")?,
feed_url: url("https://www.buildkitestatus.com/api/v2/summary.json")?,
format: FeedFormat::Statuspage,
components: ComponentFilter::Ids(
[
"r21z46hqn47c",
"1mpkbbd1kh0h",
"3n7fzwsysx86",
"56ph4s45xl9s",
]
.map(String::from)
.to_vec(),
),
})
}
#[async_trait]
impl Plugin for BuildkitePlugin {
fn metadata(&self) -> &PluginMetadata {
&self.metadata
}
fn status_source(&self) -> Option<&StatusSource> {
self.status_source.as_ref()
}
fn initialize(
&mut self, provider_id: i64, config: HashMap<String, String>,
http_client: Option<std::sync::Arc<reqwest::Client>>,
) -> PluginResult<()> {
let token = config
.get("token")
.ok_or_else(|| PluginError::InvalidConfig("Missing Buildkite API token".to_string()))?
.to_string();
let client = http_client.unwrap_or_else(|| {
std::sync::Arc::new(
reqwest::Client::builder()
.use_rustls_tls()
.pool_max_idle_per_host(10)
.timeout(Duration::from_secs(30))
.connect_timeout(Duration::from_secs(10))
.tcp_keepalive(Duration::from_secs(60))
.build()
.expect("Failed to build HTTP client"),
)
});
self.client = Some(client::BuildkiteClient::new(client, token));
self.provider_id = Some(provider_id);
self.config = config;
self.status_source = Some(buildkite_status_source()?);
Ok(())
}
async fn validate_credentials(&self) -> PluginResult<bool> {
let client = self.client()?;
client.fetch_organizations().await?;
Ok(true)
}
async fn fetch_available_pipelines(
&self, params: Option<PaginationParams>,
) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
let client = self.client()?;
client::fetch_all_available_pipelines(client, params).await
}
async fn fetch_organizations(&self) -> PluginResult<Vec<pipedash_plugin_api::Organization>> {
let client = self.client()?;
let orgs = client.fetch_organizations().await?;
Ok(orgs
.into_iter()
.map(|org| pipedash_plugin_api::Organization {
id: org.slug.clone(),
name: org.name,
description: org.description,
})
.collect())
}
async fn fetch_available_pipelines_filtered(
&self, org: Option<String>, search: Option<String>, params: Option<PaginationParams>,
) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
let client = self.client()?;
client::fetch_available_pipelines_filtered(client, org, search, params).await
}
async fn fetch_pipelines(&self) -> PluginResult<Vec<Pipeline>> {
let provider_id = self
.provider_id
.ok_or_else(|| PluginError::Internal("Provider ID not set".to_string()))?;
let pipelines = config::parse_selected_items(&self.config)?;
if pipelines.is_empty() {
return Err(PluginError::InvalidConfig(
"No pipelines configured".to_string(),
));
}
let client = self.client()?;
let futures = pipelines
.into_iter()
.map(|(org, slug)| client.fetch_pipeline(provider_id, org, slug));
let results = join_all(futures).await;
let mut all_pipelines = Vec::new();
let mut errors = Vec::new();
for result in results {
match result {
Ok(pipeline) => all_pipelines.push(pipeline),
Err(e) if e.is_transient() => return Err(e),
Err(e) => {
tracing::warn!(error = %e, "Failed to fetch Buildkite pipeline");
errors.push(e);
}
}
}
if all_pipelines.is_empty() {
if let Some(e) = errors.into_iter().next() {
return Err(e);
}
}
Ok(all_pipelines)
}
async fn fetch_run_history(
&self, pipeline_id: &str, limit: usize,
) -> PluginResult<Vec<PipelineRun>> {
let parts: Vec<&str> = pipeline_id.split("__").collect();
if parts.len() != 4 {
return Err(PluginError::InvalidConfig(format!(
"Invalid pipeline ID format: {pipeline_id}"
)));
}
let org = parts[2];
let slug = parts[3];
let client = self.client()?;
let builds = client.fetch_builds(org, slug, limit).await?;
let pipeline_runs = builds
.into_iter()
.map(|build| client::build_to_pipeline_run(build, pipeline_id))
.collect();
Ok(pipeline_runs)
}
async fn fetch_run_details(
&self, pipeline_id: &str, run_number: i64,
) -> PluginResult<PipelineRun> {
let (org, slug) = Self::org_and_slug(pipeline_id)?;
let build = self.client()?.fetch_build(org, slug, run_number).await?;
Ok(client::build_to_pipeline_run(build, pipeline_id))
}
async fn fetch_log_nodes(&self, run: RunRef<'_>) -> PluginResult<Vec<LogNode>> {
let (org, slug) = Self::org_and_slug(run.pipeline_id)?;
let jobs = self
.client()?
.fetch_build_jobs(org, slug, run.run_number)
.await?;
Ok(mapper::jobs_to_log_nodes(&jobs))
}
async fn read_log(
&self, run: RunRef<'_>, node_id: &str, sink: &mut LogSink,
) -> PluginResult<()> {
let (org, slug) = Self::org_and_slug(run.pipeline_id)?;
if node_id.is_empty() || !node_id.bytes().all(|b| b.is_ascii_hexdigit() || b == b'-') {
return Err(PluginError::InvalidConfig(format!(
"Invalid Buildkite job ID: {node_id}"
)));
}
self.client()?
.stream_job_log(org, slug, run.run_number, node_id, sink)
.await
}
fn log_dialect(&self) -> &'static dyn LogDialect {
&BuildkiteLogDialect
}
async fn fetch_workflow_parameters(
&self, _workflow_id: &str,
) -> PluginResult<Vec<WorkflowParameter>> {
Ok(vec![WorkflowParameter {
name: "branch".to_string(),
label: Some("Branch".to_string()),
description: Some("Branch to build".to_string()),
param_type: WorkflowParameterType::String {
default: Some("main".to_string()),
},
required: true,
}])
}
async fn trigger_pipeline(&self, params: TriggerParams) -> PluginResult<String> {
let parts: Vec<&str> = params.workflow_id.split("__").collect();
if parts.len() != 4 {
return Err(PluginError::InvalidConfig(format!(
"Invalid workflow ID format: {}",
params.workflow_id
)));
}
let org = parts[2];
let slug = parts[3];
let branch = params
.inputs
.as_ref()
.and_then(|inputs| inputs.get("branch"))
.and_then(|v| v.as_str())
.unwrap_or("main")
.to_string();
let client = self.client()?;
let build = client
.trigger_build(org, slug, branch.clone(), params.inputs)
.await?;
Ok(serde_json::json!({
"message": format!("Triggered build #{} on branch {}", build.number, branch),
"build_number": build.number,
"build_url": build.web_url
})
.to_string())
}
async fn fetch_agents(&self) -> PluginResult<Vec<BuildAgent>> {
let orgs: BTreeSet<String> = config::parse_selected_items(&self.config)?
.into_iter()
.map(|(org, _)| org)
.collect();
let client = self.client()?;
let mut agents = Vec::new();
for org in orgs {
agents.extend(client.fetch_agents(&org).await?);
}
Ok(agents.into_iter().map(mapper::map_agent).collect())
}
async fn fetch_artifacts(&self, run_id: &str) -> PluginResult<Vec<BuildArtifact>> {
let parts: Vec<&str> = run_id.split("__").collect();
let [_, _, org, _, build_id] = parts.as_slice() else {
return Err(PluginError::InvalidConfig(format!(
"Invalid run ID: {run_id}"
)));
};
let client = self.client()?;
let artifacts = client.fetch_artifacts(org, build_id).await?;
Ok(artifacts
.into_iter()
.map(|artifact| client::artifact_to_build_artifact(artifact, run_id))
.collect())
}
async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> PluginResult<()> {
let parts: Vec<&str> = pipeline_id.split("__").collect();
if parts.len() != 4 {
return Err(PluginError::InvalidConfig(format!(
"Invalid pipeline ID format: {pipeline_id}"
)));
}
let org = parts[2];
let slug = parts[3];
let client = self.client()?;
client.cancel_build(org, slug, run_number).await
}
}
#[cfg(test)]
mod tests {
use std::sync::atomic::{
AtomicUsize,
Ordering,
};
use std::sync::Arc;
use std::time::Duration;
use super::*;
use crate::test_support::{
stub_server,
Reply,
};
fn mixed(path: &str) -> Reply {
if path.contains("/limited") {
Reply {
status: "429 Too Many Requests",
headers: "Retry-After: 120\r\n",
body: "slow down",
}
} else if path.contains("/missing") {
Reply {
status: "404 Not Found",
headers: "",
body: "{}",
}
} else if path.contains("/builds") {
Reply {
status: "200 OK",
headers: "Content-Type: application/json\r\n",
body: "[]",
}
} else {
Reply {
status: "200 OK",
headers: "Content-Type: application/json\r\n",
body: r#"{"id":"1","slug":"ok","name":"ok","url":"u","repository":"git@github.com:acme/ok.git"}"#,
}
}
}
const BUILD_PATH: &str = "/organizations/acme/pipelines/app/builds/150";
const JOB_ID: &str = "0190d2a4-7b1e-4c55-9f0e-6f2a1c3b4d5e";
const RAW_LOG: &str = "\x1b_bk;t=1745322210692\x07~~~ Running script\r\n\
\x1b_bk;t=1745322210692\x07\x1b[90m$\x1b[0m ./script.sh\r\n";
fn build_150(path: &str) -> Reply {
let json = |body| Reply {
status: "200 OK",
headers: "Content-Type: application/json\r\n",
body,
};
match path.strip_prefix(BUILD_PATH) {
Some("?exclude_jobs=true") => json(
r#"{"id":"b-150","number":150,"state":"failed","web_url":"https://buildkite.com/acme/app/builds/150",
"created_at":"2026-10-27T10:00:00.000Z","started_at":"2026-10-27T10:00:05.000Z",
"finished_at":"2026-10-27T10:02:05.000Z","branch":"main","commit":"abc"}"#,
),
Some("") => json(
r#"{"number":150,"jobs":[{"id":"0190d2a4-7b1e-4c55-9f0e-6f2a1c3b4d5e","type":"script",
"name":"Tests","state":"failed","started_at":"2026-10-27T10:00:05.000Z",
"finished_at":"2026-10-27T10:02:05.000Z"}]}"#,
),
Some("/jobs/0190d2a4-7b1e-4c55-9f0e-6f2a1c3b4d5e/log.txt") => Reply {
status: "200 OK",
headers: "Content-Type: text/plain\r\n",
body: RAW_LOG,
},
_ => Reply {
status: "404 Not Found",
headers: "Content-Type: application/json\r\n",
body: r#"{"message":"Not Found"}"#,
},
}
}
async fn plugin_answering(
reply: fn(&str) -> Reply, selected: &str,
) -> (BuildkitePlugin, Arc<AtomicUsize>) {
let _ = rustls::crypto::ring::default_provider().install_default();
let requests = Arc::new(AtomicUsize::new(0));
let base_url = stub_server(requests.clone(), Duration::ZERO, reply).await;
let mut plugin = BuildkitePlugin::new();
let config = HashMap::from([
("token".to_string(), "token".to_string()),
("selected_items".to_string(), selected.to_string()),
]);
plugin.initialize(1, config, None).unwrap();
plugin.client = Some(
client::BuildkiteClient::new(Arc::new(reqwest::Client::new()), "token".to_string())
.with_base_url(base_url),
);
(plugin, requests)
}
async fn plugin_selecting(selected: &str) -> BuildkitePlugin {
plugin_answering(mixed, selected).await.0
}
fn run_150(metadata: &HashMap<String, serde_json::Value>) -> RunRef<'_> {
RunRef {
pipeline_id: "buildkite__1__acme__app",
run_number: 150,
run_id: "buildkite__1__acme__app__b-150",
metadata,
}
}
async fn read(plugin: &BuildkitePlugin, node_id: &str) -> PluginResult<Vec<u8>> {
let metadata = HashMap::new();
let (tx, mut rx) = tokio::sync::mpsc::channel(16);
let mut sink = LogSink::new(tx);
let result = plugin
.read_log(run_150(&metadata), node_id, &mut sink)
.await;
drop(sink);
let mut bytes = Vec::new();
while let Some(chunk) = rx.recv().await {
bytes.extend_from_slice(&chunk);
}
result.map(|()| bytes)
}
#[tokio::test]
async fn run_details_load_a_build_by_its_number() {
let (plugin, requests) = plugin_answering(build_150, "acme/app").await;
let run = plugin
.fetch_run_details("buildkite__1__acme__app", 150)
.await
.unwrap();
assert_eq!(run.run_number, 150);
assert_eq!(run.status, PipelineStatus::Failed);
assert_eq!(run.duration_seconds, Some(120));
assert_eq!(requests.load(Ordering::SeqCst), 1);
}
#[tokio::test]
async fn build_jobs_become_log_nodes() {
let (plugin, _) = plugin_answering(build_150, "acme/app").await;
let metadata = HashMap::new();
let nodes = plugin.fetch_log_nodes(run_150(&metadata)).await.unwrap();
assert_eq!(nodes.len(), 1);
assert_eq!(nodes[0].id, JOB_ID);
assert_eq!(nodes[0].name, "Tests");
assert_eq!(nodes[0].status, PipelineStatus::Failed);
assert_eq!(nodes[0].log, Some(LogAvailability::Ready));
}
#[tokio::test]
async fn job_log_streams_the_raw_text() {
let (plugin, _) = plugin_answering(build_150, "acme/app").await;
assert_eq!(read(&plugin, JOB_ID).await.unwrap(), RAW_LOG.as_bytes());
}
#[tokio::test]
async fn missing_job_log_is_not_found() {
let (plugin, _) = plugin_answering(build_150, "acme/app").await;
let error = read(&plugin, "0190d2a4-0000-0000-0000-000000000000")
.await
.unwrap_err();
assert!(
matches!(error, PluginError::PipelineNotFound(_)),
"{error:?}"
);
}
#[tokio::test]
async fn job_ids_outside_the_uuid_alphabet_are_rejected_without_a_request() {
let (plugin, requests) = plugin_answering(build_150, "acme/app").await;
for node_id in ["", "../../builds/150", "0190d2a4/log", "x?y=1"] {
let error = read(&plugin, node_id).await.unwrap_err();
assert!(
matches!(error, PluginError::InvalidConfig(_)),
"{node_id}: {error:?}"
);
}
assert_eq!(requests.load(Ordering::SeqCst), 0);
}
#[tokio::test]
async fn transient_failure_on_any_pipeline_fails_the_fetch() {
let plugin = plugin_selecting("acme/ok,acme/limited").await;
let error = plugin.fetch_pipelines().await.unwrap_err();
assert!(
matches!(error, PluginError::RateLimited { .. }),
"{error:?}"
);
}
#[tokio::test]
async fn permanent_failure_on_one_pipeline_keeps_the_rest() {
let plugin = plugin_selecting("acme/ok,acme/missing").await;
let pipelines = plugin.fetch_pipelines().await.unwrap();
assert_eq!(pipelines.len(), 1);
}
}