Sync vs Async коммуникация между сервисами

<?php

declare(strict_types=1);

namespace App\Checkout\Sync;

use Symfony\Contracts\HttpClient\HttpClientInterface;
use Symfony\Contracts\HttpClient\Exception\HttpExceptionInterface;
use Psr\Log\LoggerInterface;

/**
 * Sync orchestrator for checkout.
 *
 * Calls inventory, payment, notification services via HTTP.
 * On failure - runs compensation (saga).
 */
final class SyncCheckoutOrchestrator
{
    public function __construct(
        private readonly HttpClientInterface $inventory,
        private readonly HttpClientInterface $payment,
        private readonly HttpClientInterface $notification,
        private readonly OrderRepository $orders,
        private readonly LoggerInterface $logger,
    ) {}

    public function checkout(CheckoutCommand $cmd): CheckoutResult
    {
        $orderId = $this->orders->createPending($cmd);
        $reservationId = null;
        $paymentId = null;

        try {
            // Step 1: reserve inventory (timeout 3s)
            $resp = $this->inventory->request('POST', '/reservations', [
                'json' => ['order_id' => $orderId, 'items' => $cmd->items],
                'timeout' => 3.0,
                'headers' => ['Idempotency-Key' => $orderId],
            ]);
            $reservationId = $resp->toArray()['reservation_id'];

            // Step 2: charge payment (timeout 5s)
            $resp = $this->payment->request('POST', '/charges', [
                'json' => [
                    'order_id' => $orderId,
                    'amount' => $cmd->totalCents,
                    'currency' => $cmd->currency,
                ],
                'timeout' => 5.0,
                'headers' => ['Idempotency-Key' => $orderId],
            ]);
            $paymentId = $resp->toArray()['payment_id'];

            $this->orders->markPaid($orderId, $paymentId, $reservationId);

            // Step 3: notification is fire-and-forget even in sync flow
            $this->notification->request('POST', '/emails', [
                'json' => ['template' => 'order_confirmed', 'order_id' => $orderId],
                'timeout' => 2.0,
            ]);

            return CheckoutResult::success($orderId);
        } catch (HttpExceptionInterface $e) {
            $this->logger->error('checkout failed', [
                'order_id' => $orderId,
                'step' => $paymentId ? 'notify' : ($reservationId ? 'pay' : 'reserve'),
                'error' => $e->getMessage(),
            ]);

            // Compensation: release reservation if taken
            if ($reservationId !== null && $paymentId === null) {
                $this->safeRelease($reservationId);
            }
            // If payment succeeded but something after failed - still OK for user
            // Notification failure is non-fatal.

            $this->orders->markFailed($orderId, $e->getMessage());
            return CheckoutResult::failure($orderId, $e->getMessage());
        }
    }

    private function safeRelease(string $reservationId): void
    {
        try {
            $this->inventory->request('DELETE', "/reservations/$reservationId", [
                'timeout' => 3.0,
            ]);
        } catch (\Throwable $e) {
            // Compensation failed - send to DLQ / alert
            $this->logger->critical('compensation failed', [
                'reservation_id' => $reservationId,
                'error' => $e->getMessage(),
            ]);
        }
    }
}

package checkout

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"time"

	"log/slog"
)

// SyncOrchestrator runs checkout steps sequentially via HTTP.
type SyncOrchestrator struct {
	client     *http.Client
	inventory  string
	payment    string
	notifier   string
	orders     OrderRepo
	logger     *slog.Logger
}

func NewSyncOrchestrator(inv, pay, notif string, repo OrderRepo, log *slog.Logger) *SyncOrchestrator {
	return &SyncOrchestrator{
		client: &http.Client{
			Timeout: 10 * time.Second,
			Transport: &http.Transport{
				MaxIdleConnsPerHost: 32,
				IdleConnTimeout:     90 * time.Second,
			},
		},
		inventory: inv,
		payment:   pay,
		notifier:  notif,
		orders:    repo,
		logger:    log,
	}
}

