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 [`crate::diff`] module.
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 octocrab::Octocrab;
10use serde::Serialize;
11use time::format_description::well_known::Rfc3339;
12use time::{Duration, OffsetDateTime};
13use tokio::sync::OnceCell;
14use tracing::{debug, warn};
15
16use crate::api::{IssueComment, Notification, PrData, PullRequest, Review, ReviewComment, User};
17use crate::diff::{diff, DiffContext};
18use crate::snapshot::PrSnapshot;
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        let mut builder = Octocrab::builder().personal_token(config.token);
42        if let Some(base) = config.api_base {
43            builder = builder
44                .base_uri(base)
45                .map_err(|e| SourceError::Request(format!("invalid api_base: {e}")))?;
46        }
47        let octo = builder
48            .build()
49            .map_err(|e| SourceError::Auth(e.to_string()))?;
50        Ok(Self {
51            octo,
52            viewer: OnceCell::new(),
53        })
54    }
55
56    /// The authenticated user's login, fetched once and cached.
57    async fn viewer_login(&self) -> Result<&str, SourceError> {
58        self.viewer
59            .get_or_try_init(|| async {
60                let me: User = self.octo.get("/user", None::<&()>).await.map_err(map_err)?;
61                Ok::<_, SourceError>(me.login)
62            })
63            .await
64            .map(String::as_str)
65    }
66
67    /// List notifications updated since `since` (RFC3339), across pages.
68    async fn notifications(&self, since: Option<&str>) -> Result<Vec<Notification>, SourceError> {
69        #[derive(Serialize)]
70        struct Params<'a> {
71            all: bool,
72            per_page: u8,
73            page: u8,
74            #[serde(skip_serializing_if = "Option::is_none")]
75            since: Option<&'a str>,
76        }
77
78        let mut out = Vec::new();
79        for page in 1..=MAX_PAGES {
80            let params = Params {
81                all: true,
82                per_page: 100,
83                page,
84                since,
85            };
86            let batch: Vec<Notification> = self
87                .octo
88                .get("/notifications", Some(&params))
89                .await
90                .map_err(map_err)?;
91            let n = batch.len();
92            out.extend(batch);
93            if n < 100 {
94                break;
95            }
96        }
97        Ok(out)
98    }
99
100    /// Fetch a page-collected list from a repo sub-resource path.
101    async fn get_all<T>(&self, path: &str) -> Result<Vec<T>, SourceError>
102    where
103        T: serde::de::DeserializeOwned,
104    {
105        #[derive(Serialize)]
106        struct Page {
107            per_page: u8,
108            page: u8,
109        }
110        let mut out = Vec::new();
111        for page in 1..=MAX_PAGES {
112            let batch: Vec<T> = self
113                .octo
114                .get(
115                    path,
116                    Some(&Page {
117                        per_page: 100,
118                        page,
119                    }),
120                )
121                .await
122                .map_err(map_err)?;
123            let n = batch.len();
124            out.extend(batch);
125            if n < 100 {
126                break;
127            }
128        }
129        Ok(out)
130    }
131
132    /// Fetch everything the diff needs for one PR.
133    async fn fetch_pr(&self, owner: &str, repo: &str, number: u64) -> Result<PrData, SourceError> {
134        let pr: PullRequest = self
135            .octo
136            .get(format!("/repos/{owner}/{repo}/pulls/{number}"), None::<&()>)
137            .await
138            .map_err(map_err)?;
139        let reviews: Vec<Review> = self
140            .get_all(&format!("/repos/{owner}/{repo}/pulls/{number}/reviews"))
141            .await?;
142        let review_comments: Vec<ReviewComment> = self
143            .get_all(&format!("/repos/{owner}/{repo}/pulls/{number}/comments"))
144            .await?;
145        let issue_comments: Vec<IssueComment> = self
146            .get_all(&format!("/repos/{owner}/{repo}/issues/{number}/comments"))
147            .await?;
148        Ok(PrData {
149            pull_request: pr,
150            reviews,
151            review_comments,
152            issue_comments,
153        })
154    }
155}
156
157#[async_trait]
158impl Source for GitHubSource {
159    fn id(&self) -> &str {
160        SOURCE_ID
161    }
162
163    async fn poll(&self, state: &dyn StateStore) -> Result<Vec<Event>, SourceError> {
164        let viewer = self.viewer_login().await?.to_string();
165        let poll_start = OffsetDateTime::now_utc();
166        let since = state.get_cursor(SOURCE_ID, "notif_since").await?;
167
168        let notifs = self.notifications(since.as_deref()).await?;
169        debug!(count = notifs.len(), "fetched notifications");
170
171        let mut events = Vec::new();
172        for n in &notifs {
173            if n.subject.kind != "PullRequest" {
174                continue;
175            }
176            let Some((owner, repo, number)) = n.subject.url.as_deref().and_then(parse_pr_url)
177            else {
178                warn!(url = ?n.subject.url, "could not parse PR url from notification");
179                continue;
180            };
181            let scope = format!("{owner}/{repo}#{number}");
182
183            // Skip threads whose activity we've already processed (pure optimisation —
184            // the snapshot would suppress duplicates anyway).
185            let seen_key = format!("thread:{scope}");
186            let last_seen = state.get_cursor(SOURCE_ID, &seen_key).await?;
187            if let (Some(seen), Some(updated)) = (&last_seen, &n.updated_at) {
188                if updated.as_str() <= seen.as_str() {
189                    continue;
190                }
191            }
192
193            let pr_data = match self.fetch_pr(&owner, &repo, number).await {
194                Ok(d) => d,
195                Err(e) => {
196                    // One inaccessible PR (deleted, perms) shouldn't abort the whole poll.
197                    warn!(%scope, error = %e, "failed to fetch PR; skipping");
198                    continue;
199                }
200            };
201
202            let old: PrSnapshot = match state.get_snapshot(SOURCE_ID, &scope).await? {
203                Some(bytes) => serde_json::from_slice(&bytes)
204                    .map_err(|e| SourceError::Parse(format!("snapshot {scope}: {e}")))?,
205                None => PrSnapshot::default(),
206            };
207
208            let ctx = DiffContext {
209                source_id: SOURCE_ID.to_string(),
210                viewer_login: viewer.clone(),
211                repo: Repo {
212                    owner: owner.clone(),
213                    name: repo.clone(),
214                    url: n.repository.html_url.clone(),
215                },
216                now: poll_start,
217            };
218            let (evs, new_snapshot) = diff(&ctx, &pr_data, &old);
219
220            let bytes = serde_json::to_vec(&new_snapshot)
221                .map_err(|e| SourceError::Parse(format!("serialize snapshot {scope}: {e}")))?;
222            state.put_snapshot(SOURCE_ID, &scope, &bytes).await?;
223            if let Some(updated) = &n.updated_at {
224                state.put_cursor(SOURCE_ID, &seen_key, updated).await?;
225            }
226
227            events.extend(evs);
228        }
229
230        // Advance the list cursor with a small overlap so nothing straddling the
231        // boundary is missed on the next poll.
232        let next_since = (poll_start - SINCE_OVERLAP)
233            .format(&Rfc3339)
234            .map_err(|e| SourceError::Other(Box::new(e)))?;
235        state
236            .put_cursor(SOURCE_ID, "notif_since", &next_since)
237            .await?;
238
239        Ok(events)
240    }
241}
242
243/// Parse `https://api.github.com/repos/{owner}/{repo}/pulls/{number}` into parts.
244fn parse_pr_url(url: &str) -> Option<(String, String, u64)> {
245    let after = url.split("/repos/").nth(1)?;
246    let mut parts = after.split('/');
247    let owner = parts.next()?.to_string();
248    let repo = parts.next()?.to_string();
249    let kind = parts.next()?; // "pulls"
250    if kind != "pulls" {
251        return None;
252    }
253    let number: u64 = parts.next()?.parse().ok()?;
254    Some((owner, repo, number))
255}
256
257/// Map an octocrab error into a [`SourceError`], recognising rate limiting.
258fn map_err(err: octocrab::Error) -> SourceError {
259    let msg = err.to_string();
260    if msg.contains("rate limit") || msg.contains("403") {
261        return SourceError::RateLimited {
262            retry_after_secs: 60,
263        };
264    }
265    if msg.contains("401") || msg.contains("Bad credentials") {
266        return SourceError::Auth(msg);
267    }
268    SourceError::Request(msg)
269}
270
271#[cfg(test)]
272mod tests {
273    use super::parse_pr_url;
274
275    #[test]
276    fn parses_pr_url() {
277        assert_eq!(
278            parse_pr_url("https://api.github.com/repos/acme/widgets/pulls/12"),
279            Some(("acme".into(), "widgets".into(), 12))
280        );
281    }
282
283    #[test]
284    fn rejects_non_pull_urls() {
285        assert_eq!(
286            parse_pr_url("https://api.github.com/repos/acme/widgets/issues/12"),
287            None
288        );
289    }
290}
navi_notifier_github/source.rs

navi_notifier_github/
source.rs
