firebird-wire 0.1.5

//! DML em lote (batch) — o recurso "principal" de array DML do Firebird 4+.
//!
//! Um [`Batch`] insere/atualiza/exclui muitas linhas com uma única instrução
//! preparada, acumulando mensagens no cliente e enviando-as ao servidor em uma
//! ida só. É muito mais rápido que executar a instrução linha a linha.
//!
//! ```text
//! let mut batch = conn.create_batch(&tx, "INSERT INTO t (a, b) VALUES (?, ?)")?;
//! batch.add(&[Value::Int(1), Value::Text("um".into())])?;
//! batch.add(&[Value::Int(2), Value::Text("dois".into())])?;
//! let result = batch.execute(&mut conn, &tx)?;   // envia + executa
//! println!("{} linhas afetadas no total", result.total_affected());
//! batch.close(&mut conn)?;                        // libera no servidor
//! ```
//!
//! # Protocolo (FB4+, op codes 99–103)
//!
//! Descoberto por captura de um cliente C usando a interface OO `IBatch`:
//! 1. `op_batch_create` (99): `stmt | blr(cstring) | msglen(u32) | pb(cstring)`.
//!    O BLR descreve o formato da mensagem (igual ao `in_blr` de `op_execute`);
//!    `msglen` é o tamanho do buffer de mensagem do lado do cliente.
//! 2. `op_batch_msg` (100): `stmt | count(u32) | mensagens`, cada mensagem no
//!    mesmo formato compacto (bitmap de nulos + valores XDR) usado em `op_execute`.
//! 3. `op_batch_exec` (101): `stmt | transaction`. Responde com `op_batch_cs`.
//! 4. `op_batch_cs` (103): estado de conclusão (contagens por linha + erros).
//! 5. `op_batch_rls` (102): libera o batch no servidor.
//!
//! O cliente C agrupa create+msg e usa `op_batch_sync` (110) para drenar as
//! respostas adiadas; nós, sendo síncronos, lemos a resposta de cada op na hora.
//!
//! # BLOBs em lote (política STREAM, `op_batch_blob_stream` 105)
//!
//! Quando a instrução tem coluna BLOB, [`Connection::create_batch`] ativa a
//! política `BLOB_STREAM` no PB (`TAG_BLOB_POLICY` = 3). O chamador então usa
//! [`Batch::add_blob`] para registrar cada blob e recebe um id (quad) para pôr
//! no campo da linha. Em [`Batch::execute`], antes das mensagens, enviamos:
//!
//! `op_batch_blob_stream` (105): `stmt | length(u32) | stream`. O `stream` é a
//! concatenação CRUA (sem padding) dos blobs, cada um `id(quad 8B BE) | size(4B
//! BE) | bpb_size(4B BE) | bpb | dados`. O `length` NÃO é o número de bytes no
//! wire: é o tamanho do buffer que o servidor aloca — a soma de `align4(16 +
//! bpb + dados)` de cada blob — e deve ser múltiplo de 4 (o servidor rejeita
//! caso contrário). Ao percorrer o stream (ver `xdr_blob_stream` no servidor),
//! ele avança o ponteiro do buffer com alinhamento de 4 SEM consumir bytes do
//! wire para o padding; o laço termina quando o que resta é menor que um
//! cabeçalho (16 B) ou chega a zero. Por isso o wire carrega menos bytes que
//! `length`, e a próxima op pode começar em offset não múltiplo de 4.
//! Confirmado por captura do `11.batch.cpp` (conteúdo 30 B → length 32; 33 B →
//! 36) e pelo código do servidor. Os blobs vão **antes** das mensagens porque a
//! linha referencia o id já registrado.
//!
//! `op_batch_regblob` (104): `stmt | id_existente(quad 8B BE) | id_batch(quad)`.
//! Mapeia um BLOB já gravado fora do batch ([`Connection::write_blob`]) a um id
//! local, evitando reenviar os dados. Ver [`Batch::register_blob`].
//!
//! `op_batch_set_bpb` (106): `stmt | bpb(cstring)`. Define o BPB padrão dos
//! blobs do batch; o servidor lê o `isc_bpb_type` para marcar os blobs como
//! segmentados ou stream. Ver [`Batch::set_default_bpb`] / [`Batch::set_segmented`].
//! Em modo segmentado, cada segmento no stream é um `u32` big-endian com o
//! comprimento seguido dos bytes, SEM padding; já o campo `size` do blob segue a
//! contabilidade do buffer do servidor (cabeçalho de 2 bytes alinhado a 2), ou
//! seja `align2(2 + len)`. Capturado da Parte 3 do `11.batch.cpp`.

