oxide_batch_core/domain/
parameter.rs

1use std::collections::BTreeMap;
2use std::fmt;
3
4use sha2::{Digest, Sha256};
5
6use super::{DomainError, JobName, ParameterName};
7
8const MAX_PARAMETER_STRING_BYTES: usize = 64 * 1024;
9
10/// The stable type discriminator for a [`ParameterValue`].
11#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
12#[non_exhaustive]
13pub enum ParameterValueKind {
14    /// A UTF-8 string.
15    String,
16    /// A signed 64-bit integer.
17    I64,
18    /// An unsigned 64-bit integer.
19    U64,
20    /// A boolean.
21    Bool,
22}
23
24impl ParameterValueKind {
25    /// Returns the stable type tag for this kind.
26    ///
27    /// The tag identifies the parameter's type without exposing its value, so
28    /// redacted projections and audit records can carry it.
29    #[must_use]
30    pub const fn as_str(self) -> &'static str {
31        match self {
32            Self::String => "STRING",
33            Self::I64 => "I64",
34            Self::U64 => "U64",
35            Self::Bool => "BOOL",
36        }
37    }
38}
39
40impl fmt::Display for ParameterValueKind {
41    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
42        formatter.write_str(self.as_str())
43    }
44}
45
46/// A bounded typed job-parameter value.
47///
48/// `Debug` and `Display` intentionally redact the underlying value. Use the
49/// typed accessors only at an application or persistence boundary that is
50/// authorized to consume the parameter.
51#[derive(Clone, Eq, Hash, Ord, PartialEq, PartialOrd)]
52pub struct ParameterValue(ParameterValueInner);
53
54#[derive(Clone, Eq, Hash, Ord, PartialEq, PartialOrd)]
55enum ParameterValueInner {
56    String(String),
57    I64(i64),
58    U64(u64),
59    Bool(bool),
60}
61
62impl ParameterValue {
63    /// Validates and constructs a string parameter.
64    ///
65    /// # Errors
66    ///
67    /// Returns [`DomainError::ParameterStringTooLong`] when the UTF-8 value is
68    /// larger than 64 KiB.
69    pub fn string(value: impl Into<String>) -> Result<Self, DomainError> {
70        let value = value.into();
71        if value.len() > MAX_PARAMETER_STRING_BYTES {
72            return Err(DomainError::ParameterStringTooLong {
73                max_bytes: MAX_PARAMETER_STRING_BYTES,
74            });
75        }
76        Ok(Self(ParameterValueInner::String(value)))
77    }
78
79    /// Returns the stable type discriminator.
80    #[must_use]
81    pub const fn kind(&self) -> ParameterValueKind {
82        match self {
83            Self(ParameterValueInner::String(_)) => ParameterValueKind::String,
84            Self(ParameterValueInner::I64(_)) => ParameterValueKind::I64,
85            Self(ParameterValueInner::U64(_)) => ParameterValueKind::U64,
86            Self(ParameterValueInner::Bool(_)) => ParameterValueKind::Bool,
87        }
88    }
89
90    /// Borrows the string value when this is a string parameter.
91    #[must_use]
92    pub fn as_str(&self) -> Option<&str> {
93        match self {
94            Self(ParameterValueInner::String(value)) => Some(value),
95            _ => None,
96        }
97    }
98
99    /// Returns the signed integer when this is an `i64` parameter.
100    #[must_use]
101    pub const fn as_i64(&self) -> Option<i64> {
102        match self {
103            Self(ParameterValueInner::I64(value)) => Some(*value),
104            _ => None,
105        }
106    }
107
108    /// Returns the unsigned integer when this is a `u64` parameter.
109    #[must_use]
110    pub const fn as_u64(&self) -> Option<u64> {
111        match self {
112            Self(ParameterValueInner::U64(value)) => Some(*value),
113            _ => None,
114        }
115    }
116
117    /// Returns the boolean when this is a boolean parameter.
118    #[must_use]
119    pub const fn as_bool(&self) -> Option<bool> {
120        match self {
121            Self(ParameterValueInner::Bool(value)) => Some(*value),
122            _ => None,
123        }
124    }
125}
126
127impl From<i64> for ParameterValue {
128    fn from(value: i64) -> Self {
129        Self(ParameterValueInner::I64(value))
130    }
131}
132
133impl TryFrom<String> for ParameterValue {
134    type Error = DomainError;
135
136    fn try_from(value: String) -> Result<Self, Self::Error> {
137        Self::string(value)
138    }
139}
140
141impl TryFrom<&str> for ParameterValue {
142    type Error = DomainError;
143
144    fn try_from(value: &str) -> Result<Self, Self::Error> {
145        Self::string(value)
146    }
147}
148
149impl From<u64> for ParameterValue {
150    fn from(value: u64) -> Self {
151        Self(ParameterValueInner::U64(value))
152    }
153}
154
155impl From<bool> for ParameterValue {
156    fn from(value: bool) -> Self {
157        Self(ParameterValueInner::Bool(value))
158    }
159}
160
161impl fmt::Debug for ParameterValue {
162    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
163        formatter
164            .debug_tuple(match self.kind() {
165                ParameterValueKind::String => "String",
166                ParameterValueKind::I64 => "I64",
167                ParameterValueKind::U64 => "U64",
168                ParameterValueKind::Bool => "Bool",
169            })
170            .field(&Redacted)
171            .finish()
172    }
173}
174
175impl fmt::Display for ParameterValue {
176    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
177        formatter.write_str("<redacted>")
178    }
179}
180
181struct Redacted;
182
183impl fmt::Debug for Redacted {
184    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
185        formatter.write_str("<redacted>")
186    }
187}
188
189/// Whether a job parameter participates in job-instance identity.
190#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
191pub enum ParameterRole {
192    /// The name, type, and value participate in job-instance identity.
193    Identifying,
194    /// The parameter is launch metadata and does not select the job instance.
195    NonIdentifying,
196}
197
198/// One typed job parameter and its identity role.
199#[derive(Clone, Eq, Hash, Ord, PartialEq, PartialOrd)]
200pub struct JobParameter {
201    value: ParameterValue,
202    role: ParameterRole,
203}
204
205impl JobParameter {
206    /// Constructs a typed job parameter.
207    #[must_use]
208    pub const fn new(value: ParameterValue, role: ParameterRole) -> Self {
209        Self { value, role }
210    }
211
212    /// Borrows the typed value.
213    #[must_use]
214    pub const fn value(&self) -> &ParameterValue {
215        &self.value
216    }
217
218    /// Returns the identity role.
219    #[must_use]
220    pub const fn role(&self) -> ParameterRole {
221        self.role
222    }
223
224    /// Returns whether the parameter participates in instance identity.
225    #[must_use]
226    pub const fn is_identifying(&self) -> bool {
227        matches!(self.role, ParameterRole::Identifying)
228    }
229}
230
231impl fmt::Debug for JobParameter {
232    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
233        formatter
234            .debug_struct("JobParameter")
235            .field("kind", &self.value.kind())
236            .field("role", &self.role)
237            .field("value", &Redacted)
238            .finish()
239    }
240}
241
242/// A deterministically ordered set of typed job parameters.
243#[derive(Clone, Default, Eq, PartialEq)]
244pub struct JobParameters {
245    values: BTreeMap<ParameterName, JobParameter>,
246}
247
248impl JobParameters {
249    /// Constructs an empty parameter set.
250    #[must_use]
251    pub const fn new() -> Self {
252        Self {
253            values: BTreeMap::new(),
254        }
255    }
256
257    /// Inserts a parameter without silently replacing an existing name.
258    ///
259    /// # Errors
260    ///
261    /// Returns [`DomainError::DuplicateParameter`] when the name is already
262    /// present.
263    pub fn insert(
264        &mut self,
265        name: ParameterName,
266        parameter: JobParameter,
267    ) -> Result<(), DomainError> {
268        if self.values.contains_key(&name) {
269            return Err(DomainError::DuplicateParameter);
270        }
271        self.values.insert(name, parameter);
272        Ok(())
273    }
274
275    /// Builds a parameter set while rejecting duplicate names.
276    ///
277    /// # Errors
278    ///
279    /// Returns [`DomainError::DuplicateParameter`] when an input name occurs
280    /// more than once.
281    pub fn try_from_iter(
282        parameters: impl IntoIterator<Item = (ParameterName, JobParameter)>,
283    ) -> Result<Self, DomainError> {
284        let mut result = Self::new();
285        for (name, parameter) in parameters {
286            result.insert(name, parameter)?;
287        }
288        Ok(result)
289    }
290
291    /// Returns a parameter by its validated name.
292    #[must_use]
293    pub fn get(&self, name: &ParameterName) -> Option<&JobParameter> {
294        self.values.get(name)
295    }
296
297    /// Iterates in canonical name order.
298    #[must_use]
299    pub fn iter(&self) -> impl ExactSizeIterator<Item = (&ParameterName, &JobParameter)> {
300        self.values.iter()
301    }
302
303    /// Returns the number of parameters.
304    #[must_use]
305    pub fn len(&self) -> usize {
306        self.values.len()
307    }
308
309    /// Returns whether no parameters are present.
310    #[must_use]
311    pub fn is_empty(&self) -> bool {
312        self.values.is_empty()
313    }
314
315    /// Returns the number of parameters that participate in identity.
316    #[must_use]
317    pub fn identifying_len(&self) -> usize {
318        self.values
319            .values()
320            .filter(|parameter| parameter.is_identifying())
321            .count()
322    }
323
324    /// Hashes the complete typed parameter projection for durable flow input.
325    ///
326    /// The raw canonical bytes never leave this sensitivity-owning type. Flow
327    /// records retain only the domain-separated SHA-256 result, so parameters
328    /// are neither copied into repository rows nor exposed through diagnostics.
329    #[must_use]
330    pub fn flow_input_digest(&self) -> [u8; 32] {
331        let mut hash = Sha256::new();
332        hash.update(b"oxide-batch.flow-parameters.v1\0");
333        for (name, parameter) in &self.values {
334            hash_parameter_field(&mut hash, name.as_str().as_bytes());
335            hash.update([match parameter.role() {
336                ParameterRole::Identifying => 1,
337                ParameterRole::NonIdentifying => 0,
338            }]);
339            match &parameter.value.0 {
340                ParameterValueInner::String(value) => {
341                    hash.update([1]);
342                    hash_parameter_field(&mut hash, value.as_bytes());
343                }
344                ParameterValueInner::I64(value) => {
345                    hash.update([2]);
346                    hash.update(value.to_be_bytes());
347                }
348                ParameterValueInner::U64(value) => {
349                    hash.update([3]);
350                    hash.update(value.to_be_bytes());
351                }
352                ParameterValueInner::Bool(value) => {
353                    hash.update([4, u8::from(*value)]);
354                }
355            }
356        }
357        hash.finalize().into()
358    }
359}
360
361fn hash_parameter_field(hash: &mut Sha256, value: &[u8]) {
362    hash.update(u64::try_from(value.len()).unwrap_or(u64::MAX).to_be_bytes());
363    hash.update(value);
364}
365
366impl fmt::Debug for JobParameters {
367    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
368        formatter
369            .debug_struct("JobParameters")
370            .field("parameter_count", &self.len())
371            .field("identifying_count", &self.identifying_len())
372            .finish_non_exhaustive()
373    }
374}
375
376/// The canonical identity key for a logical job instance.
377///
378/// Parameter entries are ordered by validated name, retain their value type,
379/// and include only parameters marked [`ParameterRole::Identifying`].
380#[derive(Clone, Eq, Hash, Ord, PartialEq, PartialOrd)]
381pub struct JobInstanceKey {
382    job_name: JobName,
383    identifying_parameters: BTreeMap<ParameterName, ParameterValue>,
384}
385
386impl JobInstanceKey {
387    /// Constructs the canonical key for a named job and parameter set.
388    #[must_use]
389    pub fn new(job_name: JobName, parameters: &JobParameters) -> Self {
390        let identifying_parameters = parameters
391            .iter()
392            .filter(|(_, parameter)| parameter.is_identifying())
393            .map(|(name, parameter)| (name.clone(), parameter.value().clone()))
394            .collect();
395
396        Self {
397            job_name,
398            identifying_parameters,
399        }
400    }
401
402    /// Borrows the logical job name.
403    #[must_use]
404    pub const fn job_name(&self) -> &JobName {
405        &self.job_name
406    }
407
408    /// Returns the number of identifying parameters.
409    #[must_use]
410    pub fn identifying_parameter_count(&self) -> usize {
411        self.identifying_parameters.len()
412    }
413
414    /// Returns an identifying value for authorized application or persistence
415    /// use.
416    #[must_use]
417    pub fn identifying_value(&self, name: &ParameterName) -> Option<&ParameterValue> {
418        self.identifying_parameters.get(name)
419    }
420
421    /// Iterates over identifying parameter names and value kinds in canonical
422    /// order without exposing their values.
423    #[must_use]
424    pub fn identifying_fields(
425        &self,
426    ) -> impl ExactSizeIterator<Item = (&ParameterName, ParameterValueKind)> {
427        self.identifying_parameters
428            .iter()
429            .map(|(name, value)| (name, value.kind()))
430    }
431
432    /// Returns the canonical 32-byte digest of this identifying key.
433    ///
434    /// The encoding is version tagged, length prefixed, and type tagged, so
435    /// two keys collide only when their job name and identifying parameters
436    /// are equal. Durable adapters, operator request digests, and redacted
437    /// projections share it, and no parameter value is recoverable from the
438    /// result. The version-1 byte layout is durable data and never changes.
439    #[must_use]
440    pub fn digest(&self) -> [u8; 32] {
441        let mut encoded = Vec::new();
442        encoded.push(1);
443        push_length_prefixed(&mut encoded, self.job_name.as_str().as_bytes());
444        for (name, value) in &self.identifying_parameters {
445            push_length_prefixed(&mut encoded, name.as_str().as_bytes());
446            encoded.push(parameter_tag(value.kind()));
447            match &value.0 {
448                ParameterValueInner::String(value) => {
449                    push_length_prefixed(&mut encoded, value.as_bytes());
450                }
451                ParameterValueInner::I64(value) => {
452                    encoded.extend_from_slice(&value.to_be_bytes());
453                }
454                ParameterValueInner::U64(value) => {
455                    encoded.extend_from_slice(&value.to_be_bytes());
456                }
457                ParameterValueInner::Bool(value) => encoded.push(u8::from(*value)),
458            }
459        }
460        Sha256::digest(encoded).into()
461    }
462}
463
464fn push_length_prefixed(target: &mut Vec<u8>, value: &[u8]) {
465    let length = u32::try_from(value.len()).unwrap_or(u32::MAX);
466    target.extend_from_slice(&length.to_be_bytes());
467    target.extend_from_slice(value);
468}
469
470const fn parameter_tag(kind: ParameterValueKind) -> u8 {
471    match kind {
472        ParameterValueKind::String => 1,
473        ParameterValueKind::I64 => 2,
474        ParameterValueKind::U64 => 3,
475        ParameterValueKind::Bool => 4,
476    }
477}
478
479impl fmt::Debug for JobInstanceKey {
480    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
481        formatter
482            .debug_struct("JobInstanceKey")
483            .field("job_name", &self.job_name)
484            .field(
485                "identifying_parameter_count",
486                &self.identifying_parameter_count(),
487            )
488            .finish_non_exhaustive()
489    }
490}
oxide_batch_core/domain/parameter.rs

oxide_batch_core/domain/
parameter.rs
