navi_notifier_github/
source.rs

1//! The GitHub [`Source`]: turns notifications into fetches, fetches into diffs,
2//! and diffs into normalized events. All GitHub-specific I/O lives here; the
3//! decision logic lives in the pure `navi-notifier-forge` diff engine.
4
5use std::collections::HashSet;
6
7use async_trait::async_trait;
8use navi_notifier_core::model::{Event, Repo};
9use navi_notifier_core::traits::{Source, StateStore};
10use navi_notifier_core::SourceError;
11use navi_notifier_forge::model::{IssueComment, PrData, PullRequest, Review, ReviewComment, User};
12use navi_notifier_forge::{diff, first_sight_watermark, team_key, DiffContext, PrSnapshot};
13use octocrab::Octocrab;
14use serde::{Deserialize, Serialize};
15use time::format_description::well_known::Rfc3339;
16use time::{Duration, OffsetDateTime};
17use tokio::sync::OnceCell;
18use tracing::{debug, warn};
19
20/// A team from `GET /user/teams`, reduced to what we need to match team requests.
21#[derive(Deserialize)]
22struct GithubTeam {
23    slug: String,
24    organization: GithubOrg,
25}
26
27#[derive(Deserialize)]
28struct GithubOrg {
29    login: String,
30}
31
32use crate::notification::Notification;
33
34const SOURCE_ID: &str = "github";
35/// Safety cap on pagination per endpoint so a pathological PR can't stall a poll.
36const MAX_PAGES: u8 = 10;
37/// Overlap window when advancing the `since` cursor, to tolerate clock skew.
38const SINCE_OVERLAP: Duration = Duration::minutes(5);
39
40/// Configuration for the GitHub source.
41pub struct GitHubSourceConfig {
42    pub token: String,
43    /// API base for GitHub Enterprise Server (e.g. `https://ghe.example.com/api/v3`).
44    pub api_base: Option<String>,
45}
46
47pub struct GitHubSource {
48    octo: Octocrab,
49    /// Cached authenticated login, resolved lazily on first poll.
50    viewer: OnceCell<String>,
51}
52
53impl GitHubSource {
54    pub fn new(config: GitHubSourceConfig) -> Result<Self, SourceError> {
55        if config.token.trim().is_empty() {
56            return Err(SourceError::Auth(
57                "GitHub token is empty; set NAVI_GITHUB_TOKEN".into(),
58            ));
59        }
60        let mut builder = Octocrab::builder().personal_token(config.token);
61        if let Some(base) = config.api_base {
62            builder = builder
63                .base_uri(base)
64                .map_err(|e| SourceError::Request(format!("invalid api_base: {e}")))?;
65        }
66        let octo = builder
67            .build()
68            .map_err(|e| SourceError::Auth(e.to_string()))?;
69        Ok(Self {
70            octo,
71            viewer: OnceCell::new(),
72        })
73    }
74
75    /// The viewer's team memberships as `"org/slug"` keys, for matching team review
76    /// requests. Fetched every poll (not cached like the login) so joining or
77    /// leaving a team is picked up without a restart; it's one cheap request.
78    /// Best effort: if the token can't list teams (needs `read:org`), team requests
79    /// just won't be detected.
80    async fn viewer_team_keys(&self) -> HashSet<String> {
81        match self.get_all::<GithubTeam>("/user/teams").await {
82            Ok(teams) => teams
83                .into_iter()
84                .map(|t| team_key(&t.organization.login, &t.slug))
85                .collect(),
86            Err(e) => {
87                warn!(error = %e, "could not list your teams; team review requests won't be detected");
88                HashSet::new()
89            }
90        }
91    }
92
93    /// The authenticated user's login, fetched once and cached.
94    async fn viewer_login(&self) -> Result<&str, SourceError> {
95        self.viewer
96            .get_or_try_init(|| async {
97                let me: User = self.octo.get("/user", None::<&()>).await.map_err(map_err)?;
98                Ok::<_, SourceError>(me.login)
99            })
100            .await
101            .map(String::as_str)
102    }
103
104    /// List notifications updated since `since` (RFC3339), across pages.
105    async fn notifications(&self, since: Option<&str>) -> Result<Vec<Notification>, SourceError> {
106        #[derive(Serialize)]
107        struct Params<'a> {
108            all: bool,
109            per_page: u8,
110            page: u8,
111            #[serde(skip_serializing_if = "Option::is_none")]
112            since: Option<&'a str>,
113        }
114
115        let mut out = Vec::new();
116        for page in 1..=MAX_PAGES {
117            let params = Params {
118                all: true,
119                per_page: 100,
120                page,
121                since,
122            };
123            let batch: Vec<Notification> = self
124                .octo
125                .get("/notifications", Some(&params))
126                .await
127                .map_err(map_err)?;
128            let n = batch.len();
129            out.extend(batch);
130            if n < 100 {
131                break;
132            }
133        }
134        Ok(out)
135    }
136
137    /// Fetch a page-collected list from a repo sub-resource path.
138    async fn get_all<T>(&self, path: &str) -> Result<Vec<T>, SourceError>
139    where
140        T: serde::de::DeserializeOwned,
141    {
142        #[derive(Serialize)]
143        struct Page {
144            per_page: u8,
145            page: u8,
146        }
147        let mut out = Vec::new();
148        for page in 1..=MAX_PAGES {
149            let batch: Vec<T> = self
150                .octo
151                .get(
152                    path,
153                    Some(&Page {
154                        per_page: 100,
155                        page,
156                    }),
157                )
158                .await
159                .map_err(map_err)?;
160            let n = batch.len();
161            out.extend(batch);
162            if n < 100 {
163                break;
164            }
165        }
166        Ok(out)
167    }
168
169    /// Fetch everything the diff needs for one PR.
170    async fn fetch_pr(&self, owner: &str, repo: &str, number: u64) -> Result<PrData, SourceError> {
171        let pr: PullRequest = self
172            .octo
173            .get(format!("/repos/{owner}/{repo}/pulls/{number}"), None::<&()>)
174            .await
175            .map_err(map_err)?;
176        let reviews: Vec<Review> = self
177            .get_all(&format!("/repos/{owner}/{repo}/pulls/{number}/reviews"))
178            .await?;
179        let review_comments: Vec<ReviewComment> = self
180            .get_all(&format!("/repos/{owner}/{repo}/pulls/{number}/comments"))
181            .await?;
182        let issue_comments: Vec<IssueComment> = self
183            .get_all(&format!("/repos/{owner}/{repo}/issues/{number}/comments"))
184            .await?;
185        Ok(PrData {
186            pull_request: pr,
187            reviews,
188            review_comments,
189            issue_comments,
190        })
191    }
192}
193
194#[async_trait]
195impl Source for GitHubSource {
196    fn id(&self) -> &str {
197        SOURCE_ID
198    }
199
200    async fn poll(&self, state: &dyn StateStore) -> Result<Vec<Event>, SourceError> {
201        let viewer = self.viewer_login().await?.to_string();
202        let viewer_teams = self.viewer_team_keys().await;
203        let poll_start = OffsetDateTime::now_utc();
204        let since = state.get_cursor(SOURCE_ID, "notif_since").await?;
205
206        let notifs = self.notifications(since.as_deref()).await?;
207        debug!(count = notifs.len(), "fetched notifications");
208
209        let mut events = Vec::new();
210        for n in &notifs {
211            if n.subject.kind != "PullRequest" {
212                continue;
213            }
214            let Some((owner, repo, number)) = n.subject.url.as_deref().and_then(parse_pr_url)
215            else {
216                warn!(url = ?n.subject.url, "could not parse PR url from notification");
217                continue;
218            };
219            let scope = format!("{owner}/{repo}#{number}");
220
221            // Skip threads whose activity we've already processed. An optimisation
222            // only; the snapshot would suppress duplicates anyway.
223            let seen_key = format!("thread:{scope}");
224            let last_seen = state.get_cursor(SOURCE_ID, &seen_key).await?;
225            if let (Some(seen), Some(updated)) = (&last_seen, &n.updated_at) {
226                if updated.as_str() <= seen.as_str() {
227                    continue;
228                }
229            }
230
231            let pr_data = match self.fetch_pr(&owner, &repo, number).await {
232                Ok(d) => d,
233                Err(e) => {
234                    // One inaccessible PR (deleted, perms) shouldn't abort the whole poll.
235                    warn!(%scope, error = %e, "failed to fetch PR; skipping");
236                    continue;
237                }
238            };
239
240            let old: PrSnapshot = match state.get_snapshot(SOURCE_ID, &scope).await? {
241                Some(bytes) => serde_json::from_slice(&bytes)
242                    .map_err(|e| SourceError::Parse(format!("snapshot {scope}: {e}")))?,
243                None => PrSnapshot::default(),
244            };
245
246            let ctx = DiffContext {
247                source_id: SOURCE_ID.to_string(),
248                viewer_login: viewer.clone(),
249                repo: Repo {
250                    owner: owner.clone(),
251                    name: repo.clone(),
252                    url: n.repository.html_url.clone(),
253                },
254                now: poll_start,
255                first_sight_since: first_sight_watermark(n.updated_at.as_deref()),
256                viewer_teams: viewer_teams.clone(),
257            };
258            let (evs, new_snapshot) = diff(&ctx, &pr_data, &old);
259
260            let bytes = serde_json::to_vec(&new_snapshot)
261                .map_err(|e| SourceError::Parse(format!("serialize snapshot {scope}: {e}")))?;
262            state.put_snapshot(SOURCE_ID, &scope, &bytes).await?;
263            if let Some(updated) = &n.updated_at {
264                state.put_cursor(SOURCE_ID, &seen_key, updated).await?;
265            }
266
267            events.extend(evs);
268        }
269
270        // Advance the list cursor with a small overlap so nothing straddling the
271        // boundary is missed on the next poll.
272        let next_since = (poll_start - SINCE_OVERLAP)
273            .format(&Rfc3339)
274            .map_err(|e| SourceError::Other(Box::new(e)))?;
275        state
276            .put_cursor(SOURCE_ID, "notif_since", &next_since)
277            .await?;
278
279        Ok(events)
280    }
281}
282
283/// Parse `https://api.github.com/repos/{owner}/{repo}/pulls/{number}` into parts.
284fn parse_pr_url(url: &str) -> Option<(String, String, u64)> {
285    let after = url.split("/repos/").nth(1)?;
286    let mut parts = after.split('/');
287    let owner = parts.next()?.to_string();
288    let repo = parts.next()?.to_string();
289    let kind = parts.next()?; // "pulls"
290    if kind != "pulls" {
291        return None;
292    }
293    let number: u64 = parts.next()?.parse().ok()?;
294    Some((owner, repo, number))
295}
296
297fn map_err(err: octocrab::Error) -> SourceError {
298    classify_github_error(&err.to_string())
299}
300
301/// Classify a GitHub error message. A 403 is only a rate limit when the message
302/// says so (an unauthenticated or over-quota call); a plain 403 is a permission
303/// problem, not something to silently retry.
304fn classify_github_error(msg: &str) -> SourceError {
305    let lower = msg.to_ascii_lowercase();
306    if lower.contains("rate limit") {
307        SourceError::RateLimited {
308            retry_after_secs: 60,
309        }
310    } else if lower.contains("bad credentials")
311        || lower.contains("unauthorized")
312        || lower.contains("401")
313    {
314        SourceError::Auth(format!("invalid GitHub token: {msg}"))
315    } else if lower.contains("forbidden")
316        || lower.contains("resource not accessible")
317        || lower.contains("403")
318    {
319        SourceError::Auth(format!(
320            "GitHub returned 403 (forbidden); the token likely lacks required scopes \
321             (notifications + repo/PR read): {msg}"
322        ))
323    } else {
324        SourceError::Request(msg.to_string())
325    }
326}
327
328#[cfg(test)]
329mod tests {
330    use super::{classify_github_error, parse_pr_url};
331    use navi_notifier_core::SourceError;
332
333    #[test]
334    fn rate_limit_messages_are_rate_limited() {
335        assert!(matches!(
336            classify_github_error("API rate limit exceeded for 1.2.3.4"),
337            SourceError::RateLimited { .. }
338        ));
339        assert!(matches!(
340            classify_github_error("You have exceeded a secondary rate limit"),
341            SourceError::RateLimited { .. }
342        ));
343    }
344
345    #[test]
346    fn bad_credentials_is_auth() {
347        assert!(matches!(
348            classify_github_error("Bad credentials"),
349            SourceError::Auth(_)
350        ));
351    }
352
353    #[test]
354    fn forbidden_is_auth_not_rate_limited() {
355        match classify_github_error("Resource not accessible by personal access token") {
356            SourceError::Auth(m) => assert!(m.contains("403")),
357            other => panic!("expected Auth, got {other:?}"),
358        }
359    }
360
361    #[test]
362    fn other_errors_are_request() {
363        assert!(matches!(
364            classify_github_error("connection reset by peer"),
365            SourceError::Request(_)
366        ));
367    }
368
369    #[test]
370    fn parses_pr_url() {
371        assert_eq!(
372            parse_pr_url("https://api.github.com/repos/acme/widgets/pulls/12"),
373            Some(("acme".into(), "widgets".into(), 12))
374        );
375    }
376
377    #[test]
378    fn rejects_non_pull_urls() {
379        assert_eq!(
380            parse_pr_url("https://api.github.com/repos/acme/widgets/issues/12"),
381            None
382        );
383    }
384}
navi_notifier_github/source.rs

navi_notifier_github/
source.rs
