diff --git a/internal/db/schema/migrations/oss/postgres/78/01_deleted_tables_and_triggers.up.sql b/internal/db/schema/migrations/oss/postgres/78/01_deleted_tables_and_triggers.up.sql
new file mode 100644
index 000000000..6d47514ea
--- /dev/null
+++ b/internal/db/schema/migrations/oss/postgres/78/01_deleted_tables_and_triggers.up.sql
@@ -0,0 +1,341 @@
+-- Copyright (c) HashiCorp, Inc.
+-- SPDX-License-Identifier: BUSL-1.1
+
+begin;
+  create table target_tcp_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table target_tcp_deleted is
+    'target_tcp_deleted holds the ID and delete_time of every deleted TCP target. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table target_ssh_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table target_ssh_deleted is
+    'target_ssh_deleted holds the ID and delete_time of every deleted SSH target. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Target repository list logic is implemented
+  -- against views, so use a view for the deleted
+  -- ids too.
+  -- Note: does not end in _deleted so as
+  -- not to be attached a trigger in the below function.
+  create view target_all_subtypes_deleted_view
+  as
+    select public_id, delete_time from target_tcp_deleted
+    union
+    select public_id, delete_time from target_ssh_deleted;
+  comment on view target_all_subtypes_deleted_view is
+    'target_all_subtypes_deleted_view holds the ID and delete_time of every deleted target.';
+
+  -- Sessions
+  create table session_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table session_deleted is
+    'session_deleted holds the ID and delete_time of every deleted session. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Auth methods
+  create table auth_oidc_method_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table auth_oidc_method_deleted is
+    'auth_oidc_method_deleted holds the ID and delete_time of every deleted OIDC auth method. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table auth_ldap_method_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table auth_ldap_method_deleted is
+    'auth_ldap_method_deleted holds the ID and delete_time of every deleted LDAP auth method. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table auth_password_method_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table auth_password_method_deleted is
+    'auth_password_method_deleted holds the ID and delete_time of every deleted password auth method. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Auth tokens
+  create table auth_token_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table auth_token_deleted is
+    'auth_token_deleted holds the ID and delete_time of every deleted auth token. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Auth accounts
+  create table auth_oidc_account_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table auth_oidc_account_deleted is
+    'auth_oidc_account_deleted holds the ID and delete_time of every deleted OIDC account. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table auth_ldap_account_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table auth_ldap_account_deleted is
+    'auth_ldap_account_deleted holds the ID and delete_time of every deleted LDAP account. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table auth_password_account_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table auth_password_account_deleted is
+    'auth_password_account_deleted holds the ID and delete_time of every deleted password account. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Auth managed groups
+  create table auth_oidc_managed_group_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table auth_oidc_managed_group_deleted is
+    'auth_oidc_managed_group_deleted holds the ID and delete_time of every deleted OIDC managed group. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table auth_ldap_managed_group_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table auth_ldap_managed_group_deleted is
+    'auth_ldap_managed_group_deleted holds the ID and delete_time of every deleted LDAP managed group. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Credential libraries
+  create table credential_vault_library_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table credential_vault_library_deleted is
+    'credential_vault_library_deleted holds the ID and delete_time of every deleted Vault credential library. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table credential_vault_ssh_cert_library_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table credential_vault_ssh_cert_library_deleted is
+    'credential_vault_ssh_cert_library_deleted holds the ID and delete_time of '
+    'every deleted Vault SSH certificate credential library. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Credentials
+  create table credential_static_json_credential_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table credential_static_json_credential_deleted is
+    'credential_static_json_credential_deleted holds the ID and delete_time '
+    'of every deleted static JSON credential. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table credential_static_username_password_credential_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table credential_static_username_password_credential_deleted is
+    'credential_static_username_password_credential_deleted holds the ID and delete_time '
+    'of every deleted static username password credential. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table credential_static_ssh_private_key_credential_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table credential_static_ssh_private_key_credential_deleted is
+    'credential_static_ssh_private_key_credential_deleted holds the ID and delete_time '
+    'of every deleted static ssh private key credential. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Credential stores
+  create table credential_vault_store_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table credential_vault_store_deleted is
+    'credential_vault_store_deleted holds the ID and delete_time '
+    'of every deleted Vault credential store. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table credential_static_store_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table credential_static_store_deleted is
+    'credential_static_store_deleted holds the ID and delete_time '
+    'of every deleted static credential store. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Groups
+  create table iam_group_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table iam_group_deleted is
+    'iam_group_deleted holds the ID and delete_time of every deleted group. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Roles
+  create table iam_role_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table iam_role_deleted is
+    'iam_role_deleted holds the ID and delete_time of every deleted role. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Scopes
+  create table iam_scope_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table iam_scope_deleted is
+    'iam_scope_deleted holds the ID and delete_time of every deleted scope. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Users
+  create table iam_user_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table iam_user_deleted is
+    'iam_user_deleted holds the ID and delete_time of every deleted user. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Host catalogs
+  create table host_plugin_catalog_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table host_plugin_catalog_deleted is
+    'host_plugin_catalog_deleted holds the ID and delete_time of every deleted plugin host catalog. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table static_host_catalog_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table static_host_catalog_deleted is
+    'static_host_catalog_deleted holds the ID and delete_time of every deleted static host catalog. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Host sets
+  create table host_plugin_set_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table host_plugin_set_deleted is
+    'host_plugin_set_deleted holds the ID and delete_time of every deleted plugin host set. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table static_host_set_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table static_host_set_deleted is
+    'static_host_set_deleted holds the ID and delete_time of every deleted static host set. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Hosts
+  create table host_plugin_host_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table host_plugin_host_deleted is
+    'host_plugin_host_deleted holds the ID and delete_time of every deleted plugin host. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create table static_host_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table static_host_deleted is
+    'static_host_deleted holds the ID and delete_time of every deleted static host. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Storage buckets
+  create table storage_plugin_storage_bucket_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table storage_plugin_storage_bucket_deleted is
+    'storage_plugin_storage_bucket_deleted holds the ID and delete_time of every deleted storage bucket. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Session recordings
+  create table recording_session_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table recording_session_deleted is
+    'recording_session_deleted holds the ID and delete_time of every deleted session recording. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  -- Workers
+  create table server_worker_deleted (
+    public_id wt_public_id primary key,
+    delete_time wt_timestamp not null
+  );
+  comment on table server_worker_deleted is
+    'server_worker_deleted holds the ID and delete_time of every deleted worker. '
+    'It is automatically trimmed of records older than 30 days by a job.';
+
+  create function insert_deleted_id() returns trigger
+  as $$
+  begin
+    execute format('insert into %I (public_id, delete_time) values ($1, now()) on conflict (public_id) do update set delete_time = excluded.delete_time', tg_argv[0]) using old.public_id;
+    return old;
+  end;
+  $$ language plpgsql;
+  comment on function insert_deleted_id is
+    'insert_deleted_id is a function that inserts a record into the table '
+    'specified by the first trigger argument. It takes the public ID from the row '
+    'affected by the trigger and the current timestamp. It is used to populate rows '
+    'of the deleted tables.';
+
+  create function get_deletion_tables() returns setof name
+  as $$
+    select c.relname
+      from pg_catalog.pg_class c
+     where c.relkind in ('r')
+       and c.relname operator(pg_catalog.~) '^(.+_deleted)$' collate pg_catalog.default
+       and pg_catalog.pg_table_is_visible(c.oid);
+  $$ language sql;
+  comment on function get_deletion_tables is
+    'get_deletion_tables returns a set containing all the deleted table names by looking for '
+    'all tables that end in _deleted.';
+
+  -- Assign a "after delete" trigger on all deleted tables to run insert_deleted_id
+  do $$
+  declare
+    deletion_table_name text;
+    table_name          text;
+  begin
+    for deletion_table_name in select get_deletion_tables() loop
+      table_name := split_part(deletion_table_name, '_deleted', 1);
+      execute format('create trigger insert_deleted_id after delete on %I for each row execute function insert_deleted_id(''%I'')', table_name, deletion_table_name);
+      execute format('create index %I_delete_time_idx on %I (delete_time)', deletion_table_name, deletion_table_name);
+    end loop;
+  end;
+  $$ language plpgsql;
+
+  -- These credential indices end up truncated by postgres, so we manually
+  -- rename them to something nicer.
+  alter index credential_static_username_password_credential_deleted_delete_t rename to credential_static_username_password_deleted_delete_time_idx;
+  alter index credential_static_ssh_private_key_credential_deleted_delete_tim rename to credential_static_ssh_private_key_deleted_delete_time_idx;
+commit;
diff --git a/internal/db/sqltest/Makefile b/internal/db/sqltest/Makefile
index 3dafeebca..37bce01aa 100644
--- a/internal/db/sqltest/Makefile
+++ b/internal/db/sqltest/Makefile
@@ -33,7 +33,8 @@ TESTS ?= tests/setup/*.sql \
 		 tests/domain/*.sql \
 		 tests/history/*.sql \
 		 tests/recording/*.sql \
-		 tests/auth/*/*.sql
+		 tests/auth/*/*.sql \
+		 tests/purge/*sql
 
 POSTGRES_DOCKER_IMAGE_BASE ?= postgres
 
diff --git a/internal/db/sqltest/tests/purge/deleted_table_tests.sql b/internal/db/sqltest/tests/purge/deleted_table_tests.sql
new file mode 100644
index 000000000..9d303f2a0
--- /dev/null
+++ b/internal/db/sqltest/tests/purge/deleted_table_tests.sql
@@ -0,0 +1,86 @@
+-- Copyright (c) HashiCorp, Inc.
+-- SPDX-License-Identifier: BUSL-1.1
+
+begin;
+  create function op_table(deletion_table_name name) returns text
+  as $$
+    select split_part(deletion_table_name, '_deleted', 1);
+  $$ language sql;
+
+  create function has_operational_table(deletion_table_name name) returns text
+  as $$
+    select has_table(op_table(deletion_table_name));
+  $$ language sql;
+
+  -- tests that the deletion table has the insert trigger
+  create function has_insert_trigger(deletion_table_name name) returns text
+  as $$
+    select * from collect_tap(
+      has_trigger(op_table(deletion_table_name), 'insert_deleted_id'),
+      trigger_is(op_table(deletion_table_name), 'insert_deleted_id', 'insert_deleted_id')
+    );
+  $$ language sql;
+
+  -- tests the public_id column
+  create function has_public_id(deletion_table_name name) returns text
+  as $$
+    select * from collect_tap(
+      has_column(deletion_table_name, 'public_id'),
+      col_not_null(deletion_table_name, 'public_id'),
+      col_hasnt_default(deletion_table_name, 'public_id')
+    );
+  $$ language sql;
+
+  -- tests the delete_time column
+  create function has_delete_time(deletion_table_name name) returns text
+  as $$
+    select * from collect_tap(
+      has_column(deletion_table_name, 'delete_time'),
+      col_not_null(deletion_table_name, 'delete_time'),
+      col_hasnt_default(deletion_table_name, 'delete_time')
+    );
+  $$ language sql;
+
+  -- tests for delete_time index
+  create function has_delete_time_index(deletion_table_name name) returns text
+  as $$
+    select case when length(deletion_table_name || '_delete_time_idx') > 63
+           then hasnt_index(deletion_table_name, deletion_table_name || '_delete_time_idx', 'Index name too long: ' || deletion_table_name || '_delete_time_idx')
+           else collect_tap(
+                has_index(deletion_table_name, deletion_table_name || '_delete_time_idx', 'delete_time')
+              ) end;
+  $$ language sql;
+
+  -- tests the tables exist and follow the required naming pattern
+  create function has_correct_tables(deletetion_table_name name) returns text
+  as $$
+    select * from collect_tap(
+      has_table(deletetion_table_name),
+      has_operational_table(deletetion_table_name)
+    );
+  $$ language sql;
+
+  -- runs all the tests on a single deletion table
+  create function test_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_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()
+    ) as a;
+
+    select test_deletion_table(a)
+      from get_deletion_tables() a;
+
+  select * from finish();
+rollback;
diff --git a/internal/db/sqltest/tests/purge/trigger_insert_on_delete.sql b/internal/db/sqltest/tests/purge/trigger_insert_on_delete.sql
new file mode 100644
index 000000000..b15d839cf
--- /dev/null
+++ b/internal/db/sqltest/tests/purge/trigger_insert_on_delete.sql
@@ -0,0 +1,41 @@
+-- Copyright (c) HashiCorp, Inc.
+-- SPDX-License-Identifier: BUSL-1.1
+
+begin;
+
+  select plan(13);
+
+  -- Verify the trigger functions exist and are declared properly
+  select has_function('insert_deleted_id');
+  select volatility_is('insert_deleted_id', 'volatile');
+  select isnt_strict('insert_deleted_id');
+  select has_trigger('auth_token', 'insert_deleted_id');
+
+  -- To test the trigger that moves deleted rows into their appropriate tables, we'll use auth_token
+  -- Ensure session state table is populated, and session_deleted and auth_token_deleted tables are empty
+  select is(count(*), 1::bigint) from session_state where session_id = 's1_____clare';
+  select is(count(*), 0::bigint) from session_deleted;
+  select is(count(*), 0::bigint) from auth_token_deleted;
+
+  -- Check that we have a session for a particular auth token
+  select is(count(*), 2::bigint) from session where auth_token_id = 'tok____clare';
+
+  -- Delete auth tokens, expect no errors, check that the auth_token table is now populated
+  delete from auth_token where public_id = 'tok____clare';
+  select is(count(*), 0::bigint) from auth_token where public_id = 'tok____clare';
+  select is(count(*), 1::bigint) from auth_token_deleted;
+
+  -- Add new auth token with duplicate public_id, delete and check that there's no error
+  insert into auth_token
+    (key_id,         auth_account_id, public_id,      token)
+  values
+    ('kdkv__colors', 'apa____clare',  'tok____clare', 'tok____clare'::bytea);
+  prepare delete_again as
+    delete from auth_token where public_id = 'tok____clare';
+  select lives_ok('delete_again');
+  select is(count(*), 0::bigint) from auth_token where public_id = 'tok____clare';
+  select is(count(*), 1::bigint) from auth_token_deleted;
+
+  select * from finish();
+
+rollback;