sett 0.4.0

//! Encrypt workflow

use std::{
    collections::{BTreeMap, HashSet},
    io::{self, Write as _},
    path::{Path, PathBuf},
};

use bytes::BytesMut;
use chrono::{DateTime, Utc};
use flate2::{Compression, write::GzEncoder};
use sequoia_openpgp::policy::StandardPolicy;
use tokio::io::AsyncWriteExt;
use tracing::{debug, info, instrument, trace};

use crate::{
    destination::Destination,
    filesystem::{check_space, check_writeable, get_common_path},
    openpgp::crypto::Signer,
    package,
    progress::{ProgressDisplay, ProgressTask},
    task::{Mode, Status},
    zip::ZipArchive,
};

/// Options required by the encrypt workflow.
pub struct EncryptOpts<T, U> {
    /// Input files for encryption.
    pub files: Vec<PathBuf>,
    /// Public OpenPGP keys of recipients.
    pub recipients: Vec<crate::openpgp::cert::Fingerprint>,
    /// Private OpenPGP key for signer (used for signing data).
    pub signer: crate::openpgp::cert::Fingerprint,
    /// OpenPGP certificate store
    pub cert_store: crate::openpgp::certstore::CertStore<'static>,
    /// OpenPGP private key store
    pub key_store: crate::openpgp::keystore::KeyStore,
    /// Password for decrypting the signer's key.
    pub password: U,
    /// Algorithm used for compressing data before encryption.
    pub compression_algorithm: package::CompressionAlgorithm,
    /// Run the workflow or only perform a check.
    pub mode: Mode,
    /// Encryption progress handler.
    pub progress: Option<T>,
    /// Purpose, see [`package::Purpose`] for more details.
    pub purpose: Option<package::Purpose>,
    /// Data transfer ID, see [`package::Metadata::transfer_id`] for more details.
    pub transfer_id: Option<u32>,
    /// Timestamp of the data package.
    pub timestamp: DateTime<Utc>,
    /// Prefix for the data package name.
    pub prefix: Option<String>,
    /// Extra metadata fields, see [`package::Metadata::extra`] for more details.
    pub extra_metadata: BTreeMap<String, String>,
}

impl<T: ProgressDisplay, U> std::fmt::Debug for EncryptOpts<T, U> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("EncryptOpts")
            .field("files", &self.files)
            .field("recipients", &self.recipients)
            .field("signer", &self.signer)
            .field("compression_algorithm", &self.compression_algorithm)
            .field("mode", &self.mode)
            .field("purpose", &self.purpose)
            .field("transfer_id", &self.transfer_id)
            .field("extra_metadata", &self.extra_metadata)
            .finish()
    }
}

/// Creates a data package with encrypted and optionally compressed data along
/// with checksums and metadata.
///
/// The data package can be stored in either the local filesystem or on a
/// remote server.
#[instrument(err(Debug, level=tracing::Level::ERROR))]
pub async fn encrypt<T, F, Fut>(
    mut opts: EncryptOpts<T, F>,
    dest: Destination,
) -> Result<Status, Error>
where
    T: ProgressDisplay + Send + 'static,
    F: Fn(crate::openpgp::crypto::PasswordHint) -> Fut,
    Fut: std::future::Future<Output = super::secret::Secret>,
{
    trace!("Extract signing and encryption certificates");
    let signer_cert = opts.cert_store.get_cert_by_fingerprint(&opts.signer)?;
    let policy = Default::default();
    let valid_signer_cert = signer_cert.validate(&policy)?;
    crate::openpgp::cert::warn_if_cert_expires_soon(&valid_signer_cert.0);
    let signer = Signer::get(&valid_signer_cert.0, &mut opts.key_store, &opts.password).await?;
    let recipients = opts
        .recipients
        .iter()
        .map(|fp| {
            opts.cert_store
                .get_cert_by_fingerprint(fp)
                .map(|cert| cert.0)
        })
        .collect::<Result<Vec<_>, _>>()?;
    let status = match dest {
        Destination::Stdout => encrypt_stdout(opts, signer, recipients).await,
        Destination::Local(local) => encrypt_local(opts, &local, signer, recipients).await,
        Destination::Sftp(sftp) => encrypt_sftp(opts, &sftp, signer, recipients).await,
        Destination::S3(s3) => encrypt_s3(opts, s3, signer, recipients).await,
    }?;
    match &status {
        Status::Checked {
            destination,
            source_size,
        } => {
            debug!(destination, source_size, "Checked encryption task input");
        }
        Status::Completed {
            destination,
            source_size,
            destination_size,
            metadata,
        } => {
            info!(
                destination,
                source_size,
                destination_size,
                metadata = metadata.to_json_or_debug(),
                "Successfully created encrypted data package"
            )
        }
    }
    Ok(status)
}

