1use crate::{Checksum, ChecksumAlgorithm, ContentRef, NamespaceId, UploadId};
4use serde::{Deserialize, Serialize};
5use std::collections::BTreeMap;
6
7#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
9#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
10#[serde(deny_unknown_fields)]
11pub struct UploadContentClaim {
12 pub size_bytes: u64,
14 pub checksum: Checksum,
16}
17
18#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
20#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
21#[serde(rename_all = "snake_case")]
22pub enum UploadMode {
23 #[default]
25 ServiceProxied,
26 DirectPut,
28 DirectMultipart,
30}
31
32impl UploadMode {
33 pub fn as_str(self) -> &'static str {
35 match self {
36 Self::ServiceProxied => "service_proxied",
37 Self::DirectPut => "direct_put",
38 Self::DirectMultipart => "direct_multipart",
39 }
40 }
41}
42
43#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
45#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
46#[serde(tag = "mode", rename_all = "snake_case", deny_unknown_fields)]
47pub enum CreateUploadBody {
48 #[cfg_attr(feature = "openapi", schema(title = "CreateUploadBodyServiceProxied"))]
52 ServiceProxied {},
53 #[cfg_attr(feature = "openapi", schema(title = "CreateUploadBodyDirectPut"))]
55 DirectPut {
56 #[serde(default, skip_serializing_if = "Option::is_none")]
58 #[cfg_attr(feature = "openapi", schema(nullable = false))]
59 size_bytes: Option<u64>,
60 },
61 #[cfg_attr(feature = "openapi", schema(title = "CreateUploadBodyDirectMultipart"))]
63 DirectMultipart {
64 #[serde(default, skip_serializing_if = "Option::is_none")]
66 #[cfg_attr(feature = "openapi", schema(nullable = false))]
67 part_size_bytes: Option<u64>,
68 },
69}
70
71impl CreateUploadBody {
72 pub fn mode(&self) -> UploadMode {
74 match self {
75 Self::ServiceProxied {} => UploadMode::ServiceProxied,
76 Self::DirectPut { .. } => UploadMode::DirectPut,
77 Self::DirectMultipart { .. } => UploadMode::DirectMultipart,
78 }
79 }
80}
81
82#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
84#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
85#[serde(tag = "kind", rename_all = "snake_case")]
86pub enum ObjectTransferAccess {
87 #[cfg_attr(
89 feature = "openapi",
90 schema(title = "ObjectTransferAccessPresignedUrl")
91 )]
92 PresignedUrl {
93 method: String,
95 url: String,
97 #[serde(default, skip_serializing_if = "BTreeMap::is_empty")]
99 headers: BTreeMap<String, String>,
100 expires_at_ms: u64,
102 },
103}
104
105#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
107#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
108#[serde(deny_unknown_fields)]
109pub struct UploadPartChecksumClaim {
110 pub part_number: u32,
112 pub checksum: Checksum,
114}
115
116#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
118#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
119#[serde(deny_unknown_fields)]
120pub struct SignUploadPartsRequest {
121 pub parts: Vec<UploadPartChecksumClaim>,
123}
124
125#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
127#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
128pub struct SignedUploadPart {
129 pub part_number: u32,
131 pub access: ObjectTransferAccess,
133}
134
135#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
137#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
138pub struct SignUploadPartsResponse {
139 pub namespace_id: NamespaceId,
141 pub upload_id: UploadId,
143 pub parts: Vec<SignedUploadPart>,
145}
146
147#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
149#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
150#[serde(deny_unknown_fields)]
151pub struct CompletedUploadPart {
152 pub part_number: u32,
154 pub etag: String,
156 pub checksum: Checksum,
158}
159
160#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
162#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
163#[serde(deny_unknown_fields)]
164pub struct ContentToken {
165 pub content_ref: ContentRef,
167 pub token: String,
169}
170
171#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
173#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
174#[serde(tag = "mode", rename_all = "snake_case", deny_unknown_fields)]
175pub enum CompleteUploadBody {
176 #[cfg_attr(
178 feature = "openapi",
179 schema(title = "CompleteUploadBodyServiceProxied")
180 )]
181 ServiceProxied {},
182 #[cfg_attr(feature = "openapi", schema(title = "CompleteUploadBodyDirectPut"))]
184 DirectPut {
185 content: UploadContentClaim,
187 },
188 #[cfg_attr(
190 feature = "openapi",
191 schema(title = "CompleteUploadBodyDirectMultipart")
192 )]
193 DirectMultipart {
194 content: UploadContentClaim,
196 parts: Vec<CompletedUploadPart>,
198 },
199}
200
201impl CompleteUploadBody {
202 pub const fn mode(&self) -> UploadMode {
204 match self {
205 Self::ServiceProxied {} => UploadMode::ServiceProxied,
206 Self::DirectPut { .. } => UploadMode::DirectPut,
207 Self::DirectMultipart { .. } => UploadMode::DirectMultipart,
208 }
209 }
210}
211
212#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
214#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
215#[serde(deny_unknown_fields)]
216pub struct CompleteMultipartUploadRequest {
217 pub content: UploadContentClaim,
219 pub parts: Vec<CompletedUploadPart>,
221}
222
223#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
227#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
228#[serde(tag = "status", rename_all = "snake_case")]
229pub enum UploadSessionStatus {
230 #[cfg_attr(feature = "openapi", schema(title = "UploadSessionStatusOpen"))]
232 Open {
233 expires_at_ms: u64,
235 #[serde(default, skip_serializing_if = "Option::is_none")]
237 #[cfg_attr(feature = "openapi", schema(nullable = false))]
238 checksum_algorithm: Option<ChecksumAlgorithm>,
239 #[serde(default, skip_serializing_if = "Option::is_none")]
241 #[cfg_attr(feature = "openapi", schema(nullable = false))]
242 part_size_bytes: Option<u64>,
243 #[serde(default, skip_serializing_if = "Option::is_none")]
245 #[cfg_attr(feature = "openapi", schema(nullable = false))]
246 access: Option<ObjectTransferAccess>,
247 #[serde(default, skip_serializing_if = "Option::is_none")]
249 #[cfg_attr(feature = "openapi", schema(nullable = false))]
250 content_ref: Option<ContentRef>,
251 },
252 #[cfg_attr(feature = "openapi", schema(title = "UploadSessionStatusCompleted"))]
254 Completed {
255 completed_at_ms: u64,
257 content_ref: ContentRef,
259 #[serde(default, skip_serializing_if = "Option::is_none")]
261 #[cfg_attr(feature = "openapi", schema(nullable = false))]
262 content_token: Option<ContentToken>,
263 },
264 #[cfg_attr(feature = "openapi", schema(title = "UploadSessionStatusAborted"))]
266 Aborted {
267 aborted_at_ms: u64,
269 },
270}
271
272#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
274#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
275pub struct UploadSession {
276 pub namespace_id: NamespaceId,
278 pub upload_id: UploadId,
280 pub mode: UploadMode,
282 #[serde(flatten)]
284 pub status: UploadSessionStatus,
285}
286
287impl UploadSession {
288 pub const fn content_ref(&self) -> Option<&ContentRef> {
290 match &self.status {
291 UploadSessionStatus::Completed { content_ref, .. } => Some(content_ref),
292 UploadSessionStatus::Open { content_ref, .. } => content_ref.as_ref(),
293 UploadSessionStatus::Aborted { .. } => None,
294 }
295 }
296
297 pub const fn content_token(&self) -> Option<&ContentToken> {
299 match &self.status {
300 UploadSessionStatus::Completed { content_token, .. } => content_token.as_ref(),
301 UploadSessionStatus::Open { .. } | UploadSessionStatus::Aborted { .. } => None,
302 }
303 }
304}
305
306#[cfg(test)]
307mod tests {
308 use super::{
309 CompleteUploadBody, ContentToken, CreateUploadBody, ObjectTransferAccess,
310 UploadContentClaim, UploadMode, UploadSession, UploadSessionStatus,
311 };
312 use crate::{Checksum, ChecksumAlgorithm, ContentId, ContentRef, NamespaceId, UploadId};
313 use std::collections::BTreeMap;
314
315 #[test]
316 fn a_create_upload_body_without_a_mode_does_not_decode() {
317 assert!(serde_json::from_str::<CreateUploadBody>("{}").is_err());
318 assert_eq!(
319 serde_json::from_str::<CreateUploadBody>(r#"{"mode":"service_proxied"}"#)
320 .expect("decode proxied begin request"),
321 CreateUploadBody::ServiceProxied {}
322 );
323 }
324
325 #[test]
326 fn a_create_upload_body_carrying_another_modes_fields_does_not_decode() {
327 for body in [
328 r#"{"mode":"service_proxied","part_size_bytes":8388608}"#,
329 r#"{"mode":"service_proxied","content":{"size_bytes":5,"checksum":{"algorithm":"sha256","value":"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"}}}"#,
330 r#"{"mode":"direct_multipart","content":{"size_bytes":5,"checksum":{"algorithm":"sha256","value":"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"}}}"#,
331 r#"{"mode":"direct_put","content":{"size_bytes":5,"checksum":{"algorithm":"sha256","value":"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"}}}"#,
332 r#"{"mode":"direct_put","part_size_bytes":8388608}"#,
333 r#"{"mode":"direct_multipart","size_bytes":5}"#,
334 ] {
335 assert!(
336 serde_json::from_str::<CreateUploadBody>(body).is_err(),
337 "decoded a begin request that mixes modes: {body}"
338 );
339 }
340 }
341
342 #[test]
343 fn a_multipart_begin_names_its_part_size_beside_the_mode() {
344 assert_eq!(
345 serde_json::from_str::<CreateUploadBody>(
346 r#"{"mode":"direct_multipart","part_size_bytes":8388608}"#
347 )
348 .expect("decode multipart begin request"),
349 CreateUploadBody::DirectMultipart {
350 part_size_bytes: Some(8 * 1024 * 1024),
351 }
352 );
353 assert_eq!(
354 serde_json::from_str::<CreateUploadBody>(r#"{"mode":"direct_multipart"}"#)
355 .expect("decode multipart begin without a part size"),
356 CreateUploadBody::DirectMultipart {
357 part_size_bytes: None,
358 }
359 );
360 assert_eq!(
361 serde_json::to_value(CreateUploadBody::DirectMultipart {
362 part_size_bytes: None,
363 })
364 .expect("serialize multipart begin request"),
365 serde_json::json!({ "mode": "direct_multipart" })
366 );
367 }
368
369 #[test]
370 fn completion_requests_are_tagged_and_mode_specific() {
371 assert_eq!(
372 serde_json::from_str::<CompleteUploadBody>(r#"{"mode":"service_proxied"}"#)
373 .expect("decode proxied completion"),
374 CompleteUploadBody::ServiceProxied {}
375 );
376 let direct_put = CompleteUploadBody::DirectPut {
377 content: UploadContentClaim {
378 size_bytes: 5,
379 checksum: Checksum::crc32c(b"hello"),
380 },
381 };
382 assert_eq!(
383 serde_json::to_value(&direct_put).expect("encode direct-put completion"),
384 serde_json::json!({
385 "mode": "direct_put",
386 "content": {
387 "size_bytes": 5,
388 "checksum": Checksum::crc32c(b"hello"),
389 },
390 })
391 );
392 for body in [
393 r#"{}"#,
394 r#"{"mode":"service_proxied","content":{"size_bytes":5,"checksum":{"algorithm":"crc64nvme","value":"0123456789abcdef"}},"parts":[]}"#,
395 r#"{"mode":"direct_put"}"#,
396 r#"{"mode":"direct_multipart"}"#,
397 ] {
398 assert!(
399 serde_json::from_str::<CompleteUploadBody>(body).is_err(),
400 "decoded an invalid completion request: {body}"
401 );
402 }
403
404 let missing_parts = r#"{"mode":"direct_multipart","content":{"size_bytes":5,"checksum":{"algorithm":"crc64nvme","value":"0123456789abcdef"}}}"#;
405 let error = serde_json::from_str::<CompleteUploadBody>(missing_parts)
406 .expect_err("multipart parts are required");
407 assert!(
408 error.to_string().contains("parts"),
409 "the rejection should name the missing field: {error}"
410 );
411
412 let multipart = CompleteUploadBody::DirectMultipart {
413 content: UploadContentClaim {
414 size_bytes: 5,
415 checksum: Checksum::crc64nvme(b"hello"),
416 },
417 parts: Vec::new(),
418 };
419 let encoded = serde_json::to_string(&multipart).expect("encode multipart completion");
420 assert_eq!(
421 serde_json::from_str::<serde_json::Value>(&encoded).expect("decode multipart JSON"),
422 serde_json::json!({
423 "mode": "direct_multipart",
424 "content": {
425 "size_bytes": 5,
426 "checksum": Checksum::crc64nvme(b"hello"),
427 },
428 "parts": [],
429 })
430 );
431 }
432
433 #[test]
434 fn open_sessions_carry_only_their_modes_fields() {
435 let content_ref = ContentRef::blob_v1(
436 NamespaceId::parse("demo").expect("namespace id"),
437 ContentId::generate(),
438 b"hello",
439 );
440 let access = ObjectTransferAccess::PresignedUrl {
441 method: "PUT".to_owned(),
442 url: "https://bucket.example/object".to_owned(),
443 headers: BTreeMap::new(),
444 expires_at_ms: 1,
445 };
446 for (mode, checksum_algorithm, part_size_bytes, access, content_ref, fields) in [
447 (
448 UploadMode::DirectPut,
449 Some(ChecksumAlgorithm::Crc64nvme),
450 None,
451 Some(access.clone()),
452 None,
453 serde_json::json!({"checksum_algorithm": "crc64nvme", "access": access}),
454 ),
455 (
456 UploadMode::DirectMultipart,
457 Some(ChecksumAlgorithm::Crc64nvme),
458 Some(8388608),
459 None,
460 None,
461 serde_json::json!({"checksum_algorithm": "crc64nvme", "part_size_bytes": 8388608}),
462 ),
463 (
464 UploadMode::ServiceProxied,
465 None,
466 None,
467 None,
468 Some(content_ref.clone()),
469 serde_json::json!({"content_ref": content_ref}),
470 ),
471 (
472 UploadMode::ServiceProxied,
473 None,
474 None,
475 None,
476 None,
477 serde_json::json!({}),
478 ),
479 ] {
480 let session = UploadSession {
481 namespace_id: NamespaceId::parse("demo").expect("namespace id"),
482 upload_id: UploadId::parse("upl_00000000000000000000000000000001")
483 .expect("upload id"),
484 mode,
485 status: UploadSessionStatus::Open {
486 expires_at_ms: 1000,
487 checksum_algorithm,
488 part_size_bytes,
489 access,
490 content_ref,
491 },
492 };
493 let mut expected = serde_json::json!({
494 "namespace_id": "demo",
495 "upload_id": "upl_00000000000000000000000000000001",
496 "mode": mode,
497 "status": "open",
498 "expires_at_ms": 1000,
499 });
500 expected
501 .as_object_mut()
502 .expect("session object")
503 .extend(fields.as_object().expect("mode fields").clone());
504 assert_eq!(
505 serde_json::to_value(&session).expect("serialize session"),
506 expected
507 );
508 assert_eq!(
509 serde_json::from_value::<UploadSession>(expected).expect("decode session"),
510 session
511 );
512 }
513 }
514
515 #[test]
516 fn an_upload_content_claim_names_only_size_and_checksum() {
517 let request: CreateUploadBody =
518 serde_json::from_str(r#"{"mode":"direct_put","size_bytes":5}"#)
519 .expect("decode direct-put begin request");
520 assert_eq!(
521 request,
522 CreateUploadBody::DirectPut {
523 size_bytes: Some(5),
524 }
525 );
526 assert_eq!(
527 serde_json::from_str::<CreateUploadBody>(r#"{"mode":"direct_put"}"#)
528 .expect("decode direct-put begin without a size"),
529 CreateUploadBody::DirectPut { size_bytes: None }
530 );
531
532 assert!(
533 serde_json::from_str::<UploadContentClaim>(
534 r#"{"size_bytes":5,"checksum":{"algorithm":"sha256","value":"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"},"content_id":"con_0123456789abcdef0123456789abcdef"}"#
535 )
536 .is_err(),
537 "a client must not be able to name the content object"
538 );
539 }
540
541 #[test]
542 fn an_upload_session_is_flat_and_uses_one_status_vocabulary() {
543 let namespace_id = NamespaceId::parse("demo").expect("namespace id");
544 let upload_id = UploadId::parse("upl_00000000000000000000000000000001").expect("upload id");
545 let aborted = serde_json::to_value(UploadSession {
546 namespace_id: namespace_id.clone(),
547 upload_id: upload_id.clone(),
548 mode: UploadMode::ServiceProxied,
549 status: UploadSessionStatus::Aborted {
550 aborted_at_ms: 2_000,
551 },
552 })
553 .expect("serialize aborted status");
554 assert_eq!(aborted["status"], "aborted");
555 assert_eq!(aborted["mode"], "service_proxied");
556 assert_eq!(aborted["aborted_at_ms"], 2_000);
557 assert!(aborted.get("state").is_none());
558
559 let completed = serde_json::to_value(UploadSession {
560 namespace_id,
561 upload_id,
562 mode: UploadMode::DirectPut,
563 status: UploadSessionStatus::Completed {
564 completed_at_ms: 3_000,
565 content_ref: ContentRef::blob_v1(
566 crate::NamespaceId::parse("demo").expect("namespace id"),
567 ContentId::generate(),
568 b"hello",
569 ),
570 content_token: None,
571 },
572 })
573 .expect("serialize completed status");
574 assert_eq!(completed["status"], "completed");
575 assert_eq!(completed["mode"], "direct_put");
576 assert!(completed.get("state").is_none());
577 assert!(completed.get("status").is_some());
578 assert!(
579 completed.get("content_token").is_none(),
580 "a session past its receipt window reports no token at all"
581 );
582 }
583
584 #[test]
585 fn completion_status_and_commit_share_the_exact_content_token_shape() {
586 let namespace_id = NamespaceId::parse("demo").expect("namespace id");
587 let upload_id = UploadId::parse("upl_00000000000000000000000000000001").expect("upload id");
588 let content_ref = ContentRef::blob_v1(
589 crate::NamespaceId::parse("demo").expect("namespace id"),
590 ContentId::parse("con_0123456789abcdef0123456789abcdef").expect("content id"),
591 b"hello",
592 );
593 let content_token = ContentToken {
594 content_ref: content_ref.clone(),
595 token: "opaque-server-token".to_owned(),
596 };
597 let completion = serde_json::to_value(UploadSession {
598 namespace_id: namespace_id.clone(),
599 upload_id,
600 mode: UploadMode::ServiceProxied,
601 status: UploadSessionStatus::Completed {
602 completed_at_ms: 3_000,
603 content_ref: content_ref.clone(),
604 content_token: Some(content_token.clone()),
605 },
606 })
607 .expect("serialize completion");
608 let status = serde_json::to_value(UploadSessionStatus::Completed {
609 completed_at_ms: 3_000,
610 content_ref,
611 content_token: Some(content_token),
612 })
613 .expect("serialize completed status");
614
615 let completion_token = completion["content_token"].clone();
616 let status_token = status["content_token"].clone();
617 assert_eq!(completion_token, status_token);
618 assert_eq!(
619 completion_token,
620 serde_json::json!({
621 "content_ref": {
622 "kind": "blob_v1",
623 "owner_namespace_id": "demo",
624 "content_id": "con_0123456789abcdef0123456789abcdef",
625 "size_bytes": 5,
626 "checksum": {
627 "algorithm": "sha256",
628 "value": "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
629 }
630 },
631 "token": "opaque-server-token"
632 })
633 );
634
635 let request: crate::v0::CommitRequest = serde_json::from_value(serde_json::json!({
636 "commit_id": "same-token-shape",
637 "content_tokens": [completion_token],
638 "operations": [{
639 "kind": "create_directory",
640 "path": "/proof",
641 "parents": false
642 }]
643 }))
644 .expect("completion token decodes unchanged in a commit request");
645 assert_eq!(
646 serde_json::to_value(&request.content_tokens[0]).expect("serialize commit token"),
647 status_token
648 );
649 }
650
651 #[test]
652 fn a_content_token_rejects_unknown_fields() {
653 let token = serde_json::json!({
654 "content_ref": {
655 "kind": "blob_v1",
656 "owner_namespace_id": "demo",
657 "content_id": "con_0123456789abcdef0123456789abcdef",
658 "size_bytes": 5,
659 "checksum": {
660 "algorithm": "sha256",
661 "value": "2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
662 }
663 },
664 "token": "opaque-server-token",
665 "expires_at_ms": 1
666 });
667 assert!(serde_json::from_value::<ContentToken>(token).is_err());
668 }
669}