pharos_runtime_database/
integration.rs

1//! Bounded integration delivery operations over the configured backend.
2
3use std::collections::BTreeMap;
4
5use serde_json::{Map, Value};
6
7use crate::DbBackendConfig;
8use crate::operations::{
9    AppSchema, DatabaseError, DatabaseOperations, Mutation, Predicate, Query, SortKey, Transaction,
10    WriteValue,
11};
12
13/// Enqueues one validated delivery without importing the app's entire state.
14/// The adapter configuration and payload policy are checked by the caller.
15pub fn enqueue_delivery(
16    adapter: &dyn DatabaseOperations,
17    config: &DbBackendConfig,
18    schema: &AppSchema,
19    row: &Map<String, Value>,
20) -> Result<i64, DatabaseError> {
21    let entity = schema
22        .entities
23        .iter()
24        .find(|entity| entity.name == "integration_delivery")
25        .ok_or(DatabaseError::InvalidContract)?;
26    let mut values = BTreeMap::new();
27    for field in &entity.fields {
28        if matches!(field.name.as_str(), "id" | "created_at" | "updated_at") {
29            continue;
30        }
31        let source = match field.name.as_str() {
32            "signature_timestamp_header" => "timestamp_header",
33            name => name,
34        };
35        if let Some(value) = row.get(source).filter(|value| !value.is_null()) {
36            if value.as_str().is_some_and(str::is_empty)
37                && matches!(field.name.as_str(), "next_attempt_at" | "completed_at")
38            {
39                continue;
40            }
41            values.insert(field.name.clone(), WriteValue::Literal(value.clone()));
42        }
43    }
44    let result = adapter.transact(
45        config,
46        schema,
47        &Transaction {
48            steps: vec![Mutation::Insert {
49                entity: "integration_delivery".to_owned(),
50                values,
51            }],
52            idempotency_key: None,
53        },
54    )?;
55    result
56        .generated_ids
57        .first()
58        .copied()
59        .flatten()
60        .filter(|id| *id > 0)
61        .ok_or(DatabaseError::Unavailable)
62}
63
64/// Reads one bounded page of persisted integration records for operator inspection.
65pub fn integration_rows(
66    adapter: &dyn DatabaseOperations,
67    config: &DbBackendConfig,
68    schema: &AppSchema,
69    entity_name: &str,
70    limit: u32,
71) -> Result<Vec<Value>, DatabaseError> {
72    let entity = schema
73        .entities
74        .iter()
75        .find(|entity| entity.name == entity_name)
76        .ok_or(DatabaseError::InvalidContract)?;
77    if limit == 0 || limit > 256 {
78        return Err(DatabaseError::InvalidInput);
79    }
80    adapter
81        .query(
82            config,
83            schema,
84            &Query {
85                entity: entity_name.to_owned(),
86                fields: entity
87                    .fields
88                    .iter()
89                    .map(|field| field.name.clone())
90                    .collect(),
91                filter: None,
92                order: vec![SortKey {
93                    field: "id".to_owned(),
94                    descending: true,
95                }],
96                limit,
97                offset: 0,
98            },
99        )
100        .map(|rows| rows.into_iter().map(Value::Object).collect())
101}
102
103/// Reads one persisted delivery after an atomic claim.
104pub fn delivery_by_id(
105    adapter: &dyn DatabaseOperations,
106    config: &DbBackendConfig,
107    schema: &AppSchema,
108    delivery_id: i64,
109) -> Result<Option<Value>, DatabaseError> {
110    let entity = schema
111        .entities
112        .iter()
113        .find(|entity| entity.name == "integration_delivery")
114        .ok_or(DatabaseError::InvalidContract)?;
115    adapter
116        .query(
117            config,
118            schema,
119            &Query {
120                entity: "integration_delivery".to_owned(),
121                fields: entity
122                    .fields
123                    .iter()
124                    .map(|field| field.name.clone())
125                    .collect(),
126                filter: Some(Predicate::Equal("id".to_owned(), Value::from(delivery_id))),
127                order: Vec::new(),
128                limit: 1,
129                offset: 0,
130            },
131        )
132        .map(|rows| rows.into_iter().next().map(Value::Object))
133}
pharos_runtime_database/integration.rs

pharos_runtime_database/
integration.rs