func (o *SyncOrchestrator) Checkout(ctx context.Context, cmd CheckoutCommand) (Result, error) {
	orderID, err := o.orders.CreatePending(ctx, cmd)
	if err != nil {
		return Result{}, fmt.Errorf("create pending: %w", err)
	}

	reservationID, err := o.reserve(ctx, orderID, cmd)
	if err != nil {
		o.orders.MarkFailed(ctx, orderID, err.Error())
		return Result{OrderID: orderID, Status: "failed"}, err
	}

	paymentID, err := o.charge(ctx, orderID, cmd)
	if err != nil {
		// Compensation: release reservation
		o.safeRelease(ctx, reservationID)
		o.orders.MarkFailed(ctx, orderID, err.Error())
		return Result{OrderID: orderID, Status: "failed"}, err
	}

	if err := o.orders.MarkPaid(ctx, orderID, paymentID, reservationID); err != nil {
		o.logger.Error("mark paid failed", "order_id", orderID, "err", err)
	}

	// Notification is best-effort
	if err := o.notify(ctx, orderID); err != nil {
		o.logger.Warn("notify failed", "order_id", orderID, "err", err)
	}

	return Result{OrderID: orderID, Status: "success"}, nil
}

func (o *SyncOrchestrator) reserve(ctx context.Context, orderID string, cmd CheckoutCommand) (string, error) {
	ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()

	body, _ := json.Marshal(map[string]any{
		"order_id": orderID,
		"items":    cmd.Items,
	})
	req, _ := http.NewRequestWithContext(ctx, "POST", o.inventory+"/reservations", bytes.NewReader(body))
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Idempotency-Key", orderID)

	resp, err := o.client.Do(req)
	if err != nil {
		return "", fmt.Errorf("inventory: %w", err)
	}
	defer resp.Body.Close()
	if resp.StatusCode >= 400 {
		return "", fmt.Errorf("inventory status %d", resp.StatusCode)
	}
	var out struct {
		ReservationID string `json:"reservation_id"`
	}
	if err := json.NewDecoder(resp.Body).Decode(&out); err != nil {
		return "", err
	}
	return out.ReservationID, nil
}

// charge, notify, safeRelease - similar pattern omitted for brevity.

using System.Net.Http.Json;
using System.Text.Json.Serialization;
using Microsoft.Extensions.Logging;

public sealed record CheckoutCommand(
    string UserId,
    IReadOnlyList<OrderItem> Items,
    long TotalCents,
    string Currency);

public sealed record CheckoutResult(string OrderId, string Status, string? Error = null)
{
    public static CheckoutResult Success(string orderId) => new(orderId, "success");
    public static CheckoutResult Failure(string orderId, string error) => new(orderId, "failed", error);
}