use crate::blr::message_blr;
use crate::connection::Connection;
use crate::error::{DatabaseError, Error, Result, StatusVector};
use crate::message::{encode_row_into, message_buffer_len};
use crate::transaction::Transaction;
use crate::value::{ColumnMeta, Value};
use crate::wire::consts::*;
use crate::wire::response::{read_op, read_response, read_response_body, read_status_vector};
use crate::wire::stream::{op_name, op_packet};
use crate::wire::xdr::ParameterBuffer;

/// Um lote de mensagens vinculado a uma instrução preparada no servidor.
///
/// Crie-o com [`Connection::create_batch`], acumule linhas com [`Batch::add`] e
/// envie com [`Batch::execute`]. O batch pode ser reutilizado (adicione mais e
/// execute de novo). Ao terminar, chame [`Batch::close`] para liberá-lo.
pub struct Batch {
    handle: i32,
    params: Vec<ColumnMeta>,
    /// Mensagens já codificadas mas ainda não enviadas/executadas.
    pending: Vec<u8>,
    pending_count: u32,
    /// Stream de BLOBs acumulado (cabeçalho + dados de cada blob), ainda não
    /// enviado. Só usado em batches com coluna BLOB (política STREAM). Os blobs
    /// vão concatenados SEM padding no wire (o servidor reconstrói o alinhamento
    /// no buffer dele).
    pending_blobs: Vec<u8>,
    /// Comprimento *alinhado* do stream de blobs (campo `p_batch_blob_data` do
    /// op_batch_blob_stream): a soma de `align4(16 + dados)` de cada blob. É o
    /// tamanho do buffer que o servidor aloca, e deve ser múltiplo de
    /// `BLOB_ALIGN` (o servidor rejeita caso contrário). Sempre `>=` o número de
    /// bytes efetivamente enviados.
    blob_stream_len: usize,
    /// Registros pendentes de BLOBs pré-existentes (`op_batch_regblob`): cada par
    /// é `(id_existente, id_no_batch)`. Enviados em [`Batch::execute`] antes das
    /// mensagens, junto com o stream de blobs.
    pending_regblobs: Vec<(u64, u64)>,
    /// Próximo id (quad) a atribuir em [`Batch::add_blob`]/[`Batch::register_blob`].
    /// Começa em 1; o id 0 é reservado (quad nulo).
    next_blob_id: u64,
    /// Verdadeiro se a instrução tem ao menos uma coluna BLOB (política STREAM
    /// ativada na criação).
    blob_stream: bool,
    /// Charset da conexão, usado para codificar parâmetros de texto em
    /// [`Batch::add`] (capturado na criação, já que `add` não recebe a conexão).
    charset: crate::charset::Charset,
    /// BPB padrão a aplicar aos blobs do batch (`op_batch_set_bpb`), se definido.
    /// Enviado em [`Batch::execute`] antes do stream de blobs.
    default_bpb: Option<Vec<u8>>,
    /// Verdadeiro quando o BPB padrão marca os blobs como segmentados; nesse caso
    /// [`Batch::add_blob`] enquadra os dados em segmentos.
    blob_segmented: bool,
    closed: bool,
}

/// Alinhamento do cabeçalho de segmento em um blob segmentado
/// (`IBatch::BLOB_SEGHDR_ALIGN`). Como cada `add_blob` segmentado começa numa
/// fronteira de 4 bytes (já múltiplo de 2), não há padding a inserir.
const BLOB_SEGHDR_ALIGN: usize = 2;
const _: () = assert!(BLOB_ALIGN.is_multiple_of(BLOB_SEGHDR_ALIGN));

/// Arredonda `n` para cima até um múltiplo de `align` (potência de 2).
#[inline]
fn align_up(n: usize, align: usize) -> usize {
    (n + align - 1) & !(align - 1)
}

