pipedash_core/application/services/

provider_service.rs

use std::collections::{
    HashMap,
    HashSet,
};
use std::future::Future;
use std::sync::Arc;
use std::time::Duration;

use chrono::{
    DateTime,
    Utc,
};
use dashmap::{
    DashMap,
    DashSet,
};
use pipedash_plugin_api::{
    Plugin as PluginTrait,
    PluginRegistry,
    RateLimit,
    CONTROLLER_HEALTH_FEATURE,
};
use tokio::sync::{
    Mutex,
    RwLock,
};
use tokio::time::{
    timeout,
    Instant,
};

use super::health_service::{
    HealthHandle,
    HealthInput,
    ProviderHealthInput,
};
use crate::application::fetch_scheduler::{
    FetchScheduler,
    ForcedClaim,
    Generation,
};
use crate::application::refresh_manager::RefreshDispatcher;
use crate::domain::{
    DomainError,
    DomainResult,
    FetchStatus,
    HealthLevel,
    Provider,
    ProviderConfig,
    ProviderHealth,
    ProviderImportResult,
    ProviderSummary,
};
use crate::event::{
    CoreEvent,
    EventBus,
};
use crate::infrastructure::config::sync::ProviderSyncService;
use crate::infrastructure::config::token_ref::TokenReference;
use crate::infrastructure::config::ConfigLoader;
use crate::infrastructure::database::{
    Repository,
    FETCH_STATUS_ERROR,
    FETCH_STATUS_NEVER,
    FETCH_STATUS_SUCCESS,
};
use crate::infrastructure::providers::PluginAdapter;
use crate::infrastructure::HttpClientManager;
use crate::plugins;

const QUOTA_PAUSE_PERCENT: u64 = 10;
const ACQUIRE_ATTEMPTS: usize = 3;

pub(crate) type ProviderFactory =
    Arc<dyn Fn(&ProviderConfig) -> DomainResult<Arc<dyn Provider>> + Send + Sync>;

#[derive(Clone)]
pub struct ProviderLease {
    pub id: i64,
    pub generation: Generation,
    pub provider: Arc<dyn Provider>,
}

struct ProviderSlot {
    generation: Generation,
    provider: Arc<dyn Provider>,
}

impl ProviderSlot {
    fn lease(&self, id: i64) -> ProviderLease {
        ProviderLease {
            id,
            generation: self.generation,
            provider: Arc::clone(&self.provider),
        }
    }
}

struct ParameterFetch<'a> {
    fetches: &'a DashMap<String, Arc<Mutex<()>>>,
    key: &'a str,
    lock: Arc<Mutex<()>>,
}

impl<'a> ParameterFetch<'a> {
    fn new(fetches: &'a DashMap<String, Arc<Mutex<()>>>, key: &'a str) -> Self {
        let lock = Arc::clone(fetches.entry(key.to_string()).or_default().value());

        Self { fetches, key, lock }
    }
}

impl Drop for ParameterFetch<'_> {
    fn drop(&mut self) {
        self.fetches
            .remove_if(self.key, |_, lock| Arc::ptr_eq(lock, &self.lock));
    }
}

pub struct ProviderService {
    repository: Arc<Repository>,
    http_client_manager: Arc<HttpClientManager>,
    providers: RwLock<HashMap<i64, ProviderSlot>>,
    commit_locks: DashMap<i64, Arc<Mutex<()>>>,
    plugin_registry: Arc<PluginRegistry>,
    parameter_fetches: DashMap<String, Arc<Mutex<()>>>,
    event_bus: Arc<dyn EventBus>,
    scheduler: Arc<FetchScheduler>,
    health: HealthHandle,
    dispatcher: RefreshDispatcher,
    provider_factory: ProviderFactory,
    published_in_flight: DashSet<i64>,
}

impl ProviderService {
    pub fn new(
        repository: Arc<Repository>, http_client_manager: Arc<HttpClientManager>,
        event_bus: Arc<dyn EventBus>, scheduler: Arc<FetchScheduler>, health: HealthHandle,
        dispatcher: RefreshDispatcher,
    ) -> Self {
        let plugin_registry = Arc::new(plugins::create_plugin_registry());
        let provider_factory: ProviderFactory = {
            let registry = Arc::clone(&plugin_registry);
            let http = Arc::clone(&http_client_manager);
            Arc::new(move |config| plugin_provider(&registry, &http, config))
        };

        Self {
            repository,
            http_client_manager,
            providers: RwLock::new(HashMap::new()),
            commit_locks: DashMap::new(),
            plugin_registry,
            parameter_fetches: DashMap::new(),
            event_bus,
            scheduler,
            health,
            dispatcher,
            provider_factory,
            published_in_flight: DashSet::new(),
        }
    }

    #[cfg(test)]
    pub(crate) fn with_provider_factory(mut self, factory: ProviderFactory) -> Self {
        self.provider_factory = factory;
        self
    }

    pub fn repository(&self) -> &Arc<Repository> {
        &self.repository
    }

    pub(crate) fn scheduler(&self) -> &Arc<FetchScheduler> {
        &self.scheduler
    }

    pub(crate) fn health(&self) -> &HealthHandle {
        &self.health
    }

    pub(crate) fn dispatcher(&self) -> &RefreshDispatcher {
        &self.dispatcher
    }

    pub fn list_available_plugins(&self) -> Vec<pipedash_plugin_api::PluginMetadata> {
        let mut metadata_list = Vec::new();
        for provider_type in self.plugin_registry.provider_types() {
            if let Some(plugin) = self.plugin_registry.get(&provider_type) {
                metadata_list.push(plugin.metadata().clone());
            }
        }
        metadata_list
    }

    pub fn retains_run_history(&self, provider_type: &str) -> bool {
        self.plugin_registry
            .get(provider_type)
            .is_some_and(|plugin| plugin.metadata().capabilities.retains_run_history)
    }

    pub fn init_plugin(
        &self, provider_type: &str, config: HashMap<String, String>, provider_id: i64,
    ) -> DomainResult<Box<dyn PluginTrait>> {
        init_plugin(
            &self.http_client_manager,
            provider_type,
            config,
            provider_id,
        )
    }

    pub async fn field_options(
        &self, provider_type: &str, field_key: &str, config: HashMap<String, String>,
    ) -> DomainResult<Vec<String>> {
        let (mut plugin, client) = build_plugin(&self.http_client_manager, provider_type, &config)?;
        let init_error = plugin.initialize(0, config.clone(), Some(client)).err();

        match plugin.get_field_options(field_key, &config).await {
            Ok(options) => Ok(options),
            Err(e) => Err(match init_error {
                Some(init) => {
                    DomainError::InvalidConfig(format!("Failed to initialize plugin: {init}"))
                }
                None => PluginAdapter::map_error(e),
            }),
        }
    }

    pub(crate) fn commit_lock(&self, id: i64) -> Arc<Mutex<()>> {
        Arc::clone(self.commit_locks.entry(id).or_default().value())
    }

    async fn current(&self, id: i64) -> Option<ProviderLease> {
        self.providers
            .read()
            .await
            .get(&id)
            .map(|slot| slot.lease(id))
    }

    pub(crate) async fn generation_of(&self, id: i64) -> Option<Generation> {
        self.providers
            .read()
            .await
            .get(&id)
            .map(|slot| slot.generation)
    }

    pub(crate) async fn health_inputs(&self) -> Vec<ProviderHealthInput> {
        let leases: Vec<ProviderLease> = self
            .providers
            .read()
            .await
            .iter()
            .map(|(id, slot)| slot.lease(*id))
            .collect();
        let mut inputs = Vec::with_capacity(leases.len());
        for lease in leases {
            match self.repository.get_provider(lease.id).await {
                Ok(config) => inputs.push(self.health_input(&config, &lease)),
                Err(e) => {
                    tracing::debug!(provider_id = lease.id, error = %e, "Provider has no config row");
                }
            }
        }

        inputs
    }