/// <summary>
/// Sync orchestrator for checkout.
///
/// Calls inventory, payment and notification services over HTTP.
/// On failure — runs compensation (saga).
/// </summary>
public sealed class SyncCheckoutOrchestrator(
    IHttpClientFactory httpClientFactory,
    IOrderRepository orders,
    ILogger<SyncCheckoutOrchestrator> logger)
{
    private sealed record ReservationResponse(
        [property: JsonPropertyName("reservation_id")] string ReservationId);

    private sealed record ChargeResponse(
        [property: JsonPropertyName("payment_id")] string PaymentId);

    public async Task<CheckoutResult> CheckoutAsync(CheckoutCommand cmd, CancellationToken ct = default)
    {
        var orderId = await orders.CreatePendingAsync(cmd, ct);
        string? reservationId = null;
        string? paymentId = null;

        try
        {
            // Step 1: reserve inventory (timeout 3s)
            reservationId = await ReserveAsync(orderId, cmd, ct);

            // Step 2: charge payment (timeout 5s)
            paymentId = await ChargeAsync(orderId, cmd, ct);

            await orders.MarkPaidAsync(orderId, paymentId, reservationId, ct);

            // Step 3: notification is fire-and-forget even in the sync flow
            await NotifyAsync(orderId, ct);

            return CheckoutResult.Success(orderId);
        }
        catch (Exception ex) when (ex is HttpRequestException or TaskCanceledException)
        {
            var step = paymentId is not null ? "notify"
                : reservationId is not null ? "pay"
                : "reserve";

            logger.LogError(ex, "checkout failed: order {OrderId}, step {Step}", orderId, step);

            // Compensation: release the reservation if it was taken but never paid.
            // If payment succeeded, the order stands — notification failure is non-fatal.
            if (reservationId is not null && paymentId is null)
            {
                await SafeReleaseAsync(reservationId, ct);
            }

            await orders.MarkFailedAsync(orderId, ex.Message, ct);
            return CheckoutResult.Failure(orderId, ex.Message);
        }
    }

    private async Task<string> ReserveAsync(string orderId, CheckoutCommand cmd, CancellationToken ct)
    {
        using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
        cts.CancelAfter(TimeSpan.FromSeconds(3));

        var client = httpClientFactory.CreateClient("inventory");
        using var request = new HttpRequestMessage(HttpMethod.Post, "/reservations")
        {
            Content = JsonContent.Create(new { order_id = orderId, items = cmd.Items }),
        };
        request.Headers.Add("Idempotency-Key", orderId);

        using var response = await client.SendAsync(request, cts.Token);
        response.EnsureSuccessStatusCode();

        var body = await response.Content.ReadFromJsonAsync<ReservationResponse>(cts.Token);
        return body!.ReservationId;
    }

    private async Task<string> ChargeAsync(string orderId, CheckoutCommand cmd, CancellationToken ct)
    {
        using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
        cts.CancelAfter(TimeSpan.FromSeconds(5));

        var client = httpClientFactory.CreateClient("payment");
        using var request = new HttpRequestMessage(HttpMethod.Post, "/charges")
        {
            Content = JsonContent.Create(new
            {
                order_id = orderId,
                amount = cmd.TotalCents,
                currency = cmd.Currency,
            }),
        };
        request.Headers.Add("Idempotency-Key", orderId);

        using var response = await client.SendAsync(request, cts.Token);
        response.EnsureSuccessStatusCode();

        var body = await response.Content.ReadFromJsonAsync<ChargeResponse>(cts.Token);
        return body!.PaymentId;
    }

    private async Task NotifyAsync(string orderId, CancellationToken ct)
    {
        using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
        cts.CancelAfter(TimeSpan.FromSeconds(2));

        var client = httpClientFactory.CreateClient("notification");
        using var response = await client.PostAsJsonAsync(
            "/emails",
            new { template = "order_confirmed", order_id = orderId },
            cts.Token);

        response.EnsureSuccessStatusCode();
    }

    private async Task SafeReleaseAsync(string reservationId, CancellationToken ct)
    {
        try
        {
            using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
            cts.CancelAfter(TimeSpan.FromSeconds(3));

            var client = httpClientFactory.CreateClient("inventory");
            using var response = await client.DeleteAsync($"/reservations/{reservationId}", cts.Token);
            response.EnsureSuccessStatusCode();
        }
        catch (Exception ex)
        {
            // Compensation failed — send to DLQ / alert
            logger.LogCritical(ex, "compensation failed: reservation {ReservationId}", reservationId);
        }
    }
}

import logging
from dataclasses import dataclass
from typing import Any, Self

import httpx

logger = logging.getLogger(__name__)

@dataclass(frozen=True, slots=True)
class CheckoutCommand:
    user_id: str
    items: list[dict[str, Any]]
    total_cents: int
    currency: str

@dataclass(frozen=True, slots=True)
class CheckoutResult:
    order_id: str
    status: str
    error: str | None = None

    @classmethod
    def success(cls, order_id: str) -> Self:
        return cls(order_id, "success")

    @classmethod
    def failure(cls, order_id: str, error: str) -> Self:
        return cls(order_id, "failed", error)

