Gargalo em Banco de Dados: Mensageria e Paginação

// ❌ Abordagem ingênua — InsertOne por vez
foreach (var pedido in listaDe50MilPedidos)
{
    await context.Pedidos.AddAsync(pedido);
    await context.SaveChangesAsync(); // PROBLEMA: 1 roundtrip por registro!
}

// Oracle: configuração correta no OnModelCreating
protected override void OnModelCreating(ModelBuilder modelBuilder)
{
    modelBuilder.Entity<Pedido>(entity =>
    {
        entity.HasKey(e => e.Id);

        // Oracle 12c+: identity nativo, evita roundtrip extra por ID
        entity.Property(e => e.Id)
              .UseIdentityColumn();

        // Alternativa para Oracle 11g/legado
        // entity.Property(e => e.Id)
        //       .HasDefaultValueSql("SEQ_PEDIDOS.NEXTVAL");
    });
}

// ✅ Melhor: batch por chunk + 1 SaveChanges por lote
public async Task GravarPedidosEmLoteAsync(
    IEnumerable<Pedido> pedidos,
    CancellationToken ct = default)
{
    const int tamanhoLote = 500;

    foreach (var chunk in pedidos.Chunk(tamanhoLote))
    {
        await context.Pedidos.AddRangeAsync(chunk, ct);
        await context.SaveChangesAsync(ct);

        // Limpar o ChangeTracker para não acumular entidades em memória
        context.ChangeTracker.Clear();
    }
}

// ✅ BulkInsert para SQL Server e Oracle (via EFCore.BulkExtensions)
// dotnet add package EFCore.BulkExtensions
public async Task BulkInsertPedidosAsync(
    List<Pedido> pedidos,
    CancellationToken ct = default)
{
    var bulkConfig = new BulkConfig
    {
        BatchSize = 1000,
        UseTempDB = true,              // SQL Server: tabela temporária para staging
        SetOutputIdentity = true,      // Preenche os IDs gerados pelo banco
        PreserveInsertOrder = true
    };

    await context.BulkInsertAsync(pedidos, bulkConfig, cancellationToken: ct);
}

// dotnet add package RabbitMQ.Client
// dotnet add package EFCore.BulkExtensions

// --- Mensagem ---
public record PedidoCriadoMessage(
    Guid Id,
    string ClienteId,
    decimal Valor,
    DateTime DataCriacao);

// --- Producer ---
public class PedidoProducer(IConnection connection)
{
    private const string QueueName = "pedidos-queue";

    public async Task PublicarAsync(
        PedidoCriadoMessage mensagem,
        CancellationToken ct = default)
    {
        await using var channel = await connection.CreateChannelAsync(cancellationToken: ct);

        await channel.QueueDeclareAsync(
            queue: QueueName,
            durable: true,
            exclusive: false,
            autoDelete: false,
            cancellationToken: ct);

        var json = JsonSerializer.SerializeToUtf8Bytes(mensagem);
        var props = new BasicProperties { DeliveryMode = DeliveryModes.Persistent };

        await channel.BasicPublishAsync(
            exchange: string.Empty,
            routingKey: QueueName,
            mandatory: false,
            basicProperties: props,
            body: json,
            cancellationToken: ct);
    }
}