/// Verdadeiro se o BPB marca o blob como segmentado: clumplet `isc_bpb_type`(3)
/// com valor `isc_bpb_type_segmented`(0). Formato do BPB: byte de versão seguido
/// de clumplets `tag(1) | len(1) | valor(len)`.
fn bpb_is_segmented(bpb: &[u8]) -> bool {
    let mut i = 1; // pula o byte de versão
    while i + 1 < bpb.len() {
        let tag = bpb[i];
        let len = bpb[i + 1] as usize;
        let val = &bpb[i + 2..(i + 2 + len).min(bpb.len())];
        if tag == bpb::TYPE {
            return val.first() == Some(&bpb::TYPE_SEGMENTED);
        }
        i += 2 + len;
    }
    false
}

/// Alinhamento (em bytes) de cada blob dentro do stream de `op_batch_blob_stream`.
/// O servidor FB5 usa 4 (confirmado por captura: conteúdo de 30 B → campo de
/// comprimento 32; de 33 B → 36). É o valor que `IBatch::getBlobAlignment`
/// retorna para o protocolo remoto.
const BLOB_ALIGN: usize = 4;

impl Batch {
    /// O handle da instrução/batch no servidor.
    pub fn handle(&self) -> i32 {
        self.handle
    }

    /// Quantas mensagens estão acumuladas mas ainda não executadas.
    pub fn pending(&self) -> u32 {
        self.pending_count
    }

    /// Tamanho (alinhado) do stream de BLOBs acumulado mas ainda não enviado, em
    /// bytes. Zera após [`Batch::execute`]/[`Batch::cancel`]. Útil para o chamador
    /// drenar o lote por *volume de blob* (e não só por contagem de linhas) e assim
    /// não estourar o buffer de batch do servidor (`TAG_BUFFER_BYTES_SIZE`).
    pub fn blob_stream_len(&self) -> usize {
        self.blob_stream_len
    }

    /// Os metadados dos parâmetros de entrada (a forma de cada linha esperada).
    pub fn params(&self) -> &[ColumnMeta] {
        &self.params
    }

    /// Registra um BLOB no stream do lote e devolve seu id (quad), que deve ser
    /// posto no campo BLOB da linha correspondente como [`Value::Blob`].
    ///
    /// Só funciona em batches cuja instrução tem coluna BLOB (a política STREAM é
    /// ativada automaticamente em [`Connection::create_batch`] nesse caso). Como
    /// [`Self::add`], apenas acumula na memória; os blobs vão à rede em
    /// [`Self::execute`], **antes** das mensagens que os referenciam.
    ///
    /// ```text
    /// let id = batch.add_blob(b"conteudo do blob")?;
    /// batch.add(&[Value::Text("BAT31".into()), Value::Blob(id)])?;
    /// ```
    pub fn add_blob(&mut self, data: &[u8]) -> Result<u64> {
        if !self.blob_stream {
            return Err(Error::protocol(
                "add_blob exige uma coluna BLOB na instrução do batch",
            ));
        }
        let id = self.next_blob_id;
        self.next_blob_id += 1;
        // Cabeçalho do blob, tudo big-endian: id(quad 8B) | size(4B) | bpb_size(4B
        // = 0, pois o BPB é do batch). Os blobs vão concatenados SEM padding; o
        // servidor reconstrói o alinhamento no buffer dele.
        self.pending_blobs.extend_from_slice(&id.to_be_bytes());
        if self.blob_segmented {
            // Blob segmentado (um único segmento aqui). NO WIRE: cada segmento é
            // `u32` big-endian com o comprimento + os bytes, sem padding. Mas o
            // campo `size` (e o comprimento do stream) seguem a contabilidade do
            // *buffer* do servidor, que usa cabeçalho de 2 bytes alinhado a
            // `BLOB_SEGHDR_ALIGN`: `size = align2(2 + len)`. Ver `xdr_blob_stream`.
            let size_field = align_up(2 + data.len(), BLOB_SEGHDR_ALIGN);
            self.pending_blobs
                .extend_from_slice(&(size_field as u32).to_be_bytes());
            self.pending_blobs.extend_from_slice(&0u32.to_be_bytes()); // bpb_size
            self.pending_blobs
                .extend_from_slice(&(data.len() as u32).to_be_bytes());
            self.pending_blobs.extend_from_slice(data);
            self.blob_stream_len += align_up(16 + size_field, BLOB_ALIGN);
        } else {
            // Blob de stream (não segmentado): dados crus.
            self.pending_blobs
                .extend_from_slice(&(data.len() as u32).to_be_bytes());
            self.pending_blobs.extend_from_slice(&0u32.to_be_bytes()); // bpb_size
            self.pending_blobs.extend_from_slice(data);
            self.blob_stream_len += align_up(16 + data.len(), BLOB_ALIGN);
        }
        Ok(id)
    }

