forgetop-core 1.10.2

//! Runtime config service + resolver/section/health services.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use tokio::sync::watch;

use crate::config::*;
use crate::domain::{PipelineDefinition, PipelineRun, Section};
use crate::error::{Error, Result};
use crate::provider::*;
use crate::secret::SecretStore;

/// Holds the live config and applies mutations at runtime: each change is validated
/// against provider capabilities and persisted. Secrets go to the secret store; the
/// config only ever references them by key.
pub struct ConfigService {
    store: Arc<dyn ConfigStore>,
    secrets: Arc<dyn SecretStore>,
    registry: Arc<ProviderRegistry>,
    config: Mutex<ForgetopConfig>,
    /// Bumped whenever a persisted change alters what should be fetched — connections
    /// or section bindings, never UI preferences. The dashboard and the TUI share one
    /// `ConfigService`, so this is how a connection added in the browser reaches the
    /// terminal without waiting for the refresh tick.
    data_changed: watch::Sender<u64>,
}

impl ConfigService {
    pub fn new(store: Arc<dyn ConfigStore>, secrets: Arc<dyn SecretStore>, registry: Arc<ProviderRegistry>) -> Self {
        Self {
            store,
            secrets,
            registry,
            config: Mutex::new(ForgetopConfig::default()),
            data_changed: watch::channel(0).0,
        }
    }

    pub async fn load(&self) -> Result<()> {
        let mut cfg = self.store.load().await?;
        if migrate_pipelines_opt_in(&mut cfg) {
            self.store.save(&cfg).await?;
        }
        *self.config.lock().unwrap() = cfg;
        Ok(())
    }

    pub fn snapshot(&self) -> ForgetopConfig {
        self.config.lock().unwrap().clone()
    }

    async fn persist(&self, cfg: ForgetopConfig) -> Result<()> {
        self.store.save(&cfg).await?;
        let fetch_changed = {
            let mut current = self.config.lock().unwrap();
            let changed = !same_fetch_inputs(&current, &cfg);
            *current = cfg;
            changed
        };
        // Deliberately after the lock is released, and only for changes that alter what
        // gets fetched: a theme or sort tweak must not cost a round of network calls.
        if fetch_changed {
            self.data_changed.send_modify(|generation| *generation += 1);
        }
        Ok(())
    }

    /// Fires whenever connections or section bindings change, from anywhere in the
    /// process. A `watch` channel coalesces bursts, so the multi-step setup flow
    /// (add connection, then bind each section) settles into one reload rather than four.
    pub fn subscribe_data_changed(&self) -> watch::Receiver<u64> {
        self.data_changed.subscribe()
    }

    fn ensure_supports(&self, cfg: &ForgetopConfig, connection_id: &str, section: Section) -> Result<()> {
        let conn = cfg
            .find_connection(connection_id)
            .ok_or_else(|| Error::Config(format!("unknown connection '{connection_id}'")))?;
        let caps = self
            .registry
            .describe(conn.provider_type)
            .ok_or_else(|| Error::Config(format!("provider '{}' not registered", conn.provider_type.as_str())))?;
        if !caps.supports(section) {
            return Err(Error::Config(format!(
                "connection '{}' ({}) does not support {section:?}",
                conn.display_name,
                conn.provider_type.as_str()
            )));
        }
        Ok(())
    }

    pub async fn add_or_update_connection(&self, mut connection: Connection, secret: Option<String>) -> Result<()> {
        let cred = connection.credential_ref.clone().unwrap_or_else(|| connection.id.clone());
        connection.credential_ref = Some(cred.clone());

        if let Some(secret) = secret {
            if !self.secrets.is_writable() {
                return Err(Error::Config(
                    "secret store is read-only; provide the PAT via environment variable instead".into(),
                ));
            }
            self.secrets.set(&cred, &secret)?;
        }

        let mut cfg = self.snapshot();
        cfg.connections.retain(|c| c.id != connection.id);
        cfg.connections.push(connection);
        self.persist(cfg).await
    }

    pub async fn remove_connection(&self, connection_id: &str) -> Result<()> {
        let mut cfg = self.snapshot();
        if let Some(existing) = cfg.find_connection(connection_id).cloned() {
            if let (Some(cred), true) = (existing.credential_ref.as_ref(), self.secrets.is_writable()) {
                let _ = self.secrets.delete(cred);
            }
        }
        cfg.connections.retain(|c| c.id != connection_id);
        if let Some(b) = &mut cfg.pull_requests {
            b.connection_ids.retain(|c| c != connection_id);
            b.connection_id = b.connection_id.take().filter(|c| c != connection_id);
            if b.ids().is_empty() {
                cfg.pull_requests = None;
            }
        }
        if let Some(b) = &mut cfg.work_items {
            b.connection_ids.retain(|c| c != connection_id);
            b.connection_id = b.connection_id.take().filter(|c| c != connection_id);
            if b.ids().is_empty() {
                cfg.work_items = None;
            }
        }
        if let Some(p) = &mut cfg.pipelines {
            p.subscriptions.retain(|s| s.connection_id != connection_id);
        }
        self.persist(cfg).await
    }

    pub async fn bind_pull_requests(&self, connection_id: &str) -> Result<()> {
        let mut cfg = self.snapshot();
        self.ensure_supports(&cfg, connection_id, Section::PullRequests)?;
        let b = cfg.pull_requests.get_or_insert_with(PullRequestBinding::default);
        let mut ids = b.ids();
        if !ids.contains(&connection_id.to_string()) {
            ids.push(connection_id.into());
        }
        *b = PullRequestBinding { connection_ids: ids, connection_id: None };
        self.persist(cfg).await
    }

    pub async fn bind_work_items(&self, connection_id: &str) -> Result<()> {
        let mut cfg = self.snapshot();
        self.ensure_supports(&cfg, connection_id, Section::WorkItems)?;
        let b = cfg.work_items.get_or_insert_with(WorkItemBinding::default);
        let mut ids = b.ids();
        if !ids.contains(&connection_id.to_string()) {
            ids.push(connection_id.into());
        }
        *b = WorkItemBinding { connection_ids: ids, connection_id: None };
        self.persist(cfg).await
    }

