diff --git a/engine/apps/alerts/models/alert.py b/engine/apps/alerts/models/alert.py
index a3b5573e11..1471d3127d 100644
--- a/engine/apps/alerts/models/alert.py
+++ b/engine/apps/alerts/models/alert.py
@@ -12,6 +12,8 @@
 from apps.alerts import tasks
 from apps.alerts.constants import TASK_DELAY_SECONDS
 from apps.alerts.incident_appearance.templaters import TemplateLoader
+from apps.alerts.signals import alert_group_escalation_snapshot_built
+from apps.alerts.tasks.distribute_alert import send_alert_create_signal
 from apps.labels.alert_group_labels import assign_labels
 from common.jinja_templater import apply_jinja_template
 from common.jinja_templater.apply_jinja_template import JinjaTemplateError, JinjaTemplateWarning
@@ -102,6 +104,7 @@ def create(
         if channel_filter is None:
             channel_filter = ChannelFilter.select_filter(alert_receive_channel, raw_request_data, force_route_id)
 
+        # Get or create group
         group, group_created = AlertGroup.objects.get_or_create_grouping(
             channel=alert_receive_channel,
             channel_filter=channel_filter,
@@ -109,21 +112,7 @@ def create(
             received_at=received_at,
         )
 
-        if group_created:
-            assign_labels(group, alert_receive_channel, raw_request_data)
-            group.log_records.create(type=AlertGroupLogRecord.TYPE_REGISTERED)
-            group.log_records.create(type=AlertGroupLogRecord.TYPE_ROUTE_ASSIGNED)
-
-        mark_as_resolved = (
-            enable_autoresolve and group_data.is_resolve_signal and alert_receive_channel.allow_source_based_resolving
-        )
-        if not group.resolved and mark_as_resolved:
-            group.resolve_by_source()
-
-        mark_as_acknowledged = group_data.is_acknowledge_signal
-        if not group.acknowledged and mark_as_acknowledged:
-            group.acknowledge_by_source()
-
+        # Create alert
         alert = cls(
             is_resolve_signal=group_data.is_resolve_signal,
             title=title,
@@ -135,21 +124,36 @@ def create(
             raw_request_data=raw_request_data,
             is_the_first_alert_in_group=group_created,
         )
-
         alert.save()
+        send_alert_create_signal.apply_async((alert.pk,))
+
+        if group_created:
+            assign_labels(group, alert_receive_channel)
+            group.log_records.create(type=AlertGroupLogRecord.TYPE_REGISTERED)
+            group.log_records.create(type=AlertGroupLogRecord.TYPE_ROUTE_ASSIGNED)
+
+        if group_created or alert.group.pause_escalation:
+            # Build escalation snapshot if needed and start escalation
+            alert.group.start_escalation_if_needed(countdown=TASK_DELAY_SECONDS)
+
+        if group_created:
+            alert_group_escalation_snapshot_built.send(sender=cls.__class__, alert_group=alert.group)
+
+        mark_as_acknowledged = group_data.is_acknowledge_signal
+        if not group.acknowledged and mark_as_acknowledged:
+            group.acknowledge_by_source()
+
+        mark_as_resolved = (
+            enable_autoresolve and group_data.is_resolve_signal and alert_receive_channel.allow_source_based_resolving
+        )
+        if not group.resolved and mark_as_resolved:
+            group.resolve_by_source()
 
         # Store exact alert which resolved group.
         if group.resolved_by == AlertGroup.SOURCE and group.resolved_by_alert is None:
             group.resolved_by_alert = alert
             group.save(update_fields=["resolved_by_alert"])
 
-        if settings.DEBUG:
-            tasks.distribute_alert(alert.pk)
-        else:
-            transaction.on_commit(
-                partial(tasks.distribute_alert.apply_async, (alert.pk,), countdown=TASK_DELAY_SECONDS)
-            )
-
         if group_created:
             # all code below related to maintenance mode
             maintenance_uuid = None
diff --git a/engine/apps/alerts/tasks/distribute_alert.py b/engine/apps/alerts/tasks/distribute_alert.py
index f1d146a450..f69f80fec2 100644
--- a/engine/apps/alerts/tasks/distribute_alert.py
+++ b/engine/apps/alerts/tasks/distribute_alert.py
@@ -13,6 +13,8 @@
 def distribute_alert(alert_id):
     """
     We need this task to make task processing async and to make sure the task is delivered.
+    This task is not used anymore, but we keep it for the tasks in the queue to be processed.
+    TODO: remove this task after all the tasks in the queue are processed.
     """
     from apps.alerts.models import Alert