    async fn install(&self, id: i64, provider: Arc<dyn Provider>, interval: i64) -> ProviderLease {
        let mut providers = self.providers.write().await;
        providers
            .entry(id)
            .or_insert_with(|| {
                let generation = Generation::next();
                self.scheduler
                    .register(id, generation, interval_of(interval));
                ProviderSlot {
                    generation,
                    provider,
                }
            })
            .lease(id)
    }

    async fn replace(&self, id: i64, provider: Arc<dyn Provider>, interval: i64) -> ProviderLease {
        let generation = Generation::next();
        let mut providers = self.providers.write().await;
        providers.insert(
            id,
            ProviderSlot {
                generation,
                provider: Arc::clone(&provider),
            },
        );
        self.scheduler.reset(id, generation, interval_of(interval));

        ProviderLease {
            id,
            generation,
            provider,
        }
    }

    pub async fn acquire(&self, id: i64) -> DomainResult<ProviderLease> {
        if let Some(lease) = self.current(id).await {
            return Ok(lease);
        }

        for _ in 0..ACQUIRE_ATTEMPTS {
            let config = self.repository.get_provider(id).await?;
            let provider = (self.provider_factory)(&config)?;
            let lock = self.commit_lock(id);
            let _guard = lock.lock().await;
            if let Some(lease) = self.current(id).await {
                return Ok(lease);
            }
            let row = self.repository.get_provider(id).await?;
            if row.version == config.version {
                let lease = self.install(id, provider, row.refresh_interval).await;
                self.health.send(HealthInput::ProvidersChanged {
                    changed: vec![self.health_input(&row, &lease)],
                });
                return Ok(lease);
            }
        }

        Err(DomainError::ConcurrentModification(format!(
            "Provider {id} kept changing while it was loaded"
        )))
    }

    pub async fn commit<F, Fut, T>(&self, lease: &ProviderLease, f: F) -> Option<T>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = T>,
    {
        self.commit_at(lease.id, lease.generation, f).await
    }

    async fn commit_at<F, Fut, T>(&self, id: i64, generation: Generation, f: F) -> Option<T>
    where
        F: FnOnce() -> Fut,
        Fut: Future<Output = T>,
    {
        let lock = self.commit_lock(id);
        let _guard = lock.lock().await;
        let current = self
            .providers
            .read()
            .await
            .get(&id)
            .map(|slot| slot.generation);
        if current != Some(generation) {
            return None;
        }

        Some(f().await)
    }

    pub async fn publish_summary(&self, id: i64, generation: Generation) {
        self.commit_at(id, generation, || self.emit_status(id))
            .await;
    }

    pub(crate) async fn emit_status(&self, id: i64) {
        match self.summary_of(id).await {
            Ok(provider) => {
                if provider.health.in_flight {
                    self.published_in_flight.insert(id);
                } else {
                    self.published_in_flight.remove(&id);
                }
                self.event_bus
                    .emit(CoreEvent::ProviderStatusUpdated { provider })
                    .await;
            }
            Err(e) => {
                tracing::warn!(provider_id = id, error = %e, "Failed to build provider summary");
            }
        }
    }

    pub(crate) fn published_in_flight(&self, id: i64) -> bool {
        self.published_in_flight.contains(&id)
    }

    async fn summary_of(&self, id: i64) -> DomainResult<ProviderSummary> {
        let config = self.repository.get_provider(id).await?;

        self.build_summary(id, &config).await
    }

    async fn build_summary(
        &self, id: i64, config: &ProviderConfig,
    ) -> DomainResult<ProviderSummary> {
        let cached_pipelines = self.repository.get_cached_pipelines(Some(id)).await?;
        let (status, last_fetch_error, last_fetch_at) = match self
            .repository
            .get_provider_fetch_status(id)
            .await
        {
            Ok(status) => status,
            Err(e) => {
                tracing::warn!(provider_id = id, error = %e, "Failed to read provider fetch status");
                (FETCH_STATUS_NEVER.to_string(), None, None)
            }
        };
        let last_fetch_status = match status.as_str() {
            FETCH_STATUS_SUCCESS => FetchStatus::Success,
            FETCH_STATUS_ERROR => FetchStatus::Error,
            _ => FetchStatus::Never,
        };
        let now = Instant::now();
        let wall = Utc::now();
        let view = self.scheduler.view(id, now);
        let generation = self
            .providers
            .read()
            .await
            .get(&id)
            .map(|slot| slot.generation);
        let snapshot = self.health.snapshot();
        let state = snapshot
            .providers
            .get(&id)
            .filter(|state| Some(state.generation) == generation);

        Ok(ProviderSummary {
            id,
            name: config.display_name().to_string(),
            provider_type: config.provider_type.clone(),
            icon: self
                .plugin_registry
                .get(&config.provider_type)
                .and_then(|plugin| plugin.metadata().icon.clone()),
            pipeline_count: cached_pipelines.len(),
            last_updated: cached_pipelines.iter().map(|p| p.last_updated).max(),
            refresh_interval: config.refresh_interval,
            configured_repositories: configured_repositories(config),
            last_fetch_status,
            last_fetch_error,
            last_fetch_at: last_fetch_at.and_then(|at| {
                DateTime::parse_from_rfc3339(&at)
                    .ok()
                    .map(|at| at.with_timezone(&Utc))
            }),
            paused_until: self.scheduler.paused_until(id, wall),
            last_fetch_error_kind: view.as_ref().and_then(|view| view.last_failure),
            health: ProviderHealth {
                level: state.map_or(HealthLevel::Operational, |state| state.level),
                signals: state.map(|state| state.signals.clone()).unwrap_or_default(),
                next_attempt_at: view
                    .as_ref()
                    .filter(|view| view.failures > 0)
                    .and_then(|_| {
                        let floor = state
                            .filter(|state| state.outage_floor)
                            .map(|_| self.scheduler.policy().outage_floor);
                        self.scheduler.next_due_in(id, now, floor)
                    })
                    .and_then(|wait| chrono::Duration::from_std(wait).ok())
                    .map(|wait| wall + wait),
                in_flight: view.as_ref().is_some_and(|view| view.in_flight),
                status_page: state.and_then(|state| state.status_page.clone()),
            },
            version: config.version.unwrap_or(1),
        })
    }

    fn health_input(&self, config: &ProviderConfig, lease: &ProviderLease) -> ProviderHealthInput {
        ProviderHealthInput {
            id: lease.id,
            generation: Some(lease.generation),
            display_name: config.display_name().to_string(),
            source: lease.provider.status_source().cloned(),
            has_instance_checks: self.plugin_registry.get(&config.provider_type).is_some_and(
                |plugin| {
                    plugin
                        .metadata()
                        .features
                        .iter()
                        .any(|feature| feature.id == CONTROLLER_HEALTH_FEATURE)
                },
            ),
        }
    }

    fn queue_fetch(&self, lease: &ProviderLease) {
        if let ForcedClaim::Claimed(claim) =
            self.scheduler
                .claim_forced(lease.id, lease.generation, Instant::now())
        {
            self.dispatcher.fetch_now(vec![claim]);
        }
    }

    pub async fn add_provider(&self, config: ProviderConfig) -> DomainResult<i64> {
        let mut plugin_config = config.config.clone();
        plugin_config.insert("token".to_string(), config.token.clone());

        let plugin = self.init_plugin(&config.provider_type, plugin_config, 0)?;

        let (validation_result, permission_status) =
            tokio::join!(plugin.validate_credentials(), plugin.check_permissions());

        validation_result.map_err(|e| {
            DomainError::InvalidConfig(format!("Credential validation failed: {e}"))
        })?;

        let provider_type = plugin.provider_type();
        if provider_type != config.provider_type {
            return Err(DomainError::InvalidConfig(format!(
                "Provider type mismatch: expected '{}', got '{}'",
                config.provider_type, provider_type
            )));
        }

        let permission_status = permission_status.ok();

        let id = self.repository.add_provider(&config).await?;

        if let Some(status) = &permission_status {
            if let Err(e) = self.repository.store_provider_permissions(id, status).await {
                tracing::warn!(provider_id = id, error = %e, "Failed to store permissions");
            }
        }

        let mut config_with_id = config.clone();
        config_with_id.id = Some(id);
        let provider = (self.provider_factory)(&config_with_id)?;

        let lock = self.commit_lock(id);
        let _guard = lock.lock().await;
        let lease = self.install(id, provider, config.refresh_interval).await;
        self.health.send(HealthInput::ProvidersChanged {
            changed: vec![self.health_input(&config_with_id, &lease)],
        });
        let provider = self.summary_of(id).await?;
        self.event_bus
            .emit(CoreEvent::ProviderAdded {
                provider,
                timestamp: Utc::now().timestamp_millis(),
            })
            .await;

        Ok(id)
    }

