diff --git a/db/migrations/20250327142351_bigint_migration_events_step1.rb b/db/migrations/20250327142351_bigint_migration_events_step1.rb
index 2913a3ab9ec..8316eafe9d8 100644
--- a/db/migrations/20250327142351_bigint_migration_events_step1.rb
+++ b/db/migrations/20250327142351_bigint_migration_events_step1.rb
@@ -14,7 +14,15 @@
 
   down do
     if database_type == :postgres
+      # There is no guarantee that the table is still empty - which was the condition for simply switching the id
+      # column's type to bigint. We nevertheless want to revert the type to integer as this is the opposite procedure of
+      # the up migration. In case there is a lot of data in the table at this moment in time, this change might be
+      # problematic, e.g. take a longer time.
+      #
+      # Ideally this down migration SHOULD NEVER BE EXECUTED IN A PRODUCTION SYSTEM! (It's there for proper integration
+      # testing of the bigint migration steps.)
       VCAP::BigintMigration.revert_pk_to_integer(self, :events)
+
       VCAP::BigintMigration.drop_trigger_function(self, :events)
       VCAP::BigintMigration.drop_bigint_column(self, :events)
     end
diff --git a/db/migrations/20250603103400_bigint_migration_events_step3a.rb b/db/migrations/20250603103400_bigint_migration_events_step3a.rb
new file mode 100644
index 00000000000..ec4fa863d60
--- /dev/null
+++ b/db/migrations/20250603103400_bigint_migration_events_step3a.rb
@@ -0,0 +1,19 @@
+require 'database/bigint_migration'
+
+Sequel.migration do
+  up do
+    if database_type == :postgres && !VCAP::BigintMigration.migration_completed?(self, :events) && !VCAP::BigintMigration.migration_skipped?(self, :events)
+      begin
+        VCAP::BigintMigration.add_check_constraint(self, :events)
+      rescue Sequel::CheckConstraintViolation
+        raise "Failed to add check constraint on 'events' table!\n" \
+              "There are rows where 'id_bigint' does not match 'id', thus step 3 of the bigint migration cannot be executed.\n" \
+              "Consider running rake task 'db:bigint_backfill[events]'."
+      end
+    end
+  end
+
+  down do
+    VCAP::BigintMigration.drop_check_constraint(self, :events) if database_type == :postgres
+  end
+end
diff --git a/db/migrations/20250603103500_bigint_migration_events_step3b.rb b/db/migrations/20250603103500_bigint_migration_events_step3b.rb
new file mode 100644
index 00000000000..2560256e41f
--- /dev/null
+++ b/db/migrations/20250603103500_bigint_migration_events_step3b.rb
@@ -0,0 +1,48 @@
+require 'database/bigint_migration'
+
+Sequel.migration do
+  up do
+    if database_type == :postgres && VCAP::BigintMigration.has_check_constraint?(self, :events)
+      # Drop check constraint and trigger function
+      VCAP::BigintMigration.drop_check_constraint(self, :events)
+      VCAP::BigintMigration.drop_trigger_function(self, :events)
+
+      # Drop old id column
+      VCAP::BigintMigration.drop_pk_column(self, :events)
+
+      # Switch id_bigint -> id
+      VCAP::BigintMigration.rename_bigint_column(self, :events)
+      VCAP::BigintMigration.add_pk_constraint(self, :events)
+      VCAP::BigintMigration.add_timestamp_pk_index(self, :events)
+      VCAP::BigintMigration.set_pk_as_identity_with_correct_start_value(self, :events)
+    end
+  end
+
+  down do
+    if database_type == :postgres && VCAP::BigintMigration.migration_completed?(self, :events)
+      # Revert id -> id_bigint
+      VCAP::BigintMigration.drop_identity(self, :events)
+      VCAP::BigintMigration.drop_timestamp_pk_index(self, :events)
+      VCAP::BigintMigration.drop_pk_constraint(self, :events)
+      VCAP::BigintMigration.revert_bigint_column_name(self, :events)
+
+      # Restore old id column
+      VCAP::BigintMigration.add_id_column(self, :events)
+
+      # To restore the previous state it is necessary to backfill the id column. In case there is a lot of data in the
+      # table this might be problematic, e.g. take a longer time.
+      #
+      # Ideally this down migration SHOULD NEVER BE EXECUTED IN A PRODUCTION SYSTEM! (It's there for proper integration
+      # testing of the bigint migration steps.)
+      VCAP::BigintMigration.backfill_id(self, :events)
+
+      VCAP::BigintMigration.add_pk_constraint(self, :events)
+      VCAP::BigintMigration.add_timestamp_pk_index(self, :events)
+      VCAP::BigintMigration.set_pk_as_identity_with_correct_start_value(self, :events)
+
+      # Recreate trigger function and check constraint
+      VCAP::BigintMigration.create_trigger_function(self, :events)
+      VCAP::BigintMigration.add_check_constraint(self, :events)
+    end
+  end
+end
diff --git a/lib/database/bigint_migration.rb b/lib/database/bigint_migration.rb
index c71471193d7..191b3399822 100644
--- a/lib/database/bigint_migration.rb
+++ b/lib/database/bigint_migration.rb
@@ -50,11 +50,7 @@ def backfill(logger, db, table, batch_size: 10_000, iterations: -1)
 
       logger.info("starting bigint backfill on table '#{table}' (batch_size: #{batch_size}, iterations: #{iterations})")
       loop do
