Function transform_body

src/environmentd/src/http/webhook.rs:203–253

Transforms the body of a webhook request into a `Vec `.

(
    body: &[u8],
    format: &WebhookBodyFormat,
)

/// Transforms the body of a webhook request into a `Vec<BodyRow>`.
fn transform_body(
    body: &[u8],
    format: &WebhookBodyFormat,
) -> Result<Vec<BodyRow>, AppendWebhookError> {
    let rows = match format {
        WebhookBodyFormat::Bytes => {
            vec![Row::pack_slice(&[Datum::Bytes(body)])]
        }
        WebhookBodyFormat::Text => {
            let s = std::str::from_utf8(body)
                .map_err(|m| AppendWebhookError::InvalidUtf8Body { msg: m.to_string() })?;
            vec![Row::pack_slice(&[Datum::String(s)])]
        }
        WebhookBodyFormat::Json { array } => {
            let objects = serde_json::Deserializer::from_slice(body)
                // Automatically expand multiple JSON objects delimited by whitespace, e.g.
                // newlines, into a single batch.
                .into_iter::<serde_json::Value>()
                // Optionally expand a JSON array into separate rows, if requested.
                .flat_map(|value| match value {
                    Ok(serde_json::Value::Array(inners)) if *array => {
                        itertools::Either::Left(inners.into_iter().map(Result::Ok))
                    }
                    value => itertools::Either::Right(std::iter::once(value)),
                })
                .collect::<Result<Vec<_>, _>>()
                .map_err(|m| AppendWebhookError::InvalidJsonBody { msg: m.to_string() })?;

            // Note: `into_iter()` should be re-using the underlying allocation of the `objects`
            // vector, and it's more readable to split these into separate iterators.
            let rows = objects
                .into_iter()
                // Map a JSON object into a Row.
                .map(|o| {
                    let row = Jsonb::from_serde_json(o)
                        .map_err(|m| AppendWebhookError::InvalidJsonBody { msg: m.to_string() })?
                        .into_row();
                    Ok::<_, AppendWebhookError>(row)
                })
                .collect::<Result<_, _>>()?;

            rows
        }
    };

    // A `Row` cannot describe its schema without unpacking it. To add some safety we wrap the
    // returned `Row`s in a newtype to signify they already have the "body" column packed.
    let body_rows = rows.into_iter().map(BodyRow).collect();

    Ok(body_rows)
}

/// Pack the headers of a request into a [`Row`].
fn pack_header(