Move event streaming in to after commit queue

EE: true

app/models/audit_event.rb

 # frozen_string_literal: true
 class AuditEvent < ApplicationRecord
+  include AfterCommitQueue
   include CreatedAtFilterable
   include BulkInsertSafe
   include EachBatch

app/services/audit_event_service.rb

@@ -141,7 +141,7 @@ def save_or_track(event)
     event.save! if should_save_database?(@save_type)
     stream_event_to_external_destinations(event) if should_save_stream?(@save_type)
   rescue StandardError => e
-    Gitlab::ErrorTracking.track_exception(e, audit_event_type: event.class.to_s)
+    Gitlab::ErrorTracking.track_and_raise_for_dev_exception(e, audit_event_type: event.class.to_s)
   end
 end

ee/app/services/ee/audit_event_service.rb

@@ -238,7 +238,9 @@ def respond_to?(method, include_private = false)
     def stream_event_to_external_destinations(event)
       return if event.is_a?(AuthenticationEvent)
-      event.stream_to_external_destinations if event.entity_is_group_or_project?
+      if event.entity_is_group_or_project?
+        event.run_after_commit_or_now { event.stream_to_external_destinations }
+      end
     end
     override :base_payload

ee/spec/services/audit_event_service_spec.rb

@@ -653,6 +653,7 @@ def disable_license_audit_features
     before do
       stub_licensed_features(external_audit_events: true)
+      stub_feature_flags(ff_external_audit_events_namespace: group.root_ancestor)
     end
     subject(:event) { described_class.new(user, project, details, save_type).for_project.security_event }
@@ -660,10 +661,27 @@ def disable_license_audit_features
     context 'with save_type of :database_and_stream' do
       let(:save_type) { :database_and_stream }
-      it 'save to database and stream' do
-        expect(AuditEvents::AuditEventStreamingWorker).to receive(:perform_async).once
+      it 'saves to database' do
         expect { event }.to change(AuditEvent, :count).by(1)
       end
+
+      it 'streams the event' do
+        expect_next_instance_of(described_class) do |service|
+          expect(service).to receive(:stream_event_to_external_destinations).once
+        end
+
+        event
+      end
+
+      context 'when the event is created within a transaction' do
+        it 'does not raise an error about a job being enqueued from within a transaction' do
+          RSpec::Expectations.configuration.on_potential_false_positives = :nothing
+
+          ApplicationRecord.transaction do
+            expect { event }.not_to raise_error(Sidekiq::Worker::EnqueueFromTransactionError)
+          end
+        end
+      end
     end
     context 'with save_type of :database' do
@@ -678,9 +696,16 @@ def disable_license_audit_features
     context 'with save_type of :stream' do
       let(:save_type) { :stream }
-      it 'saves to database and stream' do
-        expect(AuditEvents::AuditEventStreamingWorker).to receive(:perform_async).once
-        expect { event }.to change(AuditEvent, :count).by(0)
+      it 'does not save to database' do
+        expect { event }.not_to change(AuditEvent, :count)
+      end
+
+      it 'streams the event' do
+        expect_next_instance_of(described_class) do |service|
+          expect(service).to receive(:stream_event_to_external_destinations).once
+        end
+
+        event
       end
     end
   end