    /// Removes a single connection from the PR section (leaves others bound).
    pub async fn unbind_pull_requests(&self, connection_id: &str) -> Result<()> {
        let mut cfg = self.snapshot();
        if let Some(b) = &mut cfg.pull_requests {
            let ids: Vec<String> = b.ids().into_iter().filter(|c| c != connection_id).collect();
            cfg.pull_requests = (!ids.is_empty()).then_some(PullRequestBinding { connection_ids: ids, connection_id: None });
        }
        self.persist(cfg).await
    }

    /// Removes a single connection from the work-item section (leaves others bound).
    pub async fn unbind_work_items(&self, connection_id: &str) -> Result<()> {
        let mut cfg = self.snapshot();
        if let Some(b) = &mut cfg.work_items {
            let ids: Vec<String> = b.ids().into_iter().filter(|c| c != connection_id).collect();
            cfg.work_items = (!ids.is_empty()).then_some(WorkItemBinding { connection_ids: ids, connection_id: None });
        }
        self.persist(cfg).await
    }

    /// Drops a connection's pipeline subscription entirely (unbinds it from the section).
    pub async fn unbind_pipelines(&self, connection_id: &str) -> Result<()> {
        let mut cfg = self.snapshot();
        if let Some(p) = &mut cfg.pipelines {
            p.subscriptions.retain(|s| s.connection_id != connection_id);
            if p.subscriptions.is_empty() {
                cfg.pipelines = None;
            }
        }
        self.persist(cfg).await
    }

    /// Replaces the PR section's connections with an explicit set (multi-bind).
    pub async fn set_pull_request_connections(&self, connection_ids: Vec<String>) -> Result<()> {
        let mut cfg = self.snapshot();
        cfg.pull_requests = Some(PullRequestBinding { connection_ids, connection_id: None });
        self.persist(cfg).await
    }

    pub async fn set_work_item_connections(&self, connection_ids: Vec<String>) -> Result<()> {
        let mut cfg = self.snapshot();
        cfg.work_items = Some(WorkItemBinding { connection_ids, connection_id: None });
        self.persist(cfg).await
    }

    pub async fn unbind_section(&self, section: Section) -> Result<()> {
        let mut cfg = self.snapshot();
        match section {
            Section::PullRequests => cfg.pull_requests = None,
            Section::WorkItems => cfg.work_items = None,
            Section::Pipelines => cfg.pipelines = None,
        }
        self.persist(cfg).await
    }

    async fn mutate_subscription(&self, connection_id: &str, f: impl FnOnce(&mut PipelineSubscription)) -> Result<()> {
        let mut cfg = self.snapshot();
        self.ensure_supports(&cfg, connection_id, Section::Pipelines)?;
        let binding = cfg.pipelines.get_or_insert_with(PipelineBinding::default);
        let sub = match binding.subscriptions.iter_mut().find(|s| s.connection_id == connection_id) {
            Some(existing) => existing,
            None => {
                binding.subscriptions.push(PipelineSubscription {
                    connection_id: connection_id.into(),
                    definition_ids: Vec::new(),
                    auto_discover_all: false,
                });
                binding.subscriptions.last_mut().unwrap()
            }
        };
        f(sub);
        self.persist(cfg).await
    }

    pub async fn subscribe_pipeline(&self, connection_id: &str, definition_id: &str) -> Result<()> {
        let definition_id = definition_id.to_string();
        self.mutate_subscription(connection_id, move |s| {
            if !s.definition_ids.contains(&definition_id) {
                s.definition_ids.push(definition_id);
            }
        })
        .await
    }

    /// Adds a connection to the Pipelines section with nothing selected — pipelines are opt-in,
    /// chosen from the picker. An existing subscription is left exactly as it is, so re-binding
    /// never undoes a selection.
    pub async fn bind_pipelines(&self, connection_id: &str) -> Result<()> {
        self.mutate_subscription(connection_id, |_| {}).await
    }

    pub async fn set_pipeline_auto_discover(&self, connection_id: &str, auto_discover_all: bool) -> Result<()> {
        self.mutate_subscription(connection_id, move |s| s.auto_discover_all = auto_discover_all).await
    }

    pub async fn unsubscribe_pipeline(&self, connection_id: &str, definition_id: &str) -> Result<()> {
        self.mutate_subscription(connection_id, move |s| s.definition_ids.retain(|d| d != definition_id)).await
    }

    pub async fn remove_pipeline_connection(&self, connection_id: &str) -> Result<()> {
        let mut cfg = self.snapshot();
        if let Some(p) = &mut cfg.pipelines {
            p.subscriptions.retain(|s| s.connection_id != connection_id);
        }
        self.persist(cfg).await
    }

    /// Replaces a connection's repository scope. `Some(vec![])` is a real, respected state — the
    /// user chose no repositories — and must not be confused with `None`, which means the scope
    /// was never established and the legacy single repository still applies.
    pub async fn set_repo_scope(&self, connection_id: &str, scope: Option<Vec<String>>) -> Result<()> {
        let mut cfg = self.snapshot();
        let conn = cfg
            .connections
            .iter_mut()
            .find(|c| c.id == connection_id)
            .ok_or_else(|| Error::Config(format!("unknown connection '{connection_id}'")))?;
        conn.repo_scope = scope;
        self.persist(cfg).await
    }

    /// A connection's stored scope, exactly as stored (including the `None` / `Some([])`
    /// distinction) — the picker needs to tell "never chosen" from "chose none".
    pub fn repo_scope(&self, connection_id: &str) -> Option<Vec<String>> {
        self.snapshot().find_connection(connection_id).and_then(|c| c.repo_scope.clone())
    }

    pub async fn set_theme(&self, theme: Option<String>) -> Result<()> {
        let mut cfg = self.snapshot();
        cfg.ui.theme = theme;
        self.persist(cfg).await
    }

