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}
22
23impl Default for GiteaPlugin {
24 fn default() -> Self {
25 Self::new()
26 }
27}
28
29impl GiteaPlugin {
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::GiteaClient> {
40 self.client
41 .as_ref()
42 .ok_or_else(|| PluginError::Internal("Plugin not initialized".to_string()))
43 }
44
45 fn provider_id(&self) -> PluginResult<i64> {
46 self.provider_id
47 .ok_or_else(|| PluginError::Internal("Provider ID not set".to_string()))
48 }
49
50 async fn fetch_repo_pipelines(
51 &self, provider_id: i64, repo_full_name: &str,
52 ) -> PluginResult<Vec<Pipeline>> {
53 let (owner, repo) = config::parse_repo(repo_full_name)?;
54 let client = self.client()?;
55 let workflows = client.list_workflows(owner, repo).await?;
56 let workflow_ids: Vec<String> = workflows.iter().map(|w| w.id.clone()).collect();
57 let latest_runs = client.latest_runs(owner, repo, &workflow_ids).await?;
58
59 Ok(workflows
60 .iter()
61 .map(|workflow| {
62 let latest_run = latest_runs.get(&workflow.id);
63 mapper::map_pipeline(provider_id, owner, repo, workflow, latest_run)
64 })
65 .collect())
66 }
67}
68
69#[async_trait]
70impl Plugin for GiteaPlugin {
71 fn metadata(&self) -> &PluginMetadata {
72 &self.metadata
73 }
74
75 fn initialize(
76 &mut self, provider_id: i64, config: HashMap<String, String>,
77 http_client: Option<Arc<reqwest::Client>>,
78 ) -> PluginResult<()> {
79 let token = config
80 .get("token")
81 .map(|token| token.trim())
82 .filter(|token| !token.is_empty())
83 .ok_or_else(|| PluginError::InvalidConfig("Missing Gitea access token".to_string()))?
84 .to_string();
85 let base_url = config::get_base_url(&config)?;
86
87 let http_client = http_client.unwrap_or_else(|| {
88 Arc::new(
89 reqwest::Client::builder()
90 .use_rustls_tls()
91 .pool_max_idle_per_host(10)
92 .timeout(Duration::from_secs(30))
93 .connect_timeout(Duration::from_secs(10))
94 .tcp_keepalive(Duration::from_secs(60))
95 .build()
96 .expect("Failed to build HTTP client"),
97 )
98 });
99
100 self.client = Some(client::GiteaClient::new(http_client, base_url, token));
101 self.provider_id = Some(provider_id);
102 self.config = config;
103 Ok(())
104 }
105
106 async fn validate_credentials(&self) -> PluginResult<bool> {
107 self.client()?.get_user().await?;
108 Ok(true)
109 }
110
111 async fn fetch_available_pipelines(
112 &self, params: Option<PaginationParams>,
113 ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
114 self.fetch_available_pipelines_filtered(None, None, params)
115 .await
116 }
117
118 async fn fetch_organizations(&self) -> PluginResult<Vec<Organization>> {
119 self.client()?.list_organizations().await
120 }
121
122 async fn fetch_available_pipelines_filtered(
123 &self, org: Option<String>, search: Option<String>, params: Option<PaginationParams>,
124 ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
125 let client = self.client()?;
126 let params = params.unwrap_or_default();
127 let (uid, exclusive) = match org {
128 Some(org) => (
129 org.parse().map_err(|_| {
130 PluginError::InvalidConfig(format!("Invalid organization ID: {org}"))
131 })?,
132 true,
133 ),
134 None => (client.get_user().await?.id, false),
135 };
136
137 let response = client
138 .search_repositories(uid, exclusive, search.as_deref(), ¶ms)
139 .await?;
140
141 Ok(PaginatedResponse {
142 items: response
143 .items
144 .iter()
145 .map(mapper::map_available_pipeline)
146 .collect(),
147 page: response.page,
148 page_size: response.page_size,
149 total_count: response.total_count,
150 total_pages: response.total_pages,
151 has_more: response.has_more,
152 })
153 }
154
155 async fn fetch_pipelines(&self) -> PluginResult<Vec<Pipeline>> {
156 let provider_id = self.provider_id()?;
157 let repositories = config::get_repositories(&self.config);
158 if repositories.is_empty() {
159 return Err(PluginError::InvalidConfig(
160 "No repositories configured".to_string(),
161 ));
162 }
163
164 let results = try_join_all(
165 repositories
166 .iter()
167 .map(|repo| self.fetch_repo_pipelines(provider_id, repo)),
168 )
169 .await?;
170
171 Ok(results.into_iter().flatten().collect())
172 }
173
174 async fn fetch_run_history(
175 &self, pipeline_id: &str, limit: usize,
176 ) -> PluginResult<Vec<PipelineRun>> {
177 let id = config::parse_pipeline_id(pipeline_id)?;
178 let runs = self
179 .client()?
180 .workflow_runs(id.owner, id.repo, id.workflow, limit)
181 .await?;
182 Ok(runs
183 .iter()
184 .map(|run| mapper::map_run(run, pipeline_id, id.owner))
185 .collect())
186 }
187
188 async fn fetch_run_details(
189 &self, pipeline_id: &str, run_number: i64,
190 ) -> PluginResult<PipelineRun> {
191 let id = config::parse_pipeline_id(pipeline_id)?;
192 let run = self
193 .client()?
194 .find_run(id.owner, id.repo, id.workflow, run_number)
195 .await?;
196 Ok(mapper::map_run(&run, pipeline_id, id.owner))
197 }
198
199 async fn fetch_workflow_parameters(
200 &self, workflow_id: &str,
201 ) -> PluginResult<Vec<WorkflowParameter>> {
202 let id = config::parse_pipeline_id(workflow_id)?;
203 let client = self.client()?;
204 let workflows = client.list_workflows(id.owner, id.repo).await?;
205 let path = workflows
206 .iter()
207 .find(|workflow| workflow.id == id.workflow)
208 .map(|workflow| workflow.path.as_str())
209 .ok_or_else(|| {
210 PluginError::PipelineNotFound(format!("Workflow {} not found", id.workflow))
211 })?;
212 let source = client.raw_file(id.owner, id.repo, path).await?;
213
214 let mut parameters = vec![WorkflowParameter {
215 name: "ref".to_string(),
216 label: Some("Ref".to_string()),
217 description: Some("Branch, tag, or commit SHA to run workflow on".to_string()),
218 param_type: WorkflowParameterType::String {
219 default: Some("main".to_string()),
220 },
221 required: true,
222 }];
223 parameters.extend(mapper::dispatch_inputs(&source)?);
224 Ok(parameters)
225 }
226
227 async fn trigger_pipeline(&self, params: TriggerParams) -> PluginResult<String> {
228 let id = config::parse_pipeline_id(¶ms.workflow_id)?;
229 let client = self.client()?;
230
231 let body = mapper::dispatch_body(params.inputs);
232
233 client
234 .dispatch_workflow(id.owner, id.repo, id.workflow, &body)
235 .await?;
236 Ok(client.actions_url(id.owner, id.repo))
237 }
238
239 async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> PluginResult<()> {
240 let id = config::parse_pipeline_id(pipeline_id)?;
241 let client = self.client()?;
242 let run = client
243 .find_run(id.owner, id.repo, id.workflow, run_number)
244 .await?;
245 client.cancel_run(id.owner, id.repo, run.id).await
246 }
247}
248
249#[cfg(test)]
250mod tests {
251 use tokio::io::{
252 AsyncReadExt,
253 AsyncWriteExt,
254 };
255
256 use super::*;
257
258 async fn stub_gitea() -> (GiteaPlugin, tokio::task::JoinHandle<()>) {
259 let _ = rustls::crypto::ring::default_provider().install_default();
260 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
261 let addr = listener.local_addr().unwrap();
262 let server = tokio::spawn(async move {
263 loop {
264 let (mut stream, _) = listener.accept().await.unwrap();
265 let mut request = Vec::new();
266 let mut buffer = [0; 4096];
267 while !request.windows(4).any(|window| window == b"\r\n\r\n") {
268 let count = stream.read(&mut buffer).await.unwrap();
269 if count == 0 {
270 break;
271 }
272 request.extend_from_slice(&buffer[..count]);
273 }
274 let request = String::from_utf8(request).unwrap();
275 let mut head = request.split_whitespace();
276 let method = head.next().unwrap();
277 let url =
278 reqwest::Url::parse(&format!("http://{addr}{}", head.next().unwrap())).unwrap();
279 let (status, body) = match (method, url.path()) {
280 ("GET", "/api/v1/user") => {
281 (200, serde_json::json!({"id": 2, "login": "claude"}))
282 }
283 ("GET", "/api/v1/repos/claude/smoke__ci/actions/workflows") => (
284 200,
285 serde_json::json!({"workflows": [
286 {"id": "build__test.yaml", "name": "Build", "path": ".gitea/workflows/build__test.yaml"},
287 {"id": "lint.yaml", "name": "Lint", "path": ".gitea/workflows/lint.yaml"}
288 ]}),
289 ),
290 ("GET", "/api/v1/repos/claude/smoke__ci/actions/runs") => {
291 let page: usize = url
292 .query_pairs()
293 .find(|(key, _)| key == "page")
294 .map(|(_, value)| value.parse().unwrap())
295 .unwrap();
296 let runs: Vec<_> = if page == 1 {
297 (0..50)
298 .map(|i| {
299 serde_json::json!({
300 "id": 100 - i, "run_number": 100 - i, "status": "queued",
301 "path": "build__test.yaml@refs/heads/main", "head_branch": "main",
302 "html_url": format!("http://g/claude/smoke__ci/actions/runs/{}", 100 - i)
303 })
304 })
305 .collect()
306 } else {
307 vec![
308 serde_json::json!({"id": 3, "run_number": 3, "status": "completed", "conclusion": "failure",
309 "path": "lint.yaml@refs/heads/main", "head_branch": "main",
310 "html_url": "http://g/claude/smoke__ci/actions/runs/3"}),
311 serde_json::json!({"id": 2, "run_number": 2, "status": "queued",
312 "path": "build__test.yaml@refs/heads/main", "head_branch": "main",
313 "html_url": "http://g/claude/smoke__ci/actions/runs/2"}),
314 serde_json::json!({"id": 1, "run_number": 1, "status": "completed", "conclusion": "success",
315 "path": "build__test.yaml@refs/heads/main", "head_branch": "main",
316 "html_url": "http://g/claude/smoke__ci/actions/runs/1"}),
317 ]
318 };
319 (200, serde_json::json!({"workflow_runs": runs}))
320 }
321 ("POST", "/api/v1/repos/claude/smoke__ci/actions/runs/2/cancel") => {
322 (404, serde_json::json!({"message": "not found"}))
323 }
324 ("GET", "/api/v1/repos/claude/missing/actions/workflows") => {
325 (404, serde_json::json!({"message": "not found"}))
326 }
327 ("GET", "/api/v1/repos/claude/smoke__ci/raw/.gitea/workflows/build__test.yaml") => {
328 (200, serde_json::json!("on:\n workflow_dispatch:\n inputs:\n greeting:\n description: Greeting\n default: hi\n"))
329 }
330 (method, path) => panic!("Unexpected stub request: {method} {path}"),
331 };
332 let body = match body {
333 serde_json::Value::String(raw) => raw,
334 json => json.to_string(),
335 };
336 let response = format!(
337 "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
338 body.len()
339 );
340 stream.write_all(response.as_bytes()).await.unwrap();
341 }
342 });
343 let mut plugin = GiteaPlugin::new();
344 plugin
345 .initialize(
346 1,
347 HashMap::from([
348 ("base_url".to_string(), format!("http://{addr}/")),
349 ("token".to_string(), "stub-token".to_string()),
350 ("selected_items".to_string(), "claude/smoke__ci".to_string()),
351 ]),
352 None,
353 )
354 .unwrap();
355 (plugin, server)
356 }
357
358 #[tokio::test]
359 async fn pipelines_find_latest_run_of_each_workflow_beyond_first_page() {
360 let (plugin, server) = stub_gitea().await;
361 let pipelines = plugin.fetch_pipelines().await.unwrap();
362
363 let summary: Vec<_> = pipelines
364 .iter()
365 .map(|p| (p.id.as_str(), p.status))
366 .collect();
367 assert_eq!(
368 summary,
369 [
370 (
371 "gitea__1__claude/smoke__ci/build__test.yaml",
372 PipelineStatus::Running
373 ),
374 (
375 "gitea__1__claude/smoke__ci/lint.yaml",
376 PipelineStatus::Failed
377 ),
378 ]
379 );
380 server.abort();
381 }
382
383 #[tokio::test]
384 async fn one_failing_repository_fails_the_whole_fetch() {
385 let (mut plugin, server) = stub_gitea().await;
386 plugin.config.insert(
387 "selected_items".to_string(),
388 "claude/smoke__ci,claude/missing".to_string(),
389 );
390 let error = plugin.fetch_pipelines().await.unwrap_err();
391 assert!(matches!(error, PluginError::PipelineNotFound(_)), "{error}");
392 server.abort();
393 }
394
395 #[tokio::test]
396 async fn workflow_parameters_include_dispatch_inputs_after_ref() {
397 let (plugin, server) = stub_gitea().await;
398 let params = plugin
399 .fetch_workflow_parameters("gitea__1__claude/smoke__ci/build__test.yaml")
400 .await
401 .unwrap();
402 assert_eq!(
403 params.iter().map(|p| p.name.as_str()).collect::<Vec<_>>(),
404 ["ref", "inputs.greeting"]
405 );
406 assert_eq!(params[1].label.as_deref(), Some("greeting"));
407 server.abort();
408 }
409
410 #[tokio::test]
411 async fn run_history_filters_by_workflow_and_respects_limit() {
412 let (plugin, server) = stub_gitea().await;
413 let runs = plugin
414 .fetch_run_history("gitea__1__claude/smoke__ci/build__test.yaml", 1)
415 .await
416 .unwrap();
417
418 assert_eq!(runs.iter().map(|r| r.run_number).collect::<Vec<_>>(), [100]);
419 let details = plugin
420 .fetch_run_details("gitea__1__claude/smoke__ci/build__test.yaml", 1)
421 .await
422 .unwrap();
423 assert_eq!(details.status, PipelineStatus::Success);
424 assert!(plugin
425 .fetch_run_details("gitea__1__claude/smoke__ci/build__test.yaml", 3)
426 .await
427 .is_err());
428 server.abort();
429 }
430
431 #[tokio::test]
432 async fn cancel_reports_not_supported_when_endpoint_is_missing() {
433 let (plugin, server) = stub_gitea().await;
434 let error = plugin
435 .cancel_run("gitea__1__claude/smoke__ci/build__test.yaml", 2)
436 .await
437 .unwrap_err();
438 assert!(matches!(error, PluginError::NotSupported(_)), "{error}");
439 server.abort();
440 }
441}