use std::{fs, path::Path, process::Command};
use anyhow::{Context, Result};
use fraiseql_core::schema::{
CompiledSchema, FieldType, InputStyle, MutationOperation, NamingConvention, content_hash_of,
};
use fraiseql_db::postgres::PostgresTlsConfig;
use tracing::{info, warn};
use crate::{
config::{ConfigSource, TomlProjectConfig},
schema::{
CompiledArtifact, ConvertOptions, IntermediateSchema, OptimizationReport, SchemaConverter,
SchemaOptimizer, SchemaValidator,
database_validator::validate_schema_against_database,
function_root_contract::validate_function_roots,
mutation_contract::{Severity, validate_mutation_contract},
pg_catalog::PgCatalog,
},
};
#[derive(Debug, Default)]
pub struct CompileOptions<'a> {
pub input: &'a str,
pub types: Option<&'a str>,
pub schema_dir: Option<&'a str>,
pub type_files: Vec<String>,
pub query_files: Vec<String>,
pub mutation_files: Vec<String>,
pub database: Option<&'a str>,
pub skip_hash: bool,
pub allow_drift: bool,
pub config: Option<ConfigSource>,
}
impl<'a> CompileOptions<'a> {
#[must_use]
pub fn new(input: &'a str) -> Self {
Self {
input,
..Default::default()
}
}
#[must_use]
pub fn with_types(mut self, types: &'a str) -> Self {
self.types = Some(types);
self
}
#[must_use]
pub fn with_schema_dir(mut self, schema_dir: &'a str) -> Self {
self.schema_dir = Some(schema_dir);
self
}
#[must_use]
pub fn with_database(mut self, database: &'a str) -> Self {
self.database = Some(database);
self
}
}
#[allow(clippy::cognitive_complexity)] pub fn load_intermediate_schema(
toml_path: &str,
type_files: &[String],
query_files: &[String],
mutation_files: &[String],
schema_dir: Option<&str>,
types_path: Option<&str>,
) -> Result<IntermediateSchema> {
if !type_files.is_empty() || !query_files.is_empty() || !mutation_files.is_empty() {
info!("Mode: Explicit file lists");
return crate::schema::SchemaMerger::merge_explicit_files(
toml_path,
type_files,
query_files,
mutation_files,
)
.context("Failed to load explicit schema files");
}
if let Some(dir) = schema_dir {
info!("Mode: Auto-discovery from directory: {}", dir);
return crate::schema::SchemaMerger::merge_from_directory(toml_path, dir)
.context("Failed to load schema from directory");
}
if let Some(types) = types_path {
info!("Mode: Language + TOML (types.json + fraiseql.toml)");
return crate::schema::SchemaMerger::merge_files(types, toml_path)
.context("Failed to merge types.json with TOML");
}
let toml_schema = crate::config::TomlSchema::from_file(toml_path)
.context(format!("Failed to load TOML from {toml_path}"))?;
if toml_schema.domain_discovery.enabled {
info!("Mode: TOML-based with domain discovery");
return crate::schema::SchemaMerger::merge_from_domains(toml_path).context(
"Failed to load schema from the configured [domain_discovery] section. Fix the \
error above, or remove the section to compile from TOML definitions only.",
);
}
if !toml_schema.includes.is_empty() {
info!("Mode: TOML-based with schema includes");
return crate::schema::SchemaMerger::merge_with_includes(toml_path).context(
"Failed to load schema from the configured [includes] section. Fix the error \
above, or remove the section to compile from TOML definitions only.",
);
}
info!("No domains or includes configured, using TOML-only definitions");
crate::schema::SchemaMerger::merge_toml_only(toml_path)
.context("Failed to load schema from TOML")
}
#[allow(clippy::cognitive_complexity)] pub async fn compile_to_schema(
opts: CompileOptions<'_>,
) -> Result<(CompiledArtifact, OptimizationReport)> {
info!("Compiling schema: {}", opts.input);
if let Some(db_url) = opts.database {
crate::connection::require_postgres(db_url)?;
}
let input_path = Path::new(opts.input);
if !input_path.exists() {
anyhow::bail!("Input file not found: {}", opts.input);
}
let is_toml = input_path
.extension()
.and_then(|ext| ext.to_str())
.is_some_and(|ext| ext.eq_ignore_ascii_case("toml"));
let mut intermediate: IntermediateSchema = if is_toml {
info!("Using TOML-based workflow");
load_intermediate_schema(
opts.input,
&opts.type_files,
&opts.query_files,
&opts.mutation_files,
opts.schema_dir,
opts.types,
)?
} else {
info!("Using legacy JSON workflow");
let schema_json = fs::read_to_string(input_path).context("Failed to read schema.json")?;
info!("Parsing intermediate schema...");
let raw: serde_json::Value =
serde_json::from_str(&schema_json).context("Failed to parse schema.json")?;
crate::schema::intermediate::reject_drifted_keys(&raw)
.map_err(|e| anyhow::anyhow!("{e}"))?;
let input_has_federation = ["federation", "federation_config"].iter().any(|key| {
raw.get(*key)
.and_then(serde_json::Value::as_object)
.is_some_and(|o| !o.is_empty())
});
let intermediate: IntermediateSchema =
serde_json::from_value(raw).context("Failed to parse schema.json")?;
if input_has_federation && intermediate.federation_config.is_none() {
anyhow::bail!(
"schema.json carries a `federation` block that did not bind into the compiled \
schema — refusing to compile a silently non-federated subgraph. This is a \
compiler bug; please report it."
);
}
intermediate
};
let config_source = match opts.config {
Some(source) => source,
None => ConfigSource::resolve(input_path, None)?,
};
info!("Project config: {config_source}");
let mut auto_error_union = false;
let mut naming_acronyms: Vec<String> = Vec::new();
let mut operation_cost_weights: std::collections::HashMap<String, usize> =
std::collections::HashMap::new();
let mut naming_convention = NamingConvention::CamelCase;
let mut database_tls = if is_toml && opts.database.is_some() {
crate::config::TomlSchema::from_file(opts.input)
.with_context(|| format!("Failed to load TOML from {}", opts.input))?
.database
.postgres_tls()?
} else {
PostgresTlsConfig::default()
};
if let Some(config_path) = config_source.project_config() {
info!("Loading security configuration from {}...", config_path.display());
let config_file = config_path.to_str().ok_or_else(|| {
anyhow::anyhow!("Config path is not valid UTF-8: {}", config_path.display())
})?;
match TomlProjectConfig::from_file(config_file) {
Ok(config) => {
info!("Validating security configuration...");
config.validate()?;
auto_error_union = config.fraiseql.mutations.auto_error_union;
naming_acronyms.clone_from(&config.fraiseql.naming.acronyms);
operation_cost_weights.clone_from(&config.fraiseql.cost_weights);
naming_convention = config.fraiseql.naming.convention;
database_tls = config.database.postgres_tls()?;
info!("Applying security configuration to schema...");
let mut security_json = config.fraiseql.security.to_json();
security_json["tenancy"] = config.fraiseql.tenancy.to_json();
intermediate.security = Some(security_json);
if config.fraiseql.session_variables
!= fraiseql_core::schema::SessionVariablesConfig::default()
{
intermediate.session_variables = Some(config.fraiseql.session_variables);
}
if let Some(locale) = config.locale {
if intermediate.locale.as_ref().is_some_and(|l| *l != locale) {
anyhow::bail!(
"[locale] in {} disagrees with the `locale` the schema document \
declares; declare it once, in fraiseql.toml",
config_path.display()
);
}
intermediate.locale = Some(locale);
}
intermediate.inject_defaults =
crate::config::inject_defaults::InjectDefaultsToml::reconcile(
config.inject_defaults.as_ref(),
intermediate.inject_defaults.take(),
)?;
info!("Security configuration applied successfully");
},
Err(e) => {
anyhow::bail!(
"Failed to parse {}: {e}\n\
Fix the configuration file or remove it to use defaults.",
config_path.display()
);
},
}
}
fraiseql_core::utils::casing::set_runtime_acronyms(&naming_acronyms);
if !is_toml {
intermediate.naming_convention = naming_convention;
}
let tenancy_row_claim: Option<String> = intermediate.security.as_ref().and_then(|sec| {
let tenancy = sec.get("tenancy")?;
let mode = tenancy.get("mode").and_then(|m| m.as_str()).unwrap_or("none");
if mode == "row" {
Some(
tenancy
.get("tenant_claim")
.and_then(|c| c.as_str())
.unwrap_or("tenant_id")
.to_string(),
)
} else {
None
}
});
if let Some(tenant_claim) = &tenancy_row_claim {
info!("Validating @tenant_id annotations for row-isolation tenancy...");
crate::schema::converter::tenancy::validate_tenant_annotations(
&mut intermediate,
tenant_claim,
)
.context("@tenant_id validation failed")?;
}
info!("Validating schema structure...");
let validation_report =
SchemaValidator::validate(&intermediate).context("Failed to validate schema")?;
if !validation_report.is_valid() {
validation_report.print();
let errors: Vec<String> = validation_report
.errors
.iter()
.filter(|e| e.severity == crate::schema::validator::ErrorSeverity::Error)
.map(|e| format!("{} ({})", e.message, e.path))
.collect();
anyhow::bail!(
"Schema validation failed with {} error(s):\n - {}",
validation_report.error_count(),
errors.join("\n - ")
);
}
if validation_report.warning_count() > 0 {
validation_report.print();
}
let mutation_defaults =
intermediate.inject_defaults.clone().unwrap_or_default().for_mutations();
let default_args: std::collections::HashMap<String, Vec<String>> = intermediate
.mutations
.iter()
.filter_map(|m| {
let mut inject = m.inject.clone();
crate::schema::intermediate::IntermediateInjectDefaults::apply_to(
&mutation_defaults,
&mut inject,
&m.exclude_inject_defaults,
&m.name,
)
.ok()
.filter(|added| !added.is_empty())
.map(|added| (m.name.clone(), added))
})
.collect();
info!("Converting to compiled format...");
let CompiledArtifact {
mut schema,
functions,
} = SchemaConverter::convert_artifact(intermediate, &ConvertOptions { auto_error_union })
.context("Failed to convert schema to compiled format")?;
schema.naming_acronyms = naming_acronyms;
if !operation_cost_weights.is_empty() {
schema.operation_cost_weights = operation_cost_weights;
}
info!("Analyzing schema for optimization opportunities...");
let report = SchemaOptimizer::optimize(&mut schema).context("Failed to optimize schema")?;
infer_native_columns_from_arg_types(&mut schema);
if let Some(db_url) = opts.database {
info!("Validating native columns for direct query arguments...");
let pool = crate::connection::postgres_pool(db_url, "database validation", &database_tls)
.await
.context("Failed to connect for database validation")?;
let pg_introspector = fraiseql_core::db::postgres::PostgresIntrospector::new(pool.clone());
let db_report = validate_schema_against_database(&schema, &pg_introspector).await?;
let mut drift_errors: Vec<String> = Vec::new();
for w in &db_report.warnings {
match w.severity() {
Severity::Error => {
warn!("{w} [drift error]");
drift_errors.push(w.to_string());
},
Severity::Warn => warn!("{w}"),
}
}
apply_database_report(&mut schema, &db_report);
info!("Validating mutation contract against the database...");
let catalog = PgCatalog::from_pool(pool);
let sources: Vec<String> = schema
.types
.iter()
.map(|t| t.sql_source.to_string())
.filter(|s| !s.is_empty())
.collect();
match catalog.tview_references(&sources).await {
Ok(references) => {
let proven =
crate::schema::native_joins::apply_native_joins(&mut schema, &references);
if proven > 0 {
info!(
"{proven} relationship(s) correlate on native columns (pg_tviews references)"
);
}
},
Err(error) => warn!(
"cannot read tviews.registry.references ({error:#}); composed reads correlate on \
document keys"
),
}
drift_errors.extend(validate_function_roots(&schema, &catalog).await?);
let contract = validate_mutation_contract(&schema, &catalog).await?;
let mut caused_by_default: std::collections::BTreeMap<String, Vec<String>> =
std::collections::BTreeMap::new();
for m in &contract.mutations {
let added = default_args.get(&m.mutation).map_or(&[][..], Vec::as_slice);
for v in &m.violations {
let kind = match v.severity() {
Severity::Error => "contract error",
Severity::Warn => "contract warning",
};
warn!("mutation `{}` (sql_source: {}): {v} [{kind}]", m.mutation, m.sql_source);
if v.severity() != Severity::Error {
continue;
}
if let Some(key) = default_caused(v, added) {
caused_by_default.entry(key).or_default().push(m.mutation.clone());
} else {
drift_errors.push(format!(
"mutation `{}` (sql_source: {}): {v}",
m.mutation, m.sql_source
));
}
}
}
for (key, mutations) in caused_by_default {
drift_errors.push(default_collision(&key, &mutations));
}
let tviews_sourced: Vec<&str> = schema
.mutations
.iter()
.filter(|m| m.cascade_source == fraiseql_core::schema::CascadeSource::PgTviews)
.map(|m| m.name.as_str())
.collect();
if !tviews_sourced.is_empty() && !catalog.pg_tviews_flush_available().await? {
for name in tviews_sourced {
drift_errors.push(format!(
"mutation `{name}`: cascade_source = \"pg_tviews\", and the database has no \
tviews.pg_tviews_flush_and_report(integer, boolean, boolean): install \
pg_tviews (preloaded), or drop cascade_source"
));
}
}
if !drift_errors.is_empty() {
if opts.allow_drift {
warn!(
"{} schema↔database drift error(s) suppressed by --allow-drift",
drift_errors.len()
);
} else {
let list =
drift_errors.iter().map(|e| format!(" - {e}")).collect::<Vec<_>>().join("\n");
anyhow::bail!(
"schema↔database drift: {} error(s):\n{list}\n\
Fix the declarations or the database objects, or pass --allow-drift \
to compile anyway (advisory mode).",
drift_errors.len()
);
}
}
} else {
for query in &schema.queries {
if query.sql_source.is_none() {
continue;
}
let unresolved: Vec<_> = query
.arguments
.iter()
.filter(|a| !NATIVE_COLUMN_SKIP_ARGS.contains(&a.name.as_str()))
.filter(|a| !query.native_columns.contains_key(&a.name))
.collect();
if !unresolved.is_empty() {
let names: Vec<_> = unresolved.iter().map(|a| a.name.as_str()).collect();
warn!(
"query `{}`: argument(s) {:?} on `{}` could not be resolved to native \
columns — no --database URL provided. These filters will use JSONB \
extraction. Provide --database or annotate with native_columns.",
query.name,
names,
query.sql_source.as_deref().unwrap_or("?"),
);
}
}
}
warn_wide_cascade_mutations(&schema);
warn_jsonb_preserve_mismatch(&schema);
refuse_unattributable_mutations(&schema)?;
refuse_unenforced_authz_declarations(&schema)?;
refuse_what_a_server_would_not_load(&schema)?;
Ok((CompiledArtifact { schema, functions }, report))
}
pub(crate) fn default_caused(
violation: &crate::schema::mutation_contract::ContractViolation,
added: &[String],
) -> Option<String> {
use crate::schema::mutation_contract::ContractViolation;
match violation {
ContractViolation::ArityMismatch { expected, found }
if !added.is_empty() && found.contains(&expected.saturating_sub(added.len())) =>
{
Some(added.join(", "))
},
ContractViolation::InjectNameMismatch { expected, .. } if added.contains(expected) => {
Some(expected.clone())
},
_ => None,
}
}
pub(crate) fn default_collision(key: &str, mutations: &[String]) -> String {
const SHOWN: usize = 5;
let shown = mutations.iter().take(SHOWN).map(|m| format!("`{m}`")).collect::<Vec<_>>();
let more = mutations.len().saturating_sub(SHOWN);
let list = if more == 0 {
shown.join(", ")
} else {
format!("{}, and {more} more", shown.join(", "))
};
format!(
"[inject_defaults] adds `{key}` to {} mutation(s) whose functions do not take it: \
{list}. Exclude it on those mutations (`exclude_inject_defaults = [\"{key}\"]`), or \
move it from the base [inject_defaults] table to [inject_defaults.queries].",
mutations.len()
)
}
fn refuse_what_a_server_would_not_load(schema: &CompiledSchema) -> Result<()> {
let body = serde_json::to_string(schema).context("Failed to serialize compiled schema")?;
let mut value: serde_json::Value = serde_json::from_str(&body)?;
let hash = content_hash_of(&value);
value
.as_object_mut()
.context("schema must serialise as JSON object")?
.insert("_content_hash".to_string(), serde_json::Value::String(hash));
CompiledSchema::from_json(&serde_json::to_string(&value)?, true).map_err(|e| {
anyhow::anyhow!("the compiled schema would be refused when a server loads it: {e}")
})?;
Ok(())
}
fn refuse_unenforced_authz_declarations(schema: &CompiledSchema) -> Result<()> {
let Some(security) = schema.security.as_ref() else {
return Ok(());
};
let mut declared: Vec<String> = Vec::new();
if !security.rules.is_empty() {
declared.push(format!("`security.rules` ({} rule(s))", security.rules.len()));
}
if !security.field_auth.is_empty() {
declared.push(format!("`security.field_auth` ({} rule(s))", security.field_auth.len()));
}
if let Some(policy) = security.default_policy.as_deref() {
declared.push(format!("`security.default_policy` (\"{policy}\")"));
}
if declared.is_empty() {
return Ok(());
}
anyhow::bail!(
"authorization declaration(s) with no enforcer: {}.\n\n\
FraiseQL enforces field-level authorization through `security.role_definitions` \
(roles → scopes) and each field's `requires_scope`; nothing reads these keys, so a \
rule written here is not applied to any request. Remove them, or express the same \
intent with role definitions and `requires_scope`.\n\n\
(A general authorization-rule engine is tracked separately; until it exists, a \
declaration that silently enforces nothing is refused rather than accepted.)",
declared.join(", ")
)
}
fn refuse_unattributable_mutations(schema: &CompiledSchema) -> Result<()> {
if !fraiseql_core::cache::declares_cacheable_views(schema) {
return Ok(());
}
let unattributable = fraiseql_core::cache::unattributable_mutations(schema);
if unattributable.is_empty() {
return Ok(());
}
let cacheable: Vec<&str> = schema
.queries
.iter()
.filter(|q| q.cache_ttl_seconds.is_some())
.filter_map(|q| q.sql_source.as_deref())
.collect();
let list = unattributable.iter().map(|m| format!(" - {m}")).collect::<Vec<_>>().join("\n");
anyhow::bail!(
"mutation(s) whose cache invalidation cannot be resolved from the schema:\n{list}\n\n\
This schema annotates {} cacheable view(s) ({}), and a mutation that resolves to no \
view invalidates nothing when it succeeds — permanently, for any view annotated \
`cache_ttl_seconds = 0`.\n\n\
Fix by declaring what each mutation writes:\n\
\x20 invalidates_views = [\"v_price\"]\n\n\
A mutation whose return type is backed by a view, or whose payload wraps an entity \
that is, already resolves and needs no annotation.",
cacheable.len(),
cacheable.join(", "),
)
}
#[allow(clippy::too_many_arguments, clippy::fn_params_excessive_bools)]
#[doc(hidden)] pub async fn run(
input: &str,
types: Option<&str>,
schema_dir: Option<&str>,
type_files: Vec<String>,
query_files: Vec<String>,
mutation_files: Vec<String>,
output: &str,
check: bool,
database: Option<&str>,
emit_ddl: Option<&str>,
check_migrations: bool,
skip_hash: bool,
allow_drift: bool,
config: Option<&str>,
) -> Result<()> {
if !check && Path::new(output) == Path::new(input) {
anyhow::bail!(
"Refusing to write compiled output over the input file '{input}'. \
Use a distinct --output path (e.g. schema.compiled.json)."
);
}
let config = ConfigSource::resolve(Path::new(input), config.map(Path::new))?;
println!("Config: {config}");
let opts = CompileOptions {
input,
types,
schema_dir,
type_files,
query_files,
mutation_files,
database,
skip_hash,
allow_drift,
config: Some(config),
};
let (artifact, optimization_report) = compile_to_schema(opts).await?;
let schema = &artifact.schema;
if check {
println!("✓ Schema is valid");
println!(" Types: {}", schema.types.len());
println!(" Queries: {}", schema.queries.len());
println!(" Mutations: {}", schema.mutations.len());
optimization_report.print();
print!("{}", localized_index_report_text(schema));
return Ok(());
}
info!("Writing compiled schema to: {output}");
let value = artifact.to_json_value()?;
let output_json = if skip_hash {
serde_json::to_string_pretty(&value).context("Failed to serialize compiled schema")?
} else {
let hash_hex = content_hash_of(&value);
let obj = value.as_object().context("schema must serialise as JSON object")?;
let mut new_obj = serde_json::Map::new();
new_obj.insert("_content_hash".to_string(), serde_json::Value::String(hash_hex));
for (k, v) in obj {
new_obj.insert(k.clone(), v.clone());
}
serde_json::to_string_pretty(&serde_json::Value::Object(new_obj))?
};
fs::write(output, output_json).context("Failed to write compiled schema")?;
println!("✓ Schema compiled successfully");
println!(" Input: {input}");
println!(" Output: {output}");
println!(" Types: {}", schema.types.len());
println!(" Queries: {}", schema.queries.len());
println!(" Mutations: {}", schema.mutations.len());
optimization_report.print();
print!("{}", localized_index_report_text(schema));
if let Some(ddl_dir) = emit_ddl {
emit_ddl_to_dir(schema, ddl_dir)?;
}
if check_migrations {
run_check_migrations(schema)?;
}
Ok(())
}
#[must_use]
pub fn localized_index_report_text(schema: &fraiseql_core::schema::CompiledSchema) -> String {
use std::fmt::Write as _;
let report = schema.localized_index_report();
if report.is_empty() {
return String::new();
}
let mut text = String::from(
"\nLocalized field indexes (one per field and allowed locale; a filter or sort on the \
field reads it):\n",
);
for advice in &report {
let subject = format!(
"{}.{} [{}] on {}",
advice.type_name, advice.field, advice.locale, advice.table
);
let _ = match &advice.index {
Some(index) => writeln!(text, " {subject}:\n {}", index.ddl),
None => writeln!(
text,
" {subject}: a view; create the index on its base table, on the expression the \
query reads"
),
};
}
text
}
pub const EMIT_DDL_FORMAT: u32 = 1;
pub fn emit_ddl_to_dir(schema: &CompiledSchema, output_dir: &str) -> Result<()> {
let mut files: std::collections::BTreeMap<String, (&str, String)> =
std::collections::BTreeMap::new();
for type_def in &schema.types {
let table_name = to_snake_case(type_def.name.as_str());
let ddl = build_create_table_ddl(&table_name, type_def)?;
if let Some((other, _)) = files.get(&table_name) {
let (first, second) = if *other < type_def.name.as_str() {
(*other, type_def.name.as_str())
} else {
(type_def.name.as_str(), *other)
};
anyhow::bail!(
"--emit-ddl: types '{first}' and '{second}' both map to the table \
`tb_{table_name}` and the file `{table_name}.sql`; rename one of them"
);
}
files.insert(table_name, (type_def.name.as_str(), ddl));
}
fs::create_dir_all(output_dir)
.context(format!("Failed to create DDL output directory: {output_dir}"))?;
for (table_name, (type_name, ddl)) in &files {
let file_path = Path::new(output_dir).join(format!("{table_name}.sql"));
fs::write(&file_path, ddl)
.context(format!("Failed to write DDL for type '{type_name}'"))?;
}
println!("✓ DDL emitted to {output_dir}/ ({} table(s))", files.len());
Ok(())
}
fn run_check_migrations(schema: &CompiledSchema) -> Result<()> {
let tmp_dir = tempfile::tempdir().context("Failed to create temporary DDL directory")?;
let tmp_path = tmp_dir.path().to_str().context("Temp directory path is not valid UTF-8")?;
emit_ddl_to_dir(schema, tmp_path)?;
info!("Running confiture migrate validate for drift detection...");
let status = Command::new("confiture").args(["migrate", "validate"]).status();
match status {
Err(_) => {
eprintln!(
"WARN: confiture is not installed; skipping migration drift check.\n\
Install it with: {}",
super::migrate::CONFITURE_INSTALL_COMMANDS.join(" (or) ")
);
Ok(())
},
Ok(s) if s.success() => {
println!("✓ No migration drift detected.");
Ok(())
},
Ok(_) => {
eprintln!(
"WARN: compiled schema diverges from database — run fraiseql migrate generate"
);
anyhow::bail!(
"Migration drift detected. Run `fraiseql migrate generate` to create a migration."
)
},
}
}
pub(crate) fn to_snake_case(name: &str) -> String {
fraiseql_core::utils::to_snake_case(name)
}
fn build_create_table_ddl(
table_name: &str,
type_def: &fraiseql_core::schema::TypeDefinition,
) -> Result<String> {
let mut lines: Vec<String> = Vec::new();
lines.push(format!("-- fraiseql emit-ddl format {EMIT_DDL_FORMAT}"));
lines.push(format!("-- Generated by fraiseql-cli {}", env!("CARGO_PKG_VERSION")));
lines.push(format!("-- Type: {}", type_def.name));
if let Some(desc) = &type_def.description {
lines.push(format!("-- {desc}"));
}
lines.push(String::new());
lines.push(format!("CREATE TABLE IF NOT EXISTS tb_{table_name} ("));
let col_lines: Vec<String> = type_def
.fields
.iter()
.map(|field| {
let col_name = to_snake_case(field.name.as_str());
let pg_type =
if field.field_type.is_searchable_vector() && field.vector_config.is_some() {
field.field_type.to_sql_type(field.vector_config.as_ref())
} else {
field_type_to_pg(&field.field_type)
};
let nullable = if field.nullable { "" } else { " NOT NULL" };
format!(" {col_name} {pg_type}{nullable}")
})
.collect();
let last = col_lines.len().saturating_sub(1);
for (i, col) in col_lines.iter().enumerate() {
if i < last {
lines.push(format!("{col},"));
} else {
lines.push(col.clone());
}
}
lines.push(");".to_string());
let vector_fields: Vec<_> = type_def
.fields
.iter()
.filter_map(|f| f.vector_config.as_ref().map(|c| (f, c)))
.collect();
if !vector_fields.is_empty() {
lines.push(String::new());
lines.push("CREATE EXTENSION IF NOT EXISTS vector;".to_string());
for (field, config) in vector_fields {
let col_name = to_snake_case(field.name.as_str());
let index_ddl = config
.index_type
.index_sql(
&format!("tb_{table_name}"),
&col_name,
&field.field_type,
config.distance_metric,
)
.with_context(|| format!("field '{}.{}'", type_def.name, field.name.as_str()))?;
if let Some(index_ddl) = index_ddl {
lines.push(format!("{index_ddl};"));
}
}
}
lines.push(String::new());
Ok(lines.join("\n"))
}
pub(crate) fn field_type_to_pg(ft: &FieldType) -> String {
match ft {
FieldType::String | FieldType::Scalar(_) => "TEXT".to_string(),
FieldType::Int => "INTEGER".to_string(),
FieldType::Float => "DOUBLE PRECISION".to_string(),
FieldType::Boolean => "BOOLEAN".to_string(),
FieldType::Id | FieldType::Uuid => "UUID".to_string(),
FieldType::DateTime => "TIMESTAMPTZ".to_string(),
FieldType::Date => "DATE".to_string(),
FieldType::Time => "TIME".to_string(),
FieldType::Json | FieldType::List { item: _, .. } | FieldType::Object(_) => {
"JSONB".to_string()
},
FieldType::Decimal => "NUMERIC".to_string(),
FieldType::Vector => "VECTOR".to_string(),
FieldType::BitVector => "VARBIT".to_string(),
FieldType::HalfVector => "HALFVEC".to_string(),
FieldType::SparseVector => "SPARSEVEC".to_string(),
FieldType::Enum(name) => name.clone(),
FieldType::Input(_) | FieldType::Interface(_) | FieldType::Union(_) => "JSONB".to_string(),
_ => "TEXT".to_string(),
}
}
pub(crate) const WIDE_FANOUT_THRESHOLD: usize = 3;
pub(crate) fn wide_cascade_mutations(
schema: &CompiledSchema,
threshold: usize,
) -> Vec<&fraiseql_core::schema::MutationDefinition> {
schema
.mutations
.iter()
.filter(|m| m.invalidates_views.len() + m.invalidates_fact_tables.len() >= threshold)
.collect()
}
fn warn_wide_cascade_mutations(schema: &CompiledSchema) {
for mutation in wide_cascade_mutations(schema, WIDE_FANOUT_THRESHOLD) {
let total = mutation.invalidates_views.len() + mutation.invalidates_fact_tables.len();
let mut targets: Vec<&str> = mutation
.invalidates_views
.iter()
.chain(mutation.invalidates_fact_tables.iter())
.map(String::as_str)
.collect();
targets.sort_unstable();
targets.dedup();
let alter_stmts: Vec<String> = targets
.iter()
.map(|&name| {
let table = name
.strip_prefix("tv_")
.or_else(|| name.strip_prefix("v_"))
.map_or_else(|| name.to_string(), |rest| format!("tb_{rest}"));
format!("ALTER TABLE {table} SET (fillfactor = 75);")
})
.collect();
warn!(
"mutation '{}' has a wide invalidation fan-out ({} targets: [{}]). \
Under high write load, HOT-update page slots on these tables may be \
exhausted, forcing full-page writes and reducing mutation throughput. \
Set fillfactor=70-80 on the backing tables: {} \
Monitor HOT efficiency: SELECT relname, \
n_tup_hot_upd * 100 / NULLIF(n_tup_upd, 0) AS hot_pct \
FROM pg_stat_user_tables WHERE n_tup_upd > 0 ORDER BY hot_pct;",
mutation.name,
total,
targets.join(", "),
alter_stmts.join(" "),
);
}
}
pub(crate) fn jsonb_preserve_mismatches(schema: &CompiledSchema) -> Vec<(String, Vec<String>)> {
if schema.naming_convention != NamingConvention::Preserve {
return Vec::new();
}
let mut out = Vec::new();
for mutation in &schema.mutations {
let single_jsonb = matches!(mutation.input_style, InputStyle::Jsonb)
|| matches!(mutation.operation, MutationOperation::Update { .. });
if !single_jsonb {
continue;
}
let input_type_name = match mutation.arguments.as_slice() {
[arg] if arg.name == "input" => match &arg.arg_type {
FieldType::Input(name) => name.as_str(),
FieldType::Object(name) if schema.find_input_type(name).is_some() => name.as_str(),
_ => continue,
},
_ => continue,
};
let Some(input_type) = schema.find_input_type(input_type_name) else {
continue;
};
let camel_fields: Vec<String> = input_type
.fields
.iter()
.filter(|f| f.name.chars().any(|c| c.is_ascii_uppercase()))
.map(|f| f.name.clone())
.collect();
if !camel_fields.is_empty() {
out.push((mutation.name.clone(), camel_fields));
}
}
out
}
fn warn_jsonb_preserve_mismatch(schema: &CompiledSchema) {
for (mutation, fields) in jsonb_preserve_mismatches(schema) {
warn!(
"mutation '{mutation}' forwards its input as a single JSONB payload but the schema \
uses naming_convention = \"preserve\", and its input type has camelCase field(s) \
[{}]. Under 'preserve' the runtime forwards input keys verbatim (no snake_case \
recasing), so a SQL function reading payload->>'snake_field' will receive these \
camelCase keys and see NULL. Set naming_convention = \"camelCase\" (the default for \
new schemas) if the function expects snake_case keys, or rename the fields. (#456)",
fields.join(", "),
);
}
}
pub(crate) async fn build_postgres_introspector(
db_url: &str,
tls: &PostgresTlsConfig,
) -> Result<fraiseql_core::db::postgres::PostgresIntrospector> {
let pool = crate::connection::postgres_pool(db_url, "database validation", tls).await?;
Ok(fraiseql_core::db::postgres::PostgresIntrospector::new(pool))
}
const NATIVE_COLUMN_SKIP_ARGS: &[&str] = &[
"where", "limit", "offset", "orderBy", "first", "last", "after", "before",
];
pub(crate) fn apply_database_report(
schema: &mut fraiseql_core::schema::CompiledSchema,
report: &crate::schema::database_validator::DatabaseValidationReport,
) {
for query in &mut schema.queries {
if let Some(cols) = report.native_columns.get(&query.name) {
query.native_columns = cols.clone();
}
if let Some(order) = report.pagination_orders.get(&query.name) {
query.pagination_order = Some(order.clone());
}
}
}
pub(crate) fn infer_native_columns_from_arg_types(schema: &mut CompiledSchema) {
for query in &mut schema.queries {
if query.sql_source.is_none()
|| query.jsonb_column.is_empty()
|| !query.sql_source_kind.is_relation()
{
continue;
}
for arg in &query.arguments {
if NATIVE_COLUMN_SKIP_ARGS.contains(&arg.name.as_str()) {
continue;
}
if query.native_columns.contains_key(&arg.name) {
continue; }
if matches!(arg.arg_type, FieldType::Id | FieldType::Uuid) {
query.native_columns.insert(
arg.name.clone(),
fraiseql_core::schema::NativeColumn::nullable("uuid"),
);
}
}
}
}