forgetop_core/

service.rs

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

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

use tokio::sync::watch;

use crate::config::*;
use crate::domain::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 cfg = self.store.load().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
    }

    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_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_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_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
    }

    /// 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
    }
}

/// Turns a configured connection id into a live [`ProviderConnection`].
pub struct ConnectionResolver {
    config: Arc<ConfigService>,
    registry: Arc<ProviderRegistry>,
    secrets: Arc<dyn SecretStore>,
}

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

    pub async fn resolve(&self, connection_id: &str) -> Result<Option<Arc<dyn ProviderConnection>>> {
        let cfg = self.config.snapshot();
        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,
        };
        Ok(Some(self.registry.create(&conn, secret)?))
    }
}

/// 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,
}

/// 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)
    }
}

/// How many repositories a brand-new connection starts scoped to — the most recently active
/// ones. Small on purpose: the cost of a refresh is one list call per repository, and the user
/// widens it from the scope picker.
pub const DEFAULT_REPO_SCOPE_SIZE: usize = 5;

/// Gives a freshly-added connection a starting repository scope: the most recently active
/// repositories its credentials can reach, capped at [`DEFAULT_REPO_SCOPE_SIZE`].
///
/// 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.
pub async fn seed_default_repo_scope(
    config: &ConfigService,
    sections: &SectionService,
    connection_id: &str,
) -> Result<Option<Vec<String>>> {
    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);
    }
    let scope: Vec<String> = page.repositories.into_iter().take(DEFAULT_REPO_SCOPE_SIZE).collect();
    config.set_repo_scope(connection_id, Some(scope.clone())).await?;
    Ok(Some(scope))
}

/// 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,
    }
    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>> {
            Ok(Arc::new(FakeConn { caps: self.caps.clone() }))
        }
    }

    fn registry() -> Arc<ProviderRegistry> {
        Arc::new(ProviderRegistry::new(vec![
            Arc::new(FakeFactory {
                provider: ProviderType::GitHub,
                caps: Capabilities { supports_pull_requests: true, supports_pipelines: true, ..Default::default() },
            }),
            Arc::new(FakeFactory {
                provider: ProviderType::Linear,
                caps: Capabilities { supports_work_items: true, ..Default::default() },
            }),
        ]))
    }

    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()]);
    }
}