pipedash_plugin_gitlab/
plugin.rs

1use std::collections::HashMap;
2use std::time::Duration;
3
4use async_trait::async_trait;
5use futures::future::join_all;
6use pipedash_plugin_api::*;
7
8use crate::{
9    client,
10    config,
11    mapper,
12    metadata,
13    types,
14};
15
16pub struct GitLabPlugin {
17    metadata: PluginMetadata,
18    client: Option<client::GitLabClient>,
19    provider_id: Option<i64>,
20    config: HashMap<String, String>,
21}
22
23impl Default for GitLabPlugin {
24    fn default() -> Self {
25        Self::new()
26    }
27}
28
29impl GitLabPlugin {
30    pub fn new() -> Self {
31        Self {
32            metadata: metadata::create_metadata(),
33            client: None,
34            provider_id: None,
35            config: HashMap::new(),
36        }
37    }
38
39    fn client(&self) -> PluginResult<&client::GitLabClient> {
40        self.client
41            .as_ref()
42            .ok_or_else(|| PluginError::Internal("Plugin not initialized".to_string()))
43    }
44
45    async fn fetch_all_projects(&self) -> PluginResult<Vec<types::Project>> {
46        let client = self.client()?;
47        let mut all_projects = Vec::new();
48        let mut page = 1;
49
50        loop {
51            let params = PaginationParams {
52                page,
53                page_size: 100,
54            };
55            let response = client.list_projects(&params).await?;
56            if response.items.is_empty() {
57                break;
58            }
59            all_projects.extend(response.items);
60
61            if !response.has_more || page > 100 {
62                break;
63            }
64            page += 1;
65        }
66
67        if let Some(selected_paths) = config::parse_selected_items(&self.config) {
68            Ok(all_projects
69                .into_iter()
70                .filter(|p| {
71                    let normalized = p.name_with_namespace.replace(" ", "");
72                    selected_paths.contains(&normalized)
73                })
74                .collect())
75        } else {
76            Ok(all_projects)
77        }
78    }
79
80    async fn fetch_projects_paginated(
81        &self, params: &PaginationParams,
82    ) -> PluginResult<PaginatedResponse<types::Project>> {
83        let client = self.client()?;
84        let response = client.list_projects(params).await?;
85
86        if let Some(selected_paths) = config::parse_selected_items(&self.config) {
87            let filtered_items: Vec<_> = response
88                .items
89                .into_iter()
90                .filter(|p| {
91                    let normalized = p.name_with_namespace.replace(" ", "");
92                    selected_paths.contains(&normalized)
93                })
94                .collect();
95            let filtered_count = filtered_items.len();
96            Ok(PaginatedResponse::new(
97                filtered_items,
98                params.page,
99                params.page_size,
100                filtered_count,
101            ))
102        } else {
103            Ok(response)
104        }
105    }
106}
107
108#[async_trait]
109impl Plugin for GitLabPlugin {
110    fn metadata(&self) -> &PluginMetadata {
111        &self.metadata
112    }
113
114    fn provider_type(&self) -> &str {
115        "gitlab"
116    }
117
118    fn initialize(
119        &mut self, provider_id: i64, config: HashMap<String, String>,
120        http_client: Option<std::sync::Arc<reqwest::Client>>,
121    ) -> PluginResult<()> {
122        let token = config
123            .get("token")
124            .ok_or_else(|| PluginError::InvalidConfig("Missing GitLab access token".to_string()))?
125            .to_string();
126
127        let base_url = config::get_base_url(&config);
128        let api_url = config::build_api_url(&base_url);
129
130        let client = http_client.unwrap_or_else(|| {
131            std::sync::Arc::new(
132                reqwest::Client::builder()
133                    .use_rustls_tls()
134                    .pool_max_idle_per_host(10)
135                    .timeout(Duration::from_secs(30))
136                    .connect_timeout(Duration::from_secs(10))
137                    .tcp_keepalive(Duration::from_secs(60))
138                    .build()
139                    .expect("Failed to build HTTP client"),
140            )
141        });
142
143        self.client = Some(client::GitLabClient::new(client, api_url, token));
144        self.provider_id = Some(provider_id);
145        self.config = config;
146
147        Ok(())
148    }
149
150    async fn validate_credentials(&self) -> PluginResult<bool> {
151        let client = self.client()?;
152        client.get_user().await?;
153        Ok(true)
154    }
155
156    async fn fetch_available_pipelines(
157        &self, params: Option<PaginationParams>,
158    ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
159        let params = params.unwrap_or_default();
160        let response = self.fetch_projects_paginated(&params).await?;
161
162        let available_pipelines = response
163            .items
164            .iter()
165            .map(mapper::map_available_pipeline)
166            .collect();
167
168        Ok(PaginatedResponse::new(
169            available_pipelines,
170            response.page,
171            response.page_size,
172            response.total_count,
173        ))
174    }
175
176    async fn fetch_organizations(&self) -> PluginResult<Vec<pipedash_plugin_api::Organization>> {
177        let client = self.client()?;
178        client.list_groups().await
179    }
180
181    async fn fetch_available_pipelines_filtered(
182        &self, org: Option<String>, search: Option<String>, params: Option<PaginationParams>,
183    ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
184        let params = params.unwrap_or_default();
185        let client = self.client()?;
186        let response = client.list_projects_filtered(org, search, &params).await?;
187
188        let available_pipelines = response
189            .items
190            .iter()
191            .map(mapper::map_available_pipeline)
192            .collect();
193
194        Ok(PaginatedResponse::new(
195            available_pipelines,
196            response.page,
197            response.page_size,
198            response.total_count,
199        ))
200    }
201
202    async fn fetch_pipelines(&self) -> PluginResult<Vec<Pipeline>> {
203        let provider_id = self
204            .provider_id
205            .ok_or_else(|| PluginError::Internal("Provider ID not set".to_string()))?;
206
207        let client = self.client()?;
208        let projects = self.fetch_all_projects().await?;
209
210        let pipeline_futures = projects.iter().map(|project| async move {
211            let pipelines = client.get_project_pipelines(project.id, 1).await.ok()?;
212            let latest_pipeline = pipelines.first();
213            Some(mapper::map_pipeline(project, latest_pipeline, provider_id))
214        });
215
216        let results: Vec<Option<Pipeline>> = join_all(pipeline_futures).await;
217        Ok(results.into_iter().flatten().collect())
218    }
219
220    async fn fetch_run_history(
221        &self, pipeline_id: &str, limit: usize,
222    ) -> PluginResult<Vec<PipelineRun>> {
223        let (provider_id, project_id) = config::parse_pipeline_id(pipeline_id)?;
224        let client = self.client()?;
225
226        let project = client.get_project(project_id).await?;
227
228        let pipeline_list = client.get_project_pipelines(project_id, limit).await?;
229
230        let parts: Vec<&str> = project.name_with_namespace.split('/').collect();
231        let namespace = if parts.len() >= 2 {
232            Some(parts[..parts.len() - 1].join("/"))
233        } else {
234            None
235        };
236
237        let detailed_pipeline_futures = pipeline_list
238            .iter()
239            .map(|p| async move { client.get_pipeline(project_id, p.id).await });
240
241        let detailed_pipelines = join_all(detailed_pipeline_futures).await;
242
243        Ok(detailed_pipelines
244            .into_iter()
245            .filter_map(|result| result.ok())
246            .map(|p| mapper::map_pipeline_run(&p, project_id, provider_id, namespace.as_deref()))
247            .collect())
248    }
249
250    async fn fetch_run_details(
251        &self, pipeline_id: &str, run_number: i64,
252    ) -> PluginResult<PipelineRun> {
253        let (provider_id, project_id) = config::parse_pipeline_id(pipeline_id)?;
254        let client = self.client()?;
255
256        let project = client.get_project(project_id).await?;
257        let pipeline = client.get_pipeline(project_id, run_number).await?;
258
259        let parts: Vec<&str> = project.name_with_namespace.split('/').collect();
260        let namespace = if parts.len() >= 2 {
261            Some(parts[..parts.len() - 1].join("/"))
262        } else {
263            None
264        };
265
266        Ok(mapper::map_pipeline_run(
267            &pipeline,
268            project_id,
269            provider_id,
270            namespace.as_deref(),
271        ))
272    }
273
274    async fn fetch_workflow_parameters(
275        &self, _workflow_id: &str,
276    ) -> PluginResult<Vec<WorkflowParameter>> {
277        Ok(vec![WorkflowParameter {
278            name: "ref".to_string(),
279            label: Some("Ref".to_string()),
280            description: Some("Branch, tag, or commit SHA to run pipeline on".to_string()),
281            param_type: WorkflowParameterType::String {
282                default: Some("main".to_string()),
283            },
284            required: true,
285        }])
286    }
287
288    async fn trigger_pipeline(&self, params: TriggerParams) -> PluginResult<String> {
289        let (_provider_id, project_id) = config::parse_pipeline_id(&params.workflow_id)?;
290
291        let client = self.client()?;
292
293        let ref_name = params
294            .inputs
295            .as_ref()
296            .and_then(|inputs| inputs.get("ref"))
297            .and_then(|v| v.as_str())
298            .unwrap_or("main")
299            .to_string();
300
301        let variables = params.inputs.as_ref().and_then(|inputs| {
302            inputs.get("variables").and_then(|vars| {
303                vars.as_object().map(|obj| {
304                    obj.iter()
305                        .map(|(key, value)| types::PipelineVariable {
306                            key: key.clone(),
307                            value: value.as_str().unwrap_or("").to_string(),
308                            variable_type: Some("env_var".to_string()),
309                        })
310                        .collect()
311                })
312            })
313        });
314
315        let pipeline = client
316            .trigger_pipeline(project_id, ref_name, variables)
317            .await?;
318        Ok(pipeline.web_url)
319    }
320
321    async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> PluginResult<()> {
322        let (_, project_id) = config::parse_pipeline_id(pipeline_id)?;
323        let client = self.client()?;
324        client.cancel_pipeline(project_id, run_number).await?;
325        Ok(())
326    }
327
328    async fn fetch_agents(&self) -> PluginResult<Vec<BuildAgent>> {
329        Err(PluginError::NotSupported(
330            "GitLab runners monitoring not implemented".to_string(),
331        ))
332    }
333
334    async fn fetch_artifacts(&self, _run_id: &str) -> PluginResult<Vec<BuildArtifact>> {
335        Err(PluginError::NotSupported(
336            "Artifacts download not implemented".to_string(),
337        ))
338    }
339
340    async fn fetch_queues(&self) -> PluginResult<Vec<BuildQueue>> {
341        Err(PluginError::NotSupported(
342            "Build queues not supported by GitLab".to_string(),
343        ))
344    }
345
346    fn get_migrations(&self) -> Vec<String> {
347        vec![]
348    }
349}
350
351#[cfg(test)]
352mod tests {
353    use tokio::io::{
354        AsyncReadExt,
355        AsyncWriteExt,
356    };
357
358    use super::*;
359
360    async fn stub_gitlab() -> (GitLabPlugin, tokio::task::JoinHandle<()>) {
361        let _ = rustls::crypto::ring::default_provider().install_default();
362        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
363        let addr = listener.local_addr().unwrap();
364        let server = tokio::spawn(async move {
365            loop {
366                let (mut stream, _) = listener.accept().await.unwrap();
367                let mut request = Vec::new();
368                let mut buffer = [0; 4096];
369                while !request.windows(4).any(|window| window == b"\r\n\r\n") {
370                    let count = stream.read(&mut buffer).await.unwrap();
371                    if count == 0 {
372                        break;
373                    }
374                    request.extend_from_slice(&buffer[..count]);
375                }
376                let request = String::from_utf8(request).unwrap();
377                let target = request.split_whitespace().nth(1).unwrap();
378                let url = reqwest::Url::parse(&format!("http://{addr}{target}")).unwrap();
379                let query: HashMap<_, _> = url.query_pairs().into_owned().collect();
380                let mut total = 0;
381                let body = match url.path() {
382                    "/api/v4/user" => serde_json::json!({
383                        "id": 7, "username": "alice", "name": "Alice"
384                    }),
385                    "/api/v4/projects/102/pipelines" => serde_json::json!([]),
386                    "/api/v4/groups/42/projects"
387                    | "/api/v4/projects"
388                    | "/api/v4/users/7/projects" => {
389                        let recursive = url.path() == "/api/v4/projects"
390                            || query
391                                .get("include_subgroups")
392                                .is_some_and(|value| value == "true");
393                        let projects: Vec<_> = [
394                            (101, "team / root"),
395                            (102, "team / backend / api"),
396                            (103, "team / frontend / api"),
397                            (104, "team / backend / nested / worker"),
398                        ]
399                        .into_iter()
400                        .filter(|(_, path)| recursive || *path == "team / root")
401                        .filter(|(_, path)| {
402                            query
403                                .get("search")
404                                .is_none_or(|search| path.contains(search))
405                        })
406                        .map(|(id, path)| {
407                            serde_json::json!({
408                                "id": id,
409                                "name": path.rsplit(" / ").next().unwrap(),
410                                "name_with_namespace": path,
411                                "web_url": format!("http://{addr}/{}", path.replace(" / ", "/"))
412                            })
413                        })
414                        .collect();
415                        total = projects.len();
416                        let page = query.get("page").unwrap().parse::<usize>().unwrap();
417                        let page_size = query.get("per_page").unwrap().parse::<usize>().unwrap();
418                        serde_json::Value::Array(
419                            projects
420                                .into_iter()
421                                .skip((page - 1) * page_size)
422                                .take(page_size)
423                                .collect(),
424                        )
425                    }
426                    path => panic!("Unexpected stub path: {path}"),
427                }
428                .to_string();
429                let response = format!(
430                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nX-Total: {total}\r\nConnection: close\r\n\r\n{body}",
431                    body.len()
432                );
433                stream.write_all(response.as_bytes()).await.unwrap();
434            }
435        });
436        let mut plugin = GitLabPlugin::new();
437        plugin
438            .initialize(
439                1,
440                HashMap::from([
441                    ("base_url".to_string(), format!("http://{addr}")),
442                    ("token".to_string(), "stub-token".to_string()),
443                ]),
444                None,
445            )
446            .unwrap();
447        (plugin, server)
448    }
449
450    #[tokio::test]
451    async fn parent_group_discovers_nested_projects_with_distinct_paths() {
452        let (plugin, server) = stub_gitlab().await;
453        let response = plugin
454            .fetch_available_pipelines_filtered(Some("42".to_string()), None, None)
455            .await
456            .unwrap();
457
458        let paths: Vec<_> = response.items.iter().map(|item| item.id.as_str()).collect();
459        assert_eq!(
460            paths,
461            [
462                "team/root",
463                "team/backend/api",
464                "team/frontend/api",
465                "team/backend/nested/worker"
466            ]
467        );
468        let displayed_paths: Vec<_> = response
469            .items
470            .iter()
471            .map(|item| {
472                format!(
473                    "{}/{}",
474                    item.organization.as_deref().unwrap(),
475                    item.repository.as_deref().unwrap()
476                )
477                .replace(' ', "")
478            })
479            .collect();
480        assert_eq!(displayed_paths, paths);
481        server.abort();
482    }
483
484    #[tokio::test]
485    async fn subgroup_search_keeps_pagination_and_total_count() {
486        let (plugin, server) = stub_gitlab().await;
487        for (page, expected, has_more) in [
488            (1, vec!["team/backend/api"], true),
489            (2, vec!["team/frontend/api"], true),
490            (3, vec![], false),
491        ] {
492            let response = plugin
493                .fetch_available_pipelines_filtered(
494                    Some("42".to_string()),
495                    Some("api".to_string()),
496                    Some(PaginationParams { page, page_size: 1 }),
497                )
498                .await
499                .unwrap();
500
501            assert_eq!(
502                response
503                    .items
504                    .iter()
505                    .map(|item| item.id.as_str())
506                    .collect::<Vec<_>>(),
507                expected
508            );
509            assert_eq!(response.total_count, 2);
510            assert_eq!(response.page, page);
511            assert_eq!(response.has_more, has_more);
512        }
513        server.abort();
514    }
515
516    #[tokio::test]
517    async fn saved_subgroup_selection_still_matches_during_refresh() {
518        let (mut plugin, server) = stub_gitlab().await;
519        plugin
520            .config
521            .insert("selected_items".to_string(), "team/backend/api".to_string());
522        let pipelines = plugin.fetch_pipelines().await.unwrap();
523
524        assert_eq!(
525            pipelines
526                .iter()
527                .map(|pipeline| pipeline.id.as_str())
528                .collect::<Vec<_>>(),
529            ["gitlab__1__102"]
530        );
531        assert_eq!(pipelines[0].repository, "team/backend/api");
532        server.abort();
533    }
534}
pipedash_plugin_gitlab/plugin.rs

pipedash_plugin_gitlab/
plugin.rs
