scm_sdk/
helpers.rs

1//! Generic, bounded native equivalents of the public SDK helpers.
2use crate::data::{FiniteNumber, Patch};
3use crate::pagination::{
4    CollectOptions, CollectionResult, Completion, Items, Page, Pages, Truncation,
5};
6use crate::{Client, ClientBuilder, Error, ErrorKind, RequestOptions, Result, Session};
7use futures_util::{Stream, StreamExt};
8use serde::Serialize;
9use serde_json::Value;
10use std::{fmt, future::Future, hash::Hash, pin::Pin};
11
12const SAFE_INTEGER: u64 = 9_007_199_254_740_991;
13pub(crate) const MAX_HELPER_BYTES: usize = 8 * 1024 * 1024;
14pub(crate) const MAX_HELPER_NODES: usize = 100_000;
15pub(crate) const MAX_HELPER_DEPTH: usize = 64;
16
17pub(crate) fn bounded_json(input: &Value) -> Result<()> {
18    let mut pending = vec![(input, 0usize)];
19    let mut nodes = 0usize;
20    while let Some((value, depth)) = pending.pop() {
21        nodes += 1;
22        if depth > MAX_HELPER_DEPTH || nodes > MAX_HELPER_NODES {
23            return Err(Error::new(ErrorKind::Limit));
24        }
25        let children = match value {
26            Value::Array(v) => v.len(),
27            Value::Object(v) => v.len(),
28            _ => 0,
29        };
30        if nodes.saturating_add(pending.len()).saturating_add(children) > MAX_HELPER_NODES {
31            return Err(Error::new(ErrorKind::Limit));
32        }
33        match value {
34            Value::Array(values) => pending.extend(values.iter().map(|v| (v, depth + 1))),
35            Value::Object(values) => pending.extend(values.values().map(|v| (v, depth + 1))),
36            Value::Number(number) if number.as_f64().is_none_or(|v| !v.is_finite()) => {
37                return Err(Error::new(ErrorKind::Validation));
38            }
39            _ => {}
40        }
41    }
42    crate::engine::serialize_bounded(input, MAX_HELPER_BYTES)?;
43    Ok(())
44}
45
46/// Build an immutable client configuration. No global singleton is modified.
47pub fn init(builder: ClientBuilder) -> Result<Client> {
48    builder.build()
49}
50
51#[derive(Clone, Default)]
52#[non_exhaustive]
53pub struct ListingOptions {
54    pub skip: Option<u64>,
55    pub limit: Option<u32>,
56    pub search: Option<String>,
57}
58impl fmt::Debug for ListingOptions {
59    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
60        f.write_str("ListingOptions([REDACTED])")
61    }
62}
63/// Preserve canonical key order and explicit zero/empty values; omit only unset fields.
64pub fn serialize_listing(options: Option<&ListingOptions>) -> Result<Vec<(String, String)>> {
65    let mut query = Vec::new();
66    if let Some(options) = options {
67        if let Some(skip) = options.skip {
68            if skip > SAFE_INTEGER {
69                return Err(Error::new(ErrorKind::Validation));
70            }
71            query.push(("skip".into(), skip.to_string()));
72        }
73        if let Some(limit) = options.limit {
74            query.push(("limit".into(), limit.to_string()));
75        }
76        if let Some(search) = &options.search {
77            if search.len() > crate::config::MAX_URL_BYTES || search.chars().any(char::is_control) {
78                return Err(Error::new(ErrorKind::Validation));
79            }
80            query.push(("search".into(), search.clone()));
81        }
82    }
83    Ok(query)
84}
85
86/// Cursor-based item iteration with a caller-owned session and one absolute budget.
87pub fn paginate<'a, T: 'a, C, F, Fut>(
88    initial: C,
89    fetch: F,
90    max_pages: usize,
91    session: Session,
92    options: RequestOptions,
93) -> Result<Items<'a, T, C>>
94where
95    C: Clone + Eq + Hash + Serialize + Send + 'a,
96    F: FnMut(C, RequestOptions) -> Fut + Send + 'a,
97    Fut: Future<Output = Result<Page<T, C>>> + Send + 'a,
98{
99    Ok(Pages::new(initial, options, session, max_pages, fetch)?.items())
100}
101
102#[derive(Clone, Copy)]
103#[non_exhaustive]
104pub struct WalkAllOptions {
105    pub limit: u32,
106    pub collect: CollectOptions,
107}
108impl Default for WalkAllOptions {
109    fn default() -> Self {
110        Self {
111            limit: 50,
112            collect: CollectOptions::default(),
113        }
114    }
115}
116#[derive(Clone, Copy)]
117#[non_exhaustive]
118pub struct SpringWalkOptions {
119    pub size: u32,
120    pub collect: CollectOptions,
121}
122impl Default for SpringWalkOptions {
123    fn default() -> Self {
124        Self {
125            size: 50,
126            collect: CollectOptions::default(),
127        }
128    }
129}
130pub struct SkipPage<T> {
131    pub items: Vec<T>,
132    pub total: Option<FiniteNumber>,
133}
134pub struct SpringPage<T> {
135    pub items: Vec<T>,
136    pub last: bool,
137}
138crate::redaction::redacted_debug!(WalkAllOptions, SpringWalkOptions);
139impl<T> fmt::Debug for SkipPage<T> {
140    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
141        f.write_str("SkipPage([REDACTED])")
142    }
143}
144impl<T> fmt::Debug for SpringPage<T> {
145    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
146        f.write_str("SpringPage([REDACTED])")
147    }
148}
149
150pub async fn collect_skip_pages<'a, T: Serialize + Send + 'a, F, Fut>(
151    mut fetch: F,
152    walk: WalkAllOptions,
153    session: Session,
154    options: RequestOptions,
155) -> CollectionResult<T, u64>
156where
157    F: FnMut(u64, u32, RequestOptions) -> Fut + Send + 'a,
158    Fut: Future<Output = Result<SkipPage<T>>> + Send + 'a,
159{
160    walk.collect.validate()?;
161    if walk.limit == 0 {
162        return Err(Error::new(ErrorKind::Validation).into());
163    }
164    let pages = Pages::new(
165        0u64,
166        options,
167        session,
168        walk.collect.max_pages,
169        move |skip, options| {
170            let future = fetch(skip, walk.limit, options);
171            async move {
172                let page = future.await?;
173                let next = skip
174                    .checked_add(page.items.len() as u64)
175                    .filter(|v| *v <= SAFE_INTEGER)
176                    .ok_or_else(|| Error::new(ErrorKind::Pagination))?;
177                let total = page
178                    .total
179                    .map(|number| {
180                        let value = number
181                            .as_f64()
182                            .ok_or_else(|| Error::new(ErrorKind::Pagination))?;
183                        if value < 0.0 || value > SAFE_INTEGER as f64 || value.fract() != 0.0 {
184                            return Err(Error::new(ErrorKind::Pagination));
185                        }
186                        Ok(value as u64)
187                    })
188                    .transpose()?;
189                let more = !page.items.is_empty()
190                    && total.map_or(page.items.len() == walk.limit as usize, |total| {
191                        next < total
192                    });
193                Ok(Page {
194                    items: page.items,
195                    next: more.then_some(next),
196                })
197            }
198        },
199    )?;
200    pages.collect(walk.collect).await
201}
202pub async fn collect_spring_pages<'a, T: Serialize + Send + 'a, F, Fut>(
203    mut fetch: F,
204    walk: SpringWalkOptions,
205    session: Session,
206    options: RequestOptions,
207) -> CollectionResult<T, u64>
208where
209    F: FnMut(u64, u32, RequestOptions) -> Fut + Send + 'a,
210    Fut: Future<Output = Result<SpringPage<T>>> + Send + 'a,
211{
212    walk.collect.validate()?;
213    if walk.size == 0 {
214        return Err(Error::new(ErrorKind::Validation).into());
215    }
216    let pages = Pages::new(
217        0u64,
218        options,
219        session,
220        walk.collect.max_pages,
221        move |number, options| {
222            let future = fetch(number, walk.size, options);
223            async move {
224                let page = future.await?;
225                if !page.last && page.items.is_empty() {
226                    return Err(Error::new(ErrorKind::Pagination));
227                }
228                let next = if page.last {
229                    None
230                } else {
231                    Some(
232                        number
233                            .checked_add(1)
234                            .filter(|v| *v <= SAFE_INTEGER)
235                            .ok_or_else(|| Error::new(ErrorKind::Pagination))?,
236                    )
237                };
238                Ok(Page {
239                    items: page.items,
240                    next,
241                })
242            }
243        },
244    )?;
245    pages.collect(walk.collect).await
246}
247
248/// Retains a pulled item when a byte cap prevents accepting it; collection can resume.
249pub struct StreamListing<S: Stream> {
250    stream: Pin<Box<S>>,
251    pending: Option<S::Item>,
252    ended: bool,
253}
254impl<S: Stream> StreamListing<S> {
255    pub fn new(stream: S) -> Self {
256        Self {
257            stream: Box::pin(stream),
258            pending: None,
259            ended: false,
260        }
261    }
262}
263impl<S: Stream> fmt::Debug for StreamListing<S> {
264    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
265        f.write_str("StreamListing([REDACTED])")
266    }
267}
268pub struct StreamCollection<T> {
269    pub items: Vec<T>,
270    pub completion: Completion,
271    pub wire_bytes: usize,
272}
273impl<T> StreamCollection<T> {
274    pub fn is_complete(&self) -> bool {
275        self.completion == Completion::Complete
276    }
277}
278impl<T> fmt::Debug for StreamCollection<T> {
279    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
280        f.debug_struct("StreamCollection")
281            .field("item_count", &self.items.len())
282            .field("completion", &self.completion)
283            .finish()
284    }
285}
286pub struct StreamCollectionError<T> {
287    error: Error,
288    partial: Box<StreamCollection<T>>,
289}
290impl<T> StreamCollectionError<T> {
291    pub fn error(&self) -> &Error {
292        &self.error
293    }
294    pub fn partial(&self) -> &StreamCollection<T> {
295        &self.partial
296    }
297    pub fn into_error(self) -> Error {
298        self.error
299    }
300    pub fn into_parts(self) -> (Error, StreamCollection<T>) {
301        (self.error, *self.partial)
302    }
303    fn interrupted(error: Error, mut partial: StreamCollection<T>) -> Self {
304        partial.completion = Completion::Interrupted;
305        Self {
306            error,
307            partial: Box::new(partial),
308        }
309    }
310}
311impl<T> From<Error> for StreamCollectionError<T> {
312    fn from(error: Error) -> Self {
313        Self::interrupted(
314            error,
315            StreamCollection {
316                items: Vec::new(),
317                completion: Completion::Interrupted,
318                wire_bytes: 2,
319            },
320        )
321    }
322}
323impl<T> fmt::Debug for StreamCollectionError<T> {
324    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
325        f.debug_struct("StreamCollectionError")
326            .field("error", &self.error)
327            .field("partial", &self.partial)
328            .finish()
329    }
330}
331impl<T> fmt::Display for StreamCollectionError<T> {
332    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
333        self.error.fmt(f)
334    }
335}
336impl<T> std::error::Error for StreamCollectionError<T> {
337    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
338        Some(&self.error)
339    }
340}
341pub type StreamCollectionResult<T> =
342    std::result::Result<StreamCollection<T>, StreamCollectionError<T>>;
343/// Collect any fallible Rust stream. Item/byte caps retain the iterator and partial rows.
344/// The page cap applies to page iterators; an arbitrary item stream has no pages.
345pub async fn collect_all<S, T>(
346    listing: &mut StreamListing<S>,
347    caps: CollectOptions,
348    session: &Session,
349    options: &RequestOptions,
350) -> StreamCollectionResult<T>
351where
352    S: Stream<Item = Result<T>>,
353    T: Serialize,
354{
355    caps.validate()?;
356    let mut result = StreamCollection {
357        items: Vec::new(),
358        completion: Completion::Complete,
359        wire_bytes: 2,
360    };
361    loop {
362        if let Err(error) = options.run(session, async { Ok(()) }).await {
363            return Err(StreamCollectionError::interrupted(error, result));
364        }
365        if listing.ended {
366            return Ok(result);
367        }
368        if result.items.len() >= caps.max_items {
369            result.completion = Completion::Truncated(Truncation::Items);
370            return Ok(result);
371        }
372        let item = if let Some(item) = listing.pending.take() {
373            Some(item)
374        } else {
375            match options
376                .run(session, async { Ok(listing.stream.next().await) })
377                .await
378            {
379                Ok(item) => item,
380                Err(error) => {
381                    return Err(StreamCollectionError::interrupted(error, result));
382                }
383            }
384        };
385        let Some(item) = item else {
386            listing.ended = true;
387            return Ok(result);
388        };
389        let item = match item {
390            Ok(item) => item,
391            Err(error) => {
392                return Err(StreamCollectionError::interrupted(error, result));
393            }
394        };
395        let comma = usize::from(!result.items.is_empty());
396        let remaining = caps
397            .max_bytes
398            .saturating_sub(result.wire_bytes)
399            .saturating_sub(comma);
400        match crate::engine::serialize_bounded(&item, remaining) {
401            Ok(bytes) => {
402                result.wire_bytes += bytes.len() + comma;
403                result.items.push(item);
404            }
405            Err(error) if error.kind() == ErrorKind::Limit => {
406                listing.pending = Some(Ok(item));
407                result.completion = Completion::Truncated(Truncation::Bytes);
408                return Ok(result);
409            }
410            Err(error) => {
411                listing.pending = Some(Ok(item));
412                return Err(StreamCollectionError::interrupted(error, result));
413            }
414        }
415    }
416}
417
418/// Native schema-composition seam: only present nonnull values reach the inner parser.
419pub fn json_nullable<T>(
420    input: Option<&Value>,
421    inner: impl FnOnce(&Value) -> Result<T>,
422) -> Result<Patch<T>> {
423    match input {
424        None => Ok(Patch::Unset),
425        Some(Value::Null) => Ok(Patch::Null),
426        Some(value) => {
427            bounded_json(value)?;
428            inner(value).map(Patch::Value)
429        }
430    }
431}
432
433/// Decode a generic Spring envelope and validate each content item with the caller's parser.
434#[cfg(feature = "models")]
435pub fn page_schema<T>(
436    input: &Value,
437    mut inner: impl FnMut(&Value) -> Result<T>,
438) -> Result<crate::dlp::models::Page<T>> {
439    bounded_json(input)?;
440    let source: crate::dlp::models::Page<crate::data::JsonValue> =
441        crate::json_decode::from_value(input.clone())
442            .map_err(|_| Error::new(ErrorKind::Validation))?;
443    let mut page = crate::dlp::models::Page::new(
444        source
445            .content
446            .iter()
447            .map(|item| inner(item.as_value()))
448            .collect::<Result<_>>()?,
449    );
450    page.empty = source.empty;
451    page.first = source.first;
452    page.last = source.last;
453    page.number = source.number;
454    page.number_of_elements = source.number_of_elements;
455    page.pageable = source.pageable;
456    page.size = source.size;
457    page.sort = source.sort;
458    page.total_elements = source.total_elements;
459    page.total_pages = source.total_pages;
460    page.extensions = source.extensions;
461    Ok(page)
462}
463
464/// Generate a scope name locally; its random suffix is not an authentication secret.
465#[cfg(feature = "iam")]
466pub fn generate_workspace_scope_name(name: &str) -> Result<String> {
467    if name.len() > 4096 {
468        return Err(Error::new(ErrorKind::Limit));
469    }
470    let mut stem = String::new();
471    let mut separator = false;
472    for ch in name.chars().flat_map(char::to_lowercase) {
473        if ch.is_ascii_lowercase() || ch.is_ascii_digit() {
474            if separator && !stem.is_empty() {
475                stem.push('_');
476            }
477            stem.push(ch);
478            separator = false;
479        } else {
480            separator = true;
481        }
482    }
483    if stem.is_empty() {
484        stem.push_str("workspace");
485    }
486    const ALPHABET: &[u8] = b"abcdefghijklmnopqrstuvwxyz0123456789";
487    let mut suffix = String::new();
488    // Rejection sampling removes modulo bias while bounding a failing entropy source.
489    for _ in 0..16 {
490        let mut bytes = [0u8; 16];
491        getrandom::fill(&mut bytes).map_err(|_| Error::new(ErrorKind::Io))?;
492        for byte in bytes {
493            if byte < 252 {
494                suffix.push(ALPHABET[usize::from(byte) % 36] as char);
495                if suffix.len() == 6 {
496                    return Ok(format!("ws_{stem}_{suffix}"));
497                }
498            }
499        }
500    }
501    Err(Error::new(ErrorKind::Io))
502}
scm_sdk/helpers.rs

scm_sdk/
helpers.rs