    pub async fn export_providers_toml(&self) -> DomainResult<String> {
        let providers = self.repository.config_backend().list_providers().await?;

        ProviderSyncService::export_toml(&providers)
    }

    pub async fn import_providers_toml(&self, content: &str) -> DomainResult<ProviderImportResult> {
        let file = ConfigLoader::parse_raw(content).map_err(|e| {
            DomainError::InvalidConfig(format!("Couldn't read the providers file: {e}"))
        })?;
        if file.providers.is_empty() {
            return Err(DomainError::InvalidConfig(
                "The file has no [providers] entries".to_string(),
            ));
        }
        if let Some((key, provider)) = file
            .providers
            .iter()
            .find(|(_, provider)| !self.plugin_registry.is_registered(&provider.provider_type))
        {
            return Err(DomainError::InvalidProviderType(format!(
                "Provider '{key}' has an unknown type '{}'",
                provider.provider_type
            )));
        }

        let existing: HashSet<String> = self
            .repository
            .config_backend()
            .list_providers()
            .await?
            .into_iter()
            .map(|provider| provider.name)
            .collect();
        let mut result = ProviderImportResult::default();
        let mut failure = None;

        for (key, provider) in &file.providers {
            let name = provider.display_name(key).to_string();
            if existing.contains(key) {
                result.skipped.push(name);
                continue;
            }

            let config = ProviderSyncService::provider_config(key, provider, String::new());
            match self.repository.add_provider(&config).await {
                Ok(_) => result.added.push(name),
                Err(e) => {
                    let added_elsewhere = matches!(e, DomainError::InvalidConfig(_))
                        && self
                            .repository
                            .config_backend()
                            .list_providers()
                            .await
                            .is_ok_and(|providers| providers.iter().any(|p| p.name == *key));
                    if !added_elsewhere {
                        failure = Some(e);
                        break;
                    }
                    result.skipped.push(name);
                }
            }
        }

        if !result.added.is_empty() {
            self.event_bus.emit(CoreEvent::ProvidersChanged).await;
        }

        failure.map_or(Ok(result), Err)
    }

    pub async fn get_provider_config(&self, id: i64) -> DomainResult<ProviderConfig> {
        self.repository.get_provider(id).await
    }

    pub async fn list_providers(&self) -> DomainResult<Vec<ProviderSummary>> {
        let configs = self.repository.list_providers().await?;
        let mut summaries = Vec::with_capacity(configs.len());

        for config in configs {
            if let Some(id) = config.id {
                summaries.push(self.build_summary(id, &config).await?);
            }
        }

        Ok(summaries)
    }

    pub async fn update_provider(&self, id: i64, config: ProviderConfig) -> DomainResult<()> {
        let mut plugin_config = config.config.clone();
        plugin_config.insert("token".to_string(), config.token.clone());

        let plugin = self.init_plugin(&config.provider_type, plugin_config, id)?;

        plugin.validate_credentials().await.map_err(|e| {
            DomainError::InvalidConfig(format!("Credential validation failed: {e}"))
        })?;

        let permission_status = plugin.check_permissions().await.ok();

        let mut config_with_id = config.clone();
        config_with_id.id = Some(id);
        let new_provider = (self.provider_factory)(&config_with_id)?;

        let lock = self.commit_lock(id);
        let guard = lock.lock().await;

        let current = self.repository.get_provider(id).await?;
        let success = self
            .repository
            .update_provider_with_version(id, &config, current.version.unwrap_or(1))
            .await?;

        if !success {
            return Err(DomainError::ConcurrentModification(format!(
                "Provider {} was modified by another client",
                id
            )));
        }

        if let Err(e) = self.repository.reset_provider_fetch_status(id).await {
            tracing::warn!(provider_id = id, error = %e, "Failed to reset fetch status");
        }
        let lease = self
            .replace(id, new_provider, config.refresh_interval)
            .await;
        self.health.send(HealthInput::ProvidersChanged {
            changed: vec![self.health_input(&config_with_id, &lease)],
        });
        match self.summary_of(id).await {
            Ok(provider) => {
                self.event_bus
                    .emit(CoreEvent::ProviderUpdated {
                        provider,
                        timestamp: Utc::now().timestamp_millis(),
                    })
                    .await;
            }
            Err(e) => {
                tracing::warn!(provider_id = id, error = %e, "Failed to build updated provider summary");
                self.event_bus.emit(CoreEvent::ProvidersChanged).await;
            }
        }
        drop(guard);

        if let Some(status) = &permission_status {
            if let Err(e) = self.repository.store_provider_permissions(id, status).await {
                tracing::warn!(provider_id = id, error = %e, "Failed to store permissions");
            }
        }

        self.queue_fetch(&lease);

        Ok(())
    }

    pub async fn update_provider_refresh_interval(
        &self, id: i64, refresh_interval: i64,
    ) -> DomainResult<()> {
        let lock = self.commit_lock(id);
        let _guard = lock.lock().await;

        let mut config = self.repository.get_provider(id).await?;
        config.refresh_interval = refresh_interval;
        self.repository.update_provider(id, &config).await?;

        let generation = self
            .providers
            .read()
            .await
            .get(&id)
            .map(|slot| slot.generation);
        if let Some(generation) = generation {
            self.scheduler
                .set_interval(id, generation, interval_of(refresh_interval));
        }

        Ok(())
    }

    pub async fn remove_provider(&self, id: i64) -> DomainResult<()> {
        let lock = self.commit_lock(id);
        let _guard = lock.lock().await;

        let config = self.repository.get_provider(id).await?;
        let provider = self.build_summary(id, &config).await?;

        self.repository.remove_provider(id).await?;
        self.providers.write().await.remove(&id);
        self.scheduler.remove(id);
        self.published_in_flight.remove(&id);
        self.health.send(HealthInput::ProvidersChanged {
            changed: vec![ProviderHealthInput {
                id,
                generation: None,
                display_name: config.display_name().to_string(),
                source: None,
                has_instance_checks: false,
            }],
        });
        self.event_bus
            .emit(CoreEvent::ProviderRemoved {
                provider,
                timestamp: Utc::now().timestamp_millis(),
            })
            .await;

        Ok(())
    }

    pub async fn get_provider(&self, id: i64) -> DomainResult<Arc<dyn Provider>> {
        Ok(self.acquire(id).await?.provider)
    }

    pub async fn check_quota(&self, lease: &ProviderLease) -> Option<DateTime<Utc>> {
        let rate_limit =
            match timeout(Duration::from_secs(10), lease.provider.fetch_rate_limit()).await {
                Ok(Ok(rate_limit)) => rate_limit,
                Ok(Err(e)) => {
                    tracing::debug!(provider_id = lease.id, error = %e, "Rate limit check failed");
                    None
                }
                Err(_) => {
                    tracing::debug!(provider_id = lease.id, "Rate limit check timed out");
                    None
                }
            };
        let paused_until = rate_limit.and_then(|rate_limit| quota_pause(&rate_limit, Utc::now()));

        self.commit(lease, || async {
            let changed = self.scheduler.set_quota_pause(
                lease.id,
                lease.generation,
                paused_until,
                Instant::now(),
                Utc::now(),
            );
            if let (true, Some(until)) = (changed, paused_until) {
                tracing::info!(provider_id = lease.id, %until, "API quota almost used up, pausing provider");
            }
            if changed {
                self.emit_status(lease.id).await;
            }

            paused_until
        })
        .await
        .flatten()
    }

    pub async fn get_provider_instance(&self, id: i64) -> DomainResult<Arc<dyn Provider>> {
        self.get_provider(id).await
    }

