pipedash_core/infrastructure/providers/
plugin_adapter.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use pipedash_plugin_api::Plugin as PluginTrait;
5
6use crate::domain::{
7    DomainError,
8    DomainResult,
9    Pipeline,
10    PipelineRun,
11    Provider,
12    TriggerParams,
13};
14
15pub struct PluginAdapter {
16    plugin: Arc<dyn PluginTrait>,
17    provider_type: String,
18    provider_id: i64,
19}
20
21impl PluginAdapter {
22    pub fn new(plugin: Box<dyn PluginTrait>, provider_type: String, provider_id: i64) -> Self {
23        Self {
24            plugin: Arc::from(plugin),
25            provider_type,
26            provider_id,
27        }
28    }
29
30    fn map_error(e: pipedash_plugin_api::PluginError) -> DomainError {
31        match e {
32            pipedash_plugin_api::PluginError::AuthenticationFailed(msg) => {
33                DomainError::AuthenticationFailed(msg)
34            }
35            pipedash_plugin_api::PluginError::ApiError(msg) => DomainError::ApiError(msg),
36            pipedash_plugin_api::PluginError::InvalidConfig(msg) => DomainError::InvalidConfig(msg),
37            pipedash_plugin_api::PluginError::PipelineNotFound(msg) => {
38                DomainError::PipelineNotFound(msg)
39            }
40            pipedash_plugin_api::PluginError::ProviderNotSupported(msg) => {
41                DomainError::InvalidProviderType(msg)
42            }
43            pipedash_plugin_api::PluginError::NetworkError(msg) => DomainError::NetworkError(msg),
44            pipedash_plugin_api::PluginError::SerializationError(msg) => {
45                DomainError::ApiError(format!("Serialization error: {}", msg))
46            }
47            pipedash_plugin_api::PluginError::DatabaseError(msg) => DomainError::DatabaseError(msg),
48            pipedash_plugin_api::PluginError::NotSupported(msg) => DomainError::NotSupported(msg),
49            pipedash_plugin_api::PluginError::Internal(msg) => DomainError::InternalError(msg),
50        }
51    }
52
53    fn convert_pipeline(
54        plugin_pipeline: pipedash_plugin_api::Pipeline, provider_id: i64, provider_type: &str,
55    ) -> Pipeline {
56        Pipeline {
57            id: plugin_pipeline.id,
58            provider_id,
59            provider_type: provider_type.to_string(),
60            name: plugin_pipeline.name,
61            status: Self::convert_status(plugin_pipeline.status),
62            last_run: plugin_pipeline.last_run,
63            last_updated: plugin_pipeline.last_updated,
64            repository: plugin_pipeline.repository,
65            branch: plugin_pipeline.branch,
66            workflow_file: plugin_pipeline.workflow_file,
67            metadata: plugin_pipeline.metadata,
68        }
69    }
70
71    fn convert_status(
72        status: pipedash_plugin_api::PipelineStatus,
73    ) -> crate::domain::PipelineStatus {
74        match status {
75            pipedash_plugin_api::PipelineStatus::Success => crate::domain::PipelineStatus::Success,
76            pipedash_plugin_api::PipelineStatus::Failed => crate::domain::PipelineStatus::Failed,
77            pipedash_plugin_api::PipelineStatus::Running => crate::domain::PipelineStatus::Running,
78            pipedash_plugin_api::PipelineStatus::Pending => crate::domain::PipelineStatus::Pending,
79            pipedash_plugin_api::PipelineStatus::Cancelled => {
80                crate::domain::PipelineStatus::Cancelled
81            }
82            pipedash_plugin_api::PipelineStatus::Skipped => crate::domain::PipelineStatus::Skipped,
83        }
84    }
85
86    fn convert_run(plugin_run: pipedash_plugin_api::PipelineRun) -> PipelineRun {
87        PipelineRun {
88            id: plugin_run.id,
89            pipeline_id: plugin_run.pipeline_id,
90            run_number: plugin_run.run_number,
91            status: Self::convert_status(plugin_run.status),
92            started_at: plugin_run.started_at,
93            concluded_at: plugin_run.concluded_at,
94            duration_seconds: plugin_run.duration_seconds,
95            logs_url: plugin_run.logs_url,
96            commit_sha: plugin_run.commit_sha,
97            commit_message: plugin_run.commit_message,
98            branch: plugin_run.branch,
99            actor: plugin_run.actor,
100            inputs: plugin_run.inputs,
101            metadata: plugin_run.metadata,
102        }
103    }
104}
105
106#[async_trait]
107impl Provider for PluginAdapter {
108    async fn fetch_pipelines(&self) -> DomainResult<Vec<Pipeline>> {
109        let pipelines = self
110            .plugin
111            .fetch_pipelines()
112            .await
113            .map_err(Self::map_error)?;
114
115        Ok(pipelines
116            .into_iter()
117            .map(|p| Self::convert_pipeline(p, self.provider_id, &self.provider_type))
118            .collect())
119    }
120
121    async fn fetch_run_history(
122        &self, pipeline_id: &str, limit: usize,
123    ) -> DomainResult<Vec<PipelineRun>> {
124        let runs = self
125            .plugin
126            .fetch_run_history(pipeline_id, limit)
127            .await
128            .map_err(Self::map_error)?;
129
130        Ok(runs.into_iter().map(Self::convert_run).collect())
131    }
132
133    async fn fetch_run_details(
134        &self, pipeline_id: &str, run_number: i64,
135    ) -> DomainResult<PipelineRun> {
136        let run = self
137            .plugin
138            .fetch_run_details(pipeline_id, run_number)
139            .await
140            .map_err(Self::map_error)?;
141
142        Ok(Self::convert_run(run))
143    }
144
145    async fn trigger_pipeline(&self, params: TriggerParams) -> DomainResult<String> {
146        let trigger_params = pipedash_plugin_api::TriggerParams {
147            workflow_id: params.workflow_id,
148            inputs: params.inputs,
149        };
150
151        self.plugin
152            .trigger_pipeline(trigger_params)
153            .await
154            .map_err(Self::map_error)
155    }
156
157    async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> DomainResult<()> {
158        self.plugin
159            .cancel_run(pipeline_id, run_number)
160            .await
161            .map_err(Self::map_error)
162    }
163
164    async fn get_workflow_parameters(
165        &self, workflow_id: &str,
166    ) -> DomainResult<Vec<pipedash_plugin_api::WorkflowParameter>> {
167        self.plugin
168            .fetch_workflow_parameters(workflow_id)
169            .await
170            .map_err(Self::map_error)
171    }
172
173    async fn validate_credentials(&self) -> DomainResult<bool> {
174        self.plugin
175            .validate_credentials()
176            .await
177            .map_err(Self::map_error)
178    }
179
180    fn provider_type(&self) -> &str {
181        &self.provider_type
182    }
183}
pipedash_core/infrastructure/providers/plugin_adapter.rs

pipedash_core/infrastructure/providers/
plugin_adapter.rs