class SyncCheckoutOrchestrator:
    """Sync orchestrator for checkout.

    Calls inventory, payment and notification services over HTTP.
    On failure — runs compensation (saga).
    """

    def __init__(
        self,
        inventory: httpx.AsyncClient,
        payment: httpx.AsyncClient,
        notification: httpx.AsyncClient,
        orders: OrderRepository,
    ) -> None:
        self._inventory = inventory
        self._payment = payment
        self._notification = notification
        self._orders = orders

    async def checkout(self, cmd: CheckoutCommand) -> CheckoutResult:
        order_id = await self._orders.create_pending(cmd)
        reservation_id: str | None = None
        payment_id: str | None = None

        try:
            # Step 1: reserve inventory (timeout 3s)
            response = await self._inventory.post(
                "/reservations",
                json={"order_id": order_id, "items": cmd.items},
                headers={"Idempotency-Key": order_id},
                timeout=3.0,
            )
            response.raise_for_status()
            reservation_id = response.json()["reservation_id"]

            # Step 2: charge payment (timeout 5s)
            response = await self._payment.post(
                "/charges",
                json={
                    "order_id": order_id,
                    "amount": cmd.total_cents,
                    "currency": cmd.currency,
                },
                headers={"Idempotency-Key": order_id},
                timeout=5.0,
            )
            response.raise_for_status()
            payment_id = response.json()["payment_id"]

            await self._orders.mark_paid(order_id, payment_id, reservation_id)

            # Step 3: notification is fire-and-forget even in the sync flow
            response = await self._notification.post(
                "/emails",
                json={"template": "order_confirmed", "order_id": order_id},
                timeout=2.0,
            )
            response.raise_for_status()

            return CheckoutResult.success(order_id)
        except httpx.HTTPError as exc:
            step = "notify" if payment_id else "pay" if reservation_id else "reserve"
            logger.error(
                "checkout failed",
                extra={"order_id": order_id, "step": step, "error": str(exc)},
            )

            # Compensation: release the reservation if it was taken but never paid.
            # If payment succeeded, the order stands — notification failure is non-fatal.
            if reservation_id is not None and payment_id is None:
                await self._safe_release(reservation_id)

            await self._orders.mark_failed(order_id, str(exc))
            return CheckoutResult.failure(order_id, str(exc))

    async def _safe_release(self, reservation_id: str) -> None:
        try:
            response = await self._inventory.delete(
                f"/reservations/{reservation_id}", timeout=3.0
            )
            response.raise_for_status()
        except httpx.HTTPError as exc:
            # Compensation failed — send to DLQ / alert
            logger.critical(
                "compensation failed",
                extra={"reservation_id": reservation_id, "error": str(exc)},
            )

<?php

declare(strict_types=1);

namespace App\Checkout\Async;

use Symfony\Component\Messenger\MessageBusInterface;
use Symfony\Component\Messenger\Attribute\AsMessageHandler;
use Psr\Log\LoggerInterface;

// ============ Events ============

final readonly class OrderCreated
{
    public function __construct(
        public string $orderId,
        public string $userId,
        public array $items,
        public int $totalCents,
        public string $currency,
    ) {}
}

final readonly class InventoryReserved
{
    public function __construct(
        public string $orderId,
        public string $reservationId,
    ) {}
}

final readonly class InventoryReservationFailed
{
    public function __construct(
        public string $orderId,
        public string $reason,
    ) {}
}

final readonly class PaymentCharged
{
    public function __construct(
        public string $orderId,
        public string $paymentId,
    ) {}
}

// ============ Handlers (choreography) ============

#[AsMessageHandler]
final class OnOrderCreated
{
    public function __construct(
        private readonly InventoryService $inventory,
        private readonly MessageBusInterface $bus,
        private readonly LoggerInterface $logger,
    ) {}

    public function __invoke(OrderCreated $e): void
    {
        // Idempotency: if already reserved for this order, skip
        if ($existing = $this->inventory->findReservation($e->orderId)) {
            $this->bus->dispatch(new InventoryReserved($e->orderId, $existing));
            return;
        }

        try {
            $reservationId = $this->inventory->reserve($e->orderId, $e->items);
            $this->bus->dispatch(new InventoryReserved($e->orderId, $reservationId));
        } catch (InsufficientStockException $ex) {
            $this->bus->dispatch(new InventoryReservationFailed($e->orderId, $ex->getMessage()));
        }
    }
}

#[AsMessageHandler]
final class OnInventoryReserved
{
    public function __construct(
        private readonly PaymentService $payment,
        private readonly MessageBusInterface $bus,
    ) {}

    public function __invoke(InventoryReserved $e): void
    {
        $order = $this->payment->loadOrder($e->orderId);

        // Idempotency check at handler level
        if ($paymentId = $this->payment->findCharge($e->orderId)) {
            $this->bus->dispatch(new PaymentCharged($e->orderId, $paymentId));
            return;
        }

        $paymentId = $this->payment->charge(
            orderId: $e->orderId,
            amount: $order->totalCents,
            currency: $order->currency,
            idempotencyKey: $e->orderId,
        );

        $this->bus->dispatch(new PaymentCharged($e->orderId, $paymentId));
    }
}

#[AsMessageHandler]
final class OnInventoryReservationFailed
{
    public function __construct(
        private readonly OrderRepository $orders,
        private readonly NotifierService $notifier,
    ) {}

    public function __invoke(InventoryReservationFailed $e): void
    {
        // No compensation needed - no payment yet.
        $this->orders->markFailed($e->orderId, $e->reason);
        $this->notifier->notifyUser($e->orderId, 'order_failed_out_of_stock');
    }
}