// --- Consumer em lote (BackgroundService) ---
public class PedidoConsumerWorker(
    IConnection connection,
    IServiceScopeFactory scopeFactory,
    ILogger<PedidoConsumerWorker> logger) : BackgroundService
{
    private const string QueueName = "pedidos-queue";
    private const int TamanhoLote = 500;

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        await using var channel = await connection.CreateChannelAsync(cancellationToken: ct);

        await channel.QueueDeclareAsync(
            queue: QueueName, durable: true,
            exclusive: false, autoDelete: false,
            cancellationToken: ct);

        // Backpressure: limita mensagens em voo para não sobrecarregar a memória
        await channel.BasicQosAsync(
            prefetchSize: 0,
            prefetchCount: TamanhoLote + 100,
            global: false,
            cancellationToken: ct);

        var lote = new List<(ulong Tag, PedidoCriadoMessage Msg)>();

        var consumer = new AsyncEventingBasicConsumer(channel);
        consumer.ReceivedAsync += async (_, ea) =>
        {
            var msg = JsonSerializer.Deserialize<PedidoCriadoMessage>(ea.Body.Span)!;
            lote.Add((ea.DeliveryTag, msg));

            if (lote.Count >= TamanhoLote)
                await ProcessarLoteAsync(channel, lote, ct);
        };

        await channel.BasicConsumeAsync(
            queue: QueueName,
            autoAck: false,
            consumer: consumer,
            cancellationToken: ct);

        // Timer de flush: força o processamento mesmo que o lote não esteja cheio
        using var timer = new PeriodicTimer(TimeSpan.FromSeconds(5));
        while (await timer.WaitForNextTickAsync(ct))
        {
            if (lote.Count > 0)
                await ProcessarLoteAsync(channel, lote, ct);
        }
    }

    private async Task ProcessarLoteAsync(
        IChannel channel,
        List<(ulong Tag, PedidoCriadoMessage Msg)> lote,
        CancellationToken ct)
    {
        var snapshot = lote.ToList();
        lote.Clear();

        var pedidos = snapshot.Select(x => new Pedido
        {
            Id           = x.Msg.Id,
            ClienteId    = x.Msg.ClienteId,
            Valor        = x.Msg.Valor,
            DataCriacao  = x.Msg.DataCriacao
        }).ToList();

        using var scope    = scopeFactory.CreateScope();
        var dbContext       = scope.ServiceProvider.GetRequiredService<AppDbContext>();
        var bulkConfig      = new BulkConfig { BatchSize = 500 };

        await dbContext.BulkInsertAsync(pedidos, bulkConfig, cancellationToken: ct);

        // ACK múltiplo: confirma todo o lote de uma só vez
        await channel.BasicAckAsync(
            deliveryTag: snapshot[^1].Tag,
            multiple: true,
            cancellationToken: ct);

        logger.LogInformation("Lote de {Count} pedidos gravado", pedidos.Count);
    }
}

// Program.cs
// dotnet add package RabbitMQ.Client
builder.Services.AddSingleton<IConnection>(_ =>
{
    var factory = new ConnectionFactory
    {
        HostName = "localhost",
        UserName = "guest",
        Password = "guest"
    };
    // CreateConnectionAsync retorna Task — resolvemos aqui para o DI container
    return factory.CreateConnectionAsync().GetAwaiter().GetResult();
});

builder.Services.AddScoped<PedidoProducer>();
builder.Services.AddHostedService<PedidoConsumerWorker>();

// dotnet add package Azure.Messaging.ServiceBus

// --- Producer com envio em lote nativo ---
public class PedidoServiceBusProducer(ServiceBusSender sender)
{
    public async Task PublicarLoteAsync(
        IEnumerable<PedidoCriadoMessage> mensagens,
        CancellationToken ct = default)
    {
        // O ServiceBusMessageBatch respeita o limite de tamanho automaticamente
        using var batch = await sender.CreateMessageBatchAsync(ct);

        foreach (var msg in mensagens)
        {
            var sbMsg = new ServiceBusMessage(
                BinaryData.FromObjectAsJson(msg))
            {
                ContentType = "application/json",
                MessageId   = msg.Id.ToString()
            };

            if (!batch.TryAddMessage(sbMsg))
                throw new InvalidOperationException(
                    $"Mensagem {msg.Id} excede o limite do lote");
        }

        await sender.SendMessagesAsync(batch, ct);
    }
}

