use std::{
collections::{HashMap, HashSet},
fmt,
};
use fraiseql_core::{
db::{
DatabaseType,
introspector::{DatabaseIntrospector, RelationInfo},
},
schema::CompiledSchema,
};
pub struct DatabaseValidationReport {
pub warnings: Vec<DatabaseWarning>,
pub native_columns: HashMap<String, HashMap<String, String>>,
}
#[derive(Debug)]
pub enum DatabaseWarning {
MissingRelation {
query_name: String,
sql_source: String,
},
MissingAdditionalView {
query_name: String,
view_name: String,
},
MissingJsonColumn {
query_name: String,
sql_source: String,
column_name: String,
},
WrongJsonColumnType {
query_name: String,
sql_source: String,
column_name: String,
actual_type: String,
},
MissingCursorColumn {
query_name: String,
sql_source: String,
column_name: String,
},
MissingJsonKey {
query_name: String,
sql_source: String,
json_column: String,
field_name: String,
json_key: String,
},
NativeColumnFallback {
query_name: String,
sql_source: String,
arg_name: String,
},
}
impl fmt::Display for DatabaseWarning {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::MissingRelation {
query_name,
sql_source,
} => {
write!(
f,
"query `{query_name}`: sql_source `{sql_source}` does not exist in database"
)
},
Self::MissingAdditionalView {
query_name,
view_name,
} => {
write!(
f,
"query `{query_name}`: additional_view `{view_name}` does not exist in database"
)
},
Self::MissingJsonColumn {
query_name,
sql_source,
column_name,
} => {
write!(
f,
"query `{query_name}`: column `{column_name}` not found on `{sql_source}`"
)
},
Self::WrongJsonColumnType {
query_name,
sql_source,
column_name,
actual_type,
} => {
write!(
f,
"query `{query_name}`: column `{column_name}` on `{sql_source}` is `{actual_type}`, expected json/jsonb"
)
},
Self::MissingCursorColumn {
query_name,
sql_source,
column_name,
} => {
write!(
f,
"query `{query_name}`: relay cursor column `{column_name}` not found on `{sql_source}`"
)
},
Self::MissingJsonKey {
query_name,
sql_source,
json_column,
field_name,
json_key,
} => {
write!(
f,
"query `{query_name}`: field `{field_name}` (key `{json_key}`) not found in `{sql_source}.{json_column}` sample data"
)
},
Self::NativeColumnFallback {
query_name,
sql_source,
arg_name,
} => {
write!(
f,
"query `{query_name}`: argument `{arg_name}` will use JSONB extraction \
(`{sql_source}.data->>''{arg_name}''`) — no native column `{arg_name}` found on \
`{sql_source}`. Add a native column with an index for O(log n) lookup."
)
},
}
}
}
pub(crate) fn is_json_type(data_type: &str, db_type: DatabaseType) -> bool {
let lower = data_type.to_lowercase();
match db_type {
DatabaseType::PostgreSQL => lower == "jsonb" || lower == "json",
DatabaseType::MySQL => lower == "json",
DatabaseType::SQLite => lower.contains("json"),
DatabaseType::SQLServer => true,
}
}
fn split_schema_qualified(sql_source: &str) -> (Option<&str>, &str) {
match sql_source.split_once('.') {
Some((schema, table)) => (Some(schema), table),
None => (None, sql_source),
}
}
fn relation_exists(
schema_qualified: &HashMap<(String, String), RelationInfo>,
unqualified: &HashMap<String, Vec<String>>,
sql_source: &str,
) -> bool {
let (schema, name) = split_schema_qualified(sql_source);
if let Some(s) = schema {
schema_qualified.contains_key(&(s.to_string(), name.to_string()))
} else {
unqualified.contains_key(name)
}
}
pub(crate) fn to_snake_case(name: &str) -> String {
let mut result = String::with_capacity(name.len() + 4);
for (i, ch) in name.chars().enumerate() {
if ch.is_uppercase() {
if i > 0 {
result.push('_');
}
result.push(ch.to_lowercase().next().unwrap_or(ch));
} else {
result.push(ch);
}
}
result
}
pub async fn validate_schema_against_database(
schema: &CompiledSchema,
introspector: &impl DatabaseIntrospector,
) -> fraiseql_core::Result<DatabaseValidationReport> {
const AUTO_PARAM_NAMES: &[&str] = &[
"where", "limit", "offset", "orderBy", "first", "last", "after", "before",
];
let mut warnings = Vec::new();
let mut native_columns: HashMap<String, HashMap<String, String>> = HashMap::new();
let db_type = introspector.database_type();
let relations = introspector.list_relations().await?;
let (schema_qualified, unqualified) = build_relation_maps(&relations);
for query in &schema.queries {
if let Some(ref source) = query.sql_source {
if !relation_exists(&schema_qualified, &unqualified, source) {
warnings.push(DatabaseWarning::MissingRelation {
query_name: query.name.clone(),
sql_source: source.clone(),
});
continue; }
let columns = introspector.get_columns(source).await?;
let column_map: HashMap<String, String> =
columns.into_iter().map(|(name, dtype, _)| (name, dtype)).collect();
let jsonb_col = &query.jsonb_column;
if !jsonb_col.is_empty() {
if let Some(actual_type) = column_map.get(jsonb_col) {
if !is_json_type(actual_type, db_type) {
warnings.push(DatabaseWarning::WrongJsonColumnType {
query_name: query.name.clone(),
sql_source: source.clone(),
column_name: jsonb_col.clone(),
actual_type: actual_type.clone(),
});
}
} else {
warnings.push(DatabaseWarning::MissingJsonColumn {
query_name: query.name.clone(),
sql_source: source.clone(),
column_name: jsonb_col.clone(),
});
}
}
if query.relay {
if let Some(ref cursor_col) = query.relay_cursor_column {
if !column_map.contains_key(cursor_col) {
warnings.push(DatabaseWarning::MissingCursorColumn {
query_name: query.name.clone(),
sql_source: source.clone(),
column_name: cursor_col.clone(),
});
}
}
}
if !jsonb_col.is_empty() {
let json_type_ok =
column_map.get(jsonb_col).is_some_and(|t| is_json_type(t, db_type));
if json_type_ok {
validate_json_keys(
schema,
query,
source,
jsonb_col,
introspector,
source, &mut warnings,
)
.await?;
}
}
let direct_args: Vec<&str> = query
.arguments
.iter()
.filter(|a| !AUTO_PARAM_NAMES.contains(&a.name.as_str()))
.map(|a| a.name.as_str())
.collect();
if !direct_args.is_empty() {
let mut query_native: HashMap<String, String> = HashMap::new();
for arg_name in &direct_args {
if let Some(col_type) = column_map.get(*arg_name) {
query_native.insert((*arg_name).to_string(), col_type.clone());
} else {
warnings.push(DatabaseWarning::NativeColumnFallback {
query_name: query.name.clone(),
sql_source: source.clone(),
arg_name: (*arg_name).to_string(),
});
}
}
if !query_native.is_empty() {
native_columns.insert(query.name.clone(), query_native);
}
}
for view in &query.additional_views {
if !relation_exists(&schema_qualified, &unqualified, view) {
warnings.push(DatabaseWarning::MissingAdditionalView {
query_name: query.name.clone(),
view_name: view.clone(),
});
}
}
}
}
for mutation in &schema.mutations {
if let Some(ref source) = mutation.sql_source {
if !relation_exists(&schema_qualified, &unqualified, source) {
warnings.push(DatabaseWarning::MissingRelation {
query_name: mutation.name.clone(),
sql_source: source.clone(),
});
}
}
}
Ok(DatabaseValidationReport {
warnings,
native_columns,
})
}
fn build_relation_maps(
relations: &[RelationInfo],
) -> (HashMap<(String, String), RelationInfo>, HashMap<String, Vec<String>>) {
let mut schema_qualified = HashMap::new();
let mut unqualified: HashMap<String, Vec<String>> = HashMap::new();
for rel in relations {
schema_qualified.insert((rel.schema.clone(), rel.name.clone()), rel.clone());
unqualified.entry(rel.name.clone()).or_default().push(rel.schema.clone());
}
(schema_qualified, unqualified)
}
async fn validate_json_keys(
schema: &CompiledSchema,
query: &fraiseql_core::schema::QueryDefinition,
source: &str,
jsonb_col: &str,
introspector: &impl DatabaseIntrospector,
table_name: &str,
warnings: &mut Vec<DatabaseWarning>,
) -> fraiseql_core::Result<()> {
let samples = introspector.get_sample_json_rows(table_name, jsonb_col, 5).await?;
if samples.is_empty() {
return Ok(());
}
let mut all_keys = HashSet::new();
for sample in &samples {
if let serde_json::Value::Object(map) = sample {
for key in map.keys() {
all_keys.insert(key.clone());
}
}
}
if all_keys.is_empty() {
return Ok(());
}
let type_def = schema.types.iter().find(|t| t.name.as_str() == query.return_type);
if let Some(type_def) = type_def {
for field in &type_def.fields {
let field_str = field.name.as_str();
let json_key = to_snake_case(field_str);
if field_str == "id" || field_str.starts_with("pk_") || field_str.starts_with("fk_") {
continue;
}
if !all_keys.contains(&json_key) && !all_keys.contains(field_str) {
warnings.push(DatabaseWarning::MissingJsonKey {
query_name: query.name.clone(),
sql_source: source.to_string(),
json_column: jsonb_col.to_string(),
field_name: field_str.to_string(),
json_key,
});
}
}
}
Ok(())
}
pub enum AnyIntrospector {
Postgres(fraiseql_core::db::PostgresIntrospector),
#[cfg(feature = "mysql")]
MySql(fraiseql_core::db::MySqlIntrospector),
#[cfg(feature = "sqlite")]
Sqlite(fraiseql_core::db::SqliteIntrospector),
#[cfg(feature = "sqlserver")]
SqlServer(fraiseql_core::db::SqlServerIntrospector),
}
impl DatabaseIntrospector for AnyIntrospector {
async fn list_fact_tables(&self) -> fraiseql_core::Result<Vec<String>> {
match self {
Self::Postgres(i) => i.list_fact_tables().await,
#[cfg(feature = "mysql")]
Self::MySql(i) => i.list_fact_tables().await,
#[cfg(feature = "sqlite")]
Self::Sqlite(i) => i.list_fact_tables().await,
#[cfg(feature = "sqlserver")]
Self::SqlServer(i) => i.list_fact_tables().await,
}
}
async fn get_columns(
&self,
table_name: &str,
) -> fraiseql_core::Result<Vec<(String, String, bool)>> {
match self {
Self::Postgres(i) => i.get_columns(table_name).await,
#[cfg(feature = "mysql")]
Self::MySql(i) => i.get_columns(table_name).await,
#[cfg(feature = "sqlite")]
Self::Sqlite(i) => i.get_columns(table_name).await,
#[cfg(feature = "sqlserver")]
Self::SqlServer(i) => i.get_columns(table_name).await,
}
}
async fn get_indexed_columns(&self, table_name: &str) -> fraiseql_core::Result<Vec<String>> {
match self {
Self::Postgres(i) => i.get_indexed_columns(table_name).await,
#[cfg(feature = "mysql")]
Self::MySql(i) => i.get_indexed_columns(table_name).await,
#[cfg(feature = "sqlite")]
Self::Sqlite(i) => i.get_indexed_columns(table_name).await,
#[cfg(feature = "sqlserver")]
Self::SqlServer(i) => i.get_indexed_columns(table_name).await,
}
}
fn database_type(&self) -> DatabaseType {
match self {
Self::Postgres(i) => i.database_type(),
#[cfg(feature = "mysql")]
Self::MySql(i) => i.database_type(),
#[cfg(feature = "sqlite")]
Self::Sqlite(i) => i.database_type(),
#[cfg(feature = "sqlserver")]
Self::SqlServer(i) => i.database_type(),
}
}
async fn get_sample_jsonb(
&self,
table_name: &str,
column_name: &str,
) -> fraiseql_core::Result<Option<serde_json::Value>> {
match self {
Self::Postgres(i) => i.get_sample_jsonb(table_name, column_name).await,
#[cfg(feature = "mysql")]
Self::MySql(i) => i.get_sample_jsonb(table_name, column_name).await,
#[cfg(feature = "sqlite")]
Self::Sqlite(i) => i.get_sample_jsonb(table_name, column_name).await,
#[cfg(feature = "sqlserver")]
Self::SqlServer(i) => i.get_sample_jsonb(table_name, column_name).await,
}
}
async fn list_relations(&self) -> fraiseql_core::Result<Vec<fraiseql_core::db::RelationInfo>> {
match self {
Self::Postgres(i) => i.list_relations().await,
#[cfg(feature = "mysql")]
Self::MySql(i) => i.list_relations().await,
#[cfg(feature = "sqlite")]
Self::Sqlite(i) => i.list_relations().await,
#[cfg(feature = "sqlserver")]
Self::SqlServer(i) => i.list_relations().await,
}
}
async fn get_sample_json_rows(
&self,
table_name: &str,
column_name: &str,
limit: usize,
) -> fraiseql_core::Result<Vec<serde_json::Value>> {
match self {
Self::Postgres(i) => i.get_sample_json_rows(table_name, column_name, limit).await,
#[cfg(feature = "mysql")]
Self::MySql(i) => i.get_sample_json_rows(table_name, column_name, limit).await,
#[cfg(feature = "sqlite")]
Self::Sqlite(i) => i.get_sample_json_rows(table_name, column_name, limit).await,
#[cfg(feature = "sqlserver")]
Self::SqlServer(i) => i.get_sample_json_rows(table_name, column_name, limit).await,
}
}
}
#[allow(clippy::unused_async)] pub async fn create_introspector(db_url: &str) -> anyhow::Result<AnyIntrospector> {
if db_url.starts_with("postgres") {
use deadpool_postgres::{Config, ManagerConfig, RecyclingMethod, Runtime};
use tokio_postgres::NoTls;
let mut cfg = Config::new();
cfg.url = Some(db_url.to_string());
cfg.manager = Some(ManagerConfig {
recycling_method: RecyclingMethod::Fast,
});
cfg.pool = Some(deadpool_postgres::PoolConfig::new(2));
let pool = cfg
.create_pool(Some(Runtime::Tokio1), NoTls)
.map_err(|e| anyhow::anyhow!("Failed to create PostgreSQL pool: {e}"))?;
Ok(AnyIntrospector::Postgres(fraiseql_core::db::PostgresIntrospector::new(pool)))
} else if db_url.starts_with("mysql") || db_url.starts_with("mariadb") {
#[cfg(feature = "mysql")]
{
use sqlx::mysql::MySqlPool;
let pool = MySqlPool::connect(db_url)
.await
.map_err(|e| anyhow::anyhow!("Failed to create MySQL pool: {e}"))?;
Ok(AnyIntrospector::MySql(fraiseql_core::db::MySqlIntrospector::new(pool)))
}
#[cfg(not(feature = "mysql"))]
{
anyhow::bail!("MySQL support not compiled in. Rebuild with `--features mysql`.")
}
} else if db_url.starts_with("sqlite")
|| std::path::Path::new(db_url)
.extension()
.is_some_and(|ext| ext.eq_ignore_ascii_case("db") || ext.eq_ignore_ascii_case("sqlite"))
{
#[cfg(feature = "sqlite")]
{
use sqlx::sqlite::SqlitePool;
let pool = SqlitePool::connect(db_url)
.await
.map_err(|e| anyhow::anyhow!("Failed to create SQLite pool: {e}"))?;
Ok(AnyIntrospector::Sqlite(fraiseql_core::db::SqliteIntrospector::new(pool)))
}
#[cfg(not(feature = "sqlite"))]
{
anyhow::bail!("SQLite support not compiled in. Rebuild with `--features sqlite`.")
}
} else if db_url.starts_with("mssql") || db_url.starts_with("server=") {
#[cfg(feature = "sqlserver")]
{
use bb8::Pool;
use bb8_tiberius::ConnectionManager;
use tiberius::Config;
let config = Config::from_ado_string(db_url).map_err(|e| {
anyhow::anyhow!("Failed to parse SQL Server connection string: {e}")
})?;
let mgr = ConnectionManager::build(config).map_err(|e| {
anyhow::anyhow!("Failed to build SQL Server connection manager: {e}")
})?;
let pool = Pool::builder()
.max_size(2)
.build(mgr)
.await
.map_err(|e| anyhow::anyhow!("Failed to create SQL Server pool: {e}"))?;
Ok(AnyIntrospector::SqlServer(fraiseql_core::db::SqlServerIntrospector::new(pool)))
}
#[cfg(not(feature = "sqlserver"))]
{
anyhow::bail!(
"SQL Server support not compiled in. Rebuild with `--features sqlserver`."
)
}
} else {
anyhow::bail!("Unrecognized database URL scheme: {db_url}")
}
}