pipedash_core/infrastructure/providers/
plugin_adapter.rs1use 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 pub(crate) fn map_error(e: pipedash_plugin_api::PluginError) -> DomainError {
31 match e {
32 pipedash_plugin_api::PluginError::AuthenticationFailed(msg) => {
33 DomainError::ProviderAuthenticationFailed(msg)
34 }
35 pipedash_plugin_api::PluginError::PermissionDenied(msg) => {
36 DomainError::PermissionDenied(msg)
37 }
38 pipedash_plugin_api::PluginError::ApiError(msg) => DomainError::ApiError(msg),
39 pipedash_plugin_api::PluginError::InvalidConfig(msg) => DomainError::InvalidConfig(msg),
40 pipedash_plugin_api::PluginError::PipelineNotFound(msg) => {
41 DomainError::PipelineNotFound(msg)
42 }
43 pipedash_plugin_api::PluginError::ProviderNotSupported(msg) => {
44 DomainError::InvalidProviderType(msg)
45 }
46 pipedash_plugin_api::PluginError::NetworkError(msg) => DomainError::NetworkError(msg),
47 pipedash_plugin_api::PluginError::SerializationError(msg) => {
48 DomainError::ApiError(format!("Serialization error: {}", msg))
49 }
50 pipedash_plugin_api::PluginError::DatabaseError(msg) => DomainError::DatabaseError(msg),
51 pipedash_plugin_api::PluginError::NotSupported(msg) => DomainError::NotSupported(msg),
52 pipedash_plugin_api::PluginError::Internal(msg) => DomainError::InternalError(msg),
53 pipedash_plugin_api::PluginError::RateLimited {
54 message,
55 retry_after,
56 } => DomainError::RateLimited {
57 message,
58 retry_after,
59 },
60 pipedash_plugin_api::PluginError::ServiceUnavailable {
61 message,
62 retry_after,
63 } => DomainError::ServiceUnavailable {
64 message,
65 retry_after,
66 },
67 pipedash_plugin_api::PluginError::Tls(msg) => DomainError::Tls(msg),
68 e => DomainError::ApiError(e.to_string()),
69 }
70 }
71
72 fn convert_pipeline(
73 plugin_pipeline: pipedash_plugin_api::Pipeline, provider_id: i64, provider_type: &str,
74 ) -> Pipeline {
75 Pipeline {
76 id: plugin_pipeline.id,
77 provider_id,
78 provider_type: provider_type.to_string(),
79 name: plugin_pipeline.name,
80 status: Self::convert_status(plugin_pipeline.status),
81 last_run: plugin_pipeline.last_run,
82 last_updated: plugin_pipeline.last_updated,
83 repository: plugin_pipeline.repository,
84 branch: plugin_pipeline.branch,
85 workflow_file: plugin_pipeline.workflow_file,
86 metadata: plugin_pipeline.metadata,
87 overdue_run: None,
88 }
89 }
90
91 fn convert_status(
92 status: pipedash_plugin_api::PipelineStatus,
93 ) -> crate::domain::PipelineStatus {
94 match status {
95 pipedash_plugin_api::PipelineStatus::Success => crate::domain::PipelineStatus::Success,
96 pipedash_plugin_api::PipelineStatus::Failed => crate::domain::PipelineStatus::Failed,
97 pipedash_plugin_api::PipelineStatus::Running => crate::domain::PipelineStatus::Running,
98 pipedash_plugin_api::PipelineStatus::Pending => crate::domain::PipelineStatus::Pending,
99 pipedash_plugin_api::PipelineStatus::Cancelled => {
100 crate::domain::PipelineStatus::Cancelled
101 }
102 pipedash_plugin_api::PipelineStatus::Skipped => crate::domain::PipelineStatus::Skipped,
103 }
104 }
105
106 fn convert_run(plugin_run: pipedash_plugin_api::PipelineRun) -> PipelineRun {
107 PipelineRun {
108 id: plugin_run.id,
109 pipeline_id: plugin_run.pipeline_id,
110 run_number: plugin_run.run_number,
111 status: Self::convert_status(plugin_run.status),
112 started_at: plugin_run.started_at,
113 concluded_at: plugin_run.concluded_at,
114 duration_seconds: plugin_run.duration_seconds,
115 logs_url: plugin_run.logs_url,
116 commit_sha: plugin_run.commit_sha,
117 commit_message: plugin_run.commit_message,
118 branch: plugin_run.branch,
119 actor: plugin_run.actor,
120 inputs: plugin_run.inputs,
121 metadata: plugin_run.metadata,
122 }
123 }
124}
125
126#[async_trait]
127impl Provider for PluginAdapter {
128 async fn fetch_pipelines(&self) -> DomainResult<Vec<Pipeline>> {
129 let pipelines = self
130 .plugin
131 .fetch_pipelines()
132 .await
133 .map_err(Self::map_error)?;
134
135 Ok(pipelines
136 .into_iter()
137 .map(|p| Self::convert_pipeline(p, self.provider_id, &self.provider_type))
138 .collect())
139 }
140
141 async fn fetch_run_history(
142 &self, pipeline_id: &str, limit: usize,
143 ) -> DomainResult<Vec<PipelineRun>> {
144 let runs = self
145 .plugin
146 .fetch_run_history(pipeline_id, limit)
147 .await
148 .map_err(Self::map_error)?;
149
150 Ok(runs.into_iter().map(Self::convert_run).collect())
151 }
152
153 async fn fetch_run_details(
154 &self, pipeline_id: &str, run_number: i64,
155 ) -> DomainResult<PipelineRun> {
156 let run = self
157 .plugin
158 .fetch_run_details(pipeline_id, run_number)
159 .await
160 .map_err(Self::map_error)?;
161
162 Ok(Self::convert_run(run))
163 }
164
165 async fn trigger_pipeline(&self, params: TriggerParams) -> DomainResult<String> {
166 let trigger_params = pipedash_plugin_api::TriggerParams {
167 workflow_id: params.workflow_id,
168 inputs: params.inputs,
169 };
170
171 self.plugin
172 .trigger_pipeline(trigger_params)
173 .await
174 .map_err(Self::map_error)
175 }
176
177 async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> DomainResult<()> {
178 self.plugin
179 .cancel_run(pipeline_id, run_number)
180 .await
181 .map_err(Self::map_error)
182 }
183
184 async fn review_run(
185 &self, pipeline_id: &str, run_number: i64, decision: pipedash_plugin_api::ReviewDecision,
186 ) -> DomainResult<()> {
187 self.plugin
188 .review_run(pipeline_id, run_number, decision)
189 .await
190 .map_err(Self::map_error)
191 }
192
193 async fn get_workflow_parameters(
194 &self, workflow_id: &str,
195 ) -> DomainResult<Vec<pipedash_plugin_api::WorkflowParameter>> {
196 self.plugin
197 .fetch_workflow_parameters(workflow_id)
198 .await
199 .map_err(Self::map_error)
200 }
201
202 async fn fetch_rate_limit(&self) -> DomainResult<Option<pipedash_plugin_api::RateLimit>> {
203 self.plugin
204 .fetch_rate_limit()
205 .await
206 .map_err(Self::map_error)
207 }
208
209 async fn validate_credentials(&self) -> DomainResult<bool> {
210 self.plugin
211 .validate_credentials()
212 .await
213 .map_err(Self::map_error)
214 }
215
216 fn provider_type(&self) -> &str {
217 &self.provider_type
218 }
219
220 fn capabilities(&self) -> &pipedash_plugin_api::PluginCapabilities {
221 &self.plugin.metadata().capabilities
222 }
223
224 fn status_source(&self) -> Option<&pipedash_plugin_api::StatusSource> {
225 self.plugin.status_source()
226 }
227
228 async fn check_health(&self) -> DomainResult<Vec<pipedash_plugin_api::ComponentHealth>> {
229 self.plugin.check_health().await.map_err(Self::map_error)
230 }
231}