detritus/

shipper.rs

use std::{
    fs, io,
    path::{Path, PathBuf},
    time::{Duration, SystemTime},
};

use detritus_protocol::{
    CrashAttachment, CrashEnvelope, CrashMetadata,
    multipart::{EnvelopeEncodings, PartEncoding},
};
use reqwest::header::{AUTHORIZATION, CONTENT_TYPE, HeaderValue};
use secrecy::{ExposeSecret, SecretString};
use url::Url;

use crate::{
    compression::{DEFAULT_COMPRESSION_LEVEL, compress, should_compress_content_type},
    install_default_crypto_provider,
    panic_hook::StoredUploadConfig,
    spool::SpoolLock,
};

#[cfg(test)]
#[path = "shipper/tests.rs"]
mod regression_tests;

/// Default number of days to keep successfully sent crash entries.
pub const DEFAULT_SENT_RETENTION_DAYS: u64 = 90;

/// Reusable crash uploader with pooled HTTP connections and an in-memory token.
///
/// Convenience shipping functions create one uploader per scan. Keep this value
/// to reuse its pool across scans, or supply a Reqwest client for private CAs,
/// mutual TLS, proxy settings, and custom timeouts.
#[derive(Debug, Clone)]
pub struct CrashShipper {
    client: reqwest::Client,
    token: SecretString,
    config: ShipConfig,
}

impl CrashShipper {
    /// Creates an uploader with a 10-second connect timeout, 30-second read
    /// timeout, and 60-second total request timeout.
    ///
    /// The offline spool owns retries; HTTP-level retries are disabled to avoid
    /// recording a crash index more than once during one upload attempt.
    ///
    /// # Errors
    ///
    /// Returns [`ShipError::Http`] if the HTTP client cannot be initialized.
    pub fn new(token: SecretString) -> Result<Self, ShipError> {
        install_default_crypto_provider();
        // https://docs.rs/reqwest/0.13.5/reqwest/struct.ClientBuilder.html
        let client = reqwest::Client::builder()
            .connect_timeout(Duration::from_secs(10))
            .read_timeout(Duration::from_secs(30))
            .timeout(Duration::from_secs(60))
            .retry(reqwest::retry::never())
            .build()?;
        Ok(Self::with_client(client, token))
    }

    /// Uses a caller-configured HTTP client. Its timeout and retry policy apply.
    #[must_use]
    pub fn with_client(client: reqwest::Client, token: SecretString) -> Self {
        Self {
            client,
            token,
            config: ShipConfig::default(),
        }
    }

    /// Sets dump and attachment compression options.
    #[must_use]
    pub fn with_config(mut self, config: ShipConfig) -> Self {
        self.config = config;
        self
    }

    /// Ships pending entries to an explicit endpoint using this client's connection pool.
    ///
    /// # Errors
    ///
    /// See [`ship_pending_crashes_with_config`]. Failed entries remain pending.
    pub async fn ship_pending(
        &self,
        spool_dir: impl AsRef<Path>,
        endpoint: Url,
    ) -> Result<usize, ShipError> {
        ship_pending_with_client(
            spool_dir.as_ref(),
            &endpoint,
            &self.token,
            &self.config,
            &self.client,
        )
        .await
    }

    /// Ships pending entries using their stored endpoints and retention policies.
    ///
    /// # Errors
    ///
    /// See [`ship_pending_crashes_using_stored_config_with_config`].
    pub async fn ship_using_stored_config(
        &self,
        spool_dir: impl AsRef<Path>,
    ) -> Result<usize, ShipError> {
        ship_stored_with_client(spool_dir.as_ref(), &self.token, &self.config, &self.client).await
    }
}

/// Configuration knobs for crash-dump upload compression.
///
/// Created with [`ShipConfig::default`] and customised with the builder
/// methods.  The defaults compress dumps at zstd level 19 and also
/// compress text-ish attachments.
#[derive(Debug, Clone)]
pub struct ShipConfig {
    /// Zstd level used for the `dump` part. Valid range is `1..=22`.
    /// Defaults to [`DEFAULT_COMPRESSION_LEVEL`] (19).
    pub dump_compression_level: i32,
    /// When `true`, text-ish attachments are compressed with zstd before
    /// upload. Defaults to `true`.
    pub attachment_compression: bool,
}

impl Default for ShipConfig {
    fn default() -> Self {
        Self {
            dump_compression_level: DEFAULT_COMPRESSION_LEVEL,
            attachment_compression: true,
        }
    }
}

impl ShipConfig {
    /// Sets the zstd compression level for the `dump` part.
    #[must_use]
    pub fn with_dump_compression_level(mut self, level: i32) -> Self {
        self.dump_compression_level = level;
        self
    }