    /// Define o BPB (blob parameter buffer) padrão dos blobs do batch
    /// (`op_batch_set_bpb`). O uso típico é marcar os blobs como **segmentados**
    /// (`isc_bpb_type = isc_bpb_type_segmented`): nesse caso [`Self::add_blob`]
    /// passa a enquadrar cada blob como um segmento. Deve ser chamado antes de
    /// [`Self::add_blob`]; o BPB vai à rede em [`Self::execute`], antes dos blobs.
    ///
    /// Conveniência: [`Self::set_segmented`] monta o BPB segmentado para você.
    pub fn set_default_bpb(&mut self, bpb: Vec<u8>) -> Result<()> {
        if !self.blob_stream {
            return Err(Error::protocol(
                "set_default_bpb exige uma coluna BLOB na instrução do batch",
            ));
        }
        // Detecta o tipo segmentado: clumplet `isc_bpb_type`(3) com valor
        // `isc_bpb_type_segmented`(0). BPB usa comprimento de 1 byte por clumplet.
        self.blob_segmented = bpb_is_segmented(&bpb);
        self.default_bpb = Some(bpb);
        Ok(())
    }

    /// Atalho para [`Self::set_default_bpb`] com um BPB que marca os blobs como
    /// segmentados (`segmentado = true`) ou stream (`false`, o padrão).
    pub fn set_segmented(&mut self, segmented: bool) -> Result<()> {
        let kind = if segmented {
            bpb::TYPE_SEGMENTED
        } else {
            bpb::TYPE_STREAM
        };
        // BPB no formato que o fbclient envia: versão(1) | tag isc_bpb_type(3) |
        // len(1) = 4 | valor(4 bytes LE). Capturado do `11.batch.cpp` Parte 3.
        self.set_default_bpb(vec![BPB_VERSION1, bpb::TYPE, 4, kind, 0, 0, 0])
    }

    /// Mapeia um BLOB **já existente** (criado fora do batch, p.ex. via
    /// [`Connection::write_blob`]/[`Connection::create_blob`]) para um id local
    /// do batch e devolve esse id, que deve ir no campo BLOB da linha como
    /// [`Value::Blob`]. Útil para reaproveitar BLOBs grandes já gravados sem
    /// reenviá-los pelo stream (`op_batch_regblob`).
    ///
    /// Como [`Self::add_blob`], só vale em batches com coluna BLOB; o registro é
    /// acumulado e enviado em [`Self::execute`], antes das mensagens.
    pub fn register_blob(&mut self, existing_id: u64) -> Result<u64> {
        if !self.blob_stream {
            return Err(Error::protocol(
                "register_blob exige uma coluna BLOB na instrução do batch",
            ));
        }
        let batch_id = self.next_blob_id;
        self.next_blob_id += 1;
        self.pending_regblobs.push((existing_id, batch_id));
        Ok(batch_id)
    }

    /// Adiciona uma linha ao lote. Os valores devem corresponder, em número e
    /// tipo, aos parâmetros da instrução (ver [`Self::params`]). Apenas acumula
    /// na memória; nada vai à rede até [`Self::execute`].
    pub fn add(&mut self, values: &[Value]) -> Result<()> {
        // Codifica direto no buffer pendente (sem Vec temporário por linha). Em erro,
        // `encode_row_into` restaura o tamanho de `pending`, então não corrompe o lote.
        encode_row_into(&mut self.pending, &self.params, values, self.charset)?;
        self.pending_count += 1;
        Ok(())
    }

