chore(pg-meta): migrate triggers (#33675)

packages/pg-meta/src/index.ts

@@ -2,6 +2,7 @@ import roles from './pg-meta-roles'
 import schemas from './pg-meta-schemas'
 import * as functions from './pg-meta-functions'
 import tablePrivileges from './pg-meta-table-privileges'
+import triggers from './pg-meta-triggers'
 import types from './pg-meta-types'
 import version from './pg-meta-version'
 import indexes from './pg-meta-indexes'
@@ -12,6 +13,7 @@ export default {
   schemas,
   functions,
   tablePrivileges,
+  triggers,
   types,
   version,
   indexes,

packages/pg-meta/src/pg-meta-triggers.ts

@@ -0,0 +1,237 @@
+import { ident, literal } from './pg-format'
+import { z } from 'zod'
+import { DEFAULT_SYSTEM_SCHEMAS } from './constants'
+import { filterByList } from './helpers'
+import { TRIGGERS_SQL } from './sql/triggers'
+
+type TriggerIdentifier = Pick<PGTrigger, 'id'> | Pick<PGTrigger, 'name' | 'schema' | 'table'>
+
+function getIdentifierWhereClause(identifier: TriggerIdentifier): string {
+  if ('id' in identifier && identifier.id) {
+    return `${ident('id')} = ${literal(identifier.id)}`
+  }
+  if ('name' in identifier && identifier.name && identifier.table && identifier.schema) {
+    return `${ident('name')} = ${literal(identifier.name)} and ${ident('schema')} = ${literal(identifier.schema)} and ${ident('table')} = ${literal(identifier.table)}`
+  }
+  throw new Error('Must provide either id or name, schema and table')
+}
+
+export const pgTriggerZod = z.object({
+  id: z.number(),
+  table_id: z.number(),
+  enabled_mode: z.enum(['DISABLED', 'ORIGIN', 'REPLICA', 'ALWAYS']),
+  function_args: z.array(z.string()),
+  name: z.string(),
+  table: z.string(),
+  schema: z.string(),
+  condition: z.string().nullable(),
+  orientation: z.string(),
+  activation: z.string(),
+  events: z.array(z.string()),
+  function_name: z.string(),
+  function_schema: z.string(),
+})
+
+export type PGTrigger = z.infer<typeof pgTriggerZod>
+
+export const pgTriggerArrayZod = z.array(pgTriggerZod)
+export const pgTriggerOptionalZod = z.optional(pgTriggerZod)
+
+export function list({
+  includeSystemSchemas = false,
+  includedSchemas,
+  excludedSchemas,
+  limit,
+  offset,
+}: {
+  includeSystemSchemas?: boolean
+  includedSchemas?: string[]
+  excludedSchemas?: string[]
+  limit?: number
+  offset?: number
+} = {}): {
+  sql: string
+  zod: typeof pgTriggerArrayZod
+} {
+  let sql = `with triggers as (${TRIGGERS_SQL}) select * from triggers`
+  const filter = filterByList(
+    includedSchemas,
+    excludedSchemas,
+    !includeSystemSchemas ? DEFAULT_SYSTEM_SCHEMAS : undefined
+  )
+  if (filter) {
+    sql += ` where schema ${filter}`
+  }
+  if (limit) {
+    sql += ` limit ${limit}`
+  }
+  if (offset) {
+    sql += ` offset ${offset}`
+  }
+  return {
+    sql,
+    zod: pgTriggerArrayZod,
+  }
+}
+
+type TriggersRetrieveReturn = {
+  sql: string
+  zod: typeof pgTriggerOptionalZod
+}
+
+export function retrieve(identifier: TriggerIdentifier): TriggersRetrieveReturn
+export function retrieve(params: TriggerIdentifier): TriggersRetrieveReturn {
+  const whereIdentifierCondition = getIdentifierWhereClause(params)
+  const sql = `with triggers as (${TRIGGERS_SQL}) select * from triggers where ${whereIdentifierCondition};`
+  return {
+    sql,
+    zod: pgTriggerOptionalZod,
+  }
+}
+
+export const pgTriggerCreateZod = z.object({
+  name: z.string(),
+  schema: z.string().optional().default('public'),
+  table: z.string(),
+  function_schema: z.string().optional().default('public'),
+  function_name: z.string(),
+  function_args: z.array(z.string()).optional(),
+  activation: z.enum(['BEFORE', 'AFTER', 'INSTEAD OF']),
+  events: z.array(z.string()),
+  orientation: z.enum(['ROW', 'STATEMENT']).optional(),
+  condition: z.string().optional(),
+})
+
+export type PGTriggerCreate = z.infer<typeof pgTriggerCreateZod>
+
+export function create({
+  name,
+  schema = 'public',
+  table,
+  function_schema = 'public',
+  function_name,
+  function_args = [],
+  activation,
+  events,
+  orientation,
+  condition,
+}: PGTriggerCreate): {
+  sql: string
+  zod: z.ZodType<void>
+} {
+  const qualifiedTableName = `${ident(schema)}.${ident(table)}`
+  const qualifiedFunctionName = `${ident(function_schema)}.${ident(function_name)}`
+  const triggerEvents = events.join(' OR ')
+  const triggerOrientation = orientation ? `FOR EACH ${orientation}` : ''
+  const triggerCondition = condition ? `WHEN (${condition})` : ''
+  const functionArgsStr = function_args.map(literal).join(',')
+
+  const sql = `CREATE TRIGGER ${ident(
+    name
+  )} ${activation} ${triggerEvents} ON ${qualifiedTableName} ${triggerOrientation} ${triggerCondition} EXECUTE FUNCTION ${qualifiedFunctionName}(${functionArgsStr});`
+
+  return {
+    sql,
+    zod: z.void(),
+  }
+}
+
+export const pgTriggerUpdateZod = z.object({
+  name: z.string().optional(),
+  enabled_mode: z.enum(['ORIGIN', 'REPLICA', 'ALWAYS', 'DISABLED']).optional(),
+})
+
+export type PGTriggerUpdate = z.infer<typeof pgTriggerUpdateZod>
+
+export function update(
+  identifier: TriggerIdentifier,
+  params: PGTriggerUpdate
+): {
+  sql: string
+} {
+  const whereIdentifierCondition = getIdentifierWhereClause(identifier)
+
+  const sql = `
+do $$
+declare
+  old record;
+begin
+  with triggers as (${TRIGGERS_SQL})
+  select * into old from triggers where ${whereIdentifierCondition};
+  
+  if old is null then
+    raise exception 'Cannot find trigger: %', ${literal(whereIdentifierCondition)};
+  end if;
+
+  ${
+    params.enabled_mode
+      ? `
+  execute(format('alter table %I.%I ${
+    params.enabled_mode === 'DISABLED'
+      ? 'DISABLE'
+      : 'ENABLE' +
+        (params.enabled_mode === 'ALWAYS' || params.enabled_mode === 'REPLICA'
+          ? ' ' + params.enabled_mode
+          : '')
+  } TRIGGER %I', 
+    old.schema, old.table, old.name));`
+      : ''
+  }
+
+  ${
+    params.name
+      ? `
+    -- Using the same name in the rename clause gives an error, so only do it if the new name is different.
+  if ${literal(params.name)} != old.name then
+    execute(format('alter trigger %I on %I.%I rename to %I;', old.name, old.schema, old.table, ${literal(params.name)}));
+  end if;`
+      : ''
+  }
+end
+$$;`
+
+  return {
+    sql,
+  }
+}
+
+export function remove(
+  identifier: TriggerIdentifier,
+  { cascade = false } = {}
+): {
+  sql: string
+  zod: z.ZodType<void>
+} {
+  const whereIdentifierCondition = getIdentifierWhereClause(identifier)
+
+  const sql = `
+do $$
+declare
+  old record;
+begin
+  with triggers as (${TRIGGERS_SQL})
+  select * into old from triggers where ${whereIdentifierCondition};
+  
+  if old is null then
+    raise exception 'Cannot find trigger';
+  end if;
+
+  execute(format('DROP TRIGGER %I ON %I.%I ${cascade ? 'CASCADE' : ''}',
+    old.name, old.schema, old.table));
+end
+$$;`
+
+  return {
+    sql,
+    zod: z.void(),
+  }
+}
+
+export default {
+  list,
+  retrieve,
+  create,
+  update,
+  remove,
+  zod: pgTriggerZod,
+}

packages/pg-meta/src/sql/triggers.ts

@@ -0,0 +1,52 @@
+export const TRIGGERS_SQL = /* SQL */ `
+SELECT
+  pg_t.oid AS id,
+  pg_t.tgrelid AS table_id,
+  CASE
+    WHEN pg_t.tgenabled = 'D' THEN 'DISABLED'
+    WHEN pg_t.tgenabled = 'O' THEN 'ORIGIN'
+    WHEN pg_t.tgenabled = 'R' THEN 'REPLICA'
+    WHEN pg_t.tgenabled = 'A' THEN 'ALWAYS'
+  END AS enabled_mode,
+  (
+    STRING_TO_ARRAY(
+      ENCODE(pg_t.tgargs, 'escape'), '\\000'
+    )
+  )[:pg_t.tgnargs] AS function_args,
+  is_t.trigger_name AS name,
+  is_t.event_object_table AS table,
+  is_t.event_object_schema AS schema,
+  is_t.action_condition AS condition,
+  is_t.action_orientation AS orientation,
+  is_t.action_timing AS activation,
+  ARRAY_AGG(is_t.event_manipulation)::text[] AS events,
+  pg_p.proname AS function_name,
+  pg_n.nspname AS function_schema
+FROM
+  pg_trigger AS pg_t
+JOIN
+  pg_class AS pg_c
+ON pg_t.tgrelid = pg_c.oid
+JOIN information_schema.triggers AS is_t
+ON is_t.trigger_name = pg_t.tgname
+AND pg_c.relname = is_t.event_object_table
+AND pg_c.relnamespace = (quote_ident(is_t.event_object_schema))::regnamespace
+JOIN pg_proc AS pg_p
+ON pg_t.tgfoid = pg_p.oid
+JOIN pg_namespace AS pg_n
+ON pg_p.pronamespace = pg_n.oid
+GROUP BY
+  pg_t.oid,
+  pg_t.tgrelid,
+  pg_t.tgenabled,
+  pg_t.tgargs,
+  pg_t.tgnargs,
+  is_t.trigger_name,
+  is_t.event_object_table,
+  is_t.event_object_schema,
+  is_t.action_condition,
+  is_t.action_orientation,
+  is_t.action_timing,
+  pg_p.proname,
+  pg_n.nspname
+`

packages/pg-meta/test/triggers.test.ts

@@ -0,0 +1,364 @@
+import { expect, test, beforeAll, afterAll } from 'vitest'
+import pgMeta from '../src/index'
+import { createTestDatabase, cleanupRoot } from './db/utils'
+
+beforeAll(async () => {
+  // Any global setup if needed
+})
+
+afterAll(async () => {
+  await cleanupRoot()
+})
+
+const withTestDatabase = (
+  name: string,
+  fn: (db: Awaited<ReturnType<typeof createTestDatabase>>) => Promise<void>
+) => {
+  test(name, async () => {
+    const db = await createTestDatabase()
+    try {
+      await fn(db)
+    } finally {
+      await db.cleanup()
+    }
+  })
+}
+
+withTestDatabase('retrieve, create, update, delete', async ({ executeQuery }) => {
+  // Create trigger
+  const { sql: createSql } = pgMeta.triggers.create({
+    name: 'test_trigger',
+    schema: 'public',
+    table: 'users_audit',
+    function_schema: 'public',
+    function_name: 'audit_action',
+    function_args: ['test1', 'test2'],
+    activation: 'AFTER',
+    events: ['UPDATE'],
+    orientation: 'ROW',
+    condition: '(old.* IS DISTINCT FROM new.*)',
+  })
+  await executeQuery(createSql)
+
+  const { sql: listTriggers, zod: listZod } = pgMeta.triggers.list()
+  const listedTriggers = await listZod.parse(await executeQuery(listTriggers))
+  const createdTriggers = listedTriggers.find(
+    (t) => t.name === 'test_trigger' && t.table === 'users_audit' && t.schema === 'public'
+  )
+  expect(createdTriggers!).toMatchInlineSnapshot(
+    { id: expect.any(Number), table_id: expect.any(Number) },
+    `
+{
+  "activation": "AFTER",
+  "condition": "(old.* IS DISTINCT FROM new.*)",
+  "enabled_mode": "ORIGIN",
+  "events": [
+    "UPDATE",
+  ],
+  "function_args": [
+    "test1",
+    "test2",
+  ],
+  "function_name": "audit_action",
+  "function_schema": "public",
+  "id": Any<Number>,
+  "name": "test_trigger",
+  "orientation": "ROW",
+  "schema": "public",
+  "table": "users_audit",
+  "table_id": Any<Number>,
+}
+  `
+  )
+
+  // Retrieve created trigger by name
+  const { sql: retrieveSqlByName, zod: retrieveZod } = pgMeta.triggers.retrieve({
+    name: 'test_trigger',
+    table: 'users_audit',
+    schema: 'public',
+  })
+  const trigger = retrieveZod.parse((await executeQuery(retrieveSqlByName))[0])
+  expect(trigger!).toMatchInlineSnapshot(
+    { id: expect.any(Number), table_id: expect.any(Number) },
+    `
+{
+  "activation": "AFTER",
+  "condition": "(old.* IS DISTINCT FROM new.*)",
+  "enabled_mode": "ORIGIN",
+  "events": [
+    "UPDATE",
+  ],
+  "function_args": [
+    "test1",
+    "test2",
+  ],
+  "function_name": "audit_action",
+  "function_schema": "public",
+  "id": Any<Number>,
+  "name": "test_trigger",
+  "orientation": "ROW",
+  "schema": "public",
+  "table": "users_audit",
+  "table_id": Any<Number>,
+}
+  `
+  )
+
+  // Retrieve created trigger by id
+  const { sql: retrieveSqlById } = pgMeta.triggers.retrieve({
+    id: trigger!.id,
+  })
+  const triggerById = retrieveZod.parse((await executeQuery(retrieveSqlById))[0])
+  expect(triggerById!).toMatchInlineSnapshot(
+    { id: expect.any(Number), table_id: expect.any(Number) },
+    `
+{
+  "activation": "AFTER",
+  "condition": "(old.* IS DISTINCT FROM new.*)",
+  "enabled_mode": "ORIGIN",
+  "events": [
+    "UPDATE",
+  ],
+  "function_args": [
+    "test1",
+    "test2",
+  ],
+  "function_name": "audit_action",
+  "function_schema": "public",
+  "id": Any<Number>,
+  "name": "test_trigger",
+  "orientation": "ROW",
+  "schema": "public",
+  "table": "users_audit",
+  "table_id": Any<Number>,
+}
+  `
+  )
+
+  // Update trigger
+  const { sql: updateSql } = pgMeta.triggers.update(trigger!, {
+    name: 'test_trigger_renamed',
+    enabled_mode: 'DISABLED',
+  })
+  await executeQuery(updateSql)
+
+  // Verify update
+  const { sql: retrieveUpdatedSql } = pgMeta.triggers.retrieve({
+    id: trigger!.id,
+  })
+  const updatedTrigger = retrieveZod.parse((await executeQuery(retrieveUpdatedSql))[0])
+  expect(updatedTrigger).toMatchInlineSnapshot(
+    { id: expect.any(Number), table_id: expect.any(Number) },
+    `
+{
+  "activation": "AFTER",
+  "condition": "(old.* IS DISTINCT FROM new.*)",
+  "enabled_mode": "DISABLED",
+  "events": [
+    "UPDATE",
+  ],
+  "function_args": [
+    "test1",
+    "test2",
+  ],
+  "function_name": "audit_action",
+  "function_schema": "public",
+  "id": Any<Number>,
+  "name": "test_trigger_renamed",
+  "orientation": "ROW",
+  "schema": "public",
+  "table": "users_audit",
+  "table_id": Any<Number>,
+}
+`
+  )
+
+  // Update trigger again
+  const { sql: updateSql2 } = pgMeta.triggers.update(updatedTrigger!, {
+    enabled_mode: 'REPLICA',
+  })
+  await executeQuery(updateSql2)
+
+  // Verify second update
+  const { sql: retrieveUpdated2Sql } = pgMeta.triggers.retrieve({
+    name: 'test_trigger_renamed',
+    table: 'users_audit',
+    schema: 'public',
+  })
+  const updatedTrigger2 = retrieveZod.parse((await executeQuery(retrieveUpdated2Sql))[0])
+  expect(updatedTrigger2).toMatchObject({
+    enabled_mode: 'REPLICA',
+  })
+
+  // Remove trigger
+  const { sql: removeSql } = pgMeta.triggers.remove(updatedTrigger2!)
+  await executeQuery(removeSql)
+
+  // Verify removal
+  const { sql: verifyRemoveSql } = pgMeta.triggers.retrieve({
+    name: 'test_trigger_renamed',
+    table: 'users_audit',
+    schema: 'public',
+  })
+  const result = retrieveZod.parse((await executeQuery(verifyRemoveSql))[0])
+  expect(result).toBeUndefined()
+})
+
+withTestDatabase('multi event', async ({ executeQuery }) => {
+  // Create trigger
+  const { sql: createSql } = pgMeta.triggers.create({
+    name: 'test_multi_event_trigger',
+    schema: 'public',
+    table: 'users_audit',
+    function_schema: 'public',
+    function_name: 'audit_action',
+    function_args: ['test1', 'test2'],
+    activation: 'AFTER',
+    events: ['insert', 'update', 'delete'],
+    orientation: 'ROW',
+  })
+  await executeQuery(createSql)
+
+  // Verify created trigger
+  const { sql: retrieveSql, zod: retrieveZod } = pgMeta.triggers.retrieve({
+    name: 'test_multi_event_trigger',
+    table: 'users_audit',
+    schema: 'public',
+  })
+  const trigger = retrieveZod.parse((await executeQuery(retrieveSql))[0])
+  expect(trigger).toMatchInlineSnapshot(
+    { id: expect.any(Number), table_id: expect.any(Number) },
+    `
+    {
+      "activation": "AFTER",
+      "condition": null,
+      "enabled_mode": "ORIGIN",
+      "events": [
+        "INSERT",
+        "DELETE",
+        "UPDATE",
+      ],
+      "function_args": [
+        "test1",
+        "test2",
+      ],
+      "function_name": "audit_action",
+      "function_schema": "public",
+      "id": Any<Number>,
+      "name": "test_multi_event_trigger",
+      "orientation": "ROW",
+      "schema": "public",
+      "table": "users_audit",
+      "table_id": Any<Number>,
+    }
+    `
+  )
+
+  // Remove trigger
+  const { sql: removeSql } = pgMeta.triggers.remove(trigger!)
+  await executeQuery(removeSql)
+
+  // Verify removal
+  const { sql: verifyRemoveSql } = pgMeta.triggers.retrieve({
+    name: 'test_multi_event_trigger',
+    table: 'users_audit',
+    schema: 'public',
+  })
+  const result = retrieveZod.parse((await executeQuery(verifyRemoveSql))[0])
+  expect(result).toBeUndefined()
+})
+
+withTestDatabase('triggers with the same name on different schemas', async ({ executeQuery }) => {
+  // Create test schemas and triggers
+  await executeQuery(`
+    create function tr_f() returns trigger language plpgsql as 'begin end';
+    create schema s1; create table s1.t(); create trigger tr before insert on s1.t execute function tr_f();
+    create schema s2; create table s2.t(); create trigger tr before insert on s2.t execute function tr_f();
+  `)
+
+  // List and verify triggers
+  const { sql: listSql, zod: listZod } = pgMeta.triggers.list()
+  const triggers = listZod.parse(await executeQuery(listSql))
+  expect(triggers.map(({ id, table_id, ...trigger }) => trigger)).toMatchInlineSnapshot(`
+    [
+      {
+        "activation": "BEFORE",
+        "condition": null,
+        "enabled_mode": "ORIGIN",
+        "events": [
+          "INSERT",
+        ],
+        "function_args": [],
+        "function_name": "tr_f",
+        "function_schema": "public",
+        "name": "tr",
+        "orientation": "STATEMENT",
+        "schema": "s1",
+        "table": "t",
+      },
+      {
+        "activation": "BEFORE",
+        "condition": null,
+        "enabled_mode": "ORIGIN",
+        "events": [
+          "INSERT",
+        ],
+        "function_args": [],
+        "function_name": "tr_f",
+        "function_schema": "public",
+        "name": "tr",
+        "orientation": "STATEMENT",
+        "schema": "s2",
+        "table": "t",
+      },
+    ]
+  `)
+})
+
+withTestDatabase('triggers on capitalized schema and table names', async ({ executeQuery }) => {
+  // Create test schema and trigger
+  await executeQuery(`
+    CREATE SCHEMA "MySchema";
+    CREATE TABLE "MySchema"."MyTable" (
+      id SERIAL PRIMARY KEY,
+      name TEXT NOT NULL,
+      created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+      updated_at TIMESTAMP
+    );
+    CREATE OR REPLACE FUNCTION "MySchema"."my_trigger_function"()
+    RETURNS TRIGGER AS $$
+    BEGIN
+      NEW.updated_at := CURRENT_TIMESTAMP;
+      RETURN NEW;
+    END;
+    $$ LANGUAGE plpgsql;
+
+    CREATE TRIGGER "my_trigger"
+    BEFORE INSERT ON "MySchema"."MyTable"
+    FOR EACH ROW
+    EXECUTE FUNCTION "MySchema"."my_trigger_function"();
+  `)
+
+  // List and verify triggers
+  const { sql: listSql, zod: listZod } = pgMeta.triggers.list()
+  const triggers = listZod.parse(await executeQuery(listSql))
+  expect(triggers.map(({ id, table_id, ...trigger }) => trigger)).toMatchInlineSnapshot(`
+    [
+      {
+        "activation": "BEFORE",
+        "condition": null,
+        "enabled_mode": "ORIGIN",
+        "events": [
+          "INSERT",
+        ],
+        "function_args": [],
+        "function_name": "my_trigger_function",
+        "function_schema": "MySchema",
+        "name": "my_trigger",
+        "orientation": "ROW",
+        "schema": "MySchema",
+        "table": "MyTable",
+      },
+    ]
+  `)
+})