spiral-sdk 0.1.12

//! References to payloads stored as dedicated objects of a Table: under its `blobs/` prefix,
//! or under a write attempt's `blobs/` when uploaded into that attempt.

use std::ops::RangeInclusive;
use std::sync::Arc;

use arrow_array::{ArrayRef, BinaryArray, StringArray, StructArray, UInt64Array};
use arrow_schema::{DataType, Field, Fields};
use spiral_contract::safelong::SafeUnsignedLong;
use uuid::Uuid;

pub use super::api::{Blob, BlobPart, BlobStat, BlobUpload};
use super::error::ScandalError;

/// The part size of an upload that names none.
pub const DEFAULT_UPLOAD_PART_BYTES: u64 = 64 << 20;

/// The part sizes an upload may choose, each a power of two.
pub const UPLOAD_PART_BYTES: RangeInclusive<u64> = (8 << 20)..=(1 << 30);

/// The most parts one upload holds, as S3 and Cloud Storage allow.
pub const MAX_UPLOAD_PARTS: u32 = 10_000;

/// The `spql.Blob` extension metadata: its property bridges in the SPQL type registry. Blobs
/// compare by identity and have no order.
const BLOB_METADATA: &str = r#"{"eq":"identity","order":"none"}"#;

impl Blob {
    /// A reference to a completed payload of `table_id`, under its `blobs/` prefix.
    pub fn new(
        table_id: Uuid,
        object_id: Uuid,
        size: u64,
        digest: [u8; 32],
        media_type: impl Into<String>,
    ) -> Self {
        Self {
            table_id,
            object_id,
            size: SafeUnsignedLong(size),
            digest: hex_digest(&digest),
            media_type: media_type.into(),
            attempt: None,
        }
    }

    /// The BLAKE3 digest of the whole payload.
    pub fn digest_bytes(&self) -> Result<[u8; 32], ScandalError> {
        parse_digest(&self.digest)
    }

    /// The payload's object, relative to its Table's root.
    pub fn key(&self) -> String {
        object_key(self.attempt, self.object_id)
    }

    /// Resource address understood by Spiral Blob projections.
    pub fn uri(&self) -> String {
        format!("spiral://{}/{}", self.table_id, self.key())
    }

    /// Arrow field for the public `spql.Blob` extension representation.
    pub fn field(name: &str) -> Field {
        let fields: Fields = vec![
            Field::new("bytes", DataType::Binary, true),
            Field::new("uri", DataType::Utf8, true),
            Field::new("offset", DataType::UInt64, true),
            Field::new("size", DataType::UInt64, true),
            Field::new("mime_type", DataType::Utf8, true),
            Field::new("version", DataType::Utf8, true),
            Field::new("etag", DataType::Utf8, true),
            Field::new("digest", DataType::Binary, true),
        ]
        .into();
        Field::new(name, DataType::Struct(fields), false).with_metadata(
            [
                ("ARROW:extension:name".into(), "spql.Blob".into()),
                ("ARROW:extension:metadata".into(), BLOB_METADATA.into()),
            ]
            .into(),
        )
    }

    /// Encode completed payload references as a Blob column without reading their bytes.
    pub fn array(blobs: &[Self]) -> Result<ArrayRef, ScandalError> {
        let n = blobs.len();
        let uris: Vec<_> = blobs.iter().map(Self::uri).collect();
        let digests = blobs
            .iter()
            .map(|blob| {
                let mut digest = vec![0x1e, 32];
                digest.extend_from_slice(&blob.digest_bytes()?);
                Ok(digest)
            })
            .collect::<Result<Vec<_>, ScandalError>>()?;
        let columns: Vec<ArrayRef> = vec![
            Arc::new(BinaryArray::from_iter((0..n).map(|_| None::<&[u8]>))),
            Arc::new(StringArray::from_iter_values(&uris)),
            Arc::new(UInt64Array::from(vec![0; n])),
            Arc::new(UInt64Array::from_iter_values(
                blobs.iter().map(|b| b.size.0),
            )),
            Arc::new(StringArray::from_iter_values(
                blobs.iter().map(|b| b.media_type.as_str()),
            )),
            Arc::new(StringArray::from(vec![None::<&str>; n])),
            Arc::new(StringArray::from(vec![None::<&str>; n])),
            Arc::new(BinaryArray::from_iter_values(
                digests.iter().map(Vec::as_slice),
            )),
        ];
        let DataType::Struct(fields) = Self::field("blob").data_type().clone() else {
            unreachable!()
        };
        Ok(Arc::new(StructArray::new(fields, columns, None)))
    }
}

impl BlobUpload {
    /// The uploaded object, relative to its Table's root.
    pub fn key(&self) -> String {
        object_key(Some(self.attempt), self.object_id)
    }

    /// The blob this upload stores once complete.
    pub fn blob(&self, size: u64, digest: [u8; 32]) -> Blob {
        Blob {
            attempt: Some(self.attempt),
            ..Blob::new(
                self.table_id,
                self.object_id,
                size,
                digest,
                self.media_type.clone(),
            )
        }
    }
}

fn object_key(attempt: Option<Uuid>, object_id: Uuid) -> String {
    match attempt {
        Some(attempt) => format!("attempts/{attempt}/blobs/{object_id}"),
        None => format!("blobs/{object_id}"),
    }
}

/// A BLAKE3 digest or chaining value as lowercase hexadecimal, as blobs carry them.
pub fn hex_digest(digest: &[u8; 32]) -> String {
    digest.iter().map(|byte| format!("{byte:02x}")).collect()
}

/// A BLAKE3 digest or chaining value from its 64 hexadecimal digits.
pub fn parse_digest(value: &str) -> Result<[u8; 32], ScandalError> {
    let invalid = || ScandalError::InvalidArgument {
        message: "a blob digest is 64 hexadecimal digits".to_owned(),
    };
    let (pairs, []) = value.as_bytes().as_chunks::<2>() else {
        return Err(invalid());
    };
    if pairs.len() != 32 {
        return Err(invalid());
    }
    let mut digest = [0; 32];
    for (byte, pair) in digest.iter_mut().zip(pairs) {
        let pair = str::from_utf8(pair).map_err(|_| invalid())?;
        *byte = u8::from_str_radix(pair, 16).map_err(|_| invalid())?;
    }
    Ok(digest)
}