Repository: saleor/saleor
Pull request: Async payload generation
This PR adds the ability to generate subscription payload for webhook in the async task. Currently enabled for the following async events:
- `checkout_created`
- `checkout_updated`

# Impact

- [ ] New migrations
- [ ] New/Updated API fields or mutations
- [ ] Deprecated API fields or mutations
- [ ] Removed API types, fields, or mutations

# Docs

<!-- Docs are stored in a separate repository: https://github.com/saleor/saleor-docs/. -->
<!-- Please provide a link to the PR that updates documentation for your changes. -->
<!-- If changes in docs are not required, please mention that in the description. -->

- [ ] Link to documentation:

# Pull Request Checklist

<!-- Please keep this section. It will make the maintainer's life easier. -->

- [ ] Privileged queries and mutations are either absent or guarded by proper permission checks
- [ ] Database queries are optimized and the number of queries is constant
- [ ] Database migrations are either absent or optimized for zero downtime
- [ ] The changes are covered by test cases
- [ ] All new fields/inputs/mutations have proper labels added (`ADDED_IN_X`, `PREVIEW_FEATURE`, etc.)
- [ ] All migrations have proper dependencies
- [ ] All indexes are added concurrently in migrations
- [ ] All RunSql and RunPython migrations have revert option defined


Problem statement:
Bug: Async events triggered by the mutation have their payloads calculated synchronously
### What are you trying to achieve?

reasonable cloud performance on the premium supported cloud subscription

### Steps to reproduce the problem

add to cart takes more tan 10 seconds

### What did you expect to happen?

add to cart takes less than 1 second. 

### Logs

@JannikZed can you help with some logs?

### Environment

Saleor version: cloud?
OS and version: …

Hints from issue discussion:
Thank you for the report, @dnplkndll. We're actively addressing the issue. The complexity stems from external synchronous webhooks, but our team is working on both resolution and mitigation.
We've identified that this is caused by events triggered by the mutation having their payloads calculated synchronously by the mutation code instead of delegating that work to a background task.

In this specific case, a fairly benign mutation triggers multiple `CHECKOUT_UPDATED` events, one of which causes taxes to be recalculated (an external HTTP call), which has a chance of causing the shipping price to be revalidated (another external HTTP call). Each of those inefficiencies seems innocent until they are all combined, and the mutation is blocked by numerous synchronous HTTP calls, which—added up—explain the observed delay.

We agree it's a significant problem, and we're addressing this by moving checkout-related asynchronous event triggers to background tasks.

Commit message: Support deferred payloads in webhookTrigger mutation

Patch to review:
diff --git a/saleor/core/utils/events.py b/saleor/core/utils/events.py
index ff49002c296..8132bcbd168 100644
--- a/saleor/core/utils/events.py
+++ b/saleor/core/utils/events.py
@@ -8,6 +8,19 @@
 from ...webhook.models import Webhook
 
 
+def get_is_deferred_payload(event_name: str) -> bool:
+    """Return True if the event has deferred payload.
+
+    When the event has deferred payload, the payload will be generated in the Celery
+    task during webhook delivery. In such case, any additional sync calls needed to
+    generated the payload are also run in this task, and we don't need to call them
+    manually before.
+    """
+    return WebhookEventAsyncType.EVENT_MAP.get(event_name, {}).get(
+        "is_deferred_payload", False
+    )
+
+
 def any_webhook_has_subscription(
     events: list[str], webhook_event_map: dict[str, set["Webhook"]]
 ) -> bool:
