use crate::prelude::*;
use axum::{
extract::{Path, Query, State},
http::StatusCode,
response::sse::{Event, KeepAlive, Sse},
Json,
};
use futures::stream::{FuturesUnordered, StreamExt};
use std::convert::Infallible;
use super::fleet::{collect_issue_repos, workspace_root};
use super::sessions::ErrorResponse;
use crate::web::github::{self, CreateIssueRequest, UpdateIssueRequest};
use crate::web::state::WebState;
fn no_github() -> (StatusCode, Json<ErrorResponse>) {
(
StatusCode::UNAUTHORIZED,
Json(ErrorResponse {
error: "GitHub token not configured".to_string(),
}),
)
}
fn github_error(e: anyhow::Error) -> (StatusCode, Json<ErrorResponse>) {
let message = e.to_string();
if message.contains("status 401") || message.contains("status 403") {
return (
StatusCode::UNAUTHORIZED,
Json(ErrorResponse {
error: "GitHub token is invalid or lacks required access".to_string(),
}),
);
}
(
StatusCode::BAD_GATEWAY,
Json(ErrorResponse { error: message }),
)
}
#[derive(Deserialize)]
pub struct ListIssuesQuery {
pub state: Option<String>,
pub repo: Option<String>,
}
#[derive(Debug, Clone, Serialize)]
pub struct RootIssue {
pub id: String,
pub title: String,
}
#[derive(Debug, Clone, Serialize)]
pub struct IssueRepoSnapshot {
pub repo: String,
pub issues: Vec<github::GitHubIssue>,
}
#[derive(Debug, Clone, Serialize)]
pub struct IssueRepoFailure {
pub repo: String,
pub error: String,
}
#[derive(Debug, Clone, Serialize)]
pub struct IssueStreamStart {
pub repos: Vec<String>,
}
#[derive(Debug, Clone, Serialize)]
pub struct IssueStreamDone {
pub repos_completed: usize,
}
#[derive(Debug, Clone, Serialize)]
pub struct IssueAggregateResponse {
pub issues: Vec<serde_json::Value>,
pub failures: Vec<IssueRepoFailure>,
}
pub async fn list_issues(
State(state): State<WebState>,
Query(params): Query<ListIssuesQuery>,
) -> Result<Json<serde_json::Value>, (StatusCode, Json<ErrorResponse>)> {
let client = github_client(&state).await?;
let issue_state = parse_issue_state(¶ms)?;
let repo_root = workspace_root()?;
let all_repos = resolve_issue_repos(&repo_root, params.repo.as_deref())?;
let aggregate = fetch_issue_aggregate(&state, &client, &all_repos, issue_state).await?;
let mut all_issues = aggregate.issues;
let failures = aggregate
.failures
.into_iter()
.map(|failure| serde_json::to_value(failure).expect("issue failure must serialize"))
.collect::<Vec<_>>();
all_issues.sort_by(|a, b| {
let a_date = a.get("updated_at").and_then(|d| d.as_str()).unwrap_or("");
let b_date = b.get("updated_at").and_then(|d| d.as_str()).unwrap_or("");
b_date.cmp(a_date)
});
Ok(Json(serde_json::json!({
"issues": all_issues,
"failures": failures,
})))
}
pub async fn stream_issues(
State(state): State<WebState>,
Query(params): Query<ListIssuesQuery>,
) -> Result<Sse<impl futures::Stream<Item = Result<Event, Infallible>>>, (StatusCode, Json<ErrorResponse>)> {
let client = github_client(&state).await?;
let issue_state = parse_issue_state(¶ms)?;
let repo_root = workspace_root()?;
let all_repos = resolve_issue_repos(&repo_root, params.repo.as_deref())?;
let repo_names = all_repos
.iter()
.map(|(owner, repo)| format!("{owner}/{repo}"))
.collect::<Vec<_>>();
let state_for_stream = state.clone();
let stream = async_stream::stream! {
if let Ok(event) = event_json("start", &IssueStreamStart { repos: repo_names.clone() }) {
yield Ok(event);
}
match load_root_issues(&state_for_stream.db, &state_for_stream.db_service).await {
Ok(root_issues) => {
if let Ok(event) = event_json("root_issues", &root_issues) {
yield Ok(event);
}
}
Err(error) => {
if let Ok(event) = event_json("root_failure", &ErrorResponse { error: error.to_string() }) {
yield Ok(event);
}
}
}
let mut pending = FuturesUnordered::new();
for (owner, repo) in all_repos {
let repo_name = format!("{owner}/{repo}");
let client = client.clone();
let state = state_for_stream.clone();
pending.push(async move {
let fetch = client.list_issues(&owner, &repo, "all").await;
match fetch {
Ok(issues) => {
let db = state.db.clone();
let repo_name_clone = repo_name.clone();
let issues_clone = issues.clone();
tokio::task::spawn_blocking(move || {
db.session_repo.cache_github_repo_issues(&repo_name_clone, &issues_clone)
}).await.unwrap_or_else(|e| Err(anyhow::anyhow!("join error: {}", e)))?;
let filtered = filter_issues_by_state(issues, issue_state);
Ok::<IssueRepoSnapshot, anyhow::Error>(IssueRepoSnapshot {
repo: repo_name,
issues: filtered,
})
}
Err(error) => {
let (status, body) = github_error(error);
if status == StatusCode::UNAUTHORIZED {
return Err(anyhow::anyhow!("AUTH:{}", body.0.error));
}
let db = state.db.clone();
let repo_name_clone = repo_name.clone();
let error_clone = body.0.error.clone();
tokio::task::spawn_blocking(move || {
db.session_repo.cache_github_repo_failure(&repo_name_clone, &error_clone)
}).await.unwrap_or_else(|e| Err(anyhow::anyhow!("join error: {}", e)))?;
Err(anyhow::anyhow!("REPO:{}:{}", repo_name, body.0.error))
}
}
});
}
let mut repos_completed = 0usize;
while let Some(result) = pending.next().await {
match result {
Ok(snapshot) => {
repos_completed += 1;
if let Ok(event) = event_json("repo_issues", &snapshot) {
yield Ok(event);
}
}
Err(error) => {
let message = error.to_string();
if let Some(auth_error) = message.strip_prefix("AUTH:") {
if let Ok(event) = event_json("fatal", &ErrorResponse { error: auth_error.to_string() }) {
yield Ok(event);
}
break;
}
if let Some((repo, repo_error)) = message
.strip_prefix("REPO:")
.and_then(|value| value.split_once(':'))
{
repos_completed += 1;
if let Ok(event) = event_json("repo_failure", &IssueRepoFailure {
repo: repo.to_string(),
error: repo_error.to_string(),
}) {
yield Ok(event);
}
} else if let Ok(event) = event_json("fatal", &ErrorResponse { error: message }) {
yield Ok(event);
break;
}
}
}
}
if let Ok(event) = event_json("done", &IssueStreamDone { repos_completed }) {
yield Ok(event);
}
};
Ok(Sse::new(stream).keep_alive(KeepAlive::default()))
}
pub async fn list_repo_issues(
State(state): State<WebState>,
Path((owner, repo)): Path<(String, String)>,
Query(params): Query<ListIssuesQuery>,
) -> Result<Json<Vec<github::GitHubIssue>>, (StatusCode, Json<ErrorResponse>)> {
let github = state.github.read().await;
let client = github.as_ref().ok_or_else(no_github)?;
let issue_state = params.state.as_deref().unwrap_or("open");
let issues = client
.list_issues(&owner, &repo, issue_state)
.await
.map_err(github_error)?;
Ok(Json(issues))
}
async fn github_client(
state: &WebState,
) -> Result<github::GitHubClient, (StatusCode, Json<ErrorResponse>)> {
let github = state.github.read().await;
github.as_ref().cloned().ok_or_else(no_github)
}
fn parse_issue_state(
params: &ListIssuesQuery,
) -> Result<IssueStateFilter, (StatusCode, Json<ErrorResponse>)> {
match params.state.as_deref().unwrap_or("open") {
"open" => Ok(IssueStateFilter::Open),
"closed" => Ok(IssueStateFilter::Closed),
"all" => Ok(IssueStateFilter::All),
other => Err((
StatusCode::BAD_REQUEST,
Json(ErrorResponse {
error: format!("unsupported issue state '{}'", other),
}),
)),
}
}
fn resolve_issue_repos(
repo_root: &std::path::Path,
filter_repo: Option<&str>,
) -> Result<Vec<(String, String)>, (StatusCode, Json<ErrorResponse>)> {
let mut repos = collect_issue_repos(repo_root)
.map_err(|e| {
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse {
error: e.to_string(),
}),
)
})?
.into_iter()
.map(|repo| (repo.github.owner, repo.github.repo))
.collect::<Vec<_>>();
if let Some(filter_repo) = filter_repo {
repos.retain(|(owner, repo)| format!("{owner}/{repo}") == filter_repo);
}
Ok(repos)
}
#[derive(Clone, Copy)]
enum IssueStateFilter {
Open,
Closed,
All,
}
fn filter_issues_by_state(
issues: Vec<github::GitHubIssue>,
state: IssueStateFilter,
) -> Vec<github::GitHubIssue> {
issues
.into_iter()
.filter(|issue| match state {
IssueStateFilter::Open => issue.state == "open",
IssueStateFilter::Closed => issue.state == "closed",
IssueStateFilter::All => true,
})
.collect()
}
async fn fetch_issue_aggregate(
state: &WebState,
client: &github::GitHubClient,
repos: &[(String, String)],
issue_state: IssueStateFilter,
) -> Result<IssueAggregateResponse, (StatusCode, Json<ErrorResponse>)> {
let mut all_issues = Vec::new();
let mut failures = Vec::new();
for (owner, repo) in repos {
let repo_name = format!("{owner}/{repo}");
match client.list_issues(owner, repo, "all").await {
Ok(issues) => {
let db = state.db.clone();
let repo_name_clone = repo_name.clone();
let issues_clone = issues.clone();
tokio::task::spawn_blocking(move || {
db.session_repo.cache_github_repo_issues(&repo_name_clone, &issues_clone)
}).await.unwrap_or_else(|e| Err(anyhow::anyhow!("join error: {}", e)))
.map_err(internal_issue_error)?;
for issue in filter_issues_by_state(issues, issue_state) {
let mut val = serde_json::to_value(&issue).expect("GitHub issue must serialize");
val.as_object_mut().expect("GitHub issue JSON object").insert(
"repo".to_string(),
serde_json::json!(repo_name),
);
all_issues.push(val);
}
}
Err(error) => {
let (status, body) = github_error(error);
if status == StatusCode::UNAUTHORIZED {
return Err((status, body));
}
let db = state.db.clone();
let repo_name_clone = repo_name.clone();
let error_clone = body.0.error.clone();
tokio::task::spawn_blocking(move || {
db.session_repo.cache_github_repo_failure(&repo_name_clone, &error_clone)
}).await.unwrap_or_else(|e| Err(anyhow::anyhow!("join error: {}", e)))
.map_err(internal_issue_error)?;
failures.push(IssueRepoFailure {
repo: repo_name,
error: body.0.error,
});
}
}
}
Ok(IssueAggregateResponse {
issues: all_issues,
failures,
})
}
async fn load_root_issues(repos: &crate::persist::global::WsbRepos, db_service: &crate::persist::global::DatabaseService) -> anyhow::Result<Vec<RootIssue>> {
let projects = repos.project_repo.list(None, None).await?;
let Some(project) = projects.into_iter().next() else {
return Ok(Vec::new());
};
let snapshot = crate::persist::local::database::create_context_snapshot(db_service, &project.id).await?;
Ok(snapshot
.active_issues
.into_iter()
.enumerate()
.map(|(index, title)| RootIssue {
id: format!("workspace-active-issue-{}", index + 1),
title,
})
.collect())
}
fn event_json<T: Serialize>(event_name: &str, payload: &T) -> Result<Event, serde_json::Error> {
let json = serde_json::to_string(payload)?;
Ok(Event::default().event(event_name).data(json))
}
fn internal_issue_error(error: anyhow::Error) -> (StatusCode, Json<ErrorResponse>) {
(
StatusCode::INTERNAL_SERVER_ERROR,
Json(ErrorResponse {
error: error.to_string(),
}),
)
}
pub async fn create_issue(
State(state): State<WebState>,
Path((owner, repo)): Path<(String, String)>,
Json(body): Json<CreateIssueRequest>,
) -> Result<Json<github::GitHubIssue>, (StatusCode, Json<ErrorResponse>)> {
let github = state.github.read().await;
let client = github.as_ref().ok_or_else(no_github)?;
let issue = client
.create_issue(&owner, &repo, &body)
.await
.map_err(github_error)?;
Ok(Json(issue))
}
pub async fn update_issue(
State(state): State<WebState>,
Path((owner, repo, number)): Path<(String, String, i64)>,
Json(body): Json<UpdateIssueRequest>,
) -> Result<Json<github::GitHubIssue>, (StatusCode, Json<ErrorResponse>)> {
let github = state.github.read().await;
let client = github.as_ref().ok_or_else(no_github)?;
let issue = client
.update_issue(&owner, &repo, number, &body)
.await
.map_err(github_error)?;
Ok(Json(issue))
}
#[derive(Deserialize)]
pub struct CommentRequest {
pub body: String,
}
pub async fn list_comments(
State(state): State<WebState>,
Path((owner, repo, number)): Path<(String, String, i64)>,
) -> Result<Json<serde_json::Value>, (StatusCode, Json<ErrorResponse>)> {
let github = state.github.read().await;
let client = github.as_ref().ok_or_else(no_github)?;
let comments = client
.list_comments(&owner, &repo, number)
.await
.map_err(github_error)?;
Ok(Json(serde_json::json!({ "comments": comments })))
}
pub async fn add_comment(
State(state): State<WebState>,
Path((owner, repo, number)): Path<(String, String, i64)>,
Json(body): Json<CommentRequest>,
) -> Result<Json<github::GitHubComment>, (StatusCode, Json<ErrorResponse>)> {
let github = state.github.read().await;
let client = github.as_ref().ok_or_else(no_github)?;
let comment = client
.add_comment(&owner, &repo, number, &body.body)
.await
.map_err(github_error)?;
Ok(Json(comment))
}
#[cfg(test)]
mod tests {
use super::{filter_issues_by_state, github_error, IssueStateFilter};
use axum::http::StatusCode;
use crate::web::github::{GitHubIssue, GitHubLabel, GitHubUser};
fn sample_issue(number: i64, state: &str) -> GitHubIssue {
GitHubIssue {
number,
title: format!("Issue {}", number),
body: None,
state: state.to_string(),
user: GitHubUser {
login: "bahram".to_string(),
avatar_url: None,
},
labels: vec![GitHubLabel {
name: "bug".to_string(),
color: "ff0000".to_string(),
description: Some("Bug".to_string()),
}],
comments: 0,
created_at: "2026-03-28T10:00:00Z".to_string(),
updated_at: "2026-03-28T10:00:00Z".to_string(),
closed_at: None,
html_url: format!("https://github.com/machinefabric/wsb/issues/{}", number),
}
}
#[test]
fn list_issues_response_shape_wraps_issues() {
let payload = serde_json::json!({
"issues": [{
"number": 1,
"title": "Broken contract",
"repo": "example/repo"
}],
"failures": [{
"repo": "example/bad-repo",
"error": "GitHub API error listing issues"
}]
});
assert!(payload.get("issues").unwrap().is_array());
assert!(payload.get("failures").unwrap().is_array());
}
#[test]
fn list_comments_response_shape_wraps_comments() {
let payload = serde_json::json!({
"comments": [{
"id": 1,
"body": "Looks good"
}]
});
assert!(payload.get("comments").unwrap().is_array());
}
#[test]
fn github_auth_errors_map_to_unauthorized() {
let (status, body) = github_error(anyhow::anyhow!("failed with GitHub API status 401"));
assert_eq!(status, StatusCode::UNAUTHORIZED);
assert_eq!(
body.0.error,
"GitHub token is invalid or lacks required access"
);
}
#[test]
fn filter_issues_by_state_respects_requested_state() {
let issues = vec![
sample_issue(1, "open"),
sample_issue(2, "closed"),
sample_issue(3, "open"),
];
let open = filter_issues_by_state(issues.clone(), IssueStateFilter::Open);
let closed = filter_issues_by_state(issues.clone(), IssueStateFilter::Closed);
let all = filter_issues_by_state(issues, IssueStateFilter::All);
assert_eq!(open.iter().map(|issue| issue.number).collect::<Vec<_>>(), vec![1, 3]);
assert_eq!(closed.iter().map(|issue| issue.number).collect::<Vec<_>>(), vec![2]);
assert_eq!(all.len(), 3);
}
}