//! Bounded, asynchronous persistence for data-plane invocation telemetry.
//!
//! Invocation responses must not wait for libSQL, so producers use
//! [`tokio::sync::mpsc::Sender::try_send`] and one task writes ordered batches.
//! When the queue is full or closed, the update is deliberately dropped and a
//! warning is emitted. These counters and events are operational telemetry;
//! they are not suitable as a billing ledger or audit log.

use std::sync::Arc;

use tokio::sync::{Mutex, mpsc, oneshot};

use crate::persistence::{InvocationUpdate, Persistence};

const QUEUE_CAPACITY: usize = 1024;
const MAX_BATCH_SIZE: usize = 64;

/// Non-blocking producer used by Wasmtime invocation threads.
///
/// All producers share one FIFO queue. Queue order is preserved for accepted
/// records, but a process crash can lose records that have not yet reached
/// libSQL.
pub(crate) struct InvocationPersistence {
    sender: mpsc::Sender<Command>,
}

enum Command {
    Record(InvocationUpdate),
    Flush(oneshot::Sender<()>),
}

impl InvocationPersistence {
    /// Starts the single ordered writer on the current Tokio runtime.
    ///
    /// `sync` is shared with full Console snapshot persistence. Taking it
    /// around each batch prevents a snapshot transaction from racing a metric
    /// delta transaction.
    pub(crate) fn start(persistence: Persistence, sync: Arc<Mutex<()>>) -> Self {
        let (sender, mut receiver) = mpsc::channel(QUEUE_CAPACITY);
        tokio::spawn(async move {
            while let Some(first) = receiver.recv().await {
                let mut batch = Vec::with_capacity(MAX_BATCH_SIZE);
                let mut flushes = Vec::new();
                match first {
                    Command::Record(update) => batch.push(update),
                    Command::Flush(flush) => flushes.push(flush),
                }
                while batch.len() < MAX_BATCH_SIZE {
                    match receiver.try_recv() {
                        Ok(Command::Record(update)) => batch.push(update),
                        Ok(Command::Flush(flush)) => flushes.push(flush),
                        Err(_) => break,
                    }
                }

                if !batch.is_empty() {
                    let _sync = sync.lock().await;
                    if let Err(error) = persistence.record_invocations(batch).await {
                        tracing::error!(%error, "failed to persist invocation telemetry");
                    }
                }
                // Flush is an ordering barrier, not a durability receipt. A
                // database error above is logged rather than returned, and a
                // flush may wait for records dequeued after the marker when
                // those records fit in the same batch.
                for flush in flushes {
                    let _ = flush.send(());
                }
            }
        });
        Self { sender }
    }

    /// Enqueues telemetry without extending API response latency.
    ///
    /// Saturation drops this update instead of applying backpressure to the
    /// Component invocation path.
    pub(crate) fn record(&self, update: InvocationUpdate) {
        if let Err(error) = self.sender.try_send(Command::Record(update)) {
            tracing::warn!(%error, "invocation telemetry queue is unavailable; update dropped");
        }
    }

    /// Waits until all telemetry queued before this call has been attempted.
    ///
    /// This method is used before a control-plane snapshot so the snapshot
    /// reads the newest in-memory counters after older delta writes. It does
    /// not report persistence failures; those are visible through tracing.
    pub(crate) async fn flush(&self) {
        let (sender, receiver) = oneshot::channel();
        if self.sender.send(Command::Flush(sender)).await.is_ok() {
            let _ = receiver.await;
        }
    }
}