    pub async fn set_hidden_sections(&self, hidden: Vec<Section>) -> Result<()> {
        let mut cfg = self.snapshot();
        cfg.ui.hidden_sections = hidden;
        self.persist(cfg).await
    }

    pub async fn set_preview_hidden(&self, hidden: Vec<Section>) -> Result<()> {
        let mut cfg = self.snapshot();
        cfg.ui.preview_hidden = Some(hidden);
        self.persist(cfg).await
    }

    pub async fn set_hidden_work_item_states(&self, hidden: Vec<String>) -> Result<()> {
        let mut cfg = self.snapshot();
        cfg.ui.hidden_work_item_states = hidden;
        self.persist(cfg).await
    }

    pub async fn set_hidden_columns(&self, hidden: [Vec<String>; 3]) -> Result<()> {
        let mut cfg = self.snapshot();
        cfg.ui.hidden_columns = Some(hidden);
        self.persist(cfg).await
    }

    pub async fn set_dismissed_launchpad_items(&self, dismissed: Vec<String>) -> Result<()> {
        let mut cfg = self.snapshot();
        cfg.ui.dismissed_launchpad_items = dismissed;
        self.persist(cfg).await
    }

    pub async fn set_dismissed_notifications(&self, dismissed: Vec<String>) -> Result<()> {
        let mut cfg = self.snapshot();
        cfg.ui.dismissed_notifications = dismissed;
        self.persist(cfg).await
    }

    pub async fn set_notifications(&self, prefs: NotificationPrefs) -> Result<()> {
        let mut cfg = self.snapshot();
        cfg.ui.notifications = prefs;
        self.persist(cfg).await
    }

    /// Persists the saved views for a section.
    pub async fn set_views(&self, section: Section, views: Vec<SavedView>) -> Result<()> {
        let mut cfg = self.snapshot();
        match section {
            Section::PullRequests => cfg.ui.pr_views = views,
            Section::WorkItems => cfg.ui.work_item_views = views,
            Section::Pipelines => cfg.ui.pipeline_views = views,
        }
        self.persist(cfg).await
    }

    /// Persists the sort for a section (`None` restores provider order).
    pub async fn set_sort(&self, section: Section, sort: Option<SortPref>) -> Result<()> {
        let mut cfg = self.snapshot();
        match section {
            Section::PullRequests => cfg.ui.pr_sort = sort,
            Section::WorkItems => cfg.ui.work_item_sort = sort,
            Section::Pipelines => cfg.ui.pipeline_sort = sort,
        }
        self.persist(cfg).await
    }

    /// Persists how the Pipelines list is grouped (`None` restores the default).
    pub async fn set_pipeline_group(&self, group: Option<String>) -> Result<()> {
        let mut cfg = self.snapshot();
        cfg.ui.pipeline_group = group;
        self.persist(cfg).await
    }

    /// Persists how the Work Items list is grouped (`None` restores the default).
    pub async fn set_work_item_group(&self, group: Option<String>) -> Result<()> {
        let mut cfg = self.snapshot();
        cfg.ui.work_item_group = group;
        self.persist(cfg).await
    }

    /// Replaces a connection's tracked pipeline definitions with an explicit set
    /// (turns off auto-discovery). An empty list tracks nothing.
    pub async fn set_pipeline_definitions(&self, connection_id: &str, definition_ids: Vec<String>) -> Result<()> {
        self.mutate_subscription(connection_id, move |s| {
            s.auto_discover_all = false;
            s.definition_ids = definition_ids;
        })
        .await
    }
}

/// The one-time switch to opt-in pipelines: a subscription saved as "every pipeline" was, until
/// now, what binding a connection did on its own, so it is reset to nothing selected. Runs once —
/// the marker keeps a later deliberate "every pipeline" from being reset. Returns whether the
/// saved config needs rewriting (a subscription was reset).
fn migrate_pipelines_opt_in(cfg: &mut ForgetopConfig) -> bool {
    if cfg.pipelines_opt_in {
        return false;
    }
    cfg.pipelines_opt_in = true;
    let Some(binding) = &mut cfg.pipelines else { return false };
    let mut changed = false;
    // An empty list with the flag off also used to mean "everything". An explicit list was a
    // real choice and is kept.
    for sub in binding.subscriptions.iter_mut().filter(|s| s.auto_discover_all || s.definition_ids.is_empty()) {
        sub.auto_discover_all = false;
        sub.definition_ids.clear();
        changed = true;
    }
    changed
}

/// Turns a configured connection id into a live [`ProviderConnection`].
pub struct ConnectionResolver {
    config: Arc<ConfigService>,
    registry: Arc<ProviderRegistry>,
    secrets: Arc<dyn SecretStore>,
    /// connection id → the live connection last created for it. Reused while the connection's
    /// configuration and credential are unchanged: a provider client keeps caches (GitHub's
    /// signed-in login and the full fetch behind each search hit) that only pay off if the client
    /// outlives one call. Created afresh per call, every reload paid for every search hit again —
    /// enough to exhaust GitHub's hourly rate limit on a busy repository.
    live: Mutex<HashMap<String, LiveConnection>>,
}

/// A created connection with what it was created from, so a change to either recreates it.
struct LiveConnection {
    connection: Connection,
    /// A fingerprint of the credential, not the credential: the client holds that already.
    secret: u64,
    live: Arc<dyn ProviderConnection>,
}

/// A change-detection fingerprint of a credential (not a secure hash — the client holds the
/// credential itself; this only says whether it is the one the client was built with).
fn secret_fingerprint(secret: Option<&str>) -> u64 {
    use std::hash::{Hash, Hasher};
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    secret.hash(&mut hasher);
    hasher.finish()
}

impl ConnectionResolver {
    pub fn new(config: Arc<ConfigService>, registry: Arc<ProviderRegistry>, secrets: Arc<dyn SecretStore>) -> Self {
        Self { config, registry, secrets, live: Mutex::new(HashMap::new()) }
    }

