pipedash_plugin_bitbucket/

plugin.rs

use std::collections::HashMap;

use pipedash_plugin_api::support::{
    resume_position,
    string_parameter,
};
use pipedash_plugin_api::*;

use crate::log_dialect::BitbucketLogDialect;
use crate::{
    client,
    config,
    mapper,
    metadata,
    types,
};

mod catalog;

#[derive(serde::Serialize, serde::Deserialize)]
struct LogPosition {
    pipeline: String,
    step: String,
    offset: u64,
}

plugin_struct!(
    BitbucketPlugin,
    client::BitbucketClient,
    metadata::create_metadata();
    status_source: Option<StatusSource> = None
);

plugin_impl!(BitbucketPlugin [
    metadata,
    status_source,
    fetch_organizations,
    ids_with_ci,
    fetch_available_pipelines,
    fetch_available_pipelines_filtered,
    fetch_pipelines,
] {
    fn initialize(
        &mut self, _provider_id: i64, config: HashMap<String, String>,
        http_client: Option<std::sync::Arc<reqwest::Client>>,
    ) -> PluginResult<()> {
        let (email, api_token) = config::get_auth(&config)?;
        let api_url = config::get_api_url();

        let credentials = format!("{}:{}", email, api_token);
        let encoded = base64::Engine::encode(
            &base64::engine::general_purpose::STANDARD,
            credentials.as_bytes(),
        );
        let auth_value = format!("Basic {}", encoded);

        let client = require_client(http_client)?;

        self.status_source = Some(config::status_source()?);
        self.client = Some(client::BitbucketClient::new(client, api_url, auth_value));
        self.config = config;

        Ok(())
    }

    async fn validate_credentials(&self) -> PluginResult<bool> {
        let client = self.client()?;
        client.get_user().await?;
        Ok(true)
    }

    async fn fetch_run_history(
        &self, pipeline_id: &str, limit: usize,
    ) -> PluginResult<Vec<PipelineRun>> {
        let (workspace, repo_slug) = config::parse_pipeline_id(pipeline_id)?;
        let client = self.client()?;

        let pipelines = client.list_pipelines(&workspace, &repo_slug, limit).await?;

        Ok(pipelines
            .iter()
            .map(|p| mapper::map_pipeline_run(p, &workspace, &repo_slug))
            .collect())
    }

    async fn fetch_run_details(
        &self, pipeline_id: &str, run_number: i64,
    ) -> PluginResult<PipelineRun> {
        let (workspace, repo_slug) = config::parse_pipeline_id(pipeline_id)?;

        let pipeline = self
            .client()?
            .find_pipeline_by_build_number(&workspace, &repo_slug, run_number)
            .await?;

        Ok(mapper::map_pipeline_run(&pipeline, &workspace, &repo_slug))
    }

    async fn fetch_log_nodes(&self, run: RunRef<'_>) -> PluginResult<Vec<LogNode>> {
        let (workspace, repo_slug) = config::parse_pipeline_id(run.pipeline_id)?;
        let client = self.client()?;

        let pipeline = client
            .find_pipeline_by_build_number(&workspace, &repo_slug, run.run_number)
            .await?;
        let steps = client
            .list_steps(&workspace, &repo_slug, &pipeline.uuid)
            .await?;

        Ok(mapper::steps_to_log_nodes(&steps))
    }

    async fn read_log(
        &self, run: RunRef<'_>, node_id: &str, from: Option<&LogCursor>, sink: &mut LogSink,
    ) -> PluginResult<LogRead> {
        let (workspace, repo_slug) = config::parse_pipeline_id(run.pipeline_id)?;
        config::check_step_uuid(node_id)?;
        let client = self.client()?;

        let pipeline = client
            .find_pipeline_by_build_number(&workspace, &repo_slug, run.run_number)
            .await?;

        let position = resume_position::<LogPosition>(from, sink, |position| {
            position.pipeline == pipeline.uuid && position.step == node_id
        })?;
        let (mode, offset) = client
            .stream_step_log(
                &workspace,
                &repo_slug,
                &pipeline.uuid,
                node_id,
                position.map(|p| p.offset),
                sink,
            )
            .await?;
        Ok(LogRead {
            mode,
            cursor: Some(LogCursor::new(&LogPosition {
                pipeline: pipeline.uuid,
                step: node_id.to_string(),
                offset,
            })?),
        })
    }

    fn log_dialect(&self) -> &'static dyn LogDialect {
        &BitbucketLogDialect
    }

    async fn fetch_workflow_parameters(
        &self, _workflow_id: &str,
    ) -> PluginResult<Vec<WorkflowParameter>> {
        Ok(vec![
            string_parameter(
                "ref",
                "Branch",
                "Branch name to run pipeline on",
                Some("main".to_string()),
                true,
            ),
            string_parameter(
                "custom_pipeline",
                "Custom Pipeline",
                "Name of custom pipeline to run (optional)",
                None,
                false,
            ),
        ])
    }

    async fn trigger_pipeline(&self, params: TriggerParams) -> PluginResult<TriggerResult> {
        let (workspace, repo_slug) = config::parse_pipeline_id(&params.workflow_id)?;
        let client = self.client()?;

        let ref_name = params
            .inputs
            .as_ref()
            .and_then(|i| i.get("ref"))
            .and_then(|v| v.as_str())
            .unwrap_or("main")
            .to_string();

        let custom_pipeline = params
            .inputs
            .as_ref()
            .and_then(|i| i.get("custom_pipeline"))
            .and_then(|v| v.as_str())
            .filter(|s| !s.is_empty());

        let request = types::TriggerPipelineRequest {
            target: types::TriggerTarget {
                target_type: "pipeline_ref_target".to_string(),
                ref_name,
                ref_type: "branch".to_string(),
                selector: custom_pipeline.map(|name| types::TriggerSelector {
                    selector_type: "custom".to_string(),
                    pattern: Some(name.to_string()),
                }),
            },
        };

        client
            .trigger_pipeline(&workspace, &repo_slug, request)
            .await?;

        Ok(TriggerResult::default())
    }

    async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> PluginResult<()> {
        let (workspace, repo_slug) = config::parse_pipeline_id(pipeline_id)?;
        let client = self.client()?;

        let pipeline = client
            .find_pipeline_by_build_number(&workspace, &repo_slug, run_number)
            .await?;

        let state = pipeline.state.name.as_str();
        if state == "PAUSED" || state == "HALTED" {
            return Err(PluginError::ApiError(format!(
                "Cannot cancel pipeline in '{}' state. Pipelines waiting for manual approval \
                 must be cancelled directly in Bitbucket UI.",
                state
            )));
        }

        if state == "COMPLETED" {
            return Err(PluginError::ApiError(
                "Cannot cancel a completed pipeline.".to_string(),
            ));
        }

        if state == "IN_PROGRESS" {
            let pipeline_is_paused = pipeline
                .state
                .stage
                .as_ref()
                .map(|s| s.name == "PAUSED")
                .unwrap_or(false);

            if pipeline_is_paused {
                return Err(PluginError::ApiError(
                    "Cannot cancel pipeline waiting for manual approval. \
                     Please cancel directly in Bitbucket UI."
                        .to_string(),
                ));
            }

            let steps = client
                .list_steps(&workspace, &repo_slug, &pipeline.uuid)
                .await?;

            let has_paused_step = steps.iter().any(|step| {
                step.state.name == "PENDING"
                    && step
                        .state
                        .stage
                        .as_ref()
                        .map(|s| s.name == "PAUSED")
                        .unwrap_or(false)
            });

            if has_paused_step {
                return Err(PluginError::ApiError(
                    "Cannot cancel pipeline with steps waiting for manual approval. \
                     Please cancel directly in Bitbucket UI."
                        .to_string(),
                ));
            }
        }

        client
            .stop_pipeline(&workspace, &repo_slug, &pipeline.uuid)
            .await
    }


});

#[cfg(test)]
mod tests;