/// Returns paths where the common prefix (to all paths) is stripped.
fn get_archive_paths(files: &[PathBuf]) -> Result<Vec<PathBuf>, std::io::Error> {
    let parent_folders = files
        .iter()
        .map(|f| {
            f.parent()
                .ok_or_else(|| std::io::Error::other("Unable to find parent folder of {f:?}"))
        })
        .collect::<Result<Vec<_>, _>>()?;
    let root_dir = get_common_path(&parent_folders)?;
    let archive_paths = files
        .iter()
        .map(|f| {
            Ok(Path::new(package::CONTENT_FOLDER).join(
                f.strip_prefix(&root_dir)
                    .map_err(|_| std::io::Error::other("Invalid archive member path: {f:?}"))?,
            ))
        })
        .collect::<Result<_, std::io::Error>>()?;
    Ok(archive_paths)
}

/// Initializes and return an OpenPGP message, which can be used for writing
/// data into it.
fn init_message<'a, W: io::Write + Send + Sync>(
    writer: &'a mut W,
    signer: sequoia_keystore::Key,
    recipients: impl IntoIterator<Item = sequoia_openpgp::serialize::stream::Recipient<'a>>,
) -> Result<sequoia_openpgp::serialize::stream::Message<'a>, crate::openpgp::error::PgpError> {
    use sequoia_openpgp::serialize::stream::{Encryptor, LiteralWriter, Message, Signer};
    LiteralWriter::new(
        Signer::new(
            Encryptor::for_recipients(Message::new(writer), recipients)
                .build()
                .map_err(crate::openpgp::error::PgpError::from)?,
            signer,
        )
        .map_err(crate::openpgp::error::PgpError::from)?
        .build()
        .map_err(crate::openpgp::error::PgpError::from)?,
    )
    .build()
    .map_err(crate::openpgp::error::PgpError::from)
}

/// Adds a file to the inner tar archive containing checksums of the individual
/// data files.
///
/// The checksum file has the following format.
///
/// ```text
/// <checksum1> <file 1 path inside the data package>
/// <checksum2> <file 2 path inside the data package>
/// ...
/// ```
fn add_checksum_file(
    archive: &mut tar::Builder<impl io::Write>,
    content: &[(String, String)],
) -> Result<(), Error> {
    use std::fmt::Write as _;
    let content = content
        .iter()
        .fold(String::new(), |mut output, (checksum, path)| {
            let _ = writeln!(output, "{checksum} {path}");
            output
        })
        .into_bytes();
    let mut header = tar::Header::new_gnu();
    header.set_entry_type(tar::EntryType::file());
    header.set_size(
        content
            .len()
            .try_into()
            .map_err(|_| std::io::Error::other("content size too big"))?,
    );
    header.set_mtime(
        Utc::now()
            .timestamp()
            .try_into()
            .map_err(|e| std::io::Error::other(format!("invalid date: {e}")))?,
    );
    header.set_mode(0o644);
    header.set_cksum();
    archive.append_data(&mut header, package::CHECKSUM_FILE, &content[..])?;
    Ok(())
}

