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

navi_notifier_github/
source.rs
