fastly 0.12.0

use std::{borrow::Cow, cell::OnceCell, collections::HashSet, time::Duration};

use fastly_shared::CacheOverride;
use fastly_sys::fastly_cache::CacheLookupState;
use http::{HeaderName, HeaderValue, StatusCode, Version};
use itertools::Itertools;
use mime::Mime;

use crate::{
    convert::{ToHeaderName, ToHeaderValue, ToStatusCode},
    handle::{BodyHandle, ResponseHandle, StreamingBodyHandle},
    http::{
        body::StreamingBody, cache, cache::HttpStorageAction, header, request::SendErrorCause,
        Body, Response,
    },
};

#[derive(Default)]
struct BodyTransform {
    transform: Option<Box<dyn FnOnce(Body, &mut StreamingBody) -> Result<(), SendErrorCause>>>,
}

impl BodyTransform {
    fn new(
        transform: impl FnOnce(Body, &mut StreamingBody) -> Result<(), SendErrorCause> + 'static,
    ) -> Self {
        Self {
            transform: Some(Box::new(transform)),
        }
    }

    // if the transform is guaranteed to not modify the body, then we can use
    // the input body length to know the output body length in advance
    fn is_no_op(&self) -> bool {
        self.transform.is_none()
    }

    fn do_transform(
        self,
        body_in: Body,
        body_out: &mut StreamingBody,
    ) -> Result<(), SendErrorCause> {
        if let Some(transform) = self.transform {
            transform(body_in, body_out)
        } else {
            // This default behavior results in host-side streaming whenever
            // possible; no reason to stream into the guest if we aren't
            // transforming the body.
            body_out.append(body_in);
            Ok(())
        }
    }

    fn apply_streaming(
        self,
        body_in: BodyHandle,
        body_out: StreamingBodyHandle,
    ) -> Result<(), SendErrorCause> {
        let mut body_out = StreamingBody::from(body_out);
        self.do_transform(body_in.into(), &mut body_out)?;
        body_out.finish()?;
        Ok(())
    }

    // Unfortunately, we cannot perform a streaming body transform when the
    // response is not going through the cache, because streaming is only
    // supported with a host-side "sink" (sending to client or into cache). This
    // will change with async fn support.
    //
    // Until then, we fake it by treating a normal body as streaming (which has
    // strictly fewer methods).
    fn apply(self, body_in: BodyHandle) -> Result<BodyHandle, SendErrorCause> {
        let mut body_out =
            StreamingBody::from(StreamingBodyHandle::from_body_handle(BodyHandle::new()));
        self.do_transform(body_in.into(), &mut body_out)?;
        Ok(body_out.into_handle()?.into_body_handle())
    }
}

/// The HTTP caching system gives "suggested" cache options based on the HTTP
/// headers for the response. A `CandidateResponse` can override those options,
/// but it can also _change_ the underlying headers. If a cache option hasn't
/// been overridden, but the headers controlling its default value are changed,
/// we want that to be reflected in e.g. `CandidateResponse::get_ttl()`.
///
/// This type is A wrapper around a `Response` that supports a lazily-loaded set
/// of suggested cache options, and guarantees that after any `&mut` access the
/// loaded options are invalidated.
///
/// The inner response does not have a body.
struct ProtectedResponse {
    response: Response,
    suggested_cache_options: OnceCell<cache::WriteOptions>,
}

impl ProtectedResponse {
    fn into_handle(self) -> ResponseHandle {
        self.response.into_response_handle()
    }

    fn into_inner(self) -> Response {
        self.response
    }
}

impl std::ops::Deref for ProtectedResponse {
    type Target = Response;

    fn deref(&self) -> &Self::Target {
        &self.response
    }
}

impl std::ops::DerefMut for ProtectedResponse {
    fn deref_mut(&mut self) -> &mut Self::Target {
        // invalidate the cache options, since the response may be modified via this reference
        drop(self.suggested_cache_options.take());
        &mut self.response
    }
}

