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

pipedash_core/infrastructure/providers/
plugin_adapter.rs
