pipedash_plugin_bitbucket/
plugin.rs1use std::collections::HashMap;
2use std::time::Duration;
3
4use async_trait::async_trait;
5use futures::future::join_all;
6use pipedash_plugin_api::*;
7
8use crate::{
9 client,
10 config,
11 mapper,
12 metadata,
13 types,
14};
15
16pub struct BitbucketPlugin {
17 metadata: PluginMetadata,
18 client: Option<client::BitbucketClient>,
19 provider_id: Option<i64>,
20 config: HashMap<String, String>,
21 status_source: Option<StatusSource>,
22}
23
24impl Default for BitbucketPlugin {
25 fn default() -> Self {
26 Self::new()
27 }
28}
29
30impl BitbucketPlugin {
31 pub fn new() -> Self {
32 Self {
33 metadata: metadata::create_metadata(),
34 client: None,
35 provider_id: None,
36 config: HashMap::new(),
37 status_source: None,
38 }
39 }
40
41 fn client(&self) -> PluginResult<&client::BitbucketClient> {
42 self.client
43 .as_ref()
44 .ok_or_else(|| PluginError::Internal("Plugin not initialized".to_string()))
45 }
46
47 async fn fetch_all_repositories(&self) -> PluginResult<Vec<types::Repository>> {
48 let client = self.client()?;
49 let mut all_repos = Vec::new();
50 let mut page = 1;
51
52 const MAX_PAGES: usize = 50;
53
54 loop {
55 let params = PaginationParams {
56 page,
57 page_size: 100,
58 };
59 let response = client.list_all_repositories(¶ms).await?;
60
61 if response.items.is_empty() {
62 break;
63 }
64
65 all_repos.extend(response.items);
66
67 if !response.has_more || page >= MAX_PAGES {
68 break;
69 }
70 page += 1;
71 }
72
73 if let Some(selected) = config::parse_selected_items(&self.config) {
74 Ok(all_repos
75 .into_iter()
76 .filter(|r| selected.contains(&r.full_name))
77 .collect())
78 } else {
79 Ok(all_repos)
80 }
81 }
82
83 async fn find_pipeline_by_build_number(
84 &self, workspace: &str, repo_slug: &str, build_number: i64,
85 ) -> PluginResult<types::Pipeline> {
86 let client = self.client()?;
87 let pipelines = client.list_pipelines(workspace, repo_slug, 100).await?;
88
89 pipelines
90 .into_iter()
91 .find(|p| p.build_number == build_number)
92 .ok_or_else(|| {
93 PluginError::PipelineNotFound(format!(
94 "Pipeline run #{} not found in recent 100 runs for {}/{}",
95 build_number, workspace, repo_slug
96 ))
97 })
98 }
99}
100
101fn collect_partial<T>(results: Vec<PluginResult<T>>, item: &str) -> PluginResult<Vec<T>> {
102 let mut kept = Vec::with_capacity(results.len());
103 for result in results {
104 match result {
105 Ok(value) => kept.push(value),
106 Err(e) if e.is_transient() => return Err(e),
107 Err(e) => tracing::warn!(error = %e, item, "Skipping failed Bitbucket item"),
108 }
109 }
110 Ok(kept)
111}
112
113#[async_trait]
114impl Plugin for BitbucketPlugin {
115 fn metadata(&self) -> &PluginMetadata {
116 &self.metadata
117 }
118
119 fn provider_type(&self) -> &str {
120 "bitbucket"
121 }
122
123 fn status_source(&self) -> Option<&StatusSource> {
124 self.status_source.as_ref()
125 }
126
127 fn initialize(
128 &mut self, provider_id: i64, config: HashMap<String, String>,
129 http_client: Option<std::sync::Arc<reqwest::Client>>,
130 ) -> PluginResult<()> {
131 let (email, api_token) = config::get_auth(&config)?;
132 let api_url = config::get_api_url();
133
134 let credentials = format!("{}:{}", email, api_token);
135 let encoded = base64::Engine::encode(
136 &base64::engine::general_purpose::STANDARD,
137 credentials.as_bytes(),
138 );
139 let auth_value = format!("Basic {}", encoded);
140
141 let client = http_client.unwrap_or_else(|| {
142 std::sync::Arc::new(
143 reqwest::Client::builder()
144 .use_rustls_tls()
145 .pool_max_idle_per_host(10)
146 .timeout(Duration::from_secs(30))
147 .connect_timeout(Duration::from_secs(10))
148 .tcp_keepalive(Duration::from_secs(60))
149 .build()
150 .expect("Failed to build HTTP client"),
151 )
152 });
153
154 self.status_source = Some(config::status_source()?);
155 self.client = Some(client::BitbucketClient::new(client, api_url, auth_value));
156 self.provider_id = Some(provider_id);
157 self.config = config;
158
159 Ok(())
160 }
161
162 async fn validate_credentials(&self) -> PluginResult<bool> {
163 let client = self.client()?;
164 client.get_user().await?;
165 Ok(true)
166 }
167
168 async fn fetch_organizations(&self) -> PluginResult<Vec<Organization>> {
169 let client = self.client()?;
170 let workspaces = client.list_workspaces().await?;
171
172 Ok(workspaces
173 .into_iter()
174 .map(|w| Organization {
175 id: w.uuid,
176 name: w.name,
177 description: Some(w.slug),
178 })
179 .collect())
180 }
181
182 async fn fetch_available_pipelines(
183 &self, params: Option<PaginationParams>,
184 ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
185 let client = self.client()?;
186 let params = params.unwrap_or_default();
187
188 let response = client.list_all_repositories(¶ms).await?;
189
190 let available = response
191 .items
192 .iter()
193 .map(mapper::map_available_pipeline)
194 .collect();
195
196 Ok(PaginatedResponse::new(
197 available,
198 response.page,
199 response.page_size,
200 response.total_count,
201 ))
202 }
203
204 async fn fetch_available_pipelines_filtered(
205 &self, org: Option<String>, search: Option<String>, params: Option<PaginationParams>,
206 ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
207 let client = self.client()?;
208 let params = params.unwrap_or_default();
209
210 let response = if let Some(workspace) = &org {
211 client.list_repositories(workspace, ¶ms).await?
212 } else {
213 client.list_all_repositories(¶ms).await?
214 };
215
216 let mut available: Vec<AvailablePipeline> = response
217 .items
218 .iter()
219 .map(mapper::map_available_pipeline)
220 .collect();
221
222 if let Some(search_term) = search {
223 let search_lower = search_term.to_lowercase();
224 available.retain(|p| {
225 p.name.to_lowercase().contains(&search_lower)
226 || p.id.to_lowercase().contains(&search_lower)
227 });
228 }
229
230 Ok(PaginatedResponse::new(
231 available,
232 params.page,
233 params.page_size,
234 response.total_count,
235 ))
236 }
237
238 async fn fetch_pipelines(&self) -> PluginResult<Vec<Pipeline>> {
239 let provider_id = self
240 .provider_id
241 .ok_or_else(|| PluginError::Internal("Provider ID not set".to_string()))?;
242
243 let client = self.client()?;
244 let repos = self.fetch_all_repositories().await?;
245
246 let pipeline_futures = repos.iter().map(|repo| async move {
247 let pipelines = client
248 .list_pipelines(&repo.workspace.slug, &repo.slug, 1)
249 .await?;
250 let latest = pipelines.first();
251 Ok(mapper::map_pipeline(repo, latest, provider_id))
252 });
253
254 collect_partial(join_all(pipeline_futures).await, "repository")
255 }
256
257 async fn fetch_run_history(
258 &self, pipeline_id: &str, limit: usize,
259 ) -> PluginResult<Vec<PipelineRun>> {
260 let (provider_id, workspace, repo_slug) = config::parse_pipeline_id(pipeline_id)?;
261 let client = self.client()?;
262
263 let pipelines = client.list_pipelines(&workspace, &repo_slug, limit).await?;
264
265 Ok(pipelines
266 .iter()
267 .map(|p| mapper::map_pipeline_run(p, &workspace, &repo_slug, provider_id))
268 .collect())
269 }
270
271 async fn fetch_run_details(
272 &self, pipeline_id: &str, run_number: i64,
273 ) -> PluginResult<PipelineRun> {
274 let (provider_id, workspace, repo_slug) = config::parse_pipeline_id(pipeline_id)?;
275
276 let pipeline = self
277 .find_pipeline_by_build_number(&workspace, &repo_slug, run_number)
278 .await?;
279
280 Ok(mapper::map_pipeline_run(
281 &pipeline,
282 &workspace,
283 &repo_slug,
284 provider_id,
285 ))
286 }
287
288 async fn fetch_workflow_parameters(
289 &self, _workflow_id: &str,
290 ) -> PluginResult<Vec<WorkflowParameter>> {
291 Ok(vec![
292 WorkflowParameter {
293 name: "ref".to_string(),
294 label: Some("Branch".to_string()),
295 description: Some("Branch name to run pipeline on".to_string()),
296 param_type: WorkflowParameterType::String {
297 default: Some("main".to_string()),
298 },
299 required: true,
300 },
301 WorkflowParameter {
302 name: "custom_pipeline".to_string(),
303 label: Some("Custom Pipeline".to_string()),
304 description: Some("Name of custom pipeline to run (optional)".to_string()),
305 param_type: WorkflowParameterType::String { default: None },
306 required: false,
307 },
308 ])
309 }
310
311 async fn trigger_pipeline(&self, params: TriggerParams) -> PluginResult<String> {
312 let (_, workspace, repo_slug) = config::parse_pipeline_id(¶ms.workflow_id)?;
313 let client = self.client()?;
314
315 let ref_name = params
316 .inputs
317 .as_ref()
318 .and_then(|i| i.get("ref"))
319 .and_then(|v| v.as_str())
320 .unwrap_or("main")
321 .to_string();
322
323 let custom_pipeline = params
324 .inputs
325 .as_ref()
326 .and_then(|i| i.get("custom_pipeline"))
327 .and_then(|v| v.as_str())
328 .filter(|s| !s.is_empty());
329
330 let request = types::TriggerPipelineRequest {
331 target: types::TriggerTarget {
332 target_type: "pipeline_ref_target".to_string(),
333 ref_name,
334 ref_type: "branch".to_string(),
335 selector: custom_pipeline.map(|name| types::TriggerSelector {
336 selector_type: "custom".to_string(),
337 pattern: Some(name.to_string()),
338 }),
339 },
340 };
341
342 let pipeline = client
343 .trigger_pipeline(&workspace, &repo_slug, request)
344 .await?;
345
346 let url = pipeline.links.html.map(|l| l.href).unwrap_or_else(|| {
347 format!(
348 "https://bitbucket.org/{}/{}/pipelines/results/{}",
349 workspace, repo_slug, pipeline.build_number
350 )
351 });
352
353 Ok(url)
354 }
355
356 async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> PluginResult<()> {
357 let (_, workspace, repo_slug) = config::parse_pipeline_id(pipeline_id)?;
358 let client = self.client()?;
359
360 let pipeline = self
361 .find_pipeline_by_build_number(&workspace, &repo_slug, run_number)
362 .await?;
363
364 let state = pipeline.state.name.as_str();
365 if state == "PAUSED" || state == "HALTED" {
366 return Err(PluginError::ApiError(format!(
367 "Cannot cancel pipeline in '{}' state. Pipelines waiting for manual approval \
368 must be cancelled directly in Bitbucket UI.",
369 state
370 )));
371 }
372
373 if state == "COMPLETED" {
374 return Err(PluginError::ApiError(
375 "Cannot cancel a completed pipeline.".to_string(),
376 ));
377 }
378
379 if state == "IN_PROGRESS" {
380 let pipeline_is_paused = pipeline
381 .state
382 .stage
383 .as_ref()
384 .map(|s| s.name == "PAUSED")
385 .unwrap_or(false);
386
387 if pipeline_is_paused {
388 return Err(PluginError::ApiError(
389 "Cannot cancel pipeline waiting for manual approval. \
390 Please cancel directly in Bitbucket UI."
391 .to_string(),
392 ));
393 }
394
395 let steps = client
396 .list_steps(&workspace, &repo_slug, &pipeline.uuid)
397 .await?;
398
399 let has_paused_step = steps.iter().any(|step| {
400 step.state.name == "PENDING"
401 && step
402 .state
403 .stage
404 .as_ref()
405 .map(|s| s.name == "PAUSED")
406 .unwrap_or(false)
407 });
408
409 if has_paused_step {
410 return Err(PluginError::ApiError(
411 "Cannot cancel pipeline with steps waiting for manual approval. \
412 Please cancel directly in Bitbucket UI."
413 .to_string(),
414 ));
415 }
416 }
417
418 client
419 .stop_pipeline(&workspace, &repo_slug, &pipeline.uuid)
420 .await
421 }
422
423 async fn fetch_agents(&self) -> PluginResult<Vec<BuildAgent>> {
424 Err(PluginError::NotSupported(
425 "Bitbucket Pipelines uses Atlassian infrastructure - no agent monitoring".to_string(),
426 ))
427 }
428
429 async fn fetch_artifacts(&self, _run_id: &str) -> PluginResult<Vec<BuildArtifact>> {
430 Err(PluginError::NotSupported(
431 "Artifact fetching not yet implemented for Bitbucket".to_string(),
432 ))
433 }
434
435 async fn fetch_queues(&self) -> PluginResult<Vec<BuildQueue>> {
436 Err(PluginError::NotSupported(
437 "Build queues not supported by Bitbucket Pipelines".to_string(),
438 ))
439 }
440
441 fn get_migrations(&self) -> Vec<String> {
442 vec![]
443 }
444}