    /// Envia as mensagens acumuladas e executa o lote, retornando o estado de
    /// conclusão (contagens por linha e erros por linha). Esvazia o buffer
    /// pendente; o batch pode então ser reutilizado.
    pub fn execute(&mut self, conn: &mut Connection, tx: &Transaction) -> Result<BatchResult> {
        if self.closed {
            return Err(Error::protocol("batch já foi fechado"));
        }
        // 0a. Define o BPB padrão (op_batch_set_bpb), se houver, antes dos blobs.
        if let Some(bpb) = self.default_bpb.take() {
            let mut w = op_packet(op::BATCH_SET_BPB);
            w.put_i32(self.handle);
            w.put_bytes(&bpb); // cstring: len(4) + bpb + pad
            conn.io().send(&w)?;
            read_response(conn.io())?;
        }
        // 0b. Envia os BLOBs pendentes (op_batch_blob_stream) ANTES das mensagens,
        //    pois cada mensagem referencia um id de blob já registrado.
        if !self.pending_blobs.is_empty() {
            let mut w = op_packet(op::BATCH_BLOB_STREAM);
            w.put_i32(self.handle);
            // Comprimento alinhado (≥ bytes no wire), depois os blobs crus. Não
            // alinhamos o pacote: o servidor recompõe o alinhamento internamente
            // e a próxima op pode começar em offset não múltiplo de 4.
            w.put_i32(self.blob_stream_len as i32);
            w.put_raw(&self.pending_blobs);
            conn.io().send(&w)?;
            read_response(conn.io())?;
            self.pending_blobs.clear();
            self.blob_stream_len = 0;
        }
        // 0b. Registra BLOBs pré-existentes (op_batch_regblob), também antes das
        //     mensagens. Layout: stmt | id_existente(quad 8B BE) | id_batch(quad).
        if !self.pending_regblobs.is_empty() {
            for (existing_id, batch_id) in std::mem::take(&mut self.pending_regblobs) {
                let mut w = op_packet(op::BATCH_REGBLOB);
                w.put_i32(self.handle);
                w.put_raw(&existing_id.to_be_bytes());
                w.put_raw(&batch_id.to_be_bytes());
                conn.io().send(&w)?;
                read_response(conn.io())?;
            }
        }
        // 1. Envia as mensagens pendentes (op_batch_msg), se houver.
        if self.pending_count > 0 {
            let mut w = op_packet(op::BATCH_MSG);
            w.put_i32(self.handle);
            w.put_i32(self.pending_count as i32);
            w.put_raw(&self.pending);
            w.align();
            conn.io().send(&w)?;
            read_response(conn.io())?;
            self.pending.clear();
            self.pending_count = 0;
        }

        // 2. Executa o lote (op_batch_exec) e lê o estado de conclusão.
        let mut w = op_packet(op::BATCH_EXEC);
        w.put_i32(self.handle);
        w.put_i32(tx.handle());
        conn.io().send(&w)?;
        read_batch_cs(conn)
    }

    /// Descarta as mensagens acumuladas que ainda não foram executadas
    /// (`op_batch_cancel`). Não afeta linhas já executadas em chamadas anteriores.
    pub fn cancel(&mut self, conn: &mut Connection) -> Result<()> {
        self.pending.clear();
        self.pending_count = 0;
        self.pending_blobs.clear();
        self.blob_stream_len = 0;
        self.pending_regblobs.clear();
        let mut w = op_packet(op::BATCH_CANCEL);
        w.put_i32(self.handle);
        conn.io().send(&w)?;
        read_response(conn.io())?;
        Ok(())
    }

    /// Libera o batch e a instrução preparada no servidor (`op_batch_rls` +
    /// `op_free_statement` com `DSQL_drop`), consumindo o handle.
    pub fn close(mut self, conn: &mut Connection) -> Result<()> {
        self.closed = true;
        let mut w = op_packet(op::BATCH_RLS);
        w.put_i32(self.handle);
        conn.io().send(&w)?;
        read_response(conn.io())?;

        let mut w = op_packet(op::FREE_STATEMENT);
        w.put_i32(self.handle);
        w.put_i32(free::DROP);
        conn.io().send(&w)?;
        read_response(conn.io())?;
        Ok(())
    }
}

impl Drop for Batch {
    fn drop(&mut self) {
        if !self.closed {
            crate::warn_unclosed("Batch", self.handle);
        }
    }
}

