1#![cfg(target_arch = "wasm32")]
2#![deny(clippy::str_to_string)]
3
4pub mod convert;
5mod error;
6
7use {
8 async_trait::async_trait,
9 convert::convert,
10 error::{ErrInto, StoreReqIntoFuture},
11 futures::stream::{empty, iter},
12 gloo_utils::format::JsValueSerdeExt,
13 gluesql_core::{
14 data::{Key, Schema, Value},
15 error::{Error, Result},
16 store::{DataRow, Metadata, Planner, RowIter, Store, StoreMut},
17 },
18 idb::{
19 CursorDirection, Database, DatabaseEvent, Factory, ObjectStoreParams, Query,
20 TransactionMode,
21 },
22 send_wrapper::SendWrapper,
23 serde_json::Value as JsonValue,
24 std::sync::{Arc, Mutex},
25 wasm_bindgen::JsValue,
26 web_sys::console,
27};
28
29trait SendWrapperExt: Sized {
30 fn send_wrapper(self) -> SendWrapper<Self> {
31 SendWrapper::new(self)
32 }
33}
34
35impl<T> SendWrapperExt for T {}
36
37const SCHEMA_STORE: &str = "gluesql-schema";
38const DEFAULT_NAMESPACE: &str = "gluesql";
39
40enum AlterType {
41 InsertSchema,
42 DeleteSchema,
43}
44
45pub struct IdbStorage {
46 namespace: String,
47 factory: Factory,
48 database: Database,
49}
50
51impl IdbStorage {
52 pub async fn new(namespace: Option<String>) -> Result<Self> {
53 let factory = Factory::new().err_into()?;
54 let namespace = namespace.as_deref().unwrap_or(DEFAULT_NAMESPACE).to_owned();
55
56 let error = Arc::new(Mutex::new(None));
57 let open_request = {
58 let error = Arc::clone(&error);
59 let mut open_request = factory.open(namespace.as_str(), None).err_into()?;
60 open_request.on_upgrade_needed(move |event| {
61 let database = match event.database().err_into() {
62 Ok(database) => database,
63 Err(e) => {
64 let mut error = match error.lock() {
65 Ok(error) => error,
66 Err(_) => {
67 let msg = JsValue::from_str("infallible - lock acquire failed");
68 console::error_1(&msg);
69 return;
70 }
71 };
72
73 *error = Some(e);
74 return;
75 }
76 };
77
78 if let Err(e) = database
79 .create_object_store(SCHEMA_STORE, ObjectStoreParams::new())
80 .err_into()
81 {
82 let mut error = match error.lock() {
83 Ok(error) => error,
84 Err(_) => {
85 let msg = JsValue::from_str("infallible - lock acquire failed");
86 console::error_1(&msg);
87 return;
88 }
89 };
90
91 *error = Some(e);
92 }
93 });
94
95 open_request
96 };
97
98 let database = open_request.await.err_into()?;
99 let mut error = error
100 .lock()
101 .map_err(|_| Error::StorageMsg("infallible - lock acquire failed".to_owned()))?;
102 if let Some(e) = error.take() {
103 return Err(e);
104 }
105
106 Ok(Self {
107 namespace,
108 factory,
109 database,
110 })
111 }
112
113 pub async fn delete(&self) -> Result<()> {
114 self.factory
115 .delete(&self.namespace)
116 .into_future()
117 .send_wrapper()
118 .await
119 .err_into()
120 }
121
122 async fn alter_object_store(
123 &mut self,
124 table_name: String,
125 alter_type: AlterType,
126 ) -> Result<()> {
127 let version = self.database.version().err_into()? + 1;
128 self.database.close();
129
130 let error = Arc::new(Mutex::new(None));
131 let open_request = {
132 let error = Arc::clone(&error);
133 let mut open_request = self
134 .factory
135 .open(self.namespace.as_str(), Some(version))
136 .err_into()?;
137
138 open_request.on_upgrade_needed(move |event| {
139 let database = match event.database().err_into() {
140 Ok(database) => database,
141 Err(e) => {
142 let mut error = match error.lock() {
143 Ok(error) => error,
144 Err(_) => {
145 let msg = JsValue::from_str("infallible - lock acquire failed");
146 console::error_1(&msg);
147 return;
148 }
149 };
150
151 *error = Some(e);
152 return;
153 }
154 };
155
156 let err = match alter_type {
157 AlterType::InsertSchema => {
158 let mut params = ObjectStoreParams::new();
159 params.auto_increment(true);
160
161 database
162 .create_object_store(&table_name, params)
163 .err_into()
164 .map(|_| ())
165 }
166 AlterType::DeleteSchema => {
167 if !database
168 .store_names()
169 .iter()
170 .any(|name| name == &table_name)
171 {
172 return;
173 }
174 database
175 .delete_object_store(table_name.as_str())
176 .err_into()
177 .map(|_| ())
178 }
179 };
180
181 if let Err(e) = err {
182 let mut error = match error.lock() {
183 Ok(error) => error,
184 Err(_) => {
185 let msg = JsValue::from_str("infallible - lock acquire failed");
186 console::error_1(&msg);
187 return;
188 }
189 };
190
191 *error = Some(e);
192 }
193 });
194
195 open_request
196 };
197
198 self.database = open_request.into_future().send_wrapper().await.err_into()?;
199 let mut error = error
200 .lock()
201 .map_err(|_| Error::StorageMsg("infallible - lock acquire failed".to_owned()))?;
202 if let Some(e) = error.take() {
203 return Err(e);
204 }
205
206 Ok(())
207 }
208}
209
210#[async_trait]
211impl Store for IdbStorage {
212 async fn fetch_all_schemas(&self) -> Result<Vec<Schema>> {
213 let transaction = self
214 .database
215 .transaction(&[SCHEMA_STORE], TransactionMode::ReadOnly)
216 .err_into()?
217 .send_wrapper();
218
219 let store = transaction
220 .object_store(SCHEMA_STORE)
221 .err_into()?
222 .send_wrapper();
223 let schemas = store
224 .get_all(None, None)
225 .into_future()
226 .send_wrapper()
227 .await
228 .err_into()?;
229 let schemas = schemas
230 .into_iter()
231 .map(|schema| {
232 schema
233 .as_string()
234 .as_deref()
235 .ok_or_else(|| {
236 Error::StorageMsg("conflict - invalid schema value: {schema:?}".to_owned())
237 })
238 .and_then(Schema::from_ddl)
239 })
240 .collect::<Result<Vec<Schema>>>()?;
241
242 transaction
243 .take()
244 .commit()
245 .into_future()
246 .send_wrapper()
247 .await
248 .err_into()?;
249 Ok(schemas)
250 }
251
252 async fn fetch_schema(&self, table_name: &str) -> Result<Option<Schema>> {
253 let transaction = self
254 .database
255 .transaction(&[SCHEMA_STORE], TransactionMode::ReadOnly)
256 .err_into()?
257 .send_wrapper();
258
259 let store = transaction
260 .object_store(SCHEMA_STORE)
261 .err_into()?
262 .send_wrapper();
263 let schema = store
264 .get(JsValue::from_str(table_name))
265 .into_future()
266 .send_wrapper()
267 .await
268 .err_into()?
269 .and_then(|schema| JsValue::as_string(&schema))
270 .map(|schema| Schema::from_ddl(schema.as_str()))
271 .transpose()?;
272
273 transaction
274 .take()
275 .commit()
276 .into_future()
277 .send_wrapper()
278 .await
279 .err_into()?;
280 Ok(schema)
281 }
282
283 async fn fetch_data(&self, table_name: &str, target: &Key) -> Result<Option<DataRow>> {
284 let column_defs = self
285 .fetch_schema(table_name)
286 .await?
287 .and_then(|schema| schema.column_defs);
288 let transaction = self
289 .database
290 .transaction(&[table_name], TransactionMode::ReadOnly)
291 .err_into()?
292 .send_wrapper();
293
294 let store = transaction
295 .object_store(table_name)
296 .err_into()?
297 .send_wrapper();
298
299 let key: Value = target.clone().into();
300 let key: JsonValue = key.try_into()?;
301 let key = JsValue::from_serde(&key).err_into()?;
302 let row = store
303 .get(key)
304 .into_future()
305 .send_wrapper()
306 .await
307 .err_into()?;
308 let row = row
309 .map(|row| convert(row, column_defs.as_deref()))
310 .transpose()?;
311
312 transaction
313 .take()
314 .commit()
315 .into_future()
316 .send_wrapper()
317 .await
318 .err_into()?;
319
320 Ok(row)
321 }
322
323 async fn scan_data<'a>(&'a self, table_name: &str) -> Result<RowIter<'a>> {
324 let column_defs = self
325 .fetch_schema(table_name)
326 .await?
327 .and_then(|schema| schema.column_defs);
328 let transaction = self
329 .database
330 .transaction(&[table_name], TransactionMode::ReadOnly)
331 .err_into()?
332 .send_wrapper();
333
334 let store = transaction
335 .object_store(table_name)
336 .err_into()?
337 .send_wrapper();
338 let mut cursor = match store
339 .open_cursor(None, Some(CursorDirection::Next))
340 .into_future()
341 .send_wrapper()
342 .await
343 .err_into()?
344 {
345 Some(cursor) => cursor.into_managed().send_wrapper(),
346 None => {
347 return Ok(Box::pin(empty()));
348 }
349 };
350
351 let mut rows = Vec::new();
352 let mut current_key = cursor
353 .key()
354 .err_into()?
355 .unwrap_or(JsValue::NULL)
356 .send_wrapper();
357 let mut current_row = cursor
358 .value()
359 .err_into()?
360 .unwrap_or(JsValue::NULL)
361 .send_wrapper();
362
363 while !current_key.as_ref().is_null() {
364 {
365 let key_js = current_key.take();
366 let row_js = current_row.take();
367
368 let key: JsonValue = key_js.into_serde().err_into()?;
369 let key: Key = Value::try_from(key)?.try_into()?;
370
371 let row = convert(row_js, column_defs.as_deref())?;
372
373 rows.push((key, row));
374 }
375
376 cursor.advance(1).send_wrapper().await.err_into()?;
377 current_key = cursor
378 .key()
379 .err_into()?
380 .unwrap_or(JsValue::NULL)
381 .send_wrapper();
382 current_row = cursor
383 .value()
384 .err_into()?
385 .unwrap_or(JsValue::NULL)
386 .send_wrapper();
387 }
388
389 transaction
390 .take()
391 .commit()
392 .into_future()
393 .send_wrapper()
394 .await
395 .err_into()?;
396
397 let rows = rows.into_iter().map(Ok);
398 Ok(Box::pin(iter(rows)))
399 }
400}
401
402#[async_trait]
403impl StoreMut for IdbStorage {
404 async fn insert_schema(&mut self, schema: &Schema) -> Result<()> {
405 let schema_exists = self
406 .fetch_schema(&schema.table_name)
407 .await
408 .map_err(|e| e.to_string())
409 .map_err(Error::StorageMsg)?
410 .is_some();
411
412 if !schema_exists {
413 self.alter_object_store(schema.table_name.to_owned(), AlterType::InsertSchema)
414 .await?;
415 }
416
417 let transaction = self
418 .database
419 .transaction(&[SCHEMA_STORE], TransactionMode::ReadWrite)
420 .err_into()?
421 .send_wrapper();
422 let store = transaction
423 .object_store(SCHEMA_STORE)
424 .err_into()?
425 .send_wrapper();
426
427 let key = JsValue::from_str(&schema.table_name).send_wrapper();
428 let schema = JsValue::from(schema.to_ddl()).send_wrapper();
429
430 if schema_exists {
431 store
432 .put(&schema, Some(&key))
433 .into_future()
434 .send_wrapper()
435 .await
436 .err_into()?;
437 } else {
438 store
439 .add(&schema, Some(&key))
440 .into_future()
441 .send_wrapper()
442 .await
443 .err_into()?;
444 };
445
446 transaction
447 .take()
448 .commit()
449 .into_future()
450 .send_wrapper()
451 .await
452 .err_into()
453 .map(|_| ())
454 }
455
456 async fn delete_schema(&mut self, table_name: &str) -> Result<()> {
457 self.alter_object_store(table_name.to_owned(), AlterType::DeleteSchema)
458 .await?;
459
460 let transaction = self
461 .database
462 .transaction(&[SCHEMA_STORE], TransactionMode::ReadWrite)
463 .err_into()?
464 .send_wrapper();
465 let store = transaction
466 .object_store(SCHEMA_STORE)
467 .err_into()?
468 .send_wrapper();
469
470 let key = JsValue::from_str(table_name);
471 store
472 .delete(Query::from(key))
473 .into_future()
474 .send_wrapper()
475 .await
476 .err_into()?;
477
478 transaction
479 .take()
480 .commit()
481 .into_future()
482 .send_wrapper()
483 .await
484 .err_into()
485 .map(|_| ())
486 }
487
488 async fn append_data(&mut self, table_name: &str, new_rows: Vec<DataRow>) -> Result<()> {
489 let transaction = self
490 .database
491 .transaction(&[table_name], TransactionMode::ReadWrite)
492 .err_into()?
493 .send_wrapper();
494 let store = transaction
495 .object_store(table_name)
496 .err_into()?
497 .send_wrapper();
498
499 for data_row in new_rows {
500 let row = match data_row {
501 DataRow::Vec(values) => Value::List(values),
502 DataRow::Map(values) => Value::Map(values),
503 };
504
505 let row = JsonValue::try_from(row)?;
506 let row = JsValue::from_serde(&row).err_into()?;
507 let row = row.send_wrapper();
508
509 store
510 .add(&row, None)
511 .into_future()
512 .send_wrapper()
513 .await
514 .err_into()?;
515 }
516
517 transaction
518 .take()
519 .commit()
520 .into_future()
521 .send_wrapper()
522 .await
523 .err_into()
524 .map(|_| ())
525 }
526
527 async fn insert_data(&mut self, table_name: &str, new_rows: Vec<(Key, DataRow)>) -> Result<()> {
528 let transaction = self
529 .database
530 .transaction(&[table_name], TransactionMode::ReadWrite)
531 .err_into()?
532 .send_wrapper();
533 let store = transaction
534 .object_store(table_name)
535 .err_into()?
536 .send_wrapper();
537
538 for (key, data_row) in new_rows {
539 let row = match data_row {
540 DataRow::Vec(values) => Value::List(values),
541 DataRow::Map(values) => Value::Map(values),
542 };
543
544 let row = JsonValue::try_from(row)?;
545 let row = JsValue::from_serde(&row).err_into()?;
546 let row = row.send_wrapper();
547
548 let key: JsonValue = Value::from(key).try_into()?;
549 let key = JsValue::from_serde(&key).err_into()?;
550 let key = key.send_wrapper();
551
552 store
553 .put(&row, Some(&key))
554 .into_future()
555 .send_wrapper()
556 .await
557 .err_into()?;
558 }
559
560 transaction
561 .take()
562 .commit()
563 .into_future()
564 .send_wrapper()
565 .await
566 .err_into()
567 .map(|_| ())
568 }
569
570 async fn delete_data(&mut self, table_name: &str, keys: Vec<Key>) -> Result<()> {
571 let transaction = self
572 .database
573 .transaction(&[table_name], TransactionMode::ReadWrite)
574 .err_into()?
575 .send_wrapper();
576 let store = transaction
577 .object_store(table_name)
578 .err_into()?
579 .send_wrapper();
580
581 for key in keys {
582 let key: JsonValue = Value::from(key).try_into()?;
583 let key = JsValue::from_serde(&key).err_into()?;
584 let key = Query::from(key);
585
586 store
587 .delete(key)
588 .into_future()
589 .send_wrapper()
590 .await
591 .err_into()?;
592 }
593
594 transaction
595 .take()
596 .commit()
597 .into_future()
598 .send_wrapper()
599 .await
600 .err_into()
601 .map(|_| ())
602 }
603}
604
605impl gluesql_core::store::AlterTable for IdbStorage {}
606impl gluesql_core::store::Index for IdbStorage {}
607impl gluesql_core::store::IndexMut for IdbStorage {}
608impl gluesql_core::store::Transaction for IdbStorage {}
609impl Metadata for IdbStorage {}
610impl gluesql_core::store::CustomFunction for IdbStorage {}
611impl gluesql_core::store::CustomFunctionMut for IdbStorage {}
612impl Planner for IdbStorage {}
613
614unsafe impl Send for IdbStorage {}
619
620unsafe impl Sync for IdbStorage {}