mod adapter_types;
mod mutations;
mod relay;
use std::sync::Arc;
pub use adapter_types::*;
use async_trait::async_trait;
use fraiseql_error::{FraiseQLError, Result};
pub use mutations::SupportsMutations;
pub use relay::RelayDatabaseAdapter;
use crate::{
types::{
DatabaseType, JsonbValue, PoolMetrics,
sql_hints::{OrderByClause, SqlProjectionHint},
},
where_clause::WhereClause,
};
#[derive(Debug, Clone, Copy)]
pub struct ChangeLogWrite<'a> {
pub object_type: &'a str,
pub modification_type: &'a str,
pub tenant_id: Option<uuid::Uuid>,
pub trace_id: Option<&'a str>,
pub schema_version: Option<&'a str>,
pub trace_context: Option<&'a str>,
pub actor_type: Option<&'a str>,
pub acting_for: Option<uuid::Uuid>,
}
impl<'a> ChangeLogWrite<'a> {
#[must_use]
pub const fn new(object_type: &'a str, modification_type: &'a str) -> Self {
Self {
object_type,
modification_type,
tenant_id: None,
trace_id: None,
schema_version: None,
trace_context: None,
actor_type: None,
acting_for: None,
}
}
#[must_use]
pub const fn with_tenant_id(mut self, tenant_id: Option<uuid::Uuid>) -> Self {
self.tenant_id = tenant_id;
self
}
#[must_use]
pub const fn with_trace_id(mut self, trace_id: Option<&'a str>) -> Self {
self.trace_id = trace_id;
self
}
#[must_use]
pub const fn with_schema_version(mut self, schema_version: Option<&'a str>) -> Self {
self.schema_version = schema_version;
self
}
#[must_use]
pub const fn with_trace_context(mut self, trace_context: Option<&'a str>) -> Self {
self.trace_context = trace_context;
self
}
#[must_use]
pub const fn with_actor_type(mut self, actor_type: Option<&'a str>) -> Self {
self.actor_type = actor_type;
self
}
#[must_use]
pub const fn with_acting_for(mut self, acting_for: Option<uuid::Uuid>) -> Self {
self.acting_for = acting_for;
self
}
}
#[async_trait]
pub trait DatabaseAdapter: Send + Sync {
async fn execute_where_query(
&self,
view: &str,
where_clause: Option<&WhereClause>,
limit: Option<u32>,
offset: Option<u32>,
order_by: Option<&[OrderByClause]>,
) -> Result<Vec<JsonbValue>>;
async fn execute_with_projection(
&self,
view: &str,
projection: Option<&SqlProjectionHint>,
where_clause: Option<&WhereClause>,
limit: Option<u32>,
offset: Option<u32>,
order_by: Option<&[OrderByClause]>,
) -> Result<Vec<JsonbValue>>;
async fn execute_where_query_arc(
&self,
view: &str,
where_clause: Option<&WhereClause>,
limit: Option<u32>,
offset: Option<u32>,
order_by: Option<&[OrderByClause]>,
) -> Result<Arc<Vec<JsonbValue>>> {
self.execute_where_query(view, where_clause, limit, offset, order_by)
.await
.map(Arc::new)
}
async fn execute_with_projection_arc(
&self,
request: &ProjectionRequest<'_>,
) -> Result<Arc<Vec<JsonbValue>>> {
self.execute_with_projection(
request.view,
request.projection,
request.where_clause,
request.limit,
request.offset,
request.order_by,
)
.await
.map(Arc::new)
}
fn database_type(&self) -> DatabaseType;
async fn health_check(&self) -> Result<()>;
fn pool_metrics(&self) -> PoolMetrics;
async fn execute_raw_query(
&self,
sql: &str,
) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>>;
async fn execute_row_query(
&self,
view_name: &str,
columns: &[crate::types::ColumnSpec],
where_sql: Option<&str>,
order_by: Option<&str>,
limit: Option<u32>,
offset: Option<u32>,
) -> Result<Vec<Vec<crate::types::ColumnValue>>> {
use crate::types::ColumnValue;
let mut sql = format!("SELECT * FROM \"{view_name}\"");
if let Some(w) = where_sql {
sql.push_str(" WHERE ");
sql.push_str(w);
}
if let Some(ob) = order_by {
sql.push_str(" ORDER BY ");
sql.push_str(ob);
}
if let Some(l) = limit {
use std::fmt::Write;
let _ = write!(sql, " LIMIT {l}");
}
if let Some(o) = offset {
use std::fmt::Write;
let _ = write!(sql, " OFFSET {o}");
}
let results = self.execute_raw_query(&sql).await?;
Ok(results
.iter()
.map(|row| {
columns
.iter()
.map(|col| {
row.get(&col.name).map_or(ColumnValue::Null, |v| match v {
serde_json::Value::Null => ColumnValue::Null,
serde_json::Value::Bool(b) => ColumnValue::Boolean(*b),
serde_json::Value::Number(n) => {
if let Some(i) = n.as_i64() {
ColumnValue::Int64(i)
} else if let Some(f) = n.as_f64() {
ColumnValue::Float64(f)
} else {
ColumnValue::Text(n.to_string())
}
},
serde_json::Value::String(s) => ColumnValue::Text(s.clone()),
other => ColumnValue::Json(other.to_string()),
})
})
.collect()
})
.collect())
}
async fn execute_parameterized_aggregate(
&self,
sql: &str,
params: &[serde_json::Value],
) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>>;
async fn execute_parameterized_aggregate_with_session(
&self,
sql: &str,
params: &[serde_json::Value],
_session_vars: &[(&str, &str)],
) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
self.execute_parameterized_aggregate(sql, params).await
}
async fn execute_function_call(
&self,
function_name: &str,
_args: &[serde_json::Value],
) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
Err(FraiseQLError::Unsupported {
message: format!(
"Mutations via function calls are not supported by this adapter. \
Function '{function_name}' cannot be executed. \
Use PostgreSQL, MySQL, or SQL Server for mutation support."
),
})
}
fn supports_mutations(&self) -> bool {
true
}
async fn bump_fact_table_versions(&self, _tables: &[String]) -> Result<()> {
Ok(())
}
async fn invalidate_views(&self, _views: &[crate::ViewName]) -> Result<u64> {
Ok(0)
}
async fn invalidate_by_entity(&self, _entity_type: &str, _entity_id: &str) -> Result<u64> {
Ok(0)
}
async fn invalidate_list_queries(&self, views: &[crate::ViewName]) -> Result<u64> {
self.invalidate_views(views).await
}
fn capabilities(&self) -> DatabaseCapabilities {
DatabaseCapabilities::from_database_type(self.database_type())
}
async fn explain_query(
&self,
_sql: &str,
_params: &[serde_json::Value],
) -> Result<serde_json::Value> {
Err(fraiseql_error::FraiseQLError::Unsupported {
message: "EXPLAIN not available for this database adapter".to_string(),
})
}
async fn explain_where_query(
&self,
_view: &str,
_where_clause: Option<&WhereClause>,
_limit: Option<u32>,
_offset: Option<u32>,
) -> Result<serde_json::Value> {
Err(fraiseql_error::FraiseQLError::Unsupported {
message: "EXPLAIN ANALYZE is not available for this database adapter. \
Only PostgreSQL supports explain_where_query."
.to_string(),
})
}
fn mutation_strategy(&self) -> MutationStrategy {
MutationStrategy::FunctionCall
}
async fn execute_function_call_with_session(
&self,
function_name: &str,
args: &[serde_json::Value],
_session_vars: &[(&str, &str)],
) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
self.execute_function_call(function_name, args).await
}
async fn execute_function_call_with_changelog(
&self,
function_name: &str,
args: &[serde_json::Value],
session_vars: &[(&str, &str)],
_changelog: Option<&ChangeLogWrite<'_>>,
) -> Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
self.execute_function_call_with_session(function_name, args, session_vars).await
}
async fn execute_where_query_arc_with_session(
&self,
view: &str,
where_clause: Option<&WhereClause>,
limit: Option<u32>,
offset: Option<u32>,
order_by: Option<&[OrderByClause]>,
_session_vars: &[(&str, &str)],
) -> Result<Arc<Vec<JsonbValue>>> {
self.execute_where_query_arc(view, where_clause, limit, offset, order_by).await
}
async fn execute_with_projection_arc_with_session(
&self,
request: &ProjectionRequest<'_>,
_session_vars: &[(&str, &str)],
) -> Result<Arc<Vec<JsonbValue>>> {
self.execute_with_projection_arc(request).await
}
async fn execute_direct_mutation(
&self,
_ctx: &DirectMutationContext<'_>,
) -> Result<Vec<serde_json::Value>> {
Err(FraiseQLError::Unsupported {
message: "Direct SQL mutations are not supported by this adapter. \
Use execute_function_call for stored-procedure mutations."
.to_string(),
})
}
async fn query_stats(&self, _limit: u32) -> Result<Vec<crate::types::QueryStatEntry>> {
Ok(vec![])
}
async fn query_stats_by_id(&self, id: &str) -> Result<Option<crate::types::QueryStatEntry>> {
let stats = self.query_stats(1000).await?;
Ok(stats.into_iter().find(|e| e.query_id == id))
}
async fn reset_query_stats(&self) -> Result<()> {
Err(FraiseQLError::Unsupported {
message: "Query stats reset is not supported by this database adapter".to_string(),
})
}
fn on_schema_reload(&self) {}
}
#[cfg(test)]
mod tests;