domani/src/domain/manager.rs

use crate::domain::{self, acme, checker, config};
use crate::error::unexpected::{self, Mappable};
use crate::origin;

use std::{future, pin, sync};
use tokio_util::sync::CancellationToken;

#[derive(thiserror::Error, Debug)]
pub enum GetConfigError {
    #[error("not found")]
    NotFound,

    #[error(transparent)]
    Unexpected(#[from] unexpected::Error),
}

impl From<config::GetError> for GetConfigError {
    fn from(e: config::GetError) -> GetConfigError {
        match e {
            config::GetError::NotFound => GetConfigError::NotFound,
            config::GetError::Unexpected(e) => GetConfigError::Unexpected(e),
        }
    }
}

#[derive(thiserror::Error, Debug)]
pub enum GetOriginError {
    #[error("not found")]
    NotFound,

    #[error(transparent)]
    Unexpected(#[from] unexpected::Error),
}

impl From<config::GetError> for GetOriginError {
    fn from(e: config::GetError) -> GetOriginError {
        match e {
            config::GetError::NotFound => GetOriginError::NotFound,
            config::GetError::Unexpected(e) => GetOriginError::Unexpected(e),
        }
    }
}

#[derive(thiserror::Error, Debug)]
pub enum SyncError {
    #[error("not found")]
    NotFound,

    #[error("already in progress")]
    AlreadyInProgress,

    #[error(transparent)]
    Unexpected(#[from] unexpected::Error),
}

impl From<config::GetError> for SyncError {
    fn from(e: config::GetError) -> SyncError {
        match e {
            config::GetError::NotFound => SyncError::NotFound,
            config::GetError::Unexpected(e) => SyncError::Unexpected(e),
        }
    }
}

#[derive(thiserror::Error, Debug)]
pub enum SyncWithConfigError {
    #[error("invalid url")]
    InvalidURL,

    #[error("invalid branch name")]
    InvalidBranchName,

    #[error("already in progress")]
    AlreadyInProgress,

    #[error("target A/AAAA not set")]
    TargetANotSet,

    #[error("challenge token not set")]
    ChallengeTokenNotSet,

    #[error(transparent)]
    Unexpected(#[from] unexpected::Error),
}

impl From<origin::store::SyncError> for SyncWithConfigError {
    fn from(e: origin::store::SyncError) -> SyncWithConfigError {
        match e {
            origin::store::SyncError::InvalidURL => SyncWithConfigError::InvalidURL,
            origin::store::SyncError::InvalidBranchName => SyncWithConfigError::InvalidBranchName,
            origin::store::SyncError::AlreadyInProgress => SyncWithConfigError::AlreadyInProgress,
            origin::store::SyncError::Unexpected(e) => SyncWithConfigError::Unexpected(e),
        }
    }
}

impl From<checker::CheckDomainError> for SyncWithConfigError {
    fn from(e: checker::CheckDomainError) -> SyncWithConfigError {
        match e {
            checker::CheckDomainError::TargetANotSet => SyncWithConfigError::TargetANotSet,
            checker::CheckDomainError::ChallengeTokenNotSet => {
                SyncWithConfigError::ChallengeTokenNotSet
            }
            checker::CheckDomainError::Unexpected(e) => SyncWithConfigError::Unexpected(e),
        }
    }
}

impl From<config::SetError> for SyncWithConfigError {
    fn from(e: config::SetError) -> SyncWithConfigError {
        match e {
            config::SetError::Unexpected(e) => SyncWithConfigError::Unexpected(e),
        }
    }
}

pub type GetAcmeHttp01ChallengeKeyError = acme::manager::GetHttp01ChallengeKeyError;

#[mockall::automock]
pub trait Manager: Sync + Send {
    fn get_config(&self, domain: &domain::Name) -> Result<config::Config, GetConfigError>;

    fn get_origin(
        &self,
        domain: &domain::Name,
    ) -> Result<sync::Arc<dyn origin::Origin>, GetOriginError>;

    fn sync_with_config<'mgr>(
        &'mgr self,
        domain: domain::Name,
        config: config::Config,
    ) -> pin::Pin<Box<dyn future::Future<Output = Result<(), SyncWithConfigError>> + Send + 'mgr>>;

