box-open-sdk 0.4.1

// Code generated by box-gantry. DO NOT EDIT.
use crate::client::Client;
use crate::models::schemas;
use crate::runtime::Error;
use base64::Engine as _;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};

/// Parts in flight at once (`join_ordered` bounds each batch), capping peak
/// buffer memory while keeping several requests moving together.
const MAX_CONCURRENT: usize = 4;

/// An orchestrator for Box chunked (multipart) uploads over a [`Client`]: it
/// creates an upload session, uploads the content's parts with bounded
/// concurrency, and commits — for a new file ([`ChunkedUpload::upload`]) or a
/// new version of an existing file ([`ChunkedUpload::upload_version`]). Box
/// requires chunked upload for files at or above its minimum session size;
/// smaller files use the single-shot upload endpoints.
pub struct ChunkedUpload<'a> {
    client: &'a Client,
}

impl Client {
    /// A chunked-upload orchestrator borrowing this client.
    pub fn chunked_upload(&self) -> ChunkedUpload<'_> {
        ChunkedUpload { client: self }
    }
}

impl ChunkedUpload<'_> {
    /// Upload `content` as a new file named `file_name` into `folder_id`.
    pub async fn upload(
        &self,
        content: &[u8],
        file_name: &str,
        folder_id: &str,
    ) -> Result<schemas::Files, Error> {
        let session = self
            .client
            .chunked_uploads
            .create_file_upload_session(schemas::CreateFileUploadSessionRequest {
                folder_id: folder_id.to_string(),
                file_size: content.len() as i64,
                file_name: file_name.to_string(),
            })
            .await?;
        self.finish(session, content).await
    }

    /// Upload `content` as a new version of the existing file `file_id`.
    pub async fn upload_version(
        &self,
        content: &[u8],
        file_name: &str,
        file_id: &str,
    ) -> Result<schemas::Files, Error> {
        let session = self
            .client
            .chunked_uploads
            .create_file_version_upload_session(
                file_id.to_string(),
                schemas::CreateFileVersionUploadSessionRequest {
                    file_size: content.len() as i64,
                    file_name: Some(file_name.to_string()),
                },
            )
            .await?;
        self.finish(session, content).await
    }

    async fn finish(
        &self,
        session: schemas::UploadSession,
        content: &[u8],
    ) -> Result<schemas::Files, Error> {
        let id = session
            .id
            .ok_or_else(|| Error::new("gantry: upload session returned no id"))?;
        let part_size = match session.part_size {
            Some(size) if size > 0 => size as usize,
            _ => {
                return Err(Error::new(
                    "gantry: upload session returned a non-positive part_size",
                ));
            }
        };
        let total = content.len();

        // Upload in batches of MAX_CONCURRENT parts at a time, each batch driven
        // concurrently (the requests are in flight together) but committed in
        // order — Box's commit lists parts in offset order.
        let offsets: Vec<usize> = (0..total).step_by(part_size).collect();
        let mut parts = Vec::with_capacity(offsets.len());
        for batch in offsets.chunks(MAX_CONCURRENT) {
            let uploads = batch
                .iter()
                .map(|&start| {
                    let end = (start + part_size).min(total);
                    self.upload_part(&id, &content[start..end], start, end, total)
                })
                .collect();
            for result in join_ordered(uploads).await {
                parts.push(result?);
            }
        }

        let digest = format!(
            "sha={}",
            base64::engine::general_purpose::STANDARD.encode(sha1(content))
        );
        self.client
            .chunked_uploads
            .commit_file_upload_session(
                id,
                digest,
                schemas::CommitFileUploadSessionRequest { parts },
                None,
            )
            .await
    }

    async fn upload_part(
        &self,
        id: &str,
        slice: &[u8],
        start: usize,
        end: usize,
        total: usize,
    ) -> Result<schemas::UploadPart, Error> {
        let digest = format!(
            "sha={}",
            base64::engine::general_purpose::STANDARD.encode(sha1(slice))
        );
        let content_range = format!("bytes {}-{}/{}", start, end - 1, total);
        let uploaded = self
            .client
            .chunked_uploads
            .update_file_upload_session(id.to_string(), digest, content_range, slice.to_vec())
            .await?;
        uploaded
            .part
            .ok_or_else(|| Error::new("gantry: upload part returned no part"))
    }
}

/// Await every future concurrently on the current task, returning results in
/// input order. No `tokio::spawn`, so it needs neither a runtime feature nor a
/// `Send`/`'static` bound — enough for I/O-bound part uploads, whose requests
/// are in flight together and driven by the reactor, without pulling in a
/// `futures`/`tokio` `rt` dependency the SDK avoids.
async fn join_ordered<F>(futures: Vec<F>) -> Vec<F::Output>
where
    F: Future,
    F::Output: Unpin,
{
    JoinOrdered {
        slots: futures
            .into_iter()
            .map(|f| (Some(Box::pin(f)), None))
            .collect(),
    }
    .await
}

