pub const SLOW_QUERIES_MAX: u32 = 50;
pub const SLOW_QUERIES_DEFAULT: u32 = 10;
pub fn parse_ref(ref_: &str) -> Result<(String, String), String> {
let trimmed = ref_.trim();
if trimmed.is_empty() {
return Err("Ref parameter is required".into());
}
let parts: Vec<&str> = trimmed.split('.').collect();
let (schema, name) = match parts.as_slice() {
[name] => ("public", *name),
[schema, name] => (*schema, *name),
_ => {
return Err(format!(
"Invalid object reference \"{ref_}\". Expected format \"schema.name\" with plain identifiers."
));
}
};
if !is_safe_ident(schema) || !is_safe_ident(name) {
return Err(format!(
"Invalid object reference \"{ref_}\". Expected format \"schema.name\" with plain identifiers."
));
}
Ok((schema.to_owned(), name.to_owned()))
}
pub fn is_safe_ident(s: &str) -> bool {
let mut chars = s.chars();
match chars.next() {
Some(c) if c.is_ascii_alphabetic() || c == '_' => {}
_ => return false,
}
chars.all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '$')
}
pub fn quote_ident(ident: &str) -> String {
format!("\"{}\"", ident.replace('"', "\"\""))
}
pub fn quote_ref(schema: &str, name: &str) -> String {
format!("{}.{}", quote_ident(schema), quote_ident(name))
}
pub fn regclass_literal(schema: &str, name: &str) -> String {
format!(
"'{}'::regclass",
quote_ref(schema, name).replace('\'', "''")
)
}
pub fn table_stats(schema: &str, table: &str) -> String {
format!(
r#"
SELECT
schemaname,
relname AS table_name,
n_live_tup AS approximate_row_count,
pg_size_pretty(pg_total_relation_size(quote_ident(schemaname) || '.' || quote_ident(relname))) AS total_size,
pg_size_pretty(pg_relation_size(quote_ident(schemaname) || '.' || quote_ident(relname))) AS table_size,
pg_size_pretty(pg_indexes_size(quote_ident(schemaname) || '.' || quote_ident(relname))) AS indexes_size,
pg_size_pretty(pg_total_relation_size(quote_ident(schemaname) || '.' || quote_ident(relname)) -
pg_relation_size(quote_ident(schemaname) || '.' || quote_ident(relname)) -
pg_indexes_size(quote_ident(schemaname) || '.' || quote_ident(relname))) AS toast_size
FROM pg_stat_user_tables
WHERE schemaname = '{schema}' AND relname = '{table}'
"#
)
.trim()
.to_owned()
}
pub fn column_stats(schema: &str, table: &str) -> String {
format!(
r#"
SELECT
attname AS column_name,
null_frac AS null_fraction,
n_distinct AS distinct_values,
avg_width AS avg_bytes,
correlation,
most_common_vals::text AS most_common_values,
most_common_freqs::text AS frequencies
FROM pg_stats
WHERE schemaname = '{schema}' AND tablename = '{table}'
ORDER BY attname
"#
)
.trim()
.to_owned()
}
pub fn column_details(schema: &str, table: &str) -> String {
let reg = regclass_literal(schema, table);
format!(
r#"
SELECT
a.attname AS column_name,
pg_catalog.format_type(a.atttypid, a.atttypmod) AS data_type,
a.attnotnull AS not_null,
COALESCE(pg_get_expr(ad.adbin, ad.adrelid), '') AS default_value,
CASE
WHEN a.attnum = ANY(pk.conkey) THEN 'PK'
WHEN a.attnum = ANY(uk.conkey) THEN 'UNIQUE'
ELSE ''
END AS key_type
FROM pg_catalog.pg_attribute a
LEFT JOIN pg_catalog.pg_attrdef ad ON (a.attrelid = ad.adrelid AND a.attnum = ad.adnum)
LEFT JOIN pg_catalog.pg_constraint pk ON (pk.conrelid = a.attrelid AND pk.contype = 'p')
LEFT JOIN pg_catalog.pg_constraint uk ON (uk.conrelid = a.attrelid AND uk.contype = 'u' AND a.attnum = ANY(uk.conkey))
WHERE a.attrelid = {reg}
AND a.attnum > 0
AND NOT a.attisdropped
ORDER BY a.attnum
"#
)
.trim()
.to_owned()
}
pub fn table_activity(schema: &str, table: &str) -> String {
format!(
r#"
SELECT
seq_scan AS sequential_scans,
seq_tup_read AS rows_seq_read,
idx_scan AS index_scans,
idx_tup_fetch AS rows_idx_fetched,
n_tup_ins AS rows_inserted,
n_tup_upd AS rows_updated,
n_tup_del AS rows_deleted,
n_tup_hot_upd AS hot_updates,
n_live_tup AS live_rows,
n_dead_tup AS dead_rows,
last_vacuum,
last_autovacuum,
last_analyze,
last_autoanalyze,
vacuum_count,
autovacuum_count,
analyze_count,
autoanalyze_count
FROM pg_stat_user_tables
WHERE schemaname = '{schema}' AND relname = '{table}'
"#
)
.trim()
.to_owned()
}
pub fn index_usage(schema: &str, table: &str) -> String {
format!(
r#"
SELECT
s.indexrelname AS index_name,
pg_size_pretty(pg_relation_size(s.indexrelid)) AS index_size,
s.idx_scan AS number_of_scans,
s.idx_tup_read AS tuples_read,
s.idx_tup_fetch AS tuples_fetched,
pg_get_indexdef(s.indexrelid) AS index_definition,
CASE
WHEN i.indisunique THEN 'UNIQUE'
WHEN i.indisprimary THEN 'PRIMARY KEY'
ELSE 'INDEX'
END AS index_type
FROM pg_stat_user_indexes s
JOIN pg_index i ON s.indexrelid = i.indexrelid
WHERE s.schemaname = '{schema}' AND s.relname = '{table}'
ORDER BY s.idx_scan DESC
"#
)
.trim()
.to_owned()
}
pub fn running_queries() -> &'static str {
r#"
SELECT pid,
usename AS user,
datname AS database,
state,
wait_event_type,
wait_event,
(now() - query_start)::text AS duration,
query_start,
LEFT(query, 500) AS query
FROM pg_stat_activity
WHERE pid != pg_backend_pid()
AND state IS DISTINCT FROM 'idle'
AND datname = current_database()
ORDER BY query_start ASC
LIMIT 100
"#
.trim()
}
pub fn blocking_locks() -> &'static str {
r#"
SELECT
blocked_locks.pid AS blocked_pid,
blocked_activity.usename AS blocked_user,
blocking_locks.pid AS blocking_pid,
blocking_activity.usename AS blocking_user,
blocked_activity.query AS blocked_query,
blocking_activity.query AS blocking_query,
blocked_locks.mode AS lock_mode,
COALESCE(c.relname, 'null') AS locked_object
FROM pg_catalog.pg_locks blocked_locks
JOIN pg_catalog.pg_stat_activity blocked_activity ON blocked_activity.pid = blocked_locks.pid
JOIN pg_catalog.pg_locks blocking_locks
ON blocking_locks.locktype = blocked_locks.locktype
AND blocking_locks.database IS NOT DISTINCT FROM blocked_locks.database
AND blocking_locks.relation IS NOT DISTINCT FROM blocked_locks.relation
AND blocking_locks.page IS NOT DISTINCT FROM blocked_locks.page
AND blocking_locks.tuple IS NOT DISTINCT FROM blocked_locks.tuple
AND blocking_locks.virtualxid IS NOT DISTINCT FROM blocked_locks.virtualxid
AND blocking_locks.transactionid IS NOT DISTINCT FROM blocked_locks.transactionid
AND blocking_locks.classid IS NOT DISTINCT FROM blocked_locks.classid
AND blocking_locks.objid IS NOT DISTINCT FROM blocked_locks.objid
AND blocking_locks.objsubid IS NOT DISTINCT FROM blocked_locks.objsubid
AND blocking_locks.pid != blocked_locks.pid
JOIN pg_catalog.pg_stat_activity blocking_activity ON blocking_activity.pid = blocking_locks.pid
LEFT JOIN pg_catalog.pg_class c ON c.oid = blocked_locks.relation
WHERE NOT blocked_locks.granted
AND blocked_activity.datname = current_database()
AND blocking_activity.datname = current_database()
"#
.trim()
}
pub fn connection_states() -> &'static str {
r#"
SELECT state, wait_event_type IS NOT NULL as waiting, count(*) as count
FROM pg_stat_activity
WHERE datname = current_database()
GROUP BY state, waiting
"#
.trim()
}
pub fn cache_hit_ratio() -> &'static str {
r#"
SELECT
blks_hit,
blks_read,
CASE WHEN blks_hit + blks_read = 0 THEN NULL
ELSE ROUND(blks_hit::numeric / (blks_hit + blks_read), 4)
END AS cache_hit_ratio,
xact_commit,
xact_rollback,
deadlocks,
temp_files,
temp_bytes
FROM pg_stat_database
WHERE datname = current_database()
"#
.trim()
}
pub fn slow_queries(limit: u32) -> String {
let capped = limit.clamp(1, SLOW_QUERIES_MAX);
format!(
r#"
SELECT
queryid::text,
LEFT(query, 500) AS query,
calls,
ROUND(mean_exec_time::numeric, 2) AS mean_ms,
ROUND(total_exec_time::numeric, 2) AS total_ms,
ROUND(stddev_exec_time::numeric, 2) AS stddev_ms,
rows
FROM pg_stat_statements
WHERE query NOT LIKE '%pg_stat_statements%'
AND query NOT LIKE 'BEGIN%'
AND query NOT LIKE 'COMMIT%'
AND query NOT LIKE 'ROLLBACK%'
AND calls >= 5
ORDER BY mean_exec_time DESC
LIMIT {capped}
"#
)
.trim()
.to_owned()
}
pub fn database_stats() -> &'static str {
r#"
SELECT
d.datname as "Database",
pg_size_pretty(pg_database_size(d.datname)) as "Size",
u.usename as "Owner",
(SELECT count(*) FROM pg_stat_activity WHERE datname = current_database()) as "Active Connections",
(SELECT count(*) FROM pg_namespace WHERE nspname NOT IN ('pg_catalog', 'information_schema')) as "Schemas",
(SELECT count(*) FROM pg_tables WHERE schemaname NOT IN ('pg_catalog', 'information_schema')) as "Tables",
(SELECT count(*) FROM pg_roles) as "Roles"
FROM pg_database d
JOIN pg_user u ON d.datdba = u.usesysid
WHERE d.datname = current_database()
"#
.trim()
}
pub fn database_maintenance_stats() -> &'static str {
r#"
SELECT
schemaname || '.' || relname as "Table",
n_dead_tup as "Dead Tuples",
n_live_tup as "Live Tuples",
last_vacuum as "Last Vacuum",
last_autovacuum as "Last Auto Vacuum",
pg_size_pretty(pg_total_relation_size(schemaname || '.' || relname)) as "Total Size"
FROM pg_stat_user_tables
WHERE n_dead_tup > 0
ORDER BY n_dead_tup DESC
LIMIT 20
"#
.trim()
}
pub fn list_extensions() -> &'static str {
r#"
SELECT e.extname AS name,
e.extversion AS version,
n.nspname AS schema
FROM pg_extension e
JOIN pg_namespace n ON n.oid = e.extnamespace
ORDER BY e.extname
"#
.trim()
}
pub fn list_roles() -> &'static str {
r#"
SELECT
r.rolname AS role,
r.rolsuper AS superuser,
r.rolcreatedb AS create_db,
r.rolcreaterole AS create_role,
r.rolcanlogin AS can_login,
r.rolreplication AS replication,
r.rolbypassrls AS bypass_rls,
r.rolinherit AS inherit,
r.rolconnlimit AS connection_limit,
r.rolvaliduntil AS valid_until
FROM pg_roles r
ORDER BY r.rolname
"#
.trim()
}
pub fn role_details() -> &'static str {
r#"
SELECT
r.rolname AS role,
r.rolsuper AS superuser,
r.rolcreatedb AS create_db,
r.rolcreaterole AS create_role,
r.rolcanlogin AS can_login,
r.rolreplication AS replication,
r.rolbypassrls AS bypass_rls,
r.rolinherit AS inherit,
r.rolconnlimit AS connection_limit,
r.rolvaliduntil AS valid_until,
pg_catalog.shobj_description(r.oid, 'pg_authid') AS description
FROM pg_roles r
WHERE r.rolname = $1
"#
.trim()
}
pub fn role_member_of() -> &'static str {
r#"
SELECT
m.rolname AS member_of,
g.rolname AS granted_by,
am.admin_option AS admin_option
FROM pg_auth_members am
JOIN pg_roles r ON r.oid = am.member
JOIN pg_roles m ON m.oid = am.roleid
JOIN pg_roles g ON g.oid = am.grantor
WHERE r.rolname = $1
ORDER BY m.rolname
"#
.trim()
}
pub fn role_has_members() -> &'static str {
r#"
SELECT
m.rolname AS has_member,
g.rolname AS granted_by,
am.admin_option AS admin_option
FROM pg_auth_members am
JOIN pg_roles r ON r.oid = am.roleid
JOIN pg_roles m ON m.oid = am.member
JOIN pg_roles g ON g.oid = am.grantor
WHERE r.rolname = $1
ORDER BY m.rolname
"#
.trim()
}
pub fn role_table_privileges() -> &'static str {
r#"
SELECT
table_schema AS schema,
table_name AS table_name,
privilege_type AS privilege,
is_grantable AS grantable
FROM information_schema.table_privileges
WHERE grantee = $1
ORDER BY table_schema, table_name, privilege_type
LIMIT 500
"#
.trim()
}
pub fn dashboard_db_info() -> &'static str {
r#"
SELECT
current_database() AS db_name,
pg_catalog.pg_get_userbyid(d.datdba) AS owner,
pg_size_pretty(pg_database_size(d.datname)) AS size,
pg_database_size(d.datname) AS size_bytes
FROM pg_database d
WHERE d.datname = current_database()
"#
.trim()
}
pub fn dashboard_top_tables() -> &'static str {
r#"
SELECT schemaname || '.' || tablename AS name,
pg_size_pretty(pg_total_relation_size(
(quote_ident(schemaname) || '.' || quote_ident(tablename))::regclass
)) AS size,
pg_total_relation_size(
(quote_ident(schemaname) || '.' || quote_ident(tablename))::regclass
) AS raw_size
FROM pg_tables
WHERE schemaname NOT IN ('pg_catalog', 'information_schema')
ORDER BY raw_size DESC
LIMIT 10
"#
.trim()
}
pub fn dashboard_object_counts() -> &'static str {
r#"
SELECT
(SELECT count(*) FROM pg_namespace
WHERE nspname NOT IN ('pg_catalog', 'information_schema')
AND nspname NOT LIKE 'pg_%') AS schemas,
(SELECT count(*) FROM pg_tables
WHERE schemaname NOT IN ('pg_catalog', 'information_schema')) AS tables,
(SELECT count(*) FROM pg_views
WHERE schemaname NOT IN ('pg_catalog', 'information_schema')) AS views,
(SELECT count(*) FROM pg_proc p
JOIN pg_namespace n ON p.pronamespace = n.oid
WHERE n.nspname NOT IN ('pg_catalog', 'information_schema')) AS functions,
(SELECT count(*) FROM pg_class c
JOIN pg_namespace n ON c.relnamespace = n.oid
WHERE c.relkind = 'S'
AND n.nspname NOT IN ('pg_catalog', 'information_schema')) AS sequences
"#
.trim()
}
pub fn dashboard_active_queries() -> &'static str {
r#"
SELECT pid, usename, datname, state,
wait_event_type, wait_event,
xact_start,
(now() - query_start)::text AS duration,
query_start,
LEFT(query, 500) AS query
FROM pg_stat_activity
WHERE pid != pg_backend_pid()
AND datname = current_database()
ORDER BY state = 'active' DESC, query_start ASC
LIMIT 50
"#
.trim()
}
pub fn dashboard_max_connections() -> &'static str {
r#"SHOW max_connections"#
}
pub fn dashboard_extension_count() -> &'static str {
r#"
SELECT count(*)::int AS count
FROM pg_extension
"#
.trim()
}
pub fn server_settings() -> &'static str {
r#"
SELECT name, setting, unit, category, short_desc
FROM pg_settings
WHERE name IN (
'max_connections', 'shared_buffers', 'work_mem', 'maintenance_work_mem',
'effective_cache_size', 'random_page_cost', 'seq_page_cost',
'statement_timeout', 'idle_in_transaction_session_timeout',
'max_wal_size', 'checkpoint_completion_target', 'wal_buffers',
'default_statistics_target', 'autovacuum', 'server_version',
'shared_preload_libraries', 'default_transaction_read_only'
)
ORDER BY name
"#
.trim()
}
pub const REPORT_LIMIT_MAX: u32 = 50;
pub const REPORT_LIMIT_DEFAULT: u32 = 20;
fn clamp_report_limit(limit: u32) -> u32 {
limit.clamp(1, REPORT_LIMIT_MAX)
}
pub fn high_seq_scan_tables(limit: u32) -> String {
let capped = clamp_report_limit(limit);
format!(
r#"
SELECT schemaname || '.' || relname AS table_name,
seq_scan,
COALESCE(idx_scan, 0) AS idx_scan,
CASE WHEN seq_scan + COALESCE(idx_scan, 0) > 0
THEN (ROUND(100.0 * seq_scan / (seq_scan + COALESCE(idx_scan, 0)), 1))::double precision
ELSE 0::double precision END AS seq_scan_pct,
n_live_tup AS row_count,
'High sequential scan ratio — consider indexes on frequently filtered/joined columns' AS rationale
FROM pg_stat_user_tables
WHERE schemaname NOT IN ('pg_catalog', 'information_schema')
AND seq_scan + COALESCE(idx_scan, 0) > 100
ORDER BY seq_scan_pct DESC, seq_scan DESC
LIMIT {capped}
"#
)
.trim()
.to_owned()
}
pub fn unindexed_fk_columns(limit: u32) -> String {
let capped = clamp_report_limit(limit);
format!(
r#"
SELECT
n.nspname || '.' || c.relname AS table_name,
a.attname AS column_name,
confrelid::regclass::text AS references_table,
'FK column without supporting index — CREATE INDEX ON ' ||
quote_ident(n.nspname) || '.' || quote_ident(c.relname) ||
' (' || quote_ident(a.attname) || ')' AS suggestion
FROM pg_constraint con
JOIN pg_class c ON c.oid = con.conrelid
JOIN pg_namespace n ON n.oid = c.relnamespace
JOIN LATERAL unnest(con.conkey) WITH ORDINALITY AS ck(attnum, ord) ON true
JOIN pg_attribute a ON a.attrelid = c.oid AND a.attnum = ck.attnum
WHERE con.contype = 'f'
AND n.nspname NOT IN ('pg_catalog', 'information_schema')
AND NOT EXISTS (
SELECT 1
FROM pg_index i
WHERE i.indrelid = c.oid
AND a.attnum = ANY (i.indkey::int2[])
)
ORDER BY n.nspname, c.relname, a.attname
LIMIT {capped}
"#
)
.trim()
.to_owned()
}
pub fn find_unused_indexes(limit: u32) -> String {
let capped = clamp_report_limit(limit);
format!(
r#"
SELECT s.schemaname || '.' || s.indexrelname AS index_name,
s.schemaname || '.' || s.relname AS table_name,
pg_size_pretty(pg_relation_size(s.indexrelid)) AS index_size,
pg_relation_size(s.indexrelid) AS raw_size,
pg_get_indexdef(s.indexrelid) AS index_definition
FROM pg_stat_user_indexes s
JOIN pg_index i
ON i.indexrelid = s.indexrelid
LEFT JOIN pg_constraint c
ON c.conindid = s.indexrelid
WHERE s.idx_scan = 0
AND s.schemaname NOT IN ('pg_catalog', 'information_schema')
AND c.oid IS NULL
AND NOT i.indisprimary
AND NOT i.indisunique
ORDER BY raw_size DESC
LIMIT {capped}
"#
)
.trim()
.to_owned()
}
pub fn bloat_report(limit: u32) -> String {
let capped = clamp_report_limit(limit);
format!(
r#"
SELECT schemaname || '.' || relname AS table_name,
n_live_tup AS live_tuples,
n_dead_tup AS dead_tuples,
CASE WHEN n_live_tup + n_dead_tup > 0
THEN ROUND(100.0 * n_dead_tup / (n_live_tup + n_dead_tup), 1)
ELSE 0 END AS bloat_pct,
pg_size_pretty(pg_relation_size(relid)) AS table_size,
last_autovacuum,
last_vacuum
FROM pg_stat_user_tables
WHERE schemaname NOT IN ('pg_catalog', 'information_schema')
AND n_dead_tup > 1000
ORDER BY bloat_pct DESC, n_dead_tup DESC
LIMIT {capped}
"#
)
.trim()
.to_owned()
}
pub fn find_missing_fks_catalog(limit: u32) -> String {
let capped = clamp_report_limit(limit);
format!(
r#"
WITH pk_tables AS (
SELECT
n.nspname AS schema_name,
c.relname AS table_name,
a.attname AS pk_column,
lower(c.relname) AS table_lower
FROM pg_constraint con
JOIN pg_class c ON c.oid = con.conrelid
JOIN pg_namespace n ON n.oid = c.relnamespace
JOIN LATERAL unnest(con.conkey) WITH ORDINALITY AS ck(attnum, ord) ON true
JOIN pg_attribute a ON a.attrelid = c.oid AND a.attnum = ck.attnum
WHERE con.contype = 'p'
AND n.nspname NOT IN ('pg_catalog', 'information_schema')
AND array_length(con.conkey, 1) = 1
),
candidates AS (
SELECT
n.nspname AS schema_name,
c.relname AS table_name,
a.attname AS column_name,
left(a.attname, length(a.attname) - 3) AS name_prefix
FROM pg_attribute a
JOIN pg_class c ON c.oid = a.attrelid AND c.relkind = 'r'
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE a.attnum > 0
AND NOT a.attisdropped
AND n.nspname NOT IN ('pg_catalog', 'information_schema')
AND a.attname ~* '_id$'
AND a.attname <> 'id'
AND NOT EXISTS (
SELECT 1
FROM pg_constraint con
JOIN LATERAL unnest(con.conkey) AS ck(attnum) ON true
WHERE con.conrelid = c.oid
AND con.contype = 'f'
AND ck.attnum = a.attnum
)
)
SELECT
cand.schema_name || '.' || cand.table_name AS from_table,
cand.column_name,
pk.schema_name || '.' || pk.table_name AS suggested_ref_table,
pk.pk_column AS suggested_ref_column,
'naming_convention' AS detection
FROM candidates cand
JOIN pk_tables pk
ON pk.schema_name = cand.schema_name
AND (
pk.table_lower = lower(cand.name_prefix)
OR pk.table_lower = lower(cand.name_prefix) || 's'
OR pk.table_lower = lower(cand.name_prefix) || 'es'
OR (right(lower(cand.name_prefix), 1) = 'y'
AND pk.table_lower = left(lower(cand.name_prefix), -1) || 'ies')
)
WHERE cand.schema_name || '.' || cand.table_name
<> pk.schema_name || '.' || pk.table_name
ORDER BY from_table, column_name
LIMIT {capped}
"#
)
.trim()
.to_owned()
}
pub fn enhance_sql_error(pg_message: &str, index_refs: &[String]) -> String {
let mut suggestions: Vec<String> = Vec::new();
if let Some(rel) = extract_pg_relation_name(pg_message) {
if !rel.contains('.') {
let matches: Vec<&String> = index_refs
.iter()
.filter(|r| r.rsplit('.').next() == Some(rel.as_str()))
.collect();
match matches.as_slice() {
[] => {
if let Some(close) = closest_index_ref(&rel, index_refs) {
suggestions.push(format!("Did you mean '{close}'?"));
}
}
[only] => {
suggestions.push(format!(
"Table '{rel}' not found in search_path. Did you mean '{only}'?"
));
}
many => {
let list: Vec<&str> = many.iter().map(|s| s.as_str()).collect();
suggestions.push(format!(
"Table '{rel}' is ambiguous across schemas: {}",
list.join(", ")
));
}
}
} else if !index_refs.iter().any(|r| r == &rel)
&& let Some(close) = closest_index_ref(&rel, index_refs)
{
suggestions.push(format!("Did you mean '{close}'?"));
}
}
if let Some(col) = extract_pg_column_name(pg_message) {
if let Some(rel) = extract_pg_relation_name(pg_message)
&& let Some(entry_ref) = resolve_ref_for_column_hint(&rel, index_refs)
{
suggestions.push(format!(
"Column '{col}' not found on '{entry_ref}' — call describe_object or inspect_or_search for valid columns."
));
} else {
suggestions.push(format!(
"Column '{col}' not found — verify spelling and schema qualification."
));
}
}
if suggestions.is_empty() {
pg_message.to_string()
} else {
format!("{pg_message}\nfix_hint: {}", suggestions.join("; "))
}
}
fn resolve_ref_for_column_hint(rel: &str, index_refs: &[String]) -> Option<String> {
if index_refs.iter().any(|r| r == rel) {
return Some(rel.to_string());
}
if !rel.contains('.') {
let matches: Vec<&String> = index_refs
.iter()
.filter(|r| r.rsplit('.').next() == Some(rel))
.collect();
if matches.len() == 1 {
return Some(matches[0].clone());
}
}
None
}
fn extract_pg_relation_name(message: &str) -> Option<String> {
let lower = message.to_ascii_lowercase();
let needle = "relation \"";
let start = lower.find(needle)? + needle.len();
let rest = &message[start..];
let end = rest.find('"')?;
Some(rest[..end].to_string())
}
fn extract_pg_column_name(message: &str) -> Option<String> {
let lower = message.to_ascii_lowercase();
let needle = "column \"";
let start = lower.find(needle)? + needle.len();
let rest = &message[start..];
let end = rest.find('"')?;
Some(rest[..end].to_string())
}
fn closest_index_ref(ref_: &str, universe: &[String]) -> Option<String> {
let mut best: Option<(String, usize)> = None;
for candidate in universe {
let dist = levenshtein(ref_.to_ascii_lowercase(), candidate.to_ascii_lowercase());
if dist > 0 && dist <= 3 && best.as_ref().is_none_or(|(_, d)| dist < *d) {
best = Some((candidate.clone(), dist));
}
}
best.map(|(s, _)| s)
}
fn levenshtein(a: String, b: String) -> usize {
let a: Vec<char> = a.chars().collect();
let b: Vec<char> = b.chars().collect();
let mut prev: Vec<usize> = (0..=b.len()).collect();
for (i, ca) in a.iter().enumerate() {
let mut cur = vec![i + 1];
for (j, cb) in b.iter().enumerate() {
let cost = if ca == cb { 0 } else { 1 };
cur.push((prev[j] + cost).min(cur[j] + 1).min(prev[j + 1] + 1));
}
prev = cur;
}
prev[b.len()]
}
pub fn map_stat_statements_error(e: &tokio_postgres::Error) -> Option<String> {
let message = nexql_conn::format_postgres_error(e);
if is_pg_stat_statements_error_message(&message) {
Some(
"pg_stat_statements is not available — CREATE EXTENSION pg_stat_statements; \
(and GRANT) or ignore slow query suggestions."
.into(),
)
} else {
None
}
}
fn is_pg_stat_statements_error_message(message: &str) -> bool {
message.to_ascii_lowercase().contains("pg_stat_statements")
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn map_stat_statements_error_matches_extension_errors() {
assert!(is_pg_stat_statements_error_message(
"relation \"pg_stat_statements\" does not exist"
));
}
#[test]
fn map_stat_statements_error_ignores_unrelated() {
assert!(!is_pg_stat_statements_error_message("connection refused"));
}
#[test]
fn parse_ref_accepts_schema_dot_name() {
assert_eq!(
parse_ref("public.users").unwrap(),
("public".into(), "users".into())
);
}
#[test]
fn parse_ref_defaults_schema() {
assert_eq!(
parse_ref("orders").unwrap(),
("public".into(), "orders".into())
);
}
#[test]
fn parse_ref_rejects_injection() {
let err = parse_ref("public.users; DROP").unwrap_err();
assert!(err.contains("Invalid object reference"), "{err}");
}
#[test]
fn parse_ref_rejects_empty() {
assert!(parse_ref("").is_err());
assert!(parse_ref(" ").is_err());
}
}