diffler_ci/providers/
gitlab.rs

1//! GitLab CI adapter (CLI-only via `glab api`, REST). The dependency graph is
2//! derived from pipeline stages (jobs in a stage depend on all jobs in the
3//! previous stage) — GitLab's default pipeline view. The exact `needs` DAG is a
4//! GraphQL refinement left for later. Logs poll the job trace by offset.
5
6use async_trait::async_trait;
7use serde::Deserialize;
8
9use crate::error::{CiError, Result};
10use crate::exec::CommandRunner;
11use crate::model::{
12    Capabilities, CiJob, CiRun, DagSource, JobId, JobStatus, LogChunk, LogMode, PullRequest,
13    RunDetail, RunExtras, RunId,
14};
15use crate::provider::{CiProvider, ProviderKind};
16
17/// Talks to GitLab CI through `glab api`. `glab` resolves the project from the
18/// repo via the `:fullpath` placeholder; an explicit `host` targets a
19/// self-hosted instance.
20pub struct GitLabProvider {
21    runner: Box<dyn CommandRunner>,
22    host: Option<String>,
23}
24
25impl GitLabProvider {
26    pub fn new(runner: Box<dyn CommandRunner>, host: Option<String>) -> Self {
27        Self { runner, host }
28    }
29
30    /// `glab api <path>`, with `--hostname` when a self-hosted host is set.
31    async fn api(&self, path: &str) -> Result<String> {
32        let mut args = vec!["api".to_owned()];
33        if let Some(host) = &self.host {
34            args.push("--hostname".to_owned());
35            args.push(host.clone());
36        }
37        args.push(path.to_owned());
38        self.runner.run("glab", &args).await
39    }
40}
41
42#[async_trait]
43impl CiProvider for GitLabProvider {
44    fn kind(&self) -> ProviderKind {
45        ProviderKind::GitLab
46    }
47
48    fn capabilities(&self) -> Capabilities {
49        Capabilities {
50            dag: DagSource::RunApi,
51            logs: LogMode::Poll,
52        }
53    }
54
55    async fn list_runs(&self, limit: usize) -> Result<Vec<CiRun>> {
56        let path = format!("projects/:fullpath/pipelines?per_page={limit}");
57        let out = self.api(&path).await?;
58        let raw: Vec<PipelineItem> = serde_json::from_str(&out).map_err(|e| CiError::Parse {
59            what: "glab pipelines".into(),
60            message: e.to_string(),
61        })?;
62        Ok(raw.into_iter().map(PipelineItem::into_run).collect())
63    }
64
65    async fn run_detail(&self, run: &RunId) -> Result<RunDetail> {
66        let meta = self
67            .api(&format!("projects/:fullpath/pipelines/{}", run.0))
68            .await?;
69        let pipeline: PipelineItem = serde_json::from_str(&meta).map_err(|e| CiError::Parse {
70            what: "glab pipeline".into(),
71            message: e.to_string(),
72        })?;
73        let jobs_out = self
74            .api(&format!("projects/:fullpath/pipelines/{}/jobs", run.0))
75            .await?;
76        let raw: Vec<JobItem> = serde_json::from_str(&jobs_out).map_err(|e| CiError::Parse {
77            what: "glab jobs".into(),
78            message: e.to_string(),
79        })?;
80        Ok(RunDetail {
81            run: pipeline.into_run(),
82            jobs: jobs_with_stage_edges(raw),
83        })
84    }
85
86    async fn job_log(&self, _run: &RunId, job: &JobId, offset: u64) -> Result<LogChunk> {
87        let trace = self
88            .api(&format!("projects/:fullpath/jobs/{}/trace", job.0))
89            .await?;
90        // resume from the saved offset, clamped to the end and floored to a char
91        // boundary, so a multibyte split or a shrunk/replaced trace yields the
92        // correct tail (or empty) instead of re-emitting the whole trace
93        let mut start = usize::try_from(offset)
94            .unwrap_or(usize::MAX)
95            .min(trace.len());
96        while start > 0 && !trace.is_char_boundary(start) {
97            start -= 1;
98        }
99        Ok(LogChunk {
100            next_offset: trace.len() as u64,
101            text: trace[start..].to_owned(),
102            steps: Vec::new(),
103            done: false,
104        })
105    }
106
107    /// GitLab exposes neither run artifacts nor annotations through this adapter.
108    async fn run_extras(&self, _run: &RunId) -> Result<RunExtras> {
109        Ok(RunExtras::default())
110    }
111
112    /// Merge-request resolution isn't wired for the GitLab adapter yet.
113    async fn current_pr(&self) -> Result<Option<PullRequest>> {
114        Ok(None)
115    }
116}
117
118/// Order jobs into stages (by first appearance) and link each job to every job
119/// in the previous stage — GitLab's stage-sequenced pipeline graph.
120fn jobs_with_stage_edges(raw: Vec<JobItem>) -> Vec<CiJob> {
121    let mut stage_order: Vec<String> = Vec::new();
122    for job in &raw {
123        if !stage_order.contains(&job.stage) {
124            stage_order.push(job.stage.clone());
125        }
126    }
127    let ids_in = |stage: &str| -> Vec<JobId> {
128        raw.iter()
129            .filter(|j| j.stage == stage)
130            .map(|j| JobId(j.id.to_string()))
131            .collect()
132    };
133    raw.iter()
134        .map(|job| {
135            let stage_idx = stage_order.iter().position(|s| *s == job.stage);
136            let needs = stage_idx
137                .and_then(|i| i.checked_sub(1))
138                .map(|prev| ids_in(&stage_order[prev]))
139                .unwrap_or_default();
140            CiJob {
141                id: JobId(job.id.to_string()),
142                name: job.name.clone(),
143                status: map_status(&job.status),
144                needs,
145            }
146        })
147        .collect()
148}
149
150fn map_status(status: &str) -> JobStatus {
151    match status {
152        "success" => JobStatus::Ok,
153        "failed" => JobStatus::Failed,
154        "running" => JobStatus::Running,
155        "canceled" | "canceling" => JobStatus::Neutral,
156        "skipped" | "manual" => JobStatus::Skipped,
157        _ => JobStatus::Queued,
158    }
159}
160
161fn parse_created(raw: &str) -> Option<time::OffsetDateTime> {
162    time::OffsetDateTime::parse(raw, &time::format_description::well_known::Rfc3339).ok()
163}
164
165#[derive(Deserialize)]
166struct PipelineItem {
167    id: u64,
168    status: String,
169    #[serde(rename = "ref")]
170    git_ref: Option<String>,
171    sha: Option<String>,
172    source: Option<String>,
173    created_at: Option<String>,
174    web_url: Option<String>,
175}
176
177impl PipelineItem {
178    fn into_run(self) -> CiRun {
179        CiRun {
180            id: RunId(self.id.to_string()),
181            name: self.source.unwrap_or_else(|| "pipeline".to_owned()),
182            title: String::new(),
183            branch: self.git_ref.unwrap_or_default(),
184            commit: self.sha.unwrap_or_default(),
185            author: String::new(),
186            created: self.created_at.as_deref().and_then(parse_created),
187            status: map_status(&self.status),
188            url: self.web_url,
189        }
190    }
191}
192
193#[derive(Deserialize)]
194struct JobItem {
195    id: u64,
196    name: String,
197    stage: String,
198    status: String,
199}
200
201#[cfg(test)]
202mod tests {
203    use super::*;
204    use crate::exec::test_support::RecordingRunner;
205
206    fn provider(responses: &[(&'static str, &str)]) -> GitLabProvider {
207        GitLabProvider::new(Box::new(RecordingRunner::new(responses)), None)
208    }
209
210    #[tokio::test]
211    async fn list_runs_parses_pipelines() {
212        let json = r#"[
213          {"id":100,"status":"running","ref":"main","sha":"deadbeef","source":"push",
214           "created_at":"2026-06-18T10:00:00Z","web_url":"https://gl/p/100"}
215        ]"#;
216        let runs = provider(&[("pipelines", json)])
217            .list_runs(10)
218            .await
219            .expect("runs");
220        assert_eq!(runs.len(), 1);
221        assert_eq!(runs[0].id, RunId("100".into()));
222        assert_eq!(runs[0].branch, "main");
223        assert_eq!(runs[0].status, JobStatus::Running);
224    }
225
226    #[tokio::test]
227    async fn run_detail_links_stages_into_a_dag() {
228        let pipeline = r#"{"id":100,"status":"running","ref":"main","sha":"d","source":"push",
229          "created_at":"2026-06-18T10:00:00Z","web_url":"u"}"#;
230        let jobs = r#"[
231          {"id":1,"name":"build","stage":"build","status":"success"},
232          {"id":2,"name":"test","stage":"test","status":"running"},
233          {"id":3,"name":"lint","stage":"test","status":"success"}
234        ]"#;
235        // "pipelines/100/jobs" must be matched before "pipelines/100"
236        let detail = provider(&[("/jobs", jobs), ("pipelines/100", pipeline)])
237            .run_detail(&RunId("100".into()))
238            .await
239            .expect("detail");
240        assert_eq!(detail.jobs.len(), 3);
241        // build is in the first stage → no upstream
242        assert!(detail.jobs[0].needs.is_empty());
243        // test-stage jobs depend on the build-stage job
244        assert_eq!(detail.jobs[1].needs, vec![JobId("1".into())]);
245        assert_eq!(detail.jobs[2].needs, vec![JobId("1".into())]);
246        assert_eq!(detail.jobs[1].status, JobStatus::Running);
247    }
248
249    #[tokio::test]
250    async fn job_log_returns_the_tail_from_offset() {
251        let chunk = provider(&[("/trace", "hello world")])
252            .job_log(&RunId("100".into()), &JobId("1".into()), 6)
253            .await
254            .expect("log");
255        assert_eq!(chunk.text, "world");
256        assert_eq!(chunk.next_offset, 11);
257    }
258
259    #[tokio::test]
260    async fn job_log_past_the_end_yields_empty_not_a_duplicate() {
261        // a shrunk/replaced trace (offset > len) must not re-emit the whole trace
262        let chunk = provider(&[("/trace", "short")])
263            .job_log(&RunId("100".into()), &JobId("1".into()), 999)
264            .await
265            .expect("log");
266        assert_eq!(chunk.text, "");
267        assert_eq!(chunk.next_offset, 5);
268    }
269}
diffler_ci/providers/gitlab.rs

diffler_ci/providers/
gitlab.rs
