Skip to main content

forgetop_server/
lib.rs

1//! Embedded HTTP server for the forgetop web dashboard.
2//!
3//! It's a thin frontend over the **same** `SectionService` / health services the TUI uses —
4//! no logic fork. Wave 1 is a read-only JSON API + a placeholder page, bound to `127.0.0.1`
5//! and gated by a per-session token so no other local process or web page can reach it.
6
7use std::net::Ipv4Addr;
8use std::sync::Arc;
9
10use axum::extract::{Query, Request, State};
11use axum::http::{header, StatusCode, Uri};
12use axum::middleware::{self, Next};
13use axum::response::{IntoResponse, Json, Response};
14use axum::routing::{get, post};
15use axum::Router;
16use forgetop_core::provider::{ItemRef, PrDecoration};
17use forgetop_core::secret::SecretStore;
18use forgetop_core::service::{ConfigService, ConnectionHealthService, SectionService};
19use rust_embed::RustEmbed;
20use serde::Deserialize;
21
22use crate::actions::ActionError;
23
24mod actions;
25mod connections;
26mod dto;
27
28/// The built dashboard SPA, baked into the binary at compile time (see `build.rs`).
29#[derive(RustEmbed)]
30#[folder = "web/dist"]
31struct Assets;
32
33/// Default dashboard port. `0` lets the OS pick a free one.
34pub const DEFAULT_PORT: u16 = 8177;
35
36/// The services the dashboard reads from and writes through — the same ones the TUI is given,
37/// so connections added here show up in the TUI (and vice versa).
38#[derive(Clone)]
39pub struct Deps {
40    pub sections: Arc<SectionService>,
41    pub health: Arc<ConnectionHealthService>,
42    pub config: Arc<ConfigService>,
43    pub secrets: Arc<dyn SecretStore>,
44}
45
46/// A bound, running server: where it lives and the token needed to reach it.
47pub struct Server {
48    pub port: u16,
49    pub token: String,
50    /// `http://127.0.0.1:<port>/?t=<token>` — the URL to open in a browser.
51    pub url: String,
52}
53
54/// Decorated PR fields, cached per `(connection, repository, id, updated_at)`.
55///
56/// `updated_at` is part of the key on purpose: it is the provider's own statement that the PR
57/// changed, so a stale entry can never outlive the change that invalidates it.
58#[derive(Default)]
59struct DecorationCache {
60    entries: std::sync::Mutex<std::collections::HashMap<String, PrDecoration>>,
61}
62
63impl DecorationCache {
64    /// Bounded so a long-running dashboard can't grow it without limit. On overflow the whole
65    /// map is dropped rather than evicted one by one — decoration is cheap to refetch.
66    const MAX: usize = 500;
67
68    fn key(conn: &str, item: &ItemRef, updated_at: Option<&str>) -> String {
69        format!("{conn}\u{1}{}\u{1}{}\u{1}{}", item.repo.as_deref().unwrap_or(""), item.id, updated_at.unwrap_or(""))
70    }
71
72    fn get(&self, key: &str) -> Option<PrDecoration> {
73        self.entries.lock().unwrap().get(key).cloned()
74    }
75
76    fn put(&self, key: String, value: PrDecoration) {
77        let mut entries = self.entries.lock().unwrap();
78        if entries.len() >= Self::MAX {
79            entries.clear();
80        }
81        entries.insert(key, value);
82    }
83}
84
85#[derive(Clone)]
86struct AppState {
87    deps: Deps,
88    token: Arc<str>,
89    decorations: Arc<DecorationCache>,
90}
91
92async fn bind(deps: Deps, port: u16) -> std::io::Result<(tokio::net::TcpListener, Server, AppState)> {
93    let token = uuid::Uuid::new_v4().to_string();
94    let listener = tokio::net::TcpListener::bind((Ipv4Addr::LOCALHOST, port)).await?;
95    let bound = listener.local_addr()?.port();
96    let url = format!("http://127.0.0.1:{bound}/?t={token}");
97    let state = AppState { deps, token: Arc::from(token.as_str()), decorations: Arc::new(DecorationCache::default()) };
98    Ok((listener, Server { port: bound, token, url }, state))
99}
100
101/// Bind + serve in the background, returning the URL (with token) for the caller to open.
102/// **Best-effort:** an `Err` just means "no dashboard" — the TUI should carry on regardless.
103pub async fn spawn(deps: Deps, port: u16) -> std::io::Result<Server> {
104    let (listener, server, state) = bind(deps, port).await?;
105    tokio::spawn(async move {
106        let _ = axum::serve(listener, router(state)).await;
107    });
108    Ok(server)
109}
110
111/// Bind + serve until the process exits (headless `--dashboard`). Calls `on_ready` with the
112/// URL once bound so the caller can open the browser.
113pub async fn serve_blocking(deps: Deps, port: u16, on_ready: impl FnOnce(&str)) -> std::io::Result<()> {
114    let (listener, server, state) = bind(deps, port).await?;
115    on_ready(&server.url);
116    axum::serve(listener, router(state)).await
117}
118
119fn router(state: AppState) -> Router {
120    // The API carries your data and can act on your behalf, so it's token-gated. The static
121    // SPA (HTML/JS/CSS) is just code — not secret — so it's served openly; the browser gets
122    // the token from the `/?t=` URL and replays it on every API call.
123    let api = Router::new()
124        .route("/api/health", get(health))
125        .route("/api/pull-requests", get(pull_requests))
126        .route("/api/work-items", get(work_items))
127        .route("/api/pipelines", get(pipelines))
128        .route("/api/notifications", get(notifications))
129        .route("/api/launchpad", get(launchpad))
130        .route("/api/pr/detail", get(pr_detail))
131        .route("/api/pr/decoration", get(pr_decoration))
132        .route("/api/pr/commit-changes", get(pr_commit_changes))
133        .route("/api/pr/vote", post(pr_vote))
134        .route("/api/pr/merge", post(pr_merge))
135        .route("/api/pr/revert", post(pr_revert))
136        .route("/api/pr/comment", post(pr_comment))
137        .route("/api/pr/reply", post(pr_reply))
138        .route("/api/pr/review", post(pr_review))
139        .route("/api/pr/resolve-thread", post(pr_resolve_thread))
140        .route("/api/pr/draft", post(pr_draft))
141        .route("/api/pr/closed", post(pr_closed))
142        .route("/api/pr/reviewers", get(pr_reviewers))
143        .route("/api/pr/request-reviewer", post(pr_request_reviewer))
144        .route("/api/wi/detail", get(wi_detail))
145        .route("/api/wi/states", get(wi_states))
146        .route("/api/wi/state", post(wi_state))
147        .route("/api/wi/comment", post(wi_comment))
148        .route("/api/wi/assignees", get(wi_assignees))
149        .route("/api/wi/assignee", post(wi_assignee))
150        .route("/api/wi/update", post(wi_update))
151        .route("/api/pipeline/detail", get(pipeline_detail))
152        .route("/api/pipeline/logs", get(pipeline_logs))
153        .route("/api/pipeline/approval", post(pipeline_approval))
154        .route("/api/pipeline/trigger", post(pipeline_trigger))
155        .route("/api/pipeline/cancel", post(pipeline_cancel))
156        .route("/api/notification/read", post(notification_read))
157        .route("/api/providers", get(providers))
158        .route("/api/connections", get(list_connections).post(save_connection))
159        .route("/api/connections/delete", post(delete_connection))
160        .route("/api/connections/test", post(test_connection))
161        .route("/api/connections/repositories", get(connection_repositories))
162        .route("/api/connections/scope", post(set_connection_scope))
163        .route("/api/pipelines/definitions", get(pipeline_definitions))
164        .route("/api/pipelines/selection", post(set_pipeline_selection))
165        .layer(middleware::from_fn_with_state(state.clone(), auth))
166        .with_state(state);
167
168    Router::new().merge(api).fallback(static_asset)
169}
170
171/// Serves an embedded SPA asset by path, falling back to `index.html` for unknown routes so
172/// client-side routing works on refresh/deep-link.
173async fn static_asset(uri: Uri) -> Response {
174    let path = uri.path().trim_start_matches('/');
175    let path = if path.is_empty() { "index.html" } else { path };
176    serve(path)
177        .or_else(|| serve("index.html"))
178        .unwrap_or_else(|| (StatusCode::NOT_FOUND, "not found").into_response())
179}
180
181fn serve(path: &str) -> Option<Response> {
182    let file = Assets::get(path)?;
183    let mime = mime_guess::from_path(path).first_or_octet_stream();
184    Some(([(header::CONTENT_TYPE, mime.as_ref().to_string())], file.data).into_response())
185}
186
187/// Session-token gate. The browser opens `/?t=<token>`; other calls may send it as the
188/// `x-forgetop-token` header. Combined with localhost-only binding, this keeps the
189/// (action-capable) API off-limits to other local processes and web pages.
190async fn auth(State(state): State<AppState>, req: Request, next: Next) -> Response {
191    if token_from(&req).as_deref() == Some(&state.token) {
192        next.run(req).await
193    } else {
194        (StatusCode::UNAUTHORIZED, "unauthorized").into_response()
195    }
196}
197
198fn token_from(req: &Request) -> Option<String> {
199    if let Some(header) = req.headers().get("x-forgetop-token").and_then(|v| v.to_str().ok()) {
200        return Some(header.to_string());
201    }
202    req.uri()
203        .query()
204        .and_then(|q| q.split('&').find_map(|pair| pair.strip_prefix("t=").map(str::to_string)))
205}
206
207async fn health(State(s): State<AppState>) -> Json<Vec<dto::HealthRow>> {
208    Json(dto::health(&s.deps.health).await)
209}
210async fn pull_requests(State(s): State<AppState>, Query(q): Query<PrListQuery>) -> Json<Vec<dto::PrRow>> {
211    Json(dto::pull_requests(&s.deps.sections, dto::PrView::parse(q.view.as_deref())).await)
212}
213async fn work_items(State(s): State<AppState>) -> Json<Vec<dto::WiRow>> {
214    Json(dto::work_items(&s.deps.sections).await)
215}
216async fn pipelines(State(s): State<AppState>) -> Json<Vec<dto::PipeRow>> {
217    Json(dto::pipelines(&s.deps.sections).await)
218}
219async fn notifications(State(s): State<AppState>) -> Json<Vec<dto::NotifRow>> {
220    Json(dto::notifications(&s.deps.sections).await)
221}
222async fn launchpad(State(s): State<AppState>) -> Json<dto::LaunchpadResponse> {
223    Json(dto::launchpad(&s.deps.sections).await)
224}
225
226/// Query params identifying an item within a connection (`?conn=…&id=…&repo=…`).
227///
228/// `repo` is a **query parameter**, never a path segment: an `owner/repo` always contains a
229/// slash. It is optional, so every link written before connections spanned an account still
230/// resolves (a single-repository connection needs no address).
231#[derive(Deserialize)]
232struct ItemQuery {
233    conn: String,
234    id: String,
235    #[serde(default)]
236    repo: Option<String>,
237}
238
239impl ItemQuery {
240    fn item(self) -> (String, ItemRef) {
241        (self.conn, ItemRef::maybe(self.repo, self.id))
242    }
243}
244
245/// Query params identifying a single commit's diff within a PR (`?conn=…&id=…&sha=…`).
246#[derive(Deserialize)]
247struct CommitQuery {
248    conn: String,
249    id: String,
250    sha: String,
251    #[serde(default)]
252    repo: Option<String>,
253}
254
255/// Query params for the PR list: which view to show (`?view=all|merged|review_requested`).
256#[derive(Deserialize)]
257struct PrListQuery {
258    #[serde(default)]
259    view: Option<String>,
260}
261
262/// Query params identifying a pipeline run within a connection (`?conn=…&run_id=…`).
263#[derive(Deserialize)]
264struct RunQuery {
265    conn: String,
266    run_id: String,
267    #[serde(default)]
268    repo: Option<String>,
269}
270
271/// Query params for pipeline logs: a run, optionally scoped to a single job (`&job=…`).
272#[derive(Deserialize)]
273struct PipelineLogsQuery {
274    conn: String,
275    run_id: String,
276    #[serde(default)]
277    repo: Option<String>,
278    #[serde(default)]
279    job: Option<String>,
280}
281
282/// Turns an action outcome into a response: `{ok:true}`, 404 (no such connection/capability),
283/// or 502 (the provider call failed).
284fn action_response(operation: &'static str, result: Result<(), ActionError>) -> Response {
285    match result {
286        Ok(()) => Json(serde_json::json!({ "ok": true })).into_response(),
287        Err(ActionError::NotFound) => (StatusCode::NOT_FOUND, "connection or capability not found").into_response(),
288        Err(ActionError::Failed(msg)) => {
289            forgetop_core::diag::log(operation, "provider action failed");
290            (StatusCode::BAD_GATEWAY, msg).into_response()
291        }
292    }
293}
294
295async fn pr_commit_changes(State(s): State<AppState>, Query(q): Query<CommitQuery>) -> Response {
296    match dto::pr_commit_changes(&s.deps.sections, &q.conn, &ItemRef::maybe(q.repo, q.id), &q.sha).await {
297        Some(changes) => Json(changes).into_response(),
298        None => (StatusCode::NOT_FOUND, "pull request not found").into_response(),
299    }
300}
301
302async fn pr_detail(State(s): State<AppState>, Query(q): Query<ItemQuery>) -> Response {
303    let (conn, item) = q.item();
304    match dto::pr_detail(&s.deps.sections, &conn, &item).await {
305        Some(detail) => Json(detail).into_response(),
306        None => (StatusCode::NOT_FOUND, "pull request not found").into_response(),
307    }
308}
309
310/// Query params for one PR's decoration. `updated_at` is the provider's own last-changed stamp
311/// from the list row, and is what makes the cache entry safe to reuse.
312#[derive(Deserialize)]
313struct DecorationQuery {
314    conn: String,
315    id: String,
316    #[serde(default)]
317    repo: Option<String>,
318    #[serde(default)]
319    updated_at: Option<String>,
320}
321
322async fn pr_decoration(State(s): State<AppState>, Query(q): Query<DecorationQuery>) -> Response {
323    let item = ItemRef::maybe(q.repo, q.id);
324    let key = DecorationCache::key(&q.conn, &item, q.updated_at.as_deref());
325    if let Some(hit) = s.decorations.get(&key) {
326        return Json(hit).into_response();
327    }
328    match dto::pr_decoration(&s.deps.sections, &q.conn, &item).await {
329        Some(decoration) => {
330            s.decorations.put(key, decoration.clone());
331            Json(decoration).into_response()
332        }
333        None => (StatusCode::NOT_FOUND, "pull request not found").into_response(),
334    }
335}
336
337async fn pr_vote(State(s): State<AppState>, Json(req): Json<actions::PrVoteReq>) -> Response {
338    action_response("dashboard.pr_vote", actions::pr_vote(&s.deps.sections, req).await)
339}
340async fn pr_merge(State(s): State<AppState>, Json(req): Json<actions::PrMergeReq>) -> Response {
341    action_response("dashboard.pr_merge", actions::pr_merge(&s.deps.sections, req).await)
342}
343async fn pr_revert(State(s): State<AppState>, Json(req): Json<actions::PrRevertReq>) -> Response {
344    action_response("dashboard.pr_revert", actions::pr_revert(&s.deps.sections, req).await)
345}
346async fn pr_comment(State(s): State<AppState>, Json(req): Json<actions::PrCommentReq>) -> Response {
347    action_response("dashboard.pr_comment", actions::pr_comment(&s.deps.sections, req).await)
348}
349async fn pr_reply(State(s): State<AppState>, Json(req): Json<actions::PrReplyReq>) -> Response {
350    action_response("dashboard.pr_reply", actions::pr_reply(&s.deps.sections, req).await)
351}
352async fn pr_review(State(s): State<AppState>, Json(req): Json<actions::PrReviewReq>) -> Response {
353    action_response("dashboard.pr_review", actions::pr_review(&s.deps.sections, req).await)
354}
355async fn pr_resolve_thread(State(s): State<AppState>, Json(req): Json<actions::PrResolveThreadReq>) -> Response {
356    action_response("dashboard.pr_resolve_thread", actions::pr_resolve_thread(&s.deps.sections, req).await)
357}
358async fn pr_draft(State(s): State<AppState>, Json(req): Json<actions::PrDraftReq>) -> Response {
359    action_response("dashboard.pr_draft", actions::pr_set_draft(&s.deps.sections, req).await)
360}
361async fn pr_closed(State(s): State<AppState>, Json(req): Json<actions::PrClosedReq>) -> Response {
362    action_response("dashboard.pr_closed", actions::pr_set_closed(&s.deps.sections, req).await)
363}
364async fn pr_reviewers(State(s): State<AppState>, Query(q): Query<ItemQuery>) -> Response {
365    let (conn, item) = q.item();
366    match actions::pr_reviewers(&s.deps.sections, &conn, &item).await {
367        Some(users) => Json(users).into_response(),
368        None => (StatusCode::NOT_FOUND, "pull request connection not found").into_response(),
369    }
370}
371async fn pr_request_reviewer(State(s): State<AppState>, Json(req): Json<actions::PrRequestReviewerReq>) -> Response {
372    action_response("dashboard.pr_request_reviewer", actions::pr_request_reviewer(&s.deps.sections, req).await)
373}
374
375async fn wi_detail(State(s): State<AppState>, Query(q): Query<ItemQuery>) -> Response {
376    let (conn, item) = q.item();
377    match dto::wi_detail(&s.deps.sections, &conn, &item).await {
378        Some(detail) => Json(detail).into_response(),
379        None => (StatusCode::NOT_FOUND, "work item not found").into_response(),
380    }
381}
382async fn wi_states(State(s): State<AppState>, Query(q): Query<ItemQuery>) -> Response {
383    let (conn, item) = q.item();
384    match actions::wi_states(&s.deps.sections, &conn, &item).await {
385        Some(states) => Json(states).into_response(),
386        None => (StatusCode::NOT_FOUND, "work item connection not found").into_response(),
387    }
388}
389async fn wi_state(State(s): State<AppState>, Json(req): Json<actions::WiStateReq>) -> Response {
390    action_response("dashboard.wi_state", actions::wi_set_state(&s.deps.sections, req).await)
391}
392async fn wi_comment(State(s): State<AppState>, Json(req): Json<actions::WiCommentReq>) -> Response {
393    action_response("dashboard.wi_comment", actions::wi_comment(&s.deps.sections, req).await)
394}
395async fn wi_assignees(State(s): State<AppState>, Query(q): Query<ItemQuery>) -> Response {
396    let (conn, item) = q.item();
397    match actions::wi_assignees(&s.deps.sections, &conn, &item).await {
398        Some(users) => Json(users).into_response(),
399        None => (StatusCode::NOT_FOUND, "work item connection not found").into_response(),
400    }
401}
402async fn wi_assignee(State(s): State<AppState>, Json(req): Json<actions::WiAssigneeReq>) -> Response {
403    action_response("dashboard.wi_assignee", actions::wi_set_assignee(&s.deps.sections, req).await)
404}
405async fn wi_update(State(s): State<AppState>, Json(req): Json<actions::WiUpdateReq>) -> Response {
406    action_response("dashboard.wi_update", actions::wi_update(&s.deps.sections, req).await)
407}
408
409async fn pipeline_detail(State(s): State<AppState>, Query(q): Query<RunQuery>) -> Response {
410    match dto::pipeline_detail(&s.deps.sections, &q.conn, &ItemRef::maybe(q.repo, q.run_id)).await {
411        Some(detail) => Json(detail).into_response(),
412        None => (StatusCode::NOT_FOUND, "pipeline run not found").into_response(),
413    }
414}
415async fn pipeline_logs(State(s): State<AppState>, Query(q): Query<PipelineLogsQuery>) -> Response {
416    match dto::pipeline_logs(&s.deps.sections, &q.conn, &ItemRef::maybe(q.repo, q.run_id), q.job.as_deref()).await {
417        Some(text) => ([(header::CONTENT_TYPE, "text/plain; charset=utf-8")], text).into_response(),
418        None => (StatusCode::NOT_FOUND, "logs not available").into_response(),
419    }
420}
421
422async fn pipeline_approval(State(s): State<AppState>, Json(req): Json<actions::PipelineApprovalReq>) -> Response {
423    action_response("dashboard.pipeline_approval", actions::pipeline_approval(&s.deps.sections, req).await)
424}
425async fn pipeline_trigger(State(s): State<AppState>, Json(req): Json<actions::PipelineTriggerReq>) -> Response {
426    action_response("dashboard.pipeline_trigger", actions::pipeline_trigger(&s.deps.sections, req).await)
427}
428async fn pipeline_cancel(State(s): State<AppState>, Json(req): Json<actions::PipelineCancelReq>) -> Response {
429    action_response("dashboard.pipeline_cancel", actions::pipeline_cancel(&s.deps.sections, req).await)
430}
431
432async fn notification_read(State(s): State<AppState>, Json(req): Json<actions::NotifReadReq>) -> Response {
433    action_response("dashboard.notification_read", actions::notif_read(&s.deps.sections, req).await)
434}
435
436// ---- connection management ----
437
438#[derive(Deserialize)]
439struct IdReq {
440    id: String,
441}
442
443async fn providers() -> Json<Vec<connections::ProviderInfo>> {
444    Json(connections::providers())
445}
446async fn list_connections(State(s): State<AppState>) -> Json<Vec<connections::ConnectionRow>> {
447    Json(connections::list(&s.deps.config, s.deps.secrets.as_ref()))
448}
449async fn save_connection(State(s): State<AppState>, Json(req): Json<connections::SaveConnectionReq>) -> Response {
450    match connections::save(&s.deps.config, &s.deps.sections, req).await {
451        Ok(id) => Json(serde_json::json!({ "ok": true, "id": id })).into_response(),
452        Err(msg) => (StatusCode::BAD_GATEWAY, msg).into_response(),
453    }
454}
455async fn delete_connection(State(s): State<AppState>, Json(req): Json<IdReq>) -> Response {
456    match connections::remove(&s.deps.config, &req.id).await {
457        Ok(()) => Json(serde_json::json!({ "ok": true })).into_response(),
458        Err(msg) => (StatusCode::BAD_GATEWAY, msg).into_response(),
459    }
460}
461async fn test_connection(State(s): State<AppState>, Json(req): Json<IdReq>) -> Response {
462    match connections::test(&s.deps.health, &req.id).await {
463        Some(healthy) => Json(serde_json::json!({ "healthy": healthy })).into_response(),
464        None => (StatusCode::NOT_FOUND, "connection not found").into_response(),
465    }
466}
467
468/// The repositories a connection could fetch from — the scope picker's candidate list. Only
469/// this endpoint calls provider discovery, so discovery being wrong shows an empty picker; it
470/// cannot stop an already-scoped connection fetching.
471async fn connection_repositories(State(s): State<AppState>, Query(q): Query<IdQuery>) -> Response {
472    match s.deps.sections.discover_repositories(&q.id).await {
473        Ok(page) => Json(page).into_response(),
474        Err(e) => (StatusCode::BAD_GATEWAY, e.to_string()).into_response(),
475    }
476}
477
478#[derive(Deserialize)]
479struct IdQuery {
480    id: String,
481}
482
483#[derive(Deserialize)]
484struct ScopeReq {
485    id: String,
486    /// The chosen repositories. An empty list is a real choice — fetch nothing — and is stored
487    /// as such; it is not the same as never having chosen.
488    scope: Vec<String>,
489}
490
491async fn set_connection_scope(State(s): State<AppState>, Json(req): Json<ScopeReq>) -> Response {
492    match s.deps.config.set_repo_scope(&req.id, Some(req.scope)).await {
493        Ok(()) => Json(serde_json::json!({ "ok": true })).into_response(),
494        Err(e) => (StatusCode::BAD_GATEWAY, e.to_string()).into_response(),
495    }
496}
497
498/// A pipeline connection's discovered definitions and which of them are fetched — the picker's
499/// candidates. Pipelines are opt-in: nothing selected (`all` false, `selected` empty) fetches
500/// nothing.
501#[derive(serde::Serialize)]
502struct PipelineDefinitionsResp {
503    definitions: Vec<forgetop_core::domain::PipelineDefinition>,
504    all: bool,
505    selected: Vec<String>,
506}
507
508async fn pipeline_definitions(State(s): State<AppState>, Query(q): Query<IdQuery>) -> Response {
509    let source = match s.deps.sections.pipeline_source_for(&q.id).await {
510        Ok(Some(source)) => source,
511        Ok(None) => return (StatusCode::NOT_FOUND, "that connection doesn't support pipelines").into_response(),
512        Err(e) => return (StatusCode::BAD_GATEWAY, e.to_string()).into_response(),
513    };
514    let definitions = match source.discover().await {
515        Ok(defs) => defs,
516        Err(e) => return (StatusCode::BAD_GATEWAY, e.to_string()).into_response(),
517    };
518    let cfg = s.deps.config.snapshot();
519    let sub = cfg.pipelines.as_ref().and_then(|p| p.subscriptions.iter().find(|s| s.connection_id == q.id));
520    Json(PipelineDefinitionsResp {
521        definitions,
522        all: sub.is_some_and(|s| s.auto_discover_all),
523        selected: sub.map(|s| s.definition_ids.clone()).unwrap_or_default(),
524    })
525    .into_response()
526}
527
528#[derive(Deserialize)]
529struct PipelineSelectionReq {
530    id: String,
531    /// Every pipeline, including ones created later. Wins over `ids`.
532    #[serde(default)]
533    all: bool,
534    /// The chosen definitions. Empty is a real choice — fetch nothing.
535    #[serde(default)]
536    ids: Vec<String>,
537}
538
539async fn set_pipeline_selection(State(s): State<AppState>, Json(req): Json<PipelineSelectionReq>) -> Response {
540    let saved = if req.all {
541        s.deps.config.set_pipeline_auto_discover(&req.id, true).await
542    } else {
543        s.deps.config.set_pipeline_definitions(&req.id, req.ids).await
544    };
545    match saved {
546        Ok(()) => Json(serde_json::json!({ "ok": true })).into_response(),
547        Err(e) => (StatusCode::BAD_GATEWAY, e.to_string()).into_response(),
548    }
549}
550
551#[cfg(test)]
552mod tests {
553    //! The optional pull-request writes (resolve a thread, draft, close/reopen, request a
554    //! reviewer) end to end through the HTTP API, against the in-memory demo provider — the same
555    //! harness `tests/api.rs` uses for the other routes.
556
557    use super::*;
558    use forgetop_core::config::{ConfigStore, ForgetopConfig, InMemoryConfigStore};
559    use forgetop_core::domain::ProviderType;
560    use forgetop_core::provider::{Connection, ProviderRegistry};
561    use forgetop_core::secret::InMemorySecretStore;
562    use forgetop_core::service::ConnectionResolver;
563    use forgetop_providers::demo::demo_factories;
564
565    /// One demo GitHub connection bound to the PR section — the shape `forgetop --demo` wires.
566    async fn demo_deps() -> Deps {
567        let registry = Arc::new(ProviderRegistry::new(demo_factories()));
568        let store: Arc<dyn ConfigStore> = Arc::new(InMemoryConfigStore::new(ForgetopConfig::default()));
569        let secrets: Arc<dyn SecretStore> = Arc::new(InMemorySecretStore::default());
570        let config = Arc::new(ConfigService::new(store, secrets.clone(), registry.clone()));
571        config.load().await.unwrap();
572        let conn = Connection {
573            id: "github".into(),
574            provider_type: ProviderType::GitHub,
575            display_name: "github".into(),
576            base_url: None,
577            organization: None,
578            project: None,
579            repository: None,
580            username: None,
581            credential_ref: None,
582            repo_scope: None,
583        };
584        config.add_or_update_connection(conn, None).await.unwrap();
585        config.bind_pull_requests("github").await.unwrap();
586        let resolver = Arc::new(ConnectionResolver::new(config.clone(), registry, secrets.clone()));
587        let sections = Arc::new(SectionService::new(config.clone(), resolver.clone()));
588        let health = Arc::new(ConnectionHealthService::new(config.clone(), resolver));
589        Deps { sections, health, config, secrets }
590    }
591
592    struct Api {
593        base: String,
594        token: String,
595        client: reqwest::Client,
596    }
597
598    impl Api {
599        async fn start() -> Api {
600            let server = spawn(demo_deps().await, 0).await.expect("server binds a free port");
601            Api { base: format!("http://127.0.0.1:{}", server.port), token: server.token, client: reqwest::Client::new() }
602        }
603        async fn get(&self, path: &str) -> serde_json::Value {
604            let r = self.client.get(format!("{}{path}", self.base)).header("x-forgetop-token", &self.token).send().await.unwrap();
605            assert_eq!(r.status(), 200, "GET {path}");
606            r.json().await.unwrap()
607        }
608        async fn post(&self, path: &str, body: serde_json::Value) -> u16 {
609            let r = self.client.post(format!("{}{path}", self.base)).header("x-forgetop-token", &self.token).json(&body).send().await.unwrap();
610            r.status().as_u16()
611        }
612        /// The open PRs the list shows, as `(id, status)`.
613        async fn open_prs(&self) -> Vec<(String, String)> {
614            let rows = self.get("/api/pull-requests").await;
615            rows.as_array()
616                .unwrap()
617                .iter()
618                .map(|r| (r["pull_request"]["id"].as_str().unwrap().to_string(), r["pull_request"]["status"].as_str().unwrap().to_string()))
619                .collect()
620        }
621        async fn detail(&self, id: &str) -> serde_json::Value {
622            self.get(&format!("/api/pr/detail?conn=github&id={id}")).await
623        }
624    }
625
626    /// The demo PRs are global to the process, so each test picks its own open PR to write to.
627    async fn open_pr(api: &Api, nth: usize) -> String {
628        let open: Vec<_> = api.open_prs().await.into_iter().filter(|(_, s)| s == "Open").collect();
629        open.get(nth).unwrap_or_else(|| panic!("demo has at least {} open PRs", nth + 1)).0.clone()
630    }
631
632    #[tokio::test]
633    async fn pr_detail_advertises_the_demo_writes() {
634        let api = Api::start().await;
635        let id = open_pr(&api, 0).await;
636        let writes = &api.detail(&id).await["writes"];
637        for k in ["resolve_threads", "draft", "close", "reopen", "request_reviewer"] {
638            assert_eq!(writes[k], true, "demo supports {k}: {writes}");
639        }
640    }
641
642    #[tokio::test]
643    async fn resolving_a_thread_flips_it_on_the_next_detail() {
644        let api = Api::start().await;
645        let id = open_pr(&api, 1).await;
646        let thread = |d: &serde_json::Value, t: &str| d["threads"].as_array().unwrap().iter().find(|x| x["id"] == t).cloned().unwrap();
647        assert_eq!(thread(&api.detail(&id).await, "t1")["is_resolved"], false, "t1 starts open");
648
649        let body = |resolved: bool| serde_json::json!({ "conn": "github", "id": id, "thread_id": "t1", "resolved": resolved });
650        assert_eq!(api.post("/api/pr/resolve-thread", body(true)).await, 200);
651        assert_eq!(thread(&api.detail(&id).await, "t1")["is_resolved"], true, "resolve persists");
652        assert_eq!(api.post("/api/pr/resolve-thread", body(false)).await, 200);
653        assert_eq!(thread(&api.detail(&id).await, "t1")["is_resolved"], false, "reopen persists");
654    }
655
656    #[tokio::test]
657    async fn draft_and_close_change_the_pr_status() {
658        let api = Api::start().await;
659        let id = open_pr(&api, 2).await;
660        let status = |d: serde_json::Value| d["pull_request"]["status"].as_str().unwrap().to_string();
661
662        assert_eq!(api.post("/api/pr/draft", serde_json::json!({ "conn": "github", "id": id, "draft": true })).await, 200);
663        assert_eq!(status(api.detail(&id).await), "Draft", "converted to a draft");
664        assert_eq!(api.post("/api/pr/draft", serde_json::json!({ "conn": "github", "id": id, "draft": false })).await, 200);
665        assert_eq!(status(api.detail(&id).await), "Open", "ready for review again");
666
667        assert_eq!(api.post("/api/pr/closed", serde_json::json!({ "conn": "github", "id": id, "closed": true })).await, 200);
668        assert_eq!(status(api.detail(&id).await), "Closed");
669        assert!(!api.open_prs().await.iter().any(|(p, _)| *p == id), "a closed PR leaves the open list");
670        assert_eq!(api.post("/api/pr/closed", serde_json::json!({ "conn": "github", "id": id, "closed": false })).await, 200);
671        assert_eq!(status(api.detail(&id).await), "Open", "reopened");
672        assert!(api.open_prs().await.iter().any(|(p, _)| *p == id), "and back on the open list");
673
674        // A missing connection is a 404, as for every other action.
675        assert_eq!(api.post("/api/pr/draft", serde_json::json!({ "conn": "nope", "id": id, "draft": true })).await, 404);
676    }
677
678    #[tokio::test]
679    async fn requesting_a_reviewer_adds_them_to_the_pr() {
680        let api = Api::start().await;
681        let id = open_pr(&api, 3).await;
682        let users = api.get(&format!("/api/pr/reviewers?conn=github&id={id}")).await;
683        let users = users.as_array().expect("reviewers is a JSON array of users");
684        assert!(!users.is_empty(), "the demo has someone to ask");
685
686        let reviewer_ids = |d: &serde_json::Value| -> Vec<String> {
687            d["pull_request"]["reviewers"].as_array().unwrap().iter().map(|r| r["user"]["id"].as_str().unwrap().to_string()).collect()
688        };
689        let before = reviewer_ids(&api.detail(&id).await);
690        let pick = users.iter().map(|u| u["id"].as_str().unwrap().to_string()).find(|u| !before.contains(u)).expect("someone not yet reviewing");
691
692        assert_eq!(api.post("/api/pr/request-reviewer", serde_json::json!({ "conn": "github", "id": id, "user_id": pick })).await, 200);
693        let after = api.detail(&id).await;
694        let added = after["pull_request"]["reviewers"].as_array().unwrap().iter().find(|r| r["user"]["id"] == pick.as_str()).cloned();
695        let added = added.expect("the requested reviewer is on the PR");
696        assert_eq!(added["vote"], "NoVote", "requested, not yet voted");
697
698        let missing = api.client.get(format!("{}/api/pr/reviewers?conn=nope&id={id}", api.base)).header("x-forgetop-token", &api.token).send().await.unwrap();
699        assert_eq!(missing.status(), 404, "a bad connection is a 404");
700    }
701}