    pub async fn load_missing_providers(&self) {
        let ids = match self.repository.list_provider_ids().await {
            Ok(ids) => ids,
            Err(e) => {
                tracing::warn!(error = %e, "Failed to list providers");
                return;
            }
        };
        let loaded: Vec<i64> = self.providers.read().await.keys().copied().collect();
        for id in ids.into_iter().filter(|id| !loaded.contains(id)) {
            if let Err(e) = self.acquire(id).await {
                tracing::debug!(provider_id = id, error = %e, "Provider not loaded yet");
            }
        }
    }

    pub async fn load_all_providers(&self) -> DomainResult<()> {
        let configs = self.repository.list_providers().await?;
        let mut changed = Vec::new();

        for config in configs {
            let Some(id) = config.id else {
                continue;
            };
            if config.token.is_empty() {
                tracing::debug!(
                    provider_id = id,
                    provider_type = %config.provider_type,
                    "Skipping provider load - token not available yet (vault initializing in background)"
                );
                continue;
            }

            match (self.provider_factory)(&config) {
                Ok(provider) => {
                    let lock = self.commit_lock(id);
                    let _guard = lock.lock().await;
                    let lease = self.replace(id, provider, config.refresh_interval).await;
                    changed.push(self.health_input(&config, &lease));
                }
                Err(e) => {
                    tracing::warn!(provider_id = id, error = %e, "Failed to load provider");
                }
            }
        }

        if !changed.is_empty() {
            self.health.send(HealthInput::ProvidersChanged { changed });
        }

        Ok(())
    }

    pub async fn get_workflow_parameters(
        &self, provider_id: i64, workflow_id: &str,
    ) -> DomainResult<Vec<pipedash_plugin_api::WorkflowParameter>> {
        if let Some(cached) = self.cached_parameters(workflow_id).await {
            return Ok(cached);
        }

        let lease = self.acquire(provider_id).await?;
        let fetch = ParameterFetch::new(&self.parameter_fetches, workflow_id);
        let _guard = fetch.lock.lock().await;

        if let Some(cached) = self.cached_parameters(workflow_id).await {
            return Ok(cached);
        }

        let parameters = lease.provider.get_workflow_parameters(workflow_id).await?;

        let cached = self
            .commit(&lease, || {
                self.repository
                    .cache_workflow_parameters(workflow_id, &parameters)
            })
            .await;
        if let Some(Err(e)) = cached {
            tracing::warn!(provider_id, workflow_id, error = %e, "Failed to cache workflow parameters");
        }

        Ok(parameters)
    }

    async fn cached_parameters(
        &self, workflow_id: &str,
    ) -> Option<Vec<pipedash_plugin_api::WorkflowParameter>> {
        match self
            .repository
            .get_cached_workflow_parameters(workflow_id)
            .await
        {
            Ok(cached) => cached,
            Err(e) => {
                tracing::warn!(workflow_id, error = %e, "Failed to read cached workflow parameters");
                None
            }
        }
    }

    pub async fn get_provider_permissions(
        &self, provider_id: i64,
    ) -> DomainResult<Option<pipedash_plugin_api::PermissionStatus>> {
        self.repository.get_provider_permissions(provider_id).await
    }

    pub async fn recheck_provider_permissions(
        &self, provider_id: i64,
    ) -> DomainResult<pipedash_plugin_api::PermissionStatus> {
        let config = self.repository.get_provider(provider_id).await?;
        let mut plugin_config = config.config.clone();
        plugin_config.insert("token".to_string(), config.token.clone());

        let plugin = self.init_plugin(&config.provider_type, plugin_config, provider_id)?;

        let status = plugin
            .check_permissions()
            .await
            .map_err(|e| DomainError::InvalidConfig(format!("Failed to check permissions: {e}")))?;

        self.repository
            .store_provider_permissions(provider_id, &status)
            .await?;

        Ok(status)
    }
}

fn interval_of(seconds: i64) -> Duration {
    Duration::from_secs(u64::try_from(seconds).unwrap_or(0))
}

fn build_plugin(
    http: &HttpClientManager, provider_type: &str, config: &HashMap<String, String>,
) -> DomainResult<(Box<dyn PluginTrait>, Arc<reqwest::Client>)> {
    let plugin = plugins::create_plugin(provider_type).ok_or_else(|| {
        DomainError::InvalidProviderType(format!("Unknown provider type: {}", provider_type))
    })?;
    let client = http.client_for(&plugin.client_options(config))?;

    Ok((plugin, client))
}

fn init_plugin(
    http: &HttpClientManager, provider_type: &str, config: HashMap<String, String>,
    provider_id: i64,
) -> DomainResult<Box<dyn PluginTrait>> {
    let (mut plugin, client) = build_plugin(http, provider_type, &config)?;
    plugin
        .initialize(provider_id, config, Some(client))
        .map_err(|e| DomainError::InvalidConfig(format!("Failed to initialize plugin: {e}")))?;

    Ok(plugin)
}

fn plugin_provider(
    registry: &PluginRegistry, http: &HttpClientManager, config: &ProviderConfig,
) -> DomainResult<Arc<dyn Provider>> {
    if !registry.is_registered(&config.provider_type) {
        return Err(DomainError::InvalidProviderType(format!(
            "Plugin not found for provider type: {}",
            config.provider_type
        )));
    }

    let provider_id = config.id.unwrap_or(0);
    let mut plugin_config = config.config.clone();

    let resolved_token = if config.token.is_empty() {
        String::new()
    } else {
        match TokenReference::parse(&config.token) {
            Ok(TokenReference::EnvVar(var_name)) => std::env::var(&var_name).unwrap_or_else(|_| {
                tracing::warn!(
                    provider_id = provider_id,
                    var_name = %var_name,
                    "Environment variable not found, using empty token"
                );
                String::new()
            }),
            Ok(TokenReference::None) => String::new(),
            Ok(_) => config.token.clone(),
            Err(_) => config.token.clone(),
        }
    };
    plugin_config.insert("token".to_string(), resolved_token);

    let plugin = init_plugin(http, &config.provider_type, plugin_config, provider_id)?;

    Ok(Arc::new(PluginAdapter::new(
        plugin,
        config.provider_type.clone(),
        provider_id,
    )))
}

fn configured_repositories(config: &ProviderConfig) -> Vec<String> {
    config
        .config
        .get("selected_items")
        .map(|items| {
            if matches!(config.provider_type.as_str(), "argocd" | "fluxcd") {
                return Vec::new();
            }

            items
                .split(',')
                .map(|s| s.trim().to_string())
                .filter(|s| !s.is_empty())
                .map(|item| {
                    if config.provider_type == "tekton" && item.contains("__") {
                        item.replace("__", "/")
                    } else if config.provider_type == "jenkins" && !item.contains('/') {
                        format!("(root)/{}", item)
                    } else {
                        item
                    }
                })
                .collect()
        })
        .unwrap_or_default()
}

fn quota_pause(rate_limit: &RateLimit, now: DateTime<Utc>) -> Option<DateTime<Utc>> {
    let low = rate_limit.remaining * 100 < rate_limit.limit * QUOTA_PAUSE_PERCENT;

    (low && rate_limit.reset_at > now).then_some(rate_limit.reset_at)
}

#[cfg(test)]
mod tests {
    use std::io::{
        Read,
        Write,
    };

    use pipedash_plugin_api::PluginError;

    use super::*;
    use crate::application::test_support::{
        pipeline,
        Gate,
        Harness,
        StubProvider,
    };
    use crate::application::FetchResult;
    use crate::domain::PipelineStatus;
    use crate::infrastructure::config::ConfigLoader;
    use crate::test_env::EnvGuard;

    const GITHUB_SECRET: &str = "ghp_exampleSECRET0123456789";
    const GITLAB_SECRET: &str = "glpat-SECRET-VALUE";
    const JENKINS_DEFAULT_SECRET: &str = "hunter2-default";