// --- Consumer com BackgroundService ---
public class PedidoServiceBusWorker(
    ServiceBusProcessor processor,
    IServiceScopeFactory scopeFactory,
    ILogger<PedidoServiceBusWorker> logger) : BackgroundService
{
    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        processor.ProcessMessageAsync += async args =>
        {
            var msg = args.Message.Body
                          .ToObjectFromJson<PedidoCriadoMessage>();

            using var scope   = scopeFactory.CreateScope();
            var dbContext      = scope.ServiceProvider
                                      .GetRequiredService<AppDbContext>();
            var pedido         = new Pedido
            {
                Id          = msg.Id,
                ClienteId   = msg.ClienteId,
                Valor       = msg.Valor,
                DataCriacao = msg.DataCriacao
            };

            await dbContext.BulkInsertAsync(
                new List<Pedido> { pedido },
                cancellationToken: ct);

            // Confirma o processamento — remove da fila
            await args.CompleteMessageAsync(args.Message, ct);
        };

        processor.ProcessErrorAsync += args =>
        {
            logger.LogError(args.Exception,
                "Erro ao processar mensagem do Service Bus");
            return Task.CompletedTask;
        };

        await processor.StartProcessingAsync(ct);
        await Task.Delay(Timeout.Infinite, ct);
    }

    public override async Task StopAsync(CancellationToken ct)
    {
        await processor.StopProcessingAsync(ct);
        await base.StopAsync(ct);
    }
}

// Program.cs
builder.Services.AddSingleton(provider =>
{
    var client = new ServiceBusClient(
        "Endpoint=sb://meu-namespace.servicebus.windows.net/;...");
    return client.CreateSender("pedidos-queue");
});

builder.Services.AddSingleton(provider =>
{
    var client = new ServiceBusClient(
        "Endpoint=sb://meu-namespace.servicebus.windows.net/;...");
    return client.CreateProcessor("pedidos-queue", new ServiceBusProcessorOptions
    {
        MaxConcurrentCalls   = 4,   // Paralelismo no consumer
        AutoCompleteMessages = false // Controle manual de ACK
    });
});

builder.Services.AddHostedService<PedidoServiceBusWorker>();

// ❌ NUNCA faça isso em produção com tabelas grandes
var todosPedidos = await context.Pedidos
    .Include(p => p.Itens)
    .Include(p => p.Cliente)
    .ToListAsync(); // Pode trazer 500 mil registros para a memória!

// Modelo de request padronizado
public record PaginacaoRequest(int Pagina = 1, int TamanhoPagina = 20)
{
    // Garante limites seguros
    public int TamanhoSeguro => Math.Min(Math.Max(TamanhoPagina, 1), 100);
    public int OffsetSeguro  => (Math.Max(Pagina, 1) - 1) * TamanhoSeguro;
}

// Modelo de resposta padronizado
public record PaginaResultado<T>(
    IReadOnlyList<T> Dados,
    int PaginaAtual,
    int TamanhoPagina,
    long TotalRegistros,
    int TotalPaginas);

// Serviço de consulta paginada
public class PedidoQueryService(AppDbContext context)
{
    public async Task<PaginaResultado<PedidoResumoDto>> ListarPedidosAsync(
        PaginacaoRequest paginacao,
        string? clienteId = null,
        CancellationToken ct = default)
    {
        // Construir query base com filtros opcionais
        var query = context.Pedidos
            .AsNoTracking()           // Leitura: sempre AsNoTracking!
            .Where(p => clienteId == null || p.ClienteId == clienteId);

        // COUNT separado para o total (EF Core 8 otimiza isso)
        var total = await query.LongCountAsync(ct);

        // Dados da página
        var dados = await query
            .OrderBy(p => p.DataCriacao)   // Orderby obrigatório para paginação consistente
            .Skip(paginacao.OffsetSeguro)
            .Take(paginacao.TamanhoSeguro)
            .Select(p => new PedidoResumoDto(   // Projection: só os campos necessários
                p.Id,
                p.ClienteId,
                p.Valor,
                p.DataCriacao,
                p.Status))
            .ToListAsync(ct);

        return new PaginaResultado<PedidoResumoDto>(
            Dados: dados,
            PaginaAtual: paginacao.Pagina,
            TamanhoPagina: paginacao.TamanhoSeguro,
            TotalRegistros: total,
            TotalPaginas: (int)Math.Ceiling(total / (double)paginacao.TamanhoSeguro));
    }
}

-- SQL Server (gerado pelo EF Core 8)
SELECT p.Id, p.ClienteId, p.Valor, p.DataCriacao, p.Status
FROM Pedidos AS p
WHERE p.ClienteId = N'cliente-123'
ORDER BY p.DataCriacao
OFFSET 40 ROWS FETCH NEXT 20 ROWS ONLY;