/// The [`Future`] backing [`join_ordered`]: it polls each not-yet-ready future
/// on every wake and finishes once all have produced a value.
struct JoinOrdered<F: Future> {
    #[allow(clippy::type_complexity)]
    slots: Vec<(Option<Pin<Box<F>>>, Option<F::Output>)>,
}

impl<F> Future for JoinOrdered<F>
where
    F: Future,
    F::Output: Unpin,
{
    type Output = Vec<F::Output>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let slots = &mut self.get_mut().slots;
        let mut pending = false;
        for (future, output) in slots.iter_mut() {
            if let Some(f) = future {
                match f.as_mut().poll(cx) {
                    Poll::Ready(value) => {
                        *output = Some(value);
                        *future = None;
                    }
                    Poll::Pending => pending = true,
                }
            }
        }
        if pending {
            Poll::Pending
        } else {
            Poll::Ready(
                slots
                    .iter_mut()
                    .map(|(_, out)| out.take().unwrap())
                    .collect(),
            )
        }
    }
}

/// SHA-1 (RFC 3174) of `data`. Hand-rolled so the SDK needs no hashing
/// dependency — its crypto stack (`sha2`, `rsa`) carries no SHA-1, and Box's
/// chunked-upload digests are `sha=<base64(sha1(bytes))>`.
fn sha1(data: &[u8]) -> [u8; 20] {
    let mut h: [u32; 5] = [
        0x6745_2301,
        0xEFCD_AB89,
        0x98BA_DCFE,
        0x1032_5476,
        0xC3D2_E1F0,
    ];
    let bit_len = (data.len() as u64).wrapping_mul(8);

    // Pad: append 0x80, zero-fill to 56 mod 64, then the 64-bit big-endian length.
    let mut msg = data.to_vec();
    msg.push(0x80);
    while msg.len() % 64 != 56 {
        msg.push(0);
    }
    msg.extend_from_slice(&bit_len.to_be_bytes());

    for block in msg.chunks_exact(64) {
        let mut w = [0u32; 80];
        for (word, bytes) in w.iter_mut().zip(block.chunks_exact(4)) {
            *word = u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]);
        }
        for i in 16..80 {
            w[i] = (w[i - 3] ^ w[i - 8] ^ w[i - 14] ^ w[i - 16]).rotate_left(1);
        }

        let [mut a, mut b, mut c, mut d, mut e] = h;
        for (i, &word) in w.iter().enumerate() {
            let (f, k) = match i {
                0..=19 => ((b & c) | ((!b) & d), 0x5A82_7999_u32),
                20..=39 => (b ^ c ^ d, 0x6ED9_EBA1),
                40..=59 => ((b & c) | (b & d) | (c & d), 0x8F1B_BCDC),
                _ => (b ^ c ^ d, 0xCA62_C1D6),
            };
            let temp = a
                .rotate_left(5)
                .wrapping_add(f)
                .wrapping_add(e)
                .wrapping_add(k)
                .wrapping_add(word);
            e = d;
            d = c;
            c = b.rotate_left(30);
            b = a;
            a = temp;
        }

        h[0] = h[0].wrapping_add(a);
        h[1] = h[1].wrapping_add(b);
        h[2] = h[2].wrapping_add(c);
        h[3] = h[3].wrapping_add(d);
        h[4] = h[4].wrapping_add(e);
    }

    let mut out = [0u8; 20];
    for (chunk, word) in out.chunks_exact_mut(4).zip(h) {
        chunk.copy_from_slice(&word.to_be_bytes());
    }
    out
}

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

    fn hex(bytes: &[u8]) -> String {
        bytes.iter().map(|b| format!("{b:02x}")).collect()
    }

    #[test]
    fn sha1_matches_known_answer_vectors() {
        // RFC 3174 / NIST vectors, plus the 55/56/64-byte cases that exercise the
        // padding boundary (56 mod 64) a hand-rolled implementation is most likely
        // to get wrong, and the classic 1e6-byte vector spanning many blocks.
        assert_eq!(hex(&sha1(b"")), "da39a3ee5e6b4b0d3255bfef95601890afd80709");
        assert_eq!(
            hex(&sha1(b"abc")),
            "a9993e364706816aba3e25717850c26c9cd0d89d"
        );
        assert_eq!(
            hex(&sha1(
                b"abcdbcdecdefdefgefghfghighijhijkijkljklmklmnlmnomnopnopq"
            )),
            "84983e441c3bd26ebaae4aa1f95129e5e54670f1"
        );
        assert_eq!(
            hex(&sha1(&[b'a'; 55])),
            "c1c8bbdc22796e28c0e15163d20899b65621d65a"
        );
        assert_eq!(
            hex(&sha1(&[b'a'; 56])),
            "c2db330f6083854c99d4b5bfb6e8f29f201be699"
        );
        assert_eq!(
            hex(&sha1(&[b'a'; 64])),
            "0098ba824b5c16427bd7a1122a5a442a25ec644d"
        );
        assert_eq!(
            hex(&sha1(&[b'a'; 1_000_000])),
            "34aa973cd4c4daa4f61eeb2bdbad27316534016f"
        );
    }
}