    fn provider(
        name: &str, provider_type: &str, token: &str, refresh_interval: i64,
        config: &[(&str, &str)],
    ) -> ProviderConfig {
        ProviderConfig {
            id: None,
            name: name.to_string(),
            provider_type: provider_type.to_string(),
            token: token.to_string(),
            config: config
                .iter()
                .map(|(k, v)| (k.to_string(), v.to_string()))
                .collect(),
            refresh_interval,
            version: None,
        }
    }

    #[tokio::test]
    async fn test_export_providers_toml_round_trips_without_tokens() {
        let source = Harness::with_real_plugins().await;
        let repository = source.repository.clone();
        let service = &source.provider_service;

        let originals = [
            provider(
                "github-main",
                "github",
                GITHUB_SECRET,
                45,
                &[("selected_items", "acme/api,acme/web")],
            ),
            provider(
                "My Jenkins",
                "jenkins",
                &format!("${{JENKINS_API_TOKEN:-{JENKINS_DEFAULT_SECRET}}}"),
                120,
                &[
                    ("display_name", "Prod Jenkins"),
                    ("server_url", "http://127.0.0.1:9"),
                    ("username", "ci"),
                    ("selected_items", "(root)/deploy"),
                ],
            ),
            provider("gitlab_main", "gitlab", GITLAB_SECRET, 30, &[]),
        ];
        for original in &originals {
            repository.add_provider(original).await.unwrap();
        }

        let exported = service.export_providers_toml().await.unwrap();

        for secret in [GITHUB_SECRET, GITLAB_SECRET, JENKINS_DEFAULT_SECRET] {
            assert!(!exported.contains(secret), "export leaked {secret}");
        }
        assert!(exported.contains("token = \"${PIPEDASH_TOKEN_GITHUB_MAIN:-}\""));
        assert!(exported.contains("token = \"${JENKINS_API_TOKEN:-}\""));
        assert!(exported.contains("token = \"${PIPEDASH_TOKEN_GITLAB_MAIN:-}\""));

        let _env = EnvGuard::set(&[
            ("PIPEDASH_TOKEN_GITHUB_MAIN", Some("env-github-token")),
            ("JENKINS_API_TOKEN", Some("env-jenkins-token")),
            ("PIPEDASH_TOKEN_GITLAB_MAIN", None),
        ]);
        let parsed = ConfigLoader::parse(&exported).unwrap();

        let mut keys: Vec<_> = parsed.providers.keys().map(String::as_str).collect();
        keys.sort_unstable();
        assert_eq!(keys, ["My Jenkins", "github-main", "gitlab_main"]);
        assert_eq!(parsed.providers["github-main"].token, "env-github-token");
        assert_eq!(parsed.providers["My Jenkins"].token, "env-jenkins-token");
        assert_eq!(parsed.providers["gitlab_main"].token, "");
        assert_eq!(
            parsed.providers["My Jenkins"].name.as_deref(),
            Some("Prod Jenkins")
        );

        let target_harness = Harness::with_real_plugins().await;
        let target = &target_harness.repository;
        crate::infrastructure::config::sync::ProviderSyncService::new()
            .sync_toml_to_db(
                parsed.providers,
                target.config_backend().as_ref(),
                &crate::infrastructure::MemoryTokenStore::new(),
                false,
            )
            .await
            .unwrap();

        let imported = target.config_backend().list_providers().await.unwrap();

        assert_eq!(imported.len(), originals.len());
        for original in &originals {
            let imported = imported.iter().find(|p| p.name == original.name).unwrap();
            assert_eq!(imported.name, original.name);
            assert_eq!(imported.provider_type, original.provider_type);
            assert_eq!(imported.refresh_interval, original.refresh_interval);
            assert_eq!(imported.config, original.config);
        }
    }

    #[tokio::test]
    async fn test_import_providers_toml_adds_missing_providers_without_tokens() {
        let source = Harness::with_real_plugins().await;
        let jenkins = provider(
            "My Jenkins",
            "jenkins",
            "${JENKINS_API_TOKEN}",
            120,
            &[
                ("display_name", "Prod Jenkins"),
                ("server_url", "http://127.0.0.1:9"),
                ("username", "ci"),
                ("selected_items", "(root)/deploy"),
            ],
        );
        for original in [
            provider(
                "github-main",
                "github",
                GITHUB_SECRET,
                45,
                &[("selected_items", "acme/api")],
            ),
            jenkins.clone(),
        ] {
            source.repository.add_provider(&original).await.unwrap();
        }
        let exported = source
            .provider_service
            .export_providers_toml()
            .await
            .unwrap();

        let target = Harness::with_real_plugins().await;
        let kept = provider(
            "github-main",
            "github",
            "target-token",
            300,
            &[("selected_items", "acme/web")],
        );
        let kept_id = target.repository.add_provider(&kept).await.unwrap();

        let _env = EnvGuard::set(&[("JENKINS_API_TOKEN", Some("env-jenkins-token"))]);
        let result = target
            .provider_service
            .import_providers_toml(&exported)
            .await
            .unwrap();

        assert_eq!(
            result,
            ProviderImportResult {
                added: vec!["Prod Jenkins".to_string()],
                skipped: vec!["github-main".to_string()],
            }
        );
        let unchanged = target.repository.get_provider(kept_id).await.unwrap();
        assert_eq!(unchanged.token, kept.token);
        assert_eq!(unchanged.refresh_interval, kept.refresh_interval);
        assert_eq!(unchanged.config, kept.config);

        let imported = target.repository.list_providers().await.unwrap();
        let imported = imported.iter().find(|p| p.name == jenkins.name).unwrap();
        let row = target
            .repository
            .config_backend()
            .get_provider(imported.id.unwrap())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(imported.token, "");
        assert_eq!(row.token, "");
        assert_eq!(imported.provider_type, jenkins.provider_type);
        assert_eq!(imported.refresh_interval, jenkins.refresh_interval);
        assert_eq!(imported.config, jenkins.config);
        assert_eq!(target.bus.count("providers-changed"), 1);
    }

    #[tokio::test]
    async fn test_import_providers_toml_rejects_unknown_type_before_adding() {
        let h = Harness::with_real_plugins().await;
        let content = r#"
[providers.github-main]
type = "github"

[providers.mystery]
type = "travis"
"#;

        let err = h
            .provider_service
            .import_providers_toml(content)
            .await
            .unwrap_err();

        assert!(matches!(err, DomainError::InvalidProviderType(_)), "{err}");
        assert!(h.repository.list_providers().await.unwrap().is_empty());
        assert_eq!(h.bus.count("providers-changed"), 0);
    }

    #[test]
    fn quota_pause_starts_under_ten_percent_until_reset() {
        let now = Utc::now();
        let reset_at = now + chrono::Duration::minutes(20);
        let pause = |remaining, reset_at| {
            quota_pause(
                &RateLimit {
                    limit: 5000,
                    remaining,
                    reset_at,
                },
                now,
            )
        };

        assert_eq!(pause(500, reset_at), None);
        assert_eq!(pause(499, reset_at), Some(reset_at));
        assert_eq!(pause(0, reset_at), Some(reset_at));
        assert_eq!(pause(0, now), None);
        assert_eq!(
            quota_pause(
                &RateLimit {
                    limit: 0,
                    remaining: 0,
                    reset_at,
                },
                now,
            ),
            None
        );
    }

    type StubGate = Option<(Arc<tokio::sync::Notify>, Arc<tokio::sync::Notify>)>;