/// Opções de criação de um [`Batch`] (clumplets do `op_batch_create`).
#[derive(Debug, Clone, Copy)]
pub struct BatchOptions {
    /// Se `true`, o servidor CONTINUA após uma linha falhar, executando as demais e
    /// reportando o erro de cada uma em [`BatchResult::errors`]. Para isso ele
    /// bracketa cada linha num savepoint interno — o que tem **custo por linha**.
    ///
    /// Se `false` (**padrão**), o lote PARA na primeira linha que falha (fail-fast):
    /// é bem mais rápido (sem savepoint por linha) e é o que se quer quando a
    /// transação é "tudo ou nada" e qualquer erro aborta a operação inteira.
    pub multierror: bool,
    /// Se `true` (**padrão**), o servidor reporta a contagem de linhas afetadas por
    /// mensagem em [`BatchResult::update_counts`] (e portanto [`BatchResult::total_affected`]).
    pub record_counts: bool,
    /// Tamanho máximo (em bytes) do buffer de batch que o servidor aloca
    /// (`TAG_BUFFER_BYTES_SIZE`). `None` (**padrão**) deixa o servidor usar o seu
    /// default. Aumentar permite acumular mais mensagens/blobs por `execute` sem
    /// estourar o buffer (o estouro aparece como "invalid BLOB ID"); o servidor
    /// impõe um teto próprio (256 MiB no Firebird).
    pub buffer_bytes: Option<u32>,
}

impl Default for BatchOptions {
    fn default() -> Self {
        // Fail-fast por padrão: a maioria dos usos de DML em lote quer abortar no
        // primeiro erro, e o multierror (savepoint por linha) só compensa quando o
        // chamador realmente vai inspecionar o erro de cada linha.
        BatchOptions {
            multierror: false,
            record_counts: true,
            buffer_bytes: None,
        }
    }
}

impl BatchOptions {
    /// Opções padrão (fail-fast, com contagens por linha).
    pub fn new() -> Self {
        Self::default()
    }

    /// Liga/desliga o modo multierro (continuar após erros por linha).
    pub fn multierror(mut self, on: bool) -> Self {
        self.multierror = on;
        self
    }

    /// Liga/desliga o reporte de contagens de linhas afetadas por mensagem.
    pub fn record_counts(mut self, on: bool) -> Self {
        self.record_counts = on;
        self
    }

    /// Define o tamanho do buffer de batch do servidor (`TAG_BUFFER_BYTES_SIZE`),
    /// em bytes. Use para acomodar lotes maiores (mais mensagens/blobs por
    /// `execute`) sem estourar o buffer. O servidor impõe um teto (256 MiB no
    /// Firebird); valores acima são limitados por ele.
    pub fn buffer_bytes(mut self, bytes: u32) -> Self {
        self.buffer_bytes = Some(bytes);
        self
    }
}

impl Connection {
    /// Prepara uma instrução e cria um lote (batch) sobre ela com as opções padrão
    /// ([`BatchOptions::default`]: fail-fast, com contagens por linha). A instrução
    /// deve ter parâmetros (`?`) — cada [`Batch::add`] fornece uma linha de valores.
    ///
    /// Para continuar após erros por linha e coletá-los todos, use
    /// [`Self::create_batch_with`] com `BatchOptions::new().multierror(true)`.
    pub fn create_batch(&mut self, tx: &Transaction, sql: &str) -> Result<Batch> {
        self.create_batch_with(tx, sql, BatchOptions::default())
    }

    /// Como [`Self::create_batch`], mas com [`BatchOptions`] explícitas.
    pub fn create_batch_with(
        &mut self,
        tx: &Transaction,
        sql: &str,
        opts: BatchOptions,
    ) -> Result<Batch> {
        let mut stmt = self.prepare(tx, sql)?;
        let handle = stmt.handle();
        let params: Vec<ColumnMeta> = stmt.params().to_vec();
        // O handle passa a viver no Batch (liberado por Batch::close), então
        // marcamos como transferido para não disparar o aviso de Drop e soltamos
        // o wrapper aqui (libera só memória).
        stmt.forget_handle();
        drop(stmt);

        let blr = message_blr(&params);
        let msglen = message_buffer_len(&params);

        // Se houver coluna BLOB, ativa a política STREAM: os blobs são enviados
        // em `op_batch_blob_stream` e a linha referencia o id (ver `add_blob`).
        let blob_stream = params
            .iter()
            .any(|c| sql_type::base(c.sql_type) == sql_type::BLOB);

        // Buffer de parâmetros do batch: byte de versão (1) seguido de clumplets
        // com comprimento LE de 4 bytes. Os tags são opcionais (ver `BatchOptions`).
        let mut pb = ParameterBuffer::new(1);
        if opts.record_counts {
            pb.bytes_be_len4(batch_tag::RECORD_COUNTS, &1u32.to_le_bytes());
        }
        if opts.multierror {
            pb.bytes_be_len4(batch_tag::MULTIERROR, &1u32.to_le_bytes());
        }
        if let Some(bytes) = opts.buffer_bytes {
            pb.bytes_be_len4(batch_tag::BUFFER_BYTES_SIZE, &bytes.to_le_bytes());
        }
        if blob_stream {
            pb.bytes_be_len4(
                batch_tag::BLOB_POLICY,
                &(blob_policy::STREAM as u32).to_le_bytes(),
            );
        }

        let mut w = op_packet(op::BATCH_CREATE);
        w.put_i32(handle);
        w.put_bytes(&blr); // cstring: len(4) + blr + pad
        w.put_i32(msglen as i32);
        w.put_bytes(pb.as_slice()); // cstring: len(4) + pb + pad
        self.io().send(&w)?;
        read_response(self.io())?;

        Ok(Batch {
            handle,
            params,
            pending: Vec::new(),
            pending_count: 0,
            pending_blobs: Vec::new(),
            blob_stream_len: 0,
            pending_regblobs: Vec::new(),
            next_blob_id: 1,
            blob_stream,
            charset: self.charset(),
            default_bpb: None,
            blob_segmented: false,
            closed: false,
        })
    }
}

