pharos_runtime_database/
sequence.rs

1//! Scoped counters whose allocation commits with the consuming app mutation.
2
3use crate::DbBackendConfig;
4use crate::operations::{
5    AppSchema, DatabaseError, DatabaseOperations, FieldType, Mutation, Predicate, Query, WriteValue,
6};
7use serde_json::Value;
8use std::collections::BTreeMap;
9
10/// Names a declared counter and its unique scope without selecting a backend.
11pub struct SequenceAllocation<'a> {
12    pub entity: &'a str,
13    pub field: &'a str,
14    pub scope: BTreeMap<String, Value>,
15    pub initial: i64,
16    pub maximum: i64,
17}
18
19/// Plans a compare-and-write allocation for inclusion in the consuming transaction.
20/// A failed consumer rolls back the counter; concurrent stale plans conflict.
21pub fn plan_sequence_allocation(
22    adapter: &dyn DatabaseOperations,
23    config: &DbBackendConfig,
24    schema: &AppSchema,
25    request: SequenceAllocation<'_>,
26) -> Result<(i64, Vec<Mutation>), DatabaseError> {
27    let definition = schema
28        .entities
29        .iter()
30        .find(|entity| entity.name == request.entity)
31        .ok_or(DatabaseError::InvalidContract)?;
32    if request.initial < 0
33        || request.maximum <= request.initial
34        || request.scope.is_empty()
35        || request.scope.contains_key(request.field)
36    {
37        return Err(DatabaseError::InvalidInput);
38    }
39    if !definition.fields.iter().any(|field| {
40        field.name == request.field && field.kind == FieldType::Integer && !field.nullable
41    }) {
42        return Err(DatabaseError::InvalidContract);
43    }
44    let keys: Vec<String> = request.scope.keys().cloned().collect();
45    let identity: Vec<String> = definition
46        .identity
47        .fields()
48        .iter()
49        .map(|field| (*field).to_owned())
50        .collect();
51    let same_keys = |candidate: &[String]| {
52        candidate.len() == keys.len() && candidate.iter().all(|key| request.scope.contains_key(key))
53    };
54    if !same_keys(&identity) && !definition.unique.iter().any(|key| same_keys(key)) {
55        return Err(DatabaseError::InvalidContract);
56    }
57    let predicates: Vec<Predicate> = request
58        .scope
59        .iter()
60        .map(|(field, value)| Predicate::Equal(field.clone(), value.clone()))
61        .collect();
62    let rows = adapter.query(
63        config,
64        schema,
65        &Query {
66            entity: request.entity.to_owned(),
67            fields: vec![request.field.to_owned()],
68            filter: Some(Predicate::All(predicates.clone())),
69            order: Vec::new(),
70            limit: 2,
71            offset: 0,
72        },
73    )?;
74    if rows.len() > 1 {
75        return Err(DatabaseError::StorageFailure);
76    }
77    let previous = match rows.first() {
78        Some(row) => row
79            .get(request.field)
80            .and_then(Value::as_i64)
81            .ok_or(DatabaseError::StorageFailure)?,
82        None => request.initial,
83    };
84    let next = previous
85        .max(request.initial)
86        .checked_add(1)
87        .filter(|next| *next <= request.maximum)
88        .ok_or(DatabaseError::CapacityExceeded)?;
89    let mut steps = Vec::with_capacity(2);
90    if rows.is_empty() {
91        let mut values: BTreeMap<String, WriteValue> = request
92            .scope
93            .into_iter()
94            .map(|(key, value)| (key, WriteValue::Literal(value)))
95            .collect();
96        values.insert(
97            request.field.to_owned(),
98            WriteValue::Literal(Value::from(previous)),
99        );
100        steps.push(Mutation::InsertIfAbsent {
101            entity: request.entity.to_owned(),
102            unique_key: keys,
103            values,
104        });
105    }
106    let mut filter = predicates;
107    filter.push(Predicate::Equal(
108        request.field.to_owned(),
109        Value::from(previous),
110    ));
111    steps.push(Mutation::Update {
112        entity: request.entity.to_owned(),
113        filter: Predicate::All(filter),
114        values: BTreeMap::from([(
115            request.field.to_owned(),
116            WriteValue::Literal(Value::from(next)),
117        )]),
118        expected_rows: Some(1),
119    });
120    Ok((next, steps))
121}
pharos_runtime_database/sequence.rs

pharos_runtime_database/
sequence.rs
