← snapshot

//! The background task that batches envelopes into NDJSON POSTs.
//!
//! Invariant: this task never propagates a failure to its caller and never
//! logs at error level. A telemetry outage must not become an outage, and
//! logging an ingest failure at error level would feed the very log-shipping
//! path that reports errors, looping the shipper into shipping itself.

use std::time::Duration;

use tokio::sync::mpsc::Receiver;
use tokio::time::interval;

use crate::S10Config;
use crate::envelope::Envelope;

/// Bounded queue capacity: enough to absorb a burst without unbounded
/// memory growth. A full queue drops the newest envelope rather than
/// blocking the caller — `try_send` in `Telemetry::emit`.
pub(crate) const QUEUE_CAP: usize = 2048;

const BATCH_SIZE: usize = 64;
const FLUSH_INTERVAL: Duration = Duration::from_secs(2);

pub(crate) async fn run(config: S10Config, mut rx: Receiver<Envelope>) {
    let client = reqwest::Client::new();
    let endpoint = format!("{}/v2/envelopes", config.url.trim_end_matches('/'));
    let mut ticker = interval(FLUSH_INTERVAL);
    let mut batch = Vec::with_capacity(BATCH_SIZE);

    loop {
        tokio::select! {
            received = rx.recv() => {
                let Some(envelope) = received else {
                    // The last `Telemetry` handle was dropped: flush what's
                    // left and stop.
                    flush(&client, &endpoint, &config.key, &mut batch).await;
                    return;
                };
                batch.push(envelope);
                if batch.len() >= BATCH_SIZE {
                    flush(&client, &endpoint, &config.key, &mut batch).await;
                }
            }
            _ = ticker.tick() => {
                flush(&client, &endpoint, &config.key, &mut batch).await;
            }
        }
    }
}

async fn flush(client: &reqwest::Client, endpoint: &str, key: &str, batch: &mut Vec<Envelope>) {
    if batch.is_empty() {
        return;
    }
    let body = batch
        .iter()
        .map(Envelope::to_ndjson_line)
        .collect::<Vec<_>>()
        .join("\n");
    batch.clear();

    // Best-effort: a dropped or non-2xx response is silently absorbed. The
    // ingest's own contract (a wrong key 401s silently) already means a
    // caller cannot distinguish "delivered" from "misconfigured" without
    // checking the console — telemetry must never gate the product on that.
    let _ = client
        .post(endpoint)
        .header("Authorization", format!("Bearer {key}"))
        .header("Content-Type", "application/x-ndjson")
        .body(body)
        .send()
        .await;
}