feat(postgres): browse PostgreSQL event triggers
This commit is contained in:
34 files changed
+632
-11
No files matched your search
@@ -50,4 +50,4 @@
|
||||
"@vue/shared": "3.5.35",
|
||||
"marked": "18.0.4"
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -138,6 +138,67 @@ pub(super) async fn list_available_extensions(
|
||||
Ok(extension_infos_from_query_result(result, false))
|
||||
}
|
||||
|
||||
/// Lists event triggers over a Kingbase agent connection. KingbaseES ships
|
||||
/// `pg_event_trigger`, so the same catalog query works; the sys/pg catalog
|
||||
/// fallback mirrors `list_extensions`. `tags`/`source` are not fetched over the
|
||||
/// agent path (text[] marshalling + `pg_get_eventtriggerdef` availability vary),
|
||||
/// so the details dialog's field-driven DDL rebuild covers them.
|
||||
pub(super) async fn list_event_triggers(
|
||||
client: Arc<db::agent_driver::PooledAgentClient>,
|
||||
database: &str,
|
||||
timeout_duration: Option<Duration>,
|
||||
) -> Result<Vec<db::EventTriggerInfo>, String> {
|
||||
let result = query_result_with_catalog_fallback(
|
||||
client,
|
||||
database,
|
||||
event_triggers_sql(ExtensionCatalog::Sys),
|
||||
event_triggers_sql(ExtensionCatalog::Pg),
|
||||
10_000,
|
||||
timeout_duration,
|
||||
)
|
||||
.await?;
|
||||
Ok(event_trigger_infos_from_query_result(result))
|
||||
}
|
||||
|
||||
fn event_triggers_sql(catalog: ExtensionCatalog) -> String {
|
||||
let cat = catalog.catalog_name();
|
||||
format!(
|
||||
"SELECT e.evtname, \
|
||||
e.evtevent, \
|
||||
COALESCE(r.rolname, '') AS owner, \
|
||||
COALESCE(format('%I.%I(%s)', pn.nspname, p.proname, pg_get_function_arguments(p.oid)), '') AS function, \
|
||||
e.evtenabled::text AS enabled, \
|
||||
NULL::text AS tags, \
|
||||
obj_description(e.oid, 'pg_event_trigger') AS comment, \
|
||||
NULL::text AS source \
|
||||
FROM {cat}.pg_event_trigger e \
|
||||
LEFT JOIN {cat}.pg_roles r ON r.oid = e.evtowner \
|
||||
LEFT JOIN {cat}.pg_proc p ON p.oid = e.evtfoid \
|
||||
LEFT JOIN {cat}.pg_namespace pn ON pn.oid = p.pronamespace \
|
||||
ORDER BY e.evtname"
|
||||
)
|
||||
}
|
||||
|
||||
fn event_trigger_infos_from_query_result(result: db::QueryResult) -> Vec<db::EventTriggerInfo> {
|
||||
result
|
||||
.rows
|
||||
.into_iter()
|
||||
.filter_map(|row| {
|
||||
let name = query_result_cell_string(&row, 0)?;
|
||||
Some(db::EventTriggerInfo {
|
||||
name,
|
||||
event: query_result_cell_string(&row, 1).unwrap_or_default(),
|
||||
owner: query_result_cell_string(&row, 2).filter(|s| !s.is_empty()),
|
||||
function: query_result_cell_string(&row, 3).filter(|s| !s.is_empty()),
|
||||
enabled: query_result_cell_string(&row, 4).filter(|s| !s.is_empty()),
|
||||
tags: None,
|
||||
comment: query_result_cell_string(&row, 6).filter(|s| !s.is_empty()),
|
||||
source: None,
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn extension_infos_from_query_result(result: db::QueryResult, include_schema: bool) -> Vec<db::ExtensionInfo> {
|
||||
result
|
||||
.rows
|
||||
|
||||
@@ -8361,6 +8361,41 @@ pub async fn list_extensions_core(
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn list_event_triggers_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
database: &str,
|
||||
) -> Result<Vec<db::EventTriggerInfo>, String> {
|
||||
retry_metadata_connection(state, connection_id, Some(database), || async {
|
||||
let pool_key = state.get_or_create_metadata_pool_for_session(connection_id, Some(database), None).await?;
|
||||
let db_config = connection_config(state, connection_id).await;
|
||||
if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::Kingbase) {
|
||||
let pool_handle = state.pool_handle(&pool_key).await;
|
||||
if let Some(client) = extract_pool!(pool_handle.as_ref(), Agent) {
|
||||
return kingbase::list_event_triggers(client, database, agent_metadata_timeout(db_config.as_ref()))
|
||||
.await;
|
||||
}
|
||||
}
|
||||
|
||||
// HighGo, Vastbase, and other PostgreSQL-compatible catalogs exposed
|
||||
// through Agent pools still ship pg_event_trigger natively, so reuse
|
||||
// the native metadata fallback used for extension metadata.
|
||||
if let Some(config) = agent_postgres_extension_fallback_config(db_config.as_ref()) {
|
||||
if let Some(pool) = native_postgres_metadata_pool(state, connection_id, database, config).await? {
|
||||
return db::postgres::list_event_triggers(&pool).await;
|
||||
}
|
||||
}
|
||||
|
||||
let pool = clone_metadata_pool(state, &pool_key).await.ok_or("Pool not found")?;
|
||||
|
||||
match &pool {
|
||||
PoolKind::Postgres(p) => db::postgres::list_event_triggers(p).await,
|
||||
_ => Ok(vec![]),
|
||||
}
|
||||
})
|
||||
.await
|
||||
}
|
||||
|
||||
pub async fn list_available_extensions_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
|
||||
@@ -36,8 +36,8 @@ use crate::types::{
|
||||
ColumnInfo, ColumnMetadataCapabilities, CompletionAssistantCandidate, CompletionAssistantCandidateKind,
|
||||
CompletionAssistantMatchMode, CompletionAssistantObjectKind, CompletionAssistantRequest,
|
||||
CompletionAssistantResponse, ConstraintInfo, CustomTypeDdl, CustomTypeDetails, CustomTypeDomainConstraint,
|
||||
CustomTypeKind, CustomTypeMember, CustomTypeProperties, DatabaseInfo, DatabaseStorageInfo, ExtensionInfo,
|
||||
ForeignKeyInfo, FunctionInfo, IndexInfo, ObjectInfo, ObjectStatistics, OwnerInfo, PgPartitionBound,
|
||||
CustomTypeKind, CustomTypeMember, CustomTypeProperties, DatabaseInfo, DatabaseStorageInfo, EventTriggerInfo,
|
||||
ExtensionInfo, ForeignKeyInfo, FunctionInfo, IndexInfo, ObjectInfo, ObjectStatistics, OwnerInfo, PgPartitionBound,
|
||||
PgPartitionKind, PgPartitionNode, PgTablePartitioning, QueryMessage, QueryResult, RuleInfo, SchemaInfo,
|
||||
SequenceInfo, SpatialColumnBuilder, TableInfo, TriggerInfo,
|
||||
};
|
||||
@@ -9760,6 +9760,84 @@ pub async fn list_available_extensions(pool: &Pool) -> Result<Vec<ExtensionInfo>
|
||||
.collect())
|
||||
}
|
||||
|
||||
fn postgres_event_trigger_catalog_exists_sql() -> &'static str {
|
||||
"SELECT EXISTS ( \
|
||||
SELECT 1 FROM pg_catalog.pg_class c \
|
||||
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace \
|
||||
WHERE n.nspname = 'pg_catalog' AND c.relname = 'pg_event_trigger' \
|
||||
)"
|
||||
}
|
||||
|
||||
async fn postgres_event_trigger_catalog_exists(client: &deadpool_postgres::Client) -> Result<bool, String> {
|
||||
let row = postgres_query_one_cached(client, postgres_event_trigger_catalog_exists_sql(), &[])
|
||||
.await
|
||||
.map_err(|e| e.to_string())?;
|
||||
Ok(pg_row_try_bool(&row, 0).unwrap_or(false))
|
||||
}
|
||||
|
||||
fn postgres_event_triggers_sql() -> &'static str {
|
||||
"SELECT e.evtname, \
|
||||
e.evtevent, \
|
||||
COALESCE(r.rolname, '') AS owner, \
|
||||
COALESCE(format('%I.%I(%s)', pn.nspname, p.proname, pg_get_function_identity_arguments(p.oid)), '') AS function, \
|
||||
e.evtenabled::text AS enabled, \
|
||||
e.evttags AS tags, \
|
||||
obj_description(e.oid, 'pg_event_trigger') AS comment, \
|
||||
pg_get_eventtriggerdef(e.oid) AS source \
|
||||
FROM pg_catalog.pg_event_trigger e \
|
||||
LEFT JOIN pg_catalog.pg_roles r ON r.oid = e.evtowner \
|
||||
LEFT JOIN pg_catalog.pg_proc p ON p.oid = e.evtfoid \
|
||||
LEFT JOIN pg_catalog.pg_namespace pn ON pn.oid = p.pronamespace \
|
||||
ORDER BY e.evtname"
|
||||
}
|
||||
|
||||
fn postgres_event_triggers_sourceless_sql() -> &'static str {
|
||||
"SELECT e.evtname, \
|
||||
e.evtevent, \
|
||||
COALESCE(r.rolname, '') AS owner, \
|
||||
COALESCE(format('%I.%I(%s)', pn.nspname, p.proname, pg_get_function_arguments(p.oid)), '') AS function, \
|
||||
e.evtenabled::text AS enabled, \
|
||||
e.evttags AS tags, \
|
||||
obj_description(e.oid, 'pg_event_trigger') AS comment, \
|
||||
NULL::text AS source \
|
||||
FROM pg_catalog.pg_event_trigger e \
|
||||
LEFT JOIN pg_catalog.pg_roles r ON r.oid = e.evtowner \
|
||||
LEFT JOIN pg_catalog.pg_proc p ON p.oid = e.evtfoid \
|
||||
LEFT JOIN pg_catalog.pg_namespace pn ON pn.oid = p.pronamespace \
|
||||
ORDER BY e.evtname"
|
||||
}
|
||||
|
||||
/// Lists PostgreSQL event triggers (`pg_event_trigger`). Event triggers are
|
||||
/// database-level objects, so the query takes no schema parameter. Stripped
|
||||
/// PostgreSQL-compatible kernels that lack the catalog report an empty list,
|
||||
/// and servers that expose the catalog without `pg_get_eventtriggerdef` fall
|
||||
/// back to a sourceless listing so the trigger identity is still visible.
|
||||
pub async fn list_event_triggers(pool: &Pool) -> Result<Vec<EventTriggerInfo>, String> {
|
||||
let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
|
||||
if !postgres_event_trigger_catalog_exists(&client).await.unwrap_or(false) {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
let rows = match postgres_query_cached(&client, postgres_event_triggers_sql(), &[]).await {
|
||||
Ok(rows) => rows,
|
||||
Err(primary_error) => postgres_query_cached(&client, postgres_event_triggers_sourceless_sql(), &[])
|
||||
.await
|
||||
.map_err(|fallback_error| format!("{primary_error}; sourceless fallback failed: {fallback_error}"))?,
|
||||
};
|
||||
Ok(rows
|
||||
.iter()
|
||||
.map(|row| EventTriggerInfo {
|
||||
name: pg_row_try_string(row, 0),
|
||||
event: pg_row_try_string(row, 1),
|
||||
owner: row.try_get::<_, Option<String>>(2).ok().flatten().filter(|s| !s.is_empty()),
|
||||
function: row.try_get::<_, Option<String>>(3).ok().flatten().filter(|s| !s.is_empty()),
|
||||
enabled: row.try_get::<_, Option<String>>(4).ok().flatten().filter(|s| !s.is_empty()),
|
||||
tags: row.try_get::<_, Option<Vec<String>>>(5).ok().flatten(),
|
||||
comment: row.try_get::<_, Option<String>>(6).ok().flatten().filter(|s| !s.is_empty()),
|
||||
source: row.try_get::<_, Option<String>>(7).ok().flatten().filter(|s| !s.is_empty()),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn list_owners(pool: &Pool, schema: &str) -> Result<Vec<OwnerInfo>, String> {
|
||||
let client = checkout_postgres_client(pool, None, super::connection_timeout()).await?;
|
||||
let rows = postgres_query_cached(&client, POSTGRES_OWNERS_SQL, &[&schema]).await.map_err(|e| e.to_string())?;
|
||||
|
||||
@@ -145,6 +145,33 @@ pub struct ExtensionInfo {
|
||||
pub schema: Option<String>,
|
||||
}
|
||||
|
||||
/// A PostgreSQL event trigger (`pg_event_trigger`). Event triggers fire on DDL
|
||||
/// commands at the database level, independent of any schema. This is distinct
|
||||
/// from MySQL events (`MysqlEventInfo`) and per-table triggers (`TriggerInfo`).
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct EventTriggerInfo {
|
||||
pub name: String,
|
||||
/// DDL event: ddl_command_start | ddl_command_end | sql_drop | table_rewrite.
|
||||
pub event: String,
|
||||
/// Owner role name.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub owner: Option<String>,
|
||||
/// `schema.function(args)` executed by the trigger.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub function: Option<String>,
|
||||
/// Session replica status char: O | A | R | D.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub enabled: Option<String>,
|
||||
/// Command tags in the WHEN clause (NULL = all tags).
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub tags: Option<Vec<String>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub comment: Option<String>,
|
||||
/// `pg_get_eventtriggerdef` reconstruction of the CREATE statement.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub source: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
|
||||
pub struct ObjectStatistics {
|
||||
pub name: String,
|
||||
|
||||
@@ -600,6 +600,7 @@ async fn serve() {
|
||||
.route("/schema/table-owner", get(routes::schema::get_table_owner))
|
||||
.route("/schema/extensions", get(routes::schema::list_extensions))
|
||||
.route("/schema/available-extensions", get(routes::schema::list_available_extensions))
|
||||
.route("/schema/event-triggers", get(routes::schema::list_event_triggers))
|
||||
.route("/schema/ddl", get(routes::schema::get_ddl))
|
||||
.route("/docs/snapshot", post(routes::docs::collect_snapshot))
|
||||
.route("/docs/annotations/load", post(routes::docs::load_annotations))
|
||||
|
||||
@@ -889,6 +889,17 @@ pub async fn list_available_extensions(
|
||||
Ok(Json(serde_json::to_value(result).map_err(|e| AppError::from(e.to_string()))?))
|
||||
}
|
||||
|
||||
pub async fn list_event_triggers(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Query(q): Query<SchemaQuery>,
|
||||
) -> Result<Json<serde_json::Value>, AppError> {
|
||||
let database = q.database.as_deref().unwrap_or("");
|
||||
let result = dbx_core::schema::list_event_triggers_core(&state.app, &q.connection_id, database)
|
||||
.await
|
||||
.map_err(AppError::from)?;
|
||||
Ok(Json(serde_json::to_value(result).map_err(|e| AppError::from(e.to_string()))?))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
Reference in new issue
Block a user