doc_db/
lib.rs

1#[cfg(all(test, not(target_arch = "wasm32")))]
2mod backend_contract;
3#[cfg(not(target_arch = "wasm32"))]
4mod dodb;
5mod memory;
6pub mod mock;
7mod remote;
8mod runtime;
9mod transaction;
10mod trx;
11mod turso;
12
13use anyhow::Result;
14use bytes::Bytes;
15pub use doc_db_protocol::DocDbRequest;
16pub use doc_db_protocol::DocDbRevision;
17use doc_db_protocol::{
18    DocDbCondition, DocDbDocument, DocDbError, DocDbKey, DocDbMutation, DocDbObservedDocument,
19    DocDbOperation, DocDbResponse, DocDbResult, DocDbTransactOutcome,
20};
21#[cfg(not(target_arch = "wasm32"))]
22pub use dodb::{
23    DodbConfig, DodbConnection, FN0_CONTROL_DODB_TENANT_ID, dodb_tenant_id, project_tenant_id,
24};
25pub use libsql_hrana::proto::Value;
26use memory::{MemoryDatabase, MemoryTransaction};
27use remote::RemoteDatabase;
28use std::future::Future;
29pub(crate) use transaction::{
30    ObservedDocument, TransactCondition, TransactConflict, TransactMutation, TransactOutcome,
31    TransactRequest, revision_from_backend, revision_to_backend, validate_transact_request,
32};
33pub use trx::{
34    ConflictDetails, ConflictKey, DocGet, DocHandle, DocKey, Document, Trx, TrxControl, TrxRead,
35    TrxResult,
36};
37use turso::{TursoDatabase, TursoTransaction};
38
39pub fn text_value(s: impl Into<String>) -> Value {
40    Value::Text {
41        value: s.into().into(),
42    }
43}
44
45pub fn integer_value(i: i64) -> Value {
46    Value::Integer { value: i }
47}
48
49pub enum WriteOp {
50    Insert {
51        pk: String,
52        sk: String,
53        data: Vec<u8>,
54    },
55    Update {
56        pk: String,
57        sk: String,
58        expected_version: i64,
59        data: Vec<u8>,
60    },
61    Delete {
62        pk: String,
63        sk: String,
64        expected_version: i64,
65    },
66}
67
68pub struct CommitOutcome {
69    pub affected_counts: Vec<u64>,
70    pub conflict: Option<ConflictInfo>,
71}
72
73pub struct ConflictInfo {
74    pub step_index: usize,
75    pub message: String,
76}
77
78pub struct RawStatement {
79    pub sql: String,
80    pub args: Vec<Value>,
81}
82
83pub struct RawStatementResult {
84    pub column_names: Vec<String>,
85    pub rows: Vec<Vec<Value>>,
86    pub affected_row_count: u64,
87    pub rows_read: u64,
88    pub rows_written: u64,
89    pub query_duration_ms: f64,
90}
91
92pub enum RawTransactionOutcome {
93    Committed {
94        statement_results: Vec<RawStatementResult>,
95    },
96    RolledBack {
97        failed_statement_index: usize,
98        error_message: String,
99    },
100}
101
102pub fn turso() -> Database {
103    let url = std::env::var("TURSO_URL").expect("TURSO_URL must be set");
104    let auth_token = std::env::var("TURSO_AUTH_TOKEN").expect("TURSO_AUTH_TOKEN must be set");
105    turso_with_config(url, auth_token)
106}
107
108pub fn turso_with_config(url: String, auth_token: String) -> Database {
109    Database {
110        inner: DatabaseInner::Turso(TursoDatabase::new(url, auth_token)),
111        mock_state: mock::MockState::default(),
112    }
113}
114
115pub fn memory() -> Database {
116    Database {
117        inner: DatabaseInner::Memory(MemoryDatabase::new()),
118        mock_state: mock::MockState::default(),
119    }
120}
121
122pub fn semantic() -> Database {
123    let url = std::env::var("FN0_DOC_DB_URL").expect("FN0_DOC_DB_URL must be set");
124    semantic_with_config(url)
125}
126
127/// Creates the backend-neutral document database used by normal fn0 guests.
128///
129/// The runtime supplies the endpoint and routes it through the semantic doc-db
130/// RPC boundary. The host backend is intentionally not part of this API.
131pub fn database() -> Database {
132    semantic()
133}
134
135pub fn semantic_with_config(url: String) -> Database {
136    Database {
137        inner: DatabaseInner::Remote(RemoteDatabase::new(url)),
138        mock_state: mock::MockState::default(),
139    }
140}
141
142#[cfg(not(target_arch = "wasm32"))]
143pub fn dodb_with_connection(connection: &DodbConnection, project_id: &str) -> Result<Database> {
144    Ok(Database {
145        inner: DatabaseInner::Dodb(connection.database(project_id)?),
146        mock_state: mock::MockState::default(),
147    })
148}
149
150#[derive(Clone)]
151pub struct Database {
152    inner: DatabaseInner,
153    mock_state: mock::MockState,
154}
155
156impl Database {
157    pub async fn get(&self, pk: &str, sk: &str) -> Result<Option<Bytes>> {
158        if let Some(result) = self.mock_state.try_match(mock::MockOp::Get, pk, sk) {
159            return match result {
160                mock::MockResult::OkGet(data) => Ok(data.map(Bytes::from)),
161                mock::MockResult::Err(msg) => Err(anyhow::anyhow!("{}", msg)),
162                _ => unreachable!(),
163            };
164        }
165        match &self.inner {
166            DatabaseInner::Turso(db) => db.get(pk, sk).await,
167            DatabaseInner::Memory(db) => db.get(pk, sk).await,
168            DatabaseInner::Remote(db) => db.get(pk, sk).await,
169            #[cfg(not(target_arch = "wasm32"))]
170            DatabaseInner::Dodb(db) => db.get(pk, sk).await,
171        }
172    }
173
174    pub async fn put(&self, pk: &str, sk: &str, data: &[u8]) -> Result<()> {
175        if let Some(result) = self.mock_state.try_match(mock::MockOp::Put, pk, sk) {
176            return match result {
177                mock::MockResult::OkVoid => Ok(()),
178                mock::MockResult::Err(msg) => Err(anyhow::anyhow!("{}", msg)),
179                _ => unreachable!(),
180            };
181        }
182        match &self.inner {
183            DatabaseInner::Turso(db) => db.put(pk, sk, data).await,
184            DatabaseInner::Memory(db) => db.put(pk, sk, data).await,
185            DatabaseInner::Remote(db) => db.put(pk, sk, data).await,
186            #[cfg(not(target_arch = "wasm32"))]
187            DatabaseInner::Dodb(db) => db.put(pk, sk, data).await,
188        }
189    }
190
191    pub async fn delete(&self, pk: &str, sk: &str) -> Result<()> {
192        if let Some(result) = self.mock_state.try_match(mock::MockOp::Delete, pk, sk) {
193            return match result {
194                mock::MockResult::OkVoid => Ok(()),
195                mock::MockResult::Err(msg) => Err(anyhow::anyhow!("{}", msg)),
196                _ => unreachable!(),
197            };
198        }
199        match &self.inner {
200            DatabaseInner::Turso(db) => db.delete(pk, sk).await,
201            DatabaseInner::Memory(db) => db.delete(pk, sk).await,
202            DatabaseInner::Remote(db) => db.delete(pk, sk).await,
203            #[cfg(not(target_arch = "wasm32"))]
204            DatabaseInner::Dodb(db) => db.delete(pk, sk).await,
205        }
206    }
207
208    // --- Mock API ---
209
210    pub fn mock_get(&self, pk: &str, sk: &str) -> mock::MockGetBuilder<'_> {
211        mock::MockGetBuilder::new(self, pk.to_string(), sk.to_string())
212    }
213
214    pub fn mock_put(&self, pk: &str, sk: &str) -> mock::MockPutBuilder<'_> {
215        mock::MockPutBuilder::new(self, pk.to_string(), sk.to_string())
216    }
217
218    pub fn mock_delete(&self, pk: &str, sk: &str) -> mock::MockDeleteBuilder<'_> {
219        mock::MockDeleteBuilder::new(self, pk.to_string(), sk.to_string())
220    }
221
222    pub fn clear_mocks(&self) {
223        self.mock_state.clear();
224    }
225
226    pub(crate) fn add_mock_rule(&self, rule: mock::MockRule) {
227        self.mock_state.push(rule);
228    }
229
230    #[tracing::instrument(skip_all, fields(pk = %pk.as_ref(), limit = limit))]
231    pub async fn query<S1: AsRef<str>, S2: AsRef<str>>(
232        &self,
233        pk: S1,
234        after_sk: Option<S2>,
235        limit: usize,
236    ) -> Result<Vec<(String, Bytes)>> {
237        match &self.inner {
238            DatabaseInner::Turso(db) => db.query(pk, after_sk, limit).await,
239            DatabaseInner::Memory(db) => db.query(pk, after_sk, limit).await,
240            DatabaseInner::Remote(db) => {
241                db.query(pk.as_ref(), after_sk.as_ref().map(AsRef::as_ref), limit)
242                    .await
243            }
244            #[cfg(not(target_arch = "wasm32"))]
245            DatabaseInner::Dodb(db) => {
246                db.query(pk.as_ref(), after_sk.as_ref().map(AsRef::as_ref), limit)
247                    .await
248            }
249        }
250    }
251
252    #[tracing::instrument(skip_all, fields(limit = limit))]
253    pub async fn scan(
254        &self,
255        after: Option<(&str, &str)>,
256        limit: usize,
257    ) -> Result<Vec<(String, String, Bytes)>> {
258        match &self.inner {
259            DatabaseInner::Turso(db) => db.scan(after, limit).await,
260            DatabaseInner::Memory(db) => db.scan(after, limit).await,
261            DatabaseInner::Remote(db) => db.scan(after, limit).await,
262            #[cfg(not(target_arch = "wasm32"))]
263            DatabaseInner::Dodb(db) => db.scan(after, limit).await,
264        }
265    }
266
267    pub async fn execute_semantic(
268        &self,
269        request: doc_db_protocol::DocDbRequest,
270    ) -> Result<doc_db_protocol::DocDbResponse> {
271        if matches!(&self.inner, DatabaseInner::Remote(_)) {
272            anyhow::bail!("semantic execution is only available on a host database")
273        }
274
275        match request.operation {
276            DocDbOperation::Get { key } => {
277                let data = self.get(&key.pk, &key.sk).await?;
278                Ok(DocDbResponse::new(DocDbResult::Get {
279                    data: data.map(|value| doc_db_protocol::BinaryDocument {
280                        data: value.to_vec(),
281                    }),
282                }))
283            }
284            DocDbOperation::Put { key, data } => {
285                self.put(&key.pk, &key.sk, &data).await?;
286                Ok(DocDbResponse::new(DocDbResult::Put))
287            }
288            DocDbOperation::Delete { key } => {
289                self.delete(&key.pk, &key.sk).await?;
290                Ok(DocDbResponse::new(DocDbResult::Delete))
291            }
292            DocDbOperation::Query {
293                pk,
294                after_sk,
295                limit,
296            } => {
297                let limit = match semantic_limit(limit) {
298                    Ok(limit) => limit,
299                    Err(message) => {
300                        return Ok(DocDbResponse::error(DocDbError::InvalidRequest { message }));
301                    }
302                };
303                let documents = self
304                    .query(&pk, after_sk.as_deref(), limit)
305                    .await?
306                    .into_iter()
307                    .map(|(sk, data)| DocDbDocument {
308                        key: DocDbKey::new(pk.clone(), sk),
309                        data: data.to_vec(),
310                    })
311                    .collect();
312                Ok(DocDbResponse::new(DocDbResult::Query { documents }))
313            }
314            DocDbOperation::Scan { after, limit } => {
315                let limit = match semantic_limit(limit) {
316                    Ok(limit) => limit,
317                    Err(message) => {
318                        return Ok(DocDbResponse::error(DocDbError::InvalidRequest { message }));
319                    }
320                };
321                let after_refs = after.as_ref().map(|key| (key.pk.as_str(), key.sk.as_str()));
322                let documents = self
323                    .scan(after_refs, limit)
324                    .await?
325                    .into_iter()
326                    .map(|(pk, sk, data)| DocDbDocument {
327                        key: DocDbKey::new(pk, sk),
328                        data: data.to_vec(),
329                    })
330                    .collect();
331                Ok(DocDbResponse::new(DocDbResult::Scan { documents }))
332            }
333            DocDbOperation::GetObserved { key } => {
334                let document = match self.get_observed(&key.pk, &key.sk).await? {
335                    ObservedDocument::Present { data, revision } => {
336                        DocDbObservedDocument::Present {
337                            data: data.to_vec(),
338                            revision,
339                        }
340                    }
341                    ObservedDocument::Missing { revision } => {
342                        DocDbObservedDocument::Missing { revision }
343                    }
344                };
345                Ok(DocDbResponse::new(DocDbResult::GetObserved { document }))
346            }
347            DocDbOperation::Transact {
348                conditions,
349                mutations,
350            } => {
351                let conditions = conditions
352                    .into_iter()
353                    .map(|condition| match condition {
354                        DocDbCondition::RevisionEquals {
355                            key,
356                            expected_revision,
357                        } => TransactCondition::RevisionEquals {
358                            pk: key.pk,
359                            sk: key.sk,
360                            expected_revision,
361                        },
362                        DocDbCondition::Exists { key } => TransactCondition::Exists {
363                            pk: key.pk,
364                            sk: key.sk,
365                        },
366                        DocDbCondition::NotExists { key } => TransactCondition::NotExists {
367                            pk: key.pk,
368                            sk: key.sk,
369                        },
370                    })
371                    .collect::<Vec<_>>();
372                let mutations = mutations
373                    .into_iter()
374                    .map(|mutation| match mutation {
375                        DocDbMutation::Put { key, data } => TransactMutation::Put {
376                            pk: key.pk,
377                            sk: key.sk,
378                            data,
379                        },
380                        DocDbMutation::Delete { key } => TransactMutation::Delete {
381                            pk: key.pk,
382                            sk: key.sk,
383                        },
384                    })
385                    .collect::<Vec<_>>();
386                let request = TransactRequest {
387                    conditions,
388                    mutations,
389                };
390                if let Err(error) = validate_transact_request(&request) {
391                    return Ok(DocDbResponse::error(DocDbError::InvalidRequest {
392                        message: error.to_string(),
393                    }));
394                }
395                let outcome = self.transact(&request).await?;
396                let outcome = match outcome.conflict {
397                    Some(conflict) => DocDbTransactOutcome::Conflict {
398                        condition_index: conflict.condition_index,
399                    },
400                    None => DocDbTransactOutcome::Committed,
401                };
402                Ok(DocDbResponse::new(DocDbResult::Transact { outcome }))
403            }
404            DocDbOperation::AdminPurgeProject { .. } => {
405                Ok(DocDbResponse::error(DocDbError::InvalidRequest {
406                    message: "admin purge must be handled by the trusted host service".to_string(),
407                }))
408            }
409        }
410    }
411
412    pub async fn admin_purge_project(&self, project_id: &str) -> Result<u64> {
413        match &self.inner {
414            DatabaseInner::Remote(db) => db.admin_purge_project(project_id).await,
415            _ => anyhow::bail!("admin project purge requires the semantic remote database"),
416        }
417    }
418
419    #[tracing::instrument(skip_all)]
420    pub async fn transaction(&self) -> Result<Transaction> {
421        match &self.inner {
422            DatabaseInner::Turso(db) => Ok(Transaction {
423                inner: TransactionInner::Turso(db.transaction().await?),
424            }),
425            DatabaseInner::Memory(db) => Ok(Transaction {
426                inner: TransactionInner::Memory(db.transaction().await?),
427            }),
428            DatabaseInner::Remote(_) => {
429                anyhow::bail!("explicit transactions are not supported by semantic doc-db RPC")
430            }
431            #[cfg(not(target_arch = "wasm32"))]
432            DatabaseInner::Dodb(_) => {
433                anyhow::bail!("explicit transactions are not supported by the dodb backend")
434            }
435        }
436    }
437
438    #[tracing::instrument(
439        skip_all,
440        fields(conditions = request.conditions.len(), mutations = request.mutations.len())
441    )]
442    pub(crate) async fn transact(&self, request: &TransactRequest) -> Result<TransactOutcome> {
443        validate_transact_request(request)?;
444        match &self.inner {
445            DatabaseInner::Turso(db) => db.transact(request).await,
446            DatabaseInner::Memory(db) => db.transact(request).await,
447            DatabaseInner::Remote(db) => db.transact(request).await,
448            #[cfg(not(target_arch = "wasm32"))]
449            DatabaseInner::Dodb(db) => db.transact(request).await,
450        }
451    }
452
453    #[tracing::instrument(skip_all)]
454    pub async fn trx<F, Fut, Out, Cancel, E>(&self, f: F) -> TrxResult<Out, Cancel, E>
455    where
456        F: FnMut(Trx) -> Fut,
457        Fut: Future<Output = Result<TrxControl<Out, Cancel>, E>>,
458        E: From<anyhow::Error>,
459    {
460        trx::run(self.clone(), f).await
461    }
462
463    #[tracing::instrument(skip_all, fields(sql = %sql))]
464    pub async fn execute_raw(
465        &self,
466        sql: &str,
467        args: Vec<Value>,
468        want_rows: bool,
469    ) -> Result<Vec<Vec<Value>>> {
470        match &self.inner {
471            DatabaseInner::Turso(db) => db.execute_raw(sql, args, want_rows).await,
472            DatabaseInner::Memory(db) => db.execute_raw(sql, args, want_rows).await,
473            DatabaseInner::Remote(_) => {
474                anyhow::bail!("raw SQL is not supported by semantic doc-db RPC")
475            }
476            #[cfg(not(target_arch = "wasm32"))]
477            DatabaseInner::Dodb(_) => {
478                anyhow::bail!("raw SQL is not supported by the dodb backend")
479            }
480        }
481    }
482
483    /// Runs the statements as a single transaction (all-or-nothing) and
484    /// returns, per statement, the result rows together with the engine's own
485    /// `rows_read` / `rows_written` counters — the numbers Turso bills by.
486    /// A statement failure rolls the whole transaction back and is reported as
487    /// [`RawTransactionOutcome::RolledBack`], not as `Err`.
488    ///
489    /// Turso backend only; the in-memory test backend rejects it.
490    #[tracing::instrument(skip_all, fields(statements = statements.len()))]
491    pub async fn execute_raw_transactional(
492        &self,
493        statements: &[RawStatement],
494    ) -> Result<RawTransactionOutcome> {
495        match &self.inner {
496            DatabaseInner::Turso(db) => db.execute_raw_transactional(statements, true).await,
497            DatabaseInner::Memory(_) => {
498                anyhow::bail!("execute_raw_transactional is only supported on the Turso backend")
499            }
500            DatabaseInner::Remote(_) => {
501                anyhow::bail!("raw SQL is not supported by semantic doc-db RPC")
502            }
503            #[cfg(not(target_arch = "wasm32"))]
504            DatabaseInner::Dodb(_) => {
505                anyhow::bail!("raw SQL is not supported by the dodb backend")
506            }
507        }
508    }
509
510    #[tracing::instrument(skip_all, fields(statements = statements.len()))]
511    pub async fn execute_raw_transactional_readonly(
512        &self,
513        statements: &[RawStatement],
514    ) -> Result<RawTransactionOutcome> {
515        if statements.iter().any(|statement| {
516            !statement
517                .sql
518                .trim_start()
519                .to_ascii_uppercase()
520                .starts_with("SELECT ")
521        }) {
522            anyhow::bail!("execute_raw_transactional_readonly accepts SELECT statements only")
523        }
524        match &self.inner {
525            DatabaseInner::Turso(db) => db.execute_raw_transactional(statements, false).await,
526            DatabaseInner::Memory(_) => {
527                anyhow::bail!(
528                    "execute_raw_transactional_readonly is only supported on the Turso backend"
529                )
530            }
531            DatabaseInner::Remote(_) => {
532                anyhow::bail!("raw SQL is not supported by semantic doc-db RPC")
533            }
534            #[cfg(not(target_arch = "wasm32"))]
535            DatabaseInner::Dodb(_) => {
536                anyhow::bail!("raw SQL is not supported by the dodb backend")
537            }
538        }
539    }
540
541    #[tracing::instrument(skip_all)]
542    pub(crate) async fn execute_op(&self, op: DbOp) -> Result<DbResult> {
543        match op {
544            DbOp::Get { pk, sk } => self.get(&pk, &sk).await.map(DbResult::Single),
545            DbOp::Query {
546                pk,
547                after_sk,
548                limit,
549            } => self
550                .query(&pk, after_sk.as_deref(), limit.unwrap_or(usize::MAX))
551                .await
552                .map(DbResult::Multiple),
553            DbOp::Put { pk, sk, data } => {
554                self.put(&pk, &sk, &data).await?;
555                Ok(DbResult::Done)
556            }
557            DbOp::Delete { pk, sk } => {
558                self.delete(&pk, &sk).await?;
559                Ok(DbResult::Done)
560            }
561        }
562    }
563
564    #[tracing::instrument(skip_all, fields(pk = %pk, sk = %sk))]
565    pub(crate) async fn get_observed(&self, pk: &str, sk: &str) -> Result<ObservedDocument> {
566        match &self.inner {
567            DatabaseInner::Turso(db) => db.get_observed(pk, sk).await,
568            DatabaseInner::Memory(db) => db.get_observed(pk, sk).await,
569            DatabaseInner::Remote(db) => db.get_observed(pk, sk).await,
570            #[cfg(not(target_arch = "wasm32"))]
571            DatabaseInner::Dodb(db) => db.get_observed(pk, sk).await,
572        }
573    }
574}
575
576fn semantic_limit(limit: u64) -> std::result::Result<usize, String> {
577    usize::try_from(limit).map_err(|_| "limit does not fit the host usize".to_string())
578}
579
580#[cfg(all(test, not(target_arch = "wasm32")))]
581mod semantic_tests {
582    use super::*;
583    use doc_db_protocol::{
584        DocDbCondition, DocDbKey, DocDbMutation, DocDbObservedDocument, DocDbOperation,
585        DocDbRequest, DocDbResult, DocDbRevision, DocDbTransactOutcome,
586    };
587    use std::sync::{Arc, Mutex};
588    use std::time::Duration;
589    use tokio::io::{AsyncReadExt, AsyncWriteExt};
590    use tokio::net::{TcpListener, TcpStream};
591    use tokio::sync::Barrier;
592
593    static DATABASE_ENV_LOCK: Mutex<()> = Mutex::new(());
594
595    #[derive(serde::Deserialize, serde::Serialize)]
596    struct TransactionTestDoc {
597        id: String,
598    }
599
600    impl Document for TransactionTestDoc {
601        fn key(&self) -> DocKey {
602            DocKey::new("TransactionTestDoc", format!("id={}", self.id))
603        }
604    }
605
606    async fn read_http_body(stream: &mut TcpStream) -> Vec<u8> {
607        let mut bytes = Vec::new();
608        let header_end;
609        loop {
610            let mut chunk = [0u8; 4096];
611            let read = stream.read(&mut chunk).await.unwrap();
612            assert!(read > 0);
613            bytes.extend_from_slice(&chunk[..read]);
614            if let Some(end) = bytes.windows(4).position(|window| window == b"\r\n\r\n") {
615                header_end = end + 4;
616                break;
617            }
618        }
619        let headers = String::from_utf8_lossy(&bytes[..header_end]);
620        let content_length = headers
621            .lines()
622            .find_map(|line| {
623                let (name, value) = line.split_once(':')?;
624                name.eq_ignore_ascii_case("content-length")
625                    .then_some(value.trim())
626            })
627            .unwrap()
628            .parse::<usize>()
629            .unwrap();
630        while bytes.len() < header_end + content_length {
631            let mut chunk = [0u8; 4096];
632            let read = stream.read(&mut chunk).await.unwrap();
633            assert!(read > 0);
634            bytes.extend_from_slice(&chunk[..read]);
635        }
636        bytes[header_end..header_end + content_length].to_vec()
637    }
638
639    async fn start_remote_server(
640        responses: Vec<DocDbResponse>,
641    ) -> (String, tokio::task::JoinHandle<Vec<DocDbOperation>>) {
642        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
643        let address = listener.local_addr().unwrap();
644        let handle = tokio::spawn(async move {
645            let mut operations = Vec::new();
646            for response in responses {
647                let (mut stream, _) = listener.accept().await.unwrap();
648                let body = read_http_body(&mut stream).await;
649                let request = doc_db_protocol::decode_request(&body).unwrap();
650                operations.push(request.operation);
651                let response_body = doc_db_protocol::encode_response(&response).unwrap();
652                let header = format!(
653                    "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
654                    response_body.len()
655                );
656                stream.write_all(header.as_bytes()).await.unwrap();
657                stream.write_all(&response_body).await.unwrap();
658            }
659            operations
660        });
661        (format!("http://{address}/rpc"), handle)
662    }
663
664    async fn start_barrier_remote_server() -> (String, tokio::task::JoinHandle<Vec<DocDbOperation>>)
665    {
666        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
667        let address = listener.local_addr().unwrap();
668        let barrier = Arc::new(Barrier::new(2));
669        let operations = Arc::new(Mutex::new(Vec::new()));
670        let handle = tokio::spawn(async move {
671            let mut handlers = Vec::new();
672            for _ in 0..2 {
673                let (stream, _) = listener.accept().await.unwrap();
674                let barrier = barrier.clone();
675                let operations = operations.clone();
676                handlers.push(tokio::spawn(async move {
677                    let mut stream = stream;
678                    let body = read_http_body(&mut stream).await;
679                    let request = doc_db_protocol::decode_request(&body).unwrap();
680                    let response = match &request.operation {
681                        DocDbOperation::Get { key } => DocDbResponse::new(DocDbResult::Get {
682                            data: Some(doc_db_protocol::BinaryDocument {
683                                data: key.sk.as_bytes().to_vec(),
684                            }),
685                        }),
686                        operation => panic!("unexpected operation: {operation:?}"),
687                    };
688                    operations.lock().unwrap().push(request.operation);
689                    barrier.wait().await;
690                    let response_body = doc_db_protocol::encode_response(&response).unwrap();
691                    let header = format!(
692                        "HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
693                        response_body.len()
694                    );
695                    stream.write_all(header.as_bytes()).await.unwrap();
696                    stream.write_all(&response_body).await.unwrap();
697                }));
698            }
699            for handler in handlers {
700                handler.await.unwrap();
701            }
702            operations.lock().unwrap().clone()
703        });
704        (format!("http://{address}/rpc"), handle)
705    }
706
707    #[tokio::test]
708    async fn maps_semantic_operations_to_the_backend_contract() {
709        let database = memory();
710        let binary_data = vec![0, 1, 127, 128, 255, 0];
711
712        let put = database
713            .execute_semantic(DocDbRequest::new(DocDbOperation::Put {
714                key: DocDbKey::new("pk", "b"),
715                data: binary_data.clone(),
716            }))
717            .await
718            .unwrap();
719        assert_eq!(put.result, DocDbResult::Put);
720
721        database
722            .execute_semantic(DocDbRequest::new(DocDbOperation::Put {
723                key: DocDbKey::new("pk", "a"),
724                data: b"a".to_vec(),
725            }))
726            .await
727            .unwrap();
728        database
729            .execute_semantic(DocDbRequest::new(DocDbOperation::Put {
730                key: DocDbKey::new("other", "a"),
731                data: b"other".to_vec(),
732            }))
733            .await
734            .unwrap();
735
736        let get = database
737            .execute_semantic(DocDbRequest::new(DocDbOperation::Get {
738                key: DocDbKey::new("pk", "b"),
739            }))
740            .await
741            .unwrap();
742        assert_eq!(
743            get.result,
744            DocDbResult::Get {
745                data: Some(doc_db_protocol::BinaryDocument { data: binary_data })
746            }
747        );
748
749        let query = database
750            .execute_semantic(DocDbRequest::new(DocDbOperation::Query {
751                pk: "pk".to_string(),
752                after_sk: Some("a".to_string()),
753                limit: 1,
754            }))
755            .await
756            .unwrap();
757        assert_eq!(
758            query.result,
759            DocDbResult::Query {
760                documents: vec![doc_db_protocol::DocDbDocument {
761                    key: DocDbKey::new("pk", "b"),
762                    data: vec![0, 1, 127, 128, 255, 0],
763                }]
764            }
765        );
766
767        let scan = database
768            .execute_semantic(DocDbRequest::new(DocDbOperation::Scan {
769                after: Some(DocDbKey::new("other", "a")),
770                limit: 1,
771            }))
772            .await
773            .unwrap();
774        assert_eq!(
775            scan.result,
776            DocDbResult::Scan {
777                documents: vec![doc_db_protocol::DocDbDocument {
778                    key: DocDbKey::new("pk", "a"),
779                    data: b"a".to_vec(),
780                }]
781            }
782        );
783
784        let observed = database
785            .execute_semantic(DocDbRequest::new(DocDbOperation::GetObserved {
786                key: DocDbKey::new("pk", "a"),
787            }))
788            .await
789            .unwrap();
790        let DocDbResult::GetObserved { document } = &observed.result else {
791            panic!("expected observed result");
792        };
793        assert!(matches!(document, DocDbObservedDocument::Present { .. }));
794        let DocDbObservedDocument::Present { revision, .. } = document else {
795            panic!("expected present observation");
796        };
797        assert_eq!(*revision, DocDbRevision::new(0));
798    }
799
800    #[tokio::test]
801    async fn maps_transaction_conditions_and_mutations_and_conflict_condition() {
802        let database = memory();
803        for (sk, data) in [
804            ("version", b"version" as &[u8]),
805            ("update", b"old" as &[u8]),
806            ("delete", b"delete" as &[u8]),
807        ] {
808            database.put("pk", sk, data).await.unwrap();
809        }
810
811        let committed = database
812            .execute_semantic(DocDbRequest::new(DocDbOperation::Transact {
813                conditions: vec![
814                    DocDbCondition::RevisionEquals {
815                        key: DocDbKey::new("pk", "version"),
816                        expected_revision: DocDbRevision::new(0),
817                    },
818                    DocDbCondition::NotExists {
819                        key: DocDbKey::new("pk", "missing"),
820                    },
821                    DocDbCondition::NotExists {
822                        key: DocDbKey::new("pk", "insert"),
823                    },
824                    DocDbCondition::RevisionEquals {
825                        key: DocDbKey::new("pk", "update"),
826                        expected_revision: DocDbRevision::new(0),
827                    },
828                    DocDbCondition::RevisionEquals {
829                        key: DocDbKey::new("pk", "delete"),
830                        expected_revision: DocDbRevision::new(0),
831                    },
832                ],
833                mutations: vec![
834                    DocDbMutation::Put {
835                        key: DocDbKey::new("pk", "insert"),
836                        data: b"insert".to_vec(),
837                    },
838                    DocDbMutation::Put {
839                        key: DocDbKey::new("pk", "update"),
840                        data: b"new".to_vec(),
841                    },
842                    DocDbMutation::Delete {
843                        key: DocDbKey::new("pk", "delete"),
844                    },
845                ],
846            }))
847            .await
848            .unwrap();
849        assert_eq!(
850            committed.result,
851            DocDbResult::Transact {
852                outcome: DocDbTransactOutcome::Committed
853            }
854        );
855        assert_eq!(
856            database.get("pk", "insert").await.unwrap(),
857            Some(Bytes::from_static(b"insert"))
858        );
859        assert_eq!(
860            database.get("pk", "update").await.unwrap(),
861            Some(Bytes::from_static(b"new"))
862        );
863        assert_eq!(database.get("pk", "delete").await.unwrap(), None);
864
865        let conflict = database
866            .execute_semantic(DocDbRequest::new(DocDbOperation::Transact {
867                conditions: vec![DocDbCondition::RevisionEquals {
868                    key: DocDbKey::new("pk", "update"),
869                    expected_revision: DocDbRevision::new(0),
870                }],
871                mutations: vec![],
872            }))
873            .await
874            .unwrap();
875        assert_eq!(
876            conflict.result,
877            DocDbResult::Transact {
878                outcome: DocDbTransactOutcome::Conflict { condition_index: 0 }
879            }
880        );
881    }
882
883    #[tokio::test]
884    async fn tuple_send_with_uses_concurrent_single_operation_requests() {
885        struct RawGet(&'static str);
886
887        impl DbRequest for RawGet {
888            type Output = Option<Vec<u8>>;
889
890            fn prepare(self) -> Prepared<Self::Output> {
891                Prepared {
892                    ops: vec![DbOp::Get {
893                        pk: "pk".to_string(),
894                        sk: self.0.to_string(),
895                    }],
896                    parse: Box::new(|iter| match iter.next().unwrap() {
897                        DbResult::Single(value) => Ok(value.map(|bytes| bytes.to_vec())),
898                        result => panic!("unexpected result: {result:?}"),
899                    }),
900                }
901            }
902        }
903
904        let (url, server) = start_barrier_remote_server().await;
905        let database = semantic_with_config(url);
906        let result = tokio::time::timeout(
907            Duration::from_secs(1),
908            (RawGet("first"), RawGet("second")).send_with(&database),
909        )
910        .await
911        .expect("tuple requests should be concurrent")
912        .unwrap();
913        assert_eq!(result, (Some(b"first".to_vec()), Some(b"second".to_vec())));
914
915        let operations = server.await.unwrap();
916        assert_eq!(operations.len(), 2);
917        assert!(
918            operations
919                .iter()
920                .all(|operation| matches!(operation, DocDbOperation::Get { .. }))
921        );
922    }
923
924    #[tokio::test]
925    async fn tuple_send_with_settles_other_operations_before_returning_error() {
926        struct FailingGet;
927
928        impl DbRequest for FailingGet {
929            type Output = Option<Vec<u8>>;
930
931            fn prepare(self) -> Prepared<Self::Output> {
932                Prepared {
933                    ops: vec![DbOp::Get {
934                        pk: "pk".to_string(),
935                        sk: "failure".to_string(),
936                    }],
937                    parse: Box::new(|iter| match iter.next().unwrap() {
938                        DbResult::Single(value) => Ok(value.map(|bytes| bytes.to_vec())),
939                        result => panic!("unexpected result: {result:?}"),
940                    }),
941                }
942            }
943        }
944
945        struct Put;
946
947        impl DbRequest for Put {
948            type Output = ();
949
950            fn prepare(self) -> Prepared<Self::Output> {
951                Prepared {
952                    ops: vec![DbOp::Put {
953                        pk: "pk".to_string(),
954                        sk: "completed".to_string(),
955                        data: b"done".to_vec(),
956                    }],
957                    parse: Box::new(|iter| match iter.next().unwrap() {
958                        DbResult::Done => Ok(()),
959                        result => panic!("unexpected result: {result:?}"),
960                    }),
961                }
962            }
963        }
964
965        let database = memory();
966        database
967            .mock_get("pk", "failure")
968            .returns_err("expected failure");
969        let error = (FailingGet, Put).send_with(&database).await.unwrap_err();
970        assert_eq!(error.to_string(), "expected failure");
971        assert_eq!(
972            database.get("pk", "completed").await.unwrap().as_deref(),
973            Some(b"done".as_slice())
974        );
975    }
976
977    #[tokio::test]
978    async fn rejects_duplicate_transaction_keys_as_invalid_requests() {
979        let database = memory();
980
981        let duplicate_condition = database
982            .execute_semantic(DocDbRequest::new(DocDbOperation::Transact {
983                conditions: vec![
984                    DocDbCondition::Exists {
985                        key: DocDbKey::new("pk", "duplicate"),
986                    },
987                    DocDbCondition::NotExists {
988                        key: DocDbKey::new("pk", "duplicate"),
989                    },
990                ],
991                mutations: vec![],
992            }))
993            .await
994            .unwrap();
995        assert!(matches!(
996            duplicate_condition.result,
997            DocDbResult::Error {
998                error: doc_db_protocol::DocDbError::InvalidRequest { .. }
999            }
1000        ));
1001
1002        let duplicate_mutation = database
1003            .execute_semantic(DocDbRequest::new(DocDbOperation::Transact {
1004                conditions: vec![],
1005                mutations: vec![
1006                    DocDbMutation::Put {
1007                        key: DocDbKey::new("pk", "duplicate"),
1008                        data: b"one".to_vec(),
1009                    },
1010                    DocDbMutation::Delete {
1011                        key: DocDbKey::new("pk", "duplicate"),
1012                    },
1013                ],
1014            }))
1015            .await
1016            .unwrap();
1017        assert!(matches!(
1018            duplicate_mutation.result,
1019            DocDbResult::Error {
1020                error: doc_db_protocol::DocDbError::InvalidRequest { .. }
1021            }
1022        ));
1023    }
1024
1025    #[tokio::test]
1026    async fn converts_single_remote_responses_to_database_results() {
1027        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
1028        let responses = vec![
1029            DocDbResponse::new(DocDbResult::Get {
1030                data: Some(doc_db_protocol::BinaryDocument {
1031                    data: vec![0, 128, 255],
1032                }),
1033            }),
1034            DocDbResponse::new(DocDbResult::Put),
1035            DocDbResponse::new(DocDbResult::Delete),
1036            DocDbResponse::new(DocDbResult::Query {
1037                documents: vec![doc_db_protocol::DocDbDocument {
1038                    key: DocDbKey::new("pk", "query-sk"),
1039                    data: b"query".to_vec(),
1040                }],
1041            }),
1042            DocDbResponse::new(DocDbResult::Scan {
1043                documents: vec![doc_db_protocol::DocDbDocument {
1044                    key: DocDbKey::new("scan-pk", "scan-sk"),
1045                    data: b"scan".to_vec(),
1046                }],
1047            }),
1048            DocDbResponse::new(DocDbResult::GetObserved {
1049                document: DocDbObservedDocument::Present {
1050                    data: b"observed".to_vec(),
1051                    revision: DocDbRevision::new(7),
1052                },
1053            }),
1054            DocDbResponse::new(DocDbResult::Transact {
1055                outcome: DocDbTransactOutcome::Conflict { condition_index: 4 },
1056            }),
1057        ];
1058        let (url, server) = start_remote_server(responses).await;
1059        let database = semantic_with_config(url);
1060
1061        assert_eq!(
1062            database.get("pk", "get").await.unwrap(),
1063            Some(Bytes::from(vec![0, 128, 255]))
1064        );
1065        database.put("pk", "put", b"put").await.unwrap();
1066        database.delete("pk", "delete").await.unwrap();
1067        assert_eq!(
1068            database.query("pk", Some("after"), 3).await.unwrap(),
1069            vec![("query-sk".to_string(), Bytes::from_static(b"query"))]
1070        );
1071        assert_eq!(
1072            database
1073                .scan(Some(("after-pk", "after-sk")), 2)
1074                .await
1075                .unwrap(),
1076            vec![(
1077                "scan-pk".to_string(),
1078                "scan-sk".to_string(),
1079                Bytes::from_static(b"scan")
1080            )]
1081        );
1082        let observed = database.get_observed("pk", "present").await.unwrap();
1083        let ObservedDocument::Present { revision, .. } = &observed else {
1084            panic!("expected present observation");
1085        };
1086        assert_eq!(*revision, DocDbRevision::new(7));
1087        let conflict = database
1088            .transact(&TransactRequest {
1089                conditions: vec![TransactCondition::NotExists {
1090                    pk: "pk".to_string(),
1091                    sk: "missing".to_string(),
1092                }],
1093                mutations: vec![],
1094            })
1095            .await
1096            .unwrap();
1097        assert_eq!(conflict.conflict.unwrap().condition_index, 4);
1098
1099        let operations = server.await.unwrap();
1100        assert!(matches!(operations[0], DocDbOperation::Get { .. }));
1101        assert!(matches!(operations[1], DocDbOperation::Put { .. }));
1102        assert!(matches!(operations[2], DocDbOperation::Delete { .. }));
1103        assert!(matches!(
1104            operations[3],
1105            DocDbOperation::Query { limit: 3, .. }
1106        ));
1107        assert!(matches!(
1108            operations[4],
1109            DocDbOperation::Scan { limit: 2, .. }
1110        ));
1111        assert!(matches!(operations[5], DocDbOperation::GetObserved { .. }));
1112        assert!(matches!(operations[6], DocDbOperation::Transact { .. }));
1113    }
1114
1115    #[tokio::test]
1116    async fn rejects_invalid_remote_transaction_conflict_index() {
1117        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
1118        let (url, server) = start_remote_server(vec![DocDbResponse::new(DocDbResult::Transact {
1119            outcome: DocDbTransactOutcome::Conflict { condition_index: 3 },
1120        })])
1121        .await;
1122        let database = semantic_with_config(url);
1123
1124        let result = database
1125            .trx(|trx| async move {
1126                let handle = trx.create(TransactionTestDoc {
1127                    id: "invalid-conflict-index".to_string(),
1128                })?;
1129                drop(handle);
1130                trx.commit::<(), ()>(())
1131            })
1132            .await;
1133
1134        let is_invalid_conflict_error = match result {
1135            TrxResult::Err(error) => error
1136                .to_string()
1137                .contains("backend returned invalid transaction conflict condition_index"),
1138            _ => false,
1139        };
1140        assert!(is_invalid_conflict_error);
1141        let operations = server.await.unwrap();
1142        assert!(matches!(
1143            operations.as_slice(),
1144            [DocDbOperation::Transact { conditions, mutations }]
1145                if conditions.len() == 1 && mutations.len() == 1
1146        ));
1147    }
1148
1149    #[tokio::test]
1150    async fn database_uses_the_semantic_rpc_endpoint() {
1151        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
1152        let (url, server) = start_remote_server(vec![DocDbResponse::new(DocDbResult::Get {
1153            data: Some(doc_db_protocol::BinaryDocument {
1154                data: b"from-semantic-rpc".to_vec(),
1155            }),
1156        })])
1157        .await;
1158        let _environment_guard = DATABASE_ENV_LOCK.lock().unwrap();
1159        let previous = std::env::var_os("FN0_DOC_DB_URL");
1160        // Environment mutation is process-global; this test serializes it with
1161        // the other constructor test state in this module.
1162        unsafe { std::env::set_var("FN0_DOC_DB_URL", &url) };
1163
1164        let result = database().get("pk", "sk").await;
1165
1166        match previous {
1167            Some(value) => unsafe { std::env::set_var("FN0_DOC_DB_URL", value) },
1168            None => unsafe { std::env::remove_var("FN0_DOC_DB_URL") },
1169        }
1170        assert_eq!(
1171            result.unwrap(),
1172            Some(Bytes::from_static(b"from-semantic-rpc"))
1173        );
1174        let operations = server.await.unwrap();
1175        assert!(matches!(
1176            operations.as_slice(),
1177            [DocDbOperation::Get { .. }]
1178        ));
1179    }
1180}
1181
1182#[derive(Clone)]
1183enum DatabaseInner {
1184    Turso(TursoDatabase),
1185    Memory(MemoryDatabase),
1186    Remote(RemoteDatabase),
1187    #[cfg(not(target_arch = "wasm32"))]
1188    Dodb(dodb::DodbDatabase),
1189}
1190
1191pub struct Transaction {
1192    inner: TransactionInner,
1193}
1194
1195enum TransactionInner {
1196    Turso(TursoTransaction),
1197    Memory(MemoryTransaction),
1198}
1199
1200impl Transaction {
1201    #[tracing::instrument(skip_all, fields(pk = %pk, sk = %sk))]
1202    pub async fn get(&mut self, pk: &str, sk: &str) -> Result<Option<Bytes>> {
1203        match &mut self.inner {
1204            TransactionInner::Turso(tx) => tx.get(pk, sk).await,
1205            TransactionInner::Memory(tx) => tx.get(pk, sk).await,
1206        }
1207    }
1208
1209    #[tracing::instrument(skip_all, fields(pk = %pk, sk = %sk, bytes = data.len()))]
1210    pub async fn put(&mut self, pk: &str, sk: &str, data: &[u8]) -> Result<()> {
1211        match &mut self.inner {
1212            TransactionInner::Turso(tx) => tx.put(pk, sk, data).await,
1213            TransactionInner::Memory(tx) => tx.put(pk, sk, data).await,
1214        }
1215    }
1216
1217    #[tracing::instrument(skip_all, fields(pk = %pk, sk = %sk))]
1218    pub async fn delete(&mut self, pk: &str, sk: &str) -> Result<()> {
1219        match &mut self.inner {
1220            TransactionInner::Turso(tx) => tx.delete(pk, sk).await,
1221            TransactionInner::Memory(tx) => tx.delete(pk, sk).await,
1222        }
1223    }
1224
1225    #[tracing::instrument(skip_all)]
1226    pub async fn commit(self) -> Result<()> {
1227        match self.inner {
1228            TransactionInner::Turso(tx) => tx.commit().await,
1229            TransactionInner::Memory(tx) => tx.commit().await,
1230        }
1231    }
1232
1233    #[tracing::instrument(skip_all)]
1234    pub async fn rollback(self) -> Result<()> {
1235        match self.inner {
1236            TransactionInner::Turso(tx) => tx.rollback().await,
1237            TransactionInner::Memory(tx) => tx.rollback().await,
1238        }
1239    }
1240}
1241
1242#[doc(hidden)]
1243#[derive(Debug)]
1244pub enum DbOp {
1245    Get {
1246        pk: String,
1247        sk: String,
1248    },
1249    Query {
1250        pk: String,
1251        after_sk: Option<String>,
1252        limit: Option<usize>,
1253    },
1254    Put {
1255        pk: String,
1256        sk: String,
1257        data: Vec<u8>,
1258    },
1259    Delete {
1260        pk: String,
1261        sk: String,
1262    },
1263}
1264
1265#[doc(hidden)]
1266#[derive(Debug)]
1267pub enum DbResult {
1268    Single(Option<Bytes>),
1269    Multiple(Vec<(String, Bytes)>),
1270    Done,
1271}
1272
1273#[doc(hidden)]
1274pub type DbResultParser<O> = Box<dyn FnOnce(&mut std::vec::IntoIter<DbResult>) -> Result<O> + Send>;
1275
1276#[doc(hidden)]
1277pub struct Prepared<O> {
1278    pub ops: Vec<DbOp>,
1279    pub parse: DbResultParser<O>,
1280}
1281
1282#[allow(async_fn_in_trait)]
1283pub trait DbRequest: Sized {
1284    type Output;
1285    fn prepare(self) -> Prepared<Self::Output>;
1286
1287    /// Executes every prepared operation concurrently.
1288    ///
1289    /// These operations are independent and non-atomic: execution order is
1290    /// unspecified, and a failure in one operation does not roll back the
1291    /// others. All started operations settle before an error is returned; the
1292    /// first error in input order wins when more than one operation fails.
1293    async fn send_with(self, db: &Database) -> Result<Self::Output> {
1294        let prepared = self.prepare();
1295        let results = futures::future::join_all(
1296            prepared
1297                .ops
1298                .into_iter()
1299                .map(|operation| db.execute_op(operation)),
1300        )
1301        .await;
1302        let mut settled = Vec::with_capacity(results.len());
1303        let mut first_error = None;
1304        for result in results {
1305            match result {
1306                Ok(result) => settled.push(result),
1307                Err(error) => {
1308                    if first_error.is_none() {
1309                        first_error = Some(error);
1310                    }
1311                }
1312            }
1313        }
1314        if let Some(error) = first_error {
1315            return Err(error);
1316        }
1317        let mut iter = settled.into_iter();
1318        (prepared.parse)(&mut iter)
1319    }
1320}
1321
1322macro_rules! impl_db_request_tuple {
1323    ($($T:ident),+) => {
1324        #[allow(non_snake_case)]
1325        impl<$($T: DbRequest),+> DbRequest for ($($T,)+)
1326        where $($T::Output: 'static),+
1327        {
1328            type Output = ($($T::Output,)+);
1329            fn prepare(self) -> Prepared<Self::Output> {
1330                let ($($T,)+) = self;
1331                $(let $T = $T.prepare();)+
1332                let mut ops = Vec::new();
1333                $(ops.extend($T.ops);)+
1334                Prepared {
1335                    ops,
1336                    parse: Box::new(move |iter| {
1337                        Ok(($(($T.parse)(iter)?,)+))
1338                    }),
1339                }
1340            }
1341        }
1342    };
1343}
1344
1345impl_db_request_tuple!(A);
1346impl_db_request_tuple!(A, B);
1347impl_db_request_tuple!(A, B, C);
1348impl_db_request_tuple!(A, B, C, D);
1349impl_db_request_tuple!(A, B, C, D, E);
1350impl_db_request_tuple!(A, B, C, D, E, F);
1351impl_db_request_tuple!(A, B, C, D, E, F, G);
1352impl_db_request_tuple!(A, B, C, D, E, F, G, H);
1353impl_db_request_tuple!(A, B, C, D, E, F, G, H, I);
1354impl_db_request_tuple!(A, B, C, D, E, F, G, H, I, J);
1355impl_db_request_tuple!(A, B, C, D, E, F, G, H, I, J, K);
1356impl_db_request_tuple!(A, B, C, D, E, F, G, H, I, J, K, L);
1357
1358impl<T: DbRequest> DbRequest for Vec<T>
1359where
1360    T::Output: 'static,
1361{
1362    type Output = Vec<T::Output>;
1363    fn prepare(self) -> Prepared<Self::Output> {
1364        let mut all_ops = Vec::new();
1365        let mut parsers: Vec<DbResultParser<T::Output>> = Vec::new();
1366        for item in self {
1367            let p = item.prepare();
1368            all_ops.extend(p.ops);
1369            parsers.push(p.parse);
1370        }
1371        Prepared {
1372            ops: all_ops,
1373            parse: Box::new(move |iter| parsers.into_iter().map(|p| p(iter)).collect()),
1374        }
1375    }
1376}
doc_db/lib.rs

doc_db/
lib.rs