    pub async fn resolve(&self, connection_id: &str) -> Result<Option<Arc<dyn ProviderConnection>>> {
        // The lock is taken before the snapshot and held across no `.await`: two resolves racing
        // a configuration change then build from the same view of it, and the later one cannot
        // put a client built from the older configuration over a newer one.
        let mut live = self.live.lock().unwrap_or_else(|poisoned| poisoned.into_inner());
        let cfg = self.config.snapshot();
        // A connection that is gone takes its client with it — the credential it was built with
        // has no business staying in memory — whichever connection this call is about.
        live.retain(|id, _| cfg.find_connection(id).is_some());
        let conn = match cfg.find_connection(connection_id) {
            Some(c) => c.clone(),
            None => return Ok(None),
        };
        if !self.registry.supports(conn.provider_type) {
            return Ok(None);
        }
        let secret = match &conn.credential_ref {
            Some(r) => self.secrets.get(r)?,
            None => None,
        };
        let fingerprint = secret_fingerprint(secret.as_deref());
        if let Some(entry) = live.get(connection_id) {
            if entry.connection == conn && entry.secret == fingerprint {
                return Ok(Some(entry.live.clone()));
            }
        }
        let created = self.registry.create(&conn, secret)?;
        live.insert(connection_id.to_string(), LiveConnection { connection: conn, secret: fingerprint, live: created.clone() });
        Ok(Some(created))
    }
}

/// One pipeline connection feeding the Pipelines section, with its subscription.
pub struct PipelineFeed {
    pub connection: Arc<dyn ProviderConnection>,
    pub source: Arc<dyn PipelineSource>,
    pub subscription: PipelineSubscription,
}

/// How many runs are fetched for each pipeline definition.
pub const RUNS_PER_DEFINITION: u32 = 10;

/// How many run queries a feed has in flight at once — polite to the provider's rate limit.
const MAX_CONCURRENT_RUN_QUERIES: usize = 4;

/// The run queries for a pipeline subscription: one per definition, never one for the whole
/// connection.
///
/// A connection-wide "latest N runs" is a window a busy pipeline fills on its own — twenty PR
/// checks push every quieter pipeline's history out, and a main-line build that ran dozens of
/// times reads as "1 run". So subscribing to everything asks each discovered definition for its
/// own recent runs, exactly as an explicit subscription does.
///
/// A definition id is only unique within its repository, so each query is addressed at the
/// repository discovery says the definition belongs to. Pipelines are opt-in, so nothing
/// selected asks for nothing. Only when there is nothing to address —
/// subscribed to everything and discovery came back empty — does it fall back to one
/// connection-wide query, so a discovery hiccup still shows something.
pub fn pipeline_run_queries(sub: &PipelineSubscription, defs: &[PipelineDefinition]) -> Vec<PipelineRunQuery> {
    // Pipelines are opt-in: nothing selected fetches nothing.
    if !sub.auto_discover_all && sub.definition_ids.is_empty() {
        return Vec::new();
    }
    let all = sub.auto_discover_all;
    let ids: Vec<&String> = if all { defs.iter().map(|d| &d.id).collect() } else { sub.definition_ids.iter().collect() };
    if ids.is_empty() {
        return vec![PipelineRunQuery { definition_id: None, repository: None, branch: None, limit: Some(20) }];
    }
    ids.into_iter()
        .map(|id| PipelineRunQuery {
            repository: defs.iter().find(|d| &d.id == id).and_then(|d| d.repository.clone()),
            definition_id: Some(id.clone()),
            branch: None,
            limit: Some(RUNS_PER_DEFINITION),
        })
        .collect()
}

impl PipelineFeed {
    /// Runs every query of this feed's subscription (see [`pipeline_run_queries`]), a few at a
    /// time, and returns each query's result in order so the caller reports failures its own way.
    pub async fn list_runs(&self, defs: &[PipelineDefinition]) -> Vec<Result<Vec<PipelineRun>>> {
        let queries = pipeline_run_queries(&self.subscription, defs);
        let mut out = Vec::with_capacity(queries.len());
        for chunk in queries.chunks(MAX_CONCURRENT_RUN_QUERIES) {
            out.extend(futures_util::future::join_all(chunk.iter().map(|q| self.source.list_runs(q))).await);
        }
        out
    }
}

/// One connection feeding the aggregated Pull Requests list.
pub struct PullRequestFeed {
    pub connection: Arc<dyn ProviderConnection>,
    pub source: Arc<dyn PullRequestSource>,
}

/// One connection feeding the aggregated Work Items list.
pub struct WorkItemFeed {
    pub connection: Arc<dyn ProviderConnection>,
    pub source: Arc<dyn WorkItemSource>,
}

/// One connection contributing to the notification inbox. Unlike the section feeds this
/// isn't binding-scoped — every connection with a notification feed contributes.
pub struct NotificationFeed {
    pub connection: Arc<dyn ProviderConnection>,
    pub source: Arc<dyn NotificationSource>,
}

/// Resolves the live source(s) backing each section from the current bindings.
pub struct SectionService {
    config: Arc<ConfigService>,
    resolver: Arc<ConnectionResolver>,
}

impl SectionService {
    pub fn new(config: Arc<ConfigService>, resolver: Arc<ConnectionResolver>) -> Self {
        Self { config, resolver }
    }

    /// One connection feeding the PR / Work-Item list (for aggregation).
    pub async fn pull_request_feeds(&self) -> Result<Vec<PullRequestFeed>> {
        let cfg = self.config.snapshot();
        let Some(binding) = cfg.pull_requests else { return Ok(Vec::new()) };
        let mut feeds = Vec::new();
        for id in binding.ids() {
            if let Some(conn) = self.resolver.resolve(&id).await? {
                if let Some(source) = conn.pull_requests() {
                    feeds.push(PullRequestFeed { connection: conn, source });
                }
            }
        }
        Ok(feeds)
    }

    pub async fn work_item_feeds(&self) -> Result<Vec<WorkItemFeed>> {
        let cfg = self.config.snapshot();
        let Some(binding) = cfg.work_items else { return Ok(Vec::new()) };
        let mut feeds = Vec::new();
        for id in binding.ids() {
            if let Some(conn) = self.resolver.resolve(&id).await? {
                if let Some(source) = conn.work_items() {
                    feeds.push(WorkItemFeed { connection: conn, source });
                }
            }
        }
        Ok(feeds)
    }