/// Resultado da execução de um lote: o estado de conclusão por mensagem.
#[derive(Debug, Clone, Default)]
pub struct BatchResult {
    /// Total de mensagens processadas nesta execução.
    pub total: u32,
    /// Contagem de linhas afetadas por mensagem, na ordem em que foram
    /// adicionadas. `>= 0` é o número de linhas; [`batch_cs::EXECUTE_FAILED`]
    /// (−1) marca uma mensagem que falhou; [`batch_cs::SUCCESS_NO_INFO`] (−2)
    /// indica sucesso sem contagem reportada.
    pub update_counts: Vec<i32>,
    /// Erros detalhados por mensagem (índice da mensagem + erro do servidor).
    pub errors: Vec<BatchError>,
}

impl BatchResult {
    /// Verdadeiro se nenhuma mensagem falhou.
    pub fn all_succeeded(&self) -> bool {
        self.errors.is_empty() && !self.update_counts.contains(&batch_cs::EXECUTE_FAILED)
    }

    /// Soma das linhas afetadas pelas mensagens bem-sucedidas (ignora as que
    /// falharam ou não reportaram contagem).
    pub fn total_affected(&self) -> u64 {
        self.update_counts
            .iter()
            .filter(|&&c| c >= 0)
            .map(|&c| c as u64)
            .sum()
    }
}

/// Um erro detalhado de uma mensagem específica do lote.
#[derive(Debug, Clone)]
pub struct BatchError {
    /// Índice (base zero) da mensagem que falhou, na ordem de adição.
    pub message_index: u32,
    /// O erro reportado pelo servidor para essa mensagem.
    pub error: DatabaseError,
}