/// Writes files into a tarball.
fn create_tarball(
    writer: &mut impl io::Write,
    file_path: &[(PathBuf, PathBuf)],
    progress: &mut Option<ProgressTask<impl ProgressDisplay>>,
) -> Result<(), Error> {
    enum Message {
        Payload(BytesMut),
        Finalize(std::path::PathBuf),
    }
    impl crate::io::Message for Message {
        fn from_bytes(bytes: BytesMut) -> Self {
            Self::Payload(bytes)
        }
    }

    let (result_tx, mut result_rx) = tokio::sync::mpsc::channel(4);
    let (pool_tx, mut pool_rx) = tokio::sync::mpsc::channel(4);
    for _ in 0..4 {
        pool_tx.blocking_send(BytesMut::with_capacity(1 << 19))?;
    }
    let checksums_len = file_path.len();
    let checksums_handle = std::thread::spawn(move || -> Result<_, Error> {
        use sequoia_openpgp::types::HashAlgorithm::SHA256;

        let mut checksums = Vec::with_capacity(checksums_len);
        let mut hasher = SHA256
            .context()
            .map_err(crate::openpgp::error::PgpError::from)?
            .for_digest();
        while let Some(message) = result_rx.blocking_recv() {
            match message {
                Message::Payload(payload) => {
                    hasher.update(&payload);
                    if pool_tx.blocking_send(payload).is_err() {
                        // Under "normal" conditions this can only happen once, when all data is
                        // read and the receiver end of this channel is closed.
                        trace!("Failed to send chunk back to the pool (channel closed)");
                    }
                }
                Message::Finalize(path) => {
                    let path = crate::filesystem::to_posix_path(&path)
                        .ok_or_else(|| {
                            std::io::Error::other(format!(
                                "Path contains non-UTF-8 characters: {path:?}"
                            ))
                        })?
                        .to_string();
                    let cs = std::mem::replace(
                        &mut hasher,
                        SHA256
                            .context()
                            .map_err(crate::openpgp::error::PgpError::from)?
                            .for_digest(),
                    );
                    checksums.push((
                        crate::utils::to_hex_string(
                            &cs.into_digest()
                                .map_err(crate::openpgp::error::PgpError::from)?,
                        ),
                        path,
                    ));
                }
            }
        }
        Ok(checksums)
    });

    let mut archive = tar::Builder::new(writer);
    for (fs_path, archive_path) in file_path {
        let f = std::fs::File::open(fs_path)?;
        let mut header = tar::Header::new_gnu();
        header.set_metadata(&f.metadata()?);
        header.set_cksum();
        let hash_reader = crate::io::Tee::new(f, &mut pool_rx, &result_tx)?;
        if let Some(p) = progress.as_mut() {
            let progress_reader = p.wrap_reader(hash_reader);
            archive.append_data(&mut header, archive_path, progress_reader)?;
        } else {
            archive.append_data(&mut header, archive_path, hash_reader)?;
        }
        result_tx.blocking_send(Message::Finalize(archive_path.clone()))?;
    }
    drop(result_tx);
    let checksums = checksums_handle
        .join()
        .map_err(|_| PrimaryError::Thread)??;
    add_checksum_file(&mut archive, &checksums)?;
    archive.finish()?;
    Ok(())
}

/// Adds the metadata file to the data package (outer zip file).
fn add_metadata_file<S: sequoia_openpgp::crypto::Signer + Send + Sync>(
    archive: &mut ZipArchive<impl io::Write>,
    metadata: &package::Metadata,
    signer_key: S,
) -> Result<(), Error> {
    let zip_file_opts = Default::default();
    archive.add(package::METADATA_FILE, &zip_file_opts)?;
    let metadata_json = serde_json::to_string(&metadata)
        .map_err(std::io::Error::other)?
        .as_bytes()
        .to_owned();
    archive.write_all(&metadata_json)?;
    archive.add(package::METADATA_SIG_FILE, &zip_file_opts)?;
    archive.write_all(&crate::openpgp::crypto::sign_detached(
        &metadata_json,
        signer_key,
    )?)?;
    Ok(())
}

