use std::sync::Arc;
use forgetop_core::domain::{
CheckRun, CommentThread, Commit, FileChange, Notification, PipelineApproval, PipelineRun,
ProviderType, PullRequest, PullRequestStatus, TimelineEvent, TimelineEventKind, WorkItem,
};
use forgetop_core::filter::run_triggered_by;
use forgetop_core::launchpad::{self, EntryItem, PipeInput, PrInput, WiInput};
use forgetop_core::provider::{
ItemRef, PipelineSource, PrDecoration, PrWriteSupport, PullRequestFilter, PullRequestQuery,
PullRequestSource, WorkItemQuery, WorkItemSource,
};
use forgetop_core::service::{ConnectionHealthService, SectionService};
use serde::Serialize;
fn log_fetch_failure(operation: &'static str) {
forgetop_core::diag::log(operation, "provider fetch failed");
}
#[derive(Serialize)]
pub struct PrRow {
pub connection_id: String,
pub connection: String,
pub provider: ProviderType,
pub pull_request: PullRequest,
pub needs_decoration: bool,
}
#[derive(Serialize)]
pub struct WiRow {
pub connection_id: String,
pub connection: String,
pub provider: ProviderType,
pub work_item: WorkItem,
}
#[derive(Serialize)]
pub struct PipeRow {
pub connection_id: String,
pub connection: String,
pub provider: ProviderType,
pub run: PipelineRun,
pub definition_name: Option<String>,
pub approvals: Vec<PipelineApproval>,
}
#[derive(Serialize)]
pub struct NotifRow {
pub connection_id: String,
pub connection: String,
pub provider: ProviderType,
pub notification: Notification,
}
#[derive(Serialize)]
pub struct HealthRow {
pub connection_id: String,
pub display_name: String,
pub provider: ProviderType,
pub healthy: bool,
}
#[derive(Clone, Copy)]
pub enum PrView {
All,
Yours,
MergedByYou,
ReviewRequested,
}
impl PrView {
pub fn parse(s: Option<&str>) -> PrView {
match s {
Some("yours") => PrView::Yours,
Some("merged") => PrView::MergedByYou,
Some("review_requested") => PrView::ReviewRequested,
_ => PrView::All,
}
}
fn query(self) -> PullRequestQuery {
let (filter, include_completed) = match self {
PrView::MergedByYou => (PullRequestFilter::Mine, true),
PrView::Yours => (PullRequestFilter::Mine, false),
PrView::ReviewRequested => (PullRequestFilter::ReviewRequested, false),
PrView::All => (PullRequestFilter::All, false),
};
PullRequestQuery { filter, include_completed, limit: Some(50), decorate: false }
}
fn keep(self, pr: &PullRequest) -> bool {
match self {
PrView::MergedByYou => matches!(pr.status, PullRequestStatus::Merged),
_ => true,
}
}
}
pub async fn pull_requests(sections: &SectionService, view: PrView) -> Vec<PrRow> {
let mut out = Vec::new();
if let Ok(feeds) = sections.pull_request_feeds().await.inspect_err(|_| log_fetch_failure("dashboard.pull_requests.feeds")) {
let query = view.query();
for feed in feeds {
let needs_decoration = feed.source.list_omits_decoration();
if let Ok(list) = feed.source.list(&query).await.inspect_err(|_| log_fetch_failure("dashboard.pull_requests.list")) {
out.extend(list.into_iter().filter(|pr| view.keep(pr)).map(|pr| PrRow {
connection_id: feed.connection.connection_id().to_string(),
connection: feed.connection.display_name().to_string(),
provider: feed.connection.provider_type(),
pull_request: pr,
needs_decoration,
}));
}
}
}
out
}
pub async fn work_items(sections: &SectionService) -> Vec<WiRow> {
let mut out = Vec::new();
if let Ok(feeds) = sections.work_item_feeds().await.inspect_err(|_| log_fetch_failure("dashboard.work_items.feeds")) {
let query = WorkItemQuery { mine_only: true, include_completed: false, limit: Some(50) };
for feed in feeds {
if let Ok(list) = feed.source.list(&query).await.inspect_err(|_| log_fetch_failure("dashboard.work_items.list")) {
out.extend(list.into_iter().map(|wi| WiRow {
connection_id: feed.connection.connection_id().to_string(),
connection: feed.connection.display_name().to_string(),
provider: feed.connection.provider_type(),
work_item: wi,
}));
}
}
}
out
}
pub async fn pipelines(sections: &SectionService) -> Vec<PipeRow> {
let mut out = Vec::new();
if let Ok(feeds) = sections.pipeline_feeds().await.inspect_err(|_| log_fetch_failure("dashboard.pipelines.feeds")) {
for feed in feeds {
let defs = feed
.source
.discover()
.await
.inspect_err(|_| log_fetch_failure("dashboard.pipelines.discover"))
.unwrap_or_default();
let def_names: std::collections::HashMap<String, String> =
defs.iter().map(|d| (d.id.clone(), d.name.clone())).collect();
let supports = feed.source.supports_approvals();
for result in feed.list_runs(&defs).await {
if let Ok(runs) = result.inspect_err(|_| log_fetch_failure("dashboard.pipelines.list")) {
for mut run in runs {
let approvals = if supports && run.status.is_active() {
feed.source
.pending_approvals(&run.item_ref())
.await
.inspect_err(|_| log_fetch_failure("dashboard.pipelines.pending_approvals"))
.unwrap_or_default()
} else {
Vec::new()
};
run.status = run.shown_status(approvals.iter().any(|a| a.blocks_run));
out.push(PipeRow {
connection_id: feed.connection.connection_id().to_string(),
connection: feed.connection.display_name().to_string(),
provider: feed.connection.provider_type(),
definition_name: def_names.get(&run.definition_id).cloned(),
approvals,
run,
});
}
}
}
}
}
out
}
pub async fn notifications(sections: &SectionService) -> Vec<NotifRow> {
let mut out = Vec::new();
if let Ok(feeds) = sections.notification_feeds().await.inspect_err(|_| log_fetch_failure("dashboard.notifications.feeds")) {
for feed in feeds {
if let Ok(list) = feed.source.list().await.inspect_err(|_| log_fetch_failure("dashboard.notifications.list")) {
out.extend(list.into_iter().map(|n| NotifRow {
connection_id: feed.connection.connection_id().to_string(),
connection: feed.connection.display_name().to_string(),
provider: feed.connection.provider_type(),
notification: n,
}));
}
}
}
out.sort_by_key(|r| std::cmp::Reverse(r.notification.updated_at)); out
}
#[derive(Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum LaunchpadItem {
Pr { pull_request: PullRequest },
Wi { work_item: WorkItem },
Pipe { run: PipelineRun, definition_name: Option<String> },
}
#[derive(Serialize)]
pub struct LaunchpadRow {
pub bucket: &'static str,
pub bucket_title: &'static str,
pub column: usize,
pub muted: bool,
pub connection_id: String,
pub connection: String,
pub provider: ProviderType,
#[serde(flatten)]
pub item: LaunchpadItem,
}
fn launchpad_query(filter: PullRequestFilter, include_completed: bool) -> PullRequestQuery {
PullRequestQuery { filter, include_completed, limit: Some(50), decorate: true }
}
async fn pr_inputs(sections: &SectionService, filter: PullRequestFilter, include_completed: bool) -> Vec<PrInput> {
let mut out = Vec::new();
if let Ok(feeds) = sections.pull_request_feeds().await.inspect_err(|_| log_fetch_failure("dashboard.launchpad.pull_request_feeds")) {
let query = launchpad_query(filter, include_completed);
for feed in feeds {
if let Ok(list) = feed.source.list(&query).await.inspect_err(|_| log_fetch_failure("dashboard.launchpad.pull_requests")) {
out.extend(list.into_iter().map(|pr| PrInput {
connection_id: feed.connection.connection_id().to_string(),
connection: feed.connection.display_name().to_string(),
provider: feed.connection.provider_type(),
pr,
}));
}
}
}
out
}
async fn wi_inputs(sections: &SectionService) -> Vec<WiInput> {
let mut out = Vec::new();
if let Ok(feeds) = sections.work_item_feeds().await.inspect_err(|_| log_fetch_failure("dashboard.launchpad.work_item_feeds")) {
let query = WorkItemQuery { mine_only: true, include_completed: false, limit: Some(50) };
for feed in feeds {
if let Ok(list) = feed.source.list(&query).await.inspect_err(|_| log_fetch_failure("dashboard.launchpad.work_items")) {
out.extend(list.into_iter().map(|wi| WiInput {
connection_id: feed.connection.connection_id().to_string(),
connection: feed.connection.display_name().to_string(),
provider: feed.connection.provider_type(),
wi,
}));
}
}
}
out
}
async fn pipe_inputs(sections: &SectionService) -> Vec<PipeInput> {
let mut out = Vec::new();
if let Ok(feeds) = sections.pipeline_feeds().await.inspect_err(|_| log_fetch_failure("dashboard.launchpad.pipeline_feeds")) {
for feed in feeds {
let defs = feed
.source
.discover()
.await
.inspect_err(|_| log_fetch_failure("dashboard.launchpad.pipeline_discovery"))
.unwrap_or_default();
let def_names: std::collections::HashMap<String, String> =
defs.iter().map(|d| (d.id.clone(), d.name.clone())).collect();
let supports = feed.source.supports_approvals();
let me = feed
.source
.current_user()
.await
.inspect_err(|_| log_fetch_failure("dashboard.launchpad.pipeline_current_user"))
.unwrap_or_default();
for result in feed.list_runs(&defs).await {
if let Ok(runs) = result.inspect_err(|_| log_fetch_failure("dashboard.launchpad.pipelines")) {
for mut run in runs {
let approvals = if supports && run.status.is_active() {
feed.source
.pending_approvals(&run.item_ref())
.await
.inspect_err(|_| log_fetch_failure("dashboard.launchpad.pipeline_approvals"))
.unwrap_or_default()
} else {
Vec::new()
};
run.status = run.shown_status(approvals.iter().any(|a| a.blocks_run));
let awaiting_approval = approvals.iter().any(|x| x.can_respond);
out.push(PipeInput {
connection_id: feed.connection.connection_id().to_string(),
connection: feed.connection.display_name().to_string(),
provider: feed.connection.provider_type(),
definition_name: def_names.get(&run.definition_id).cloned(),
awaiting_approval,
triggered_by_me: run_triggered_by(&run, me.as_deref()),
run,
});
}
}
}
}
}
out
}
#[derive(Serialize)]
pub struct LaunchpadMore {
pub needs_review: bool,
pub your_work: bool,
pub your_open_prs: bool,
pub recently_merged: bool,
pub recent_pipelines: bool,
}
#[derive(Serialize)]
pub struct LaunchpadResponse {
pub rows: Vec<LaunchpadRow>,
pub more: LaunchpadMore,
}
pub async fn launchpad(sections: &SectionService) -> LaunchpadResponse {
let review = pr_inputs(sections, PullRequestFilter::ReviewRequested, false).await;
let mine = pr_inputs(sections, PullRequestFilter::Mine, true).await;
let wis = wi_inputs(sections).await;
let pipes = pipe_inputs(sections).await;
let built = launchpad::build(&review, &mine, &wis, &pipes);
let rows = built
.entries
.into_iter()
.map(|e| LaunchpadRow {
bucket: e.bucket.key(),
bucket_title: e.bucket.title(),
column: e.bucket.column(),
muted: e.bucket.muted(),
connection_id: e.connection_id,
connection: e.connection,
provider: e.provider,
item: match e.item {
EntryItem::Pr(pr) => LaunchpadItem::Pr { pull_request: pr },
EntryItem::Wi(wi) => LaunchpadItem::Wi { work_item: wi },
EntryItem::Pipe { run, definition_name } => LaunchpadItem::Pipe { run, definition_name },
},
})
.collect();
LaunchpadResponse {
rows,
more: LaunchpadMore {
needs_review: built.overflow.needs_review,
your_work: built.overflow.your_work,
your_open_prs: built.overflow.your_open_prs,
recently_merged: built.overflow.recently_merged,
recent_pipelines: built.overflow.recent_pipelines,
},
}
}
#[derive(Serialize)]
pub struct PrDetail {
pub pull_request: PullRequest,
pub threads: Vec<CommentThread>,
pub timeline: Vec<TimelineEvent>,
pub changes: Vec<FileChange>,
pub checks: Vec<CheckRun>,
pub commits: Vec<Commit>,
#[serde(skip_serializing_if = "Option::is_none")]
pub me: Option<String>,
pub writes: PrWriteSupport,
}
pub async fn pr_source(sections: &SectionService, conn: &str) -> Option<Arc<dyn PullRequestSource>> {
sections
.pull_request_feeds()
.await
.inspect_err(|_| log_fetch_failure("dashboard.pull_requests.feeds"))
.ok()?
.into_iter()
.find(|f| f.connection.connection_id() == conn)
.map(|f| f.source)
}
fn with_comment_events(mut timeline: Vec<TimelineEvent>, threads: &[CommentThread]) -> Vec<TimelineEvent> {
for t in threads.iter().filter(|t| t.file_path.is_none()) {
for c in &t.comments {
timeline.push(TimelineEvent {
actor: Some(c.author.clone()),
kind: TimelineEventKind::Commented,
summary: "commented".into(),
at: c.created_at,
});
}
}
timeline.sort_by_key(|a| a.at);
timeline
}
pub async fn pr_detail(sections: &SectionService, conn: &str, item: &ItemRef) -> Option<PrDetail> {
let source = pr_source(sections, conn).await?;
let pull_request = source.get(item).await.inspect_err(|_| log_fetch_failure("dashboard.pr_detail.get")).ok()?;
let threads = source
.threads(item)
.await
.inspect_err(|_| log_fetch_failure("dashboard.pr_detail.threads"))
.unwrap_or_default();
let timeline = with_comment_events(
source
.timeline(item)
.await
.inspect_err(|_| log_fetch_failure("dashboard.pr_detail.timeline"))
.unwrap_or_default(),
&threads,
);
Some(PrDetail {
pull_request,
threads,
timeline,
changes: source
.changes(item)
.await
.inspect_err(|_| log_fetch_failure("dashboard.pr_detail.changes"))
.unwrap_or_default(),
checks: source
.checks(item)
.await
.inspect_err(|_| log_fetch_failure("dashboard.pr_detail.checks"))
.unwrap_or_default(),
commits: source
.commits(item)
.await
.inspect_err(|_| log_fetch_failure("dashboard.pr_detail.commits"))
.unwrap_or_default(),
me: source
.current_user()
.await
.inspect_err(|_| log_fetch_failure("dashboard.pr_detail.current_user"))
.ok()
.flatten(),
writes: source.pr_writes(),
})
}
pub async fn pr_decoration(sections: &SectionService, conn: &str, item: &ItemRef) -> Option<PrDecoration> {
let source = pr_source(sections, conn).await?;
source.decorate(item).await.inspect_err(|_| log_fetch_failure("dashboard.pr_decoration")).ok()
}
pub async fn pr_commit_changes(sections: &SectionService, conn: &str, item: &ItemRef, sha: &str) -> Option<Vec<FileChange>> {
let source = pr_source(sections, conn).await?;
Some(
source
.commit_changes(item, sha)
.await
.inspect_err(|_| log_fetch_failure("dashboard.pr_commit_changes"))
.unwrap_or_default(),
)
}
#[derive(Serialize)]
pub struct WiDetail {
pub work_item: WorkItem,
pub threads: Vec<CommentThread>,
pub timeline: Vec<TimelineEvent>,
}
pub async fn wi_source(sections: &SectionService, conn: &str) -> Option<Arc<dyn WorkItemSource>> {
sections
.work_item_feeds()
.await
.inspect_err(|_| log_fetch_failure("dashboard.work_items.feeds"))
.ok()?
.into_iter()
.find(|f| f.connection.connection_id() == conn)
.map(|f| f.source)
}
pub async fn wi_detail(sections: &SectionService, conn: &str, item: &ItemRef) -> Option<WiDetail> {
let source = wi_source(sections, conn).await?;
let work_item = source.get(item).await.inspect_err(|_| log_fetch_failure("dashboard.wi_detail.get")).ok()?;
let threads = source
.threads(item)
.await
.inspect_err(|_| log_fetch_failure("dashboard.wi_detail.threads"))
.unwrap_or_default();
let timeline = with_comment_events(
source
.timeline(item)
.await
.inspect_err(|_| log_fetch_failure("dashboard.wi_detail.timeline"))
.unwrap_or_default(),
&threads,
);
Some(WiDetail { work_item, threads, timeline })
}
#[derive(Serialize)]
pub struct PipelineDetail {
pub run: PipelineRun,
pub approvals: Vec<PipelineApproval>,
}
pub async fn pipe_source(sections: &SectionService, conn: &str) -> Option<Arc<dyn PipelineSource>> {
sections
.pipeline_feeds()
.await
.inspect_err(|_| log_fetch_failure("dashboard.pipelines.feeds"))
.ok()?
.into_iter()
.find(|f| f.connection.connection_id() == conn)
.map(|f| f.source)
}
pub async fn pipeline_detail(sections: &SectionService, conn: &str, run: &ItemRef) -> Option<PipelineDetail> {
let source = pipe_source(sections, conn).await?;
let mut run = source.get_run(run).await.inspect_err(|_| log_fetch_failure("dashboard.pipeline_detail.get")).ok()?;
let approvals = if source.supports_approvals() && run.status.is_active() {
source
.pending_approvals(&run.item_ref())
.await
.inspect_err(|_| log_fetch_failure("dashboard.pipeline_detail.pending_approvals"))
.unwrap_or_default()
} else {
Vec::new()
};
run.status = run.shown_status(approvals.iter().any(|a| a.blocks_run));
Some(PipelineDetail { run, approvals })
}
pub async fn pipeline_logs(sections: &SectionService, conn: &str, run: &ItemRef, job_id: Option<&str>) -> Option<String> {
let source = pipe_source(sections, conn).await?;
source.logs(run, job_id).await.inspect_err(|_| log_fetch_failure("dashboard.pipeline_logs.get")).ok()
}
pub async fn health(svc: &ConnectionHealthService) -> Vec<HealthRow> {
svc.check_all()
.await
.into_iter()
.map(|h| HealthRow {
connection_id: h.connection.id.clone(),
display_name: h.connection.display_name.clone(),
provider: h.connection.provider_type,
healthy: h.healthy,
})
.collect()
}
#[cfg(test)]
mod tests {
use super::*;
use forgetop_core::domain::{CheckStatus, MergeableState, Reviewer, ReviewVote, User};
fn pr(reviewers: Vec<Reviewer>) -> PullRequest {
PullRequest {
repository: Some("acme/pay".into()),
id: "7".into(),
number: Some(7),
title: "Add retries".into(),
description: None,
author: User { id: "me".into(), display_name: "me".into(), handle: Some("me".into()), avatar_url: None },
status: PullRequestStatus::Open,
is_draft: false,
source_ref: None,
target_ref: None,
reviewers,
labels: vec![],
checks: CheckStatus::Passed,
check_summary: None,
mergeable: MergeableState::Mergeable,
changed_files: 0,
additions: 0,
deletions: 0,
created_at: None,
updated_at: None,
url: None,
}
}
fn approver() -> Reviewer {
Reviewer {
user: User { id: "priya".into(), display_name: "Priya".into(), handle: None, avatar_url: None },
vote: ReviewVote::Approved,
is_required: false,
}
}
#[test]
fn the_launchpad_decorates_every_fetch_it_ranks_or_renders_from() {
for filter in [PullRequestFilter::Mine, PullRequestFilter::ReviewRequested, PullRequestFilter::All] {
assert!(launchpad_query(filter, false).decorate, "{filter:?} feeds rows the Command Center renders checks on");
}
let ready = pr(vec![approver()]);
let mut conflicting = pr(vec![approver()]);
conflicting.mergeable = MergeableState::Conflicting;
assert_ne!(
launchpad::classify_pr(&ready, launchpad::PrRole::Author).map(|b| b.key()),
launchpad::classify_pr(&conflicting, launchpad::PrRole::Author).map(|b| b.key()),
);
let mut decorated = pr(vec![]);
decorated.mergeable = MergeableState::Conflicting;
decorated.checks = CheckStatus::Failed;
assert_eq!(
launchpad::classify_pr(&pr(vec![]), launchpad::PrRole::Reviewer).map(|b| b.key()),
launchpad::classify_pr(&decorated, launchpad::PrRole::Reviewer).map(|b| b.key()),
"ranking alone would say the review fetch needs no decoration"
);
assert_ne!(decorated.checks, CheckStatus::None, "checks is what an undecorated row would lose");
}
#[tokio::test]
async fn only_providers_whose_list_omits_decoration_ask_for_a_per_row_fetch() {
use forgetop_providers::{bitbucket, github, gitlab};
use forgetop_core::provider::{Connection, ProviderFactory};
let conn = |provider, organization, repository, username| Connection {
id: "c".into(),
provider_type: provider,
display_name: "c".into(),
base_url: None,
organization,
project: None,
repository,
username,
credential_ref: None,
repo_scope: None,
};
let gh = github::GitHubFactory.create(&conn(ProviderType::GitHub, None, Some("acme/pay".into()), None), None).unwrap();
assert!(gh.pull_requests().unwrap().list_omits_decoration(), "GitHub's /pulls omits them");
let gl = gitlab::GitLabFactory.create(&conn(ProviderType::GitLab, None, Some("g/p".into()), None), None).unwrap();
assert!(!gl.pull_requests().unwrap().list_omits_decoration(), "GitLab fills them from the list");
let bb = bitbucket::BitbucketFactory
.create(&conn(ProviderType::Bitbucket, Some("ws".into()), Some("repo".into()), Some("u".into())), None)
.unwrap();
assert!(!bb.pull_requests().unwrap().list_omits_decoration(), "Bitbucket fills them from the list");
}
}