teaql_runtime/
business_id.rs

1use std::collections::HashMap;
2use std::sync::{Arc, Mutex};
3
4use chrono::{NaiveDate, Utc};
5use teaql_core::business_id::{
6    BusinessIdAllocation, BusinessIdAllocator, BusinessIdDefinition, BusinessIdError,
7    BusinessIdErrorCode, BusinessIdGenerationRequest, BusinessIdPlan, BusinessIdProfile,
8    BusinessIdScope, BusinessIdSlot, BusinessIdValue, DEFAULT_BUSINESS_ID_PROFILE,
9};
10
11use crate::UserContext;
12
13/// Provider-owned Business ID schema contribution.
14///
15/// Installing an allocator is passive. The runtime invokes this hook only from
16/// [`UserContext::ensure_schema`], so constructing a context never performs DDL.
17pub trait BusinessIdSchemaContributor: Send + Sync {
18    fn ensure_schema(&self, context: &UserContext) -> Result<(), BusinessIdError>;
19}
20
21#[derive(Clone)]
22pub(crate) struct BusinessIdSchemaService {
23    contributor: Arc<dyn BusinessIdSchemaContributor>,
24}
25
26impl BusinessIdSchemaService {
27    pub(crate) fn from_shared(contributor: Arc<dyn BusinessIdSchemaContributor>) -> Self {
28        Self { contributor }
29    }
30
31    pub(crate) fn ensure_schema(&self, context: &UserContext) -> Result<(), BusinessIdError> {
32        self.contributor.ensure_schema(context)
33    }
34}
35
36#[derive(Debug, Clone, Copy, PartialEq, Eq)]
37pub struct BusinessDate(pub NaiveDate);
38
39impl BusinessDate {
40    pub fn today() -> Self {
41        Self(Utc::now().date_naive())
42    }
43}
44
45#[derive(Debug, Default, Clone, Copy)]
46pub struct DailySequenceBusinessIdProfile;
47
48impl BusinessIdProfile for DailySequenceBusinessIdProfile {
49    fn plan(
50        &self,
51        request: &BusinessIdGenerationRequest<'_>,
52    ) -> Result<BusinessIdPlan, BusinessIdError> {
53        let definition = request.definition;
54        if definition.profile != DEFAULT_BUSINESS_ID_PROFILE {
55            return Err(BusinessIdError::new(
56                BusinessIdErrorCode::InvalidDefinition,
57                format!("unsupported business ID profile {}", definition.profile),
58            ));
59        }
60        if definition.prefix.trim().is_empty()
61            || definition.namespace.trim().is_empty()
62            || definition.separator.is_empty()
63        {
64            return Err(BusinessIdError::new(
65                BusinessIdErrorCode::InvalidDefinition,
66                "prefix, namespace and separator must not be blank",
67            ));
68        }
69        let date_text = definition
70            .split_by_date
71            .then(|| request.business_date.format("%Y%m%d").to_string());
72        Ok(BusinessIdPlan {
73            scope: BusinessIdScope {
74                tenant: request.tenant.to_owned(),
75                namespace: definition.namespace.clone(),
76                date_partition: date_text.clone(),
77            },
78            prefix: definition.prefix.clone(),
79            date_text,
80            preserve_digits: definition.preserve_digits,
81            separator: definition.separator.clone(),
82            max_sequence: definition.max_sequence()?,
83        })
84    }
85
86    fn format(
87        &self,
88        plan: &BusinessIdPlan,
89        allocation: &BusinessIdAllocation,
90    ) -> Result<BusinessIdValue, BusinessIdError> {
91        if allocation.scope != plan.scope || allocation.sequence > plan.max_sequence {
92            return Err(BusinessIdError::new(
93                BusinessIdErrorCode::Exhausted,
94                "business ID allocation does not belong to the plan or exceeds its range",
95            ));
96        }
97        let sequence = format!(
98            "{:0width$}",
99            allocation.sequence,
100            width = usize::from(plan.preserve_digits)
101        );
102        let value = match &plan.date_text {
103            Some(date) => [plan.prefix.as_str(), date, sequence.as_str()].join(&plan.separator),
104            None => [plan.prefix.as_str(), sequence.as_str()].join(&plan.separator),
105        };
106        Ok(BusinessIdValue(value))
107    }
108
109    fn validate(
110        &self,
111        definition: &BusinessIdDefinition,
112        value: &str,
113    ) -> Result<BusinessIdValue, BusinessIdError> {
114        if definition.profile != DEFAULT_BUSINESS_ID_PROFILE
115            || definition.prefix.trim().is_empty()
116            || definition.namespace.trim().is_empty()
117            || definition.separator.is_empty()
118        {
119            return Err(BusinessIdError::new(
120                BusinessIdErrorCode::InvalidDefinition,
121                "invalid daily-sequence Business ID definition",
122            ));
123        }
124        let parts = value.split(&definition.separator).collect::<Vec<_>>();
125        let valid = if definition.split_by_date {
126            parts.len() == 3
127                && parts[0] == definition.prefix
128                && parts[1].len() == 8
129                && parts[1].bytes().all(|value| value.is_ascii_digit())
130                && chrono::NaiveDate::parse_from_str(parts[1], "%Y%m%d").is_ok()
131                && valid_sequence(parts[2], definition.preserve_digits)
132        } else {
133            parts.len() == 2
134                && parts[0] == definition.prefix
135                && valid_sequence(parts[1], definition.preserve_digits)
136        };
137        if !valid {
138            return Err(BusinessIdError::new(
139                BusinessIdErrorCode::InvalidFormat,
140                format!("invalid daily-sequence Business ID: {value}"),
141            ));
142        }
143        Ok(BusinessIdValue(value.to_owned()))
144    }
145}
146
147fn valid_sequence(value: &str, preserve_digits: u8) -> bool {
148    value.len() == usize::from(preserve_digits) && value.bytes().all(|value| value.is_ascii_digit())
149}
150
151#[derive(Debug, Default)]
152pub struct InMemoryBusinessIdAllocator {
153    levels: Mutex<HashMap<String, u64>>,
154}
155
156impl BusinessIdAllocator for InMemoryBusinessIdAllocator {
157    fn allocate(&self, plan: &BusinessIdPlan) -> Result<BusinessIdAllocation, BusinessIdError> {
158        let key = plan.scope.canonical_key();
159        let mut levels = self.levels.lock().map_err(|_| {
160            BusinessIdError::new(
161                BusinessIdErrorCode::Allocation,
162                "in-memory business ID allocator lock is poisoned",
163            )
164        })?;
165        let next = levels
166            .get(&key)
167            .copied()
168            .unwrap_or(0)
169            .checked_add(1)
170            .ok_or_else(|| {
171                BusinessIdError::new(
172                    BusinessIdErrorCode::Exhausted,
173                    "business ID sequence overflow",
174                )
175            })?;
176        if next > plan.max_sequence {
177            return Err(BusinessIdError::new(
178                BusinessIdErrorCode::Exhausted,
179                format!("business ID sequence exhausted for scope {key}"),
180            ));
181        }
182        levels.insert(key, next);
183        Ok(BusinessIdAllocation {
184            scope: plan.scope.clone(),
185            sequence: next,
186        })
187    }
188}
189
190#[derive(Clone)]
191pub struct BusinessIdService {
192    allocator: Arc<dyn BusinessIdAllocator>,
193    profile: DailySequenceBusinessIdProfile,
194}
195
196impl BusinessIdService {
197    pub fn new(allocator: impl BusinessIdAllocator + 'static) -> Self {
198        Self {
199            allocator: Arc::new(allocator),
200            profile: DailySequenceBusinessIdProfile,
201        }
202    }
203
204    pub fn from_shared(allocator: Arc<dyn BusinessIdAllocator>) -> Self {
205        Self {
206            allocator,
207            profile: DailySequenceBusinessIdProfile,
208        }
209    }
210
211    pub fn ensure<S: BusinessIdSlot>(
212        &self,
213        context: &UserContext,
214        definition: &BusinessIdDefinition,
215        tenant: &str,
216        aggregate_type: &str,
217        slot: &mut S,
218    ) -> Result<BusinessIdValue, BusinessIdError> {
219        if let Some(current) = slot.current_business_id().filter(|value| !value.is_empty()) {
220            return self.profile.validate(definition, current);
221        }
222        if !slot.is_new_aggregate() {
223            return Err(BusinessIdError::new(
224                BusinessIdErrorCode::Immutable,
225                format!(
226                    "{} is immutable after the aggregate is persisted",
227                    definition.field_name
228                ),
229            ));
230        }
231        let business_date = context
232            .get_resource::<BusinessDate>()
233            .copied()
234            .unwrap_or_else(BusinessDate::today)
235            .0;
236        let request = BusinessIdGenerationRequest {
237            definition,
238            tenant,
239            aggregate_type,
240            business_date,
241        };
242        let plan = self.profile.plan(&request)?;
243        let allocation = self.allocator.allocate(&plan)?;
244        let value = self.profile.format(&plan, &allocation)?;
245        slot.assign_business_id(value.clone());
246        Ok(value)
247    }
248}
teaql_runtime/business_id.rs

teaql_runtime/
business_id.rs