/// Processes input data files.
///
/// Returns:
///
/// - A vector of pairs (tuples) where the first element is the absolute path
///   to the file on disk, and the second element is its path inside the data
///   package, i.e. a path where the common prefix (for all files) is stripped.
/// - The combined size of all files.
///
/// Processing steps:
///
/// - Walk directories and return individual files.
/// - Normalize paths (make them absolute).
/// - Remove duplicated paths.
/// - Verify that archive paths contain only valid UTF-8 characters
///   (later in the workflow it's necessary to write paths as strings).
/// - Calculate the combined size of the input.
fn process_files<P: AsRef<Path>>(
    files: &[P],
) -> Result<(Vec<(PathBuf, PathBuf)>, u64), std::io::Error> {
    let mut total_input_size = 0u64;
    let mut unique_files = std::collections::HashSet::new();
    for entry in files.iter().flat_map(walkdir::WalkDir::new) {
        let entry = entry?;
        if entry.file_type().is_file() {
            total_input_size += entry.metadata()?.len();
            unique_files.insert(entry.path().canonicalize()?.to_owned());
        }
    }
    let files = Vec::from_iter(unique_files);
    let archive_paths = get_archive_paths(&files)?;
    archive_paths.iter().try_for_each(|p| {
        p.to_str().and(Some(())).ok_or_else(|| {
            std::io::Error::other(format!("Path contains non-UTF-8 characters: {p:?}"))
        })
    })?;
    let file_archive_path = files.into_iter().zip(archive_paths).collect();
    Ok((file_archive_path, total_input_size))
}

/// Spawns a blocking task to handle the encryption and compression pipeline.
fn spawn_encryption_task<T: ProgressDisplay + Send + 'static, F>(
    opts: EncryptOpts<T, F>,
    file_path: Vec<(PathBuf, PathBuf)>,
    total_input_size: u64,
    signer: Signer,
    recipients: Vec<sequoia_openpgp::Cert>,
    source: crate::io::Source,
    sink: tokio::sync::mpsc::Sender<BytesMut>,
) -> tokio::task::JoinHandle<Result<(u64, package::Metadata), Error>> {
    struct Message(BytesMut);

    impl crate::io::Message for Message {
        fn from_bytes(bytes: BytesMut) -> Self {
            Self(bytes)
        }
    }

    tokio::task::spawn_blocking(move || -> Result<_, Error> {
        let policy = StandardPolicy::new();
        let (recipients, recipient_fingerprints) =
            crate::openpgp::crypto::get_recipients(&recipients, &policy)?;
        let writer = crate::io::ChannelWriter::new(source, sink)?;
        let mut zip = ZipArchive::new(writer);
        zip.add(package::DATA_FILE_ENCRYPTED, &Default::default())?;

        let (result_tx, mut result_rx) = tokio::sync::mpsc::channel::<Message>(4);
        let (pool_tx, mut pool_rx) = tokio::sync::mpsc::channel(4);
        for _ in 0..4 {
            pool_tx.blocking_send(BytesMut::with_capacity(1 << 19))?;
        }
        let checksums_handle = std::thread::spawn(move || -> Result<_, Error> {
            use sequoia_openpgp::types::HashAlgorithm::SHA256;

            let mut hasher = SHA256
                .context()
                .map_err(crate::openpgp::error::PgpError::from)?
                .for_digest();
            while let Some(message) = result_rx.blocking_recv() {
                hasher.update(&message.0);
                if pool_tx.blocking_send(message.0).is_err() {
                    // Under "normal" conditions this can only happen once, when all data is
                    // read and the receiver end of this channel is closed.
                    trace!("Failed to send chunk back to the pool (channel closed)");
                }
            }
            Ok(crate::utils::to_hex_string(
                &hasher
                    .into_digest()
                    .map_err(crate::openpgp::error::PgpError::from)?,
            ))
        });

        let mut checksum_writer = crate::io::Tee::new(&mut zip, &mut pool_rx, &result_tx)?;
        let mut encrypted_message =
            init_message(&mut checksum_writer, signer.key.clone(), recipients)?;

        let mut progress = opts.progress.map(|p| p.start(total_input_size));
        crate::io::write_parallel(&mut encrypted_message, |w| {
            match opts.compression_algorithm {
                package::CompressionAlgorithm::Stored => {
                    create_tarball(w, &file_path, &mut progress)
                }
                package::CompressionAlgorithm::Gzip(level) => {
                    let mut enc = GzEncoder::new(
                        w,
                        Compression::new(level.unwrap_or_else(|| Compression::default().level())),
                    );
                    create_tarball(&mut enc, &file_path, &mut progress)
                }
                package::CompressionAlgorithm::Zstandard(level) => {
                    let mut enc = zstd::stream::write::Encoder::new(
                        w,
                        level.unwrap_or(zstd::DEFAULT_COMPRESSION_LEVEL),
                    )?;
                    enc.multithread(2)?;
                    create_tarball(&mut enc, &file_path, &mut progress)?;
                    enc.finish()?;
                    Ok(())
                }
            }
        })?;

        encrypted_message
            .finalize()
            .map_err(crate::openpgp::error::PgpError::from)?;
        checksum_writer.flush_channel()?;
        drop(result_tx);
        let checksum = checksums_handle
            .join()
            .map_err(|_| PrimaryError::Thread)??;

        let metadata = package::Metadata {
            sender: signer.cert_fingerprint.to_hex(),
            recipients: HashSet::<&sequoia_openpgp::Fingerprint>::from_iter(
                &recipient_fingerprints,
            )
            .iter()
            .map(|fp| fp.to_hex())
            .collect(),
            checksum,
            timestamp: opts.timestamp,
            version: package::default_version(),
            checksum_algorithm: Default::default(),
            compression_algorithm: opts.compression_algorithm,
            transfer_id: opts.transfer_id,
            purpose: opts.purpose,
            extra: opts.extra_metadata,
        };
        add_metadata_file(&mut zip, &metadata, signer.key)?;
        let size = zip.finish()?;

        Ok((size, metadata))
    })
}