-        updated_rows = db.
-                       from(table, :batch).
-                       with(:batch, db[table].select(:id).where(id_bigint: nil).order(:id).limit(batch_size).for_update.skip_locked).
-                       where(Sequel.qualify(table, :id) => :batch__id).
-                       update(id_bigint: :batch__id)
+        updated_rows = backfill_batch(db, table, :id, :id_bigint, batch_size)
         logger.info("updated #{updated_rows} rows")
         iterations -= 1 if iterations > 0
         break if updated_rows < batch_size || iterations == 0
@@ -62,10 +58,116 @@ def backfill(logger, db, table, batch_size: 10_000, iterations: -1)
       logger.info("finished bigint backfill on table '#{table}'")
     end
 
+    def migration_completed?(db, table)
+      column_type(db, table, :id) == 'bigint'
+    end
+
+    def migration_skipped?(db, table)
+      !column_exists?(db, table, :id_bigint)
+    end
+
+    def add_check_constraint(db, table)
+      return if has_check_constraint?(db, table)
+
+      constraint_name = check_constraint_name(table)
+      db.alter_table(table) do
+        add_constraint(constraint_name) do
+          Sequel.lit('id_bigint IS NOT NULL AND id_bigint = id')
+        end
+      end
+    end
+
+    def drop_check_constraint(db, table)
+      return unless has_check_constraint?(db, table)
+
+      constraint_name = check_constraint_name(table)
+      db.alter_table(table) do
+        drop_constraint(constraint_name)
+      end
+    end
+
+    def has_check_constraint?(db, table)
+      check_constraint_exists?(db, table, check_constraint_name(table))
+    end
+
+    def drop_pk_column(db, table)
+      db.drop_column(table, :id, if_exists: true)
+    end
+
+    def add_id_column(db, table)
+      db.add_column(table, :id, :integer, if_not_exists: true)
+    end
+
+    def rename_bigint_column(db, table)
+      db.rename_column(table, :id_bigint, :id) if column_exists?(db, table, :id_bigint) && !column_exists?(db, table, :id)
+    end
+
+    def revert_bigint_column_name(db, table)
+      db.rename_column(table, :id, :id_bigint) if column_exists?(db, table, :id) && column_type(db, table, :id) == 'bigint' && !column_exists?(db, table, :id_bigint)
+    end
+
+    def add_pk_constraint(db, table)
+      return if db.primary_key(table) == 'id'
+
+      db.alter_table(table) do
+        add_primary_key([:id])
+      end
+    end
+
+    def drop_pk_constraint(db, table)
+      return unless db.primary_key(table) == 'id'
+
+      constraint_name = pk_constraint_name(table)
+      db.alter_table(table) do
+        drop_constraint(constraint_name)
+        set_column_allow_null(:id, true)
+      end
+    end
+
+    def add_timestamp_pk_index(db, table)
+      db.add_index(table, %i[timestamp id], name: timestamp_id_index_name(table), unique: false, if_not_exists: true)
+    end
+
+    def drop_timestamp_pk_index(db, table)
+      db.drop_index(table, %i[timestamp id], name: timestamp_id_index_name(table), if_exists: true)
+    end
+
+    def set_pk_as_identity_with_correct_start_value(db, table)
+      return if column_attribute(db, table, :id, :auto_increment) == true
+
+      block = <<~BLOCK
+        DO $$
+        DECLARE
+          max_id BIGINT;
+        BEGIN
+          SELECT COALESCE(MAX(id), 0) + 1 INTO max_id FROM #{table};
+
+          EXECUTE format('ALTER TABLE #{table} ALTER COLUMN id ADD GENERATED BY DEFAULT AS IDENTITY (START WITH %s)', max_id);
+        END $$;
+      BLOCK
+      db.run(block)
+    end
+
+    def drop_identity(db, table)
+      db.run("ALTER TABLE #{table} ALTER COLUMN id DROP IDENTITY IF EXISTS")
+    end
+
+    def backfill_id(db, table)
+      batch_size = 10_000
+      loop do
+        updated_rows = backfill_batch(db, table, :id_bigint, :id, batch_size)
+        break if updated_rows < batch_size
+      end
+    end
+
     private
 
