use crate::infra::utils::{quote_ident, quote_literal};
use super::super::PostgresAdapter;
impl PostgresAdapter {
pub(in crate::infra::adapters::postgres) fn tables_query() -> &'static str {
r#"
SELECT json_agg(row_to_json(t))
FROM (
SELECT
n.nspname as schema,
c.relname as name,
c.reltuples::bigint as row_count_estimate,
c.relrowsecurity as has_rls
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE c.relkind = 'r'
AND n.nspname NOT IN ('pg_catalog', 'information_schema', 'pg_toast')
AND (
has_table_privilege(c.oid, 'SELECT')
OR has_table_privilege(c.oid, 'INSERT')
OR has_table_privilege(c.oid, 'UPDATE')
OR has_table_privilege(c.oid, 'DELETE')
OR has_table_privilege(c.oid, 'TRUNCATE')
OR has_table_privilege(c.oid, 'REFERENCES')
OR has_table_privilege(c.oid, 'TRIGGER')
)
ORDER BY n.nspname, c.relname
) t
"#
}
pub(in crate::infra::adapters::postgres) fn table_signatures_query() -> &'static str {
r#"
SELECT json_agg(row_to_json(t))
FROM (
SELECT
n.nspname AS schema,
c.relname AS name,
md5(
COALESCE((
SELECT string_agg(
a.attname || ':' || pg_catalog.format_type(a.atttypid, a.atttypmod)
|| ':' || a.attnotnull::text
|| ':' || COALESCE(pg_get_expr(d.adbin, d.adrelid), ''),
'|' ORDER BY a.attnum
)
FROM pg_attribute a
LEFT JOIN pg_attrdef d ON d.adrelid = c.oid AND d.adnum = a.attnum
WHERE a.attrelid = c.oid AND a.attnum > 0 AND NOT a.attisdropped
), '')
|| '##FK##'
|| COALESCE((
SELECT string_agg(
con.conname || ':'
|| n2.nspname || '.' || c2.relname,
'|' ORDER BY con.conname
)
FROM pg_constraint con
JOIN pg_class c2 ON c2.oid = con.confrelid
JOIN pg_namespace n2 ON n2.oid = c2.relnamespace
WHERE con.conrelid = c.oid AND con.contype = 'f'
), '')
) AS signature
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE c.relkind = 'r'
AND n.nspname NOT IN ('pg_catalog', 'information_schema', 'pg_toast')
AND (
has_table_privilege(c.oid, 'SELECT')
OR has_table_privilege(c.oid, 'INSERT')
OR has_table_privilege(c.oid, 'UPDATE')
OR has_table_privilege(c.oid, 'DELETE')
OR has_table_privilege(c.oid, 'TRUNCATE')
OR has_table_privilege(c.oid, 'REFERENCES')
OR has_table_privilege(c.oid, 'TRIGGER')
)
ORDER BY n.nspname, c.relname
) t
"#
}
pub(in crate::infra::adapters::postgres) fn schemas_query() -> &'static str {
r#"
SELECT json_agg(row_to_json(s))
FROM (
SELECT nspname as name
FROM pg_namespace
WHERE nspname NOT IN ('pg_catalog', 'information_schema', 'pg_toast')
AND nspname NOT LIKE 'pg_temp_%'
AND nspname NOT LIKE 'pg_toast_temp_%'
ORDER BY nspname
) s
"#
}
pub(in crate::infra::adapters::postgres) fn columns_query(schema: &str, table: &str) -> String {
format!(
r#"
SELECT json_agg(row_to_json(c) ORDER BY c.ordinal_position)
FROM (
SELECT
a.attname as name,
pg_catalog.format_type(a.atttypid, a.atttypmod) as data_type,
NOT a.attnotnull as nullable,
pg_get_expr(d.adbin, d.adrelid) as default,
EXISTS (
SELECT 1 FROM pg_index i
WHERE i.indrelid = cl.oid
AND i.indisprimary
AND a.attnum = ANY(i.indkey)
) as is_primary_key,
EXISTS (
SELECT 1 FROM pg_index i
WHERE i.indrelid = cl.oid
AND i.indisunique
AND NOT i.indisprimary
AND array_length(i.indkey, 1) = 1
AND a.attnum = ANY(i.indkey)
) as is_unique,
col_description(cl.oid, a.attnum) as comment,
a.attnum as ordinal_position
FROM pg_class cl
JOIN pg_namespace n ON n.oid = cl.relnamespace
JOIN pg_attribute a ON a.attrelid = cl.oid
LEFT JOIN pg_attrdef d ON d.adrelid = cl.oid AND d.adnum = a.attnum
WHERE n.nspname = {}
AND cl.relname = {}
AND a.attnum > 0
AND NOT a.attisdropped
) c
"#,
quote_literal(schema),
quote_literal(table)
)
}
pub(in crate::infra::adapters::postgres) fn preview_pk_columns_query(
schema: &str,
table: &str,
) -> String {
format!(
r#"
SELECT COALESCE(json_agg(a.attname ORDER BY array_position(i.indkey, a.attnum)), '[]'::json)
FROM pg_index i
JOIN pg_class c ON c.oid = i.indrelid
JOIN pg_namespace n ON n.oid = c.relnamespace
JOIN pg_attribute a ON a.attrelid = c.oid AND a.attnum = ANY(i.indkey)
WHERE i.indisprimary
AND n.nspname = {}
AND c.relname = {}
"#,
quote_literal(schema),
quote_literal(table)
)
}
pub(in crate::infra::adapters::postgres) fn build_preview_query(
schema: &str,
table: &str,
order_columns: &[String],
limit: usize,
offset: usize,
) -> String {
let order_clause = if order_columns.is_empty() {
String::new()
} else {
let cols = order_columns
.iter()
.map(|col| quote_ident(col))
.collect::<Vec<_>>()
.join(", ");
format!(" ORDER BY {}", cols)
};
format!(
"SELECT * FROM {}.{}{} LIMIT {} OFFSET {}",
quote_ident(schema),
quote_ident(table),
order_clause,
limit,
offset
)
}
pub(in crate::infra::adapters::postgres) fn indexes_query(schema: &str, table: &str) -> String {
format!(
r#"
SELECT json_agg(row_to_json(i))
FROM (
SELECT
idx.relname as name,
array_agg(a.attname ORDER BY array_position(ix.indkey, a.attnum)) as columns,
ix.indisunique as is_unique,
ix.indisprimary as is_primary,
am.amname as index_type,
pg_get_indexdef(idx.oid) as definition
FROM pg_index ix
JOIN pg_class idx ON idx.oid = ix.indexrelid
JOIN pg_class tbl ON tbl.oid = ix.indrelid
JOIN pg_namespace n ON n.oid = tbl.relnamespace
JOIN pg_am am ON am.oid = idx.relam
JOIN pg_attribute a ON a.attrelid = tbl.oid AND a.attnum = ANY(ix.indkey)
WHERE n.nspname = {}
AND tbl.relname = {}
GROUP BY idx.relname, ix.indisunique, ix.indisprimary, am.amname, idx.oid
ORDER BY idx.relname
) i
"#,
quote_literal(schema),
quote_literal(table)
)
}
pub(in crate::infra::adapters::postgres) fn foreign_keys_query(
schema: &str,
table: &str,
) -> String {
format!(
r#"
SELECT json_agg(row_to_json(fk))
FROM (
SELECT
con.conname as name,
n1.nspname as from_schema,
c1.relname as from_table,
array_agg(a1.attname ORDER BY array_position(con.conkey, a1.attnum)) as from_columns,
n2.nspname as to_schema,
c2.relname as to_table,
array_agg(a2.attname ORDER BY array_position(con.confkey, a2.attnum)) as to_columns,
con.confdeltype as on_delete,
con.confupdtype as on_update
FROM pg_constraint con
JOIN pg_class c1 ON c1.oid = con.conrelid
JOIN pg_namespace n1 ON n1.oid = c1.relnamespace
JOIN pg_class c2 ON c2.oid = con.confrelid
JOIN pg_namespace n2 ON n2.oid = c2.relnamespace
JOIN pg_attribute a1 ON a1.attrelid = c1.oid AND a1.attnum = ANY(con.conkey)
JOIN pg_attribute a2 ON a2.attrelid = c2.oid AND a2.attnum = ANY(con.confkey)
WHERE con.contype = 'f'
AND n1.nspname = {}
AND c1.relname = {}
GROUP BY con.conname, n1.nspname, c1.relname, n2.nspname, c2.relname, con.confdeltype, con.confupdtype
) fk
"#,
quote_literal(schema),
quote_literal(table)
)
}
pub(in crate::infra::adapters::postgres) fn rls_query(schema: &str, table: &str) -> String {
format!(
r#"
SELECT json_build_object(
'enabled', c.relrowsecurity,
'force', c.relforcerowsecurity,
'policies', COALESCE((
SELECT json_agg(json_build_object(
'name', p.polname,
'permissive', p.polpermissive,
'roles', (
SELECT array_agg(r.rolname)
FROM pg_roles r
WHERE r.oid = ANY(p.polroles)
),
'cmd', p.polcmd,
'qual', pg_get_expr(p.polqual, p.polrelid),
'with_check', pg_get_expr(p.polwithcheck, p.polrelid)
))
FROM pg_policy p
WHERE p.polrelid = c.oid
), '[]'::json)
)
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname = {}
AND c.relname = {}
"#,
quote_literal(schema),
quote_literal(table)
)
}
pub(in crate::infra::adapters::postgres) fn triggers_query(
schema: &str,
table: &str,
) -> String {
format!(
r#"
SELECT json_agg(row_to_json(t) ORDER BY t.name)
FROM (
SELECT
tg.tgname AS name,
CASE
WHEN (tg.tgtype & 2) != 0 THEN 'BEFORE'
WHEN (tg.tgtype & 2) = 0 AND (tg.tgtype & 64) != 0 THEN 'INSTEAD OF'
ELSE 'AFTER'
END AS timing,
array_remove(ARRAY[
CASE WHEN (tg.tgtype & 4) != 0 THEN 'INSERT' END,
CASE WHEN (tg.tgtype & 8) != 0 THEN 'DELETE' END,
CASE WHEN (tg.tgtype & 16) != 0 THEN 'UPDATE' END,
CASE WHEN (tg.tgtype & 32) != 0 THEN 'TRUNCATE' END
], NULL) AS events,
p.proname AS function_name,
p.prosecdef AS security_definer
FROM pg_trigger tg
JOIN pg_class c ON c.oid = tg.tgrelid
JOIN pg_namespace n ON n.oid = c.relnamespace
JOIN pg_proc p ON p.oid = tg.tgfoid
WHERE NOT tg.tgisinternal
AND n.nspname = {}
AND c.relname = {}
) t
"#,
quote_literal(schema),
quote_literal(table)
)
}
pub(in crate::infra::adapters::postgres) fn table_info_query(
schema: &str,
table: &str,
) -> String {
format!(
r#"
SELECT row_to_json(t)
FROM (
SELECT
pg_get_userbyid(c.relowner) AS owner,
obj_description(c.oid) AS comment,
c.reltuples::bigint AS row_count_estimate
FROM pg_class c
JOIN pg_namespace n ON n.oid = c.relnamespace
WHERE n.nspname = {}
AND c.relname = {}
) t
"#,
quote_literal(schema),
quote_literal(table)
)
}
pub(in crate::infra::adapters::postgres) fn table_detail_light_query(
schema: &str,
table: &str,
) -> String {
format!(
r#"
SELECT json_build_object(
'columns', ({columns}),
'foreign_keys', ({fks})
)
"#,
columns = Self::columns_query(schema, table).trim(),
fks = Self::foreign_keys_query(schema, table).trim(),
)
}
pub(in crate::infra::adapters::postgres) fn table_detail_query(
schema: &str,
table: &str,
) -> String {
format!(
r#"
SELECT json_build_object(
'columns', ({columns}),
'indexes', ({indexes}),
'foreign_keys', ({fks}),
'rls', ({rls}),
'triggers', ({triggers}),
'table_info', ({table_info})
)
"#,
columns = Self::columns_query(schema, table).trim(),
indexes = Self::indexes_query(schema, table).trim(),
fks = Self::foreign_keys_query(schema, table).trim(),
rls = Self::rls_query(schema, table).trim(),
triggers = Self::triggers_query(schema, table).trim(),
table_info = Self::table_info_query(schema, table).trim(),
)
}
}
#[cfg(test)]
mod tests {
use crate::infra::adapters::postgres::PostgresAdapter;
mod preview_query {
use super::*;
#[test]
fn with_primary_key_columns_returns_ordered_preview_query() {
let sql = PostgresAdapter::build_preview_query(
"public",
"users",
&["id".to_string(), "tenant_id".to_string()],
100,
200,
);
assert_eq!(
sql,
"SELECT * FROM \"public\".\"users\" ORDER BY \"id\", \"tenant_id\" LIMIT 100 OFFSET 200"
);
}
#[test]
fn without_primary_key_columns_returns_unordered_preview_query() {
let sql = PostgresAdapter::build_preview_query("public", "users", &[], 100, 0);
assert_eq!(sql, "SELECT * FROM \"public\".\"users\" LIMIT 100 OFFSET 0");
}
#[test]
fn primary_key_query_returns_json_aggregate_sql() {
let sql = PostgresAdapter::preview_pk_columns_query("public", "users");
assert!(
sql.contains("json_agg(a.attname ORDER BY array_position(i.indkey, a.attnum))")
);
assert!(sql.contains("n.nspname = 'public'"));
assert!(sql.contains("c.relname = 'users'"));
}
}
mod table_signatures_query {
use super::*;
#[test]
fn contains_md5_hash() {
let sql = PostgresAdapter::table_signatures_query();
assert!(sql.contains("md5("));
}
#[test]
fn uses_same_filter_as_tables_query() {
let sig_sql = PostgresAdapter::table_signatures_query();
let tab_sql = PostgresAdapter::tables_query();
assert!(sig_sql.contains("c.relkind = 'r'"));
assert!(
sig_sql
.contains("n.nspname NOT IN ('pg_catalog', 'information_schema', 'pg_toast')")
);
assert!(sig_sql.contains("has_table_privilege(c.oid, 'SELECT')"));
for fragment in [
"c.relkind = 'r'",
"n.nspname NOT IN ('pg_catalog', 'information_schema', 'pg_toast')",
"has_table_privilege(c.oid, 'SELECT')",
"has_table_privilege(c.oid, 'INSERT')",
"has_table_privilege(c.oid, 'TRIGGER')",
] {
assert!(
tab_sql.contains(fragment) && sig_sql.contains(fragment),
"Filter mismatch: {fragment}"
);
}
}
#[test]
fn includes_fk_separator() {
let sql = PostgresAdapter::table_signatures_query();
assert!(sql.contains("'##FK##'"));
}
#[test]
fn returns_json_aggregate() {
let sql = PostgresAdapter::table_signatures_query();
assert!(sql.contains("json_agg(row_to_json(t))"));
}
}
mod preview_query_edge_cases {
use super::*;
#[test]
fn schema_name_with_double_quote_is_escaped() {
let sql = PostgresAdapter::build_preview_query("my\"schema", "users", &[], 100, 0);
assert_eq!(
sql,
"SELECT * FROM \"my\"\"schema\".\"users\" LIMIT 100 OFFSET 0"
);
}
#[test]
fn table_name_with_double_quote_is_escaped() {
let sql = PostgresAdapter::build_preview_query("public", "my\"table", &[], 100, 0);
assert_eq!(
sql,
"SELECT * FROM \"public\".\"my\"\"table\" LIMIT 100 OFFSET 0"
);
}
#[test]
fn order_by_column_with_double_quote_is_escaped() {
let sql = PostgresAdapter::build_preview_query(
"public",
"users",
&["my\"col".to_string()],
100,
0,
);
assert_eq!(
sql,
"SELECT * FROM \"public\".\"users\" ORDER BY \"my\"\"col\" LIMIT 100 OFFSET 0"
);
}
}
mod table_detail_query {
use super::*;
#[test]
fn wraps_all_six_categories_in_json_build_object() {
let sql = PostgresAdapter::table_detail_query("public", "users");
assert!(sql.contains("json_build_object("));
for key in [
"'columns'",
"'indexes'",
"'foreign_keys'",
"'rls'",
"'triggers'",
"'table_info'",
] {
assert!(sql.contains(key), "Missing key: {key}");
}
}
#[test]
fn uses_quoted_schema_and_table() {
let sql = PostgresAdapter::table_detail_query("my_schema", "my_table");
assert!(sql.contains("'my_schema'"));
assert!(sql.contains("'my_table'"));
}
}
mod table_detail_light_query {
use super::*;
#[test]
fn wraps_columns_and_fks_only_in_json_build_object() {
let sql = PostgresAdapter::table_detail_light_query("public", "users");
assert!(sql.contains("json_build_object("));
assert!(sql.contains("'columns'"));
assert!(sql.contains("'foreign_keys'"));
assert!(!sql.contains("'indexes'"));
assert!(!sql.contains("'rls'"));
assert!(!sql.contains("'triggers'"));
assert!(!sql.contains("'table_info'"));
}
#[test]
fn uses_quoted_schema_and_table() {
let sql = PostgresAdapter::table_detail_light_query("my_schema", "my_table");
assert!(sql.contains("'my_schema'"));
assert!(sql.contains("'my_table'"));
}
}
mod metadata_query_injection {
use super::*;
use rstest::rstest;
const HOSTILE: &str = "'; DROP TABLE users; --";
const ESCAPED: &str = "'''; DROP TABLE users; --'";
#[rstest]
#[case("columns_query", PostgresAdapter::columns_query(HOSTILE, "t"))]
#[case(
"columns_query_table",
PostgresAdapter::columns_query("public", HOSTILE)
)]
#[case("indexes_query", PostgresAdapter::indexes_query(HOSTILE, "t"))]
#[case(
"foreign_keys_query",
PostgresAdapter::foreign_keys_query(HOSTILE, "t")
)]
#[case("rls_query", PostgresAdapter::rls_query(HOSTILE, "t"))]
#[case("triggers_query", PostgresAdapter::triggers_query(HOSTILE, "t"))]
#[case(
"table_detail_query",
PostgresAdapter::table_detail_query(HOSTILE, "t")
)]
#[case(
"table_detail_query_table",
PostgresAdapter::table_detail_query("public", HOSTILE)
)]
#[case(
"table_detail_light_query",
PostgresAdapter::table_detail_light_query(HOSTILE, "t")
)]
#[case(
"table_detail_light_query_table",
PostgresAdapter::table_detail_light_query("public", HOSTILE)
)]
fn hostile_input_is_escaped(#[case] _label: &str, #[case] sql: String) {
assert!(
sql.contains(ESCAPED),
"Hostile input must be quote_literal-escaped in: {sql}"
);
}
}
}