1use std::collections::HashMap;
2use std::sync::Arc;
3use std::time::Duration;
4
5use async_trait::async_trait;
6use futures::future::try_join_all;
7use pipedash_plugin_api::*;
8
9use crate::{
10 client,
11 config,
12 mapper,
13 metadata,
14};
15
16pub struct GiteaPlugin {
17 metadata: PluginMetadata,
18 client: Option<client::GiteaClient>,
19 provider_id: Option<i64>,
20 config: HashMap<String, String>,
21 status_source: Option<StatusSource>,
22}
23
24impl Default for GiteaPlugin {
25 fn default() -> Self {
26 Self::new()
27 }
28}
29
30impl GiteaPlugin {
31 pub fn new() -> Self {
32 Self {
33 metadata: metadata::create_metadata(),
34 client: None,
35 provider_id: None,
36 config: HashMap::new(),
37 status_source: None,
38 }
39 }
40
41 fn client(&self) -> PluginResult<&client::GiteaClient> {
42 self.client
43 .as_ref()
44 .ok_or_else(|| PluginError::Internal("Plugin not initialized".to_string()))
45 }
46
47 fn provider_id(&self) -> PluginResult<i64> {
48 self.provider_id
49 .ok_or_else(|| PluginError::Internal("Provider ID not set".to_string()))
50 }
51
52 async fn fetch_repo_pipelines(
53 &self, provider_id: i64, repo_full_name: &str,
54 ) -> PluginResult<Vec<Pipeline>> {
55 let (owner, repo) = config::parse_repo(repo_full_name)?;
56 let client = self.client()?;
57 let workflows = client.list_workflows(owner, repo).await?;
58 let workflow_ids: Vec<String> = workflows.iter().map(|w| w.id.clone()).collect();
59 let latest_runs = client.latest_runs(owner, repo, &workflow_ids).await?;
60
61 Ok(workflows
62 .iter()
63 .map(|workflow| {
64 let latest_run = latest_runs.get(&workflow.id);
65 mapper::map_pipeline(provider_id, owner, repo, workflow, latest_run)
66 })
67 .collect())
68 }
69}
70
71#[async_trait]
72impl Plugin for GiteaPlugin {
73 fn metadata(&self) -> &PluginMetadata {
74 &self.metadata
75 }
76
77 fn status_source(&self) -> Option<&StatusSource> {
78 self.status_source.as_ref()
79 }
80
81 fn initialize(
82 &mut self, provider_id: i64, config: HashMap<String, String>,
83 http_client: Option<Arc<reqwest::Client>>,
84 ) -> PluginResult<()> {
85 let token = config
86 .get("token")
87 .map(|token| token.trim())
88 .filter(|token| !token.is_empty())
89 .ok_or_else(|| PluginError::InvalidConfig("Missing Gitea access token".to_string()))?
90 .to_string();
91 let base_url = config::get_base_url(&config)?;
92 let status_source = custom_status_source(&config)?;
93
94 let http_client = http_client.unwrap_or_else(|| {
95 Arc::new(
96 reqwest::Client::builder()
97 .use_rustls_tls()
98 .pool_max_idle_per_host(10)
99 .timeout(Duration::from_secs(30))
100 .connect_timeout(Duration::from_secs(10))
101 .tcp_keepalive(Duration::from_secs(60))
102 .build()
103 .expect("Failed to build HTTP client"),
104 )
105 });
106
107 self.status_source = status_source;
108 self.client = Some(client::GiteaClient::new(http_client, base_url, token));
109 self.provider_id = Some(provider_id);
110 self.config = config;
111 Ok(())
112 }
113
114 async fn validate_credentials(&self) -> PluginResult<bool> {
115 self.client()?.get_user().await?;
116 Ok(true)
117 }
118
119 async fn fetch_available_pipelines(
120 &self, params: Option<PaginationParams>,
121 ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
122 self.fetch_available_pipelines_filtered(None, None, params)
123 .await
124 }
125
126 async fn fetch_organizations(&self) -> PluginResult<Vec<Organization>> {
127 self.client()?.list_organizations().await
128 }
129
130 async fn fetch_available_pipelines_filtered(
131 &self, org: Option<String>, search: Option<String>, params: Option<PaginationParams>,
132 ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
133 let client = self.client()?;
134 let params = params.unwrap_or_default();
135 let (uid, exclusive) = match org {
136 Some(org) => (
137 org.parse().map_err(|_| {
138 PluginError::InvalidConfig(format!("Invalid organization ID: {org}"))
139 })?,
140 true,
141 ),
142 None => (client.get_user().await?.id, false),
143 };
144
145 let response = client
146 .search_repositories(uid, exclusive, search.as_deref(), ¶ms)
147 .await?;
148
149 Ok(PaginatedResponse {
150 items: response
151 .items
152 .iter()
153 .map(mapper::map_available_pipeline)
154 .collect(),
155 page: response.page,
156 page_size: response.page_size,
157 total_count: response.total_count,
158 total_pages: response.total_pages,
159 has_more: response.has_more,
160 })
161 }
162
163 async fn fetch_pipelines(&self) -> PluginResult<Vec<Pipeline>> {
164 let provider_id = self.provider_id()?;
165 let repositories = config::get_repositories(&self.config);
166 if repositories.is_empty() {
167 return Err(PluginError::InvalidConfig(
168 "No repositories configured".to_string(),
169 ));
170 }
171
172 let results = try_join_all(
173 repositories
174 .iter()
175 .map(|repo| self.fetch_repo_pipelines(provider_id, repo)),
176 )
177 .await?;
178
179 Ok(results.into_iter().flatten().collect())
180 }
181
182 async fn fetch_run_history(
183 &self, pipeline_id: &str, limit: usize,
184 ) -> PluginResult<Vec<PipelineRun>> {
185 let id = config::parse_pipeline_id(pipeline_id)?;
186 let runs = self
187 .client()?
188 .workflow_runs(id.owner, id.repo, id.workflow, limit)
189 .await?;
190 Ok(runs
191 .iter()
192 .map(|run| mapper::map_run(run, pipeline_id, id.owner))
193 .collect())
194 }
195
196 async fn fetch_run_details(
197 &self, pipeline_id: &str, run_number: i64,
198 ) -> PluginResult<PipelineRun> {
199 let id = config::parse_pipeline_id(pipeline_id)?;
200 let run = self
201 .client()?
202 .find_run(id.owner, id.repo, id.workflow, run_number)
203 .await?;
204 Ok(mapper::map_run(&run, pipeline_id, id.owner))
205 }
206
207 async fn fetch_workflow_parameters(
208 &self, workflow_id: &str,
209 ) -> PluginResult<Vec<WorkflowParameter>> {
210 let id = config::parse_pipeline_id(workflow_id)?;
211 let client = self.client()?;
212 let workflows = client.list_workflows(id.owner, id.repo).await?;
213 let path = workflows
214 .iter()
215 .find(|workflow| workflow.id == id.workflow)
216 .map(|workflow| workflow.path.as_str())
217 .ok_or_else(|| {
218 PluginError::PipelineNotFound(format!("Workflow {} not found", id.workflow))
219 })?;
220 let source = client.raw_file(id.owner, id.repo, path).await?;
221
222 let mut parameters = vec![WorkflowParameter {
223 name: "ref".to_string(),
224 label: Some("Ref".to_string()),
225 description: Some("Branch, tag, or commit SHA to run workflow on".to_string()),
226 param_type: WorkflowParameterType::String {
227 default: Some("main".to_string()),
228 },
229 required: true,
230 }];
231 parameters.extend(mapper::dispatch_inputs(&source)?);
232 Ok(parameters)
233 }
234
235 async fn trigger_pipeline(&self, params: TriggerParams) -> PluginResult<String> {
236 let id = config::parse_pipeline_id(¶ms.workflow_id)?;
237 let client = self.client()?;
238
239 let body = mapper::dispatch_body(params.inputs);
240
241 client
242 .dispatch_workflow(id.owner, id.repo, id.workflow, &body)
243 .await?;
244 Ok(client.actions_url(id.owner, id.repo))
245 }
246
247 async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> PluginResult<()> {
248 let id = config::parse_pipeline_id(pipeline_id)?;
249 let client = self.client()?;
250 let run = client
251 .find_run(id.owner, id.repo, id.workflow, run_number)
252 .await?;
253 client.cancel_run(id.owner, id.repo, run.id).await
254 }
255}
256
257#[cfg(test)]
258mod tests {
259 use tokio::io::{
260 AsyncReadExt,
261 AsyncWriteExt,
262 };
263
264 use super::*;
265
266 const STUB_REPOS: usize = 150;
267 const STUB_ORGS: usize = 70;
268 const MAX_RESPONSE_ITEMS: usize = 50;
269 const DEFAULT_PAGING_NUM: usize = 30;
270
271 fn stub_page(url: &reqwest::Url, total: usize) -> std::ops::Range<usize> {
272 let query = |key: &str| {
273 url.query_pairs()
274 .find(|(k, _)| k == key)
275 .map(|(_, v)| v.parse::<usize>().unwrap())
276 };
277 let limit = query("limit")
278 .unwrap_or(DEFAULT_PAGING_NUM)
279 .min(MAX_RESPONSE_ITEMS);
280 let page = query("page").unwrap_or(1);
281 ((page - 1) * limit).min(total)..(page * limit).min(total)
282 }
283
284 async fn stub_gitea() -> (GiteaPlugin, tokio::task::JoinHandle<()>) {
285 let _ = rustls::crypto::ring::default_provider().install_default();
286 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
287 let addr = listener.local_addr().unwrap();
288 let server = tokio::spawn(async move {
289 loop {
290 let (mut stream, _) = listener.accept().await.unwrap();
291 let mut request = Vec::new();
292 let mut buffer = [0; 4096];
293 while !request.windows(4).any(|window| window == b"\r\n\r\n") {
294 let count = stream.read(&mut buffer).await.unwrap();
295 if count == 0 {
296 break;
297 }
298 request.extend_from_slice(&buffer[..count]);
299 }
300 let request = String::from_utf8(request).unwrap();
301 let mut head = request.split_whitespace();
302 let method = head.next().unwrap();
303 let url =
304 reqwest::Url::parse(&format!("http://{addr}{}", head.next().unwrap())).unwrap();
305 let mut total_count = None;
306 let (status, body) = match (method, url.path()) {
307 ("GET", "/api/v1/repos/search") => {
308 total_count = Some(STUB_REPOS);
309 let data: Vec<_> = stub_page(&url, STUB_REPOS)
310 .map(|i| {
311 serde_json::json!({
312 "name": format!("r{i}"),
313 "full_name": format!("org/r{i}"),
314 "owner": {"login": "org"}
315 })
316 })
317 .collect();
318 (200, serde_json::json!({"data": data}))
319 }
320 ("GET", "/api/v1/user/orgs") => {
321 total_count = Some(STUB_ORGS);
322 let orgs: Vec<_> = stub_page(&url, STUB_ORGS)
323 .map(|i| serde_json::json!({"id": 100 + i, "username": format!("o{i}")}))
324 .collect();
325 (200, serde_json::json!(orgs))
326 }
327 ("GET", "/api/v1/user") => {
328 (200, serde_json::json!({"id": 2, "login": "claude"}))
329 }
330 ("GET", "/api/v1/repos/claude/smoke__ci/actions/workflows") => (
331 200,
332 serde_json::json!({"workflows": [
333 {"id": "build__test.yaml", "name": "Build", "path": ".gitea/workflows/build__test.yaml"},
334 {"id": "lint.yaml", "name": "Lint", "path": ".gitea/workflows/lint.yaml"}
335 ]}),
336 ),
337 ("GET", "/api/v1/repos/claude/smoke__ci/actions/runs") => {
338 let page: usize = url
339 .query_pairs()
340 .find(|(key, _)| key == "page")
341 .map(|(_, value)| value.parse().unwrap())
342 .unwrap();
343 let runs: Vec<_> = if page == 1 {
344 (0..50)
345 .map(|i| {
346 serde_json::json!({
347 "id": 100 - i, "run_number": 100 - i, "status": "queued",
348 "path": "build__test.yaml@refs/heads/main", "head_branch": "main",
349 "html_url": format!("http://g/claude/smoke__ci/actions/runs/{}", 100 - i)
350 })
351 })
352 .collect()
353 } else {
354 vec![
355 serde_json::json!({"id": 3, "run_number": 3, "status": "completed", "conclusion": "failure",
356 "path": "lint.yaml@refs/heads/main", "head_branch": "main",
357 "html_url": "http://g/claude/smoke__ci/actions/runs/3"}),
358 serde_json::json!({"id": 2, "run_number": 2, "status": "queued",
359 "path": "build__test.yaml@refs/heads/main", "head_branch": "main",
360 "html_url": "http://g/claude/smoke__ci/actions/runs/2"}),
361 serde_json::json!({"id": 1, "run_number": 1, "status": "completed", "conclusion": "success",
362 "path": "build__test.yaml@refs/heads/main", "head_branch": "main",
363 "html_url": "http://g/claude/smoke__ci/actions/runs/1"}),
364 ]
365 };
366 (200, serde_json::json!({"workflow_runs": runs}))
367 }
368 ("POST", "/api/v1/repos/claude/smoke__ci/actions/runs/2/cancel") => {
369 (404, serde_json::json!({"message": "not found"}))
370 }
371 ("GET", "/api/v1/repos/claude/missing/actions/workflows") => {
372 (404, serde_json::json!({"message": "not found"}))
373 }
374 ("GET", "/api/v1/repos/claude/smoke__ci/raw/.gitea/workflows/build__test.yaml") => {
375 (200, serde_json::json!("on:\n workflow_dispatch:\n inputs:\n greeting:\n description: Greeting\n default: hi\n"))
376 }
377 (method, path) => panic!("Unexpected stub request: {method} {path}"),
378 };
379 let body = match body {
380 serde_json::Value::String(raw) => raw,
381 json => json.to_string(),
382 };
383 let total_header = total_count
384 .map(|total| format!("X-Total-Count: {total}\r\n"))
385 .unwrap_or_default();
386 let response = format!(
387 "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\n{total_header}Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
388 body.len()
389 );
390 stream.write_all(response.as_bytes()).await.unwrap();
391 }
392 });
393 let mut plugin = GiteaPlugin::new();
394 plugin
395 .initialize(
396 1,
397 HashMap::from([
398 ("base_url".to_string(), format!("http://{addr}/")),
399 ("token".to_string(), "stub-token".to_string()),
400 ("selected_items".to_string(), "claude/smoke__ci".to_string()),
401 ]),
402 None,
403 )
404 .unwrap();
405 (plugin, server)
406 }
407
408 #[tokio::test]
409 async fn pipelines_find_latest_run_of_each_workflow_beyond_first_page() {
410 let (plugin, server) = stub_gitea().await;
411 let pipelines = plugin.fetch_pipelines().await.unwrap();
412
413 let summary: Vec<_> = pipelines
414 .iter()
415 .map(|p| (p.id.as_str(), p.status))
416 .collect();
417 assert_eq!(
418 summary,
419 [
420 (
421 "gitea__1__claude/smoke__ci/build__test.yaml",
422 PipelineStatus::Running
423 ),
424 (
425 "gitea__1__claude/smoke__ci/lint.yaml",
426 PipelineStatus::Failed
427 ),
428 ]
429 );
430 server.abort();
431 }
432
433 #[tokio::test]
434 async fn one_failing_repository_fails_the_whole_fetch() {
435 let (mut plugin, server) = stub_gitea().await;
436 plugin.config.insert(
437 "selected_items".to_string(),
438 "claude/smoke__ci,claude/missing".to_string(),
439 );
440 let error = plugin.fetch_pipelines().await.unwrap_err();
441 assert!(matches!(error, PluginError::PipelineNotFound(_)), "{error}");
442 server.abort();
443 }
444
445 #[tokio::test]
446 async fn workflow_parameters_include_dispatch_inputs_after_ref() {
447 let (plugin, server) = stub_gitea().await;
448 let params = plugin
449 .fetch_workflow_parameters("gitea__1__claude/smoke__ci/build__test.yaml")
450 .await
451 .unwrap();
452 assert_eq!(
453 params.iter().map(|p| p.name.as_str()).collect::<Vec<_>>(),
454 ["ref", "inputs.greeting"]
455 );
456 assert_eq!(params[1].label.as_deref(), Some("greeting"));
457 server.abort();
458 }
459
460 #[tokio::test]
461 async fn run_history_filters_by_workflow_and_respects_limit() {
462 let (plugin, server) = stub_gitea().await;
463 let runs = plugin
464 .fetch_run_history("gitea__1__claude/smoke__ci/build__test.yaml", 1)
465 .await
466 .unwrap();
467
468 assert_eq!(runs.iter().map(|r| r.run_number).collect::<Vec<_>>(), [100]);
469 let details = plugin
470 .fetch_run_details("gitea__1__claude/smoke__ci/build__test.yaml", 1)
471 .await
472 .unwrap();
473 assert_eq!(details.status, PipelineStatus::Success);
474 assert!(plugin
475 .fetch_run_details("gitea__1__claude/smoke__ci/build__test.yaml", 3)
476 .await
477 .is_err());
478 server.abort();
479 }
480
481 #[tokio::test]
482 async fn cancel_reports_not_supported_when_endpoint_is_missing() {
483 let (plugin, server) = stub_gitea().await;
484 let error = plugin
485 .cancel_run("gitea__1__claude/smoke__ci/build__test.yaml", 2)
486 .await
487 .unwrap_err();
488 assert!(matches!(error, PluginError::NotSupported(_)), "{error}");
489 server.abort();
490 }
491
492 #[tokio::test]
493 async fn dispatch_is_not_resent_after_a_server_error() {
494 let _ = rustls::crypto::ring::default_provider().install_default();
495 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
496 let addr = listener.local_addr().unwrap();
497 let requests = Arc::new(std::sync::atomic::AtomicUsize::new(0));
498 let counter = requests.clone();
499 let server = tokio::spawn(async move {
500 loop {
501 let (mut stream, _) = listener.accept().await.unwrap();
502 let mut buffer = [0; 4096];
503 let _ = stream.read(&mut buffer).await.unwrap();
504 counter.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
505 stream
506 .write_all(b"HTTP/1.1 503 X\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
507 .await
508 .unwrap();
509 }
510 });
511 let mut plugin = GiteaPlugin::new();
512 plugin
513 .initialize(
514 1,
515 HashMap::from([
516 ("base_url".to_string(), format!("http://{addr}/")),
517 ("token".to_string(), "stub-token".to_string()),
518 ]),
519 None,
520 )
521 .unwrap();
522
523 let error = plugin
524 .client()
525 .unwrap()
526 .dispatch_workflow("o", "r", "w.yaml", &serde_json::json!({}))
527 .await
528 .unwrap_err();
529 server.abort();
530
531 assert!(
532 matches!(error, PluginError::ServiceUnavailable { .. }),
533 "{error}"
534 );
535 assert_eq!(requests.load(std::sync::atomic::Ordering::SeqCst), 1);
536 }
537
538 #[tokio::test]
539 async fn repo_picker_pages_reach_every_repo_when_gitea_caps_the_limit() {
540 let (plugin, server) = stub_gitea().await;
541 let mut names = Vec::new();
542 let mut has_more = true;
543 let mut page = 1;
544 while has_more {
545 let response = plugin
546 .fetch_available_pipelines_filtered(
547 Some("7".into()),
548 None,
549 Some(PaginationParams {
550 page,
551 page_size: 100,
552 }),
553 )
554 .await
555 .unwrap();
556 names.extend(response.items.into_iter().map(|item| item.name));
557 has_more = response.has_more;
558 page += 1;
559 }
560 server.abort();
561
562 let expected: Vec<_> = (0..STUB_REPOS).map(|i| format!("org/r{i}")).collect();
563 assert_eq!(names, expected);
564 }
565
566 #[tokio::test]
567 async fn organizations_include_every_page() {
568 let (plugin, server) = stub_gitea().await;
569 let orgs = plugin.fetch_organizations().await.unwrap();
570 server.abort();
571
572 assert_eq!(orgs.len(), STUB_ORGS + 1);
573 assert_eq!(orgs.last().unwrap().name, format!("o{}", STUB_ORGS - 1));
574 }
575}