async fn encrypt_to_writer<T: ProgressDisplay + Send + 'static, F>(
    output: impl tokio::io::AsyncWrite,
    opts: EncryptOpts<T, F>,
    file_path: Vec<(PathBuf, PathBuf)>,
    total_input_size: u64,
    signer: Signer,
    recipients: Vec<sequoia_openpgp::Cert>,
) -> Result<(u64, package::Metadata), Error> {
    let (result_tx, mut result_rx) = tokio::sync::mpsc::channel(3);
    let (buf_pool_tx, buf_pool_rx) = tokio::sync::mpsc::channel(3);
    for _ in 0..3 {
        buf_pool_tx.send(BytesMut::with_capacity(1 << 22)).await?;
    }
    let encrypt_task_handle = spawn_encryption_task(
        opts,
        file_path,
        total_input_size,
        signer,
        recipients,
        crate::io::Source::Channel(buf_pool_rx),
        result_tx,
    );
    tokio::pin!(output);
    while let Some(chunk) = result_rx.recv().await {
        output.write_all(&chunk).await?;
        if buf_pool_tx.send(chunk).await.is_err() {
            // Under "normal" conditions this can only happen once, when the encryption thread
            // is done and the receiver end of this channel is closed.
            trace!("Failed to send chunk back to the pool (channel closed)");
        }
    }
    encrypt_task_handle.await?
}