    /// Every connection that exposes a notification feed (across all connections, not just
    /// section-bound ones) — the inbox aggregates them all.
    pub async fn notification_feeds(&self) -> Result<Vec<NotificationFeed>> {
        let cfg = self.config.snapshot();
        let mut feeds = Vec::new();
        for c in &cfg.connections {
            if let Some(conn) = self.resolver.resolve(&c.id).await? {
                if let Some(source) = conn.notifications() {
                    feeds.push(NotificationFeed { connection: conn, source });
                }
            }
        }
        Ok(feeds)
    }

    /// The first bound PR source (back-compat for single-source call sites).
    pub async fn pull_request_source(&self) -> Result<Option<Arc<dyn PullRequestSource>>> {
        Ok(self.pull_request_feeds().await?.into_iter().next().map(|f| f.source))
    }

    pub async fn work_item_source(&self) -> Result<Option<Arc<dyn WorkItemSource>>> {
        Ok(self.work_item_feeds().await?.into_iter().next().map(|f| f.source))
    }

    /// Resolves a connection's pipeline source directly, regardless of whether it is
    /// currently subscribed — used to discover definitions before subscribing.
    pub async fn pipeline_source_for(&self, connection_id: &str) -> Result<Option<Arc<dyn PipelineSource>>> {
        Ok(self.resolver.resolve(connection_id).await?.and_then(|c| c.pipelines()))
    }

    /// The repositories a connection's credentials can reach, for its scope picker.
    pub async fn discover_repositories(&self, connection_id: &str) -> Result<RepositoryPage> {
        match self.resolver.resolve(connection_id).await? {
            Some(conn) => conn.discover_repositories().await,
            None => Ok(RepositoryPage::default()),
        }
    }

    pub async fn pipeline_feeds(&self) -> Result<Vec<PipelineFeed>> {
        let cfg = self.config.snapshot();
        let Some(binding) = cfg.pipelines else { return Ok(Vec::new()) };
        let mut feeds = Vec::new();
        for sub in binding.subscriptions {
            if let Some(conn) = self.resolver.resolve(&sub.connection_id).await? {
                if let Some(source) = conn.pipelines() {
                    feeds.push(PipelineFeed { connection: conn, source, subscription: sub });
                }
            }
        }
        Ok(feeds)
    }
}

/// Gives a freshly-added connection a starting repository scope: none of them, chosen.
///
/// Discovery still runs, so the repositories the credentials reach are known up front and the
/// user picks from them ("0 of 38"); nothing is fetched until they do. Returns that discovery
/// page, for a frontend to show the count without discovering again.
///
/// Best-effort and deliberately inert on anything already established — it only acts when the
/// scope has never been set *and* no legacy single repository was supplied, so it can neither
/// overwrite a user's choice nor change an existing connection. If discovery fails or returns
/// nothing, the scope stays `None` and the connection behaves exactly as it did before: an
/// empty scope is only a useful starting point when there is something to pick.
pub async fn seed_default_repo_scope(
    config: &ConfigService,
    sections: &SectionService,
    connection_id: &str,
) -> Result<Option<RepositoryPage>> {
    let Some(conn) = config.snapshot().find_connection(connection_id).cloned() else { return Ok(None) };
    if conn.repo_scope.is_some() || conn.repository.is_some() {
        return Ok(None);
    }
    let page = sections.discover_repositories(connection_id).await.unwrap_or_default();
    if page.repositories.is_empty() {
        return Ok(None);
    }
    config.set_repo_scope(connection_id, Some(Vec::new())).await?;
    Ok(Some(page))
}

/// A configured connection and whether it's currently reachable/authed.
pub struct ConnectionHealth {
    pub connection: Connection,
    pub healthy: bool,
}

/// Probes each configured connection for the connections health bar.
pub struct ConnectionHealthService {
    config: Arc<ConfigService>,
    resolver: Arc<ConnectionResolver>,
}

impl ConnectionHealthService {
    pub fn new(config: Arc<ConfigService>, resolver: Arc<ConnectionResolver>) -> Self {
        Self { config, resolver }
    }

    pub async fn check_all(&self) -> Vec<ConnectionHealth> {
        let cfg = self.config.snapshot();
        let mut out = Vec::new();
        for connection in cfg.connections {
            let healthy = match self.resolver.resolve(&connection.id).await {
                Ok(Some(live)) => live.check().await,
                _ => false,
            };
            out.push(ConnectionHealth { connection, healthy });
        }
        out
    }
}


