feat: upgrade v8 runtime crates

src/exec/fixtures/sqlite.rs

@@ -0,0 +1,241 @@
+use deno_core::futures::StreamExt;
+use deno_core::{Extension, OpState};
+use elm_project_utils::ElmResult;
+use serde::{Deserialize, Serialize};
+use serde_json::value::Number;
+use serde_json::Value;
+use sqlx::sqlite::SqlitePool;
+use sqlx::Row;
+use sqlx::{Column, TypeInfo, ValueRef};
+use std::thread::JoinHandle;
+use std::time::Instant;
+use tokio;
+use tracing::{info_span, Instrument};
+
+type SQLWorkerMailbox = std::sync::mpsc::Sender<(
+    oneshot::Sender<ElmResult<Vec<Vec<String>>, AstridQueryError>>,
+    Vec<(bool, String, Vec<String>)>,
+)>;
+
+pub(crate) fn init(db_pool: SqlitePool) -> Result<(Extension, JoinHandle<()>), ()> {
+    let (worker_mailbox, rx) = std::sync::mpsc::channel::<(
+        oneshot::Sender<ElmResult<_, _>>,
+        Vec<(bool, String, Vec<String>)>,
+    )>();
+
+    let sql_worker_thread = std::thread::spawn(move || {
+        let worker = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .unwrap();
+        loop {
+            if let Ok((response, queries)) = rx.recv() {
+                // I am not sure if I should only work on one database task at a time, or
+                // submit as many takes as possible.  Just spawning the future onto this
+                // exectutor does not seem to work, even though the docs say the thread pool
+                // will poll the future until it completes.
+                let db_pool_clone = db_pool.clone();
+                let span = info_span!("inside sql queries futures");
+                let f = async move {
+                    let _start = Instant::now();
+                    let db_pool = db_pool_clone;
+                    let mut result: Vec<Vec<String>> = vec![];
+                    let mut failure: Option<AstridQueryError> = None;
+                    for (fetch_all, sql, _args) in queries {
+                        let mut acc = Vec::new();
+                        if fetch_all {
+                            let mut stream = sqlx::query(&sql).fetch(&db_pool);
+                            loop {
+                                match stream.next().await {
+                                    None => break,
+                                    Some(Ok(row)) => {
+                                        // Try to auto marshall the row into a javascript
+                                        // object to make it easier on our no-code users.
+                                        match try_marshal(&row) {
+                                            Ok(json) => acc.push(json.to_string()),
+                                            Err(err) => {
+                                                failure = Some(AstridQueryError::Execute {
+                                                    sql: sql.clone(),
+                                                    message: err.to_string(),
+                                                });
+                                                break;
+                                            }
+                                        }
+                                    }
+                                    Some(Err(err)) => {
+                                        eprintln!("got fetch_all sql error {:?}", err);
+                                        failure = Some(AstridQueryError::Execute {
+                                            sql: sql.clone(),
+                                            message: err.to_string(),
+                                        });
+                                        break;
+                                    }
+                                }
+                            }
+                            result.push(acc);
+                        } else {
+                            match sqlx::query(&sql)
+                                .fetch_one(&db_pool)
+                                .await
+                                .and_then(|row| try_marshal(&row).map(|json| json.to_string()))
+                            {
+                                Ok(s) => result.push(vec![s]),
+                                Err(sqlx::Error::RowNotFound) => {
+                                    failure = Some(AstridQueryError::NotFound { sql });
+                                }
+                                Err(err) => {
+                                    eprintln!("got fetchOne sql error {:?}", err);
+                                    failure = Some(AstridQueryError::Execute {
+                                        sql,
+                                        message: err.to_string(),
+                                    });
+                                }
+                            }
+                        }
+                        if failure.is_some() {
+                            break;
+                        }
+                    }
+                    if let Some(msg) = failure {
+                        response.send(ElmResult::err(msg))
+                    } else {
+                        response.send(ElmResult::ok(result))
+                    }
+                };
+
+                // I found it interesting that the runtime of the future from the viewpoint of
+                // tracing was around 230us for a trivial select 2 rows query, but walltime I
+                // measured was around 700us. So polling the future or waiting for file IO is
+                // more expensive than I thought.
+                worker.block_on(f.instrument(span)).unwrap();
+            } else {
+                break;
+            }
+        }
+    });
+    let worker_mailbox_clone = worker_mailbox.clone();
+    let extension = Extension::builder()
+        .ops(vec![op_starmelon_batch_queries::decl()])
+        .state(move |state| {
+            state.put(worker_mailbox_clone.clone());
+
+            Ok(())
+        })
+        .build();
+
+    Ok((extension, sql_worker_thread))
+}
+
+#[deno_core::op]
+fn op_starmelon_batch_queries(
+    state: &mut OpState,
+    queries: Vec<(bool, String, Vec<String>)>,
+) -> Result<ElmResult<Vec<Vec<String>>, AstridQueryError>, deno_core::error::AnyError> {
+    let worker_mailbox_clone = state.borrow::<SQLWorkerMailbox>();
+
+    let worker_mailbox = worker_mailbox_clone.clone();
+    let (sender, receiver) = oneshot::channel::<ElmResult<Vec<Vec<String>>, _>>();
+
+    let span = info_span!("run sql");
+    let timing_guard = span.enter();
+
+    worker_mailbox.send((sender, queries)).unwrap();
+    let elm_result = receiver.recv().unwrap();
+
+    drop(timing_guard);
+
+    Ok(elm_result)
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+#[serde(tag = "$")]
+enum AstridQueryError {
+    Execute {
+        #[serde(rename = "a")]
+        sql: String,
+        #[serde(rename = "b")]
+        message: String,
+    },
+    NotFound {
+        #[serde(rename = "a")]
+        sql: String,
+    },
+}
+
+fn try_marshal(row: &sqlx::sqlite::SqliteRow) -> Result<serde_json::Value, sqlx::Error> {
+    use serde_json::value::Map;
+    use serde_json::Value;
+    use sqlx::{Column, TypeInfo};
+
+    let mut object = Map::new();
+    for i in 0..row.len() {
+        let value = try_decode_column(row, i)?;
+        let key = row.column(i).name().to_owned();
+        object.insert(key, value);
+    }
+
+    Ok(Value::Object(object))
+}
+
+fn try_decode_column(
+    row: &sqlx::sqlite::SqliteRow,
+    i: usize,
+) -> Result<serde_json::Value, sqlx::Error> {
+    let value = row
+        .try_get_raw(i)
+        .map_err(|err| sqlx::Error::ColumnDecode {
+            index: row.column(i).name().to_owned(),
+            source: Box::new(StringError(format!("{}", err))),
+        })?;
+    match value.type_info().name() {
+        "NULL" => {
+            if let Ok(_x) = row.try_get::<i64, _>(i) {
+                eprintln!("it was actually an int");
+            };
+            Ok(Value::Null)
+        }
+        "TEXT" => Ok(Value::String(row.try_get::<String, _>(i)?)),
+        "REAL" => {
+            let x = row.try_get::<f64, _>(i)?;
+            match Number::from_f64(x) {
+                Some(n) => Ok(Value::Number(n)),
+                None => {
+                    Err(sqlx::Error::ColumnDecode {
+                        index: row.column(i).name().to_owned(),
+                        source: Box::new(StringError(format!("While parsing a SQL type `REAL` I expected a finite number but got {} instead ", x))),
+                    })
+                }
+            }
+        }
+        //"BLOB" =>
+        "INTEGER" => Ok(Value::Number(row.try_get::<i64, _>(i)?.into())),
+        //"NUMERIC" =>
+        "BOOLEAN" => Ok(Value::Bool(row.try_get::<bool, _>(i)?)),
+        //"DATE" =>
+        //"TIME" =>
+        //"DATETIME" =>
+        unknown => Err(sqlx::Error::ColumnDecode {
+            index: row.column(i).name().to_owned(),
+            source: Box::new(StringError(format!(
+                "I don't know how to automatically convert the SQL type `{}`` into a JSON value.",
+                unknown
+            ))),
+        }),
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct StringError(String);
+
+impl ::std::fmt::Display for StringError {
+    fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
+        self.0.fmt(f)
+    }
+}
+
+impl std::error::Error for StringError {
+    #[inline]
+    fn description(&self) -> &str {
+        &self.0
+    }
+}