async fn encrypt_to_blocking_writer<T: ProgressDisplay + Send + 'static, F>(
    output: &mut impl std::io::Write,
    opts: EncryptOpts<T, F>,
    file_path: Vec<(PathBuf, PathBuf)>,
    total_input_size: u64,
    signer: Signer,
    recipients: Vec<sequoia_openpgp::Cert>,
) -> Result<(u64, package::Metadata), Error> {
    let (result_tx, mut result_rx) = tokio::sync::mpsc::channel(3);
    let (buf_pool_tx, buf_pool_rx) = tokio::sync::mpsc::channel(3);
    for _ in 0..3 {
        buf_pool_tx.send(BytesMut::with_capacity(1 << 22)).await?;
    }
    let encrypt_task_handle = spawn_encryption_task(
        opts,
        file_path,
        total_input_size,
        signer,
        recipients,
        crate::io::Source::Channel(buf_pool_rx),
        result_tx,
    );
    while let Some(chunk) = result_rx.recv().await {
        output.write_all(&chunk)?;
        if buf_pool_tx.send(chunk).await.is_err() {
            // Under "normal" conditions this can only happen once, when the encryption thread
            // is done and the receiver end of this channel is closed.
            trace!("Failed to send chunk back to the pool (channel closed)");
        }
    }
    encrypt_task_handle.await?
}

/// Encrypt to a file on disk.
async fn encrypt_local<T: ProgressDisplay + Send + 'static, F>(
    opts: EncryptOpts<T, F>,
    dest: &crate::destination::Local,
    signer: Signer,
    recipients: Vec<sequoia_openpgp::Cert>,
) -> Result<Status, Error> {
    trace!("Verify files to encrypt");
    let (file_path, source_size) = process_files(&opts.files)?;
    let output_path = dest
        .path
        .join(dest.package_name(&opts.timestamp, opts.prefix.as_deref()));
    let destination = output_path.to_string_lossy().into_owned();
    let parent = output_path.parent().ok_or_else(|| {
        std::io::Error::new(
            std::io::ErrorKind::NotFound,
            "Unable to find output directory",
        )
    })?;
    trace!("Verify available disk space");
    check_space(source_size, parent, opts.mode)?;
    trace!("Verify whether destination is writeable");
    check_writeable(parent, opts.mode)?;
    Ok(if let Mode::Check = opts.mode {
        Status::Checked {
            destination,
            source_size,
        }
    } else {
        let output = tokio::fs::File::create(&output_path).await?;
        let mut buffered_output = tokio::io::BufWriter::with_capacity(1 << 22, output);
        let (package_size, metadata) = encrypt_to_writer(
            &mut buffered_output,
            opts,
            file_path,
            source_size,
            signer,
            recipients,
        )
        .await
        .map_err(|e| {
            // If an error occurs while creating the `.zip` file, delete
            // the partial file.
            if let Err(msg) = std::fs::remove_file(&output_path) {
                e.with_cleanup_error(destination.clone(), msg.into())
            } else {
                e
            }
        })?;
        buffered_output.flush().await?;
        Status::Completed {
            destination,
            source_size,
            destination_size: package_size,
            metadata,
        }
    })
}

/// Encrypt to standard output.
async fn encrypt_stdout<T: ProgressDisplay + Send + 'static, F>(
    opts: EncryptOpts<T, F>,
    signer: Signer,
    recipients: Vec<sequoia_openpgp::Cert>,
) -> Result<Status, Error> {
    trace!("Verify files to encrypt");
    let (file_path, source_size) = process_files(&opts.files)?;
    let destination = "stdout".to_string();
    Ok(if let Mode::Check = opts.mode {
        Status::Checked {
            destination,
            source_size,
        }
    } else {
        let (package_size, metadata) = encrypt_to_writer(
            tokio::io::stdout(),
            opts,
            file_path,
            source_size,
            signer,
            recipients,
        )
        .await?;
        Status::Completed {
            destination,
            source_size,
            destination_size: package_size,
            metadata,
        }
    })
}

