use std::cmp::Ordering;
use std::collections::{HashMap, HashSet};
use std::time::Duration;
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use futures::stream::{FuturesUnordered, StreamExt};
use octocrab::Octocrab;
use serde::Deserialize;
use thiserror::Error;
use tokio::time::{self, timeout};
const FETCH_TIMEOUT: Duration = Duration::from_secs(30);
const MAX_CONCURRENT_FETCHES: usize = 4;
#[derive(Debug, Clone)]
pub struct RepoItem {
pub kind: ItemKind,
pub number: u64,
pub title: String,
pub created_at: DateTime<Utc>,
pub updated_at: DateTime<Utc>,
pub author: String,
pub pr_draft: Option<bool>,
}
#[derive(Debug, Clone, PartialEq)]
pub enum ItemKind {
PullRequest,
Issue,
}
pub type SubscribedItems = HashMap<String, HashSet<u64>>;
#[derive(Debug, Clone)]
pub struct RepoResult {
pub repo: String,
pub status: RepoStatus,
}
#[derive(Debug, Clone)]
pub enum RepoStatus {
Items(Vec<RepoItem>),
NotFound,
Error(RepoError),
}
#[derive(Debug, Clone, PartialEq, Eq, Error)]
pub enum RepoError {
#[error("timeout after 30s")]
Timeout,
#[error("{0}")]
Api(String),
}
#[derive(Debug, Error)]
pub enum GithubError {
#[error("repo not found: {0}")]
NotFound(String),
#[error("api error: {0}")]
Api(#[from] octocrab::Error),
}
fn map_github_err<T>(
res: std::result::Result<T, octocrab::Error>,
repo_label: &str,
) -> std::result::Result<T, GithubError> {
match res {
Ok(v) => Ok(v),
Err(octocrab::Error::GitHub { source, .. }) if source.status_code.as_u16() == 404 => {
Err(GithubError::NotFound(repo_label.to_owned()))
}
Err(e) => Err(GithubError::Api(e)),
}
}
pub(crate) fn describe_api_error(e: &octocrab::Error) -> String {
match e {
octocrab::Error::GitHub { source, .. } => {
let mut message = format!("HTTP {} {}", source.status_code, source.message);
if source.status_code.as_u16() == 401 {
message.push_str(" — set GITHUB_TOKEN");
}
message
}
other => other.to_string(),
}
}
pub fn split_repo(s: &str) -> Option<(&str, &str)> {
let (owner, name) = s.split_once('/')?;
if owner.is_empty() || name.is_empty() {
return None;
}
Some((owner, name))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AccountKind {
User,
Organization,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ListSource {
Authenticated,
Org(String),
PublicUser(String),
}
pub fn account_kind_from_type(profile_type: &str) -> AccountKind {
if profile_type == "Organization" {
AccountKind::Organization
} else {
AccountKind::User
}
}
pub fn resolve_list_source(
all: bool,
username: Option<&str>,
auth_login: &str,
kind: Option<AccountKind>,
) -> ListSource {
if all {
return ListSource::Authenticated;
}
let Some(username) = username else {
return ListSource::Authenticated;
};
if username.eq_ignore_ascii_case(auth_login) {
return ListSource::Authenticated;
}
match kind {
Some(AccountKind::Organization) => ListSource::Org(username.to_owned()),
_ => ListSource::PublicUser(username.to_owned()),
}
}
pub fn item_cmp(a: &RepoItem, b: &RepoItem) -> Ordering {
match (&a.kind, &b.kind) {
(ItemKind::PullRequest, ItemKind::Issue) => Ordering::Less,
(ItemKind::Issue, ItemKind::PullRequest) => Ordering::Greater,
_ => b.number.cmp(&a.number),
}
}
pub async fn list_user_repos(crab: &Octocrab, username: &str) -> Result<Vec<String>> {
let first_page = crab
.users(username)
.repos()
.r#type(octocrab::params::users::repos::Type::Owner)
.per_page(100)
.send()
.await
.context("listing user repositories")?;
let all_pages = crab
.all_pages(first_page)
.await
.context("paginating user repositories")?;
let mut names: Vec<String> = all_pages.into_iter().filter_map(|r| r.full_name).collect();
names.sort();
Ok(names)
}
pub async fn list_authenticated_repos(crab: &Octocrab) -> Result<Vec<String>> {
let first_page = crab
.current()
.list_repos_for_authenticated_user()
.visibility("all")
.affiliation("owner,collaborator,organization_member")
.per_page(100)
.send()
.await
.context("listing repositories for the authenticated user")?;
let all_pages = crab
.all_pages(first_page)
.await
.context("paginating authenticated repositories")?;
let mut names: Vec<String> = all_pages.into_iter().filter_map(|r| r.full_name).collect();
names.sort();
names.dedup();
Ok(names)
}
pub async fn list_org_repos(crab: &Octocrab, org: &str) -> Result<Vec<String>> {
let first_page = crab
.orgs(org)
.list_repos()
.repo_type(octocrab::params::repos::Type::All)
.per_page(100)
.send()
.await
.context("listing organization repositories")?;
let all_pages = crab
.all_pages(first_page)
.await
.context("paginating organization repositories")?;
let mut names: Vec<String> = all_pages.into_iter().filter_map(|r| r.full_name).collect();
names.sort();
Ok(names)
}
pub async fn authenticated_login(crab: &Octocrab) -> Result<Option<String>> {
match crab.current().user().await {
Ok(user) => Ok(Some(user.login)),
Err(octocrab::Error::GitHub { source, .. }) if source.status_code.as_u16() == 401 => {
Ok(None)
}
Err(e) => Err(e).context("identifying the authenticated user"),
}
}
pub async fn account_kind(crab: &Octocrab, username: &str) -> Result<AccountKind> {
let profile = crab
.users(username)
.profile()
.await
.with_context(|| format!("fetching profile for {username}"))?;
Ok(account_kind_from_type(&profile.r#type))
}
pub async fn resolve_source_for(crab: &Octocrab, username: &str) -> Result<ListSource> {
let auth_login = authenticated_login(crab).await?;
if let Some(login) = &auth_login
&& login.eq_ignore_ascii_case(username)
{
return Ok(ListSource::Authenticated);
}
let kind = account_kind(crab, username).await?;
Ok(resolve_list_source(
false,
Some(username),
auth_login.as_deref().unwrap_or(""),
Some(kind),
))
}
#[derive(Debug, Deserialize)]
struct SubscribedIssue {
number: u64,
repository_url: String,
}
pub async fn fetch_subscribed_items(crab: &Octocrab) -> Result<SubscribedItems> {
let first_page = crab
.get::<octocrab::Page<SubscribedIssue>, _, _>(
"/issues?filter=subscribed&state=open&per_page=100",
None::<&()>,
)
.await
.context("listing subscribed issues and pull requests (GITHUB_TOKEN is required)")?;
let issues = crab
.all_pages(first_page)
.await
.context("paginating subscribed issues and pull requests")?;
Ok(index_subscribed_items(issues))
}
fn index_subscribed_items(issues: Vec<SubscribedIssue>) -> SubscribedItems {
let mut subscribed = SubscribedItems::new();
for issue in issues {
let Some(repo) = repo_name_from_api_url(&issue.repository_url) else {
continue;
};
subscribed.entry(repo).or_default().insert(issue.number);
}
subscribed
}
fn repo_name_from_api_url(url: &str) -> Option<String> {
let (_, path) = url.split_once("/repos/")?;
let mut segments = path.split('/');
let owner = segments.next()?;
let name = segments.next()?;
if owner.is_empty() || name.is_empty() {
return None;
}
Some(format!("{owner}/{name}").to_ascii_lowercase())
}
pub async fn fetch_repo_items(
crab: &Octocrab,
repo: &str,
subscribed_numbers: Option<&HashSet<u64>>,
) -> RepoResult {
let Some((owner, name)) = split_repo(repo) else {
return RepoResult {
repo: repo.to_owned(),
status: RepoStatus::NotFound,
};
};
match fetch_items_inner(crab, owner, name, subscribed_numbers).await {
Ok(items) => RepoResult {
repo: repo.to_owned(),
status: RepoStatus::Items(items),
},
Err(GithubError::NotFound(_)) => RepoResult {
repo: repo.to_owned(),
status: RepoStatus::NotFound,
},
Err(GithubError::Api(e)) => RepoResult {
repo: repo.to_owned(),
status: RepoStatus::Error(RepoError::Api(describe_api_error(&e))),
},
}
}
pub(crate) async fn fetch_repos_rest(
crab: &Octocrab,
repos: &[String],
subscribed: Option<&SubscribedItems>,
) -> Vec<RepoResult> {
let empty_subscriptions = HashSet::new();
let mut results = vec![None; repos.len()];
let mut in_flight = FuturesUnordered::new();
let mut next = 0;
while next < repos.len() && in_flight.len() < MAX_CONCURRENT_FETCHES {
let repo = repos[next].clone();
let repo_key = repo.to_ascii_lowercase();
let subscribed_numbers =
subscribed.map(|items| items.get(&repo_key).unwrap_or(&empty_subscriptions));
in_flight.push(fetch_repo_with_timeout(
crab,
next,
repo,
subscribed_numbers,
));
next += 1;
}
let deadline = time::sleep(FETCH_TIMEOUT);
tokio::pin!(deadline);
while !in_flight.is_empty() {
tokio::select! {
_ = &mut deadline => break,
Some((index, result)) = in_flight.next() => {
results[index] = Some(result);
if next < repos.len() {
let repo = repos[next].clone();
let repo_key = repo.to_ascii_lowercase();
let subscribed_numbers = subscribed
.map(|items| items.get(&repo_key).unwrap_or(&empty_subscriptions));
in_flight.push(fetch_repo_with_timeout(
crab,
next,
repo,
subscribed_numbers,
));
next += 1;
}
}
}
}
results
.into_iter()
.enumerate()
.map(|(index, result)| result.unwrap_or_else(|| timeout_result(repos[index].clone())))
.collect()
}
async fn fetch_repo_with_timeout(
crab: &Octocrab,
index: usize,
repo: String,
subscribed_numbers: Option<&HashSet<u64>>,
) -> (usize, RepoResult) {
let result = match timeout(
FETCH_TIMEOUT,
fetch_repo_items(crab, &repo, subscribed_numbers),
)
.await
{
Ok(result) => result,
Err(_) => timeout_result(repo),
};
(index, result)
}
fn timeout_result(repo: String) -> RepoResult {
RepoResult {
repo,
status: RepoStatus::Error(RepoError::Timeout),
}
}
async fn fetch_items_inner(
crab: &Octocrab,
owner: &str,
name: &str,
subscribed_numbers: Option<&HashSet<u64>>,
) -> std::result::Result<Vec<RepoItem>, GithubError> {
let label = format!("{owner}/{name}");
let issues_handler = crab.issues(owner, name);
let issues_future = issues_handler
.list()
.state(octocrab::params::State::Open)
.per_page(100)
.send();
let prs_handler = crab.pulls(owner, name);
let prs_future = prs_handler
.list()
.state(octocrab::params::State::Open)
.per_page(100)
.send();
let (issues_res, prs_res) = futures::future::join(issues_future, prs_future).await;
let issues_page = map_github_err(issues_res, &label)?;
let prs_page = map_github_err(prs_res, &label)?;
let all_issues = crab
.all_pages(issues_page)
.await
.map_err(GithubError::Api)?;
let all_prs = crab.all_pages(prs_page).await.map_err(GithubError::Api)?;
let mut items: Vec<RepoItem> = Vec::new();
for issue in all_issues {
if issue.pull_request.is_some() {
continue;
}
let author = issue.user.login.clone();
let created_at = issue.created_at;
let updated_at = issue.updated_at;
items.push(RepoItem {
kind: ItemKind::Issue,
number: issue.number,
title: issue.title,
created_at,
updated_at,
author,
pr_draft: None,
});
}
for pr in all_prs {
let author = pr.user.login.clone();
let created_at = pr.created_at;
let updated_at = pr.updated_at;
let pr_draft = pr.draft;
items.push(RepoItem {
kind: ItemKind::PullRequest,
number: pr.number,
title: pr.title,
created_at,
updated_at,
author,
pr_draft,
});
}
retain_subscribed(&mut items, subscribed_numbers);
items.sort_by(item_cmp);
Ok(items)
}
pub(crate) fn retain_subscribed(
items: &mut Vec<RepoItem>,
subscribed_numbers: Option<&HashSet<u64>>,
) {
if let Some(numbers) = subscribed_numbers {
items.retain(|item| numbers.contains(&item.number));
}
}
#[cfg(test)]
mod tests {
use super::*;
fn make_item(kind: ItemKind, number: u64) -> RepoItem {
RepoItem {
kind,
number,
title: format!("item {number}"),
created_at: Utc::now(),
updated_at: Utc::now(),
author: "user".into(),
pr_draft: None,
}
}
#[test]
fn account_kind_organization() {
assert_eq!(
account_kind_from_type("Organization"),
AccountKind::Organization
);
}
#[test]
fn account_kind_user() {
assert_eq!(account_kind_from_type("User"), AccountKind::User);
}
#[test]
fn account_kind_unknown_defaults_to_user() {
assert_eq!(account_kind_from_type("Bot"), AccountKind::User);
}
#[test]
fn all_flag_lists_authenticated() {
assert_eq!(
resolve_list_source(true, None, "me", None),
ListSource::Authenticated
);
}
#[test]
fn own_username_lists_authenticated() {
assert_eq!(
resolve_list_source(false, Some("me"), "me", Some(AccountKind::User)),
ListSource::Authenticated
);
}
#[test]
fn own_username_is_case_insensitive() {
assert_eq!(
resolve_list_source(false, Some("ME"), "me", Some(AccountKind::User)),
ListSource::Authenticated
);
}
#[test]
fn org_target_lists_org_repos() {
assert_eq!(
resolve_list_source(false, Some("acme"), "me", Some(AccountKind::Organization)),
ListSource::Org("acme".to_owned())
);
}
#[test]
fn third_party_user_lists_public_only() {
assert_eq!(
resolve_list_source(false, Some("octocat"), "me", Some(AccountKind::User)),
ListSource::PublicUser("octocat".to_owned())
);
}
#[test]
fn split_repo_valid() {
assert_eq!(split_repo("a/b"), Some(("a", "b")));
}
#[test]
fn split_repo_no_slash() {
assert_eq!(split_repo("abc"), None);
}
#[test]
fn split_repo_trailing_slash() {
assert_eq!(split_repo("a/"), None);
}
#[test]
fn split_repo_leading_slash() {
assert_eq!(split_repo("/b"), None);
}
#[test]
fn split_repo_many_slashes() {
assert_eq!(split_repo("a/b/c"), Some(("a", "b/c")));
}
#[test]
fn subscribed_index_and_filter_keep_only_matching_repo_items() {
let subscribed = index_subscribed_items(vec![
SubscribedIssue {
number: 7,
repository_url: "https://api.github.com/repos/Acme/Widget".into(),
},
SubscribedIssue {
number: 99,
repository_url: "https://api.github.com/users/octocat".into(),
},
]);
let mut items = vec![
make_item(ItemKind::Issue, 7),
make_item(ItemKind::PullRequest, 9),
];
retain_subscribed(&mut items, subscribed.get("acme/widget"));
assert_eq!(items.len(), 1);
assert_eq!(items[0].number, 7);
}
#[test]
fn missing_subscription_filter_keeps_all_repo_items() {
let mut items = vec![
make_item(ItemKind::Issue, 7),
make_item(ItemKind::PullRequest, 9),
];
retain_subscribed(&mut items, None);
assert_eq!(items.len(), 2);
}
#[test]
fn item_cmp_sorts_prs_before_issues_then_number_desc() {
let mut items = [
make_item(ItemKind::Issue, 5),
make_item(ItemKind::PullRequest, 2),
make_item(ItemKind::Issue, 10),
make_item(ItemKind::PullRequest, 8),
];
items.sort_by(item_cmp);
let numbers: Vec<u64> = items.iter().map(|i| i.number).collect();
assert_eq!(numbers, vec![8, 2, 10, 5]);
}
}