/// A response from a backend that is a candidate for caching.
pub struct CandidateResponse {
    cache_handle: cache::HttpCacheHandle,
    response: ProtectedResponse,
    // the body received from the backend, if any
    body: BodyHandle,
    body_transform: BodyTransform,
    suggested_storage_action: cache::HttpStorageAction,
    // a space-separated list of keys, provided by `Request::set_surrogate_key`
    extra_surrogate_keys_abi: Option<String>,
    override_storage_action: Option<cache::HttpStorageAction>,
    override_ttl: Option<Duration>,
    override_stale_while_revalidate: Option<Duration>,
    // a space-separated list of headers
    override_vary_rule_abi: Option<String>,
    // a space-separated list of keys
    override_surrogate_keys_abi: Option<String>,
    override_pci: Option<bool>,
}

impl CandidateResponse {
    pub(crate) fn new(
        cache_handle: cache::HttpCacheHandle,
        cache_override: &CacheOverride,
        mut handle: ResponseHandle,
        body: BodyHandle,
    ) -> Result<Self, SendErrorCause> {
        let mut suggested_storage_action =
            cache_handle.prepare_response_for_storage(&mut handle)?;
        let response = Response::from_response_handle(handle);

        // Fastly-specific heuristic: by default, we do not cache responses that set cookies
        if response.contains_header(http::header::SET_COOKIE)
            && suggested_storage_action != HttpStorageAction::DoNotStore
        {
            suggested_storage_action = HttpStorageAction::RecordUncacheable;
        }

        let cand = Self {
            cache_handle,
            response: ProtectedResponse {
                response,
                suggested_cache_options: OnceCell::new(),
            },
            body,
            body_transform: BodyTransform::default(),
            suggested_storage_action,
            extra_surrogate_keys_abi: cache_override
                // this is a misnomer: it's actually _keys_ separated by spaces
                .get_surrogate_key()
                // TODO: throw error if surrogate key is invalid?
                .and_then(|k| HeaderValue::to_str(k).ok())
                .map(str::to_string),
            override_storage_action: None,
            override_pci: cache_override.get_pci(),
            override_stale_while_revalidate: cache_override
                .get_stale_while_revalidate()
                .map(|t| Duration::from_secs(t as u64)),
            override_surrogate_keys_abi: None,
            override_ttl: cache_override
                .get_ttl()
                .map(|t| Duration::from_secs(t as u64)),
            override_vary_rule_abi: None,
        };

        Ok(cand)
    }

    // The suggested cache options are not just what we get from the hostcall;
    // we also need to mix in other avenues for specifying surrogate keys, like
    // the request cache override, or response headers.
    fn build_fresh_suggested_cache_options(&self) -> cache::WriteOptions {
        let mut options = self
            .cache_handle
            .get_suggested_cache_options(self.response.get_response_handle());
        let keys = &mut options.surrogate_keys_abi;

        if let Some(extra_keys) = &self.extra_surrogate_keys_abi {
            if !keys.is_empty() {
                keys.push(' ');
            }
            keys.push_str(extra_keys);
        }

        for key in self
            .response
            .get_header_all(header::fastly::SURROGATE_KEY)
            .filter_map(|v| v.to_str().ok())
        {
            if !keys.is_empty() {
                keys.push(' ');
            }
            keys.push_str(key);
        }

        options
    }

    fn get_suggested_cache_options(&self) -> &cache::WriteOptions {
        self.response
            .suggested_cache_options
            .get_or_init(|| self.build_fresh_suggested_cache_options())
    }

    /// Combines the suggested and override cache settings into a final set of options
    fn finalize_options(&mut self) -> (HttpStorageAction, cache::WriteOptions) {
        let storage_action = self
            .override_storage_action
            .take()
            .unwrap_or(self.suggested_storage_action);

        let suggested = self
            .response
            .suggested_cache_options
            .take()
            .unwrap_or_else(|| self.build_fresh_suggested_cache_options());

        let write_options = cache::WriteOptions {
            max_age: self
                .override_ttl
                .map(|ttl| ttl + suggested.initial_age)
                .unwrap_or(suggested.max_age),
            initial_age: suggested.initial_age,
            stale_while_revalidate: self
                .override_stale_while_revalidate
                .unwrap_or(suggested.stale_while_revalidate),
            vary_rule_abi: self
                .override_vary_rule_abi
                .take()
                .unwrap_or(suggested.vary_rule_abi),
            surrogate_keys_abi: self
                .override_surrogate_keys_abi
                .take()
                .unwrap_or(suggested.surrogate_keys_abi),
            sensitive_data: self.override_pci.unwrap_or(suggested.sensitive_data),
            length: self
                .body
                .known_length()
                .filter(|_| self.body_transform.is_no_op()),
        };

        (storage_action, write_options)
    }