    fn get_acme_http01_challenge_key(
        &self,
        token: &str,
    ) -> Result<String, GetAcmeHttp01ChallengeKeyError>;

    fn all_domains(&self) -> Result<Vec<domain::Name>, unexpected::Error>;
}

pub struct ManagerImpl {
    origin_store: sync::Arc<dyn origin::store::Store>,
    domain_config_store: sync::Arc<dyn config::Store>,
    domain_checker: checker::DNSChecker,
    acme_manager: Option<sync::Arc<dyn acme::manager::Manager>>,

    canceller: CancellationToken,
    origin_sync_handler: tokio::task::JoinHandle<()>,
}

fn sync_origins(origin_store: &dyn origin::store::Store) {
    match origin_store.all_descrs() {
        Ok(iter) => iter.into_iter(),
        Err(err) => {
            log::error!("Error fetching origin descriptors: {err}");
            return;
        }
    }
    .for_each(|descr| {
        if let Err(err) = origin_store.sync(descr.clone(), origin::store::Limits {}) {
            log::error!("Failed to sync store for {:?}: {err}", descr);
            return;
        }
    });
}

pub fn new(
    origin_store: sync::Arc<dyn origin::store::Store>,
    domain_config_store: sync::Arc<dyn config::Store>,
    domain_checker: checker::DNSChecker,
    acme_manager: Option<sync::Arc<dyn acme::manager::Manager>>,
) -> ManagerImpl {
    let canceller = CancellationToken::new();

    let origin_sync_handler = {
        let origin_store = origin_store.clone();
        let canceller = canceller.clone();
        tokio::spawn(async move {
            let mut interval = tokio::time::interval(tokio::time::Duration::from_secs(20 * 60));

            loop {
                tokio::select! {
                _ = interval.tick() => sync_origins(origin_store.as_ref()),
                _ = canceller.cancelled() => return,
                }
            }
        })
    };

    ManagerImpl {
        origin_store,
        domain_config_store,
        domain_checker,
        acme_manager,
        canceller,
        origin_sync_handler,
    }
}

impl ManagerImpl {
    pub fn stop(self) -> tokio::task::JoinHandle<()> {
        self.canceller.cancel();
        self.origin_sync_handler
    }
}

impl Manager for ManagerImpl {
    fn get_config(&self, domain: &domain::Name) -> Result<config::Config, GetConfigError> {
        Ok(self.domain_config_store.get(domain)?)
    }

    fn get_origin(
        &self,
        domain: &domain::Name,
    ) -> Result<sync::Arc<dyn origin::Origin>, GetOriginError> {
        let config = self.domain_config_store.get(domain)?;
        let origin = self
            .origin_store
            .get(config.origin_descr)
            // if there's a config there should be an origin, any error here is unexpected
            .or_unexpected()?;
        Ok(origin)
    }

    fn sync_with_config<'mgr>(
        &'mgr self,
        domain: domain::Name,
        config: config::Config,
    ) -> pin::Pin<Box<dyn future::Future<Output = Result<(), SyncWithConfigError>> + Send + 'mgr>>
    {
        Box::pin(async move {
            let config_hash = config
                .hash()
                .or_unexpected_while("calculating config hash")?;

            self.domain_checker
                .check_domain(&domain, &config_hash)
                .await?;

            self.origin_store
                .sync(config.origin_descr.clone(), origin::store::Limits {})?;

            self.domain_config_store.set(&domain, &config)?;

            if let Some(ref acme_manager) = self.acme_manager {
                acme_manager.sync_domain(domain.clone()).await?;
            }

            Ok(())
        })
    }

    fn get_acme_http01_challenge_key(
        &self,
        token: &str,
    ) -> Result<String, GetAcmeHttp01ChallengeKeyError> {
        if let Some(ref acme_manager) = self.acme_manager {
            return acme_manager.get_http01_challenge_key(token);
        }

        Err(GetAcmeHttp01ChallengeKeyError::NotFound)
    }

    fn all_domains(&self) -> Result<Vec<domain::Name>, unexpected::Error> {
        self.domain_config_store.all_domains()
    }
}