    /// Enables or disables zstd compression for text-ish attachment parts.
    #[must_use]
    pub fn with_attachment_compression(mut self, enabled: bool) -> Self {
        self.attachment_compression = enabled;
        self
    }
}

/// Errors returned while shipping pending crash reports.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ShipError {
    /// Filesystem operation failed.
    #[error("crash spool I/O error: {0}")]
    Io(#[from] io::Error),
    /// JSON metadata failed to parse or encode.
    #[error("crash spool JSON error: {0}")]
    Json(#[from] serde_json::Error),
    /// HTTP client failed.
    #[error("crash upload HTTP error: {0}")]
    Http(#[from] reqwest::Error),
    /// Bearer token contains bytes that are not valid in an HTTP header.
    #[error("crash upload bearer token is not a valid HTTP header value")]
    InvalidToken,
    /// Server rejected the upload.
    #[error("crash upload failed with status {0}")]
    Status(reqwest::StatusCode),
    /// Multipart encoding failed.
    #[error("crash multipart encoding failed: {0}")]
    Protocol(#[from] detritus_protocol::ProtocolError),
    /// A spool entry has no usable stored upload config: `sdk-config.json` is
    /// absent when required or its stored endpoint is not a valid URL.
    #[error("entry {} has no usable stored upload config", .0.display())]
    MissingStoredConfig(PathBuf),
}

/// Ships pending crash artifacts and moves successful entries to `sent/`.
///
/// A filesystem lock on `spool_dir/.lock` prevents two processes from scanning
/// the same pending directory concurrently. One spool directory per process is
/// still the recommended default.
///
/// Dump bytes and text-ish attachments are compressed with zstd
/// (level [`DEFAULT_COMPRESSION_LEVEL`]) **before** the SHA-256 is computed,
/// so content-addressed dedup is based on the compressed bytes.
///
/// # Errors
///
/// See [`ship_pending_crashes_with_config`].
pub async fn ship_pending_crashes(
    spool_dir: impl AsRef<Path>,
    endpoint: Url,
    token: SecretString,
) -> Result<usize, ShipError> {
    ship_pending_crashes_with_config(spool_dir, endpoint, token, ShipConfig::default()).await
}

/// Like [`ship_pending_crashes`] but with explicit compression settings.
///
/// # Errors
///
/// Returns [`ShipError::Io`] on spool filesystem errors, [`ShipError::Json`] if
/// stored metadata cannot be parsed, [`ShipError::Protocol`] if multipart
/// encoding fails, [`ShipError::InvalidToken`] for malformed bearer headers,
/// [`ShipError::Http`] if the upload request fails, or
/// [`ShipError::Status`] if the server rejects an upload.
pub async fn ship_pending_crashes_with_config(
    spool_dir: impl AsRef<Path>,
    endpoint: Url,
    token: SecretString,
    config: ShipConfig,
) -> Result<usize, ShipError> {
    CrashShipper::new(token)?
        .with_config(config)
        .ship_pending(spool_dir, endpoint)
        .await
}

async fn ship_pending_with_client(
    spool_dir: &Path,
    endpoint: &Url,
    token: &SecretString,
    config: &ShipConfig,
    client: &reqwest::Client,
) -> Result<usize, ShipError> {
    let _lock = SpoolLock::acquire(spool_dir)?;
    let pending = spool_dir.join("pending");
    let sent = spool_dir.join("sent");
    fs::create_dir_all(&pending)?;
    fs::create_dir_all(&sent)?;
    cleanup_sent(&sent, DEFAULT_SENT_RETENTION_DAYS)?;

    let mut shipped = 0;
    for entry in pending_entries(&pending)? {
        ship_one(&entry, &sent, endpoint, token, config, client).await?;
        shipped += 1;
    }
    Ok(shipped)
}

/// Ships pending crash artifacts using each entry's stored endpoint.
///
/// The caller supplies the bearer token. The token is never read from the spool;
/// each pending entry only contributes its stored endpoint and sent-entry
/// retention from `sdk-config.json`.
///
/// A filesystem lock on `spool_dir/.lock` prevents two processes from scanning
/// the same pending directory concurrently. The lock is held for the whole
/// scan-and-ship run.
///
/// # Errors
///
/// Returns [`ShipError::Io`] on spool filesystem errors, [`ShipError::Json`] if
/// stored metadata or stored upload config cannot be parsed,
/// [`ShipError::MissingStoredConfig`] if a pending entry lacks a usable
/// `sdk-config.json`, [`ShipError::Protocol`] if multipart encoding fails,
/// [`ShipError::InvalidToken`] for malformed bearer headers,
/// [`ShipError::Http`] if an upload request fails, or [`ShipError::Status`] if
/// the server rejects an upload. The first error aborts the run.
pub async fn ship_pending_crashes_using_stored_config(
    spool_dir: impl AsRef<Path>,
    token: SecretString,
) -> Result<usize, ShipError> {
    ship_pending_crashes_using_stored_config_with_config(spool_dir, token, ShipConfig::default())
        .await
}

/// Like [`ship_pending_crashes_using_stored_config`] but with explicit
/// compression settings.
///
/// Each pending entry is posted to the endpoint recovered from that entry's
/// `sdk-config.json`. Successfully shipped entries are moved to `sent/`, then
/// `sent/` is cleaned using the maximum stored retention across the entries
/// shipped by this run. If no entry was shipped, cleanup uses
/// [`DEFAULT_SENT_RETENTION_DAYS`].
///
/// # Errors
///
/// Returns [`ShipError::Io`] on spool filesystem errors, [`ShipError::Json`] if
/// stored metadata or stored upload config cannot be parsed,
/// [`ShipError::MissingStoredConfig`] if a pending entry lacks a usable
/// `sdk-config.json`, [`ShipError::Protocol`] if multipart encoding fails,
/// [`ShipError::Http`] if an upload request fails, or [`ShipError::Status`] if
/// the server rejects an upload. The first error aborts the run.
pub async fn ship_pending_crashes_using_stored_config_with_config(
    spool_dir: impl AsRef<Path>,
    token: SecretString,
    config: ShipConfig,
) -> Result<usize, ShipError> {
    CrashShipper::new(token)?
        .with_config(config)
        .ship_using_stored_config(spool_dir)
        .await
}

async fn ship_stored_with_client(
    spool_dir: &Path,
    token: &SecretString,
    config: &ShipConfig,
    client: &reqwest::Client,
) -> Result<usize, ShipError> {
    let _lock = SpoolLock::acquire(spool_dir)?;
    let pending = spool_dir.join("pending");
    let sent = spool_dir.join("sent");
    fs::create_dir_all(&pending)?;
    fs::create_dir_all(&sent)?;

    let mut shipped = 0;
    let mut max_retention_days = None;
    for entry in pending_entries(&pending)? {
        let (endpoint, retention_days) = resolve_stored_endpoint(&entry)?;
        ship_one(&entry, &sent, &endpoint, token, config, client).await?;
        record_retention(&mut max_retention_days, retention_days);
        shipped += 1;
    }
    cleanup_sent(&sent, retention_or_default(max_retention_days))?;
    Ok(shipped)
}

fn pending_entries(pending: &Path) -> io::Result<Vec<PathBuf>> {
    let mut entries = match fs::read_dir(pending) {
        Ok(entries) => entries
            .filter_map(Result::ok)
            .map(|entry| entry.path())
            .filter(|path| path.is_dir())
            .collect::<Vec<_>>(),
        Err(error) if error.kind() == io::ErrorKind::NotFound => Vec::new(),
        Err(error) => return Err(error),
    };
    entries.sort();
    Ok(entries)
}

async fn ship_one(
    entry: &Path,
    sent: &Path,
    endpoint: &Url,
    token: &SecretString,
    config: &ShipConfig,
    client: &reqwest::Client,
) -> Result<(), ShipError> {
    let envelope = read_envelope(entry)?;
    post_envelope(endpoint, token, &envelope, config, client).await?;
    let destination = sent.join(
        entry
            .file_name()
            .ok_or_else(|| io::Error::other("pending entry has no file name"))?,
    );
    if destination.exists() {
        fs::remove_dir_all(&destination)?;
    }
    fs::rename(entry, destination)?;
    Ok(())
}

fn read_envelope(entry: &Path) -> Result<CrashEnvelope, ShipError> {
    let metadata: CrashMetadata = serde_json::from_slice(&fs::read(entry.join("metadata.json"))?)?;
    let dump = fs::read(entry.join("dump.bin"))?;
    let attachments = metadata
        .attachments
        .iter()
        .filter_map(|manifest| {
            let filename = manifest.filename.as_ref()?;
            let path = entry.join(filename);
            Some((manifest, path))
        })
        .map(|(manifest, path)| {
            Ok(CrashAttachment {
                key: manifest.key.clone(),
                content_type: manifest.content_type.clone(),
                bytes: fs::read(path)?,
            })
        })
        .collect::<io::Result<Vec<_>>>()?;
    Ok(CrashEnvelope {
        metadata,
        dump,
        attachments,
    })
}

/// Applies zstd compression to the envelope's dump (and optionally its
/// attachments), then serialises to multipart and POSTs it.
///
/// Compressed bytes replace the originals in-place inside a locally-owned
/// clone so the original `CrashEnvelope` is not mutated.
async fn post_envelope(
    endpoint: &Url,
    token: &SecretString,
    envelope: &CrashEnvelope,
    config: &ShipConfig,
    client: &reqwest::Client,
) -> Result<(), ShipError> {
    install_default_crypto_provider();

    // Build a locally-owned envelope with compressed bytes plus the matching
    // EnvelopeEncodings that records which parts are zstd-encoded.
    let mut compressed_envelope = envelope.clone();
    let mut encodings = EnvelopeEncodings {
        dump: PartEncoding::default(),
        attachments: Vec::with_capacity(envelope.attachments.len()),
    };

    // Compress the dump unconditionally.
    let compressed_dump = compress(&envelope.dump, config.dump_compression_level)?;
    compressed_envelope.dump = compressed_dump;
    encodings.dump = PartEncoding {
        content_encoding: Some("zstd".to_owned()),
    };

    // Optionally compress text-ish attachments.
    for attachment in &envelope.attachments {
        let should =
            config.attachment_compression && should_compress_content_type(&attachment.content_type);
        if should {
            let compressed = compress(&attachment.bytes, config.dump_compression_level)?;
            // Find the matching attachment in our local clone and replace bytes.
            if let Some(local) = compressed_envelope
                .attachments
                .iter_mut()
                .find(|a| a.key == attachment.key)
            {
                local.bytes = compressed;
            }
            encodings.attachments.push(PartEncoding {
                content_encoding: Some("zstd".to_owned()),
            });
        } else {
            encodings.attachments.push(PartEncoding::default());
        }
    }

    let mut body = Vec::new();
    compressed_envelope
        .write_to_with_boundary_and_encodings(
            &mut body,
            detritus_protocol::multipart::DEFAULT_BOUNDARY,
            &encodings,
        )
        .await?;

    let url = crash_url(endpoint);
    let mut authorization = HeaderValue::from_str(&format!("Bearer {}", token.expose_secret()))
        .map_err(|_| ShipError::InvalidToken)?;
    authorization.set_sensitive(true);
    let mut response = client
        .post(url)
        .header(AUTHORIZATION, authorization)
        .header(
            CONTENT_TYPE,
            format!(
                "multipart/form-data; boundary={}",
                detritus_protocol::multipart::DEFAULT_BOUNDARY
            ),
        )
        .body(body)
        .send()
        .await?;
    if response.status().is_success() {
        // HTTP/1 connections return to the pool only after the response body is
        // consumed. Stream and discard it to keep memory bounded; the client's
        // read and total timeouts also apply while draining.
        while response.chunk().await?.is_some() {}
        Ok(())
    } else {
        Err(ShipError::Status(response.status()))
    }
}

fn crash_url(endpoint: &Url) -> Url {
    if endpoint.path().ends_with("/v1/crashes") {
        endpoint.clone()
    } else {
        endpoint
            .join("/v1/crashes")
            .unwrap_or_else(|_| endpoint.clone())
    }
}

fn cleanup_sent(sent: &Path, retention_days: u64) -> io::Result<()> {
    // saturating_mul avoids an overflow panic if retention_days is misconfigured
    // to an absurd value; the checked_sub below then saturates the cutoff to
    // UNIX_EPOCH, i.e. "keep everything", which is the intended extreme behavior.
    let cutoff = SystemTime::now()
        .checked_sub(Duration::from_secs(
            retention_days.saturating_mul(24 * 60 * 60),
        ))
        .unwrap_or(SystemTime::UNIX_EPOCH);
    for entry in pending_entries(sent)? {
        let modified = entry.metadata()?.modified().unwrap_or(SystemTime::now());
        if modified < cutoff {
            fs::remove_dir_all(entry)?;
        }
    }
    Ok(())
}

fn record_retention(max_retention_days: &mut Option<u64>, retention_days: u64) {
    *max_retention_days = Some(
        max_retention_days
            .map(|current| current.max(retention_days))
            .unwrap_or(retention_days),
    );
}

fn retention_or_default(max_retention_days: Option<u64>) -> u64 {
    max_retention_days.unwrap_or(DEFAULT_SENT_RETENTION_DAYS)
}

fn read_stored_upload_config(entry: &Path) -> Result<Option<StoredUploadConfig>, ShipError> {
    let path = entry.join("sdk-config.json");
    if path.exists() {
        Ok(Some(serde_json::from_slice(&fs::read(path)?)?))
    } else {
        Ok(None)
    }
}

/// Resolve a spool entry's stored upload endpoint and retention.
///
/// Returns the parsed endpoint URL and retention, or an error if the entry lacks
/// a usable `sdk-config.json`.
fn resolve_stored_endpoint(entry: &Path) -> Result<(Url, u64), ShipError> {
    let config = read_stored_upload_config(entry)?
        .ok_or_else(|| ShipError::MissingStoredConfig(entry.to_path_buf()))?;
    let endpoint = Url::parse(&config.endpoint)
        .map_err(|_| ShipError::MissingStoredConfig(entry.to_path_buf()))?;
    Ok((endpoint, config.sent_retention_days))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[cfg_attr(coverage_nightly, coverage(off))]
    fn write_stored_config(
        entry: &Path,
        endpoint: &str,
        sent_retention_days: u64,
    ) -> Result<(), ShipError> {
        fs::write(
            entry.join("sdk-config.json"),
            serde_json::to_vec_pretty(&StoredUploadConfig {
                endpoint: endpoint.to_owned(),
                sent_retention_days,
            })?,
        )?;
        Ok(())
    }

    #[test]
    #[cfg_attr(coverage_nightly, coverage(off))]
    fn read_stored_upload_config_round_trips_producer_format() {
        let temp = tempfile::tempdir().expect("temp dir is created");
        write_stored_config(temp.path(), "https://crashes.example.test/upload", 14)
            .expect("stored config is written");

        let config = read_stored_upload_config(temp.path())
            .expect("stored config is readable")
            .expect("stored config is present");

        assert_eq!(config.endpoint, "https://crashes.example.test/upload");
        assert_eq!(config.sent_retention_days, 14);
    }

    #[test]
    #[cfg_attr(coverage_nightly, coverage(off))]
    fn read_stored_upload_config_returns_none_when_missing() {
        let temp = tempfile::tempdir().expect("temp dir is created");

        let config =
            read_stored_upload_config(temp.path()).expect("missing config is not an error");

        assert!(config.is_none());
    }

    #[test]
    #[cfg_attr(coverage_nightly, coverage(off))]
    fn read_stored_upload_config_reports_malformed_json() {
        let temp = tempfile::tempdir().expect("temp dir is created");
        fs::write(temp.path().join("sdk-config.json"), b"{not-json")
            .expect("malformed config is written");

        assert!(matches!(
            read_stored_upload_config(temp.path()),
            Err(ShipError::Json(_))
        ));
    }

    #[test]
    #[cfg_attr(coverage_nightly, coverage(off))]
    fn resolve_stored_endpoint_reports_missing_config() {
        let temp = tempfile::tempdir().expect("temp dir is created");

        assert!(matches!(
            resolve_stored_endpoint(temp.path()),
            Err(ShipError::MissingStoredConfig(path)) if path == temp.path()
        ));
    }

    #[test]
    #[cfg_attr(coverage_nightly, coverage(off))]
    fn resolve_stored_endpoint_returns_url_and_retention() {
        let temp = tempfile::tempdir().expect("temp dir is created");
        write_stored_config(temp.path(), "https://crashes.example.test/base", 30)
            .expect("stored config is written");

        let (endpoint, retention_days) =
            resolve_stored_endpoint(temp.path()).expect("stored endpoint is usable");

        assert_eq!(endpoint.as_str(), "https://crashes.example.test/base");
        assert_eq!(retention_days, 30);
    }

    #[test]
    #[cfg_attr(coverage_nightly, coverage(off))]
    fn resolve_stored_endpoint_reports_invalid_url() {
        let temp = tempfile::tempdir().expect("temp dir is created");
        write_stored_config(temp.path(), "not a url", 30).expect("stored config is written");

        assert!(matches!(
            resolve_stored_endpoint(temp.path()),
            Err(ShipError::MissingStoredConfig(path)) if path == temp.path()
        ));
    }

    #[test]
    #[cfg_attr(coverage_nightly, coverage(off))]
    fn record_retention_keeps_maximum_value() {
        let mut retention = None;

        record_retention(&mut retention, 7);
        record_retention(&mut retention, 30);
        record_retention(&mut retention, 0);

        assert_eq!(retention_or_default(retention), 30);
    }

    #[test]
    #[cfg_attr(coverage_nightly, coverage(off))]
    fn retention_or_default_falls_back_when_no_entry_was_shipped() {
        assert_eq!(retention_or_default(None), DEFAULT_SENT_RETENTION_DAYS);
    }
}