/// Lê a resposta `op_batch_cs` de um `op_batch_exec`.
///
/// Layout (confirmado por captura, inclusive com erros forçados):
/// `op | stmt | reccount | updates | vectors | errors |`
/// `updates×i32 (contagens) | vectors×(pos u32 + status vector) | errors×u32`.
fn read_batch_cs(conn: &mut Connection) -> Result<BatchResult> {
    let code = read_op(conn.io())?;
    if code == op::RESPONSE {
        // Falha global (não por linha) veio como op_response.
        read_response_body(conn.io())?.into_result()?;
        return Err(Error::protocol(
            "op_batch_exec retornou op_response sem erro",
        ));
    }
    if code != op::BATCH_CS {
        return Err(Error::protocol(format!(
            "esperava op_batch_cs, veio {} ({code})",
            op_name(code)
        )));
    }

    let _stmt = conn.io().read_i32()?;
    let reccount = conn.io().read_i32()? as u32;
    let updates = conn.io().read_i32()? as u32;
    let vectors = conn.io().read_i32()? as u32;
    let errors = conn.io().read_i32()? as u32;

    let mut update_counts = Vec::with_capacity(updates as usize);
    for _ in 0..updates {
        update_counts.push(conn.io().read_i32()?);
    }

    let mut batch_errors = Vec::with_capacity(vectors as usize);
    for _ in 0..vectors {
        let pos = conn.io().read_i32()? as u32;
        let status = read_status_vector(conn.io())?;
        batch_errors.push(BatchError {
            message_index: pos,
            error: DatabaseError::new(status),
        });
    }
    // Lista simples de posições com erro (quando os detalhes não são pedidos).
    for _ in 0..errors {
        let pos = conn.io().read_i32()? as u32;
        if !batch_errors.iter().any(|e| e.message_index == pos) {
            let empty = StatusVector {
                args: Vec::new(),
                sql_state: None,
            };
            batch_errors.push(BatchError {
                message_index: pos,
                error: DatabaseError::new(empty),
            });
        }
    }

    Ok(BatchResult {
        total: reccount,
        update_counts,
        errors: batch_errors,
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    /// Cria um `Batch` "de mentira" (sem conexão/servidor) só para exercitar a
    /// contabilidade EM MEMÓRIA de `add_blob`. `closed: true` evita o aviso de Drop.
    fn fake_blob_batch() -> Batch {
        Batch {
            handle: 0,
            params: Vec::new(),
            pending: Vec::new(),
            pending_count: 0,
            pending_blobs: Vec::new(),
            blob_stream_len: 0,
            pending_regblobs: Vec::new(),
            next_blob_id: 1,
            blob_stream: true,
            charset: crate::charset::Charset::Utf8,
            default_bpb: None,
            blob_segmented: false,
            closed: true,
        }
    }

    /// Bytes que cada blob (não-segmentado) ocupa no buffer do servidor:
    /// `align4(16 + dados)` (cabeçalho de 16 B + dados, alinhado a `BLOB_ALIGN`).
    fn entry_len(data_len: usize) -> usize {
        align_up(16 + data_len, BLOB_ALIGN)
    }

    #[test]
    fn add_blob_atribui_ids_sequenciais_a_partir_de_1() {
        let mut b = fake_blob_batch();
        assert_eq!(b.add_blob(b"a").unwrap(), 1);
        assert_eq!(b.add_blob(b"bb").unwrap(), 2);
        assert_eq!(b.add_blob(b"ccc").unwrap(), 3);
    }

    #[test]
    fn add_blob_exige_coluna_blob() {
        let mut b = fake_blob_batch();
        b.blob_stream = false;
        assert!(b.add_blob(b"x").is_err());
    }

    /// O invariante do qual o flush por bytes depende: `blob_stream_len()` é a soma
    /// de `align4(16 + dados)` de cada blob — inclusive (e principalmente) para
    /// tamanhos NÃO múltiplos de 4, que arredondam pra cima.
    #[test]
    fn blob_stream_len_acumula_tamanhos_alinhados() {
        let mut b = fake_blob_batch();
        for data in [&b"x"[..], &b"yy"[..], &b"zzz"[..], &b"wwww"[..]] {
            b.add_blob(data).unwrap();
        }
        let esperado = entry_len(1) + entry_len(2) + entry_len(3) + entry_len(4);
        assert_eq!(b.blob_stream_len(), esperado);
        // A soma é sempre múltipla de BLOB_ALIGN (cada parcela é).
        assert_eq!(b.blob_stream_len() % BLOB_ALIGN, 0);
    }

    /// Casos pontuais do tamanho declarado para um único blob (cabeçalho 16 B +
    /// dados, alinhado a 4): 30 -> align4(46)=48; 33 -> align4(49)=52.
    #[test]
    fn blob_stream_len_de_um_unico_blob() {
        let mut b = fake_blob_batch();
        b.add_blob(&[0u8; 30]).unwrap();
        assert_eq!(b.blob_stream_len(), 48);

        let mut b = fake_blob_batch();
        b.add_blob(&[0u8; 33]).unwrap();
        assert_eq!(b.blob_stream_len(), 52);
    }

    /// `BatchOptions` por padrão não fixa o buffer (deixa o default do servidor); o
    /// builder `buffer_bytes` registra o limite a enviar como `TAG_BUFFER_BYTES_SIZE`.
    #[test]
    fn batch_options_buffer_bytes() {
        assert_eq!(BatchOptions::new().buffer_bytes, None);
        let opts = BatchOptions::new().buffer_bytes(16 * 1024 * 1024);
        assert_eq!(opts.buffer_bytes, Some(16 * 1024 * 1024));
    }
}