@@ -68,9 +81,12 @@ def webhook_async_event_requires_sync_webhooks_to_trigger(
     means that sync webhook should be triggered first, before calling async webhook.
     """
     _validate_event_name(event_name, webhook_event_map)
-
     _validate_webhook_event_map(webhook_event_map, possible_sync_events)
 
+    is_deferred_payload = get_is_deferred_payload(event_name)
+    if is_deferred_payload:
+        return False
+
     if not webhook_event_map[event_name] and not webhook_event_map.get(
         WebhookEventAsyncType.ANY
     ):
diff --git a/saleor/graphql/checkout/tests/mutations/test_checkout_create.py b/saleor/graphql/checkout/tests/mutations/test_checkout_create.py
index e13b897d447..a4834951872 100644
--- a/saleor/graphql/checkout/tests/mutations/test_checkout_create.py
+++ b/saleor/graphql/checkout/tests/mutations/test_checkout_create.py
@@ -2753,24 +2753,23 @@ def test_checkout_create_triggers_webhooks(
         webhook_id=checkout_created_webhook.id
     ).exists()
 
-    shipping_methods_call, filter_shipping_call, tax_delivery_call = (
-        mocked_send_webhook_request_sync.mock_calls
-    )
-    shipping_methods_delivery = shipping_methods_call.args[0]
+    sync_deliveries = {
+        call.args[0].event_type: call.args[0]
+        for call in mocked_send_webhook_request_sync.mock_calls
+    }
+
+    assert WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT in sync_deliveries
+    shipping_methods_delivery = sync_deliveries[
+        WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
+    ]
     assert shipping_methods_delivery.webhook_id == shipping_webhook.id
-    assert (
-        shipping_methods_delivery.event_type
-        == WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
-    )
-    assert shipping_methods_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    filter_shipping_delivery = filter_shipping_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS in sync_deliveries
+    filter_shipping_delivery = sync_deliveries[
+        WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
+    ]
     assert filter_shipping_delivery.webhook_id == shipping_filter_webhook.id
-    assert (
-        filter_shipping_delivery.event_type
-        == WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
-    )
-    assert filter_shipping_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    tax_delivery = tax_delivery_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES in sync_deliveries
+    tax_delivery = sync_deliveries[WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES]
     assert tax_delivery.webhook_id == tax_webhook.id
diff --git a/saleor/graphql/checkout/tests/mutations/test_checkout_delivery_method_update.py b/saleor/graphql/checkout/tests/mutations/test_checkout_delivery_method_update.py
index 4b31434c77d..aeb26d2c732 100644
--- a/saleor/graphql/checkout/tests/mutations/test_checkout_delivery_method_update.py
+++ b/saleor/graphql/checkout/tests/mutations/test_checkout_delivery_method_update.py
@@ -21,47 +21,45 @@
 from ....tests.utils import get_graphql_content
 
 MUTATION_UPDATE_DELIVERY_METHOD = """
-    mutation checkoutDeliveryMethodUpdate(
-            $id: ID, $deliveryMethodId: ID) {
-        checkoutDeliveryMethodUpdate(
-            id: $id,
-            deliveryMethodId: $deliveryMethodId) {
-            checkout {
+    mutation checkoutDeliveryMethodUpdate($id: ID, $deliveryMethodId: ID) {
+      checkoutDeliveryMethodUpdate(id: $id, deliveryMethodId: $deliveryMethodId) {
+        checkout {
+          token
+          id
+          shippingAddress {
             id
-            shippingAddress {
-                id
-                firstName
+            firstName
+          }
+          deliveryMethod {
+            __typename
+            ... on ShippingMethod {
+              name
+              id
+              translation(languageCode: EN_US) {
+                name
+              }
             }
-            deliveryMethod {
-                __typename
-                ... on ShippingMethod {
-                    name
-                    id
-                    translation(languageCode: EN_US) {
-                        name
-                    }
-                }
-                ... on Warehouse {
-                   name
-                   id
-                }
+            ... on Warehouse {
+              name
+              id
             }
-            totalPrice {
-                gross {
-                    amount
-                }
-                net {
-                    amount
-                }
+          }
+          totalPrice {
+            gross {
+              amount
             }
+            net {
+              amount
+            }
+          }
         }
         errors {
-            field
-            message
-            code
+          field
+          message
+          code
         }
+      }
     }
-}
 """
 
 
@@ -1105,31 +1103,30 @@ def test_checkout_delivery_method_update_triggers_webhooks(
     )
 
     # confirm each sync webhook was called without saving event delivery
-    assert mocked_send_webhook_request_sync.call_count == 3
+    assert mocked_send_webhook_request_sync.call_count == 4
     assert not EventDelivery.objects.exclude(
         webhook_id=checkout_updated_webhook.id
     ).exists()
 
-    shipping_methods_call, filter_shipping_call, tax_delivery_call = (
-        mocked_send_webhook_request_sync.mock_calls
-    )
-    shipping_methods_delivery = shipping_methods_call.args[0]
+    sync_deliveries = {
+        call.args[0].event_type: call.args[0]
+        for call in mocked_send_webhook_request_sync.mock_calls
+    }
+
+    assert WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT in sync_deliveries
+    shipping_methods_delivery = sync_deliveries[
+        WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
+    ]
     assert shipping_methods_delivery.webhook_id == shipping_webhook.id
-    assert (
-        shipping_methods_delivery.event_type
-        == WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
-    )
-    assert shipping_methods_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    filter_shipping_delivery = filter_shipping_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS in sync_deliveries
+    filter_shipping_delivery = sync_deliveries[
+        WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
+    ]
     assert filter_shipping_delivery.webhook_id == shipping_filter_webhook.id
-    assert (
-        filter_shipping_delivery.event_type
-        == WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
-    )
-    assert filter_shipping_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    tax_delivery = tax_delivery_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES in sync_deliveries
+    tax_delivery = sync_deliveries[WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES]
     assert tax_delivery.webhook_id == tax_webhook.id
 
 
@@ -1184,7 +1181,6 @@ def test_checkout_delivery_method_update_cc_triggers_webhooks(
     # then
     content = get_graphql_content(response)
     assert not content["data"]["checkoutDeliveryMethodUpdate"]["errors"]
-
     assert wrapped_call_checkout_info_event.called
 
     # confirm that event delivery was generated for each async webhook.
@@ -1199,32 +1195,35 @@ def test_checkout_delivery_method_update_cc_triggers_webhooks(
         retry_kwargs={"max_retries": 5},
     )
 
+    # Shipping sync webhooks are called twice - first call before saving the changes in
+    # the checkout instance, second when generating the payload. In the second call
+    # checkout instance has different data and cache cannot be used.
+    assert mocked_send_webhook_request_sync.call_count == 5
+
     # confirm each sync webhook was called without saving event delivery
-    assert mocked_send_webhook_request_sync.call_count == 3
     assert not EventDelivery.objects.exclude(
         webhook_id=checkout_updated_webhook.id
     ).exists()
 
-    shipping_methods_call, filter_shipping_call, tax_delivery_call = (
-        mocked_send_webhook_request_sync.mock_calls
-    )
-    shipping_methods_delivery = shipping_methods_call.args[0]
+    sync_deliveries = {
+        call.args[0].event_type: call.args[0]
+        for call in mocked_send_webhook_request_sync.mock_calls
+    }
+
+    assert WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT in sync_deliveries
+    shipping_methods_delivery = sync_deliveries[
+        WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
+    ]
     assert shipping_methods_delivery.webhook_id == shipping_webhook.id
-    assert (
-        shipping_methods_delivery.event_type
-        == WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
-    )
-    assert shipping_methods_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    filter_shipping_delivery = filter_shipping_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS in sync_deliveries
+    filter_shipping_delivery = sync_deliveries[
+        WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
+    ]
     assert filter_shipping_delivery.webhook_id == shipping_filter_webhook.id
-    assert (
-        filter_shipping_delivery.event_type
-        == WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
-    )
-    assert filter_shipping_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    tax_delivery = tax_delivery_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES in sync_deliveries
+    tax_delivery = sync_deliveries[WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES]
     assert tax_delivery.webhook_id == tax_webhook.id
 
 
diff --git a/saleor/graphql/checkout/tests/mutations/test_checkout_lines_add.py b/saleor/graphql/checkout/tests/mutations/test_checkout_lines_add.py
index 6ccba9a2a09..3728149a3ca 100644
--- a/saleor/graphql/checkout/tests/mutations/test_checkout_lines_add.py
+++ b/saleor/graphql/checkout/tests/mutations/test_checkout_lines_add.py
@@ -29,6 +29,8 @@
 from .....warehouse.models import Reservation, Stock
 from .....warehouse.tests.utils import get_available_quantity_for_stock
 from .....webhook.event_types import WebhookEventAsyncType, WebhookEventSyncType
+from .....webhook.transport.asynchronous.transport import send_webhook_request_async
+from .....webhook.transport.utils import WebhookResponse
 from ....core.utils import to_global_id_or_none
 from ....tests.utils import assert_no_permission, get_graphql_content
 from ...mutations.utils import update_checkout_shipping_method_if_invalid
@@ -1849,10 +1851,15 @@ def test_with_active_problems_flow(
 )
 @patch("saleor.webhook.transport.synchronous.transport.send_webhook_request_sync")
 @patch(
-    "saleor.webhook.transport.asynchronous.transport.send_webhook_request_async.apply_async"
+    "saleor.webhook.transport.asynchronous.transport.send_webhook_request_async.apply_async",
+    wraps=send_webhook_request_async.apply_async,
+)
+@patch(
+    "saleor.webhook.transport.asynchronous.transport.send_webhook_using_scheme_method"
 )
 @override_settings(PLUGINS=["saleor.plugins.webhook.plugin.WebhookPlugin"])
 def test_checkout_lines_add_triggers_webhooks(
+    mocked_send_webhook_using_scheme_method,
     mocked_send_webhook_request_async,
     mocked_send_webhook_request_sync,
     wrapped_call_checkout_info_event,
@@ -1863,6 +1870,7 @@ def test_checkout_lines_add_triggers_webhooks(
     stock,
 ):
     # given
+    mocked_send_webhook_using_scheme_method.return_value = WebhookResponse(content="")
     mocked_send_webhook_request_sync.return_value = []
     (
         tax_webhook,
@@ -1897,18 +1905,7 @@ def test_checkout_lines_add_triggers_webhooks(
     content = get_graphql_content(response)
     assert not content["data"]["checkoutLinesAdd"]["errors"]
     assert wrapped_call_checkout_info_event.called
-
-    # confirm that event delivery was generated for each async webhook.
-    checkout_update_delivery = EventDelivery.objects.get(
-        webhook_id=checkout_updated_webhook.id
-    )
-    mocked_send_webhook_request_async.assert_called_once_with(
-        kwargs={"event_delivery_id": checkout_update_delivery.id},
-        queue=settings.CHECKOUT_WEBHOOK_EVENTS_CELERY_QUEUE_NAME,
-        bind=True,
-        retry_backoff=10,
-        retry_kwargs={"max_retries": 5},
-    )
+    assert mocked_send_webhook_request_async.call_count == 1
 
     # confirm each sync webhook was called without saving event delivery
     assert mocked_send_webhook_request_sync.call_count == 3
@@ -1916,24 +1913,23 @@ def test_checkout_lines_add_triggers_webhooks(
         webhook_id=checkout_updated_webhook.id
     ).exists()
 
-    shipping_methods_call, filter_shipping_call, tax_delivery_call = (
-        mocked_send_webhook_request_sync.mock_calls
-    )
-    shipping_methods_delivery = shipping_methods_call.args[0]
+    sync_deliveries = {
+        call.args[0].event_type: call.args[0]
+        for call in mocked_send_webhook_request_sync.mock_calls
+    }
+
+    assert WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT in sync_deliveries
+    shipping_methods_delivery = sync_deliveries[
+        WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
+    ]
     assert shipping_methods_delivery.webhook_id == shipping_webhook.id
-    assert (
-        shipping_methods_delivery.event_type
-        == WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
-    )
-    assert shipping_methods_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    filter_shipping_delivery = filter_shipping_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS in sync_deliveries
+    filter_shipping_delivery = sync_deliveries[
+        WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
+    ]
     assert filter_shipping_delivery.webhook_id == shipping_filter_webhook.id
-    assert (
-        filter_shipping_delivery.event_type
-        == WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
-    )
-    assert filter_shipping_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    tax_delivery = tax_delivery_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES in sync_deliveries
+    tax_delivery = sync_deliveries[WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES]
     assert tax_delivery.webhook_id == tax_webhook.id
diff --git a/saleor/graphql/checkout/tests/mutations/test_checkout_lines_update.py b/saleor/graphql/checkout/tests/mutations/test_checkout_lines_update.py
index dd542b4b250..6d24ad2b12e 100644
--- a/saleor/graphql/checkout/tests/mutations/test_checkout_lines_update.py
+++ b/saleor/graphql/checkout/tests/mutations/test_checkout_lines_update.py
@@ -23,6 +23,8 @@
 from .....product.models import ProductChannelListing
 from .....warehouse.models import Reservation, Stock
 from .....webhook.event_types import WebhookEventAsyncType, WebhookEventSyncType
+from .....webhook.transport.asynchronous.transport import send_webhook_request_async
+from .....webhook.transport.utils import WebhookResponse
 from ....core.utils import to_global_id_or_none
 from ....tests.utils import assert_no_permission, get_graphql_content
 from ...mutations.utils import update_checkout_shipping_method_if_invalid
@@ -1460,10 +1462,15 @@ def test_checkout_lines_update_quantity_gift(user_api_client, checkout_with_item
 )
 @patch("saleor.webhook.transport.synchronous.transport.send_webhook_request_sync")
 @patch(
-    "saleor.webhook.transport.asynchronous.transport.send_webhook_request_async.apply_async"
+    "saleor.webhook.transport.asynchronous.transport.send_webhook_request_async.apply_async",
+    wraps=send_webhook_request_async.apply_async,
+)
+@patch(
+    "saleor.webhook.transport.asynchronous.transport.send_webhook_using_scheme_method"
 )
 @override_settings(PLUGINS=["saleor.plugins.webhook.plugin.WebhookPlugin"])
 def test_checkout_lines_update_triggers_webhooks(
+    mocked_send_webhook_using_scheme_method,
     mocked_send_webhook_request_async,
     mocked_send_webhook_request_sync,
     wrapped_call_checkout_info_event,
@@ -1474,6 +1481,7 @@ def test_checkout_lines_update_triggers_webhooks(
     product_with_single_variant,
 ):
     # given
+    mocked_send_webhook_using_scheme_method.return_value = WebhookResponse(content="")
     mocked_send_webhook_request_sync.return_value = []
     (
         tax_webhook,
@@ -1503,20 +1511,8 @@ def test_checkout_lines_update_triggers_webhooks(
     # then
     content = get_graphql_content(response)
     assert not content["data"]["checkoutLinesUpdate"]["errors"]
-
     assert wrapped_call_checkout_info_event.called
-
-    # confirm that event delivery was generated for each async webhook.
-    checkout_update_delivery = EventDelivery.objects.get(
-        webhook_id=checkout_updated_webhook.id
-    )
-    mocked_send_webhook_request_async.assert_called_once_with(
-        kwargs={"event_delivery_id": checkout_update_delivery.id},
-        queue=settings.CHECKOUT_WEBHOOK_EVENTS_CELERY_QUEUE_NAME,
-        bind=True,
-        retry_backoff=10,
-        retry_kwargs={"max_retries": 5},
-    )
+    assert mocked_send_webhook_request_async.call_count == 1
 
     # confirm each sync webhook was called without saving event delivery
     assert mocked_send_webhook_request_sync.call_count == 3
@@ -1524,24 +1520,23 @@ def test_checkout_lines_update_triggers_webhooks(
         webhook_id=checkout_updated_webhook.id
     ).exists()
 
-    shipping_methods_call, filter_shipping_call, tax_delivery_call = (
-        mocked_send_webhook_request_sync.mock_calls
-    )
-    shipping_methods_delivery = shipping_methods_call.args[0]
+    sync_deliveries = {
+        call.args[0].event_type: call.args[0]
+        for call in mocked_send_webhook_request_sync.mock_calls
+    }
+
+    assert WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT in sync_deliveries
+    shipping_methods_delivery = sync_deliveries[
+        WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
+    ]
     assert shipping_methods_delivery.webhook_id == shipping_webhook.id
-    assert (
-        shipping_methods_delivery.event_type
-        == WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
-    )
-    assert shipping_methods_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    filter_shipping_delivery = filter_shipping_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS in sync_deliveries
+    filter_shipping_delivery = sync_deliveries[
+        WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
+    ]
     assert filter_shipping_delivery.webhook_id == shipping_filter_webhook.id
-    assert (
-        filter_shipping_delivery.event_type
-        == WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
-    )
-    assert filter_shipping_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    tax_delivery = tax_delivery_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES in sync_deliveries
+    tax_delivery = sync_deliveries[WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES]
     assert tax_delivery.webhook_id == tax_webhook.id
diff --git a/saleor/graphql/checkout/tests/mutations/test_checkout_remove_promo_code.py b/saleor/graphql/checkout/tests/mutations/test_checkout_remove_promo_code.py
index 5fd3b1d3cee..296a16b004a 100644
--- a/saleor/graphql/checkout/tests/mutations/test_checkout_remove_promo_code.py
+++ b/saleor/graphql/checkout/tests/mutations/test_checkout_remove_promo_code.py
@@ -16,6 +16,8 @@
 from .....discount.models import Voucher, VoucherChannelListing, VoucherCode
 from .....plugins.manager import get_plugins_manager
 from .....webhook.event_types import WebhookEventAsyncType, WebhookEventSyncType
+from .....webhook.transport.asynchronous.transport import send_webhook_request_async
+from .....webhook.transport.utils import WebhookResponse
 from ....core.utils import to_global_id_or_none
 from ....tests.utils import get_graphql_content
 
@@ -766,10 +768,15 @@ def test_with_active_problems_flow(
 )
 @patch("saleor.webhook.transport.synchronous.transport.send_webhook_request_sync")
 @patch(
-    "saleor.webhook.transport.asynchronous.transport.send_webhook_request_async.apply_async"
+    "saleor.webhook.transport.asynchronous.transport.send_webhook_request_async.apply_async",
+    wraps=send_webhook_request_async.apply_async,
+)
+@patch(
+    "saleor.webhook.transport.asynchronous.transport.send_webhook_using_scheme_method"
 )
 @override_settings(PLUGINS=["saleor.plugins.webhook.plugin.WebhookPlugin"])
 def test_checkout_remove_triggers_webhooks(
+    mocked_send_webhook_using_scheme_method,
     mocked_send_webhook_request_async,
     mocked_send_webhook_request_sync,
     wrapped_call_checkout_info_event,
@@ -779,6 +786,7 @@ def test_checkout_remove_triggers_webhooks(
     checkout_with_voucher,
 ):
     # given
+    mocked_send_webhook_using_scheme_method.return_value = WebhookResponse(content="")
     mocked_send_webhook_request_sync.return_value = []
     (
         tax_webhook,
@@ -797,20 +805,8 @@ def test_checkout_remove_triggers_webhooks(
 
     # then
     assert not content["errors"]
-
     assert wrapped_call_checkout_info_event.called
-
-    # confirm that event delivery was generated for each async webhook.
-    checkout_update_delivery = EventDelivery.objects.get(
-        webhook_id=checkout_updated_webhook.id
-    )
-    mocked_send_webhook_request_async.assert_called_once_with(
-        kwargs={"event_delivery_id": checkout_update_delivery.id},
-        queue=settings.CHECKOUT_WEBHOOK_EVENTS_CELERY_QUEUE_NAME,
-        bind=True,
-        retry_backoff=10,
-        retry_kwargs={"max_retries": 5},
-    )
+    assert mocked_send_webhook_request_async.call_count == 1
 
     # confirm each sync webhook was called without saving event delivery
     assert mocked_send_webhook_request_sync.call_count == 3
@@ -818,24 +814,23 @@ def test_checkout_remove_triggers_webhooks(
         webhook_id=checkout_updated_webhook.id
     ).exists()
 
-    shipping_methods_call, filter_shipping_call, tax_delivery_call = (
-        mocked_send_webhook_request_sync.mock_calls
-    )
-    shipping_methods_delivery = shipping_methods_call.args[0]
+    sync_deliveries = {
+        call.args[0].event_type: call.args[0]
+        for call in mocked_send_webhook_request_sync.mock_calls
+    }
+
+    assert WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT in sync_deliveries
+    shipping_methods_delivery = sync_deliveries[
+        WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
+    ]
     assert shipping_methods_delivery.webhook_id == shipping_webhook.id
-    assert (
-        shipping_methods_delivery.event_type
-        == WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
-    )
-    assert shipping_methods_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    filter_shipping_delivery = filter_shipping_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS in sync_deliveries
+    filter_shipping_delivery = sync_deliveries[
+        WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
+    ]
     assert filter_shipping_delivery.webhook_id == shipping_filter_webhook.id
-    assert (
-        filter_shipping_delivery.event_type
-        == WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
-    )
-    assert filter_shipping_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    tax_delivery = tax_delivery_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES in sync_deliveries
+    tax_delivery = sync_deliveries[WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES]
     assert tax_delivery.webhook_id == tax_webhook.id
diff --git a/saleor/graphql/checkout/tests/mutations/test_checkout_shipping_method_update.py b/saleor/graphql/checkout/tests/mutations/test_checkout_shipping_method_update.py
index 039049ac96d..0d5b6e097da 100644
--- a/saleor/graphql/checkout/tests/mutations/test_checkout_shipping_method_update.py
+++ b/saleor/graphql/checkout/tests/mutations/test_checkout_shipping_method_update.py
@@ -16,6 +16,8 @@
 from .....shipping import models as shipping_models
 from .....shipping.utils import convert_to_shipping_method_data
 from .....webhook.event_types import WebhookEventAsyncType, WebhookEventSyncType
+from .....webhook.transport.asynchronous.transport import send_webhook_request_async
+from .....webhook.transport.utils import WebhookResponse
 from ....core.utils import to_global_id_or_none
 from ....tests.utils import get_graphql_content
 
@@ -525,45 +527,33 @@ def test_checkout_shipping_method_update_triggers_webhooks(
     assert not content["data"]["checkoutShippingMethodUpdate"]["errors"]
 
     assert wrapped_call_checkout_info_event.called
-
-    # confirm that event delivery was generated for each async webhook.
-    checkout_update_delivery = EventDelivery.objects.get(
-        webhook_id=checkout_updated_webhook.id
-    )
-    mocked_send_webhook_request_async.assert_called_once_with(
-        kwargs={"event_delivery_id": checkout_update_delivery.id},
-        queue=settings.CHECKOUT_WEBHOOK_EVENTS_CELERY_QUEUE_NAME,
-        bind=True,
-        retry_backoff=10,
-        retry_kwargs={"max_retries": 5},
-    )
+    assert mocked_send_webhook_request_async.call_count == 1
 
     # confirm each sync webhook was called without saving event delivery
-    assert mocked_send_webhook_request_sync.call_count == 3
+    assert mocked_send_webhook_request_sync.call_count == 4
     assert not EventDelivery.objects.exclude(
         webhook_id=checkout_updated_webhook.id
     ).exists()
 
-    shipping_methods_call, filter_shipping_call, tax_delivery_call = (
-        mocked_send_webhook_request_sync.mock_calls
-    )
-    shipping_methods_delivery = shipping_methods_call.args[0]
+    sync_deliveries = {
+        call.args[0].event_type: call.args[0]
+        for call in mocked_send_webhook_request_sync.mock_calls
+    }
+
+    assert WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT in sync_deliveries
+    shipping_methods_delivery = sync_deliveries[
+        WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
+    ]
     assert shipping_methods_delivery.webhook_id == shipping_webhook.id
-    assert (
-        shipping_methods_delivery.event_type
-        == WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
-    )
-    assert shipping_methods_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    filter_shipping_delivery = filter_shipping_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS in sync_deliveries
+    filter_shipping_delivery = sync_deliveries[
+        WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
+    ]
     assert filter_shipping_delivery.webhook_id == shipping_filter_webhook.id
-    assert (
-        filter_shipping_delivery.event_type
-        == WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
-    )
-    assert filter_shipping_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    tax_delivery = tax_delivery_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES in sync_deliveries
+    tax_delivery = sync_deliveries[WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES]
     assert tax_delivery.webhook_id == tax_webhook.id
 
 
@@ -573,10 +563,15 @@ def test_checkout_shipping_method_update_triggers_webhooks(
 )
 @patch("saleor.webhook.transport.synchronous.transport.send_webhook_request_sync")
 @patch(
-    "saleor.webhook.transport.asynchronous.transport.send_webhook_request_async.apply_async"
+    "saleor.webhook.transport.asynchronous.transport.send_webhook_request_async.apply_async",
+    wraps=send_webhook_request_async.apply_async,
+)
+@patch(
+    "saleor.webhook.transport.asynchronous.transport.send_webhook_using_scheme_method"
 )
 @override_settings(PLUGINS=["saleor.plugins.webhook.plugin.WebhookPlugin"])
 def test_checkout_shipping_method_update_external_shipping_triggers_webhooks(
+    mocked_send_webhook_using_scheme_method,
     mocked_send_webhook_request_async,
     mocked_send_webhook_request_sync,
     wrapped_call_checkout_info_event,
@@ -587,6 +582,7 @@ def test_checkout_shipping_method_update_external_shipping_triggers_webhooks(
     checkout_with_item,
 ):
     # given
+    mocked_send_webhook_using_scheme_method.return_value = WebhookResponse(content="")
     (
         tax_webhook,
         shipping_webhook,
@@ -628,45 +624,34 @@ def test_checkout_shipping_method_update_external_shipping_triggers_webhooks(
     assert not content["data"]["checkoutShippingMethodUpdate"]["errors"]
 
     assert wrapped_call_checkout_info_event.called
-
-    # confirm that event delivery was generated for each async webhook.
-    checkout_update_delivery = EventDelivery.objects.get(
-        webhook_id=checkout_updated_webhook.id
-    )
-    mocked_send_webhook_request_async.assert_called_once_with(
-        kwargs={"event_delivery_id": checkout_update_delivery.id},
-        queue=settings.CHECKOUT_WEBHOOK_EVENTS_CELERY_QUEUE_NAME,
-        bind=True,
-        retry_backoff=10,
-        retry_kwargs={"max_retries": 5},
-    )
+    assert wrapped_call_checkout_info_event.called
+    assert mocked_send_webhook_request_async.call_count == 1
 
     # confirm each sync webhook was called without saving event delivery
-    assert mocked_send_webhook_request_sync.call_count == 3
+    assert mocked_send_webhook_request_sync.call_count == 5
     assert not EventDelivery.objects.exclude(
         webhook_id=checkout_updated_webhook.id
     ).exists()
 
-    shipping_methods_call, filter_shipping_call, tax_delivery_call = (
-        mocked_send_webhook_request_sync.mock_calls
-    )
-    shipping_methods_delivery = shipping_methods_call.args[0]
+    sync_deliveries = {
+        call.args[0].event_type: call.args[0]
+        for call in mocked_send_webhook_request_sync.mock_calls
+    }
+
+    assert WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT in sync_deliveries
+    shipping_methods_delivery = sync_deliveries[
+        WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
+    ]
     assert shipping_methods_delivery.webhook_id == shipping_webhook.id
-    assert (
-        shipping_methods_delivery.event_type
-        == WebhookEventSyncType.SHIPPING_LIST_METHODS_FOR_CHECKOUT
-    )
-    assert shipping_methods_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    filter_shipping_delivery = filter_shipping_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS in sync_deliveries
+    filter_shipping_delivery = sync_deliveries[
+        WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
+    ]
     assert filter_shipping_delivery.webhook_id == shipping_filter_webhook.id
-    assert (
-        filter_shipping_delivery.event_type
-        == WebhookEventSyncType.CHECKOUT_FILTER_SHIPPING_METHODS
-    )
-    assert filter_shipping_call.kwargs["timeout"] == settings.WEBHOOK_SYNC_TIMEOUT
 
-    tax_delivery = tax_delivery_call.args[0]
+    assert WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES in sync_deliveries
+    tax_delivery = sync_deliveries[WebhookEventSyncType.CHECKOUT_CALCULATE_TAXES]
     assert tax_delivery.webhook_id == tax_webhook.id
 
 
diff --git a/saleor/graphql/core/types/taxes.py b/saleor/graphql/core/types/taxes.py
index b517d7abad5..74effff67c9 100644
--- a/saleor/graphql/core/types/taxes.py
+++ b/saleor/graphql/core/types/taxes.py
@@ -7,6 +7,7 @@
 
 from ....checkout import base_calculations
 from ....checkout.models import Checkout, CheckoutLine
+from ....core.db.connection import allow_writer_in_context
 from ....core.prices import quantize_price
 from ....discount import DiscountType
 from ....discount.utils.checkout import has_checkout_order_promotion
@@ -185,6 +186,7 @@ def calculate_charge_taxes(tax_configs_per_country):
         if isinstance(root, CheckoutLine):
             checkout = CheckoutByTokenLoader(info.context).load(root.checkout_id)
 
+            @allow_writer_in_context(info.context)
             def load_channel(checkout):
                 country_code = checkout.get_country()
                 load_tax_config_with_country = partial(
diff --git a/saleor/graphql/webhook/mutations/webhook_trigger.py b/saleor/graphql/webhook/mutations/webhook_trigger.py
index 8fd0eb50778..2ff62243359 100644
--- a/saleor/graphql/webhook/mutations/webhook_trigger.py
+++ b/saleor/graphql/webhook/mutations/webhook_trigger.py
@@ -1,3 +1,5 @@
+from dataclasses import asdict
+
 import graphene
 from celery.exceptions import Retry
 from django.db.models import Exists, OuterRef
@@ -5,11 +7,15 @@
 
 from ....app.models import App
 from ....core import EventDeliveryStatus
+from ....core import models as core_models
+from ....core.utils.events import get_is_deferred_payload
 from ....discount import models as discount_models
+from ....graphql.utils import get_user_or_app_from_context
 from ....permission.auth_filters import AuthorizationFilters
 from ....webhook.error_codes import WebhookTriggerErrorCode
 from ....webhook.event_types import WebhookEventAsyncType
 from ....webhook.models import Webhook
+from ....webhook.transport.utils import prepare_deferred_payload_data
 from ...core import ResolveInfo
 from ...core.descriptions import ADDED_IN_311, PREVIEW_FEATURE
 from ...core.doc_category import DOC_CATEGORY_WEBHOOKS
@@ -148,29 +154,50 @@ def get_instance(cls, info: ResolveInfo, object_id):
     def perform_mutation(cls, _root, info, **data):
         from ....webhook.transport.asynchronous.transport import (
             create_deliveries_for_subscriptions,
+            generate_deferred_payloads,
             send_webhook_request_async,
         )
 
         event_type, object, webhook = cls.validate_input(info, **data)
         delivery = None
+        requestor = get_user_or_app_from_context(info.context)
 
-        if all([event_type, object, webhook]):
+        if event_type and object and webhook:
             if event_type in [
                 WebhookEventAsyncType.VOUCHER_CODES_CREATED,
                 WebhookEventAsyncType.VOUCHER_CODES_DELETED,
             ]:
                 object = [object]
 
-            deliveries = create_deliveries_for_subscriptions(
-                event_type, object, [webhook]
-            )
-            if deliveries:
-                delivery = deliveries[0]
-                try:
-                    send_webhook_request_async(delivery.id)
-                    return WebhookTrigger(delivery=delivery)
-                except Retry:
-                    delivery.status = EventDeliveryStatus.FAILED
-                    delivery.save(update_fields=["status"])
+            is_deferred_payload = get_is_deferred_payload(event_type)
+            if is_deferred_payload:
+                delivery = core_models.EventDelivery(
+                    status=EventDeliveryStatus.PENDING,
+                    event_type=event_type,
+                    webhook=webhook,
+                )
+                delivery.save()
+                deferred_payload_data = prepare_deferred_payload_data(
+                    object, requestor, None
+                )
+                generate_deferred_payloads.apply_async(
+                    kwargs={
+                        "event_delivery_ids": [delivery.pk],
+                        "deferred_payload_data": asdict(deferred_payload_data),
+                    },
+                    bind=True,
+                )
+            else:
+                deliveries = create_deliveries_for_subscriptions(
+                    event_type, object, [webhook]
+                )
+                if deliveries:
+                    delivery = deliveries[0]
+                    try:
+                        send_webhook_request_async(delivery.id)
+                        return WebhookTrigger(delivery=delivery)
+                    except Retry:
+                        delivery.status = EventDeliveryStatus.FAILED
+                        delivery.save(update_fields=["status"])
 
         return WebhookTrigger(delivery=delivery)
diff --git a/saleor/graphql/webhook/tests/mutations/test_webhook_trigger.py b/saleor/graphql/webhook/tests/mutations/test_webhook_trigger.py
index c29bedf2b6e..cdcca616265 100644
--- a/saleor/graphql/webhook/tests/mutations/test_webhook_trigger.py
+++ b/saleor/graphql/webhook/tests/mutations/test_webhook_trigger.py
@@ -7,6 +7,7 @@
 from .....graphql.tests.utils import get_graphql_content
 from .....webhook.error_codes import WebhookTriggerErrorCode
 from .....webhook.event_types import WebhookEventAsyncType
+from .....webhook.transport.asynchronous.transport import generate_deferred_payloads
 from ....tests.utils import assert_no_permission
 from ...subscription_types import WEBHOOK_TYPES_MAP
 
@@ -19,6 +20,7 @@
                 message
             }
             delivery {
+                id
                 status
                 eventType
                 payload
@@ -350,3 +352,50 @@ def test_webhook_trigger_for_removed_app(
     assert app_data["delivery"] is None
     assert app_data["errors"][0]["code"] == WebhookTriggerErrorCode.NOT_FOUND.name
     assert app_data["errors"][0]["field"] == "webhookId"
+
+
+@mock.patch(
+    "saleor.webhook.transport.asynchronous.transport.send_webhook_request_async.apply_async"
+)
+@mock.patch(
+    "saleor.webhook.transport.asynchronous.transport.generate_deferred_payloads.apply_async",
+    wraps=generate_deferred_payloads.apply_async,
+)
+def test_webhook_trigger_for_deferred_payload(
+    mocked_generate_deferred_payloads,
+    mocked_send_webhook_request_async,
+    staff_api_client,
+    checkout,
+    subscription_checkout_updated_webhook,
+    permission_manage_checkouts,
+):
+    # given
+    query = WEBHOOK_TRIGGER_MUTATION
+    staff_api_client.user.user_permissions.add(permission_manage_checkouts)
+    checkout_id = graphene.Node.to_global_id("Checkout", checkout.pk)
+    webhook = subscription_checkout_updated_webhook
+    webhook_id = graphene.Node.to_global_id("Webhook", webhook.id)
+
+    variables = {"webhookId": webhook_id, "objectId": checkout_id}
+
+    # when
+    response = staff_api_client.post_graphql(query, variables)
+
+    # then
+    content = get_graphql_content(response)
+    data = content["data"]["webhookTrigger"]
+
+    assert not data["errors"]
+    delivery_node_id = data["delivery"]["id"]
+    _, delivery_pk = graphene.Node.from_global_id(delivery_node_id)
+    delivery_pk = int(delivery_pk)
+    assert mocked_generate_deferred_payloads.called
+    generate_payload_kwargs = mocked_generate_deferred_payloads.call_args.kwargs[
+        "kwargs"
+    ]
+    assert generate_payload_kwargs["event_delivery_ids"] == [delivery_pk]
+    assert generate_payload_kwargs["deferred_payload_data"]["object_id"] == checkout.pk
+
+    assert mocked_send_webhook_request_async.called
+    send_webhook_kwargs = mocked_send_webhook_request_async.call_args.kwargs["kwargs"]
+    assert send_webhook_kwargs["event_delivery_id"] == delivery_pk
diff --git a/saleor/tests/fixtures.py b/saleor/tests/fixtures.py
index 977dfab4e1c..1fe52e26969 100644
--- a/saleor/tests/fixtures.py
+++ b/saleor/tests/fixtures.py
@@ -9625,7 +9625,22 @@ def setup_checkout_webhooks(
     permission_manage_checkouts,
 ):
     subscription_async_webhooks = """
+    fragment IssuingPrincipal on IssuingPrincipal {
+      ... on App {
+        id
+        name
+      }
+      ... on User {
+        id
+        email
+      }
+    }
+
     fragment CheckoutFragment on Checkout {
+      shippingMethods {
+        id
+        name
+      }
       shippingPrice {
         gross {
           amount
@@ -9641,21 +9656,33 @@ def setup_checkout_webhooks(
     subscription {
       event {
         ... on CheckoutCreated {
+          issuingPrincipal {
+            ...IssuingPrincipal
+          }
           checkout {
             ...CheckoutFragment
           }
         }
         ... on CheckoutUpdated {
+          issuingPrincipal {
+            ...IssuingPrincipal
+          }
           checkout {
             ...CheckoutFragment
           }
         }
         ... on CheckoutFullyPaid {
+          issuingPrincipal {
+            ...IssuingPrincipal
+          }
           checkout {
             ...CheckoutFragment
           }
         }
         ... on CheckoutMetadataUpdated {
+          issuingPrincipal {
+            ...IssuingPrincipal
+          }
           checkout {
             ...CheckoutFragment
           }
diff --git a/saleor/webhook/event_types.py b/saleor/webhook/event_types.py
index 6622679c255..73154e1082d 100644
--- a/saleor/webhook/event_types.py
+++ b/saleor/webhook/event_types.py
@@ -600,10 +600,12 @@ class WebhookEventAsyncType:
         CHECKOUT_CREATED: {
             "name": "Checkout created",
             "permission": CheckoutPermissions.MANAGE_CHECKOUTS,
+            "is_deferred_payload": True,
         },
         CHECKOUT_UPDATED: {
             "name": "Checkout updated",
             "permission": CheckoutPermissions.MANAGE_CHECKOUTS,
+            "is_deferred_payload": True,
         },
         CHECKOUT_FULLY_PAID: {
             "name": "Checkout fully paid",
diff --git a/saleor/webhook/transport/asynchronous/tests/test_deferred_payload.py b/saleor/webhook/transport/asynchronous/tests/test_deferred_payload.py
new file mode 100644
index 00000000000..3ec8250a3e9
--- /dev/null
+++ b/saleor/webhook/transport/asynchronous/tests/test_deferred_payload.py
@@ -0,0 +1,199 @@
+import json
+import uuid
+from dataclasses import asdict
+from unittest import mock
+
+import pytest
+
+from .....checkout.calculations import fetch_checkout_data
+from .....checkout.fetch import fetch_checkout_info, fetch_checkout_lines
+from .....core import EventDeliveryStatus
+from .....core.models import EventDelivery
+from ....event_types import WebhookEventAsyncType
+from ...utils import get_webhooks_for_event
+from ..transport import (
+    DeferredPayloadData,
+    generate_deferred_payloads,
+    trigger_webhooks_async,
+)
+
+
+@pytest.fixture
+def fetch_kwargs(checkout_with_items, plugins_manager):
+    lines, _ = fetch_checkout_lines(checkout_with_items)
+    return {
+        "checkout_info": fetch_checkout_info(
+            checkout_with_items, lines, plugins_manager
+        ),
+        "manager": plugins_manager,
+        "lines": lines,
+        "address": checkout_with_items.shipping_address
+        or checkout_with_items.billing_address,
+    }
+
+
+@mock.patch(
+    "saleor.webhook.transport.asynchronous.transport.generate_deferred_payloads.apply_async"
+)
+def test_call_trigger_webhook_async_deferred_payload(
+    mocked_generate_deferred_payloads,
+    checkout_with_item,
+    setup_checkout_webhooks,
+    staff_user,
+):
+    # given
+    checkout = checkout_with_item
+    event_type = WebhookEventAsyncType.CHECKOUT_UPDATED
+    _, _, _, checkout_updated_webhook = setup_checkout_webhooks(event_type)
+    webhooks = get_webhooks_for_event(event_type)
+
+    # when
+    trigger_webhooks_async(
+        data=None,
+        event_type=event_type,
+        webhooks=webhooks,
+        subscribable_object=checkout,
+        requestor=staff_user,
+    )
+
+    # then
+    delivery = checkout_updated_webhook.eventdelivery_set.first()
+    call_kwargs = mocked_generate_deferred_payloads.call_args.kwargs
+    deferred_payload_data = call_kwargs["kwargs"]["deferred_payload_data"]
+
+    assert deferred_payload_data["model_name"] == "checkout.checkout"
+    assert deferred_payload_data["object_id"] == checkout.pk
+    assert deferred_payload_data["requestor_model_name"] == "account.user"
+    assert deferred_payload_data["requestor_object_id"] == staff_user.pk
+    assert call_kwargs["kwargs"]["event_delivery_ids"] == [delivery.id]
+
+    assert delivery.event_type == event_type
+    assert delivery.payload is None
+    assert delivery.webhook == checkout_updated_webhook
+
+
+@mock.patch(
+    "saleor.webhook.transport.asynchronous.transport.send_webhook_request_async.apply_async"
+)
+def test_generate_deferred_payload(
+    mocked_send_webhook_request_async,
+    checkout_with_item,
+    setup_checkout_webhooks,
+    staff_user,
+    fetch_kwargs,
+):
+    # given
+    checkout = checkout_with_item
+    fetch_checkout_data(**fetch_kwargs)
+
+    event_type = WebhookEventAsyncType.CHECKOUT_UPDATED
+    _, _, _, checkout_updated_webhook = setup_checkout_webhooks(event_type)
+    deferred_payload_data = DeferredPayloadData(
+        model_name="checkout.checkout",
+        object_id=checkout.pk,
+        requestor_model_name="account.user",
+        requestor_object_id=staff_user.pk,
+        request_time=None,
+    )
+    delivery = EventDelivery(
+        event_type=event_type,
+        webhook=checkout_updated_webhook,
+        status=EventDeliveryStatus.PENDING,
+    )
+    delivery.save()
+
+    # when
+    generate_deferred_payloads.delay(
+        event_delivery_ids=[delivery.pk],
+        deferred_payload_data=asdict(deferred_payload_data),
+    )
+
+    # then
+    checkout.refresh_from_db()
+    delivery.refresh_from_db()
+    assert delivery.payload
+
+    data = delivery.payload.get_payload()
+    data = json.loads(data)
+
+    assert data["issuingPrincipal"]["email"] == staff_user.email
+    assert (
+        data["checkout"]["totalPrice"]["gross"]["amount"] == checkout.total_gross_amount
+    )
+
+    assert mocked_send_webhook_request_async.call_count == 1
+    call_kwargs = mocked_send_webhook_request_async.call_args.kwargs
+    assert call_kwargs["kwargs"]["event_delivery_id"] == delivery.pk
+
+
+def test_generate_deferred_payload_model_pk_does_not_exist(
+    checkout_with_item, setup_checkout_webhooks, staff_user, fetch_kwargs
+):
+    # given
+    event_type = WebhookEventAsyncType.CHECKOUT_UPDATED
+    _, _, _, checkout_updated_webhook = setup_checkout_webhooks(event_type)
+    deferred_payload_data = DeferredPayloadData(
+        model_name="checkout.checkout",
+        object_id=uuid.uuid4(),
+        requestor_model_name="account.user",
+        requestor_object_id=999999,
+        request_time=None,
+    )
+    delivery = EventDelivery(
+        event_type=event_type,
+        webhook=checkout_updated_webhook,
+        status=EventDeliveryStatus.PENDING,
+    )
+    delivery.save()
+
+    # when
+    generate_deferred_payloads.delay(
+        event_delivery_ids=[delivery.pk],
+        deferred_payload_data=asdict(deferred_payload_data),
+    )
+
+    # then
+    delivery.refresh_from_db()
+    assert delivery.status == EventDeliveryStatus.FAILED
+    assert delivery.payload is None
+
+
+@mock.patch(
+    "saleor.webhook.transport.asynchronous.transport.send_webhook_request_async.apply_async"
+)
+@mock.patch(
+    "saleor.webhook.transport.asynchronous.transport.generate_deferred_payloads.apply_async",
+    wraps=generate_deferred_payloads.apply_async,
+)
+def test_pass_queue_to_send_webhook_request_async(
+    mocked_generate_deferred_payloads,
+    mocked_send_webhook_request_async,
+    checkout_with_item,
+    setup_checkout_webhooks,
+    staff_user,
+):
+    # given
+    checkout = checkout_with_item
+    event_type = WebhookEventAsyncType.CHECKOUT_UPDATED
+    _, _, _, checkout_updated_webhook = setup_checkout_webhooks(event_type)
+    webhooks = get_webhooks_for_event(event_type)
+    queue = "checkout_queue"
+
+    # when
+    trigger_webhooks_async(
+        data=None,
+        event_type=event_type,
+        webhooks=webhooks,
+        subscribable_object=checkout,
+        requestor=staff_user,
+        queue=queue,
+    )
+
+    # then
+    call_kwargs_generate_payloads = mocked_generate_deferred_payloads.call_args.kwargs
+    assert call_kwargs_generate_payloads["queue"] == queue
+
+    call_kwargs_send_webhook_request = (
+        mocked_send_webhook_request_async.call_args.kwargs
+    )
+    assert call_kwargs_send_webhook_request["queue"] == queue
diff --git a/saleor/webhook/transport/asynchronous/transport.py b/saleor/webhook/transport/asynchronous/transport.py
index a31ca136869..bd77a198c42 100644
--- a/saleor/webhook/transport/asynchronous/transport.py
+++ b/saleor/webhook/transport/asynchronous/transport.py
@@ -1,13 +1,15 @@
 import datetime
 import json
 import logging
+from collections import defaultdict
 from collections.abc import Sequence
-from dataclasses import dataclass
+from dataclasses import asdict, dataclass
 from typing import TYPE_CHECKING, Any, Callable, Optional
 from urllib.parse import urlparse
 
 from celery import group
 from celery.utils.log import get_task_logger
+from django.apps import apps
 from django.conf import settings
 from django.db import transaction
 
@@ -20,6 +22,7 @@
 from ....graphql.core.dataloaders import DataLoader
 from ....graphql.webhook.subscription_payload import (
     generate_payload_from_subscription,
+    generate_payload_promise_from_subscription,
     get_pre_save_payload_key,
     initialize_request,
 )
@@ -28,6 +31,8 @@
 from ...event_types import WebhookEventAsyncType, WebhookEventSyncType
 from ...observability import WebhookData
 from ..utils import (
+    DeferredPayloadData,
+    RequestorModelName,
     WebhookResponse,
     WebhookSchemes,
     attempt_update,
@@ -35,7 +40,9 @@
     create_attempt,
     delivery_update,
     get_delivery_for_webhook,
+    get_multiple_deliveries_for_webhooks,
     handle_webhook_retry,
+    prepare_deferred_payload_data,
     send_webhook_using_scheme_method,
 )
 
@@ -181,9 +188,9 @@ def create_deliveries_for_multiple_subscription_objects(
 
 
 def create_deliveries_for_subscriptions(
-    event_type,
+    event_type: str,
     subscribable_object,
-    webhooks,
+    webhooks: Sequence["Webhook"],
     requestor=None,
     allow_replica=False,
     pre_save_payloads: Optional[dict] = None,
@@ -212,11 +219,16 @@ def create_deliveries_for_subscriptions(
     )
 
 
-def group_webhooks_by_subscription(webhooks):
-    subscription = [webhook for webhook in webhooks if webhook.subscription_query]
-    regular = [webhook for webhook in webhooks if not webhook.subscription_query]
+def group_webhooks_by_subscription(
+    webhooks: Sequence["Webhook"],
+) -> tuple[list["Webhook"], list["Webhook"]]:
+    """Group webhooks by subscription query.
 
-    return regular, subscription
+    Returns a tuple of two lists: legacy webhooks and subscription webhooks.
+    """
+    subscription = [webhook for webhook in webhooks if webhook.subscription_query]
+    legacy = [webhook for webhook in webhooks if not webhook.subscription_query]
+    return legacy, subscription
 
 
 @allow_writer()
@@ -269,10 +281,23 @@ def trigger_webhooks_async_for_multiple_objects(
     :param allow_replica: use a replica database.
     :param queue: defines the queue to which the event should be sent.
     """
-    regular_webhooks, subscription_webhooks = group_webhooks_by_subscription(webhooks)
-    deliveries = []
+    legacy_webhooks, subscription_webhooks = group_webhooks_by_subscription(webhooks)
+
+    is_deferred_payload = WebhookEventAsyncType.EVENT_MAP.get(event_type, {}).get(
+        "is_deferred_payload", False
+    )
+
+    # List of deliveries with payloads.
+    deliveries: list[EventDelivery] = []
+
+    # List of deliveries and data to generate deferred payloads for each subscribable
+    # object. Note: we assume that all subscribable objects are of the same type.
+    deferred_deliveries_per_object: dict[
+        int, list[tuple[EventDelivery, DeferredPayloadData]]
+    ] = defaultdict(list)
+
     for webhook_payload_detail in webhook_payloads_data:
-        if regular_webhooks:
+        if legacy_webhooks:
             data = webhook_payload_detail.data
             if webhook_payload_detail.legacy_data_generator:
                 data = webhook_payload_detail.legacy_data_generator()
@@ -282,32 +307,69 @@ def trigger_webhooks_async_for_multiple_objects(
                 )
 
             with allow_writer():
-                # Use transaction to ensure EventPayload and EventDelivery are created together, preventing inconsistent DB state.
+                # Use transaction to ensure EventPayload and EventDelivery are created
+                # together, preventing inconsistent DB state.
                 with transaction.atomic():
                     payload = EventPayload.objects.create_with_payload_file(data)
                     deliveries.extend(
                         create_event_delivery_list_for_webhooks(
-                            webhooks=regular_webhooks,
+                            webhooks=legacy_webhooks,
                             event_payload=payload,
                             event_type=event_type,
                         )
                     )
 
     if subscription_webhooks:
-        deliveries.extend(
-            create_deliveries_for_multiple_subscription_objects(
-                event_type=event_type,
-                subscribable_objects=[
-                    webhook_payload_data.subscribable_object
-                    for webhook_payload_data in webhook_payloads_data
-                ],
-                webhooks=subscription_webhooks,
-                requestor=requestor,
-                allow_replica=allow_replica,
-                pre_save_payloads=pre_save_payloads,
-                request_time=request_time,
+        subscribable_objects = [
+            webhook_payload_data.subscribable_object
+            for webhook_payload_data in webhook_payloads_data
+        ]
+        if is_deferred_payload:
+            _deferred_deliveries_to_create = []
+            for subscribable_object in subscribable_objects:
+                for webhook in subscription_webhooks:
+                    delivery = EventDelivery(
+                        status=EventDeliveryStatus.PENDING,
+                        event_type=event_type,
+                        webhook=webhook,
+                    )
+                    _deferred_deliveries_to_create.append(delivery)
+                    deferred_payload_data = prepare_deferred_payload_data(
+                        subscribable_object, requestor, request_time
+                    )
+                    deferred_deliveries_per_object[subscribable_object.pk].append(
+                        (delivery, deferred_payload_data)
+                    )
+
+            EventDelivery.objects.bulk_create(_deferred_deliveries_to_create)
+        else:
+            deliveries.extend(
+                create_deliveries_for_multiple_subscription_objects(
+                    event_type=event_type,
+                    subscribable_objects=subscribable_objects,
+                    webhooks=subscription_webhooks,
+                    requestor=requestor,
+                    allow_replica=allow_replica,
+                    pre_save_payloads=pre_save_payloads,
+                    request_time=request_time,
+                )
             )
-        )
+
+    if deferred_deliveries_per_object:
+        for _, deferred_deliveries in deferred_deliveries_per_object.items():
+            event_delivery_ids = [delivery.id for delivery, _ in deferred_deliveries]
+
+            # Trigger deferred payload generation task for each subscribable object.
+            generate_deferred_payloads.apply_async(
+                kwargs={
+                    "event_delivery_ids": event_delivery_ids,
+                    "deferred_payload_data": asdict(deferred_payload_data),
+                    "queue": queue,
+                },
+                queue=queue or settings.WEBHOOK_CELERY_QUEUE_NAME,
+                bind=True,
+            )
+
     for delivery in deliveries:
         send_webhook_request_async.apply_async(
             kwargs={"event_delivery_id": delivery.id},
@@ -364,6 +426,81 @@ def trigger_webhooks_async(
     )
 
 
+@app.task(
+    queue=settings.WEBHOOK_CELERY_QUEUE_NAME,
+    bind=True,
+)
+def generate_deferred_payloads(
+    self,
+    event_delivery_ids: list,
+    deferred_payload_data: dict,
+    queue: Optional[str] = None,
+):
+    deliveries = list(get_multiple_deliveries_for_webhooks(event_delivery_ids).values())
+    args_obj = DeferredPayloadData(**deferred_payload_data)
+    requestor = None
+    if args_obj.requestor_object_id and args_obj.requestor_model_name in (
+        RequestorModelName.APP,
+        RequestorModelName.USER,
+    ):
+        model = apps.get_model(args_obj.requestor_model_name)
+        requestor = model.objects.filter(pk=args_obj.requestor_object_id).first()
+
+    subscribable_object = (
+        apps.get_model(args_obj.model_name)
+        .objects.filter(pk=args_obj.object_id)
+        .first()
+    )
+    if not subscribable_object:
+        EventDelivery.objects.filter(pk__in=event_delivery_ids).update(
+            status=EventDeliveryStatus.FAILED
+        )
+        return
+
+    for delivery in deliveries:
+        event_type = delivery.event_type
+        webhook = delivery.webhook
+        request = initialize_request(
+            requestor,
+            event_type in WebhookEventSyncType.ALL,
+            event_type=event_type,
+            allow_replica=True,
+            request_time=args_obj.request_time,
+        )
+        data_promise = generate_payload_promise_from_subscription(
+            event_type=event_type,
+            subscribable_object=subscribable_object,
+            subscription_query=webhook.subscription_query,  # type: ignore
+            request=request,
+            app=webhook.app,
+        )
+
+        if data_promise:
+            data = data_promise.get()
+            if data:
+                data_json = json.dumps({**data})
+                with allow_writer():
+                    event_payload = EventPayload.objects.create_with_payload_file(
+                        data_json
+                    )
+                    delivery.payload = event_payload
+                    delivery.save(update_fields=["payload"])
+
+                # Trigger webhook delivery task when the payload is ready.
+                send_webhook_request_async.apply_async(
+                    kwargs={
+                        "event_delivery_id": delivery.id,
+                    },
+                    queue=get_queue_name_for_webhook(
+                        delivery.webhook,
+                        default_queue=queue or settings.WEBHOOK_CELERY_QUEUE_NAME,
+                    ),
+                    bind=True,
+                    retry_backoff=10,
+                    retry_kwargs={"max_retries": 5},
+                )
+
+
 @app.task(
     queue=settings.WEBHOOK_CELERY_QUEUE_NAME,
     bind=True,
@@ -379,6 +516,7 @@ def send_webhook_request_async(self, event_delivery_id):
     domain = get_domain()
     attempt = create_attempt(delivery, self.request.id)
     delivery_status = EventDeliveryStatus.SUCCESS
+
     try:
         if not delivery.payload:
             raise ValueError(
@@ -412,6 +550,7 @@ def send_webhook_request_async(self, event_delivery_id):
         response = WebhookResponse(content=str(e), status=EventDeliveryStatus.FAILED)
         attempt_update(attempt, response)
         delivery_update(delivery=delivery, status=EventDeliveryStatus.FAILED)
+
     observability.report_event_delivery_attempt(attempt)
     clear_successful_delivery(delivery)
 
diff --git a/saleor/webhook/transport/tests/test_utils.py b/saleor/webhook/transport/tests/test_utils.py
index 375675b1ede..e98d888dcb1 100644
--- a/saleor/webhook/transport/tests/test_utils.py
+++ b/saleor/webhook/transport/tests/test_utils.py
@@ -8,9 +8,15 @@
 from celery.utils.threads import LocalStack
 
 from ....core import EventDeliveryStatus
-from ....core.models import EventDeliveryAttempt
+from ....core.models import EventDelivery, EventDeliveryAttempt
 from ...event_types import WebhookEventAsyncType
-from ..utils import WebhookResponse, handle_webhook_retry, send_webhook_using_aws_sqs
+from ..utils import (
+    WebhookResponse,
+    get_delivery_for_webhook,
+    get_multiple_deliveries_for_webhooks,
+    handle_webhook_retry,
+    send_webhook_using_aws_sqs,
+)
 
 
 @pytest.fixture
@@ -187,3 +193,75 @@ def test_send_webhook_using_aws_sqs_with_fifo_queue(mocked_boto3_client):
     # then
     _, send_message_kwargs = mocked_boto3_client.send_message.call_args
     assert send_message_kwargs["MessageGroupId"] == domain
+
+
+def test_get_delivery_for_webhook(event_delivery):
+    # when
+    delivery = get_delivery_for_webhook(event_delivery.pk)
+
+    # then
+    assert delivery == event_delivery
+
+
+def test_get_delivery_for_webhook_invalid_id(caplog):
+    # when
+    invalid_pk = 99999
+    delivery = get_delivery_for_webhook(invalid_pk)
+
+    # then
+    assert delivery is None
+    assert caplog.records[0].message == (f"Event delivery id: {invalid_pk} not found")
+
+
+def test_get_delivery_for_webhook_inactive_webhook(event_delivery, caplog):
+    # given
+    event_delivery.webhook.is_active = False
+    event_delivery.webhook.save(update_fields=["is_active"])
+
+    # when
+    delivery = get_delivery_for_webhook(event_delivery.pk)
+
+    # then
+    assert delivery is None
+    event_delivery.refresh_from_db()
+    assert event_delivery.status == EventDeliveryStatus.FAILED
+    assert caplog.records[0].message == (
+        f"Event delivery id: {event_delivery.pk} webhook is disabled."
+    )
+
+
+def test_get_multiple_deliveries_for_webhooks(event_deliveries):
+    # given
+    all_deliveries = EventDelivery.objects.all()
+    ids = [event_delivery.pk for event_delivery in all_deliveries]
+
+    # when
+    deliveries = get_multiple_deliveries_for_webhooks(ids)
+
+    # then
+    assert len(all_deliveries) == len(deliveries)
+    assert set(deliveries.keys()) == set(ids)
+
+
+def test_get_multiple_deliveries_for_webhooks_with_inactive(
+    any_webhook, event_deliveries
+):
+    # given
+    all_deliveries = EventDelivery.objects.all()
+    ids = [event_delivery.pk for event_delivery in all_deliveries]
+    inactive_delivery = all_deliveries[0]
+
+    any_webhook.is_active = False
+    any_webhook.save(update_fields=["is_active"])
+    inactive_delivery.webhook = any_webhook
+    inactive_delivery.save(update_fields=["webhook"])
+
+    # when
+    deliveries = get_multiple_deliveries_for_webhooks(ids)
+
+    # then
+    assert len(deliveries) == len(all_deliveries) - 1
+    assert set(deliveries.keys()) == set(ids) - {inactive_delivery.pk}
+
+    inactive_delivery.refresh_from_db()
+    assert inactive_delivery.status == EventDeliveryStatus.FAILED
diff --git a/saleor/webhook/transport/utils.py b/saleor/webhook/transport/utils.py
index 12b97052716..c0e37c68721 100644
--- a/saleor/webhook/transport/utils.py
+++ b/saleor/webhook/transport/utils.py
@@ -1,3 +1,4 @@
+import datetime
 import decimal
 import hashlib
 import json
@@ -6,8 +7,9 @@
 from dataclasses import dataclass
 from enum import Enum
 from time import time
-from typing import Any, Callable, Optional
+from typing import Any, Callable, Optional, Union
 from urllib.parse import unquote, urlparse, urlunparse
+from uuid import UUID
 
 import boto3
 from botocore.exceptions import ClientError
@@ -86,6 +88,41 @@ class WebhookResponse:
     duration: float = 0.0
 
 
+class RequestorModelName:
+    # lowercase, as it is returned as such by `model._meta.model_name`
+    APP = "app.app"
+    USER = "account.user"
+
+
+@dataclass
+class DeferredPayloadData:
+    model_name: str
+    object_id: Union[int, UUID]
+    requestor_model_name: Optional[str]
+    requestor_object_id: Optional[Union[int, UUID]]
+    request_time: Optional[datetime.datetime]
+
+
+def prepare_deferred_payload_data(
+    subscribable_object, requestor, request_time
+) -> DeferredPayloadData:
+    model_name = (
+        f"{subscribable_object._meta.app_label}.{subscribable_object._meta.model_name}"
+    )
+    requestor_model_name = (
+        f"{requestor._meta.app_label}.{requestor._meta.model_name}"
+        if requestor
+        else None
+    )
+    return DeferredPayloadData(
+        model_name=model_name,
+        object_id=subscribable_object.pk,
+        request_time=request_time,
+        requestor_model_name=requestor_model_name,
+        requestor_object_id=(requestor.pk if requestor else None),
+    )
+
+
 def generate_cache_key_for_webhook(
     key_data: dict, webhook_url: str, event: str, app_id: int
 ) -> str:
@@ -360,19 +397,39 @@ def handle_webhook_retry(
 
 
 def get_delivery_for_webhook(event_delivery_id) -> Optional["EventDelivery"]:
-    try:
-        delivery = EventDelivery.objects.select_related("payload", "webhook__app").get(
-            id=event_delivery_id
+    delivery = get_multiple_deliveries_for_webhooks([event_delivery_id])
+    return delivery.get(event_delivery_id)
+
+
+def get_multiple_deliveries_for_webhooks(
+    event_delivery_ids,
+) -> dict[int, "EventDelivery"]:
+    deliveries = EventDelivery.objects.select_related("payload", "webhook__app").filter(
+        id__in=event_delivery_ids
+    )
+
+    active_deliveries = {}
+    inactive_delivery_ids = set()
+
+    not_found_delivery_ids = set(event_delivery_ids) - set(
+        delivery.pk for delivery in deliveries
+    )
+    for not_found_delivery_id in not_found_delivery_ids:
+        logger.warning("Event delivery id: %r not found", not_found_delivery_id)
+
+    for delivery in deliveries:
+        if delivery.webhook.is_active:
+            active_deliveries[delivery.pk] = delivery
+        else:
+            logger.info("Event delivery id: %r webhook is disabled.", delivery.pk)
+            inactive_delivery_ids.add(delivery.pk)
+
+    if inactive_delivery_ids:
+        EventDelivery.objects.filter(id__in=inactive_delivery_ids).update(
+            status=EventDeliveryStatus.FAILED
         )
-    except EventDelivery.DoesNotExist:
-        logger.warning("Event delivery id: %r not found", event_delivery_id)
-        return None
 
-    if not delivery.webhook.is_active:
-        delivery_update(delivery=delivery, status=EventDeliveryStatus.FAILED)
-        logger.info("Event delivery id: %r webhook is disabled.", event_delivery_id)
-        return None
-    return delivery
+    return active_deliveries
 
 
 @contextmanager


Code review comments:
diff_hunk: @@ -282,32 +307,69 @@ def trigger_webhooks_async_for_multiple_objects(
                 )
 
             with allow_writer():
-                # Use transaction to ensure EventPayload and EventDelivery are created together, preventing inconsistent DB state.
+                # Use transaction to ensure EventPayload and EventDelivery are created
+                # together, preventing inconsistent DB state.
                 with transaction.atomic():
                     payload = EventPayload.objects.create_with_payload_file(data)
                     deliveries.extend(
                         create_event_delivery_list_for_webhooks(
-                            webhooks=regular_webhooks,
+                            webhooks=legacy_webhooks,
                             event_payload=payload,
                             event_type=event_type,
                         )
                     )
 
     if subscription_webhooks:
-        deliveries.extend(
-            create_deliveries_for_multiple_subscription_objects(
-                event_type=event_type,
-                subscribable_objects=[
-                    webhook_payload_data.subscribable_object
-                    for webhook_payload_data in webhook_payloads_data
-                ],
-                webhooks=subscription_webhooks,
-                requestor=requestor,
-                allow_replica=allow_replica,
-                pre_save_payloads=pre_save_payloads,
-                request_time=request_time,
+        subscribable_objects = [
+            webhook_payload_data.subscribable_object
+            for webhook_payload_data in webhook_payloads_data
+        ]
+        if is_deferred_payload:
+            _deferred_deliveries_to_create = []
+            for subscribable_object in subscribable_objects:
+                for webhook in subscription_webhooks:
+                    delivery = EventDelivery(
+                        status=EventDeliveryStatus.PENDING,
+                        event_type=event_type,
+                        webhook=webhook,
+                    )
+                    _deferred_deliveries_to_create.append(delivery)
+                    deferred_payload_data = prepare_deferred_payload_data(
+                        subscribable_object, requestor, request_time
+                    )
+                    deferred_deliveries_per_object[subscribable_object.pk].append(
+                        (delivery, deferred_payload_data)
+                    )
+
+            EventDelivery.objects.bulk_create(_deferred_deliveries_to_create)
+        else:
+            deliveries.extend(
+                create_deliveries_for_multiple_subscription_objects(
+                    event_type=event_type,
+                    subscribable_objects=subscribable_objects,
+                    webhooks=subscription_webhooks,
+                    requestor=requestor,
+                    allow_replica=allow_replica,
+                    pre_save_payloads=pre_save_payloads,
+                    request_time=request_time,
+                )
             )
-        )
+
+    if deferred_deliveries_per_object:
+        for _, deferred_deliveries in deferred_deliveries_per_object.items():
+            event_delivery_ids = [delivery.id for delivery, _ in deferred_deliveries]
+
+            # Trigger deferred payload generation task for each subscribable object.
+            generate_deferred_payloads.apply_async(
+                kwargs={
+                    "event_delivery_ids": event_delivery_ids,
+                    "deferred_payload_data": asdict(deferred_payload_data),
+                    "queue": queue,
+                },
+                queue=queue or settings.WEBHOOK_CELERY_QUEUE_NAME,
path: saleor/webhook/transport/asynchronous/transport.py
text: @user1:
This means: "use the same celery queue as the one for sending the webhooks". 
And IMO this is wrong as it means, that single queue will be shared between sending webhooks & building the webhook payloads. Few downsizes of using the same queue:
- in case of high traffic and a huge amount of webhooks to send in the queue, the building payload could be delayed, which means that the body could be changed, or object could not exist anymore. 
- sending webhooks could be delayed as right now the same queue also builds the payload. So assuming that you have checkout with sync webhooks to call, this could take a few seconds (max 20 secs) to process single task. It means, that this could downgrade the maximal number of webhooks that we're able to send by `WEBHOOK_CELERY_QUEUE_NAME`

IMHO, generating payloads should have its own queue, as we want to build a payload as fast as possible. Or at least using the default one, but using the same as the one for sending webhooks could cause a problem in case of huge amount of webhooks on queue. 

@user2:
>in case of high traffic and a huge amount of webhooks to send in the queue, the building payload could be delayed, which means that the body could be changed, or object could not exist anymore.

That should not be an issue as there is similarly little value in delivering long-obsolete data.

>sending webhooks could be delayed as right now the same queue also builds the payload. So assuming that you have checkout with sync webhooks to call, this could take a few seconds (max 20 secs) to process single task.

That is a valid concern, generating events would happen at the expense of our webhook delivery capacity. We could use the default queue for payload generation until we implement the delivery fairness improvements we're working on.

@user3:
I also agree to use the default queue for this task. 
I don't see any more objections.

@author:
Changed to use default queue for payload generation.

diff_hunk: @@ -282,32 +307,69 @@ def trigger_webhooks_async_for_multiple_objects(
                 )
 
             with allow_writer():
-                # Use transaction to ensure EventPayload and EventDelivery are created together, preventing inconsistent DB state.
+                # Use transaction to ensure EventPayload and EventDelivery are created
+                # together, preventing inconsistent DB state.
                 with transaction.atomic():
                     payload = EventPayload.objects.create_with_payload_file(data)
                     deliveries.extend(
                         create_event_delivery_list_for_webhooks(
-                            webhooks=regular_webhooks,
+                            webhooks=legacy_webhooks,
                             event_payload=payload,
                             event_type=event_type,
                         )
                     )
 
     if subscription_webhooks:
-        deliveries.extend(
-            create_deliveries_for_multiple_subscription_objects(
-                event_type=event_type,
-                subscribable_objects=[
-                    webhook_payload_data.subscribable_object
-                    for webhook_payload_data in webhook_payloads_data
-                ],
-                webhooks=subscription_webhooks,
-                requestor=requestor,
-                allow_replica=allow_replica,
-                pre_save_payloads=pre_save_payloads,
-                request_time=request_time,
+        subscribable_objects = [
+            webhook_payload_data.subscribable_object
+            for webhook_payload_data in webhook_payloads_data
+        ]
+        if is_deferred_payload:
+            _deferred_deliveries_to_create = []
+            for subscribable_object in subscribable_objects:
path: saleor/webhook/transport/asynchronous/transport.py
text: @user1:
Can we move this logic to separate function? Like we already have for `create_deliveries_for_multiple_subscription_objects(`. It would be more readable IMO.

diff_hunk: @@ -364,6 +426,81 @@ def trigger_webhooks_async(
     )
 
 
+@app.task(
+    queue=settings.WEBHOOK_CELERY_QUEUE_NAME,
+    bind=True,
+)
+def generate_deferred_payloads(
+    self,
+    event_delivery_ids: list,
+    deferred_payload_data: dict,
+    queue: Optional[str] = None,
+):
+    deliveries = list(get_multiple_deliveries_for_webhooks(event_delivery_ids).values())
+    args_obj = DeferredPayloadData(**deferred_payload_data)
+    requestor = None
+    if args_obj.requestor_object_id and args_obj.requestor_model_name in (
+        RequestorModelName.APP,
+        RequestorModelName.USER,
+    ):
+        model = apps.get_model(args_obj.requestor_model_name)
+        requestor = model.objects.filter(pk=args_obj.requestor_object_id).first()
+
+    subscribable_object = (
+        apps.get_model(args_obj.model_name)
+        .objects.filter(pk=args_obj.object_id)
+        .first()
+    )
+    if not subscribable_object:
+        EventDelivery.objects.filter(pk__in=event_delivery_ids).update(
+            status=EventDeliveryStatus.FAILED
+        )
+        return
+
+    for delivery in deliveries:
+        event_type = delivery.event_type
+        webhook = delivery.webhook
+        request = initialize_request(
+            requestor,
+            event_type in WebhookEventSyncType.ALL,
+            event_type=event_type,
+            allow_replica=True,
+            request_time=args_obj.request_time,
+        )
+        data_promise = generate_payload_promise_from_subscription(
+            event_type=event_type,
+            subscribable_object=subscribable_object,
+            subscription_query=webhook.subscription_query,  # type: ignore
+            request=request,
+            app=webhook.app,
+        )
+
+        if data_promise:
+            data = data_promise.get()
+            if data:
+                data_json = json.dumps({**data})
+                with allow_writer():
+                    event_payload = EventPayload.objects.create_with_payload_file(
path: saleor/webhook/transport/asynchronous/transport.py
text: @user1:
In each place, payload generation is wrapped in transaction.atomic(). Shouldn't we also do this here?
Also this means, that for each delivery in deliveries, we are calling 2 or 3 db queries. Can't we make it in bulk? For example, using: `bulk_create_with_payload_files` ?