package checkout

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log/slog"

	"github.com/segmentio/kafka-go"
)

// OrderCreated is emitted by OrderService after persisting a pending order.
type OrderCreated struct {
	OrderID    string  `json:"order_id"`
	UserID     string  `json:"user_id"`
	Items      []Item  `json:"items"`
	TotalCents int64   `json:"total_cents"`
	Currency   string  `json:"currency"`
}

type InventoryReserved struct {
	OrderID       string `json:"order_id"`
	ReservationID string `json:"reservation_id"`
}

// InventoryConsumer processes OrderCreated -> reserves stock -> emits next event.
type InventoryConsumer struct {
	reader    *kafka.Reader
	writer    *kafka.Writer
	inventory InventorySvc
	idem      IdempotencyStore
	logger    *slog.Logger
}

func (c *InventoryConsumer) Run(ctx context.Context) error {
	for {
		msg, err := c.reader.FetchMessage(ctx)
		if err != nil {
			return err
		}

		var evt OrderCreated
		if err := json.Unmarshal(msg.Value, &evt); err != nil {
			c.logger.Error("bad payload", "err", err)
			c.reader.CommitMessages(ctx, msg) // skip poison message
			continue
		}

		// Idempotency: have we processed this order already?
		seen, err := c.idem.Seen(ctx, "inv:"+evt.OrderID)
		if err != nil {
			c.logger.Error("idempotency check", "err", err)
			continue // do not commit - retry
		}
		if seen {
			c.reader.CommitMessages(ctx, msg)
			continue
		}

		if err := c.handle(ctx, evt); err != nil {
			if errors.Is(err, ErrRetryable) {
				c.logger.Warn("retryable", "order_id", evt.OrderID, "err", err)
				continue // leave offset - Kafka will redeliver
			}
			// Non-retryable: emit failure event
			c.emitFailed(ctx, evt.OrderID, err.Error())
		}

		c.idem.Mark(ctx, "inv:"+evt.OrderID)
		if err := c.reader.CommitMessages(ctx, msg); err != nil {
			c.logger.Error("commit", "err", err)
		}
	}
}

func (c *InventoryConsumer) handle(ctx context.Context, evt OrderCreated) error {
	resID, err := c.inventory.Reserve(ctx, evt.OrderID, evt.Items)
	if err != nil {
		return fmt.Errorf("reserve: %w", err)
	}
	return c.emit(ctx, "inventory.reserved", InventoryReserved{
		OrderID:       evt.OrderID,
		ReservationID: resID,
	})
}

func (c *InventoryConsumer) emit(ctx context.Context, topic string, v any) error {
	body, _ := json.Marshal(v)
	return c.writer.WriteMessages(ctx, kafka.Message{
		Topic: topic,
		Key:   []byte(extractOrderID(v)),
		Value: body,
	})
}

using MassTransit;
using Microsoft.Extensions.Logging;

// ============ Events ============

public sealed record OrderCreated(
    string OrderId,
    string UserId,
    IReadOnlyList<OrderItem> Items,
    long TotalCents,
    string Currency);

public sealed record InventoryReserved(string OrderId, string ReservationId);

public sealed record InventoryReservationFailed(string OrderId, string Reason);

public sealed record PaymentCharged(string OrderId, string PaymentId);

// ============ Consumers (choreography) ============
// MassTransit consumers play the role of Symfony Messenger handlers: each one
// reacts to a single event and publishes the next; no service knows the others.

public sealed class OnOrderCreated(
    IInventoryService inventory,
    ILogger<OnOrderCreated> logger) : IConsumer<OrderCreated>
{
    public async Task Consume(ConsumeContext<OrderCreated> context)
    {
        var e = context.Message;
        var ct = context.CancellationToken;

        // Idempotency: if already reserved for this order, just re-emit
        var existing = await inventory.FindReservationAsync(e.OrderId, ct);
        if (existing is not null)
        {
            await context.Publish(new InventoryReserved(e.OrderId, existing), ct);
            return;
        }

        try
        {
            var reservationId = await inventory.ReserveAsync(e.OrderId, e.Items, ct);
            await context.Publish(new InventoryReserved(e.OrderId, reservationId), ct);
        }
        catch (InsufficientStockException ex)
        {
            logger.LogWarning(ex, "out of stock for order {OrderId}", e.OrderId);
            await context.Publish(new InventoryReservationFailed(e.OrderId, ex.Message), ct);
        }
    }
}