+    def column_attribute(db, table, column, attribute)
+      db.schema(table).find { |col, _| col == column }&.dig(1, attribute)
+    end
+
     def column_type(db, table, column)
-      db.schema(table).find { |col, _| col == column }&.dig(1, :db_type)
+      column_attribute(db, table, column, :db_type)
     end
 
     def function_name(table)
@@ -79,5 +181,29 @@ def trigger_name(table)
     def column_exists?(db, table, column)
       db[table].columns.include?(column)
     end
+
+    def check_constraint_name(table)
+      :"#{table}_check_id_bigint_matches_id"
+    end
+
+    def check_constraint_exists?(db, table, constraint_name)
+      db.check_constraints(table).include?(constraint_name)
+    end
+
+    def pk_constraint_name(table)
+      :"#{table}_pkey"
+    end
+
+    def timestamp_id_index_name(table)
+      :"#{table}_timestamp_id_index"
+    end
+
+    def backfill_batch(db, table, from_column, to_column, batch_size)
+      db.
+        from(table, :batch).
+        with(:batch, db[table].select(from_column).where(to_column => nil).order(from_column).limit(batch_size).for_update.skip_locked).
+        where(Sequel.qualify(table, from_column) => :"batch__#{from_column}").
+        update(to_column => :"batch__#{from_column}")
+    end
   end
 end
diff --git a/spec/migrations/20250603103400_bigint_migration_events_step3a_spec.rb b/spec/migrations/20250603103400_bigint_migration_events_step3a_spec.rb
new file mode 100644
index 00000000000..9d2920a0ee2
--- /dev/null
+++ b/spec/migrations/20250603103400_bigint_migration_events_step3a_spec.rb
@@ -0,0 +1,17 @@
+require 'spec_helper'
+require 'migrations/helpers/bigint_migration_step3_shared_context'
+
+RSpec.describe 'bigint migration - events table - step3a', isolation: :truncation, type: :migration do
+  include_context 'bigint migration step3a' do
+    let(:migration_filename_step1) { '20250327142351_bigint_migration_events_step1.rb' }
+    let(:migration_filename_step3a) { '20250603103400_bigint_migration_events_step3a.rb' }
+    let(:table) { :events }
+    let(:insert) do
+      lambda do |db|
+        db[:events].insert(guid: SecureRandom.uuid, timestamp: Time.now.utc, type: 'type',
+                           actor: 'actor', actor_type: 'actor_type',
+                           actee: 'actee', actee_type: 'actee_type')
+      end
+    end
+  end
+end
diff --git a/spec/migrations/20250603103500_bigint_migration_events_step3b_spec.rb b/spec/migrations/20250603103500_bigint_migration_events_step3b_spec.rb
new file mode 100644
index 00000000000..078d7f58f76
--- /dev/null
+++ b/spec/migrations/20250603103500_bigint_migration_events_step3b_spec.rb
@@ -0,0 +1,18 @@
+require 'spec_helper'
+require 'migrations/helpers/bigint_migration_step3_shared_context'
+
+RSpec.describe 'bigint migration - events table - step3b', isolation: :truncation, type: :migration do
+  include_context 'bigint migration step3b' do
+    let(:migration_filename_step1) { '20250327142351_bigint_migration_events_step1.rb' }
+    let(:migration_filename_step3a) { '20250603103400_bigint_migration_events_step3a.rb' }
+    let(:migration_filename_step3b) { '20250603103500_bigint_migration_events_step3b.rb' }
+    let(:table) { :events }
+    let(:insert) do
+      lambda do |db|
+        db[:events].insert(guid: SecureRandom.uuid, timestamp: Time.now.utc, type: 'type',
+                           actor: 'actor', actor_type: 'actor_type',
+                           actee: 'actee', actee_type: 'actee_type')
+      end
+    end
+  end
+end
diff --git a/spec/migrations/helpers/bigint_migration_step1_shared_context.rb b/spec/migrations/helpers/bigint_migration_step1_shared_context.rb
index bb15bc12a0b..49a4ff9133f 100644
--- a/spec/migrations/helpers/bigint_migration_step1_shared_context.rb
+++ b/spec/migrations/helpers/bigint_migration_step1_shared_context.rb
@@ -60,6 +60,10 @@
       context 'when the table is not empty' do
         let!(:old_id) { insert.call(db) }
 
