trendkit_core/
github.rs

1//! GitHub source: Search-API trending collection and user repo listing.
2//!
3//! Two clients, two policies, on purpose:
4//!
5//! - `repos` client keeps fetch-kit's built-in retry (3 attempts) — the
6//!   core REST API allows 5000 req/h and transient failures are cheap.
7//! - `search` client disables built-in retry and drives its own loop:
8//!   requests are paced to the authenticated Search budget (30 req/min)
9//!   and 403/429 answers back off via `retry-backoff`, whose exponential
10//!   schedule reaches 60s — beyond fetch-kit's 30s built-in max.
11
12use std::time::Duration;
13
14use fetch_kit::{Client, FetchError};
15use retry_backoff::backoffs::impl_exponential_backoff::Backoff;
16use retry_backoff::RetryBackoff;
17use serde::de::DeserializeOwned;
18use serde::{Deserialize, Serialize};
19
20use crate::clock::{iso_date_from_unix, unix_seconds};
21use crate::error::{Result, TrendError};
22use crate::model::{Observation, RepoListed, SourceKind, SubjectRecord};
23use crate::source::{CollectCtx, CollectReport, TrendSource, SEARCH_PACING};
24
25const API_ROOT: &str = "https://api.github.com";
26const SEARCH_PATH: &str = "/search/repositories";
27const USER_AGENT: &str = "trendkit/0.2";
28const API_VERSION: &str = "2022-11-28";
29const PER_PAGE: u32 = 100;
30/// Attempts per search page before declaring the endpoint rate-limited.
31const RATE_LIMIT_ATTEMPTS: u32 = 4;
32/// Cap for profile listings: 5 pages x 100 covers 500 repos.
33const USER_REPOS_MAX_PAGES: u32 = 5;
34
35/// GitHub trend source.
36#[derive(Debug, Clone)]
37pub struct GithubSource {
38    search: Client,
39    repos: Client,
40    token: Option<String>,
41    backoff: Backoff,
42}
43
44impl GithubSource {
45    /// Build a source; `token` is a GitHub PAT (env `GH_TOKEN` in the CLI).
46    #[must_use]
47    pub fn new(token: Option<String>) -> GithubSource {
48        let search = Client::builder()
49            .user_agent(USER_AGENT)
50            // HeaderName::from_static demands lowercase — uppercase panics.
51            .default_header("accept", "application/vnd.github+json")
52            .default_header("x-github-api-version", API_VERSION)
53            .timeout(Duration::from_secs(30))
54            .retries(0)
55            .build();
56        let repos = Client::builder()
57            .user_agent(USER_AGENT)
58            .default_header("accept", "application/vnd.github+json")
59            .default_header("x-github-api-version", API_VERSION)
60            .timeout(Duration::from_secs(30))
61            .build();
62        let backoff = Backoff::new(Duration::from_secs(2), Some(Duration::from_secs(60)));
63        GithubSource {
64            search,
65            repos,
66            token,
67            backoff,
68        }
69    }
70
71    /// Collect trending new repos: created within the window, sorted by stars.
72    async fn search_trending(&self, ctx: &CollectCtx) -> Result<usize> {
73        let now = unix_seconds(ctx.clock.as_ref());
74        let since = iso_date_from_unix(now - i64::from(ctx.window_days) * 86_400);
75        let query = format!("created:>{since}");
76        let mut total = 0usize;
77
78        for page in 1..=ctx.pages {
79            let params = SearchQuery {
80                q: &query,
81                sort: "stars",
82                order: "desc",
83                per_page: PER_PAGE,
84                page,
85            };
86            let response: SearchResponse = self.search_get(SEARCH_PATH, &params).await?;
87            let count = response.items.len();
88            for item in response.items {
89                let observation = Observation {
90                    ts: now,
91                    score: item.stargazers_count,
92                    extra1: item.forks_count,
93                    extra2: item.watchers_count,
94                };
95                let subject = SubjectRecord::from(item);
96                let subject_id = ctx.store.upsert_subject(&subject)?;
97                ctx.store.insert_observation(subject_id, &observation)?;
98                total += 1;
99            }
100            if count < PER_PAGE as usize || page == ctx.pages {
101                break;
102            }
103            tokio::time::sleep(SEARCH_PACING).await;
104        }
105
106        ctx.store
107            .record_run(now, self.name(), i64::from(ctx.pages), total as i64)?;
108        Ok(total)
109    }
110
111    /// GET with search-endpoint rate-limit handling (403/429 → backoff).
112    async fn search_get<Q, T>(&self, path: &str, params: &Q) -> Result<T>
113    where
114        Q: Serialize,
115        T: DeserializeOwned,
116    {
117        let url = format!("{API_ROOT}{path}");
118        let mut attempt: u32 = 0;
119        loop {
120            attempt += 1;
121            let mut request = self.search.get(&url).query(params);
122            if let Some(token) = &self.token {
123                request = request.bearer_auth(token);
124            }
125            match request.json_response::<T>().await {
126                Ok(value) => return Ok(value),
127                Err(FetchError::StatusCode {
128                    status: 403 | 429, ..
129                }) => {
130                    if attempt >= RATE_LIMIT_ATTEMPTS {
131                        return Err(TrendError::RateLimited {
132                            endpoint: "search/repositories".to_owned(),
133                            attempts: RATE_LIMIT_ATTEMPTS,
134                        });
135                    }
136                    let wait = RetryBackoff::delay(&self.backoff, attempt as usize);
137                    tracing::warn!(
138                        attempt,
139                        wait_secs = wait.as_secs(),
140                        "github search rate limited; backing off"
141                    );
142                    tokio::time::sleep(wait).await;
143                }
144                Err(err) => return Err(TrendError::Fetch(err)),
145            }
146        }
147    }
148
149    /// List a user's own (non-fork-inclusive) repos for the personal profile.
150    ///
151    /// Core REST endpoint: fetch-kit's built-in retry applies, no manual
152    /// pacing needed at 5000 req/h.
153    pub async fn list_user_repos(&self, owner: &str) -> Result<Vec<RepoListed>> {
154        let mut out = Vec::new();
155        for page in 1..=USER_REPOS_MAX_PAGES {
156            let url = format!("{API_ROOT}/users/{owner}/repos");
157            let params = ListQuery {
158                per_page: PER_PAGE,
159                page,
160                sort: "pushed",
161                list_type: "owner",
162            };
163            let response: Vec<RepoJson> = self.repos_get(&url, &params).await?;
164            let count = response.len();
165            out.extend(response.into_iter().map(RepoListed::from));
166            if count < PER_PAGE as usize {
167                break;
168            }
169        }
170        Ok(out)
171    }
172
173    async fn repos_get<Q, T>(&self, url: &str, params: &Q) -> Result<T>
174    where
175        Q: Serialize,
176        T: DeserializeOwned,
177    {
178        let mut request = self.repos.get(url).query(params);
179        if let Some(token) = &self.token {
180            request = request.bearer_auth(token);
181        }
182        Ok(request.json_response::<T>().await?)
183    }
184}
185
186impl TrendSource for GithubSource {
187    fn name(&self) -> &'static str {
188        "github"
189    }
190
191    // The future is deliberately not `Send`: the SQLite connection is
192    // confined to the collecting task, and API sources are sequential by
193    // design (pacing). Multi-source concurrency happens above one store.
194    fn collect(
195        &self,
196        ctx: &CollectCtx,
197    ) -> impl std::future::Future<Output = Result<CollectReport>> {
198        let ctx = ctx.clone();
199        async move {
200            let items = self.search_trending(&ctx).await?;
201            Ok(CollectReport {
202                source: self.name(),
203                items,
204                pages: ctx.pages as usize,
205            })
206        }
207    }
208}
209
210#[derive(Debug, Serialize)]
211struct SearchQuery<'a> {
212    q: &'a str,
213    sort: &'a str,
214    order: &'a str,
215    per_page: u32,
216    page: u32,
217}
218
219#[derive(Debug, Serialize)]
220struct ListQuery<'a> {
221    per_page: u32,
222    page: u32,
223    sort: &'a str,
224    #[serde(rename = "type")]
225    list_type: &'a str,
226}
227
228/// Deserialize GitHub search results.
229#[derive(Debug, Deserialize)]
230struct SearchResponse {
231    #[serde(default)]
232    items: Vec<RepoJson>,
233}
234
235/// The subset of the GitHub repository payload trendkit consumes.
236#[derive(Debug, Clone, Deserialize)]
237#[allow(dead_code)] // counter fields are payload-derived and kept for fidelity
238pub(crate) struct RepoJson {
239    #[serde(default)]
240    pub(crate) full_name: String,
241    #[serde(default)]
242    pub(crate) owner: Option<Owner>,
243    #[serde(default)]
244    pub(crate) name: Option<String>,
245    #[serde(default)]
246    pub(crate) description: Option<String>,
247    #[serde(default)]
248    pub(crate) language: Option<String>,
249    #[serde(default)]
250    pub(crate) html_url: Option<String>,
251    #[serde(default)]
252    pub(crate) created_at: Option<String>,
253    #[serde(default)]
254    pub(crate) pushed_at: Option<String>,
255    #[serde(default)]
256    pub(crate) stargazers_count: i64,
257    #[serde(default)]
258    pub(crate) forks_count: i64,
259    #[serde(default)]
260    pub(crate) watchers_count: i64,
261    #[serde(default)]
262    pub(crate) open_issues_count: i64,
263    #[serde(default)]
264    pub(crate) fork: bool,
265    #[serde(default)]
266    pub(crate) topics: Vec<String>,
267}
268
269/// `owner` object; only the login is used.
270#[derive(Debug, Clone, Deserialize)]
271#[allow(dead_code)] // payload-derived; kept for payload fidelity and debugging
272pub(crate) struct Owner {
273    pub(crate) login: String,
274}
275
276impl From<RepoJson> for SubjectRecord {
277    fn from(json: RepoJson) -> Self {
278        let full_name = json.full_name;
279        let url = json.html_url.clone().unwrap_or_else(|| {
280            if full_name.is_empty() {
281                String::new()
282            } else if full_name.contains('/') {
283                format!("{GITHUB_WEB_ROOT}/{full_name}")
284            } else {
285                // Degenerate payload without an owner object: synthesize
286                // a same-name owner path so the subject still gets a URL.
287                format!("{GITHUB_WEB_ROOT}/{full_name}/{full_name}")
288            }
289        });
290        SubjectRecord {
291            kind: SourceKind::GithubRepo,
292            key: full_name,
293            title: json.description,
294            lang: json.language,
295            url: Some(url),
296            topics: json.topics,
297            created_at: json.created_at,
298        }
299    }
300}
301
302/// Web root used to synthesize repo URLs when the payload omits them.
303pub const GITHUB_WEB_ROOT: &str = "https://github.com";
304
305impl From<RepoJson> for RepoListed {
306    fn from(json: RepoJson) -> Self {
307        RepoListed {
308            full_name: json.full_name,
309            language: json.language,
310            topics: json.topics,
311            description: json.description,
312            is_fork: json.fork,
313            pushed_at: json.pushed_at,
314            stars: json.stargazers_count,
315        }
316    }
317}
318
319/// Parse a GitHub search response body into subjects (used by tests and
320/// available to callers who capture raw payloads).
321///
322/// # Errors
323/// Fails when the payload is not a GitHub search response.
324pub fn parse_search_json(payload: &str) -> Result<Vec<SubjectRecord>> {
325    let response: SearchResponse = serde_json::from_str(payload)
326        .map_err(|e| TrendError::Config(format!("invalid search payload: {e}")))?;
327    Ok(response
328        .items
329        .into_iter()
330        .map(SubjectRecord::from)
331        .collect())
332}
333
334#[cfg(test)]
335mod tests {
336    #![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
337
338    use super::*;
339
340    const SAMPLE: &str = r#"{
341        "total_count": 2,
342        "incomplete_results": false,
343        "items": [
344            {
345                "id": 1,
346                "full_name": "tokio-rs/tokio",
347                "name": "tokio",
348                "owner": { "login": "tokio-rs" },
349                "description": "A runtime for Rust",
350                "language": "Rust",
351                "html_url": "https://github.com/tokio-rs/tokio",
352                "created_at": "2026-10-10T00:00:00Z",
353                "pushed_at": "2026-10-16T00:00:00Z",
354                "stargazers_count": 42,
355                "forks_count": 3,
356                "watchers_count": 42,
357                "open_issues_count": 5,
358                "fork": false,
359                "topics": ["async", "runtime"]
360            },
361            {
362                "id": 2,
363                "full_name": "sparse-repo",
364                "stargazers_count": 7
365            }
366        ]
367    }"#;
368
369    #[test]
370    fn parse_extracts_subjects_with_fallbacks() {
371        let subjects = parse_search_json(SAMPLE).unwrap();
372        assert_eq!(subjects.len(), 2);
373
374        assert_eq!(subjects[0].kind, SourceKind::GithubRepo);
375        assert_eq!(subjects[0].key, "tokio-rs/tokio");
376        assert_eq!(subjects[0].title.as_deref(), Some("A runtime for Rust"));
377        assert_eq!(subjects[0].lang.as_deref(), Some("Rust"));
378        assert_eq!(
379            subjects[0].url.as_deref(),
380            Some("https://github.com/tokio-rs/tokio")
381        );
382        assert_eq!(
383            subjects[0].topics,
384            vec!["async".to_owned(), "runtime".to_owned()]
385        );
386
387        // Degenerate second item: url synthesized from full_name.
388        assert_eq!(subjects[1].key, "sparse-repo");
389        assert_eq!(
390            subjects[1].url.as_deref(),
391            Some("https://github.com/sparse-repo/sparse-repo")
392        );
393    }
394
395    #[test]
396    fn listed_rows_carry_counters() {
397        let response: SearchResponse = serde_json::from_str(SAMPLE).unwrap();
398        let listed: Vec<RepoListed> = response.items.into_iter().map(RepoListed::from).collect();
399        assert_eq!(listed[0].stars, 42);
400        assert!(!listed[0].is_fork);
401        assert_eq!(listed[0].topics.len(), 2);
402    }
403
404    #[test]
405    fn parse_rejects_garbage() {
406        assert!(parse_search_json("not json").is_err());
407    }
408}
trendkit_core/github.rs

trendkit_core/
github.rs
