datasources/
stored_procedure_bulk_write_strategy.rs

1use crate::bulk_write_strategy::BulkWriteStrategy;
2use crate::command::Command;
3use crate::connection::Connection;
4use crate::crud_repository_builder::{write_statements, CRUDRepositoryBuilder};
5use crate::crud_repository_options::CRUDRepositoryOptions;
6use crate::datasource_error::DatasourceError;
7use crate::IWhere;
8use crate::parameter::Parameter;
9use crate::parameter_direction::ParameterDirection;
10use crate::response::Response;
11use crate::result::Result;
12use crate::row::Row;
13use crate::value::Value;
14use crate::ExecuteStoredProcedureStatement;
15use async_trait::async_trait;
16use base64::Engine;
17use chrono::{DateTime, Utc};
18use rust_decimal::Decimal;
19use std::collections::BTreeMap;
20use std::sync::{Arc, OnceLock};
21use uuid::Uuid;
22
23pub struct StoredProcedureBulkWriteStrategy {
24    create: String,
25    update: String,
26    delete: String,
27    command: OnceLock<Arc<dyn Command>>,
28    options: OnceLock<CRUDRepositoryOptions>,
29    connection: OnceLock<Arc<dyn Connection>>,
30}
31
32impl StoredProcedureBulkWriteStrategy {
33    pub fn new(
34        create: impl Into<String>,
35        update: impl Into<String>,
36        delete: impl Into<String>,
37    ) -> Self {
38        Self {
39            create: create.into(),
40            update: update.into(),
41            delete: delete.into(),
42            command: OnceLock::new(),
43            options: OnceLock::new(),
44            connection: OnceLock::new(),
45        }
46    }
47
48    fn command(&self) -> Result<Arc<dyn Command>> {
49        self.command
50            .get()
51            .cloned()
52            .ok_or_else(|| DatasourceError::message("Command required"))
53    }
54
55    fn options(&self) -> Result<&CRUDRepositoryOptions> {
56        self.options
57            .get()
58            .ok_or_else(|| DatasourceError::message("Options required"))
59    }
60
61    fn reverse(&self, item: &Row) -> Result<Row> {
62        let options = self.options()?;
63        let row = match &options.object_converter {
64            Some(converter) => converter.reverse_convert(item)?,
65            None => item.clone(),
66        };
67        match &options.row_converter {
68            Some(converter) => converter.reverse_convert(&row),
69            None => Ok(row),
70        }
71    }
72
73    fn column(&self, property: &str) -> String {
74        self.options()
75            .ok()
76            .and_then(|options| options.property_mappings.as_ref())
77            .map(|mapper| mapper.map(property.to_string()))
78            .unwrap_or_else(|| property.to_string())
79    }
80
81    fn row_of(&self, item: &Row, create: bool) -> Result<Row> {
82        let stored = self.reverse(item)?;
83        let connection = self
84            .connection
85            .get()
86            .cloned()
87            .ok_or_else(|| DatasourceError::message("Connection required"))?;
88        let built = CRUDRepositoryBuilder::parts(connection, self.options()?.clone())?;
89        let statements = write_statements(&built.statements);
90        let mut row = Row::new();
91        for name in statements.read_columns() {
92            let physical = self.column(name);
93            if create && self.identity(name) && !assigned(stored.get(name).or_else(|| stored.get(self.ids()[0].as_str()))) {
94                continue;
95            }
96            let value = stored
97                .iter()
98                .find(|(key, _)| *key == name || self.column(key) == physical)
99                .map(|(_, value)| value.clone())
100                .unwrap_or(Value::Null);
101            row.insert(physical, value);
102        }
103        Ok(row)
104    }
105
106    fn identity(&self, column: &str) -> bool {
107        self.ids().len() == 1 && self.ids()[0] == column
108    }
109
110    async fn call(&self, name: &str, items: &[Row]) -> Result<Response> {
111        self.call_with(name, items, false).await
112    }
113
114    async fn call_with(&self, name: &str, items: &[Row], generated: bool) -> Result<Response> {
115        if items.is_empty() {
116            return Ok(Response::new(Value::Null, 0, None));
117        }
118        let rows = items
119            .iter()
120            .map(|item| self.row_of(item, generated))
121            .collect::<Result<Vec<_>>>()?;
122        let payload = Parameter::new("payload", json(&rows));
123        let parameters = if generated && self.ids().len() == 1 {
124            vec![
125                payload,
126                Parameter::with_direction("ids", Value::Null, ParameterDirection::InputOutput),
127            ]
128        } else {
129            vec![payload]
130        };
131        let statement = ExecuteStoredProcedureStatement::new(name, &parameters);
132        let response = self
133            .command()?
134            .execute_statement(&statement, Some(&parameters))
135            .await?;
136        Ok(response)
137    }
138
139    fn ids(&self) -> &[String] {
140        self.options
141            .get()
142            .map(|options| options.ids.as_slice())
143            .unwrap_or(&[])
144    }
145
146    fn apply_ids(&self, items: &mut [Row], parameters: Option<&[Parameter]>) {
147        if self.ids().len() != 1 {
148            return;
149        }
150        let found = parameters.and_then(|list| list.iter().find(|parameter| parameter.name == "ids"));
151        let ids = parse_ids(found.map(|parameter| &parameter.value));
152        let column = self.ids()[0].clone();
153        for (item, id) in items.iter_mut().zip(ids) {
154            item.insert(column.clone(), Value::Int(id));
155        }
156    }
157}
158
159#[async_trait]
160impl BulkWriteStrategy for StoredProcedureBulkWriteStrategy {
161    fn attach(
162        &self,
163        command: Arc<dyn Command>,
164        options: CRUDRepositoryOptions,
165        connection: Arc<dyn Connection>,
166    ) {
167        let _ = self.command.set(command);
168        let _ = self.options.set(options);
169        let _ = self.connection.set(connection);
170    }
171
172    fn persist_children(&self) -> bool {
173        true
174    }
175
176    async fn create(&self, items: &mut [Row]) -> Result<Response> {
177        let response = self.call_with(&self.create, items, true).await?;
178        self.apply_ids(items, response.parameters.as_deref());
179        Ok(response)
180    }
181
182    async fn update(&self, items: &[Row], _by_where: Option<&dyn IWhere>) -> Result<Response> {
183        self.call(&self.update, items).await
184    }
185
186    async fn delete(&self, items: &[Row], _by_where: Option<&dyn IWhere>) -> Result<Response> {
187        self.call(&self.delete, items).await
188    }
189}
190
191fn assigned(value: Option<&Value>) -> bool {
192    match value {
193        None | Some(Value::Null) => false,
194        Some(Value::Int(0)) => false,
195        Some(Value::Text(text)) => !text.is_empty(),
196        Some(_) => true,
197    }
198}
199
200fn parse_ids(value: Option<&Value>) -> Vec<i64> {
201    let text = match value {
202        Some(Value::Text(text)) => text.trim(),
203        Some(Value::Int(number)) => return vec![*number],
204        Some(Value::Binary(bytes)) => std::str::from_utf8(bytes).unwrap_or("").trim(),
205        _ => return vec![],
206    };
207    if text.is_empty() || text == "[]" || text == "null" {
208        return vec![];
209    }
210    text.trim_start_matches('[')
211        .trim_end_matches(']')
212        .split(',')
213        .filter_map(|part| part.trim().parse().ok())
214        .collect()
215}
216
217fn json(rows: &[Row]) -> String {
218    let objects: Vec<String> = rows.iter().map(object).collect();
219    format!("[{}]", objects.join(","))
220}
221
222fn object(row: &Row) -> String {
223    let fields: BTreeMap<&str, &Value> = row
224        .iter()
225        .map(|(key, value)| (key.as_str(), value))
226        .collect();
227    let body: Vec<String> = fields
228        .into_iter()
229        .map(|(key, value)| format!("{}:{}", quote(key), scalar(value)))
230        .collect();
231    format!("{{{}}}", body.join(","))
232}
233
234fn scalar(value: &Value) -> String {
235    match value {
236        Value::Null => "null".to_string(),
237        Value::Bool(flag) => if *flag { "true" } else { "false" }.to_string(),
238        Value::Int(number) => number.to_string(),
239        Value::Float(number) => number.to_string(),
240        Value::Text(text) => quote(text),
241        Value::DateTime(date) => quote(&round_trip(*date)),
242        Value::Uuid(id) => quote(&uuid_text(id)),
243        Value::Decimal(number) => quote(&decimal_text(number)),
244        Value::Binary(bytes) => quote(&base64::engine::general_purpose::STANDARD.encode(bytes)),
245    }
246}
247
248fn round_trip(date: DateTime<Utc>) -> String {
249    format!(
250        "{}.{:07}Z",
251        date.format("%Y-%m-%dT%H:%M:%S"),
252        date.timestamp_subsec_nanos() / 100
253    )
254}
255
256fn uuid_text(id: &Uuid) -> String {
257    id.to_string()
258}
259
260fn decimal_text(number: &Decimal) -> String {
261    number.to_string()
262}
263
264fn quote(text: &str) -> String {
265    let mut result = String::from("\"");
266    for ch in text.chars() {
267        match ch {
268            '\\' => result.push_str("\\\\"),
269            '"' => result.push_str("\\\""),
270            '\u{0008}' => result.push_str("\\b"),
271            '\u{000c}' => result.push_str("\\f"),
272            '\n' => result.push_str("\\n"),
273            '\r' => result.push_str("\\r"),
274            '\t' => result.push_str("\\t"),
275            ch if (ch as u32) < 0x20 => result.push_str(&format!("\\u{:04x}", ch as u32)),
276            ch => result.push(ch),
277        }
278    }
279    result.push('"');
280    result
281}
datasources/stored_procedure_bulk_write_strategy.rs

datasources/
stored_procedure_bulk_write_strategy.rs
