pharos_runtime_database/
effects.rs

1//! Transaction-local outbox composition without selecting a database vendor.
2use crate::DbBackendConfig;
3use crate::operations::*;
4use serde_json::{Map, Value};
5use std::collections::BTreeMap;
6use std::sync::Arc;
7
8/// Binds app-declared inserts to each domain write transaction. Read-only
9/// operations never emit effects, and domain result indices remain unchanged.
10pub struct EffectAdapter {
11    inner: Arc<dyn DatabaseOperations>,
12    effects: Vec<Mutation>,
13    resource: Option<ResourceBinding>,
14}
15
16impl EffectAdapter {
17    /// Accepts a bounded insertion-only outbox plan with local identity references.
18    pub fn new(
19        inner: Arc<dyn DatabaseOperations>,
20        schema: &AppSchema,
21        effects: Vec<Mutation>,
22    ) -> Result<Self, DatabaseError> {
23        if effects.is_empty()
24            || effects.len() > 64
25            || effects
26                .iter()
27                .any(|step| !matches!(step, Mutation::Insert { .. }))
28        {
29            return Err(DatabaseError::InvalidInput);
30        }
31        validate_transaction(
32            schema,
33            &Transaction {
34                steps: effects.clone(),
35                idempotency_key: None,
36            },
37        )?;
38        Ok(Self {
39            inner,
40            effects,
41            resource: None,
42        })
43    }
44    /// Binds one declared resource to its generated domain identity at commit.
45    pub fn with_resource(
46        mut self,
47        schema: &AppSchema,
48        resource: ResourceBinding,
49    ) -> Result<Self, DatabaseError> {
50        validate_resource_binding(schema, &resource)?;
51        self.resource = Some(resource);
52        Ok(self)
53    }
54}
55
56/// Maps an app's logical entity label to its declared storage entity.
57#[derive(Clone)]
58pub struct ResourceBinding {
59    pub entity: String,
60    pub prefix: String,
61}
62
63/// Validates the resource declaration on both public composition entry points.
64fn validate_resource_binding(
65    schema: &AppSchema,
66    resource: &ResourceBinding,
67) -> Result<(), DatabaseError> {
68    entity(schema, &resource.entity)?;
69    if resource.prefix.is_empty() || resource.prefix.len() > 128 || !resource.prefix.ends_with(':')
70    {
71        return Err(DatabaseError::InvalidInput);
72    }
73    Ok(())
74}
75
76impl DatabaseOperations for EffectAdapter {
77    /// Delegates backend name through the same configured operation interface.
78    fn backend_name(&self) -> &'static str {
79        self.inner.backend_name()
80    }
81    /// Preserves the underlying adapter artifact contract through this wrapper.
82    fn local_artifact_format(&self) -> Option<&'static str> {
83        self.inner.local_artifact_format()
84    }
85    /// Delegates ensure database through the same configured operation interface.
86    fn ensure_database(
87        &self,
88        config: &DbBackendConfig,
89        schema: &AppSchema,
90    ) -> Result<(), DatabaseError> {
91        self.inner.ensure_database(config, schema)
92    }
93    /// Delegates user credentials through the same configured operation interface.
94    fn user_credentials(
95        &self,
96        config: &DbBackendConfig,
97        schema: &AppSchema,
98        username: &str,
99    ) -> Result<Option<UserCredentials>, DatabaseError> {
100        self.inner.user_credentials(config, schema, username)
101    }
102    /// Delegates update user password through the same configured operation interface.
103    fn update_user_password(
104        &self,
105        config: &DbBackendConfig,
106        user_id: i64,
107        password_hash: &str,
108    ) -> Result<(), DatabaseError> {
109        self.inner
110            .update_user_password(config, user_id, password_hash)
111    }
112    /// Delegates issue session through the same configured operation interface.
113    fn issue_session(
114        &self,
115        config: &DbBackendConfig,
116        user_id: i64,
117        session_token_hash: &str,
118        expires_at: &str,
119    ) -> Result<(), DatabaseError> {
120        self.inner
121            .issue_session(config, user_id, session_token_hash, expires_at)
122    }
123    /// Delegates revoke session through the same configured operation interface.
124    fn revoke_session(
125        &self,
126        config: &DbBackendConfig,
127        session_token_hash: &str,
128    ) -> Result<(), DatabaseError> {
129        self.inner.revoke_session(config, session_token_hash)
130    }
131    /// Delegates revoke user sessions through the same configured operation interface.
132    fn revoke_user_sessions(
133        &self,
134        config: &DbBackendConfig,
135        user_id: i64,
136    ) -> Result<(), DatabaseError> {
137        self.inner.revoke_user_sessions(config, user_id)
138    }
139    /// Delegates active session through the same configured operation interface.
140    fn active_session(
141        &self,
142        config: &DbBackendConfig,
143        schema: &AppSchema,
144        session_token_hash: &str,
145        now_epoch_seconds: u64,
146    ) -> Result<Option<ActiveSession>, DatabaseError> {
147        self.inner
148            .active_session(config, schema, session_token_hash, now_epoch_seconds)
149    }
150    /// Delegates append rate limit event through the same configured operation interface.
151    fn append_rate_limit_event(
152        &self,
153        config: &DbBackendConfig,
154        policy_key: &str,
155        bucket_key: &str,
156        outcome: &str,
157    ) -> Result<(), DatabaseError> {
158        self.inner
159            .append_rate_limit_event(config, policy_key, bucket_key, outcome)
160    }
161    /// Delegates admit rate limited request through the same configured operation interface.
162    fn admit_rate_limited_request(
163        &self,
164        config: &DbBackendConfig,
165        policy_key: &str,
166        bucket_key: &str,
167        maximum: usize,
168        window_seconds: u64,
169    ) -> Result<bool, DatabaseError> {
170        self.inner.admit_rate_limited_request(
171            config,
172            policy_key,
173            bucket_key,
174            maximum,
175            window_seconds,
176        )
177    }
178    /// Delegates apply contract through the same configured operation interface.
179    fn apply_contract(
180        &self,
181        config: &DbBackendConfig,
182        contract: &crate::contract::AppDataContract,
183    ) -> Result<(), DatabaseError> {
184        self.inner.apply_contract(config, contract)
185    }
186    /// Delegates migration ledger through the same configured operation interface.
187    fn migration_ledger(
188        &self,
189        config: &DbBackendConfig,
190        app_id: &str,
191    ) -> Result<BTreeMap<String, String>, DatabaseError> {
192        self.inner.migration_ledger(config, app_id)
193    }
194    /// Delegates applied migration checksum through the same configured operation interface.
195    fn applied_migration_checksum(
196        &self,
197        config: &DbBackendConfig,
198        app_id: &str,
199        migration_id: &str,
200    ) -> Result<Option<String>, DatabaseError> {
201        self.inner
202            .applied_migration_checksum(config, app_id, migration_id)
203    }
204    /// Delegates verify contract through the same configured operation interface.
205    fn verify_contract(
206        &self,
207        config: &DbBackendConfig,
208        contract: &crate::contract::AppDataContract,
209    ) -> Result<(), DatabaseError> {
210        self.inner.verify_contract(config, contract)
211    }
212    /// Delegates ready through the same configured operation interface.
213    fn ready(&self, config: &DbBackendConfig, schema: &AppSchema) -> Result<(), DatabaseError> {
214        self.inner.ready(config, schema)
215    }
216    /// Delegates field exists through the same configured operation interface.
217    fn field_exists(
218        &self,
219        config: &DbBackendConfig,
220        schema: &AppSchema,
221        entity_name: &str,
222        field_name: &str,
223    ) -> Result<bool, DatabaseError> {
224        self.inner
225            .field_exists(config, schema, entity_name, field_name)
226    }
227    /// Delegates migrate through the same configured operation interface.
228    fn migrate(
229        &self,
230        config: &DbBackendConfig,
231        schema: &AppSchema,
232        migration: &Migration,
233    ) -> Result<(), DatabaseError> {
234        self.inner.migrate(config, schema, migration)
235    }
236    /// Delegates adopt legacy migration through the same configured operation interface.
237    fn adopt_legacy_migration(
238        &self,
239        config: &DbBackendConfig,
240        schema: &AppSchema,
241        migration: &Migration,
242        expected_legacy: &BTreeMap<String, String>,
243    ) -> Result<(), DatabaseError> {
244        self.inner
245            .adopt_legacy_migration(config, schema, migration, expected_legacy)
246    }
247    /// Delegates query through the same configured operation interface.
248    fn query(
249        &self,
250        config: &DbBackendConfig,
251        schema: &AppSchema,
252        query: &Query,
253    ) -> Result<Vec<Map<String, Value>>, DatabaseError> {
254        self.inner.query(config, schema, query)
255    }
256    /// Delegates join query through the same configured operation interface.
257    fn join_query(
258        &self,
259        config: &DbBackendConfig,
260        schema: &AppSchema,
261        request: &JoinedQuery,
262    ) -> Result<Vec<JoinedRecord>, DatabaseError> {
263        self.inner.join_query(config, schema, request)
264    }
265    /// Delegates aggregate through the same configured operation interface.
266    fn aggregate(
267        &self,
268        config: &DbBackendConfig,
269        schema: &AppSchema,
270        query: &AggregateQuery,
271    ) -> Result<Value, DatabaseError> {
272        self.inner.aggregate(config, schema, query)
273    }
274    /// Delegates transact through the same configured operation interface.
275    fn transact(
276        &self,
277        config: &DbBackendConfig,
278        schema: &AppSchema,
279        transaction: &Transaction,
280    ) -> Result<TransactionResult, DatabaseError> {
281        transact_with_resource_effects(
282            self.inner.as_ref(),
283            config,
284            schema,
285            transaction,
286            &self.effects,
287            self.resource.as_ref(),
288        )
289    }
290    /// Delegates claim one through the same configured operation interface.
291    fn claim_one(
292        &self,
293        config: &DbBackendConfig,
294        schema: &AppSchema,
295        claim: &Claim,
296    ) -> Result<Option<i64>, DatabaseError> {
297        self.inner.claim_one(config, schema, claim)
298    }
299    /// Delegates export page through the same configured operation interface.
300    fn export_page(
301        &self,
302        config: &DbBackendConfig,
303        schema: &AppSchema,
304        entity: &str,
305        after_identity: Option<Value>,
306        limit: u32,
307    ) -> Result<Vec<Record>, DatabaseError> {
308        self.inner
309            .export_page(config, schema, entity, after_identity, limit)
310    }
311    /// Delegates export snapshot through the same configured operation interface.
312    fn export_snapshot(
313        &self,
314        config: &DbBackendConfig,
315        schema: &AppSchema,
316        limits: SnapshotLimits,
317    ) -> Result<Vec<Record>, DatabaseError> {
318        self.inner.export_snapshot(config, schema, limits)
319    }
320    /// Delegates restore snapshot through the same configured operation interface.
321    fn restore_snapshot(
322        &self,
323        config: &DbBackendConfig,
324        schema: &AppSchema,
325        records: &[Record],
326        limits: SnapshotLimits,
327        counts: &[crate::snapshot::RestoreCount],
328    ) -> Result<(), DatabaseError> {
329        self.inner
330            .restore_snapshot(config, schema, records, limits, counts)
331    }
332    /// Delegates import batch through the same configured operation interface.
333    fn import_batch(
334        &self,
335        config: &DbBackendConfig,
336        schema: &AppSchema,
337        records: &[Record],
338    ) -> Result<(), DatabaseError> {
339        self.inner.import_batch(config, schema, records)
340    }
341}
342
343/// Commits durable outbox inserts with the original domain transaction while
344/// keeping all result positions relative to the caller's original step list.
345pub fn transact_with_effects(
346    adapter: &dyn DatabaseOperations,
347    config: &DbBackendConfig,
348    schema: &AppSchema,
349    transaction: &Transaction,
350    effects: &[Mutation],
351) -> Result<TransactionResult, DatabaseError> {
352    transact_with_resource_effects(adapter, config, schema, transaction, effects, None)
353}
354
355#[derive(Clone, Copy, PartialEq, Eq)]
356enum ResourceIdentity {
357    Generated(usize),
358    Existing(i64),
359}
360
361/// Requires every domain write to this entity to identify the same record.
362/// Broad updates, ambiguous predicates and multi-record batches remain aggregate actions.
363fn resource_identity(
364    schema: &AppSchema,
365    transaction: &Transaction,
366    resource: &ResourceBinding,
367) -> Option<ResourceIdentity> {
368    let identity = entity(schema, &resource.entity)
369        .ok()?
370        .identity
371        .generated_field()?;
372    fn known_id(filter: &Predicate, field: &str) -> Option<i64> {
373        match filter {
374            Predicate::Equal(name, value) if name == field => value.as_i64().filter(|id| *id > 0),
375            Predicate::All(parts) => {
376                let ids = parts
377                    .iter()
378                    .filter_map(|part| known_id(part, field))
379                    .collect::<std::collections::BTreeSet<_>>();
380                if ids.len() == 1 {
381                    ids.into_iter().next()
382                } else {
383                    None
384                }
385            }
386            _ => None,
387        }
388    }
389    let mut result = None;
390    for (index, step) in transaction.steps.iter().enumerate() {
391        let candidate = match step {
392            Mutation::Insert { entity, .. } | Mutation::InsertIfAbsent { entity, .. }
393                if entity == &resource.entity =>
394            {
395                ResourceIdentity::Generated(index)
396            }
397            Mutation::Update { entity, filter, .. } | Mutation::Delete { entity, filter, .. }
398                if entity == &resource.entity =>
399            {
400                ResourceIdentity::Existing(known_id(filter, identity)?)
401            }
402            Mutation::Upsert { entity, .. } | Mutation::Reserve { entity, .. }
403                if entity == &resource.entity =>
404            {
405                return None;
406            }
407            _ => continue,
408        };
409        if result.is_some_and(|prior| prior != candidate) {
410            return None;
411        }
412        result = Some(candidate);
413    }
414    result
415}
416
417/// Resolves a single created resource inside the same transaction as its outbox.
418/// Batch actions retain their declared aggregate resource instead of choosing
419/// an arbitrary record identity.
420pub fn transact_with_resource_effects(
421    adapter: &dyn DatabaseOperations,
422    config: &DbBackendConfig,
423    schema: &AppSchema,
424    transaction: &Transaction,
425    effects: &[Mutation],
426    resource: Option<&ResourceBinding>,
427) -> Result<TransactionResult, DatabaseError> {
428    if let Some(resource) = resource {
429        validate_resource_binding(schema, resource)?;
430    }
431    if effects.is_empty()
432        || transaction
433            .steps
434            .iter()
435            .all(|step| matches!(step, Mutation::RequireCount { .. }))
436    {
437        return adapter.transact(config, schema, transaction);
438    }
439    if effects.len() > 64
440        || effects
441            .iter()
442            .any(|step| !matches!(step, Mutation::Insert { .. }))
443    {
444        return Err(DatabaseError::InvalidContract);
445    }
446    validate_transaction(
447        schema,
448        &Transaction {
449            steps: effects.to_vec(),
450            idempotency_key: None,
451        },
452    )?;
453    let domain_steps = transaction.steps.len();
454    let binding = resource.and_then(|resource| {
455        resource_identity(schema, transaction, resource).map(|identity| (resource, identity))
456    });
457    let mut composed = transaction.clone();
458    for step in effects {
459        let Mutation::Insert { entity, values } = step else {
460            return Err(DatabaseError::InvalidContract);
461        };
462        let mut values = values.clone();
463        for value in values.values_mut() {
464            value.relocate(domain_steps)?;
465        }
466        if let Some((resource, identity)) = binding {
467            {
468                let index = match identity {
469                    ResourceIdentity::Generated(index) => index,
470                    ResourceIdentity::Existing(_) => 0,
471                };
472                let marker = format!("{}0", resource.prefix);
473                let declaration = crate::operations::entity(schema, entity)?;
474                for (name, value) in &mut values {
475                    let field = declaration
476                        .fields
477                        .iter()
478                        .find(|field| field.name == *name)
479                        .ok_or(DatabaseError::InvalidContract)?;
480                    if let WriteValue::Literal(literal) = value {
481                        if field.kind == FieldType::Text
482                            && literal.as_str() == Some(marker.as_str())
483                        {
484                            *value = WriteValue::GeneratedText {
485                                index,
486                                prefix: resource.prefix.clone(),
487                            };
488                        } else if field.kind == FieldType::Json {
489                            *value = WriteValue::GeneratedJson {
490                                index,
491                                marker: marker.clone(),
492                                template: literal.clone(),
493                            };
494                        } else if name == "entity_id"
495                            && field.kind == FieldType::Integer
496                            && literal.is_null()
497                        {
498                            *value = WriteValue::GeneratedId(index);
499                        }
500                    }
501                    if let ResourceIdentity::Existing(id) = identity {
502                        // Existing-record references resolve before adapter admission.
503                        if value.generated_index() == Some(0)
504                            && matches!(
505                                value,
506                                WriteValue::GeneratedText { .. } | WriteValue::GeneratedJson { .. }
507                            )
508                        {
509                            *value = WriteValue::Literal(resolve_write_value(value, &[Some(id)])?);
510                        } else if name == "entity_id" && matches!(value, WriteValue::GeneratedId(0))
511                        {
512                            *value = WriteValue::Literal(Value::from(id));
513                        }
514                    }
515                }
516            }
517        }
518        composed.steps.push(Mutation::Insert {
519            entity: entity.clone(),
520            values,
521        });
522    }
523    let mut result = adapter.transact(config, schema, &composed)?;
524    result.affected.truncate(domain_steps);
525    result.generated_ids.truncate(domain_steps);
526    Ok(result)
527}
pharos_runtime_database/effects.rs

pharos_runtime_database/
effects.rs