    /// Performs the cache actions specified by the candidate response, and
    /// returns the final response _from_ the cache.
    ///
    /// That setup is important: the backend may stream the body into the cache
    /// more quickly than the client reads it out. By pulling the final response
    /// from the cache after streaming insertion, we allow backend and client to
    /// go at different speeds.
    pub(crate) fn apply_and_stream_back(mut self) -> Result<Response, SendErrorCause> {
        use cache::HttpStorageAction::*;
        let (storage_action, options) = self.finalize_options();
        let mut resp = match storage_action {
            Insert => {
                let (body_out, readback_entry) = self
                    .cache_handle
                    .transaction_insert_and_stream_back(self.response.into_handle(), &options)?;
                self.body_transform.apply_streaming(self.body, body_out)?;
                readback_entry.get_found_response(false)?
            }
            Update => self
                .cache_handle
                .transaction_update_and_return_fresh(self.response.into_handle(), &options)?
                .get_found_response(true)?,
            DoNotStore => {
                // Use `abandon` to only wake a single waiter in the
                // non-hit-for-pass case, so concurrent requests remain
                // serialized.
                self.cache_handle.transaction_abandon()?;
                self.into_response(options)?
            }
            RecordUncacheable => {
                self.cache_handle
                    .transaction_record_not_cacheable(options.max_age, &options.vary_rule_abi)?;
                self.into_response(options)?
            }
        };
        resp.metadata.cache_storage_action = Some(storage_action);
        Ok(resp)
    }

    /// Produces a final response without going through the cache
    fn into_response(self, cache_options: cache::WriteOptions) -> Result<Response, SendErrorCause> {
        let mut response = self.response.into_inner();
        response.set_body(self.body_transform.apply(self.body)?);
        response.metadata.cache_options = Some(cache_options);
        Ok(response)
    }

    /// Performs the cache actions specified by the candidate response, but does
    /// not return the finalized response; appropriate for background
    /// revalidation during SWR. We don't currently implement that via an actual
    /// background task, but the application logic is the same regardless.
    pub(crate) fn apply_in_background(mut self) -> Result<(), SendErrorCause> {
        use cache::HttpStorageAction::*;
        let (storage_action, options) = self.finalize_options();
        match storage_action {
            Insert => {
                let body_out = self
                    .cache_handle
                    .transaction_insert(self.response.into_handle(), &options)?;
                self.body_transform.apply_streaming(self.body, body_out)?;
            }
            Update => self
                .cache_handle
                .transaction_update(self.response.into_handle(), &options)?,
            DoNotStore => {
                // Use `abandon` to only wake a single waiter in the
                // non-hit-for-pass case, so concurrent requests remain
                // serialized.
                self.cache_handle.transaction_abandon()?;
            }
            RecordUncacheable => self
                .cache_handle
                .transaction_record_not_cacheable(options.max_age, &options.vary_rule_abi)?,
        };
        Ok(())
    }

    /// Get the MIME type described by the response's
    /// [`Content-Type`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Type)
    /// header, or `None` if that header is absent or contains an invalid MIME type.
    ///
    /// # Examples
    ///
    /// See [`Response::get_content_type`].
    pub fn get_content_type(&self) -> Option<Mime> {
        self.response.get_content_type()
    }

    /// Builder-style equivalent of [`set_content_type()`][`Self::set_content_type()`].
    pub fn with_content_type(mut self, mime: Mime) -> Self {
        self.set_content_type(mime);
        self
    }

    /// Set the MIME type described by the response's
    /// [`Content-Type`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Type)
    /// header.
    ///
    /// Any existing `Content-Type` header values will be overwritten.
    ///
    /// # Examples
    ///
    /// See [`Response::set_content_type`].
    pub fn set_content_type(&mut self, mime: Mime) {
        self.response.set_content_type(mime);
    }

    /// Get the value of the response's
    /// [`Content-Length`](https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Length)
    /// header, if it exists.
    pub fn get_content_length(&self) -> Option<usize> {
        self.get_header(http::header::CONTENT_LENGTH)
            .and_then(|v| v.to_str().ok())
            .and_then(|v| v.parse().ok())
    }

    /// Returns whether the given header name is present in the response.
    ///
    #[doc = include_str!("../../docs/snippets/header-name-argument.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::contains_header`].
    pub fn contains_header(&self, name: impl ToHeaderName) -> bool {
        self.response.contains_header(name)
    }