/// Encrypts and transfers data to an S3 instance.
async fn encrypt_s3<T: ProgressDisplay + Send + 'static, F>(
    opts: EncryptOpts<T, F>,
    dest: crate::remote::s3::Location,
    signer: Signer,
    recipients: Vec<sequoia_openpgp::Cert>,
) -> Result<Status, Error> {
    trace!("Verify files to encrypt");
    let (file_path, source_size) = process_files(&opts.files)?;

    // Name and path of data package when uploaded to the S3 object store.
    let object_name = package::generate_package_name(&opts.timestamp, opts.prefix.as_deref());
    let destination = dest.object_path(&object_name);

    // Dry run mode: retrieve the S3 instance endpoint (URL) and bucket name
    // based on the provided DTR ID, but no encryption, compression or transfer.
    if let Mode::Check = opts.mode {
        return Ok(Status::Checked {
            destination,
            source_size,
        });
    }

    trace!("Encrypt and transfer data to S3 object store: {destination}");
    let (result_tx, result_rx) = tokio::sync::mpsc::channel(3);
    let chunk_size = crate::remote::s3::compute_chunk_size(source_size)
        .map_err(crate::remote::s3::error::UploadError::from)?;
    let encrypt_task_handle = spawn_encryption_task(
        opts,
        file_path,
        source_size,
        signer,
        recipients,
        crate::io::Source::New(chunk_size),
        result_tx,
    );
    dest.put_object(&object_name, result_rx).await?;
    let (package_size, metadata) = encrypt_task_handle.await??;
    Ok(Status::Completed {
        destination,
        source_size,
        destination_size: package_size,
        metadata,
    })
}

/// Encrypts to a file or a remote SFTP server.
async fn encrypt_sftp<T: ProgressDisplay + Send + 'static, F>(
    opts: EncryptOpts<T, F>,
    dest: &crate::destination::Sftp,
    signer: Signer,
    recipients: Vec<sequoia_openpgp::Cert>,
) -> Result<Status, Error> {
    trace!("Verify files to encrypt");
    let (file_path, source_size) = process_files(&opts.files)?;
    let client = dest.client().connect()?;
    let upload_dir = crate::remote::sftp::UploadDir::new(dest.base_path(), &client);
    let dpkg_path = crate::remote::sftp::DpkgPath::new(
        &upload_dir.path,
        package::generate_package_name(&opts.timestamp, opts.prefix.as_deref()),
        &client,
    );
    let destination = client.get_url(&dpkg_path.path);
    if let Mode::Check = opts.mode {
        return Ok(Status::Checked {
            destination,
            source_size,
        });
    }
    upload_dir.create(None)?;
    let mut buffered_output =
        std::io::BufWriter::with_capacity(1 << 22, client.inner.create(&dpkg_path.tmp)?);
    let (package_size, metadata) = encrypt_to_blocking_writer(
        &mut buffered_output,
        opts,
        file_path,
        source_size,
        signer,
        recipients,
    )
    .await
    .map_err(|e| {
        // If an error occurs while creating the `.zip` file, delete
        // the partial file on the SFTP server.
        if let Err(msg) = upload_dir.delete() {
            e.with_cleanup_error(destination.clone(), msg.into())
        } else {
            e
        }
    })?;
    buffered_output.flush()?;
    dpkg_path.finalize()?;
    upload_dir.finalize()?;
    Ok(Status::Completed {
        source_size,
        destination_size: package_size,
        destination,
        metadata,
    })
}

/// Encryption error
#[derive(Debug)]
pub struct Error {
    /// Actual error
    pub primary: PrimaryError,
    /// Additional error potentially happening during cleanup
    pub cleanup: Option<(String, PrimaryError)>,
}

impl Error {
    fn with_cleanup_error(self, path: String, e: PrimaryError) -> Self {
        Self {
            cleanup: Some((path, e)),
            ..self
        }
    }
}

impl std::fmt::Display for Error {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "Encryption failed")?;
        if let Some((destination, cleanup_error)) = &self.cleanup {
            write!(
                f,
                ". Additionally, failed to clean partial file '{destination}', reason: {cleanup_error}"
            )?;
        }
        Ok(())
    }
}

