pipedash_core/infrastructure/providers/
plugin_adapter.rs

1use std::sync::Arc;
2
3use async_trait::async_trait;
4use chrono::{
5    DateTime,
6    Utc,
7};
8use pipedash_plugin_api::{
9    Plugin as PluginTrait,
10    RunRef,
11};
12
13use crate::domain::{
14    DomainError,
15    DomainResult,
16    LogDialect,
17    LogNode,
18    LogSink,
19    Pipeline,
20    PipelineRun,
21    Provider,
22    TriggerParams,
23};
24
25pub struct PluginAdapter {
26    plugin: Arc<dyn PluginTrait>,
27    provider_type: String,
28    provider_id: i64,
29    renew_at: Option<DateTime<Utc>>,
30}
31
32impl PluginAdapter {
33    pub fn new(
34        plugin: Box<dyn PluginTrait>, provider_type: String, provider_id: i64,
35        renew_at: Option<DateTime<Utc>>,
36    ) -> Self {
37        Self {
38            plugin: Arc::from(plugin),
39            provider_type,
40            provider_id,
41            renew_at,
42        }
43    }
44
45    pub(crate) fn map_error(e: pipedash_plugin_api::PluginError) -> DomainError {
46        match e {
47            pipedash_plugin_api::PluginError::AuthenticationFailed(msg) => {
48                DomainError::ProviderAuthenticationFailed(msg)
49            }
50            pipedash_plugin_api::PluginError::PermissionDenied(msg) => {
51                DomainError::PermissionDenied(msg)
52            }
53            pipedash_plugin_api::PluginError::ApiError(msg) => DomainError::ApiError(msg),
54            pipedash_plugin_api::PluginError::InvalidConfig(msg) => DomainError::InvalidConfig(msg),
55            pipedash_plugin_api::PluginError::PipelineNotFound(msg) => {
56                DomainError::PipelineNotFound(msg)
57            }
58            pipedash_plugin_api::PluginError::ProviderNotSupported(msg) => {
59                DomainError::InvalidProviderType(msg)
60            }
61            pipedash_plugin_api::PluginError::NetworkError(msg) => DomainError::NetworkError(msg),
62            pipedash_plugin_api::PluginError::SerializationError(msg) => {
63                DomainError::ApiError(format!("Serialization error: {}", msg))
64            }
65            pipedash_plugin_api::PluginError::DatabaseError(msg) => DomainError::DatabaseError(msg),
66            pipedash_plugin_api::PluginError::NotSupported(msg) => DomainError::NotSupported(msg),
67            pipedash_plugin_api::PluginError::Internal(msg) => DomainError::InternalError(msg),
68            pipedash_plugin_api::PluginError::RateLimited {
69                message,
70                retry_after,
71            } => DomainError::RateLimited {
72                message,
73                retry_after,
74            },
75            pipedash_plugin_api::PluginError::ServiceUnavailable {
76                message,
77                retry_after,
78            } => DomainError::ServiceUnavailable {
79                message,
80                retry_after,
81            },
82            pipedash_plugin_api::PluginError::Tls(msg) => DomainError::Tls(msg),
83            e => DomainError::ApiError(e.to_string()),
84        }
85    }
86
87    fn convert_pipeline(
88        plugin_pipeline: pipedash_plugin_api::Pipeline, provider_id: i64, provider_type: &str,
89    ) -> Pipeline {
90        Pipeline {
91            id: plugin_pipeline.id,
92            provider_id,
93            provider_type: provider_type.to_string(),
94            name: plugin_pipeline.name,
95            status: Self::convert_status(plugin_pipeline.status),
96            last_run: plugin_pipeline.last_run,
97            last_updated: plugin_pipeline.last_updated,
98            repository: plugin_pipeline.repository,
99            branch: plugin_pipeline.branch,
100            workflow_file: plugin_pipeline.workflow_file,
101            metadata: plugin_pipeline.metadata,
102            overdue_run: None,
103        }
104    }
105
106    fn convert_status(
107        status: pipedash_plugin_api::PipelineStatus,
108    ) -> crate::domain::PipelineStatus {
109        match status {
110            pipedash_plugin_api::PipelineStatus::Success => crate::domain::PipelineStatus::Success,
111            pipedash_plugin_api::PipelineStatus::Failed => crate::domain::PipelineStatus::Failed,
112            pipedash_plugin_api::PipelineStatus::Running => crate::domain::PipelineStatus::Running,
113            pipedash_plugin_api::PipelineStatus::Pending => crate::domain::PipelineStatus::Pending,
114            pipedash_plugin_api::PipelineStatus::Cancelled => {
115                crate::domain::PipelineStatus::Cancelled
116            }
117            pipedash_plugin_api::PipelineStatus::Skipped => crate::domain::PipelineStatus::Skipped,
118        }
119    }
120
121    fn convert_run(plugin_run: pipedash_plugin_api::PipelineRun) -> PipelineRun {
122        PipelineRun {
123            id: plugin_run.id,
124            pipeline_id: plugin_run.pipeline_id,
125            run_number: plugin_run.run_number,
126            status: Self::convert_status(plugin_run.status),
127            started_at: plugin_run.started_at,
128            concluded_at: plugin_run.concluded_at,
129            duration_seconds: plugin_run.duration_seconds,
130            logs_url: plugin_run.logs_url,
131            commit_sha: plugin_run.commit_sha,
132            commit_message: plugin_run.commit_message,
133            branch: plugin_run.branch,
134            actor: plugin_run.actor,
135            inputs: plugin_run.inputs,
136            metadata: plugin_run.metadata,
137        }
138    }
139
140    fn run_ref(run: &PipelineRun) -> RunRef<'_> {
141        RunRef {
142            pipeline_id: &run.pipeline_id,
143            run_number: run.run_number,
144            run_id: &run.id,
145            metadata: &run.metadata,
146        }
147    }
148}
149
150#[async_trait]
151impl Provider for PluginAdapter {
152    async fn fetch_pipelines(&self) -> DomainResult<Vec<Pipeline>> {
153        let pipelines = self
154            .plugin
155            .fetch_pipelines()
156            .await
157            .map_err(Self::map_error)?;
158
159        Ok(pipelines
160            .into_iter()
161            .map(|p| Self::convert_pipeline(p, self.provider_id, &self.provider_type))
162            .collect())
163    }
164
165    async fn fetch_run_history(
166        &self, pipeline_id: &str, limit: usize,
167    ) -> DomainResult<Vec<PipelineRun>> {
168        let runs = self
169            .plugin
170            .fetch_run_history(pipeline_id, limit)
171            .await
172            .map_err(Self::map_error)?;
173
174        Ok(runs.into_iter().map(Self::convert_run).collect())
175    }
176
177    async fn fetch_run_details(
178        &self, pipeline_id: &str, run_number: i64,
179    ) -> DomainResult<PipelineRun> {
180        let run = self
181            .plugin
182            .fetch_run_details(pipeline_id, run_number)
183            .await
184            .map_err(Self::map_error)?;
185
186        Ok(Self::convert_run(run))
187    }
188
189    async fn trigger_pipeline(&self, params: TriggerParams) -> DomainResult<String> {
190        let trigger_params = pipedash_plugin_api::TriggerParams {
191            workflow_id: params.workflow_id,
192            inputs: params.inputs,
193        };
194
195        self.plugin
196            .trigger_pipeline(trigger_params)
197            .await
198            .map_err(Self::map_error)
199    }
200
201    async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> DomainResult<()> {
202        self.plugin
203            .cancel_run(pipeline_id, run_number)
204            .await
205            .map_err(Self::map_error)
206    }
207
208    async fn review_run(
209        &self, pipeline_id: &str, run_number: i64, decision: pipedash_plugin_api::ReviewDecision,
210    ) -> DomainResult<()> {
211        self.plugin
212            .review_run(pipeline_id, run_number, decision)
213            .await
214            .map_err(Self::map_error)
215    }
216
217    async fn rerun_failed_jobs(&self, pipeline_id: &str, run_number: i64) -> DomainResult<()> {
218        self.plugin
219            .rerun_failed_jobs(pipeline_id, run_number)
220            .await
221            .map_err(Self::map_error)
222    }
223
224    async fn rerun_run(&self, pipeline_id: &str, run_number: i64) -> DomainResult<()> {
225        self.plugin
226            .rerun_run(pipeline_id, run_number)
227            .await
228            .map_err(Self::map_error)
229    }
230
231    async fn get_workflow_parameters(
232        &self, workflow_id: &str,
233    ) -> DomainResult<Vec<pipedash_plugin_api::WorkflowParameter>> {
234        self.plugin
235            .fetch_workflow_parameters(workflow_id)
236            .await
237            .map_err(Self::map_error)
238    }
239
240    async fn fetch_rate_limit(&self) -> DomainResult<Option<pipedash_plugin_api::RateLimit>> {
241        self.plugin
242            .fetch_rate_limit()
243            .await
244            .map_err(Self::map_error)
245    }
246
247    async fn validate_credentials(&self) -> DomainResult<bool> {
248        self.plugin
249            .validate_credentials()
250            .await
251            .map_err(Self::map_error)
252    }
253
254    fn provider_type(&self) -> &str {
255        &self.provider_type
256    }
257
258    fn capabilities(&self) -> &pipedash_plugin_api::PluginCapabilities {
259        &self.plugin.metadata().capabilities
260    }
261
262    fn renew_at(&self) -> Option<DateTime<Utc>> {
263        self.renew_at
264    }
265
266    fn status_source(&self) -> Option<&pipedash_plugin_api::StatusSource> {
267        self.plugin.status_source()
268    }
269
270    async fn check_health(&self) -> DomainResult<Vec<pipedash_plugin_api::ComponentHealth>> {
271        self.plugin.check_health().await.map_err(Self::map_error)
272    }
273
274    fn log_dialect(&self) -> &'static dyn LogDialect {
275        self.plugin.log_dialect()
276    }
277
278    async fn fetch_log_nodes(&self, run: &PipelineRun) -> DomainResult<Vec<LogNode>> {
279        let nodes = self
280            .plugin
281            .fetch_log_nodes(Self::run_ref(run))
282            .await
283            .map_err(Self::map_error)?;
284        Ok(nodes
285            .into_iter()
286            .map(|n| LogNode {
287                id: n.id,
288                parent_id: n.parent_id,
289                kind: n.kind,
290                name: n.name,
291                status: Self::convert_status(n.status),
292                started_at: n.started_at,
293                finished_at: n.finished_at,
294                log: n.log,
295                web_url: n.web_url,
296            })
297            .collect())
298    }
299
300    async fn read_log(
301        &self, run: &PipelineRun, node_id: &str, sink: &mut LogSink,
302    ) -> DomainResult<()> {
303        self.plugin
304            .read_log(Self::run_ref(run), node_id, sink)
305            .await
306            .map_err(Self::map_error)
307    }
308}
pipedash_core/infrastructure/providers/plugin_adapter.rs

pipedash_core/infrastructure/providers/
plugin_adapter.rs