public sealed class OnInventoryReserved(IPaymentService payment) : IConsumer<InventoryReserved>
{
    public async Task Consume(ConsumeContext<InventoryReserved> context)
    {
        var e = context.Message;
        var ct = context.CancellationToken;

        // Idempotency check at handler level
        var existing = await payment.FindChargeAsync(e.OrderId, ct);
        if (existing is not null)
        {
            await context.Publish(new PaymentCharged(e.OrderId, existing), ct);
            return;
        }

        var order = await payment.LoadOrderAsync(e.OrderId, ct);
        var paymentId = await payment.ChargeAsync(
            orderId: e.OrderId,
            amountCents: order.TotalCents,
            currency: order.Currency,
            idempotencyKey: e.OrderId,
            ct: ct);

        await context.Publish(new PaymentCharged(e.OrderId, paymentId), ct);
    }
}

public sealed class OnInventoryReservationFailed(
    IOrderRepository orders,
    INotifierService notifier) : IConsumer<InventoryReservationFailed>
{
    public async Task Consume(ConsumeContext<InventoryReservationFailed> context)
    {
        var e = context.Message;
        var ct = context.CancellationToken;

        // No compensation needed — no payment happened yet
        await orders.MarkFailedAsync(e.OrderId, e.Reason, ct);
        await notifier.NotifyUserAsync(e.OrderId, "order_failed_out_of_stock", ct);
    }
}

import logging
from dataclasses import dataclass
from typing import Any

from faststream.kafka import KafkaBroker

logger = logging.getLogger(__name__)

# ============ Events ============

@dataclass(frozen=True, slots=True)
class OrderCreated:
    order_id: str
    user_id: str
    items: list[dict[str, Any]]
    total_cents: int
    currency: str

@dataclass(frozen=True, slots=True)
class InventoryReserved:
    order_id: str
    reservation_id: str

@dataclass(frozen=True, slots=True)
class InventoryReservationFailed:
    order_id: str
    reason: str

@dataclass(frozen=True, slots=True)
class PaymentCharged:
    order_id: str
    payment_id: str

# ============ Handlers (choreography) ============

def register_handlers(
    broker: KafkaBroker,
    inventory: InventoryService,
    payment: PaymentService,
    orders: OrderRepository,
    notifier: NotifierService,
) -> None:
    """Wire choreography handlers.

    Each handler reacts to exactly one event and publishes the next one;
    no service knows about the others. The Kafka key is the order id, so
    all events for one order land on the same partition and stay ordered.
    """

    @broker.subscriber("order.created", group_id="inventory")
    @broker.publisher("inventory.reserved")
    async def on_order_created(event: OrderCreated) -> InventoryReserved | None:
        # Idempotency: if already reserved for this order, just re-emit
        existing = await inventory.find_reservation(event.order_id)
        if existing is not None:
            return InventoryReserved(event.order_id, existing)

        try:
            reservation_id = await inventory.reserve(event.order_id, event.items)
        except InsufficientStockError as exc:
            logger.warning(
                "out of stock", extra={"order_id": event.order_id, "error": str(exc)}
            )
            await broker.publish(
                InventoryReservationFailed(event.order_id, str(exc)),
                topic="inventory.reservation_failed",
                key=event.order_id.encode(),
            )
            return None

        return InventoryReserved(event.order_id, reservation_id)

    @broker.subscriber("inventory.reserved", group_id="payment")
    @broker.publisher("payment.charged")
    async def on_inventory_reserved(event: InventoryReserved) -> PaymentCharged:
        # Idempotency check at handler level
        existing = await payment.find_charge(event.order_id)
        if existing is not None:
            return PaymentCharged(event.order_id, existing)

        order = await payment.load_order(event.order_id)
        payment_id = await payment.charge(
            order_id=event.order_id,
            amount_cents=order.total_cents,
            currency=order.currency,
            idempotency_key=event.order_id,
        )
        return PaymentCharged(event.order_id, payment_id)

    @broker.subscriber("inventory.reservation_failed", group_id="orders")
    async def on_reservation_failed(event: InventoryReservationFailed) -> None:
        # No compensation needed — no payment happened yet
        await orders.mark_failed(event.order_id, event.reason)
        await notifier.notify_user(event.order_id, "order_failed_out_of_stock")