services:
  postgres:
    image: postgres:latest
    container_name: postgres
    environment:
      - TZ
      - POSTGRES_USER=admin
      - POSTGRES_PASSWORD=admin
      - POSTGRES_DB=shipping
    volumes:
      - ./docker_data/pgdata:/var/lib/postgresql/data
    ports:
      - "5432:5432"
    restart: always
    networks:
      - docker-web

  rabbitmq:
    image: rabbitmq:3-management
    container_name: rabbitmq
    ports:
      - "15672:15672"
      - "5672:5672"
    restart: always
    volumes:
      - ./docker_data/rabbitmq:/var/lib/rabbitmq
    networks:
      - docker-web

  seq:
    image: datalust/seq:2024.3
    container_name: seq
    restart: always
    environment:
      - ACCEPT_EULA=Y
    volumes:
      - ./docker_data/seq:/data
    ports:
      - "5341:5341"
      - "8081:80"
    networks:
      - docker-web

networks:
  docker-web:
    driver: bridge

dotnet add package MassTransit.RabbitMQ

var rabbitMqConfiguration = configuration.GetSection(nameof(RabbitMQConfiguration)).Get<RabbitMQConfiguration>()!;

services.AddMassTransit(busConfig =>
{
    busConfig.SetKebabCaseEndpointNameFormatter();
    busConfig.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(new Uri(rabbitMqConfiguration.Host), h =>
        {
            h.Username(rabbitMqConfiguration.Username);
            h.Password(rabbitMqConfiguration.Password);
        });
        cfg.ConfigureEndpoints(context);
    });
});

{
  "RabbitMQConfiguration": {
    "Host": "amqp://127.0.0.1:5672",
    "Username": "guest",
    "Password": "guest"
  }
}

public async Task<ErrorOr<CreateShipmentResponse>> Handle(CreateShipmentCommand request, CancellationToken cancellationToken)
{
    var shipmentAlreadyExists = await context.Shipments.Where(s => s.OrderId == request.OrderId).AnyAsync(cancellationToken);
    if (shipmentAlreadyExists)
    {
        logger.LogInformation("Shipment for order '{OrderId}' is already created", request.OrderId);
        return Error.Failure($"Shipment for order '{request.OrderId}' is already created");
    }

    var shipmentNumber = new Faker().Commerce.Ean8();
    var shipment = CreateShipment(request, shipmentNumber);
    context.Shipments.Add(shipment);

    var shipmentCreatedEvent = CreateShipmentCreatedEvent(shipment);
    await publishEndpoint.Publish(shipmentCreatedEvent, cancellationToken);

    await context.SaveChangesAsync(cancellationToken);
    logger.LogInformation("Created shipment: {@Shipment}", shipment);

    return new CreateShipmentResponse(shipment.Number);
}

services.AddMassTransit(busConfig =>
{
    busConfig.AddEntityFrameworkOutbox<EfCoreDbContext>(o =>
    {
        o.QueryDelay = TimeSpan.FromSeconds(30);
        o.UsePostgres().UseBusOutbox();
    });

    busConfig.SetKebabCaseEndpointNameFormatter();

    busConfig.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(new Uri(rabbitMqConfiguration.Host), h =>
        {
            h.Username(rabbitMqConfiguration.Username);
            h.Password(rabbitMqConfiguration.Password);
        });

        cfg.UseMessageRetry(r => r.Exponential(10, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(60), TimeSpan.FromSeconds(5)));
        cfg.ConfigureEndpoints(context);
    });
});

public class ShipmentCreatedConsumer : IConsumer<ShipmentCreatedEvent>
{
    private readonly IPublishEndpoint publishEndpoint;
    private readonly ILogger<ShipmentCreatedConsumer> logger;

    public ShipmentCreatedConsumer(IPublishEndpoint publishEndpoint, ILogger<ShipmentCreatedConsumer> logger)
    {
        this.publishEndpoint = publishEndpoint;
        this.logger = logger;
    }

    public async Task Consume(ConsumeContext<ShipmentCreatedEvent> context)
    {
        var message = context.Message;
        logger.LogInformation("Received shipment created event: {@Event}", message);

        // Xử lý tạo Order và publish event
        var orderEvent = new OrderCreatedEvent(message.OrderId, DateTime.UtcNow, 100, message.ReceiverEmail);
        await publishEndpoint.Publish(orderEvent);
    }
}

dotnet add package MassTransit.Azure.ServiceBus.Core

busConfig.UsingAzureServiceBus((context, cfg) =>
{
    cfg.Host(azureServiceBusConfiguration.ConnectionString);

    cfg.UseMessageRetry(r => r.Exponential(10, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(60), TimeSpan.FromSeconds(5)));
    cfg.ConfigureEndpoints(context);
});