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