/// Whether two configs would produce the same fetches. Compares every field except
/// `ui`, so a new data-bearing field is included by default rather than being silently
/// forgotten — the failure mode we want is a redundant refresh, not a stale screen.
fn same_fetch_inputs(a: &ForgetopConfig, b: &ForgetopConfig) -> bool {
    a.connections == b.connections
        && a.pull_requests == b.pull_requests
        && a.work_items == b.work_items
        && a.pipelines == b.pipelines
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::domain::ProviderType;
    use crate::secret::InMemorySecretStore;
    use async_trait::async_trait;

    struct FakeConn {
        caps: Capabilities,
    }
    #[async_trait]
    impl ProviderConnection for FakeConn {
        fn connection_id(&self) -> &str { "fake" }
        fn provider_type(&self) -> ProviderType { ProviderType::GitHub }
        fn display_name(&self) -> &str { "Fake" }
        fn capabilities(&self) -> &Capabilities { &self.caps }
        fn pull_requests(&self) -> Option<Arc<dyn PullRequestSource>> { None }
        fn work_items(&self) -> Option<Arc<dyn WorkItemSource>> { None }
        fn pipelines(&self) -> Option<Arc<dyn PipelineSource>> { None }
        async fn check(&self) -> bool { true }
    }
    struct FakeFactory {
        provider: ProviderType,
        caps: Capabilities,
        /// How many connections this factory has built.
        created: Arc<std::sync::atomic::AtomicUsize>,
    }
    impl ProviderFactory for FakeFactory {
        fn provider_type(&self) -> ProviderType { self.provider }
        fn describe_capabilities(&self) -> Capabilities { self.caps.clone() }
        fn create(&self, _c: &Connection, _s: Option<String>) -> Result<Arc<dyn ProviderConnection>> {
            self.created.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
            Ok(Arc::new(FakeConn { caps: self.caps.clone() }))
        }
    }

    fn registry() -> Arc<ProviderRegistry> {
        counting_registry().0
    }

    /// The registry, and how many GitHub connections it has built so far.
    fn counting_registry() -> (Arc<ProviderRegistry>, Arc<std::sync::atomic::AtomicUsize>) {
        let created = Arc::new(std::sync::atomic::AtomicUsize::new(0));
        let registry = Arc::new(ProviderRegistry::new(vec![
            Arc::new(FakeFactory {
                provider: ProviderType::GitHub,
                caps: Capabilities { supports_pull_requests: true, supports_pipelines: true, ..Default::default() },
                created: created.clone(),
            }),
            Arc::new(FakeFactory {
                provider: ProviderType::Linear,
                caps: Capabilities { supports_work_items: true, ..Default::default() },
                created: Arc::default(),
            }),
        ]));
        (registry, created)
    }

    /// A resolver over a fresh config, with the count of GitHub connections built.
    fn resolver() -> (ConnectionResolver, Arc<ConfigService>, Arc<std::sync::atomic::AtomicUsize>) {
        let (registry, created) = counting_registry();
        let secrets = Arc::new(InMemorySecretStore::default());
        let svc = Arc::new(ConfigService::new(Arc::new(InMemoryConfigStore::default()), secrets.clone(), registry.clone()));
        (ConnectionResolver::new(svc.clone(), registry, secrets), svc, created)
    }

    #[tokio::test]
    async fn resolving_a_connection_again_reuses_the_live_one() {
        let (resolver, svc, created) = resolver();
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), Some("pat".into())).await.unwrap();
        let first = resolver.resolve("gh-1").await.unwrap().expect("resolved");
        let again = resolver.resolve("gh-1").await.unwrap().expect("resolved");
        assert!(Arc::ptr_eq(&first, &again), "the same live connection, caches and all");
        assert_eq!(created.load(std::sync::atomic::Ordering::SeqCst), 1, "built once");
    }

    #[tokio::test]
    async fn a_changed_credential_or_configuration_recreates_the_connection() {
        let (resolver, svc, created) = resolver();
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), Some("pat".into())).await.unwrap();
        let first = resolver.resolve("gh-1").await.unwrap().unwrap();

        // A new token: the old client would keep sending the old one.
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), Some("pat-2".into())).await.unwrap();
        let rekeyed = resolver.resolve("gh-1").await.unwrap().unwrap();
        assert!(!Arc::ptr_eq(&first, &rekeyed), "rebuilt with the new credential");
        assert_eq!(created.load(std::sync::atomic::Ordering::SeqCst), 2);

        // A new scope: the client fetches from the repositories it was built with.
        let mut rescoped = conn("gh-1", ProviderType::GitHub);
        rescoped.repo_scope = Some(vec!["acme/pay".into()]);
        svc.add_or_update_connection(rescoped, None).await.unwrap();
        let widened = resolver.resolve("gh-1").await.unwrap().unwrap();
        assert!(!Arc::ptr_eq(&rekeyed, &widened), "rebuilt with the new scope");
        assert_eq!(created.load(std::sync::atomic::Ordering::SeqCst), 3);

        // Unchanged since: reused.
        let same = resolver.resolve("gh-1").await.unwrap().unwrap();
        assert!(Arc::ptr_eq(&widened, &same));
        assert_eq!(created.load(std::sync::atomic::Ordering::SeqCst), 3);
    }

    #[tokio::test]
    async fn a_removed_connection_is_let_go_on_the_next_resolve_of_any_connection() {
        let (resolver, svc, _) = resolver();
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), Some("pat".into())).await.unwrap();
        svc.add_or_update_connection(conn("gh-2", ProviderType::GitHub), Some("pat".into())).await.unwrap();
        resolver.resolve("gh-1").await.unwrap().expect("resolved");
        resolver.resolve("gh-2").await.unwrap().expect("resolved");
        svc.remove_connection("gh-1").await.unwrap();
        // Nothing resolves a removed id again (feeds iterate the connections that exist), so the
        // pruning has to ride on whichever resolve comes next.
        resolver.resolve("gh-2").await.unwrap().expect("still there");
        let live: Vec<String> = resolver.live.lock().unwrap().keys().cloned().collect();
        assert_eq!(live, vec!["gh-2".to_string()], "the removed connection's client is gone");
        assert!(resolver.resolve("gh-1").await.unwrap().is_none());
    }

    fn conn(id: &str, provider: ProviderType) -> Connection {
        Connection {
            id: id.into(),
            provider_type: provider,
            display_name: id.into(),
            base_url: None,
            organization: None,
            project: None,
            repository: None,
            username: None,
            credential_ref: None,
            repo_scope: None,
        }
    }

    fn service() -> (Arc<ConfigService>, Arc<InMemorySecretStore>) {
        let secrets = Arc::new(InMemorySecretStore::default());
        let svc = Arc::new(ConfigService::new(
            Arc::new(InMemoryConfigStore::default()),
            secrets.clone(),
            registry(),
        ));
        (svc, secrets)
    }

    #[tokio::test]
    async fn add_connection_persists_and_stores_secret() {
        let (svc, secrets) = service();
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), Some("pat".into())).await.unwrap();
        assert_eq!(svc.snapshot().connections.len(), 1);
        assert_eq!(secrets.get("gh-1").unwrap().as_deref(), Some("pat"));
    }

    #[tokio::test]
    async fn connection_and_binding_changes_notify_subscribers() {
        let (svc, _) = service();
        let mut rx = svc.subscribe_data_changed();

        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), Some("pat".into())).await.unwrap();
        assert!(rx.has_changed().unwrap(), "adding a connection must wake the TUI");
        rx.borrow_and_update();

        svc.bind_pull_requests("gh-1").await.unwrap();
        assert!(rx.has_changed().unwrap(), "binding a section must wake the TUI");
        rx.borrow_and_update();

        svc.remove_connection("gh-1").await.unwrap();
        assert!(rx.has_changed().unwrap(), "removing a connection must wake the TUI");
    }

    #[tokio::test]
    async fn ui_preferences_do_not_notify_subscribers() {
        let (svc, _) = service();
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), None).await.unwrap();
        let rx = svc.subscribe_data_changed();

        // None of these change what gets fetched, so none should cost a round of
        // network calls. This is the half of the behaviour that regresses silently.
        svc.set_theme(Some("slate".into())).await.unwrap();
        svc.set_hidden_sections(vec![Section::Pipelines]).await.unwrap();
        svc.set_notifications(NotificationPrefs::default()).await.unwrap();

        assert!(!rx.has_changed().unwrap(), "a UI preference must not trigger a refetch");
    }

    #[tokio::test]
    async fn rewriting_an_identical_connection_does_not_notify() {
        let (svc, _) = service();
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), None).await.unwrap();
        let rx = svc.subscribe_data_changed();

        // The dashboard re-saves on every form submit, including no-op ones.
        let mut same = conn("gh-1", ProviderType::GitHub);
        same.credential_ref = Some("gh-1".into());
        svc.add_or_update_connection(same, None).await.unwrap();

        assert!(!rx.has_changed().unwrap(), "an unchanged rewrite must not trigger a refetch");
    }

    #[tokio::test]
    async fn bind_validates_capability() {
        let (svc, _) = service();
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), None).await.unwrap();
        svc.add_or_update_connection(conn("lin-1", ProviderType::Linear), None).await.unwrap();
        svc.bind_pull_requests("gh-1").await.unwrap();
        assert_eq!(svc.snapshot().pull_requests.unwrap().ids(), vec!["gh-1".to_string()]);
        assert!(svc.bind_pull_requests("lin-1").await.is_err());
    }

    #[tokio::test]
    async fn remove_connection_cascades() {
        let (svc, secrets) = service();
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), Some("pat".into())).await.unwrap();
        svc.bind_pull_requests("gh-1").await.unwrap();
        svc.subscribe_pipeline("gh-1", "ci").await.unwrap();
        svc.remove_connection("gh-1").await.unwrap();
        let cfg = svc.snapshot();
        assert!(cfg.connections.is_empty());
        assert!(cfg.pull_requests.is_none());
        assert!(cfg.pipelines.unwrap().subscriptions.is_empty());
        assert_eq!(secrets.get("gh-1").unwrap(), None);
    }

    #[tokio::test]
    async fn repo_scope_persists_including_the_empty_choice() {
        let (svc, _) = service();
        let mut c = conn("gh-1", ProviderType::GitHub);
        c.repository = Some("acme/pay".into());
        svc.add_or_update_connection(c, None).await.unwrap();

        // Never set → absent, so the legacy repository is still what the factory resolves.
        assert_eq!(svc.repo_scope("gh-1"), None);

        svc.set_repo_scope("gh-1", Some(vec!["acme/pay".into(), "acme/ledger".into()])).await.unwrap();
        assert_eq!(svc.repo_scope("gh-1").unwrap().len(), 2);

        // "I chose none" is a real, persisted state — not the same as never having chosen.
        svc.set_repo_scope("gh-1", Some(vec![])).await.unwrap();
        assert_eq!(svc.repo_scope("gh-1"), Some(vec![]));
        svc.load().await.unwrap();
        assert_eq!(svc.repo_scope("gh-1"), Some(vec![]), "an emptied scope survives a reload");

        assert!(svc.set_repo_scope("nope", None).await.is_err());
    }

    #[tokio::test]
    async fn pipeline_subscribe_unsubscribe() {
        let (svc, _) = service();
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), None).await.unwrap();
        svc.subscribe_pipeline("gh-1", "ci").await.unwrap();
        svc.subscribe_pipeline("gh-1", "release").await.unwrap();
        svc.subscribe_pipeline("gh-1", "ci").await.unwrap(); // duplicate ignored
        assert_eq!(svc.snapshot().pipelines.unwrap().subscriptions[0].definition_ids.len(), 2);
        svc.unsubscribe_pipeline("gh-1", "ci").await.unwrap();
        assert_eq!(svc.snapshot().pipelines.unwrap().subscriptions[0].definition_ids.len(), 1);
    }

    #[tokio::test]
    async fn set_pipeline_definitions_replaces_and_disables_auto() {
        let (svc, _) = service();
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), None).await.unwrap();
        svc.set_pipeline_auto_discover("gh-1", true).await.unwrap();
        svc.set_pipeline_definitions("gh-1", vec!["ci".into(), "release".into()]).await.unwrap();
        let cfg = svc.snapshot();
        let sub = &cfg.pipelines.unwrap().subscriptions[0];
        assert!(!sub.auto_discover_all);
        assert_eq!(sub.definition_ids, vec!["ci".to_string(), "release".to_string()]);
    }

    fn def(repo: &str, id: &str) -> PipelineDefinition {
        PipelineDefinition { repository: Some(repo.into()), id: id.into(), name: id.into(), path: None, url: None }
    }

    /// A subscribed definition id is only unique within its repository, so each pipeline query
    /// must be addressed at the repository discovery says the definition belongs to.
    #[test]
    fn pipeline_queries_are_addressed_at_the_definition_s_own_repository() {
        let defs = vec![def("acme/pay", "ci"), def("acme/web", "release")];
        let sub = PipelineSubscription {
            connection_id: "gh".into(),
            definition_ids: vec!["ci".into(), "release".into()],
            auto_discover_all: false,
        };
        let queries = pipeline_run_queries(&sub, &defs);
        assert_eq!(queries[0].repository.as_deref(), Some("acme/pay"));
        assert_eq!(queries[1].repository.as_deref(), Some("acme/web"));
    }

    /// Subscribing to everything must not mean "the connection's latest 20 runs": a busy PR check
    /// fills that window alone and a main-line pipeline with dozens of runs reads as one. Each
    /// discovered definition gets its own query instead.
    #[test]
    fn subscribing_to_everything_queries_every_definition_on_its_own() {
        let defs = vec![def("proj/app", "pr-check"), def("proj/app", "mainline"), def("proj/web", "deploy")];
        for sub in [
            PipelineSubscription { connection_id: "ado".into(), definition_ids: vec![], auto_discover_all: true },
            // Auto-discovery wins over a stale explicit list.
            PipelineSubscription { connection_id: "ado".into(), definition_ids: vec!["pr-check".into()], auto_discover_all: true },
        ] {
            let queries = pipeline_run_queries(&sub, &defs);
            let ids: Vec<_> = queries.iter().map(|q| q.definition_id.as_deref()).collect();
            assert_eq!(ids, vec![Some("pr-check"), Some("mainline"), Some("deploy")]);
            assert!(queries.iter().all(|q| q.limit == Some(RUNS_PER_DEFINITION)));
            assert_eq!(queries[2].repository.as_deref(), Some("proj/web"));
        }
    }

    /// With nothing discovered there is no definition to ask about, so one connection-wide query
    /// still shows whatever recent runs there are.
    #[test]
    fn an_empty_discovery_falls_back_to_one_connection_wide_query() {
        let sub = PipelineSubscription { connection_id: "ado".into(), definition_ids: vec![], auto_discover_all: true };
        let queries = pipeline_run_queries(&sub, &[]);
        assert_eq!(queries.len(), 1);
        assert_eq!(queries[0].definition_id, None);
    }

    /// Pipelines are opt-in: a connection with nothing selected asks for no runs at all.
    #[test]
    fn nothing_selected_fetches_nothing() {
        let sub = PipelineSubscription { connection_id: "ado".into(), definition_ids: vec![], auto_discover_all: false };
        assert!(pipeline_run_queries(&sub, &[def("proj/app", "ci")]).is_empty());
        assert!(pipeline_run_queries(&sub, &[]).is_empty());
    }

    /// Binding adds the connection with nothing selected, and re-binding (the web form re-saves
    /// every section on each edit) leaves a selection alone.
    #[tokio::test]
    async fn bind_pipelines_selects_nothing_and_keeps_an_existing_selection() {
        let (svc, _) = service();
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), None).await.unwrap();
        svc.bind_pipelines("gh-1").await.unwrap();
        let sub = svc.snapshot().pipelines.unwrap().subscriptions[0].clone();
        assert!(!sub.auto_discover_all && sub.definition_ids.is_empty());

        svc.set_pipeline_definitions("gh-1", vec!["ci".into()]).await.unwrap();
        svc.bind_pipelines("gh-1").await.unwrap();
        let subs = svc.snapshot().pipelines.unwrap().subscriptions;
        assert_eq!(subs.len(), 1);
        assert_eq!(subs[0].definition_ids, vec!["ci".to_string()]);
    }

    fn sub(id: &str, ids: &[&str], all: bool) -> PipelineSubscription {
        PipelineSubscription {
            connection_id: id.into(),
            definition_ids: ids.iter().map(|s| s.to_string()).collect(),
            auto_discover_all: all,
        }
    }

    /// The switch to opt-in resets "every pipeline" subscriptions once, keeps explicit
    /// selections, and never runs again — so a later "tick everything" survives restarts.
    #[test]
    fn opt_in_migration_resets_every_pipeline_once() {
        let mut cfg = ForgetopConfig {
            pipelines: Some(PipelineBinding {
                subscriptions: vec![sub("ado", &[], true), sub("gh", &[], false), sub("gl", &["ci"], false)],
            }),
            ..Default::default()
        };
        assert!(migrate_pipelines_opt_in(&mut cfg));
        assert!(cfg.pipelines_opt_in);
        let subs = &cfg.pipelines.as_ref().unwrap().subscriptions;
        assert_eq!(subs[0], sub("ado", &[], false));
        assert_eq!(subs[1], sub("gh", &[], false));
        assert_eq!(subs[2], sub("gl", &["ci"], false), "an explicit selection is kept");

        cfg.pipelines.as_mut().unwrap().subscriptions[0].auto_discover_all = true;
        assert!(!migrate_pipelines_opt_in(&mut cfg), "runs once");
        assert!(cfg.pipelines.unwrap().subscriptions[0].auto_discover_all);
    }

    /// Through the real load path: the reset is saved, and a later "every pipeline" choice
    /// survives the next launch's load.
    #[tokio::test]
    async fn opt_in_reset_is_saved_and_not_repeated_on_the_next_load() {
        let store = Arc::new(InMemoryConfigStore::default());
        let svc = ConfigService::new(store.clone(), Arc::new(InMemorySecretStore::default()), registry());
        svc.add_or_update_connection(conn("gh-1", ProviderType::GitHub), None).await.unwrap();
        svc.set_pipeline_auto_discover("gh-1", true).await.unwrap();

        svc.load().await.unwrap();
        let saved = store.load().await.unwrap();
        assert!(saved.pipelines_opt_in);
        assert!(!saved.pipelines.unwrap().subscriptions[0].auto_discover_all, "reset on disk");

        svc.set_pipeline_auto_discover("gh-1", true).await.unwrap();
        svc.load().await.unwrap();
        assert!(svc.snapshot().pipelines.unwrap().subscriptions[0].auto_discover_all, "kept after restart");
    }

    /// A config with no pipelines is only marked, never rewritten on disk for it.
    #[test]
    fn opt_in_migration_without_pipelines_needs_no_save() {
        let mut cfg = ForgetopConfig::default();
        assert!(!migrate_pipelines_opt_in(&mut cfg));
        assert!(cfg.pipelines_opt_in);
    }
}