#[cfg(any(test, feature = "conformance-state"))]
pub mod conformance;
#[cfg(all(feature = "state-memory", not(target_arch = "wasm32")))]
pub mod memory;
#[cfg(all(feature = "state-file", not(target_arch = "wasm32")))]
pub mod file;
use async_trait::async_trait;
use base64::Engine;
use chrono::{DateTime, Utc};
use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
use std::borrow::Cow;
use std::collections::HashMap;
use std::iter::FromIterator;
use crate::error::{Error, Result};
use crate::runtime::MaybeSendSync;
use crate::storage::StorageHandle;
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
pub trait StateStore: MaybeSendSync {
fn name(&self) -> &'static str;
async fn set(&self, state: &UploadState, mode: WriteMode) -> Result<()>;
async fn get(&self, id: &str) -> Result<Option<UploadState>>;
async fn delete(&self, id: &str) -> Result<()>;
async fn list_expired(&self, before: DateTime<Utc>) -> Result<Vec<String>>;
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum WriteMode {
CreateNew,
Update,
}
#[cfg_attr(not(target_arch = "wasm32"), async_trait)]
#[cfg_attr(target_arch = "wasm32", async_trait(?Send))]
pub trait UploadInventory: MaybeSendSync {
async fn list_upload_ids(&self, limit: usize, offset: usize) -> Result<Vec<String>>;
}
pub const UPLOAD_STATE_SCHEMA_VERSION: u32 = 1;
fn default_schema_version() -> u32 {
1
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UploadState {
#[serde(default = "default_schema_version")]
schema_version: u32,
id: String,
offset: u64,
length: Option<u64>,
storage_key: Option<String>,
created_at: DateTime<Utc>,
expires_at: Option<DateTime<Utc>>,
is_partial: bool,
is_final: bool,
parts: Option<Vec<String>>,
metadata: UploadMetadata,
#[serde(default)]
internal: HashMap<String, String>,
}
impl UploadState {
pub fn new(id: impl Into<String>) -> Self {
Self {
schema_version: UPLOAD_STATE_SCHEMA_VERSION,
id: id.into(),
offset: 0,
length: None,
storage_key: None,
created_at: Utc::now(),
expires_at: None,
is_partial: false,
is_final: false,
parts: None,
metadata: UploadMetadata::new(),
internal: HashMap::new(),
}
}
pub fn new_random() -> Self {
Self::new(uuid::Uuid::new_v4().to_string())
}
pub fn schema_version(&self) -> u32 {
self.schema_version
}
#[must_use]
pub fn with_length(mut self, length: u64) -> Self {
self.length = Some(length);
self
}
#[must_use]
pub fn with_expiration(mut self, expires_at: DateTime<Utc>) -> Self {
self.expires_at = Some(expires_at);
self
}
#[must_use]
pub fn with_metadata(mut self, metadata: impl Into<UploadMetadata>) -> Self {
self.metadata = metadata.into();
self
}
#[must_use]
pub fn with_partial(mut self) -> Self {
self.is_partial = true;
self
}
#[must_use]
pub fn with_final(mut self, parts: Vec<String>) -> Self {
self.is_final = true;
self.parts = Some(parts);
self
}
pub fn id(&self) -> &str {
&self.id
}
pub fn offset(&self) -> u64 {
self.offset
}
pub(crate) fn set_offset(&mut self, offset: u64) {
self.offset = offset;
}
pub fn length(&self) -> Option<u64> {
self.length
}
pub(crate) fn set_length(&mut self, length: u64) {
self.length = Some(length);
}
pub fn storage_handle(&self) -> Option<StorageHandle> {
self.storage_key
.clone()
.map(|key| StorageHandle::from_parts(key, self.internal.clone()))
}
pub(crate) fn require_storage_handle(&self) -> Result<StorageHandle> {
self.storage_handle().ok_or(Error::StorageKeyMissing)
}
pub fn set_storage_handle(&mut self, handle: StorageHandle) {
let (key, internal) = handle.into_parts();
self.storage_key = Some(key);
self.internal = internal;
}
pub fn created_at(&self) -> &DateTime<Utc> {
&self.created_at
}
pub fn expires_at(&self) -> Option<&DateTime<Utc>> {
self.expires_at.as_ref()
}
pub(crate) fn set_expiration(&mut self, expires_at: DateTime<Utc>) {
self.expires_at = Some(expires_at);
}
pub fn is_partial(&self) -> bool {
self.is_partial
}
pub(crate) fn mark_partial(&mut self) {
self.is_partial = true;
}
pub fn is_final(&self) -> bool {
self.is_final
}
pub(crate) fn mark_final(&mut self, parts: Vec<String>) {
self.is_final = true;
self.parts = Some(parts);
}
pub fn parts(&self) -> Option<&[String]> {
self.parts.as_deref()
}
pub fn metadata(&self) -> &UploadMetadata {
&self.metadata
}
pub fn metadata_mut(&mut self) -> &mut UploadMetadata {
&mut self.metadata
}
pub fn set_metadata(&mut self, metadata: impl Into<UploadMetadata>) {
self.metadata = metadata.into();
}
pub fn is_complete(&self) -> bool {
match self.length {
Some(length) => self.offset >= length,
None => false, }
}
pub fn expires_before(&self, before: DateTime<Utc>) -> bool {
crate::expiration::ProtocolExpiration::for_upload(self).expires_before(before)
}
pub fn is_expired(&self) -> bool {
crate::expiration::ProtocolExpiration::for_upload(self).is_expired()
}
pub fn remaining(&self) -> Option<u64> {
self.length.map(|len| len.saturating_sub(self.offset))
}
pub fn expires_header(&self) -> Option<String> {
crate::expiration::ProtocolExpiration::for_upload(self).header()
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
pub struct UploadMetadata(HashMap<String, MetadataValue>);
impl UploadMetadata {
pub fn new() -> Self {
Self(HashMap::new())
}
pub fn insert(
&mut self,
key: impl Into<String>,
value: impl Into<MetadataValue>,
) -> Option<MetadataValue> {
self.0.insert(key.into(), value.into())
}
pub fn get(&self, key: &str) -> Option<&MetadataValue> {
self.0.get(key)
}
pub fn contains_key(&self, key: &str) -> bool {
self.0.contains_key(key)
}
pub fn remove(&mut self, key: &str) -> Option<MetadataValue> {
self.0.remove(key)
}
pub fn is_empty(&self) -> bool {
self.0.is_empty()
}
pub fn len(&self) -> usize {
self.0.len()
}
pub fn iter(&self) -> impl Iterator<Item = (&String, &MetadataValue)> {
self.0.iter()
}
pub fn into_inner(self) -> HashMap<String, MetadataValue> {
self.0
}
}
impl From<HashMap<String, MetadataValue>> for UploadMetadata {
fn from(metadata: HashMap<String, MetadataValue>) -> Self {
Self(metadata)
}
}
impl From<&HashMap<String, MetadataValue>> for UploadMetadata {
fn from(metadata: &HashMap<String, MetadataValue>) -> Self {
metadata.clone().into()
}
}
impl From<HashMap<String, String>> for UploadMetadata {
fn from(metadata: HashMap<String, String>) -> Self {
metadata
.into_iter()
.map(|(key, value)| (key, MetadataValue::from(value)))
.collect()
}
}
impl From<&HashMap<String, String>> for UploadMetadata {
fn from(metadata: &HashMap<String, String>) -> Self {
metadata.clone().into()
}
}
impl From<&UploadMetadata> for UploadMetadata {
fn from(metadata: &UploadMetadata) -> Self {
metadata.clone()
}
}
impl FromIterator<(String, MetadataValue)> for UploadMetadata {
fn from_iter<T: IntoIterator<Item = (String, MetadataValue)>>(iter: T) -> Self {
Self(HashMap::from_iter(iter))
}
}
impl FromIterator<(String, String)> for UploadMetadata {
fn from_iter<T: IntoIterator<Item = (String, String)>>(iter: T) -> Self {
iter.into_iter()
.map(|(key, value)| (key, MetadataValue::from(value)))
.collect::<HashMap<_, _>>()
.into()
}
}
impl IntoIterator for UploadMetadata {
type Item = (String, MetadataValue);
type IntoIter = std::collections::hash_map::IntoIter<String, MetadataValue>;
fn into_iter(self) -> Self::IntoIter {
self.0.into_iter()
}
}
impl<'a> IntoIterator for &'a UploadMetadata {
type Item = (&'a String, &'a MetadataValue);
type IntoIter = std::collections::hash_map::Iter<'a, String, MetadataValue>;
fn into_iter(self) -> Self::IntoIter {
self.0.iter()
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq, Hash)]
pub struct MetadataValue(Vec<u8>);
impl MetadataValue {
pub fn new(bytes: impl Into<Vec<u8>>) -> Self {
Self(bytes.into())
}
pub fn as_bytes(&self) -> &[u8] {
&self.0
}
pub fn into_bytes(self) -> Vec<u8> {
self.0
}
pub fn as_str(&self) -> Option<&str> {
std::str::from_utf8(&self.0).ok()
}
pub fn to_string_lossy(&self) -> Cow<'_, str> {
String::from_utf8_lossy(&self.0)
}
}
impl From<Vec<u8>> for MetadataValue {
fn from(bytes: Vec<u8>) -> Self {
Self::new(bytes)
}
}
impl From<&[u8]> for MetadataValue {
fn from(bytes: &[u8]) -> Self {
Self::new(bytes.to_vec())
}
}
impl From<String> for MetadataValue {
fn from(value: String) -> Self {
Self::new(value.into_bytes())
}
}
impl From<&str> for MetadataValue {
fn from(value: &str) -> Self {
Self::new(value.as_bytes().to_vec())
}
}
impl Serialize for MetadataValue {
fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
where
S: Serializer,
{
if let Some(value) = self.as_str() {
return serializer.serialize_str(value);
}
use serde::ser::SerializeMap;
let mut map = serializer.serialize_map(Some(1))?;
map.serialize_entry(
"base64",
&base64::engine::general_purpose::STANDARD.encode(&self.0),
)?;
map.end()
}
}
impl<'de> Deserialize<'de> for MetadataValue {
fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
#[serde(untagged)]
enum PersistedMetadataValue {
Text(String),
Binary { base64: String },
}
match PersistedMetadataValue::deserialize(deserializer)? {
PersistedMetadataValue::Text(value) => Ok(Self::from(value)),
PersistedMetadataValue::Binary { base64 } => {
let bytes = base64::engine::general_purpose::STANDARD
.decode(base64.as_bytes())
.map_err(de::Error::custom)?;
Ok(Self(bytes))
}
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_upload_state_new() {
let state = UploadState::new("test-id");
assert_eq!(state.id(), "test-id");
assert_eq!(state.offset(), 0);
assert!(state.length().is_none());
assert!(!state.is_partial());
assert!(!state.is_final());
}
#[test]
fn legacy_state_without_schema_version_deserializes_as_v1() {
let mut value = serde_json::to_value(UploadState::new("upload-1")).unwrap();
value
.as_object_mut()
.unwrap()
.remove("schema_version")
.expect("serialized state should include schema_version");
let state: UploadState = serde_json::from_value(value).unwrap();
assert_eq!(state.schema_version(), 1);
}
#[test]
fn serialized_state_includes_current_schema_version() {
let value = serde_json::to_value(UploadState::new("upload-1")).unwrap();
assert_eq!(
value.get("schema_version").and_then(|v| v.as_u64()),
Some(u64::from(UPLOAD_STATE_SCHEMA_VERSION))
);
}
#[test]
fn test_upload_state_new_random() {
let state = UploadState::new_random();
assert!(!state.id().is_empty());
assert_eq!(state.id().len(), 36); }
#[test]
fn test_upload_state_builder() {
let mut metadata = HashMap::new();
metadata.insert("filename".to_string(), MetadataValue::from("test.txt"));
let state = UploadState::new("test")
.with_length(1024)
.with_metadata(metadata);
assert_eq!(state.length(), Some(1024));
assert_eq!(
state.metadata().get("filename").and_then(|v| v.as_str()),
Some("test.txt")
);
}
#[test]
fn upload_metadata_exposes_intentional_escape_hatch() {
let metadata =
UploadMetadata::from_iter([("filename".to_string(), MetadataValue::from("test.txt"))]);
let inner = metadata.into_inner();
assert_eq!(
inner.get("filename").and_then(|v| v.as_str()),
Some("test.txt")
);
}
#[test]
fn upload_metadata_clones_from_borrowed_upload_metadata() {
let mut metadata = UploadMetadata::new();
metadata.insert("filename", "test.txt");
let cloned = UploadMetadata::from(&metadata);
assert_eq!(cloned, metadata);
}
#[test]
fn upload_metadata_converts_from_borrowed_text_map() {
let metadata = HashMap::from([("filename".to_string(), "test.txt".to_string())]);
let converted = UploadMetadata::from(&metadata);
assert_eq!(
converted.get("filename").and_then(|value| value.as_str()),
Some("test.txt")
);
}
#[test]
fn upload_metadata_converts_from_borrowed_binary_map() {
let metadata = HashMap::from([("bin".to_string(), MetadataValue::from(&b"\xFF\xFE"[..]))]);
let converted = UploadMetadata::from(&metadata);
assert_eq!(converted.get("bin").unwrap().as_bytes(), b"\xFF\xFE");
}
#[test]
fn metadata_value_deserializes_legacy_text_json() {
let value: MetadataValue = serde_json::from_str(r#""test.txt""#).unwrap();
assert_eq!(value.as_bytes(), b"test.txt");
}
#[test]
fn metadata_value_serializes_text_as_plain_json_string() {
let json = serde_json::to_string(&MetadataValue::from("test.txt")).unwrap();
assert_eq!(json, r#""test.txt""#);
}
#[test]
fn metadata_value_serializes_binary_as_base64_object() {
let json = serde_json::to_string(&MetadataValue::from(vec![0xFF, 0xFE, 0xFD])).unwrap();
assert_eq!(json, r#"{"base64":"//79"}"#);
let decoded: MetadataValue = serde_json::from_str(&json).unwrap();
assert_eq!(decoded.as_bytes(), [0xFF, 0xFE, 0xFD]);
}
#[test]
fn test_is_complete() {
let mut state = UploadState::new("test").with_length(100);
assert!(!state.is_complete());
state.set_offset(50);
assert!(!state.is_complete());
state.set_offset(100);
assert!(state.is_complete());
let deferred = UploadState::new("test2");
assert!(!deferred.is_complete());
}
#[test]
fn test_is_expired() {
let state = UploadState::new("test");
assert!(!state.is_expired());
let expired =
UploadState::new("test2").with_expiration(Utc::now() - chrono::Duration::hours(1));
assert!(expired.is_expired());
let future =
UploadState::new("test3").with_expiration(Utc::now() + chrono::Duration::hours(1));
assert!(!future.is_expired());
}
#[test]
fn test_remaining() {
let mut state = UploadState::new("test").with_length(1000);
assert_eq!(state.remaining(), Some(1000));
state.set_offset(300);
assert_eq!(state.remaining(), Some(700));
state.set_offset(1000);
assert_eq!(state.remaining(), Some(0));
let deferred = UploadState::new("test2");
assert_eq!(deferred.remaining(), None);
}
#[test]
fn test_serialization() {
let state = UploadState::new("test-id").with_length(1024).with_partial();
let json = serde_json::to_string(&state).unwrap();
let deserialized: UploadState = serde_json::from_str(&json).unwrap();
assert_eq!(deserialized.id(), "test-id");
assert_eq!(deserialized.length(), Some(1024));
assert!(deserialized.is_partial());
}
#[test]
fn test_partial_and_final() {
let partial = UploadState::new("part1").with_partial();
assert!(partial.is_partial());
assert!(!partial.is_final());
let final_upload =
UploadState::new("final").with_final(vec!["part1".to_string(), "part2".to_string()]);
assert!(!final_upload.is_partial());
assert!(final_upload.is_final());
assert_eq!(
final_upload.parts(),
Some(vec!["part1".to_string(), "part2".to_string()]).as_deref()
);
}
#[test]
fn test_expires_header_rfc7231_format() {
use chrono::TimeZone;
let dt = Utc.with_ymd_and_hms(2025, 6, 25, 14, 30, 0).unwrap();
let state = UploadState::new("test").with_expiration(dt);
let header = state.expires_header().unwrap();
assert_eq!(header, "Wed, 25 Jun 2025 14:30:00 GMT");
let dt2 = Utc.with_ymd_and_hms(1994, 11, 6, 8, 49, 37).unwrap();
let state2 = UploadState::new("test2").with_expiration(dt2);
let header2 = state2.expires_header().unwrap();
assert_eq!(header2, "Sun, 06 Nov 1994 08:49:37 GMT");
}
#[test]
fn test_expires_header_none_when_no_expiration() {
let state = UploadState::new("test");
assert!(state.expires_header().is_none());
}
#[test]
fn completed_upload_expiration_policy_distinguishes_deliverable_and_partial_uploads() {
use chrono::TimeZone;
let expires_at = Utc.with_ymd_and_hms(2030, 6, 25, 14, 30, 0).unwrap();
let mut deliverable = UploadState::new("deliverable")
.with_length(5)
.with_expiration(expires_at);
deliverable.set_offset(5);
let mut partial = UploadState::new("partial")
.with_length(5)
.with_expiration(expires_at)
.with_partial();
partial.set_offset(5);
assert_eq!(deliverable.expires_header(), None);
assert_eq!(
partial.expires_header().as_deref(),
Some("Tue, 25 Jun 2030 14:30:00 GMT")
);
}
}