impl core::error::Error for Error {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        use PrimaryError::*;
        match &self.primary {
            IO(source) => Some(source),
            Send(_) => None,
            AsyncTask(source) => Some(source),
            Pgp(source) => Some(source),
            Sftp(source) => Some(source),
            S3(source) => Some(source),
            Thread => None,
        }
    }
}

impl<E> From<E> for Error
where
    PrimaryError: From<E>,
{
    fn from(value: E) -> Self {
        Self {
            primary: value.into(),
            cleanup: None,
        }
    }
}

/// Encryption error
#[derive(Debug)]
pub enum PrimaryError {
    /// I/O related error
    IO(std::io::Error),
    /// Async send error
    Send(String),
    /// PGP error
    Pgp(crate::openpgp::error::PgpError),
    /// Async task join
    AsyncTask(tokio::task::JoinError),
    /// Thread join
    Thread,
    /// sftp error
    Sftp(SftpError),
    /// S3 error
    S3(crate::remote::s3::error::UploadError),
}

impl std::fmt::Display for PrimaryError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::IO(err) => err.fmt(f),
            Self::Send(err) => err.fmt(f),
            Self::Pgp(err) => err.fmt(f),
            Self::Sftp(err) => err.fmt(f),
            Self::S3(err) => err.fmt(f),
            Self::AsyncTask(err) => err.fmt(f),
            Self::Thread => write!(f, "Thread join error"),
        }
    }
}

impl From<std::io::Error> for PrimaryError {
    fn from(value: std::io::Error) -> Self {
        Self::IO(value)
    }
}
impl<T> From<tokio::sync::mpsc::error::SendError<T>> for PrimaryError {
    fn from(value: tokio::sync::mpsc::error::SendError<T>) -> Self {
        Self::Send(format!("{value}"))
    }
}
impl From<crate::openpgp::error::PgpError> for PrimaryError {
    fn from(value: crate::openpgp::error::PgpError) -> Self {
        Self::Pgp(value)
    }
}
impl From<ssh2::Error> for PrimaryError {
    fn from(value: ssh2::Error) -> Self {
        Self::Sftp(SftpError::Ssh(value))
    }
}
impl From<crate::remote::s3::error::UploadError> for PrimaryError {
    fn from(value: crate::remote::s3::error::UploadError) -> Self {
        Self::S3(value)
    }
}
impl From<crate::remote::sftp::error::ConnectionError> for PrimaryError {
    fn from(value: crate::remote::sftp::error::ConnectionError) -> Self {
        Self::Sftp(SftpError::Sftp(value))
    }
}
impl From<crate::remote::sftp::error::DeleteError> for PrimaryError {
    fn from(value: crate::remote::sftp::error::DeleteError) -> Self {
        Self::Sftp(SftpError::Delete(value))
    }
}
impl From<crate::io::error::BufferExchangeError> for PrimaryError {
    fn from(value: crate::io::error::BufferExchangeError) -> Self {
        Self::IO(value.into())
    }
}
impl From<crate::io::error::ChannelClosedError> for PrimaryError {
    fn from(value: crate::io::error::ChannelClosedError) -> Self {
        Self::IO(value.into())
    }
}
impl From<tokio::task::JoinError> for PrimaryError {
    fn from(value: tokio::task::JoinError) -> Self {
        Self::AsyncTask(value)
    }
}

/// sftp encryption error
#[derive(Debug)]
pub enum SftpError {
    /// ssh error
    Ssh(ssh2::Error),
    /// sftp error
    Sftp(crate::remote::sftp::error::ConnectionError),
    /// sftp delete error
    Delete(crate::remote::sftp::error::DeleteError),
}

impl std::fmt::Display for SftpError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Ssh(err) => err.fmt(f),
            Self::Sftp(err) => err.fmt(f),
            Self::Delete(err) => err.fmt(f),
        }
    }
}

impl core::error::Error for SftpError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            Self::Ssh(source) => Some(source),
            Self::Sftp(source) => Some(source),
            Self::Delete(source) => Some(source),
        }
    }
}