pipedash_plugin_buildkite/
plugin.rs

1use std::collections::{
2    BTreeSet,
3    HashMap,
4};
5use std::time::Duration;
6
7use async_trait::async_trait;
8use futures::future::join_all;
9use pipedash_plugin_api::*;
10
11use crate::{
12    client,
13    config,
14    mapper,
15    metadata,
16};
17
18pub struct BuildkitePlugin {
19    metadata: PluginMetadata,
20    client: Option<client::BuildkiteClient>,
21    provider_id: Option<i64>,
22    config: HashMap<String, String>,
23}
24
25impl Default for BuildkitePlugin {
26    fn default() -> Self {
27        Self::new()
28    }
29}
30
31impl BuildkitePlugin {
32    pub fn new() -> Self {
33        Self {
34            metadata: metadata::create_metadata(),
35            client: None,
36            provider_id: None,
37            config: HashMap::new(),
38        }
39    }
40
41    fn client(&self) -> PluginResult<&client::BuildkiteClient> {
42        self.client
43            .as_ref()
44            .ok_or_else(|| PluginError::Internal("Plugin not initialized".to_string()))
45    }
46}
47
48#[async_trait]
49impl Plugin for BuildkitePlugin {
50    fn metadata(&self) -> &PluginMetadata {
51        &self.metadata
52    }
53
54    fn initialize(
55        &mut self, provider_id: i64, config: HashMap<String, String>,
56        http_client: Option<std::sync::Arc<reqwest::Client>>,
57    ) -> PluginResult<()> {
58        let token = config
59            .get("token")
60            .ok_or_else(|| PluginError::InvalidConfig("Missing Buildkite API token".to_string()))?
61            .to_string();
62
63        let client = http_client.unwrap_or_else(|| {
64            std::sync::Arc::new(
65                reqwest::Client::builder()
66                    .use_rustls_tls()
67                    .pool_max_idle_per_host(10)
68                    .timeout(Duration::from_secs(30))
69                    .connect_timeout(Duration::from_secs(10))
70                    .tcp_keepalive(Duration::from_secs(60))
71                    .build()
72                    .expect("Failed to build HTTP client"),
73            )
74        });
75
76        self.client = Some(client::BuildkiteClient::new(client, token));
77        self.provider_id = Some(provider_id);
78        self.config = config;
79
80        Ok(())
81    }
82
83    async fn validate_credentials(&self) -> PluginResult<bool> {
84        let client = self.client()?;
85
86        client.fetch_organizations().await?;
87
88        Ok(true)
89    }
90
91    async fn fetch_available_pipelines(
92        &self, params: Option<PaginationParams>,
93    ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
94        let client = self.client()?;
95        client::fetch_all_available_pipelines(client, params).await
96    }
97
98    async fn fetch_organizations(&self) -> PluginResult<Vec<pipedash_plugin_api::Organization>> {
99        let client = self.client()?;
100        let orgs = client.fetch_organizations().await?;
101
102        Ok(orgs
103            .into_iter()
104            .map(|org| pipedash_plugin_api::Organization {
105                id: org.slug.clone(),
106                name: org.name,
107                description: org.description,
108            })
109            .collect())
110    }
111
112    async fn fetch_available_pipelines_filtered(
113        &self, org: Option<String>, search: Option<String>, params: Option<PaginationParams>,
114    ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
115        let client = self.client()?;
116        client::fetch_available_pipelines_filtered(client, org, search, params).await
117    }
118
119    async fn fetch_pipelines(&self) -> PluginResult<Vec<Pipeline>> {
120        let provider_id = self
121            .provider_id
122            .ok_or_else(|| PluginError::Internal("Provider ID not set".to_string()))?;
123
124        let pipelines = config::parse_selected_items(&self.config)?;
125
126        if pipelines.is_empty() {
127            return Err(PluginError::InvalidConfig(
128                "No pipelines configured".to_string(),
129            ));
130        }
131
132        let client = self.client()?;
133        let futures = pipelines
134            .into_iter()
135            .map(|(org, slug)| client.fetch_pipeline(provider_id, org, slug));
136
137        let results = join_all(futures).await;
138
139        let mut all_pipelines = Vec::new();
140        let mut errors = Vec::new();
141
142        for result in results {
143            match result {
144                Ok(pipeline) => all_pipelines.push(pipeline),
145                Err(e) => errors.push(e),
146            }
147        }
148
149        if !errors.is_empty() && all_pipelines.is_empty() {
150            return Err(errors.into_iter().next().unwrap());
151        }
152
153        Ok(all_pipelines)
154    }
155
156    async fn fetch_run_history(
157        &self, pipeline_id: &str, limit: usize,
158    ) -> PluginResult<Vec<PipelineRun>> {
159        let parts: Vec<&str> = pipeline_id.split("__").collect();
160        if parts.len() != 4 {
161            return Err(PluginError::InvalidConfig(format!(
162                "Invalid pipeline ID format: {pipeline_id}"
163            )));
164        }
165
166        let org = parts[2];
167        let slug = parts[3];
168
169        let client = self.client()?;
170        let builds = client.fetch_builds(org, slug, limit).await?;
171
172        let pipeline_runs = builds
173            .into_iter()
174            .map(|build| client::build_to_pipeline_run(build, pipeline_id))
175            .collect();
176
177        Ok(pipeline_runs)
178    }
179
180    async fn fetch_run_details(
181        &self, pipeline_id: &str, run_number: i64,
182    ) -> PluginResult<PipelineRun> {
183        let parts: Vec<&str> = pipeline_id.split("__").collect();
184        if parts.len() != 4 {
185            return Err(PluginError::InvalidConfig(format!(
186                "Invalid pipeline ID format: {pipeline_id}"
187            )));
188        }
189
190        let org = parts[2];
191        let slug = parts[3];
192
193        let client = self.client()?;
194        let builds = client.fetch_builds(org, slug, 100).await?;
195
196        let build = builds
197            .into_iter()
198            .find(|b| b.number == run_number)
199            .ok_or_else(|| {
200                PluginError::PipelineNotFound(format!(
201                    "Build #{run_number} not found for pipeline {pipeline_id}"
202                ))
203            })?;
204
205        Ok(client::build_to_pipeline_run(build, pipeline_id))
206    }
207
208    async fn fetch_workflow_parameters(
209        &self, _workflow_id: &str,
210    ) -> PluginResult<Vec<WorkflowParameter>> {
211        Ok(vec![WorkflowParameter {
212            name: "branch".to_string(),
213            label: Some("Branch".to_string()),
214            description: Some("Branch to build".to_string()),
215            param_type: WorkflowParameterType::String {
216                default: Some("main".to_string()),
217            },
218            required: true,
219        }])
220    }
221
222    async fn trigger_pipeline(&self, params: TriggerParams) -> PluginResult<String> {
223        let parts: Vec<&str> = params.workflow_id.split("__").collect();
224        if parts.len() != 4 {
225            return Err(PluginError::InvalidConfig(format!(
226                "Invalid workflow ID format: {}",
227                params.workflow_id
228            )));
229        }
230
231        let org = parts[2];
232        let slug = parts[3];
233
234        let branch = params
235            .inputs
236            .as_ref()
237            .and_then(|inputs| inputs.get("branch"))
238            .and_then(|v| v.as_str())
239            .unwrap_or("main")
240            .to_string();
241
242        let client = self.client()?;
243        let build = client
244            .trigger_build(org, slug, branch.clone(), params.inputs)
245            .await?;
246
247        Ok(serde_json::json!({
248            "message": format!("Triggered build #{} on branch {}", build.number, branch),
249            "build_number": build.number,
250            "build_url": build.web_url
251        })
252        .to_string())
253    }
254
255    async fn fetch_agents(&self) -> PluginResult<Vec<BuildAgent>> {
256        let orgs: BTreeSet<String> = config::parse_selected_items(&self.config)?
257            .into_iter()
258            .map(|(org, _)| org)
259            .collect();
260
261        let client = self.client()?;
262        let mut agents = Vec::new();
263        for org in orgs {
264            agents.extend(client.fetch_agents(&org).await?);
265        }
266
267        Ok(agents.into_iter().map(mapper::map_agent).collect())
268    }
269
270    async fn fetch_artifacts(&self, run_id: &str) -> PluginResult<Vec<BuildArtifact>> {
271        let parts: Vec<&str> = run_id.split("__").collect();
272        let [_, _, org, _, build_id] = parts.as_slice() else {
273            return Err(PluginError::InvalidConfig(format!(
274                "Invalid run ID: {run_id}"
275            )));
276        };
277
278        let client = self.client()?;
279        let artifacts = client.fetch_artifacts(org, build_id).await?;
280
281        Ok(artifacts
282            .into_iter()
283            .map(|artifact| client::artifact_to_build_artifact(artifact, run_id))
284            .collect())
285    }
286
287    async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> PluginResult<()> {
288        let parts: Vec<&str> = pipeline_id.split("__").collect();
289        if parts.len() != 4 {
290            return Err(PluginError::InvalidConfig(format!(
291                "Invalid pipeline ID format: {pipeline_id}"
292            )));
293        }
294
295        let org = parts[2];
296        let slug = parts[3];
297
298        let client = self.client()?;
299        client.cancel_build(org, slug, run_number).await
300    }
301}
pipedash_plugin_buildkite/plugin.rs

pipedash_plugin_buildkite/
plugin.rs
