queryer merger test progress

src/database/executors/mock.rs

@@ -0,0 +1,64 @@
+use crate::database::executors::DatabaseExecutor;
+use serde_json::Value;
+
+#[cfg(test)]
+use std::sync::Mutex;
+
+#[cfg(test)]
+pub struct MockExecutor {
+  pub query_responses: Mutex<Vec<Result<Value, String>>>,
+  pub execute_responses: Mutex<Vec<Result<(), String>>>,
+  pub captured_queries: Mutex<Vec<String>>,
+}
+
+#[cfg(test)]
+impl MockExecutor {
+  pub fn new() -> Self {
+    Self {
+      query_responses: Mutex::new(Vec::new()),
+      execute_responses: Mutex::new(Vec::new()),
+      captured_queries: Mutex::new(Vec::new()),
+    }
+  }
+}
+
+#[cfg(test)]
+impl DatabaseExecutor for MockExecutor {
+  fn query(&self, sql: &str, _args: Option<&[Value]>) -> Result<Value, String> {
+    println!("DEBUG SQL QUERY: {}", sql);
+    self.captured_queries.lock().unwrap().push(sql.to_string());
+    let mut responses = self.query_responses.lock().unwrap();
+    if responses.is_empty() {
+      return Ok(Value::Array(vec![]));
+    }
+    responses.remove(0)
+  }
+
+  fn execute(&self, sql: &str, _args: Option<&[Value]>) -> Result<(), String> {
+    println!("DEBUG SQL EXECUTE: {}", sql);
+    self.captured_queries.lock().unwrap().push(sql.to_string());
+    let mut responses = self.execute_responses.lock().unwrap();
+    if responses.is_empty() {
+      return Ok(());
+    }
+    responses.remove(0)
+  }
+
+  fn auth_user_id(&self) -> Result<String, String> {
+    Ok("00000000-0000-0000-0000-000000000000".to_string())
+  }
+
+  fn timestamp(&self) -> Result<String, String> {
+    Ok("2026-03-10T00:00:00Z".to_string())
+  }
+
+  #[cfg(test)]
+  fn get_queries(&self) -> Vec<String> {
+    self.captured_queries.lock().unwrap().clone()
+  }
+
+  #[cfg(test)]
+  fn reset_mocks(&self) {
+    self.captured_queries.lock().unwrap().clear();
+  }
+}

src/database/executors/mod.rs

@@ -0,0 +1,28 @@
+pub mod mock;
+
+#[cfg(not(test))]
+pub mod pgrx;
+
+use serde_json::Value;
+
+/// An abstraction over database execution to allow for isolated unit testing
+/// without a live Postgres SPI connection.
+pub trait DatabaseExecutor: Send + Sync {
+  /// Executes a query expecting a single JSONB return, representing rows.
+  fn query(&self, sql: &str, args: Option<&[Value]>) -> Result<Value, String>;
+
+  /// Executes an operation (INSERT, UPDATE, DELETE, or pg_notify) that does not return rows.
+  fn execute(&self, sql: &str, args: Option<&[Value]>) -> Result<(), String>;
+
+  /// Returns the current authenticated user's ID
+  fn auth_user_id(&self) -> Result<String, String>;
+
+  /// Returns the current transaction timestamp
+  fn timestamp(&self) -> Result<String, String>;
+
+  #[cfg(test)]
+  fn get_queries(&self) -> Vec<String>;
+
+  #[cfg(test)]
+  fn reset_mocks(&self);
+}

src/database/executors/pgrx.rs

@@ -0,0 +1,96 @@
+use crate::database::executors::DatabaseExecutor;
+use pgrx::prelude::*;
+use serde_json::Value;
+
+/// The production executor that wraps `pgrx::spi::Spi`.
+pub struct SpiExecutor;
+
+impl SpiExecutor {
+  pub fn new() -> Self {
+    Self {}
+  }
+}
+
+impl DatabaseExecutor for SpiExecutor {
+  fn query(&self, sql: &str, args: Option<&[Value]>) -> Result<Value, String> {
+    let mut json_args = Vec::new();
+    let mut args_with_oid: Vec<pgrx::datum::DatumWithOid> = Vec::new();
+    if let Some(params) = args {
+      for val in params {
+        json_args.push(pgrx::JsonB(val.clone()));
+      }
+      for j_val in json_args.into_iter() {
+        args_with_oid.push(pgrx::datum::DatumWithOid::from(j_val));
+      }
+    }
+
+    Spi::connect(|client| {
+      match client.select(sql, Some(args_with_oid.len() as i64), &args_with_oid) {
+        Ok(tup_table) => {
+          let mut results = Vec::new();
+          for row in tup_table {
+            if let Ok(Some(jsonb)) = row.get::<pgrx::JsonB>(1) {
+              results.push(jsonb.0);
+            }
+          }
+          Ok(Value::Array(results))
+        }
+        Err(e) => Err(format!("SPI Query Fetch Failure: {}", e)),
+      }
+    })
+  }
+
+  fn execute(&self, sql: &str, args: Option<&[Value]>) -> Result<(), String> {
+    let mut json_args = Vec::new();
+    let mut args_with_oid: Vec<pgrx::datum::DatumWithOid> = Vec::new();
+    if let Some(params) = args {
+      for val in params {
+        json_args.push(pgrx::JsonB(val.clone()));
+      }
+      for j_val in json_args.into_iter() {
+        args_with_oid.push(pgrx::datum::DatumWithOid::from(j_val));
+      }
+    }
+
+    Spi::connect_mut(|client| {
+      match client.update(sql, Some(args_with_oid.len() as i64), &args_with_oid) {
+        Ok(_) => Ok(()),
+        Err(e) => Err(format!("SPI Execution Failure: {}", e)),
+      }
+    })
+  }
+
+  fn auth_user_id(&self) -> Result<String, String> {
+    Spi::connect(|client| {
+      let mut tup_table = client
+        .select(
+          "SELECT COALESCE(current_setting('auth.user_id', true), 'ffffffff-ffff-ffff-ffff-ffffffffffff')",
+          None,
+          &[],
+        )
+        .map_err(|e| format!("SPI Select Error: {}", e))?;
+
+      let row = tup_table
+        .next()
+        .ok_or("No user id setting returned from context".to_string())?;
+      let user_id: Option<String> = row.get(1).map_err(|e| e.to_string())?;
+
+      user_id.ok_or("Missing user_id".to_string())
+    })
+  }
+
+  fn timestamp(&self) -> Result<String, String> {
+    Spi::connect(|client| {
+      let mut tup_table = client
+        .select("SELECT clock_timestamp()::text", None, &[])
+        .map_err(|e| format!("SPI Select Error: {}", e))?;
+
+      let row = tup_table
+        .next()
+        .ok_or("No clock timestamp returned".to_string())?;
+      let timestamp: Option<String> = row.get(1).map_err(|e| e.to_string())?;
+
+      timestamp.ok_or("Missing timestamp".to_string())
+    })
+  }
+}