    /// Builder-style equivalent of [`append_header()`][`Self::append_header()`].
    pub fn with_header(mut self, name: impl ToHeaderName, value: impl ToHeaderValue) -> Self {
        self.append_header(name, value);
        self
    }

    /// Builder-style equivalent of [`set_header()`][`Self::set_header()`].
    pub fn with_set_header(mut self, name: impl ToHeaderName, value: impl ToHeaderValue) -> Self {
        self.set_header(name, value);
        self
    }

    /// Get the value of a header as a string, or `None` if the header is not present.
    ///
    /// If there are multiple values for the header, only one is returned, which may be any of the
    /// values. See [`get_header_all_str()`][`Self::get_header_all_str()`] if you need to get all of
    /// the values.
    ///
    #[doc = include_str!("../../docs/snippets/header-name-argument.md")]
    ///
    /// # Panics
    ///
    #[doc = include_str!("../../docs/snippets/panics-reqresp-header-utf8.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::get_header_str`].
    pub fn get_header_str(&self, name: impl ToHeaderName) -> Option<&str> {
        self.response.get_header_str(name)
    }

    /// Get the value of a header as a string, including invalid characters, or `None` if the header
    /// is not present.
    ///
    #[doc = include_str!("../../docs/snippets/utf8-replacement.md")]
    ///
    /// If there are multiple values for the header, only one is returned, which may be any of the
    /// values. See [`get_header_all_str_lossy()`][`Self::get_header_all_str_lossy()`] if you need
    /// to get all of the values.
    ///
    #[doc = include_str!("../../docs/snippets/header-name-argument.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::get_header_str_lossy`].
    pub fn get_header_str_lossy(&self, name: impl ToHeaderName) -> Option<Cow<'_, str>> {
        self.response.get_header_str_lossy(name)
    }

    /// Get the value of a header, or `None` if the header is not present.
    ///
    /// If there are multiple values for the header, only one is returned, which may be any of the
    /// values. See [`get_header_all()`][`Self::get_header_all()`] if you need to get all of the
    /// values.
    ///
    #[doc = include_str!("../../docs/snippets/header-name-argument.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::get_header`].
    pub fn get_header(&self, name: impl ToHeaderName) -> Option<&HeaderValue> {
        self.response.get_header(name)
    }

    /// Get all values of a header as strings, or an empty vector if the header is not present.
    ///
    #[doc = include_str!("../../docs/snippets/header-name-argument.md")]
    ///
    /// # Panics
    ///
    #[doc = include_str!("../../docs/snippets/panics-reqresp-headers-utf8.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::get_header_all_str`].
    pub fn get_header_all_str(&self, name: impl ToHeaderName) -> Vec<&str> {
        self.response.get_header_all_str(name)
    }

    /// Get an iterator of all the response's header names and values.
    ///
    /// # Examples
    ///
    /// See [`Response::get_headers`].
    pub fn get_headers(&self) -> impl Iterator<Item = (&HeaderName, &HeaderValue)> {
        self.response.get_headers()
    }

    /// Get all values of a header as strings, including invalid characters, or an empty vector if the header is not present.
    ///
    #[doc = include_str!("../../docs/snippets/utf8-replacement.md")]
    ///
    #[doc = include_str!("../../docs/snippets/header-name-argument.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::get_header_all_str_lossy`].
    pub fn get_header_all_str_lossy(&self, name: impl ToHeaderName) -> Vec<Cow<'_, str>> {
        self.response.get_header_all_str_lossy(name)
    }

    /// Get an iterator of all the values of a header.
    ///
    #[doc = include_str!("../../docs/snippets/header-name-argument.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::get_header_all`].
    pub fn get_header_all(&self, name: impl ToHeaderName) -> impl Iterator<Item = &HeaderValue> {
        self.response.get_header_all(name)
    }

    /// Get all of the response's header names as strings, or an empty vector if no headers are
    /// present.
    ///
    /// # Examples
    ///
    /// See [`Response::get_header_names_str`].
    pub fn get_header_names_str(&self) -> Vec<&str> {
        self.response.get_header_names_str()
    }

    /// Get an iterator of all the response's header names.
    ///
    /// # Examples
    ///
    /// See [`Response::get_header_names`].
    pub fn get_header_names(&self) -> impl Iterator<Item = &HeaderName> {
        self.response.get_header_names()
    }

    /// Set a response header to the given value, discarding any previous values for the given
    /// header name.
    ///
    #[doc = include_str!("../../docs/snippets/header-name-value-argument.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::set_header`].
    pub fn set_header(&mut self, name: impl ToHeaderName, value: impl ToHeaderValue) {
        self.response.set_header(name, value);
    }

    /// Add a response header with given value.
    ///
    /// Unlike [`set_header()`][`Self::set_header()`], this does not discard existing values for the
    /// same header name.
    ///
    #[doc = include_str!("../../docs/snippets/header-name-value-argument.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::append_header`].
    pub fn append_header(&mut self, name: impl ToHeaderName, value: impl ToHeaderValue) {
        self.response.append_header(name, value);
    }

    /// Remove all response headers of the given name, and return one of the removed header values
    /// if any were present.
    ///
    /// If the header has multiple values, one is returned arbitrarily. To get all of the removed
    /// header values, or to get a specific value, use
    /// [`get_header_all()`][`Self::get_header_all()`].
    ///
    #[doc = include_str!("../../docs/snippets/header-name-argument.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::remove_header`].
    pub fn remove_header(&mut self, name: impl ToHeaderName) -> Option<HeaderValue> {
        self.response.remove_header(name)
    }

    /// Remove all response headers of the given name, and return one of the removed header values
    /// as a string if any were present.
    ///
    #[doc = include_str!("../../docs/snippets/removes-one-header.md")]
    ///
    #[doc = include_str!("../../docs/snippets/header-name-argument.md")]
    ///
    /// # Panics
    ///
    #[doc = include_str!("../../docs/snippets/panics-reqresp-remove-header-utf8.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::remove_header_str`].
    pub fn remove_header_str(&mut self, name: impl ToHeaderName) -> Option<String> {
        self.response.remove_header_str(name)
    }

    /// Remove all response headers of the given name, and return one of the removed header values
    /// as a string, including invalid characters, if any were present.
    ///
    #[doc = include_str!("../../docs/snippets/utf8-replacement.md")]
    ///
    #[doc = include_str!("../../docs/snippets/removes-one-header.md")]
    ///
    #[doc = include_str!("../../docs/snippets/header-name-argument.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::remove_header_str_lossy`].
    pub fn remove_header_str_lossy(&mut self, name: impl ToHeaderName) -> Option<String> {
        self.response.remove_header_str_lossy(name)
    }

    /// Get the HTTP status code of the response.
    pub fn get_status(&self) -> StatusCode {
        self.response.get_status()
    }

    /// Builder-style equivalent of [`set_status()`][`Self::set_status()`].
    pub fn with_status(mut self, status: impl ToStatusCode) -> Self {
        self.set_status(status);
        self
    }

    /// Set the HTTP status code of the response.
    ///
    #[doc = include_str!("../../docs/snippets/statuscode-argument.md")]
    ///
    /// # Examples
    ///
    /// See [`Response::set_status`].
    pub fn set_status(&mut self, status: impl ToStatusCode) {
        self.response.set_status(status)
    }

    /// Get the HTTP version of this response.
    pub fn get_version(&self) -> Version {
        self.response.get_version()
    }

    /// Builder-style equivalent of [`set_version()`][`Self::set_version()`].
    pub fn with_version(mut self, version: Version) -> Self {
        self.set_version(version);
        self
    }

    /// Set the HTTP version of this response.
    pub fn set_version(&mut self, version: Version) {
        self.response.set_version(version)
    }

    /// Get the Time to Live (TTL) in the cache for this response.
    ///
    /// The TTL determines the duration of "freshness" for the cached response
    /// after it is inserted into the cache.
    pub fn get_ttl(&self) -> Duration {
        self.override_ttl.unwrap_or(
            self.get_suggested_cache_options()
                .max_age
                .saturating_sub(self.get_age()),
        )
    }

    /// Builder-style equivalent of [`set_ttl()`][`Self::set_ttl()`].
    pub fn with_ttl(mut self, ttl: Duration) -> Self {
        self.set_ttl(ttl);
        self
    }

    /// Set the Time to Live (TTL) in the cache for this response.
    ///
    /// The TTL determines the duration of "freshness" for the cached response
    /// after it is inserted into the cache.
    pub fn set_ttl(&mut self, ttl: Duration) {
        self.override_ttl = Some(ttl);
    }

    /// The current age of the cached item, relative to the originating backend.
    pub fn get_age(&self) -> Duration {
        self.get_suggested_cache_options().initial_age
    }

    /// Gets the time for which a cached item can safely be used despite being considered stale.
    pub fn get_stale_while_revalidate(&self) -> Duration {
        self.override_stale_while_revalidate
            .unwrap_or(self.get_suggested_cache_options().stale_while_revalidate)
    }

    /// Builder-style equivalent of [`set_stale_while_revalidate()`][`Self::set_stale_while_revalidate()`].
    pub fn with_stale_while_revalidate(mut self, stale_while_revalidate: Duration) -> Self {
        self.set_stale_while_revalidate(stale_while_revalidate);
        self
    }

    /// Sets the time for which a cached item can safely be used despite being considered stale.
    pub fn set_stale_while_revalidate(&mut self, stale_while_revalidate: Duration) {
        self.override_stale_while_revalidate = Some(stale_while_revalidate)
    }

    /// Determines whether the cached response is stale.
    ///
    /// A cached response is stale if it has been in the cache beyond its TTL period,
    /// or if it has been [soft purged].
    ///
    /// [soft purged]: https://www.fastly.com/documentation/guides/concepts/edge-state/cache/purging/#soft-vs-hard-purging
    pub fn is_stale(&self) -> bool {
        self.cache_handle
            .get_state()
            .contains(CacheLookupState::STALE)
    }

    /// Gets the set of request headers for which the response may vary.
    pub fn get_vary(&self) -> impl Iterator<Item = &str> {
        self.override_vary_rule_abi
            .as_ref()
            .unwrap_or(&self.get_suggested_cache_options().vary_rule_abi)
            .split(" ")
    }

    /// Builder-style equivalent of [`set_vary()`][`Self::set_vary()`].
    pub fn with_vary<'a>(mut self, headers: impl IntoIterator<Item = &'a HeaderName>) -> Self {
        self.set_vary(headers);
        self
    }

    /// Sets the set of request headers for which the response may vary.
    pub fn set_vary<'a>(&mut self, headers: impl IntoIterator<Item = &'a HeaderName>) {
        self.override_vary_rule_abi = Some(
            Itertools::intersperse(headers.into_iter().map(HeaderName::as_str), " ").collect(),
        );
    }

    /// Adds zero or more headers to the set of request headers for which the response may vary.
    pub fn append_vary<'a>(&mut self, headers: impl IntoIterator<Item = &'a HeaderName>) {
        let mut varies: HashSet<&str> = self.get_vary().collect();
        for header in headers {
            varies.insert(header.as_str());
        }
        self.override_vary_rule_abi =
            Some(Itertools::intersperse(varies.into_iter(), " ").collect());
    }

    /// Adds a header to the set of request headers for which the response may vary.
    pub fn push_vary(&mut self, header: &HeaderName) {
        self.append_vary(std::iter::once(header));
    }

    /// Gets the surrogate keys for the cached response.
    pub fn get_surrogate_keys(&self) -> impl Iterator<Item = &str> {
        self.override_surrogate_keys_abi
            .as_ref()
            .unwrap_or(&self.get_suggested_cache_options().surrogate_keys_abi)
            .split(" ")
    }

    /// Builder-style equivalent of [`set_surrogate_keys()`][`Self::set_surrogate_keys()`].
    pub fn with_surrogate_keys<'a>(mut self, keys: impl IntoIterator<Item = &'a str>) -> Self {
        self.set_surrogate_keys(keys);
        self
    }

    /// Sets the surrogate keys for the cached response.
    ///
    /// Surrogate keys must contain only printable ASCII characters (those between `0x21` and
    /// `0x7E`, inclusive). Any invalid keys will be ignored.
    ///
    /// See the [Fastly surrogate keys
    /// guide](https://docs.fastly.com/en/guides/purging-api-cache-with-surrogate-keys) for details.
    pub fn set_surrogate_keys<'a>(&mut self, keys: impl IntoIterator<Item = &'a str>) {
        self.override_surrogate_keys_abi =
            Some(Itertools::intersperse(keys.into_iter(), " ").collect());
    }

    /// Gets whether this response should only be stored via PCI/HIPAA-compliant non-volatile caching.
    ///
    /// See the [Fastly PCI-Compliant Caching and Delivery
    /// documentation](https://docs.fastly.com/products/pci-compliant-caching-and-delivery) for
    /// details.
    pub fn is_sensitive_data(&self) -> bool {
        self.override_pci
            .unwrap_or(self.get_suggested_cache_options().sensitive_data)
    }

    /// Builder-style equivalent of [`set_sensitive_data()`][`Self::set_sensitive_data()`].
    pub fn with_sensitive_data(mut self, is_sensitive_data: bool) -> Self {
        self.set_sensitive_data(is_sensitive_data);
        self
    }

    /// Set the caching behavior of this response to enable or disable PCI/HIPAA-compliant
    /// non-volatile caching.
    ///
    /// By default, this is `false`, which means the response may not be PCI/HIPAA-compliant. Set it
    /// to `true` to enable compliant caching.
    ///
    /// See the [Fastly PCI-Compliant Caching and Delivery
    /// documentation](https://docs.fastly.com/products/pci-compliant-caching-and-delivery) for
    /// details.
    pub fn set_sensitive_data(&mut self, is_sensitive_data: bool) {
        self.override_pci = Some(is_sensitive_data);
    }

    /// Gets whether this response should be cached.
    pub fn is_cacheable(&self) -> bool {
        let action = self
            .override_storage_action
            .unwrap_or(self.suggested_storage_action);
        matches!(
            action,
            cache::HttpStorageAction::Insert | cache::HttpStorageAction::Update
        )
    }

    /// Builder-style equivalent of [`set_cacheable()`][`Self::set_cacheable()`]
    /// and [`set_uncacheable()`][`Self::set_uncacheable()`]
    pub fn with_cacheable(mut self, is_cacheable: bool, record_uncacheable: bool) -> Self {
        if is_cacheable {
            self.set_cacheable()
        } else {
            self.set_uncacheable(record_uncacheable)
        }
        self
    }

    /// Forces this response to be stored in the cache, even if its headers or
    /// status would normally prevent that.
    pub fn set_cacheable(&mut self) {
        // Revalidating responses _always_ have a suggested action of `Update`; retain that
        // if we toggled between storing / not storing
        self.override_storage_action =
            if self.suggested_storage_action == cache::HttpStorageAction::Update {
                Some(cache::HttpStorageAction::Update)
            } else {
                Some(cache::HttpStorageAction::Insert)
            };
    }

    /// Set the response to not be stored in the cache.
    ///
    /// Optionally, if `record_uncacheable` is true, the cache will record that
    /// the originating request led to an uncacheable response, so that future
    /// cache lookups will result in immediately going to the backend, rather
    /// than attempting to coordinate concurrent requests to reduce backend
    /// traffic.
    ///
    /// See the [Fastly request collapsing
    /// guide](https://www.fastly.com/documentation/guides/concepts/edge-state/cache/request-collapsing/)
    /// for more details on the mechanism that `record_uncacheable` disables.
    pub fn set_uncacheable(&mut self, record_uncacheable: bool) {
        self.override_storage_action = if record_uncacheable {
            Some(cache::HttpStorageAction::RecordUncacheable)
        } else {
            Some(cache::HttpStorageAction::DoNotStore)
        };
    }

    /// Sets a callback to be used for transforming the response body prior to
    /// caching.
    ///
    /// Body transformations are performed via a callback, rather than by
    /// directly working with the body, because not every response contains a
    /// fresh body: 304 Not Modified responses, which are used to revalidate a
    /// stale cached response, are valuable precisely because they do not
    /// retransmit the body.
    ///
    /// For any other response status, the backend response will contain a
    /// relevant body, and the `transform` callback will be invoked. The
    /// original backend body of type `Body` is given as the first parameter to
    /// the transform. The transform is responsible for writing the new body
    /// into the `StreamingBody` given as its second argument. Upon returning from
    /// the transform, [`StreamingBody::finish()`] will be called implicitly. This
    /// setup allows the transform to work with streamed chunks of the backend body,
    /// rather than necessarily reading it entirely into memory.
    pub fn set_body_transform(
        &mut self,
        transform: impl FnOnce(Body, &mut StreamingBody) -> Result<(), SendErrorCause> + 'static,
    ) {
        self.body_transform = BodyTransform::new(transform)
    }
}