Skip to main content

oxigdal_streaming/cloud/
object_store.rs

1//! Cloud object storage abstractions: URL parsing, byte-range requests,
2//! credentials, presigned URLs, multipart upload, and range coalescing.
3
4use std::collections::HashMap;
5use thiserror::Error;
6
7// ─────────────────────────────────────────────────────────────────────────────
8// CloudError
9// ─────────────────────────────────────────────────────────────────────────────
10
11/// Errors produced by the cloud I/O layer.
12#[derive(Debug, Error, Clone, PartialEq, Eq)]
13pub enum CloudError {
14    /// The URL string could not be parsed.
15    #[error("invalid URL: {0}")]
16    InvalidUrl(String),
17
18    /// The URL scheme is not supported.
19    #[error("unsupported scheme: {0}")]
20    UnsupportedScheme(String),
21
22    /// Credentials are required but were not provided.
23    #[error("missing credentials")]
24    MissingCredentials,
25
26    /// The supplied credentials are malformed or expired.
27    #[error("invalid credentials: {0}")]
28    InvalidCredentials(String),
29
30    /// An error occurred while generating a presigned URL.
31    #[error("presign error: {0}")]
32    PresignError(String),
33
34    /// The requested byte range exceeds the object size.
35    #[error("range out of bounds: [{start}, {end}) vs size {size}")]
36    RangeOutOfBounds {
37        /// Start offset of the requested range.
38        start: u64,
39        /// End offset (exclusive) of the requested range.
40        end: u64,
41        /// Actual size of the object.
42        size: u64,
43    },
44
45    /// An HTTP error was returned by the remote server.
46    #[error("HTTP error: status {status}, url: {url}")]
47    HttpError {
48        /// The HTTP status code.
49        status: u16,
50        /// The URL that returned the error.
51        url: String,
52    },
53
54    /// The HTTP client encountered a network or protocol error.
55    #[error("HTTP transport error: {0}")]
56    TransportError(String),
57
58    /// The remote exhausted all retry attempts.
59    #[error("exhausted all {attempts} retry attempts")]
60    RetryExhausted {
61        /// How many attempts were made.
62        attempts: u32,
63    },
64
65    /// The credential type is not supported for the requested operation.
66    #[error("unsupported credentials: {0}")]
67    UnsupportedCredentials(String),
68
69    /// A required response header was missing or malformed.
70    #[error("missing or malformed header: {0}")]
71    MissingHeader(String),
72}
73
74// ─────────────────────────────────────────────────────────────────────────────
75// CloudScheme / ObjectUrl
76// ─────────────────────────────────────────────────────────────────────────────
77
78/// Supported cloud URL schemes.
79#[derive(Debug, Clone, PartialEq, Eq)]
80pub enum CloudScheme {
81    /// Amazon S3 (`s3://`)
82    S3,
83    /// Google Cloud Storage (`gs://`)
84    Gs,
85    /// Azure Blob Storage (`az://` or `abfs://`)
86    Az,
87    /// Plain HTTP (`http://`)
88    Http,
89    /// HTTPS (`https://`)
90    Https,
91}
92
93/// A parsed cloud object URL.
94#[derive(Debug, Clone, PartialEq, Eq)]
95pub struct ObjectUrl {
96    /// Scheme of the URL.
97    pub scheme: CloudScheme,
98    /// Bucket or container name.
99    pub bucket: String,
100    /// Object key / path within the bucket.
101    pub key: String,
102    /// AWS / GCS region (if present in the URL).
103    pub region: Option<String>,
104    /// Custom endpoint override.
105    pub endpoint: Option<String>,
106}
107
108impl ObjectUrl {
109    /// Parse a cloud URL string into an [`ObjectUrl`].
110    ///
111    /// Supported forms:
112    /// - `s3://bucket/key`
113    /// - `gs://bucket/key`
114    /// - `az://container/blob`
115    /// - `abfs://container@account.dfs.core.windows.net/path`
116    /// - `http://host/path`
117    /// - `https://host/path`
118    pub fn parse(url: &str) -> Result<Self, CloudError> {
119        let (scheme_str, rest) = url
120            .split_once("://")
121            .ok_or_else(|| CloudError::InvalidUrl(format!("no scheme separator in '{url}'")))?;
122
123        let scheme = match scheme_str.to_ascii_lowercase().as_str() {
124            "s3" => CloudScheme::S3,
125            "gs" => CloudScheme::Gs,
126            "az" | "abfs" => CloudScheme::Az,
127            "http" => CloudScheme::Http,
128            "https" => CloudScheme::Https,
129            other => return Err(CloudError::UnsupportedScheme(other.to_owned())),
130        };
131
132        match &scheme {
133            CloudScheme::Http | CloudScheme::Https => {
134                // For http(s) the "bucket" is the host and the "key" is the path.
135                let (host, path) = if let Some(idx) = rest.find('/') {
136                    (&rest[..idx], &rest[idx + 1..])
137                } else {
138                    (rest, "")
139                };
140                if host.is_empty() {
141                    return Err(CloudError::InvalidUrl(format!("no host in '{url}'")));
142                }
143                Ok(ObjectUrl {
144                    scheme,
145                    bucket: host.to_owned(),
146                    key: path.to_owned(),
147                    region: None,
148                    endpoint: None,
149                })
150            }
151            _ => {
152                // s3://, gs://, az://  →  bucket/key
153                let (bucket, key) = if let Some(idx) = rest.find('/') {
154                    (&rest[..idx], &rest[idx + 1..])
155                } else {
156                    (rest, "")
157                };
158                if bucket.is_empty() {
159                    return Err(CloudError::InvalidUrl(format!("no bucket in '{url}'")));
160                }
161                Ok(ObjectUrl {
162                    scheme,
163                    bucket: bucket.to_owned(),
164                    key: key.to_owned(),
165                    region: None,
166                    endpoint: None,
167                })
168            }
169        }
170    }
171
172    /// Convert the cloud URL to an HTTPS URL.
173    ///
174    /// - S3: `https://{bucket}.s3.{region}.amazonaws.com/{key}`
175    /// - GCS: `https://storage.googleapis.com/{bucket}/{key}`
176    /// - Azure: `https://{account}.blob.core.windows.net/{container}/{key}`
177    /// - HTTP/HTTPS: return as-is (upgraded to https for http)
178    pub fn to_https_url(&self, endpoint_override: Option<&str>) -> String {
179        if let Some(ep) = endpoint_override {
180            let base = ep.trim_end_matches('/');
181            return format!("{base}/{}/{}", self.bucket, self.key);
182        }
183        match &self.scheme {
184            CloudScheme::S3 => {
185                let region = self.region.as_deref().unwrap_or("us-east-1");
186                format!(
187                    "https://{}.s3.{}.amazonaws.com/{}",
188                    self.bucket, region, self.key
189                )
190            }
191            CloudScheme::Gs => {
192                format!(
193                    "https://storage.googleapis.com/{}/{}",
194                    self.bucket, self.key
195                )
196            }
197            CloudScheme::Az => {
198                // bucket is treated as "account/container"
199                // We store just "container" in bucket; account may be in endpoint.
200                format!("https://{}.blob.core.windows.net/{}", self.bucket, self.key)
201            }
202            CloudScheme::Http => {
203                format!("https://{}/{}", self.bucket, self.key)
204            }
205            CloudScheme::Https => {
206                format!("https://{}/{}", self.bucket, self.key)
207            }
208        }
209    }
210
211    /// Build the canonical host string used when signing requests.
212    pub fn signing_host(&self) -> String {
213        match &self.scheme {
214            CloudScheme::S3 => {
215                let region = self.region.as_deref().unwrap_or("us-east-1");
216                format!("{}.s3.{}.amazonaws.com", self.bucket, region)
217            }
218            CloudScheme::Gs => "storage.googleapis.com".to_owned(),
219            CloudScheme::Az => {
220                format!("{}.blob.core.windows.net", self.bucket)
221            }
222            CloudScheme::Http | CloudScheme::Https => self.bucket.clone(),
223        }
224    }
225
226    /// Return the URL path component used for signing.
227    pub fn signing_path(&self) -> String {
228        let key = if self.key.starts_with('/') {
229            self.key.clone()
230        } else {
231            format!("/{}", self.key)
232        };
233        match &self.scheme {
234            CloudScheme::Gs | CloudScheme::Az => {
235                format!("/{}{}", self.bucket, key)
236            }
237            _ => key,
238        }
239    }
240}
241
242// ─────────────────────────────────────────────────────────────────────────────
243// ByteRangeRequest
244// ─────────────────────────────────────────────────────────────────────────────
245
246/// A request for a specific byte range of a cloud object.
247#[derive(Debug, Clone, PartialEq, Eq)]
248pub struct ByteRangeRequest {
249    /// The object URL to read from.
250    pub url: ObjectUrl,
251    /// Byte range (exclusive end).
252    pub range: std::ops::Range<u64>,
253}
254
255impl ByteRangeRequest {
256    /// Create a new byte-range request.
257    pub fn new(url: ObjectUrl, start: u64, end: u64) -> Self {
258        ByteRangeRequest {
259            url,
260            range: start..end,
261        }
262    }
263
264    /// Return the HTTP `Range` header value: `bytes=start-end_inclusive`.
265    pub fn to_http_range_header(&self) -> String {
266        let end_inclusive = self.range.end.saturating_sub(1);
267        format!("bytes={}-{}", self.range.start, end_inclusive)
268    }
269
270    /// Number of bytes in this range.
271    pub fn length(&self) -> u64 {
272        self.range.end.saturating_sub(self.range.start)
273    }
274}
275
276// ─────────────────────────────────────────────────────────────────────────────
277// ObjectMetadata
278// ─────────────────────────────────────────────────────────────────────────────
279
280/// Metadata for a cloud object.
281#[derive(Debug, Clone, PartialEq, Eq)]
282pub struct ObjectMetadata {
283    /// The object URL.
284    pub url: ObjectUrl,
285    /// Total size in bytes.
286    pub size: u64,
287    /// MIME content type, if available.
288    pub content_type: Option<String>,
289    /// ETag string, if available.
290    pub etag: Option<String>,
291    /// Last-modified Unix timestamp, if available.
292    pub last_modified: Option<u64>,
293    /// User-defined metadata key/value pairs.
294    pub user_metadata: HashMap<String, String>,
295}
296
297// ─────────────────────────────────────────────────────────────────────────────
298// CloudCredentials
299// ─────────────────────────────────────────────────────────────────────────────
300
301/// Authentication credentials for cloud object storage.
302#[derive(Debug, Clone, PartialEq, Eq)]
303pub enum CloudCredentials {
304    /// No authentication (public buckets).
305    Anonymous,
306    /// AWS / GCS access-key credentials.
307    AccessKey {
308        /// Access key ID.
309        access_key_id: String,
310        /// Secret access key.
311        secret_access_key: String,
312        /// Optional STS session token.
313        session_token: Option<String>,
314    },
315    /// GCS service-account JSON file path.
316    ServiceAccountFile {
317        /// Path to the service account JSON file.
318        path: String,
319    },
320    /// Azure Shared Key authentication.
321    AzureSharedKey {
322        /// Azure storage account name.
323        account_name: String,
324        /// Base64-encoded account key.
325        account_key: String,
326    },
327    /// Azure SAS token.
328    SasToken {
329        /// The SAS token string.
330        token: String,
331    },
332    /// Generic OAuth2 bearer token.
333    Bearer {
334        /// The bearer token string.
335        token: String,
336    },
337}
338
339// ─────────────────────────────────────────────────────────────────────────────
340// PresignedUrlConfig / HttpMethod
341// ─────────────────────────────────────────────────────────────────────────────
342
343/// HTTP verb used in a presigned URL.
344#[derive(Debug, Clone, PartialEq, Eq)]
345pub enum HttpMethod {
346    /// GET request.
347    Get,
348    /// PUT request.
349    Put,
350    /// DELETE request.
351    Delete,
352    /// HEAD request.
353    Head,
354}
355
356impl HttpMethod {
357    fn as_str(&self) -> &'static str {
358        match self {
359            HttpMethod::Get => "GET",
360            HttpMethod::Put => "PUT",
361            HttpMethod::Delete => "DELETE",
362            HttpMethod::Head => "HEAD",
363        }
364    }
365}
366
367/// Configuration for presigned URL generation.
368#[derive(Debug, Clone, PartialEq, Eq)]
369pub struct PresignedUrlConfig {
370    /// How many seconds the URL should remain valid.
371    pub expires_in_secs: u64,
372    /// HTTP method the presigned URL will allow.
373    pub method: HttpMethod,
374    /// Optional content-type constraint.
375    pub content_type: Option<String>,
376}
377
378impl PresignedUrlConfig {
379    /// Create a GET presigned URL configuration.
380    pub fn get(expires_in_secs: u64) -> Self {
381        PresignedUrlConfig {
382            expires_in_secs,
383            method: HttpMethod::Get,
384            content_type: None,
385        }
386    }
387
388    /// Create a PUT presigned URL configuration with a content type.
389    pub fn put(expires_in_secs: u64, content_type: impl Into<String>) -> Self {
390        PresignedUrlConfig {
391            expires_in_secs,
392            method: HttpMethod::Put,
393            content_type: Some(content_type.into()),
394        }
395    }
396}
397
398// ─────────────────────────────────────────────────────────────────────────────
399// Pure-Rust SHA-256 and HMAC-SHA256
400// ─────────────────────────────────────────────────────────────────────────────
401
402/// SHA-256 round constants.
403const K: [u32; 64] = [
404    0x428a2f98, 0x71374491, 0xb5c0fbcf, 0xe9b5dba5, 0x3956c25b, 0x59f111f1, 0x923f82a4, 0xab1c5ed5,
405    0xd807aa98, 0x12835b01, 0x243185be, 0x550c7dc3, 0x72be5d74, 0x80deb1fe, 0x9bdc06a7, 0xc19bf174,
406    0xe49b69c1, 0xefbe4786, 0x0fc19dc6, 0x240ca1cc, 0x2de92c6f, 0x4a7484aa, 0x5cb0a9dc, 0x76f988da,
407    0x983e5152, 0xa831c66d, 0xb00327c8, 0xbf597fc7, 0xc6e00bf3, 0xd5a79147, 0x06ca6351, 0x14292967,
408    0x27b70a85, 0x2e1b2138, 0x4d2c6dfc, 0x53380d13, 0x650a7354, 0x766a0abb, 0x81c2c92e, 0x92722c85,
409    0xa2bfe8a1, 0xa81a664b, 0xc24b8b70, 0xc76c51a3, 0xd192e819, 0xd6990624, 0xf40e3585, 0x106aa070,
410    0x19a4c116, 0x1e376c08, 0x2748774c, 0x34b0bcb5, 0x391c0cb3, 0x4ed8aa4a, 0x5b9cca4f, 0x682e6ff3,
411    0x748f82ee, 0x78a5636f, 0x84c87814, 0x8cc70208, 0x90befffa, 0xa4506ceb, 0xbef9a3f7, 0xc67178f2,
412];
413
414/// Initial hash values (first 32 bits of the fractional parts of the square roots of the first 8 primes).
415const H0: [u32; 8] = [
416    0x6a09e667, 0xbb67ae85, 0x3c6ef372, 0xa54ff53a, 0x510e527f, 0x9b05688c, 0x1f83d9ab, 0x5be0cd19,
417];
418
419/// Compute SHA-256 of `data`.  Pure-Rust implementation — no external crates.
420pub fn sha256(data: &[u8]) -> [u8; 32] {
421    let mut h = H0;
422
423    // Pre-processing: add padding
424    let bit_len = (data.len() as u64).wrapping_mul(8);
425    let mut msg: Vec<u8> = data.to_vec();
426    msg.push(0x80);
427    while msg.len() % 64 != 56 {
428        msg.push(0x00);
429    }
430    msg.extend_from_slice(&bit_len.to_be_bytes());
431
432    // Process each 512-bit (64-byte) block
433    for block in msg.chunks_exact(64) {
434        let mut w = [0u32; 64];
435        for (i, chunk) in block.chunks_exact(4).enumerate().take(16) {
436            w[i] = u32::from_be_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
437        }
438        for i in 16..64 {
439            let s0 = w[i - 15].rotate_right(7) ^ w[i - 15].rotate_right(18) ^ (w[i - 15] >> 3);
440            let s1 = w[i - 2].rotate_right(17) ^ w[i - 2].rotate_right(19) ^ (w[i - 2] >> 10);
441            w[i] = w[i - 16]
442                .wrapping_add(s0)
443                .wrapping_add(w[i - 7])
444                .wrapping_add(s1);
445        }
446
447        let [mut a, mut b, mut c, mut d, mut e, mut f, mut g, mut hh] = h;
448
449        for i in 0..64 {
450            let s1 = e.rotate_right(6) ^ e.rotate_right(11) ^ e.rotate_right(25);
451            let ch = (e & f) ^ ((!e) & g);
452            let temp1 = hh
453                .wrapping_add(s1)
454                .wrapping_add(ch)
455                .wrapping_add(K[i])
456                .wrapping_add(w[i]);
457            let s0 = a.rotate_right(2) ^ a.rotate_right(13) ^ a.rotate_right(22);
458            let maj = (a & b) ^ (a & c) ^ (b & c);
459            let temp2 = s0.wrapping_add(maj);
460
461            hh = g;
462            g = f;
463            f = e;
464            e = d.wrapping_add(temp1);
465            d = c;
466            c = b;
467            b = a;
468            a = temp1.wrapping_add(temp2);
469        }
470
471        h[0] = h[0].wrapping_add(a);
472        h[1] = h[1].wrapping_add(b);
473        h[2] = h[2].wrapping_add(c);
474        h[3] = h[3].wrapping_add(d);
475        h[4] = h[4].wrapping_add(e);
476        h[5] = h[5].wrapping_add(f);
477        h[6] = h[6].wrapping_add(g);
478        h[7] = h[7].wrapping_add(hh);
479    }
480
481    let mut out = [0u8; 32];
482    for (i, &word) in h.iter().enumerate() {
483        out[i * 4..(i + 1) * 4].copy_from_slice(&word.to_be_bytes());
484    }
485    out
486}
487
488/// Compute HMAC-SHA256.
489pub fn hmac_sha256(key: &[u8], data: &[u8]) -> [u8; 32] {
490    const BLOCK_SIZE: usize = 64;
491
492    // Derive the actual HMAC key (hash if longer than block size)
493    let mut k = [0u8; BLOCK_SIZE];
494    if key.len() > BLOCK_SIZE {
495        let hashed = sha256(key);
496        k[..32].copy_from_slice(&hashed);
497    } else {
498        k[..key.len()].copy_from_slice(key);
499    }
500
501    let mut ipad = [0u8; BLOCK_SIZE];
502    let mut opad = [0u8; BLOCK_SIZE];
503    for i in 0..BLOCK_SIZE {
504        ipad[i] = k[i] ^ 0x36;
505        opad[i] = k[i] ^ 0x5c;
506    }
507
508    let mut inner = ipad.to_vec();
509    inner.extend_from_slice(data);
510    let inner_hash = sha256(&inner);
511
512    let mut outer = opad.to_vec();
513    outer.extend_from_slice(&inner_hash);
514    sha256(&outer)
515}
516
517/// Compute HMAC-SHA256 and return the result as a lowercase hex string.
518pub fn hmac_sha256_hex(key: &[u8], data: &[u8]) -> String {
519    hex_encode(&hmac_sha256(key, data))
520}
521
522/// Encode a byte slice as a lowercase hexadecimal string.
523pub fn hex_encode(bytes: &[u8]) -> String {
524    bytes
525        .iter()
526        .fold(String::with_capacity(bytes.len() * 2), |mut s, b| {
527            use std::fmt::Write;
528            let _ = write!(s, "{b:02x}");
529            s
530        })
531}
532
533// ─────────────────────────────────────────────────────────────────────────────
534// PresignedUrlGenerator
535// ─────────────────────────────────────────────────────────────────────────────
536
537/// Generates presigned URLs using AWS SigV4 / GCS v4 signing.
538///
539/// The implementation is entirely pure Rust — no external cryptographic crates.
540pub struct PresignedUrlGenerator {
541    /// The credentials used for signing.
542    pub credentials: CloudCredentials,
543    /// AWS / GCS region.
544    pub region: String,
545}
546
547impl PresignedUrlGenerator {
548    /// Create a new generator.
549    pub fn new(credentials: CloudCredentials, region: impl Into<String>) -> Self {
550        PresignedUrlGenerator {
551            credentials,
552            region: region.into(),
553        }
554    }
555
556    /// Format a Unix timestamp as an AWS date string (`YYYYMMDD`).
557    fn format_date(ts: u64) -> String {
558        // Days since Unix epoch
559        let days = ts / 86_400;
560        let (year, month, day) = days_to_ymd(days);
561        format!("{year:04}{month:02}{day:02}")
562    }
563
564    /// Format a Unix timestamp as an AWS datetime string (`YYYYMMDDTHHmmSSZ`).
565    fn format_datetime(ts: u64) -> String {
566        let date = Self::format_date(ts);
567        let rem = ts % 86_400;
568        let h = rem / 3600;
569        let m = (rem % 3600) / 60;
570        let s = rem % 60;
571        format!("{date}T{h:02}{m:02}{s:02}Z")
572    }
573
574    /// Derive the AWS SigV4 signing key.
575    fn derive_signing_key(secret: &str, date: &str, region: &str, service: &str) -> [u8; 32] {
576        let k_date = hmac_sha256(format!("AWS4{secret}").as_bytes(), date.as_bytes());
577        let k_region = hmac_sha256(&k_date, region.as_bytes());
578        let k_service = hmac_sha256(&k_region, service.as_bytes());
579        hmac_sha256(&k_service, b"aws4_request")
580    }
581
582    /// Percent-encode a string using AWS URI encoding rules.
583    fn uri_encode(s: &str, encode_slash: bool) -> String {
584        let mut out = String::with_capacity(s.len());
585        for byte in s.bytes() {
586            match byte {
587                b'A'..=b'Z' | b'a'..=b'z' | b'0'..=b'9' | b'-' | b'_' | b'.' | b'~' => {
588                    out.push(byte as char);
589                }
590                b'/' if !encode_slash => out.push('/'),
591                other => {
592                    use std::fmt::Write;
593                    let _ = write!(out, "%{other:02X}");
594                }
595            }
596        }
597        out
598    }
599
600    /// Build a sorted canonical query string from key-value pairs.
601    pub fn canonical_query_string(&self, params: &[(String, String)]) -> String {
602        let mut sorted: Vec<(String, String)> = params
603            .iter()
604            .map(|(k, v)| (Self::uri_encode(k, true), Self::uri_encode(v, true)))
605            .collect();
606        sorted.sort_by(|(a, _), (b, _)| a.cmp(b));
607        sorted
608            .iter()
609            .map(|(k, v)| format!("{k}={v}"))
610            .collect::<Vec<_>>()
611            .join("&")
612    }
613
614    /// Generate an AWS SigV4 presigned URL.
615    pub fn generate_s3(
616        &self,
617        url: &ObjectUrl,
618        config: &PresignedUrlConfig,
619        timestamp_unix: u64,
620    ) -> Result<String, CloudError> {
621        let (access_key_id, secret_access_key) = match &self.credentials {
622            CloudCredentials::AccessKey {
623                access_key_id,
624                secret_access_key,
625                ..
626            } => (access_key_id.as_str(), secret_access_key.as_str()),
627            _ => return Err(CloudError::MissingCredentials),
628        };
629
630        let service = "s3";
631        let date = Self::format_date(timestamp_unix);
632        let datetime = Self::format_datetime(timestamp_unix);
633
634        let credential = format!(
635            "{access_key_id}/{date}/{}/{service}/aws4_request",
636            self.region
637        );
638
639        let host = url.signing_host();
640        let path = Self::uri_encode(&url.key, false);
641        let canonical_path = format!("/{path}");
642
643        let mut query_params: Vec<(String, String)> = vec![
644            ("X-Amz-Algorithm".to_owned(), "AWS4-HMAC-SHA256".to_owned()),
645            ("X-Amz-Credential".to_owned(), credential.clone()),
646            ("X-Amz-Date".to_owned(), datetime.clone()),
647            (
648                "X-Amz-Expires".to_owned(),
649                config.expires_in_secs.to_string(),
650            ),
651            ("X-Amz-SignedHeaders".to_owned(), "host".to_owned()),
652        ];
653
654        if let CloudCredentials::AccessKey {
655            session_token: Some(tok),
656            ..
657        } = &self.credentials
658        {
659            query_params.push(("X-Amz-Security-Token".to_owned(), tok.clone()));
660        }
661
662        let canonical_query = self.canonical_query_string(&query_params);
663
664        let canonical_headers = format!("host:{host}\n");
665        let signed_headers = "host";
666
667        // For presigned URLs, the payload hash is "UNSIGNED-PAYLOAD"
668        let payload_hash = "UNSIGNED-PAYLOAD";
669
670        let canonical_request = format!(
671            "{method}\n{path}\n{query}\n{headers}\n{signed}\n{payload}",
672            method = config.method.as_str(),
673            path = canonical_path,
674            query = canonical_query,
675            headers = canonical_headers,
676            signed = signed_headers,
677            payload = payload_hash,
678        );
679
680        let scope = format!("{date}/{}/{service}/aws4_request", self.region);
681        let string_to_sign = format!(
682            "AWS4-HMAC-SHA256\n{datetime}\n{scope}\n{hash}",
683            hash = hex_encode(&sha256(canonical_request.as_bytes())),
684        );
685
686        let signing_key = Self::derive_signing_key(secret_access_key, &date, &self.region, service);
687        let signature = hmac_sha256_hex(&signing_key, string_to_sign.as_bytes());
688
689        let mut final_params = query_params;
690        final_params.push(("X-Amz-Signature".to_owned(), signature));
691        let final_query = self.canonical_query_string(&final_params);
692
693        let base_url = format!("https://{host}{canonical_path}");
694        Ok(format!("{base_url}?{final_query}"))
695    }
696
697    /// Generate a GCS v4 presigned URL (same signing algorithm as S3 SigV4).
698    pub fn generate_gcs(
699        &self,
700        url: &ObjectUrl,
701        config: &PresignedUrlConfig,
702        timestamp_unix: u64,
703    ) -> Result<String, CloudError> {
704        let (access_key_id, secret_access_key) = match &self.credentials {
705            CloudCredentials::AccessKey {
706                access_key_id,
707                secret_access_key,
708                ..
709            } => (access_key_id.as_str(), secret_access_key.as_str()),
710            CloudCredentials::ServiceAccountFile { path } => {
711                // In a real implementation we'd parse the JSON; here we use the path
712                // as a stand-in identifier so the signature is deterministic in tests.
713                return Err(CloudError::PresignError(format!(
714                    "service account file signing requires JSON parsing (path: {path})"
715                )));
716            }
717            _ => return Err(CloudError::MissingCredentials),
718        };
719
720        let service = "storage";
721        let date = Self::format_date(timestamp_unix);
722        let datetime = Self::format_datetime(timestamp_unix);
723        let host = "storage.googleapis.com";
724        let canonical_path = format!("/{}/{}", url.bucket, Self::uri_encode(&url.key, false));
725
726        let credential = format!(
727            "{access_key_id}/{date}/{}/{service}/goog4_request",
728            self.region
729        );
730
731        let query_params: Vec<(String, String)> = vec![
732            (
733                "X-Goog-Algorithm".to_owned(),
734                "GOOG4-HMAC-SHA256".to_owned(),
735            ),
736            ("X-Goog-Credential".to_owned(), credential.clone()),
737            ("X-Goog-Date".to_owned(), datetime.clone()),
738            (
739                "X-Goog-Expires".to_owned(),
740                config.expires_in_secs.to_string(),
741            ),
742            ("X-Goog-SignedHeaders".to_owned(), "host".to_owned()),
743        ];
744
745        let canonical_query = self.canonical_query_string(&query_params);
746        let canonical_headers = format!("host:{host}\n");
747        let signed_headers = "host";
748        let payload_hash = "UNSIGNED-PAYLOAD";
749
750        let canonical_request = format!(
751            "{method}\n{path}\n{query}\n{headers}\n{signed}\n{payload}",
752            method = config.method.as_str(),
753            path = canonical_path,
754            query = canonical_query,
755            headers = canonical_headers,
756            signed = signed_headers,
757            payload = payload_hash,
758        );
759
760        let scope = format!("{date}/{}/{service}/goog4_request", self.region);
761        let string_to_sign = format!(
762            "GOOG4-HMAC-SHA256\n{datetime}\n{scope}\n{hash}",
763            hash = hex_encode(&sha256(canonical_request.as_bytes())),
764        );
765
766        let signing_key = Self::derive_signing_key(secret_access_key, &date, &self.region, service);
767        let signature = hmac_sha256_hex(&signing_key, string_to_sign.as_bytes());
768
769        let mut final_params = query_params;
770        final_params.push(("X-Goog-Signature".to_owned(), signature));
771        let final_query = self.canonical_query_string(&final_params);
772
773        Ok(format!("https://{host}{canonical_path}?{final_query}"))
774    }
775}
776
777// ─────────────────────────────────────────────────────────────────────────────
778// Date arithmetic helper
779// ─────────────────────────────────────────────────────────────────────────────
780
781/// Convert a count of days since the Unix epoch (1970-01-01) to (year, month, day).
782fn days_to_ymd(days: u64) -> (u32, u32, u32) {
783    // Use the civil calendar algorithm (proleptic Gregorian).
784    // Reference: http://howardhinnant.github.io/date_algorithms.html
785    let z = days as i64 + 719_468;
786    let era: i64 = if z >= 0 { z } else { z - 146_096 } / 146_097;
787    let doe = (z - era * 146_097) as u64;
788    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
789    let y = yoe as i64 + era * 400;
790    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
791    let mp = (5 * doy + 2) / 153;
792    let d = doy - (153 * mp + 2) / 5 + 1;
793    let m = if mp < 10 { mp + 3 } else { mp - 9 };
794    let y_adj = if m <= 2 { y + 1 } else { y };
795    (y_adj as u32, m as u32, d as u32)
796}
797
798// ─────────────────────────────────────────────────────────────────────────────
799// MultipartUploadState
800// ─────────────────────────────────────────────────────────────────────────────
801
802/// Tracks the state of an S3 multipart upload.
803#[derive(Debug, Clone, PartialEq, Eq)]
804pub struct CompletedPart {
805    /// 1-based part number.
806    pub part_number: u16,
807    /// ETag returned by the server for this part.
808    pub etag: String,
809    /// Size of this part in bytes.
810    pub size: u64,
811}
812
813/// State machine for an in-progress multipart upload.
814#[derive(Debug, Clone, PartialEq, Eq)]
815pub struct MultipartUploadState {
816    /// The upload ID assigned by the cloud provider.
817    pub upload_id: String,
818    /// The target object URL.
819    pub url: ObjectUrl,
820    /// Parts that have been successfully uploaded.
821    pub parts: Vec<CompletedPart>,
822    /// Nominal part size in bytes.
823    pub part_size: u64,
824}
825
826impl MultipartUploadState {
827    /// Create a new multipart upload tracker.
828    pub fn new(upload_id: impl Into<String>, url: ObjectUrl, part_size: u64) -> Self {
829        MultipartUploadState {
830            upload_id: upload_id.into(),
831            url,
832            parts: Vec::new(),
833            part_size,
834        }
835    }
836
837    /// Record a completed part.
838    pub fn add_part(&mut self, part_number: u16, etag: impl Into<String>, size: u64) {
839        self.parts.push(CompletedPart {
840            part_number,
841            etag: etag.into(),
842            size,
843        });
844    }
845
846    /// Total uploaded bytes across all parts.
847    pub fn total_size(&self) -> u64 {
848        self.parts.iter().map(|p| p.size).sum()
849    }
850
851    /// Number of parts recorded.
852    pub fn part_count(&self) -> usize {
853        self.parts.len()
854    }
855
856    /// Build the S3 `CompleteMultipartUpload` XML body.
857    ///
858    /// Parts are emitted in ascending part-number order.
859    pub fn to_xml(&self) -> String {
860        let mut sorted = self.parts.clone();
861        sorted.sort_by_key(|p| p.part_number);
862
863        let mut xml = String::from("<CompleteMultipartUpload>\n");
864        for part in &sorted {
865            xml.push_str("  <Part>\n");
866            xml.push_str(&format!(
867                "    <PartNumber>{}</PartNumber>\n",
868                part.part_number
869            ));
870            xml.push_str(&format!("    <ETag>{}</ETag>\n", part.etag));
871            xml.push_str("  </Part>\n");
872        }
873        xml.push_str("</CompleteMultipartUpload>");
874        xml
875    }
876}
877
878// ─────────────────────────────────────────────────────────────────────────────
879// CloudRangeCoalescer
880// ─────────────────────────────────────────────────────────────────────────────
881
882/// Merges nearby byte-range requests into larger ones to reduce round-trip
883/// overhead when reading from cloud object storage.
884pub struct CloudRangeCoalescer {
885    /// Maximum gap between two ranges that should still be merged.
886    pub max_gap_bytes: u64,
887    /// Maximum total size of a coalesced request.
888    pub max_request_size: u64,
889    /// Minimum size of a request (avoid sending tiny reads).
890    pub min_request_size: u64,
891}
892
893impl Default for CloudRangeCoalescer {
894    fn default() -> Self {
895        Self::new()
896    }
897}
898
899impl CloudRangeCoalescer {
900    /// Create a coalescer with sensible defaults for cloud storage:
901    /// - `max_gap_bytes` = 512 KiB
902    /// - `max_request_size` = 8 MiB
903    /// - `min_request_size` = 64 KiB
904    pub fn new() -> Self {
905        CloudRangeCoalescer {
906            max_gap_bytes: 512 * 1024,
907            max_request_size: 8 * 1024 * 1024,
908            min_request_size: 64 * 1024,
909        }
910    }
911
912    /// Coalesce a list of byte-range requests.
913    ///
914    /// All input requests must target the **same** URL.  Requests are sorted by
915    /// start offset and then merged when:
916    /// - the gap between consecutive ranges is ≤ `max_gap_bytes`, **and**
917    /// - the resulting coalesced range would not exceed `max_request_size`.
918    pub fn coalesce(&self, mut ranges: Vec<ByteRangeRequest>) -> Vec<ByteRangeRequest> {
919        if ranges.is_empty() {
920            return ranges;
921        }
922
923        // Sort by start offset
924        ranges.sort_by_key(|r| r.range.start);
925
926        let url = ranges[0].url.clone();
927        let mut coalesced: Vec<ByteRangeRequest> = Vec::new();
928        let mut current_start = ranges[0].range.start;
929        let mut current_end = ranges[0].range.end;
930
931        for req in ranges.into_iter().skip(1) {
932            let gap = req.range.start.saturating_sub(current_end);
933            let new_end = req.range.end.max(current_end);
934            let new_size = new_end - current_start;
935
936            if gap <= self.max_gap_bytes && new_size <= self.max_request_size {
937                // Merge
938                current_end = new_end;
939            } else {
940                coalesced.push(ByteRangeRequest::new(
941                    url.clone(),
942                    current_start,
943                    current_end,
944                ));
945                current_start = req.range.start;
946                current_end = req.range.end;
947            }
948        }
949        coalesced.push(ByteRangeRequest::new(url, current_start, current_end));
950        coalesced
951    }
952
953    /// Extract a sub-range from a coalesced response buffer.
954    ///
955    /// `coalesced_start` is the byte offset at which `coalesced_data` begins.
956    /// `sub_range` is the desired slice within the full object.
957    ///
958    /// # Panics
959    ///
960    /// Panics if `sub_range` does not fall within the coalesced data window.
961    pub fn slice_response<'a>(
962        coalesced_data: &'a [u8],
963        coalesced_start: u64,
964        sub_range: &std::ops::Range<u64>,
965    ) -> &'a [u8] {
966        let offset = (sub_range.start - coalesced_start) as usize;
967        let len = (sub_range.end - sub_range.start) as usize;
968        &coalesced_data[offset..offset + len]
969    }
970}