refact(db): Replace get_deleted_tables function with a view

This function was returning the set of deletion tables. A view seems
better suited for this task since it would allow for applying additional
filters of the result set. This was particularly necessary to easily
make changes to some sqltests due to switching the delete trigger for
the session table.

(cherry picked from commit 32d41639892950b21fafc39b50c10b997c6d5121)

internal/db/schema/migrations/oss/postgres/81/01_deleted_tables_and_triggers.up.sql

@@ -308,6 +308,7 @@ begin;
     'affected by the trigger and the current timestamp. It is used to populate rows '
     'of the deleted tables.';
 
+  -- Removed in 91/05_deletion_tables_view and replaced with a view.
   create function get_deletion_tables() returns setof name
   as $$
     select c.relname

internal/db/schema/migrations/oss/postgres/91/05_deletion_tables_view.up.sql

@@ -0,0 +1,18 @@
+-- Copyright (c) HashiCorp, Inc.
+-- SPDX-License-Identifier: BUSL-1.1
+
+begin;
+  -- Originially added in 81/01_deleted_tables_and_triggers.up.sql
+  -- This is being replaced with a view.
+  drop function get_deletion_tables;
+
+  -- This view uses the pg_catalog to find all tables that end in _deleted and are visibile.
+  -- See: https://www.postgresql.org/docs/current/catalog-pg-class.html
+  --      https://www.postgresql.org/docs/current/functions-info.html#FUNCTIONS-INFO-SCHEMA
+  create view deletion_table as
+    select c.relname as tablename
+      from pg_catalog.pg_class c
+     where c.relkind in ('r') -- r = ordinary table
+       and c.relname operator(pg_catalog.~) '^(.+_deleted)$' collate pg_catalog.default
+       and pg_catalog.pg_table_is_visible(c.oid);
+commit;

internal/db/sqltest/tests/purge/deleted_table_tests.sql

@@ -21,6 +21,15 @@ begin;
     );
   $$ language sql;
 
+  -- tests that the deletion table has the bulk insert trigger
+  create function has_bulk_insert_trigger(deletion_table_name name) returns text
+  as $$
+    select * from collect_tap(
+      has_trigger(op_table(deletion_table_name), 'bulk_insert_deleted_ids'),
+      trigger_is(op_table(deletion_table_name), 'bulk_insert_deleted_ids', 'bulk_insert_deleted_ids')
+    );
+  $$ language sql;
+
   -- tests the public_id column
   create function has_public_id(deletion_table_name name) returns text
   as $$
@@ -72,15 +81,32 @@ begin;
     );
   $$ language sql;
 
+  -- like above, but using the bulk delete trigger
+  create function test_bulk_deletion_table(deletion_table_name name) returns text
+  as $$
+    select * from collect_tap(
+      has_correct_tables(deletion_table_name),
+      has_public_id(deletion_table_name),
+      has_delete_time(deletion_table_name),
+      has_delete_time_index(deletion_table_name),
+      has_bulk_insert_trigger(deletion_table_name)
+    );
+  $$ language sql;
+
   -- 11 tests for each deletion table
   select plan(a.table_count::integer)
     from (
       select 11 * count(*) as table_count
-        from get_deletion_tables()
+        from deletion_table
     ) as a;
 
-    select test_deletion_table(a)
-      from get_deletion_tables() a;
+    select test_deletion_table(a.tablename)
+      from deletion_table a
+     where a.tablename not in ('session_deleted');
+
+    select test_bulk_deletion_table(a.tablename)
+      from deletion_table a
+     where a.tablename in ('session_deleted');
 
   select * from finish();
 rollback;

internal/pagination/purge/purge_test.go

@@ -26,7 +26,7 @@ func TestPurgeTables(t *testing.T) {
 		t.Errorf("error getting db connection %s", err)
 	}
 
-	rows, err := db.Query("select get_deletion_tables()")
+	rows, err := db.Query("select tablename from deletion_table")
 	if err != nil {
 		t.Errorf("unable to query for deletion tables %s", err)
 	}

internal/pagination/purge/query.go

@@ -5,7 +5,8 @@ package purge
 
 const (
 	getDeletionTablesQuery = `
-select get_deletion_tables();
+select tablename
+  from deletion_table;
 `
 	deleteQueryTemplate = `
 delete from %s where delete_time < now() - interval '30 days'