    async fn stub_github(
        remaining: u64, reset: i64, gate: StubGate,
    ) -> (String, Arc<Mutex<Vec<String>>>) {
        use tokio::io::{
            AsyncReadExt,
            AsyncWriteExt,
        };

        let _ = rustls::crypto::ring::default_provider().install_default();
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();
        let requests: Arc<Mutex<Vec<String>>> = Arc::default();
        let log = requests.clone();
        tokio::spawn(async move {
            loop {
                let (mut stream, _) = listener.accept().await.unwrap();
                let mut request = Vec::new();
                let mut buffer = [0; 4096];
                while !request.windows(4).any(|window| window == b"\r\n\r\n") {
                    let count = stream.read(&mut buffer).await.unwrap();
                    if count == 0 {
                        break;
                    }
                    request.extend_from_slice(&buffer[..count]);
                }
                let request = String::from_utf8(request).unwrap();
                let path = request.split_whitespace().nth(1).unwrap_or_default();
                let path = path.split('?').next().unwrap().to_string();
                let (status, body) = if path == "/api/v3/rate_limit" {
                    if let Some((arrived, release)) = &gate {
                        arrived.notify_one();
                        release.notified().await;
                    }
                    let rate = serde_json::json!({
                        "limit": 5000, "used": 5000 - remaining, "remaining": remaining, "reset": reset
                    });
                    let search = serde_json::json!({"limit": 30, "used": 0, "remaining": 30, "reset": reset});
                    let body = serde_json::json!({
                        "resources": {"core": rate, "search": search},
                        "rate": rate
                    });
                    ("200 OK", body.to_string())
                } else {
                    ("404 Not Found", r#"{"message":"Not Found"}"#.to_string())
                };
                log.lock().await.push(path);
                let response = format!(
                    "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
                    body.len()
                );
                let _ = stream.write_all(response.as_bytes()).await;
            }
        });

        (format!("http://{addr}"), requests)
    }

    #[tokio::test]
    async fn refresh_skips_a_github_provider_with_its_quota_almost_used_up() {
        let h = Harness::with_real_plugins().await;
        h.start().await;
        let reset = Utc::now().timestamp() + 1200;
        let (base_url, requests) = stub_github(0, reset, None).await;
        let id = h
            .repository
            .add_provider(&provider(
                "github",
                "github",
                "ghp_stub",
                30,
                &[("base_url", &base_url), ("selected_items", "acme/app")],
            ))
            .await
            .unwrap();

        let first = h
            .pipeline_service
            .fetch_now(&[id], Duration::from_secs(5))
            .await;
        let again = h
            .pipeline_service
            .fetch_now(&[id], Duration::from_secs(5))
            .await;
        let summaries = h.provider_service.list_providers().await.unwrap();

        assert_eq!(*requests.lock().await, ["/api/v3/rate_limit"]);
        assert_eq!(
            summaries[0].paused_until,
            DateTime::from_timestamp(reset, 0)
        );
        assert!(matches!(summaries[0].last_fetch_status, FetchStatus::Never));
        assert_eq!(first[0].state, crate::domain::RefreshState::Recent);
        assert_eq!(again[0].state, crate::domain::RefreshState::Recent);
        assert_eq!(again[0].kind, Some(crate::domain::FailureKind::RateLimited));
        assert_eq!(*requests.lock().await, ["/api/v3/rate_limit"]);
    }

    #[tokio::test]
    async fn refresh_fetches_a_github_provider_with_quota_left() {
        let h = Harness::with_real_plugins().await;
        h.start().await;
        let (base_url, requests) = stub_github(4000, Utc::now().timestamp() + 1200, None).await;
        let id = h
            .repository
            .add_provider(&provider(
                "github",
                "github",
                "ghp_stub",
                30,
                &[("base_url", &base_url), ("selected_items", "acme/app")],
            ))
            .await
            .unwrap();

        h.pipeline_service
            .fetch_now(&[id], Duration::from_secs(5))
            .await;
        let summaries = h.provider_service.list_providers().await.unwrap();
        let requests = requests.lock().await;

        assert_eq!(requests[0], "/api/v3/rate_limit");
        assert!(requests
            .iter()
            .any(|path| path.starts_with("/api/v3/repos/acme/app")));
        assert_eq!(summaries[0].paused_until, None);
    }

    #[tokio::test]
    async fn quota_answer_for_replaced_credentials_does_not_pause_the_new_ones() {
        let h = Harness::with_real_plugins().await;
        let reset = Utc::now().timestamp() + 1200;
        let arrived = Arc::new(tokio::sync::Notify::new());
        let release = Arc::new(tokio::sync::Notify::new());
        let (base_url, _) = stub_github(0, reset, Some((arrived.clone(), release.clone()))).await;
        let id = h
            .repository
            .add_provider(&provider(
                "github",
                "github",
                "ghp_old",
                30,
                &[("base_url", &base_url), ("selected_items", "acme/app")],
            ))
            .await
            .unwrap();
        let service = Arc::clone(&h.provider_service);
        let lease = service.acquire(id).await.unwrap();

        let check = tokio::spawn({
            let service = Arc::clone(&service);
            let lease = lease.clone();
            async move { service.check_quota(&lease).await }
        });
        arrived.notified().await;
        let replacement =
            (service.provider_factory)(&h.repository.get_provider(id).await.unwrap()).unwrap();
        service.replace(id, replacement, 30).await;
        release.notify_one();

        assert_eq!(check.await.unwrap(), None);
        assert_eq!(h.scheduler.paused_until(id, Utc::now()), None);
    }

    fn updated(h: &Harness, id: i64) -> ProviderConfig {
        let mut config = h.config("renamed", 45);
        config.id = Some(id);
        config
    }

    #[tokio::test]
    async fn an_update_during_a_blocked_fetch_supersedes_the_old_attempt() {
        let h = Harness::new().await;
        h.start().await;
        let old_gate = Gate::new();
        let (id, old) = h
            .add_stub(|id| {
                StubProvider::new(id)
                    .gated(&old_gate)
                    .returning(vec![pipeline(id, "old", PipelineStatus::Success)])
            })
            .await;
        let new = h.stubs.queue(StubProvider::new(id).returning(vec![pipeline(
            id,
            "new",
            PipelineStatus::Failed,
        )]));
        let blocked = h.spawn_fetch(id).await;
        old_gate.arrived().await;
        let before = h.provider_service.acquire(id).await.unwrap().generation;

        h.provider_service
            .update_provider(id, updated(&h, id))
            .await
            .unwrap();
        let updated_at = h
            .bus
            .names()
            .iter()
            .position(|name| *name == "provider-updated")
            .unwrap();
        h.bus
            .wait_for("the fetch queued by the update", |events| {
                events[updated_at..]
                    .iter()
                    .any(|event| matches!(event, CoreEvent::PipelinesUpdated { .. }))
            })
            .await;
        old_gate.open();
        let old_result = blocked.await.unwrap();

        assert!(matches!(old_result, FetchResult::Superseded));
        assert_eq!(h.cached_names(id).await, ["new"]);
        let published: Vec<_> = h.bus.events()[updated_at..]
            .iter()
            .filter_map(|event| match event {
                CoreEvent::PipelinesUpdated { pipelines, .. } => {
                    Some(pipelines.iter().map(|p| p.name.clone()).collect::<Vec<_>>())
                }
                _ => None,
            })
            .collect();
        assert_eq!(published, vec![vec!["new".to_string()]]);
        let view = h.scheduler.view(id, Instant::now()).unwrap();
        assert_eq!(
            (view.failures, view.last_failure, view.in_flight),
            (0, None, false)
        );
        assert_ne!(
            h.provider_service.acquire(id).await.unwrap().generation,
            before
        );
        assert_eq!((old.fetch_count(), new.fetch_count()), (1, 1));
    }

    #[tokio::test]
    async fn an_update_resets_the_fetch_status_and_the_schedule() {
        let h = Harness::new().await;
        let (id, _) = h
            .add_stub(|id| StubProvider::new(id).failing(DomainError::NetworkError("down".into())))
            .await;
        h.fetch_directly(id).await;
        h.health_inputs();
        assert_eq!(h.scheduler.view(id, Instant::now()).unwrap().failures, 1);

        h.provider_service
            .update_provider(id, updated(&h, id))
            .await
            .unwrap();
        let inputs = h.health_inputs();

        let event = h
            .bus
            .events()
            .into_iter()
            .find_map(|event| match event {
                CoreEvent::ProviderUpdated { provider, .. } => Some(provider),
                _ => None,
            })
            .unwrap();
        assert!(matches!(event.last_fetch_status, FetchStatus::Never));
        assert_eq!(event.last_fetch_error, None);
        assert_eq!(event.last_fetch_error_kind, None);
        assert_eq!(event.health.next_attempt_at, None);
        assert_eq!(event.name, "renamed");
        assert_eq!(h.scheduler.view(id, Instant::now()).unwrap().failures, 0);
        let lease = h.provider_service.acquire(id).await.unwrap();
        assert!(matches!(
            inputs.as_slice(),
            [HealthInput::ProvidersChanged { changed }]
                if changed.len() == 1 && changed[0].generation == Some(lease.generation)
        ));
    }

    #[tokio::test]
    async fn an_old_failure_finishing_after_an_update_leaves_the_new_schedule_clean() {
        let h = Harness::new().await;
        h.start().await;
        let old_gate = Gate::new();
        let new_gate = Gate::new();
        let (id, _) = h
            .add_stub(|id| {
                StubProvider::new(id)
                    .gated(&old_gate)
                    .failing(DomainError::NetworkError("down".into()))
            })
            .await;
        h.stubs.queue(StubProvider::new(id).gated(&new_gate));
        let blocked = h.spawn_fetch(id).await;
        old_gate.arrived().await;

        h.provider_service
            .update_provider(id, updated(&h, id))
            .await
            .unwrap();
        new_gate.arrived().await;
        old_gate.open();
        let old_result = blocked.await.unwrap();

        assert!(matches!(old_result, FetchResult::Superseded));
        let view = h.scheduler.view(id, Instant::now()).unwrap();
        assert_eq!(
            (view.failures, view.last_failure, view.in_flight),
            (0, None, true)
        );
        let (status, error, _) = h.repository.get_provider_fetch_status(id).await.unwrap();
        assert_eq!((status.as_str(), error), ("never", None));
        new_gate.open();
        h.bus
            .wait_for("the new fetch", |events| {
                events
                    .iter()
                    .any(|event| matches!(event, CoreEvent::PipelinesUpdated { .. }))
            })
            .await;
        assert_eq!(h.scheduler.view(id, Instant::now()).unwrap().failures, 0);
    }

    #[tokio::test]
    async fn a_removal_during_a_blocked_fetch_leaves_nothing_behind() {
        let h = Harness::new().await;
        let gate = Gate::new();
        let (id, _) = h.add_stub(|id| StubProvider::new(id).gated(&gate)).await;
        let blocked = h.spawn_fetch(id).await;
        gate.arrived().await;
        h.health_inputs();

        h.provider_service.remove_provider(id).await.unwrap();
        let removed_at = h.bus.events().len();
        gate.open();
        let result = blocked.await.unwrap();
        let inputs = h.health_inputs();

        assert!(matches!(result, FetchResult::Superseded));
        assert!(h.repository.get_provider(id).await.is_err());
        assert!(h.repository.get_provider_fetch_status(id).await.is_err());
        assert!(h.cached_names(id).await.is_empty());
        assert!(h.scheduler.view(id, Instant::now()).is_none());
        assert!(h.bus.names()[removed_at..].is_empty());
        assert_eq!(h.bus.names()[removed_at - 1], "provider-removed");
        assert!(matches!(
            inputs.as_slice(),
            [HealthInput::ProvidersChanged { changed }]
                if changed[0].id == id && changed[0].generation.is_none()
        ));
    }

    #[tokio::test]
    async fn concurrent_acquires_share_one_provider() {
        let h = Harness::new().await;
        let id = h.add_provider("shared").await;

        let leases =
            futures::future::join_all((0..8).map(|_| h.provider_service.acquire(id))).await;

        let first = leases[0].as_ref().unwrap();
        for lease in &leases {
            let lease = lease.as_ref().unwrap();
            assert!(Arc::ptr_eq(&lease.provider, &first.provider));
            assert_eq!(lease.generation, first.generation);
        }
    }

    #[tokio::test]
    async fn a_lazy_acquire_racing_a_removal_does_not_bring_the_provider_back() {
        let h = Harness::new().await;
        let id = h.add_provider("racing").await;
        let lock = h.provider_service.commit_lock(id);
        let held = lock.lock().await;
        let removal = tokio::spawn({
            let service = Arc::clone(&h.provider_service);
            async move { service.remove_provider(id).await }
        });
        tokio::time::sleep(Duration::from_millis(100)).await;
        let acquire = tokio::spawn({
            let service = Arc::clone(&h.provider_service);
            async move { service.acquire(id).await }
        });
        tokio::time::sleep(Duration::from_millis(100)).await;

        drop(held);
        removal.await.unwrap().unwrap();
        let acquired = acquire.await.unwrap();

        assert!(matches!(acquired, Err(DomainError::ProviderNotFound(_))));
        assert!(h.provider_service.current(id).await.is_none());
        assert!(h.scheduler.view(id, Instant::now()).is_none());
    }

    #[tokio::test]
    async fn changing_the_refresh_interval_keeps_the_generation_and_reschedules() {
        let h = Harness::new().await;
        h.start().await;
        let (id, _) = h.add_stub(StubProvider::new).await;
        h.pipeline_service
            .fetch_now(&[id], Duration::from_secs(5))
            .await;
        let before = h.provider_service.acquire(id).await.unwrap().generation;
        assert!(
            h.scheduler
                .view(id, Instant::now())
                .unwrap()
                .next_attempt_in
                > Duration::from_secs(20)
        );

        h.provider_service
            .update_provider_refresh_interval(id, 5)
            .await
            .unwrap();

        let after = h.provider_service.acquire(id).await.unwrap().generation;
        assert_eq!(before, after);
        assert!(
            h.scheduler
                .view(id, Instant::now())
                .unwrap()
                .next_attempt_in
                <= Duration::from_secs(5)
        );
        assert_eq!(
            h.repository
                .get_provider(id)
                .await
                .unwrap()
                .refresh_interval,
            5
        );
    }

    #[tokio::test]
    async fn health_hears_about_every_provider_change() {
        let h = Harness::new().await;

        let id = h
            .provider_service
            .add_provider(h.config("argo", 30))
            .await
            .unwrap();
        let added = h.health_inputs();
        let added_generation = h.provider_service.acquire(id).await.unwrap().generation;
        h.provider_service
            .update_provider(id, updated(&h, id))
            .await
            .unwrap();
        let updated_inputs = h.health_inputs();
        h.provider_service.load_all_providers().await.unwrap();
        let reloaded = h.health_inputs();
        let reloaded_generation = h.provider_service.acquire(id).await.unwrap().generation;
        h.provider_service.remove_provider(id).await.unwrap();
        let removed = h.health_inputs();

        let generations = |inputs: &[HealthInput]| match inputs {
            [HealthInput::ProvidersChanged { changed }] => changed
                .iter()
                .map(|input| (input.id, input.generation))
                .collect::<Vec<_>>(),
            other => panic!("unexpected health inputs: {other:?}"),
        };
        assert_eq!(generations(&added), [(id, Some(added_generation))]);
        let updated_generation = generations(&updated_inputs)[0].1;
        assert!(updated_generation.is_some());
        assert_ne!(updated_generation, Some(added_generation));
        assert_eq!(generations(&reloaded), [(id, Some(reloaded_generation))]);
        assert_ne!(Some(reloaded_generation), updated_generation);
        assert_eq!(generations(&removed), [(id, None)]);
        assert_eq!(h.bus.count("provider-added"), 1);
        assert_eq!(h.bus.count("provider-updated"), 1);
        assert_eq!(h.bus.count("provider-removed"), 1);
    }

    #[tokio::test]
    async fn workflow_parameters_are_cached_for_ten_minutes() {
        let h = Harness::new().await;
        let (id, stub) = h.add_stub(StubProvider::new).await;
        let workflow = format!("github__{id}__web");

        let first = h
            .provider_service
            .get_workflow_parameters(id, &workflow)
            .await
            .unwrap();
        h.provider_service
            .get_workflow_parameters(id, &workflow)
            .await
            .unwrap();
        let cached_fetches = stub
            .parameter_fetches
            .load(std::sync::atomic::Ordering::SeqCst);
        sqlx::query(
            "UPDATE workflow_parameters_cache SET cached_at = datetime('now', '-11 minutes')",
        )
        .execute(&h.pool)
        .await
        .unwrap();
        h.provider_service
            .get_workflow_parameters(id, &workflow)
            .await
            .unwrap();

        assert_eq!(first.len(), 1);
        assert_eq!(cached_fetches, 1);
        assert_eq!(
            stub.parameter_fetches
                .load(std::sync::atomic::Ordering::SeqCst),
            2
        );
    }

    #[tokio::test]
    async fn workflow_parameter_failures_keep_their_kind_and_leave_no_lock_behind() {
        let h = Harness::new().await;
        let (id, stub) = h.add_stub(StubProvider::new).await;
        *stub.parameters.lock().unwrap() = Err(DomainError::RateLimited {
            message: "slow".to_string(),
            retry_after: None,
        });

        let failed = h
            .provider_service
            .get_workflow_parameters(id, "github__1__web")
            .await;
        let missing = h
            .provider_service
            .get_workflow_parameters(4242, "github__4242__web")
            .await;

        assert!(matches!(failed, Err(DomainError::RateLimited { .. })));
        assert!(matches!(missing, Err(DomainError::ProviderNotFound(_))));
        assert!(h.provider_service.parameter_fetches.is_empty());
    }

    #[tokio::test]
    async fn fetching_workflow_parameters_does_not_block_the_provider_map() {
        let h = Harness::new().await;
        let gate = Gate::new();
        let (id, _) = h
            .add_stub(|id| StubProvider::new(id).parameter_gated(&gate))
            .await;
        let other = h.add_provider("other").await;
        let fetch = tokio::spawn({
            let service = Arc::clone(&h.provider_service);
            async move { service.get_workflow_parameters(id, "github__1__web").await }
        });
        gate.arrived().await;

        let acquired =
            tokio::time::timeout(Duration::from_secs(2), h.provider_service.acquire(other)).await;
        gate.open();

        assert!(acquired.is_ok());
        fetch.await.unwrap().unwrap();
    }

    #[tokio::test]
    async fn workflow_parameters_fetched_across_a_removal_are_not_cached() {
        let h = Harness::new().await;
        let gate = Gate::new();
        let (id, _) = h
            .add_stub(|id| StubProvider::new(id).parameter_gated(&gate))
            .await;
        let workflow = format!("github__{id}__web");
        let fetch = tokio::spawn({
            let service = Arc::clone(&h.provider_service);
            let workflow = workflow.clone();
            async move { service.get_workflow_parameters(id, &workflow).await }
        });
        gate.arrived().await;

        h.provider_service.remove_provider(id).await.unwrap();
        gate.open();

        assert_eq!(fetch.await.unwrap().unwrap().len(), 1);
        assert!(h
            .repository
            .get_cached_workflow_parameters(&workflow)
            .await
            .unwrap()
            .is_none());
    }

    #[tokio::test]
    async fn field_options_work_when_the_plugin_rejects_the_partial_form() {
        let h = Harness::with_real_plugins().await;
        let kubeconfig = h._temp_dir.path().join("kubeconfig");
        std::fs::write(
            &kubeconfig,
            "apiVersion: v1\nkind: Config\nclusters:\n- name: c\n  cluster: {server: 'https://127.0.0.1:6443'}\nusers:\n- name: u\n  user: {token: t}\ncontexts:\n- name: ctx-b\n  context: {cluster: c, user: u}\n- name: ctx-a\n  context: {cluster: c, user: u}\ncurrent-context: ctx-a\n",
        )
        .unwrap();
        let config = HashMap::from([
            ("kubeconfig".to_string(), kubeconfig.display().to_string()),
            ("namespaces".to_string(), "Not_A_Namespace".to_string()),
        ]);
        assert!(h
            .provider_service
            .init_plugin("fluxcd", config.clone(), 0)
            .is_err());

        let contexts = h
            .provider_service
            .field_options("fluxcd", "context", config)
            .await
            .unwrap();

        assert_eq!(contexts, ["ctx-a", "ctx-b"]);
    }

    #[tokio::test]
    async fn field_options_that_need_a_client_use_the_initialized_plugin() {
        let h = Harness::with_real_plugins().await;
        let ready = HashMap::from([
            ("server_url".to_string(), h.argocd_url.clone()),
            ("token".to_string(), "token".to_string()),
        ]);
        let incomplete = HashMap::from([("token".to_string(), "token".to_string())]);

        let organizations = h
            .provider_service
            .field_options("argocd", "organizations", ready)
            .await;
        let rejected = h
            .provider_service
            .field_options("argocd", "organizations", incomplete)
            .await;

        assert_eq!(organizations.unwrap(), Vec::<String>::new());
        assert!(
            matches!(rejected, Err(DomainError::InvalidConfig(message)) if message.contains("initialize"))
        );
    }

    fn serve_tls(stream: std::net::TcpStream, config: Arc<rustls::ServerConfig>) {
        let Ok(connection) = rustls::ServerConnection::new(config) else {
            return;
        };
        let mut tls = rustls::StreamOwned::new(connection, stream);
        let mut request = Vec::new();
        let mut buffer = [0; 4096];
        while !request.windows(4).any(|window| window == b"\r\n\r\n") {
            match tls.read(&mut buffer) {
                Ok(0) | Err(_) => return,
                Ok(count) => request.extend_from_slice(&buffer[..count]),
            }
        }
        let body = r#"{"items":[]}"#;
        let _ = write!(
            tls,
            "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
            body.len()
        );
        let _ = tls.flush();
        tls.conn.send_close_notify();
        let _ = tls.flush();
    }

    fn self_signed_argocd() -> String {
        let _ = rustls::crypto::ring::default_provider().install_default();
        let cert = rustls::pki_types::CertificateDer::from(
            include_bytes!("../../../tests/fixtures/self_signed_cert.der").to_vec(),
        );
        let key = rustls::pki_types::PrivateKeyDer::Pkcs8(
            include_bytes!("../../../tests/fixtures/self_signed_key.der")
                .to_vec()
                .into(),
        );
        let config = Arc::new(
            rustls::ServerConfig::builder()
                .with_no_client_auth()
                .with_single_cert(vec![cert], key)
                .unwrap(),
        );
        let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
        let addr = listener.local_addr().unwrap();
        std::thread::spawn(move || {
            for stream in listener.incoming().flatten() {
                let config = Arc::clone(&config);
                std::thread::spawn(move || serve_tls(stream, config));
            }
        });

        format!("https://{addr}")
    }

    fn argocd_settings(url: &str, insecure: bool) -> HashMap<String, String> {
        HashMap::from([
            ("server_url".to_string(), url.to_string()),
            ("token".to_string(), "token".to_string()),
            ("insecure".to_string(), insecure.to_string()),
        ])
    }

    #[tokio::test]
    async fn argocd_insecure_tls_is_honored_by_add_validate_and_preview() {
        let h = Harness::with_real_plugins().await;
        let service = &h.provider_service;
        let url = self_signed_argocd();
        let page = Some(pipedash_plugin_api::PaginationParams {
            page: 1,
            page_size: 10,
        });
        let add = |insecure| {
            let mut config = h.config("argo", 30);
            config.config = argocd_settings(&url, insecure);
            config.config.remove("token");
            service.add_provider(config)
        };

        let id = add(true).await.unwrap();
        let insecure = service
            .init_plugin("argocd", argocd_settings(&url, true), 0)
            .unwrap();
        let secure = service
            .init_plugin("argocd", argocd_settings(&url, false), 0)
            .unwrap();
        let rejected = add(false).await.unwrap_err();

        assert!(id > 0);
        assert!(insecure.validate_credentials().await.unwrap());
        assert!(insecure
            .fetch_available_pipelines_filtered(None, None, page.clone())
            .await
            .unwrap()
            .items
            .is_empty());
        assert!(matches!(
            secure.validate_credentials().await,
            Err(PluginError::Tls(_))
        ));
        assert!(matches!(
            secure
                .fetch_available_pipelines_filtered(None, None, page)
                .await,
            Err(PluginError::Tls(_))
        ));
        assert!(matches!(rejected, DomainError::InvalidConfig(message) if message.contains("TLS")));
    }
}