-- Oracle 12c+ (gerado pelo Oracle.EntityFrameworkCore 8)
SELECT p.Id, p.ClienteId, p.Valor, p.DataCriacao, p.Status
FROM Pedidos p
WHERE p.ClienteId = :clienteId_0
ORDER BY p.DataCriacao
OFFSET 40 ROWS FETCH NEXT 20 ROWS ONLY;

// Request com cursor (keyset)
public record KeysetRequest(
    Guid? UltimoId = null,
    DateTime? UltimaDataCriacao = null,
    int Limite = 20);

// Resultado com cursor para a próxima página
public record KeysetResultado<T>(
    IReadOnlyList<T> Dados,
    bool TemProximaPagina,
    Guid? ProximoCursorId,
    DateTime? ProximoCursorData);

public class PedidoKeysetService(AppDbContext context)
{
    public async Task<KeysetResultado<PedidoResumoDto>> ListarComCursorAsync(
        KeysetRequest request,
        CancellationToken ct = default)
    {
        // Filtro de keyset: busca registros "após" o cursor
        // Usando (DataCriacao, Id) como chave composta para estabilidade
        var query = context.Pedidos
            .AsNoTracking()
            .Where(p =>
                request.UltimaDataCriacao == null ||
                p.DataCriacao > request.UltimaDataCriacao ||
                (p.DataCriacao == request.UltimaDataCriacao &&
                 p.Id.CompareTo(request.UltimoId!.Value) > 0))
            .OrderBy(p => p.DataCriacao)
            .ThenBy(p => p.Id);

        // Busca Limite + 1 para saber se há próxima página
        var dados = await query
            .Take(request.Limite + 1)
            .Select(p => new PedidoResumoDto(p.Id, p.ClienteId, p.Valor, p.DataCriacao, p.Status))
            .ToListAsync(ct);

        var temProxima = dados.Count > request.Limite;
        if (temProxima) dados.RemoveAt(dados.Count - 1); // Remove o item extra

        var ultimo = dados.LastOrDefault();

        return new KeysetResultado<PedidoResumoDto>(
            Dados: dados,
            TemProximaPagina: temProxima,
            ProximoCursorId: ultimo?.Id,
            ProximoCursorData: ultimo?.DataCriacao);
    }
}

-- SQL gerado (SQL Server / Oracle)
SELECT TOP(21) p.Id, p.ClienteId, p.Valor, p.DataCriacao, p.Status
FROM Pedidos AS p
WHERE p.DataCriacao > '2025-06-01'
   OR (p.DataCriacao = '2025-06-01' AND p.Id > 'abc-123')
ORDER BY p.DataCriacao, p.Id;

-- SQL Server
CREATE INDEX IX_Pedidos_Keyset ON Pedidos (DataCriacao, Id) INCLUDE (ClienteId, Valor, Status);

-- Oracle
CREATE INDEX IX_Pedidos_Keyset ON Pedidos (DataCriacao, Id);

// Endpoint ASP.NET Core — escrita assíncrona
app.MapPost("/pedidos/lote", async (
    [FromBody] List<CriarPedidoRequest> requests,
    PedidoService service,
    CancellationToken ct) =>
{
    var ids = await Task.WhenAll(
        requests.Select(r => service.CriarPedidoAsync(r, ct)));

    // 202 Accepted: os pedidos foram enfileirados, não necessariamente gravados
    return Results.Accepted("/pedidos/status", new { Ids = ids });
});

// Endpoint ASP.NET Core — leitura paginada
app.MapGet("/pedidos", async (
    [AsParameters] PaginacaoRequest paginacao,
    [FromQuery] string? clienteId,
    PedidoQueryService queryService,
    CancellationToken ct) =>
{
    var resultado = await queryService.ListarPedidosAsync(paginacao, clienteId, ct);
    return Results.Ok(resultado);
});

// Streaming com IAsyncEnumerable — sem carregar tudo na memória
await foreach (var pedido in context.Pedidos
    .AsNoTracking()
    .AsAsyncEnumerable()
    .WithCancellation(ct))
{
    await exportador.EscreverLinhaAsync(pedido, ct);
}