+        after do
+          db[table].delete # Necessary to successfully run subsequent migrations in the after block of the migration shared context...
+        end
+
         it "does not change the id column's type" do
           expect(db).to have_table_with_column_and_type(table, :id, 'integer')
 
@@ -186,6 +190,10 @@
         run_migration
       end
 
+      after do
+        db[table].delete # Necessary to successfully run subsequent migrations in the after block of the migration shared context...
+      end
+
       it 'drops the id_bigint column' do
         expect(db).to have_table_with_column(table, :id_bigint)
 
diff --git a/spec/migrations/helpers/bigint_migration_step3_shared_context.rb b/spec/migrations/helpers/bigint_migration_step3_shared_context.rb
new file mode 100644
index 00000000000..8ecc2f4ed19
--- /dev/null
+++ b/spec/migrations/helpers/bigint_migration_step3_shared_context.rb
@@ -0,0 +1,263 @@
+require 'migrations/helpers/migration_shared_context'
+require 'database/bigint_migration'
+
+RSpec.shared_context 'bigint migration step3a' do
+  subject(:run_migration_step1) { Sequel::Migrator.run(db, migrations_path, target: current_migration_index, allow_missing_migration_files: true) }
+
+  subject(:run_migration_step3a) { Sequel::Migrator.run(db, migrations_path, target: current_migration_index_step3a, allow_missing_migration_files: true) }
+
+  let(:migration_filename) { migration_filename_step1 }
+  let(:current_migration_index_step3a) { migration_filename_step3a.match(/\A\d+/)[0].to_i }
+
+  include_context 'migration'
+
+  let(:skip_bigint_id_migration) { false }
+  let(:logger) { double(:logger, info: nil) }
+
+  before do
+    skip unless db.database_type == :postgres
+
+    allow_any_instance_of(VCAP::CloudController::Config).to receive(:get).with(:skip_bigint_id_migration).and_return(skip_bigint_id_migration)
+  end
+
+  describe 'up' do
+    context 'when migration step 1 was executed' do
+      context 'when the id_bigint column was added' do
+        before do
+          insert.call(db)
+          run_migration_step1
+        end
+
+        context 'when backfilling was completed' do
+          before do
+            VCAP::BigintMigration.backfill(logger, db, table)
+          end
+
+          it 'adds a check constraint' do
+            expect(db).not_to have_table_with_check_constraint(table)
+
+            run_migration_step3a
+
+            expect(db).to have_table_with_check_constraint(table)
+          end
+        end
+
+        context 'when backfilling was not completed' do
+          after do
+            db[table].delete # Necessary as the migration will be executed again in the after block of the migration shared context - and should not fail...
+          end
+
+          it 'fails ...' do
+            expect do
+              run_migration_step3a
+            end.to raise_error(/Failed to add check constraint on 'events' table!/)
+          end
+        end
+      end
+
+      context "when the migration was concluded (id column's type switched)" do
+        before do
+          db[table].delete
+          run_migration_step1
+        end
+
+        it 'does not add a check constraint' do
+          expect(db).not_to have_table_with_check_constraint(table)
+
+          run_migration_step3a
+
+          expect(db).not_to have_table_with_check_constraint(table)
+        end
+      end
+    end
+
+    context 'when migration step 1 was skipped' do
+      let(:skip_bigint_id_migration) { true }
+
+      before do
+        run_migration_step1
+      end
+
+      it 'does not add a check constraint' do
+        expect(db).not_to have_table_with_check_constraint(table)
+
+        run_migration_step3a
+
+        expect(db).not_to have_table_with_check_constraint(table)
+      end
+    end
+  end
+
+  describe 'down' do
+    subject(:run_rollback_step3a) { Sequel::Migrator.run(db, migrations_path, target: current_migration_index_step3a - 1, allow_missing_migration_files: true) }
+
+    context 'when migration step 3a was executed' do
+      before do
+        insert.call(db)
+        run_migration_step1
+        VCAP::BigintMigration.backfill(logger, db, table)
+        run_migration_step3a
+      end
+
+      it 'drops the check constraint' do
+        expect(db).to have_table_with_check_constraint(table)
+
+        run_rollback_step3a
+
+        expect(db).not_to have_table_with_check_constraint(table)
+      end
+    end
+  end
+end
+
+RSpec.shared_context 'bigint migration step3b' do
+  subject(:run_migration_step1) { Sequel::Migrator.run(db, migrations_path, target: current_migration_index, allow_missing_migration_files: true) }
+
+  subject(:run_migration_step3a) { Sequel::Migrator.run(db, migrations_path, target: current_migration_index_step3a, allow_missing_migration_files: true) }
+
+  subject(:run_migration_step3b) { Sequel::Migrator.run(db, migrations_path, target: current_migration_index_step3b, allow_missing_migration_files: true) }
+
+  let(:migration_filename) { migration_filename_step1 }
+  let(:current_migration_index_step3a) { migration_filename_step3a.match(/\A\d+/)[0].to_i }
+  let(:current_migration_index_step3b) { migration_filename_step3b.match(/\A\d+/)[0].to_i }
+
+  include_context 'migration'
+
+  let(:skip_bigint_id_migration) { false }
+  let(:logger) { double(:logger, info: nil) }
+
+  before do
+    skip unless db.database_type == :postgres
+
+    allow_any_instance_of(VCAP::CloudController::Config).to receive(:get).with(:skip_bigint_id_migration).and_return(skip_bigint_id_migration)
+  end
+
+  describe 'up' do
+    context 'when migration step 3a was executed' do
+      before do
+        insert.call(db)
+        run_migration_step1
+        VCAP::BigintMigration.backfill(logger, db, table)
+        run_migration_step3a
+      end
+
+      it 'drops the check constraint' do
+        expect(db).to have_table_with_check_constraint(table)
+
+        run_migration_step3b
+
+        expect(db).not_to have_table_with_check_constraint(table)
+      end
+
+      it 'drops the trigger function' do
+        expect(db).to have_trigger_function_for_table(table)
+
+        run_migration_step3b
+
+        expect(db).not_to have_trigger_function_for_table(table)
+      end
+
+      it 'drops the id column and renames the id_bigint column to id' do
+        expect(db).to have_table_with_column_and_type(table, :id, 'integer')
+        expect(db).to have_table_with_column_and_type(table, :id_bigint, 'bigint')
+
+        run_migration_step3b
+
+        expect(db).to have_table_with_column_and_type(table, :id, 'bigint')
+        expect(db).not_to have_table_with_column(table, :id_bigint)
+      end
+
+      it 'uses the (bigint) id column as primary key' do
+        expect(db).to have_table_with_primary_key(table, :id)
+
+        run_migration_step3b
+
+        expect(db).to have_table_with_primary_key(table, :id)
+      end
+
+      it 'has an index on timestamp + (bigint) id column' do
+        expect(db).to have_table_with_index_on_columns(table, %i[timestamp id])
+
+        run_migration_step3b
+
+        expect(db).to have_table_with_index_on_columns(table, %i[timestamp id])
+      end
+
+      it 'uses an identity with correct start value for the (bigint) id column' do
+        last_id_before_migration = insert.call(db)
+
+        run_migration_step3b
+
+        first_id_after_migration = insert.call(db)
+        expect(first_id_after_migration).to eq(last_id_before_migration + 1)
+      end
+    end
+  end
+
+  describe 'down' do
+    subject(:run_rollback_step3b) { Sequel::Migrator.run(db, migrations_path, target: current_migration_index_step3b - 1, allow_missing_migration_files: true) }
+
+    context 'when migration step 3b was executed' do
+      before do
+        insert.call(db)
+        run_migration_step1
+        VCAP::BigintMigration.backfill(logger, db, table)
+        run_migration_step3a
+        run_migration_step3b
+      end
+
+      it 'uses an identity with correct start value for the (integer) id column' do
+        last_id_before_migration = insert.call(db)
+
+        run_rollback_step3b
+
+        first_id_after_migration = insert.call(db)
+        expect(first_id_after_migration).to eq(last_id_before_migration + 1)
+      end
+
+      it 'has an index on timestamp + (integer) id column' do
+        expect(db).to have_table_with_index_on_columns(table, %i[timestamp id])
+
+        run_rollback_step3b
+
+        expect(db).to have_table_with_index_on_columns(table, %i[timestamp id])
+        expect(db).not_to have_table_with_index_on_columns(table, %i[timestamp id_bigint])
+      end
+
+      it 'uses the (integer) id column as primary key' do
+        expect(db).to have_table_with_primary_key(table, :id)
+
+        run_rollback_step3b
+
+        expect(db).to have_table_with_primary_key(table, :id)
+      end
+
+      it 'renames the id column to id_bigint and re-adds the (integer) id column' do
+        expect(db).to have_table_with_column_and_type(table, :id, 'bigint')
+        expect(db).not_to have_table_with_column(table, :id_bigint)
+
+        run_rollback_step3b
+
+        expect(db).to have_table_with_column_and_type(table, :id, 'integer')
+        expect(db).to have_table_with_column_and_type(table, :id_bigint, 'bigint')
+        expect(db).to have_table_with_column_and_attribute(table, :id_bigint, :allow_null, true)
+      end
+
+      it 're-creates the trigger function' do
+        expect(db).not_to have_trigger_function_for_table(table)
+
+        run_rollback_step3b
+
+        expect(db).to have_trigger_function_for_table(table)
+      end
+
+      it 're-adds the check constraint (this also ensures that id was correctly backfilled)' do
+        expect(db).not_to have_table_with_check_constraint(table)
+
+        run_rollback_step3b
+
+        expect(db).to have_table_with_check_constraint(table)
+      end
+    end
+  end
+end
diff --git a/spec/migrations/helpers/matchers.rb b/spec/migrations/helpers/matchers.rb
index 5dea2dc569c..1fd6e191dfd 100644
--- a/spec/migrations/helpers/matchers.rb
+++ b/spec/migrations/helpers/matchers.rb
@@ -3,7 +3,7 @@
 
   match do |passed_options|
     options = passed_options.keys
-    !options.delete(:name).nil? && (options - %i[where if_not_exists concurrently]).empty?
+    !options.delete(:name).nil? && (options - %i[where if_not_exists concurrently unique]).empty?
   end
 end
 
@@ -56,3 +56,31 @@
     db[table].where(column => nil).any?
   end
 end
+
+RSpec::Matchers.define :have_table_with_check_constraint do |table|
+  match do |db|
+    constraint_name = :"#{table}_check_id_bigint_matches_id"
+
+    db.check_constraints(table).include?(constraint_name)
+  end
+end
+
+RSpec::Matchers.define :have_table_with_primary_key do |table, column|
+  match do |db|
+    db.primary_key(table).to_sym == column
+  end
+end
+
+RSpec::Matchers.define :have_table_with_column_and_attribute do |table, column, attribute, value|
+  match do |db|
+    expect(db).to have_table_with_column(table, column)
+
+    db.schema(table).find { |col, _| col == column }&.dig(1, attribute) == value
+  end
+end
+
+RSpec::Matchers.define :have_table_with_index_on_columns do |table, columns|
+  match do |db|
+    db.indexes(table).any? { |_, index| index[:columns] == columns }
+  end
+end