//! OTLP/gRPC sink — batches OTLP protobuf data and delivers it via gRPC to an
//! OpenTelemetry Collector.
//!
//! This sink is designed to work with the [`OtlpEncoder`](crate::encoder::otlp::OtlpEncoder),
//! which writes length-prefixed protobuf `Metric` or `LogRecord` bytes. The sink:
//!
//! 1. Receives raw bytes from the encoder via `write()`.
//! 2. Parses each length-prefixed message and accumulates it in a batch.
//! 3. When the batch reaches `batch_size` entries (or on `flush()`), wraps all
//!    accumulated items into an `ExportMetricsServiceRequest` or
//!    `ExportLogsServiceRequest`, and sends it via gRPC unary call.
//!
//! Requires the `otlp` feature flag.

use std::marker::PhantomData;
use std::time::{Duration, Instant};

use async_trait::async_trait;
use bytes::Buf;
use prost::Message;
use tonic::client::Grpc;
use tonic::codec::{Codec, Decoder, Encoder as TonicEncoder};
use tonic::transport::Channel;
use tonic::Status;

use crate::encoder::otlp::{
    self, ExportLogsServiceRequest, ExportLogsServiceResponse, ExportMetricsServiceRequest,
    ExportMetricsServiceResponse, InstrumentationScope, KeyValue, LogRecord, Metric, Resource,
    ResourceLogs, ResourceMetrics, ScopeLogs, ScopeMetrics,
};
use crate::sink::retry::RetryPolicy;
use crate::sink::Sink;
use crate::SondaError;

/// Default batch size in data point / log record entries — sized so low-rate scenarios flush within seconds.
pub const DEFAULT_BATCH_SIZE: usize = 5;

/// The gRPC service path for the OTLP metrics export RPC.
const METRICS_EXPORT_PATH: &str = "/opentelemetry.proto.collector.metrics.v1.MetricsService/Export";

/// The gRPC service path for the OTLP logs export RPC.
const LOGS_EXPORT_PATH: &str = "/opentelemetry.proto.collector.logs.v1.LogsService/Export";

// ---------------------------------------------------------------------------
// Custom ProstCodec for tonic 0.14+
//
// tonic 0.14 removed the built-in ProstCodec. We implement a minimal codec
// that uses prost for encoding/decoding, matching the pattern from earlier
// tonic versions.
// ---------------------------------------------------------------------------

/// A gRPC codec that uses prost for protobuf encoding and decoding.
///
/// Type parameters `T` and `U` are the request and response message types.
#[derive(Debug, Clone)]
struct OtlpCodec<T, U>(PhantomData<(T, U)>);

impl<T, U> Default for OtlpCodec<T, U> {
    fn default() -> Self {
        Self(PhantomData)
    }
}

impl<T, U> Codec for OtlpCodec<T, U>
where
    T: Message + 'static,
    U: Message + Default + 'static,
{
    type Encode = T;
    type Decode = U;
    type Encoder = OtlpProstEncoder<T>;
    type Decoder = OtlpProstDecoder<U>;

    fn encoder(&mut self) -> Self::Encoder {
        OtlpProstEncoder(PhantomData)
    }

    fn decoder(&mut self) -> Self::Decoder {
        OtlpProstDecoder(PhantomData)
    }
}

/// Prost-based encoder for gRPC request messages.
#[derive(Debug)]
struct OtlpProstEncoder<T>(PhantomData<T>);

impl<T: Message + 'static> TonicEncoder for OtlpProstEncoder<T> {
    type Item = T;
    type Error = Status;

    fn encode(
        &mut self,
        item: Self::Item,
        dst: &mut tonic::codec::EncodeBuf<'_>,
    ) -> Result<(), Self::Error> {
        item.encode(dst)
            .map_err(|e| Status::internal(format!("protobuf encode error: {e}")))
    }
}

/// Prost-based decoder for gRPC response messages.
#[derive(Debug)]
struct OtlpProstDecoder<T>(PhantomData<T>);

impl<T: Message + Default + 'static> Decoder for OtlpProstDecoder<T> {
    type Item = T;
    type Error = Status;

    fn decode(
        &mut self,
        src: &mut tonic::codec::DecodeBuf<'_>,
    ) -> Result<Option<Self::Item>, Self::Error> {
        let buf = src.copy_to_bytes(src.remaining());
        if buf.is_empty() {
            return Ok(None);
        }
        T::decode(buf)
            .map(Some)
            .map_err(|e| Status::internal(format!("protobuf decode error: {e}")))
    }
}

// ---------------------------------------------------------------------------
// Signal type
// ---------------------------------------------------------------------------

/// Selects which OTLP signal type the sink handles.
///
/// Determines both the gRPC path and the parsing/batching strategy.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[cfg_attr(feature = "config", derive(serde::Serialize, serde::Deserialize))]
#[cfg_attr(feature = "config", serde(rename_all = "lowercase"))]
pub enum OtlpSignalType {
    /// Metric signal — data is parsed as OTLP `Metric` messages and sent
    /// to the `MetricsService/Export` gRPC endpoint.
    Metrics,
    /// Log signal — data is parsed as OTLP `LogRecord` messages and sent
    /// to the `LogsService/Export` gRPC endpoint.
    Logs,
}

// ---------------------------------------------------------------------------
// Sink implementation
// ---------------------------------------------------------------------------

/// Delivers OTLP protobuf telemetry to an OpenTelemetry Collector via gRPC.
///
/// Encoded data points accumulate in an internal batch. When the batch reaches
/// `batch_size`, or when `flush()` is called, the batch is wrapped in the
/// appropriate OTLP export request and sent via a tonic gRPC unary call.
pub struct OtlpGrpcSink {
    channel: Channel,
    metric_batch: Vec<Metric>,
    log_batch: Vec<LogRecord>,
    batch_size: usize,
    signal_type: OtlpSignalType,
    resource_attrs: Vec<KeyValue>,
    endpoint: String,
    retry_policy: Option<RetryPolicy>,
    /// `Duration::ZERO` disables time-based flushing.
    max_buffer_age: Duration,
    last_flush_at: Instant,
    last_write_delivered: bool,
}

impl OtlpGrpcSink {
    /// Create a new `OtlpGrpcSink` connected to the given OTLP endpoint.
    ///
    /// # Arguments
    ///
    /// - `endpoint` — the gRPC endpoint URL, e.g. `"http://localhost:4317"`.
    /// - `signal_type` — whether to send metrics or logs.
    /// - `batch_size` — flush threshold in number of data points / log records.
    ///   Use [`DEFAULT_BATCH_SIZE`] if no override is needed.
    /// - `resource_attrs` — key-value pairs for the OTLP `Resource` (typically
    ///   from scenario labels).
    /// - `max_buffer_age` — maximum age a non-empty batch may reach before a
    ///   time-based flush. `Duration::ZERO` disables time-based flushing.
    ///
    pub async fn new(
        endpoint: &str,
        signal_type: OtlpSignalType,
        batch_size: usize,
        resource_attrs: Vec<KeyValue>,
        retry_policy: Option<RetryPolicy>,
        max_buffer_age: Duration,
    ) -> Result<Self, SondaError> {
        let endpoint_str = endpoint.to_owned();
        let channel = Channel::from_shared(endpoint_str.clone())
            .map_err(|e| {
                std::io::Error::other(format!(
                    "otlp grpc sink: invalid endpoint '{}': {}",
                    endpoint_str, e
                ))
            })
            .map_err(SondaError::Sink)?
            .connect()
            .await
            .map_err(|e| {
                std::io::Error::new(
                    std::io::ErrorKind::ConnectionRefused,
                    format!(
                        "otlp grpc sink: failed to connect to '{}': {}",
                        endpoint_str, e
                    ),
                )
            })
            .map_err(SondaError::Sink)?;

        Ok(Self {
            channel,
            metric_batch: Vec::with_capacity(batch_size),
            log_batch: Vec::with_capacity(batch_size),
            batch_size,
            signal_type,
            resource_attrs,
            endpoint: endpoint.to_owned(),
            retry_policy,
            max_buffer_age,
            last_flush_at: Instant::now(),
            last_write_delivered: false,
        })
    }

    /// Build the OTLP `Resource` from the stored resource attributes.
    fn build_resource(&self) -> Resource {
        Resource {
            attributes: self.resource_attrs.clone(),
        }
    }

    /// Build the standard instrumentation scope for Sonda.
    fn build_scope() -> InstrumentationScope {
        InstrumentationScope {
            name: "sonda".to_string(),
            version: env!("CARGO_PKG_VERSION").to_string(),
        }
    }

    async fn flush_metrics(&mut self) -> Result<(), SondaError> {
        if self.metric_batch.is_empty() {
            return Ok(());
        }
        self.last_flush_at = Instant::now();
        let metrics =
            std::mem::replace(&mut self.metric_batch, Vec::with_capacity(self.batch_size));

        let channel = self.channel.clone();
        let endpoint = self.endpoint.clone();
        let resource = self.build_resource();
        let send_once = || async {
            let request = ExportMetricsServiceRequest {
                resource_metrics: vec![ResourceMetrics {
                    resource: Some(resource.clone()),
                    scope_metrics: vec![ScopeMetrics {
                        scope: Some(Self::build_scope()),
                        metrics: metrics.clone(),
                    }],
                }],
            };
            send_grpc_unary::<ExportMetricsServiceRequest, ExportMetricsServiceResponse>(
                channel.clone(),
                &endpoint,
                request,
                METRICS_EXPORT_PATH,
            )
            .await
        };

        run_with_retry(self.retry_policy.as_ref(), send_once).await
    }

    async fn flush_logs(&mut self) -> Result<(), SondaError> {
        if self.log_batch.is_empty() {
            return Ok(());
        }
        self.last_flush_at = Instant::now();
        let log_records =
            std::mem::replace(&mut self.log_batch, Vec::with_capacity(self.batch_size));

        let channel = self.channel.clone();
        let endpoint = self.endpoint.clone();
        let resource = self.build_resource();
        let send_once = || async {
            let request = ExportLogsServiceRequest {
                resource_logs: vec![ResourceLogs {
                    resource: Some(resource.clone()),
                    scope_logs: vec![ScopeLogs {
                        scope: Some(Self::build_scope()),
                        log_records: log_records.clone(),
                    }],
                }],
            };
            send_grpc_unary::<ExportLogsServiceRequest, ExportLogsServiceResponse>(
                channel.clone(),
                &endpoint,
                request,
                LOGS_EXPORT_PATH,
            )
            .await
        };

        run_with_retry(self.retry_policy.as_ref(), send_once).await
    }

    /// Classify whether a gRPC error is retryable.
    ///
    /// Transport errors (connection refused, broken pipe) and gRPC status
    /// codes UNAVAILABLE, DEADLINE_EXCEEDED, and RESOURCE_EXHAUSTED are
    /// retryable.
    fn is_retryable(err: &SondaError) -> bool {
        if let SondaError::Sink(io_err) = err {
            let msg = io_err.to_string();
            // UNAVAILABLE, DEADLINE_EXCEEDED, RESOURCE_EXHAUSTED in gRPC
            // status codes are retryable. Also transport errors.
            if msg.contains("UNAVAILABLE")
                || msg.contains("DEADLINE_EXCEEDED")
                || msg.contains("RESOURCE_EXHAUSTED")
                || io_err.kind() == std::io::ErrorKind::ConnectionRefused
                || io_err.kind() == std::io::ErrorKind::BrokenPipe
            {
                return true;
            }
            // Generic transport errors are retryable.
            if msg.contains("not ready") || msg.contains("connection") {
                return true;
            }
        }
        false
    }
}

async fn send_grpc_unary<T, U>(
    channel: Channel,
    endpoint: &str,
    request: T,
    path: &'static str,
) -> Result<(), SondaError>
where
    T: Message + 'static,
    U: Message + Default + 'static,
{
    let mut client = Grpc::new(channel);
    client
        .ready()
        .await
        .map_err(|e| {
            std::io::Error::new(
                std::io::ErrorKind::ConnectionRefused,
                format!("otlp grpc sink: service not ready at '{}': {}", endpoint, e),
            )
        })
        .map_err(SondaError::Sink)?;

    let grpc_path = http::uri::PathAndQuery::from_static(path);
    let codec: OtlpCodec<T, U> = OtlpCodec::default();
    let tonic_request = tonic::Request::new(request);

    client
        .unary(tonic_request, grpc_path, codec)
        .await
        .map_err(|e| {
            std::io::Error::new(
                std::io::ErrorKind::BrokenPipe,
                format!("otlp grpc sink: gRPC call to '{}' failed: {}", endpoint, e),
            )
        })
        .map_err(SondaError::Sink)?;
    Ok(())
}

async fn run_with_retry<F, Fut>(
    policy: Option<&RetryPolicy>,
    mut send_once: F,
) -> Result<(), SondaError>
where
    F: FnMut() -> Fut,
    Fut: std::future::Future<Output = Result<(), SondaError>>,
{
    let mut last_error = match send_once().await {
        Ok(()) => return Ok(()),
        Err(e) => e,
    };
    let Some(policy) = policy else {
        return Err(last_error);
    };
    for attempt in 0..policy.max_attempts() {
        if !OtlpGrpcSink::is_retryable(&last_error) {
            return Err(last_error);
        }
        let backoff = policy.jittered_backoff(attempt);
        eprintln!(
            "sonda: retry {}/{} after {}ms (error: {})",
            attempt + 1,
            policy.max_attempts(),
            backoff.as_millis(),
            last_error,
        );
        tokio::time::sleep(backoff).await;
        match send_once().await {
            Ok(()) => return Ok(()),
            Err(e) => last_error = e,
        }
    }
    eprintln!(
        "sonda: all {} retries exhausted (last error: {})",
        policy.max_attempts(),
        last_error,
    );
    Err(last_error)
}

#[async_trait]
impl Sink for OtlpGrpcSink {
    async fn write(&mut self, data: &[u8]) -> Result<(), SondaError> {
        match self.signal_type {
            OtlpSignalType::Metrics => {
                let metrics = otlp::parse_length_prefixed_metrics(data)?;
                self.metric_batch.extend(metrics);
                let size_reached = self.metric_batch.len() >= self.batch_size;
                let age_reached = !self.max_buffer_age.is_zero()
                    && self.last_flush_at.elapsed() >= self.max_buffer_age;
                let should_flush = size_reached || age_reached;
                if should_flush {
                    self.flush_metrics().await?;
                }
                self.last_write_delivered = should_flush;
            }
            OtlpSignalType::Logs => {
                let records = otlp::parse_length_prefixed_log_records(data)?;
                self.log_batch.extend(records);
                let size_reached = self.log_batch.len() >= self.batch_size;
                let age_reached = !self.max_buffer_age.is_zero()
                    && self.last_flush_at.elapsed() >= self.max_buffer_age;
                let should_flush = size_reached || age_reached;
                if should_flush {
                    self.flush_logs().await?;
                }
                self.last_write_delivered = should_flush;
            }
        }
        Ok(())
    }

    async fn flush(&mut self) -> Result<(), SondaError> {
        match self.signal_type {
            OtlpSignalType::Metrics => self.flush_metrics().await,
            OtlpSignalType::Logs => self.flush_logs().await,
        }
    }

    fn last_write_delivered(&self) -> bool {
        self.last_write_delivered
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::encoder::otlp::{
        self, any_value, metric, number_data_point, AnyValue, Gauge, Metric, NumberDataPoint,
    };
    use crate::sink::{create_sink, SinkConfig};

    // -----------------------------------------------------------------------
    // Helpers
    // -----------------------------------------------------------------------

    /// Build a sink with a lazy (non-connecting) channel pointed at a dead address.
    fn lazy_sink(
        signal_type: OtlpSignalType,
        batch_size: usize,
        max_buffer_age: Duration,
    ) -> OtlpGrpcSink {
        let channel = tonic::transport::Endpoint::from_static("http://127.0.0.1:1").connect_lazy();
        OtlpGrpcSink {
            channel,
            metric_batch: Vec::with_capacity(batch_size),
            log_batch: Vec::with_capacity(batch_size),
            batch_size,
            signal_type,
            resource_attrs: vec![],
            endpoint: "http://127.0.0.1:1".to_string(),
            retry_policy: None,
            max_buffer_age,
            last_flush_at: Instant::now(),
            last_write_delivered: false,
        }
    }

    /// Length-prefixed bytes for a single OTLP metric, as the encoder produces.
    fn one_metric_bytes(name: &str) -> Vec<u8> {
        let metric = Metric {
            name: name.to_string(),
            description: String::new(),
            unit: String::new(),
            data: Some(metric::Data::Gauge(Gauge {
                data_points: vec![NumberDataPoint {
                    attributes: vec![],
                    time_unix_nano: 1_700_000_000_000_000_000,
                    value: Some(number_data_point::Value::AsDouble(1.0)),
                }],
            })),
        };
        let proto = metric.encode_to_vec();
        let mut buf = (proto.len() as u32).to_le_bytes().to_vec();
        buf.extend_from_slice(&proto);
        buf
    }

    /// Length-prefixed bytes for a single OTLP log record.
    fn one_log_bytes() -> Vec<u8> {
        let record = otlp::LogRecord {
            time_unix_nano: 1_700_000_000_000_000_000,
            severity_number: 9,
            severity_text: "INFO".to_string(),
            body: Some(AnyValue {
                value: Some(any_value::Value::StringValue("hello".to_string())),
            }),
            attributes: vec![],
        };
        let proto = record.encode_to_vec();
        let mut buf = (proto.len() as u32).to_le_bytes().to_vec();
        buf.extend_from_slice(&proto);
        buf
    }

    // -----------------------------------------------------------------------
    // Constants
    // -----------------------------------------------------------------------

    #[test]
    fn default_batch_size_is_5() {
        assert_eq!(DEFAULT_BATCH_SIZE, 5);
    }

    // -----------------------------------------------------------------------
    // OtlpSignalType
    // -----------------------------------------------------------------------

    #[test]
    fn signal_type_is_cloneable_and_debuggable() {
        let st = OtlpSignalType::Metrics;
        let cloned = st;
        assert_eq!(cloned, OtlpSignalType::Metrics);
        let s = format!("{st:?}");
        assert!(s.contains("Metrics"));
    }

    #[cfg(feature = "config")]
    #[test]
    fn signal_type_deserializes_metrics() {
        let json = "\"metrics\"";
        let st: OtlpSignalType = serde_json::from_str(json).expect("deser ok");
        assert_eq!(st, OtlpSignalType::Metrics);
    }

    #[cfg(feature = "config")]
    #[test]
    fn signal_type_deserializes_logs() {
        let json = "\"logs\"";
        let st: OtlpSignalType = serde_json::from_str(json).expect("deser ok");
        assert_eq!(st, OtlpSignalType::Logs);
    }

    // -----------------------------------------------------------------------
    // Send + Sync contract
    // -----------------------------------------------------------------------

    #[test]
    fn otlp_grpc_sink_is_send_and_sync() {
        fn assert_send_sync<T: Send + Sync>() {}
        assert_send_sync::<OtlpGrpcSink>();
    }

    // -----------------------------------------------------------------------
    // SinkConfig deserialization
    // -----------------------------------------------------------------------

    #[cfg(feature = "config")]
    #[test]
    fn sink_config_otlp_grpc_deserializes_with_all_fields() {
        let yaml = r#"
type: otlp_grpc
endpoint: "http://localhost:4317"
signal_type: metrics
batch_size: 50
"#;
        let config: SinkConfig = serde_yaml_ng::from_str(yaml).expect("deser ok");
        match config {
            SinkConfig::OtlpGrpc {
                endpoint,
                signal_type,
                batch_size,
                ..
            } => {
                assert_eq!(endpoint, "http://localhost:4317");
                assert_eq!(signal_type, OtlpSignalType::Metrics);
                assert_eq!(batch_size, Some(50));
            }
            other => panic!("expected OtlpGrpc, got {other:?}"),
        }
    }

    #[cfg(feature = "config")]
    #[test]
    fn sink_config_otlp_grpc_batch_size_is_optional() {
        let yaml = r#"
type: otlp_grpc
endpoint: "http://localhost:4317"
signal_type: logs
"#;
        let config: SinkConfig = serde_yaml_ng::from_str(yaml).expect("deser ok");
        match config {
            SinkConfig::OtlpGrpc {
                batch_size,
                signal_type,
                ..
            } => {
                assert!(batch_size.is_none());
                assert_eq!(signal_type, OtlpSignalType::Logs);
            }
            other => panic!("expected OtlpGrpc, got {other:?}"),
        }
    }

    #[cfg(feature = "config")]
    #[test]
    fn sink_config_otlp_grpc_requires_endpoint() {
        let yaml = "type: otlp_grpc\nsignal_type: metrics";
        let result: Result<SinkConfig, _> = serde_yaml_ng::from_str(yaml);
        assert!(
            result.is_err(),
            "otlp_grpc without endpoint should fail deserialization"
        );
    }

    #[cfg(feature = "config")]
    #[test]
    fn sink_config_otlp_grpc_requires_signal_type() {
        let yaml = "type: otlp_grpc\nendpoint: \"http://localhost:4317\"";
        let result: Result<SinkConfig, _> = serde_yaml_ng::from_str(yaml);
        assert!(
            result.is_err(),
            "otlp_grpc without signal_type should fail deserialization"
        );
    }

    #[test]
    fn sink_config_otlp_grpc_is_cloneable_and_debuggable() {
        let config = SinkConfig::OtlpGrpc {
            endpoint: "http://localhost:4317".to_string(),
            signal_type: OtlpSignalType::Metrics,
            batch_size: Some(100),
            max_buffer_age: None,
            retry: None,
        };
        let cloned = config.clone();
        let s = format!("{cloned:?}");
        assert!(s.contains("OtlpGrpc"));
        assert!(s.contains("4317"));
    }

    // -----------------------------------------------------------------------
    // Request construction helpers (unit tests without network)
    // -----------------------------------------------------------------------

    /// Verify that `ExportMetricsServiceRequest` wrapping produces valid protobuf.
    #[test]
    fn export_metrics_request_wraps_metrics_correctly() {
        let metric = Metric {
            name: "test_gauge".to_string(),
            description: String::new(),
            unit: String::new(),
            data: Some(metric::Data::Gauge(Gauge {
                data_points: vec![NumberDataPoint {
                    attributes: vec![],
                    time_unix_nano: 1_700_000_000_000_000_000,
                    value: Some(number_data_point::Value::AsDouble(42.0)),
                }],
            })),
        };

        let attrs = vec![otlp::KeyValue {
            key: "service.name".to_string(),
            value: Some(AnyValue {
                value: Some(any_value::Value::StringValue("sonda".to_string())),
            }),
        }];

        let req = ExportMetricsServiceRequest {
            resource_metrics: vec![ResourceMetrics {
                resource: Some(Resource { attributes: attrs }),
                scope_metrics: vec![ScopeMetrics {
                    scope: Some(InstrumentationScope {
                        name: "sonda".to_string(),
                        version: "test".to_string(),
                    }),
                    metrics: vec![metric],
                }],
            }],
        };

        // Roundtrip through protobuf
        let mut buf = Vec::new();
        req.encode(&mut buf).expect("encode");
        let decoded = ExportMetricsServiceRequest::decode(buf.as_slice()).expect("decode");
        assert_eq!(decoded.resource_metrics.len(), 1);
        assert_eq!(
            decoded.resource_metrics[0].scope_metrics[0].metrics.len(),
            1
        );
        assert_eq!(
            decoded.resource_metrics[0].scope_metrics[0].metrics[0].name,
            "test_gauge"
        );
    }

    /// Verify that `ExportLogsServiceRequest` wrapping produces valid protobuf.
    #[test]
    fn export_logs_request_wraps_log_records_correctly() {
        let record = otlp::LogRecord {
            time_unix_nano: 1_700_000_000_000_000_000,
            severity_number: 9,
            severity_text: "INFO".to_string(),
            body: Some(AnyValue {
                value: Some(any_value::Value::StringValue("hello".to_string())),
            }),
            attributes: vec![],
        };

        let req = ExportLogsServiceRequest {
            resource_logs: vec![ResourceLogs {
                resource: Some(Resource { attributes: vec![] }),
                scope_logs: vec![ScopeLogs {
                    scope: Some(InstrumentationScope {
                        name: "sonda".to_string(),
                        version: "test".to_string(),
                    }),
                    log_records: vec![record],
                }],
            }],
        };

        let mut buf = Vec::new();
        req.encode(&mut buf).expect("encode");
        let decoded = ExportLogsServiceRequest::decode(buf.as_slice()).expect("decode");
        assert_eq!(decoded.resource_logs.len(), 1);
        assert_eq!(decoded.resource_logs[0].scope_logs[0].log_records.len(), 1);
        assert_eq!(
            decoded.resource_logs[0].scope_logs[0].log_records[0].severity_text,
            "INFO"
        );
    }

    // -----------------------------------------------------------------------
    // Construction failure: unreachable endpoint
    // -----------------------------------------------------------------------

    #[tokio::test]
    #[ignore = "requires network timeout; run with --ignored when desired"]
    async fn new_with_unreachable_endpoint_returns_sink_error() {
        let result = OtlpGrpcSink::new(
            "http://127.0.0.1:1",
            OtlpSignalType::Metrics,
            DEFAULT_BATCH_SIZE,
            vec![],
            None,
            Duration::ZERO,
        )
        .await;
        match result {
            Err(err) => {
                let msg = err.to_string();
                assert!(
                    msg.contains("127.0.0.1:1") || msg.contains("otlp"),
                    "error should reference the endpoint: {msg}"
                );
            }
            Ok(_) => panic!("construction must fail when endpoint is unreachable"),
        }
    }

    #[tokio::test]
    async fn new_with_invalid_endpoint_returns_error() {
        let result = OtlpGrpcSink::new(
            "not a url",
            OtlpSignalType::Metrics,
            DEFAULT_BATCH_SIZE,
            vec![],
            None,
            Duration::ZERO,
        )
        .await;
        assert!(result.is_err(), "invalid endpoint URL must be rejected");
    }

    // -----------------------------------------------------------------------
    // Full scenario YAML: otlp_grpc sink variant
    // -----------------------------------------------------------------------

    #[cfg(feature = "config")]
    #[test]
    fn scenario_yaml_with_otlp_metrics_deserializes() {
        use crate::config::ScenarioConfig;
        use crate::encoder::EncoderConfig;

        let yaml = r#"
name: otlp_test
rate: 10.0
generator:
  type: constant
  value: 1.0
encoder:
  type: otlp
sink:
  type: otlp_grpc
  endpoint: "http://localhost:4317"
  signal_type: metrics
"#;
        let config: ScenarioConfig = serde_yaml_ng::from_str(yaml).expect("deser ok");
        assert_eq!(config.name, "otlp_test");
        assert!(matches!(config.encoder, EncoderConfig::Otlp));
        assert!(matches!(
            config.sink,
            SinkConfig::OtlpGrpc {
                ref endpoint,
                signal_type: OtlpSignalType::Metrics,
                ..
            } if endpoint == "http://localhost:4317"
        ));
    }

    #[cfg(feature = "config")]
    #[test]
    fn scenario_yaml_with_otlp_logs_deserializes() {
        use crate::config::LogScenarioConfig;
        use crate::encoder::EncoderConfig;

        let yaml = r#"
name: otlp_logs_test
rate: 5.0
generator:
  type: template
  templates:
    - message: "Request processed"
encoder:
  type: otlp
sink:
  type: otlp_grpc
  endpoint: "http://localhost:4317"
  signal_type: logs
  batch_size: 50
"#;
        let config: LogScenarioConfig = serde_yaml_ng::from_str(yaml).expect("deser ok");
        assert_eq!(config.name, "otlp_logs_test");
        assert!(matches!(config.encoder, EncoderConfig::Otlp));
        match &config.sink {
            SinkConfig::OtlpGrpc {
                endpoint,
                signal_type,
                batch_size,
                ..
            } => {
                assert_eq!(endpoint, "http://localhost:4317");
                assert_eq!(*signal_type, OtlpSignalType::Logs);
                assert_eq!(*batch_size, Some(50));
            }
            other => panic!("expected OtlpGrpc, got {other:?}"),
        }
    }

    // -----------------------------------------------------------------------
    // Time-based flush
    // -----------------------------------------------------------------------

    #[tokio::test]
    async fn time_based_flush_fires_when_buffer_age_exceeded() {
        let mut sink = lazy_sink(
            OtlpSignalType::Metrics,
            1_000_000,
            Duration::from_millis(50),
        );

        sink.write(&one_metric_bytes("first"))
            .await
            .expect("write 1");
        assert_eq!(sink.metric_batch.len(), 1, "first write only buffers");

        tokio::time::sleep(Duration::from_millis(200)).await;
        let result = sink.write(&one_metric_bytes("second")).await;
        assert!(
            result.is_err(),
            "second write past max_buffer_age must attempt a flush"
        );
        assert!(
            sink.metric_batch.is_empty(),
            "the time-based flush must have drained the batch"
        );
    }

    #[tokio::test]
    async fn time_based_flush_fires_for_log_signal() {
        let mut sink = lazy_sink(OtlpSignalType::Logs, 1_000_000, Duration::from_millis(50));

        sink.write(&one_log_bytes()).await.expect("write 1");
        assert_eq!(sink.log_batch.len(), 1, "first write only buffers");

        tokio::time::sleep(Duration::from_millis(200)).await;
        let result = sink.write(&one_log_bytes()).await;
        assert!(
            result.is_err(),
            "second log write past max_buffer_age must attempt a flush"
        );
        assert!(
            sink.log_batch.is_empty(),
            "the time-based flush must have drained the log batch"
        );
    }

    #[tokio::test]
    async fn zero_max_buffer_age_disables_time_based_flush() {
        let mut sink = lazy_sink(OtlpSignalType::Metrics, 1_000_000, Duration::ZERO);

        sink.write(&one_metric_bytes("first"))
            .await
            .expect("write 1");
        tokio::time::sleep(Duration::from_millis(150)).await;
        sink.write(&one_metric_bytes("second"))
            .await
            .expect("write 2");
        tokio::time::sleep(Duration::from_millis(150)).await;
        sink.write(&one_metric_bytes("third"))
            .await
            .expect("write 3");

        assert_eq!(
            sink.metric_batch.len(),
            3,
            "zero max_buffer_age must disable time-based flush"
        );
    }

    #[tokio::test]
    async fn size_triggered_flush_still_works_and_resets_the_timer() {
        let mut sink = lazy_sink(OtlpSignalType::Metrics, 2, Duration::from_secs(60));

        sink.write(&one_metric_bytes("a"))
            .await
            .expect("write 1 buffers");
        assert_eq!(sink.metric_batch.len(), 1, "below batch_size: no flush");

        let result = sink.write(&one_metric_bytes("b")).await;
        assert!(result.is_err(), "filling the batch must attempt a flush");
        assert!(
            sink.metric_batch.is_empty(),
            "size-triggered flush must drain the batch"
        );

        sink.write(&one_metric_bytes("c"))
            .await
            .expect("partial write after a size flush must not time-flush immediately");
        assert_eq!(sink.metric_batch.len(), 1, "partial write only buffers");
    }

    #[tokio::test]
    async fn last_write_delivered_is_false_when_metric_write_only_buffers() {
        let mut sink = lazy_sink(OtlpSignalType::Metrics, 1_000_000, Duration::ZERO);

        sink.write(&one_metric_bytes("buffered"))
            .await
            .expect("write 1");

        assert!(
            !sink.last_write_delivered(),
            "a metric write that only buffers must report last_write_delivered() == false"
        );
    }

    #[tokio::test]
    async fn last_write_delivered_is_false_when_log_write_only_buffers() {
        let mut sink = lazy_sink(OtlpSignalType::Logs, 1_000_000, Duration::ZERO);

        sink.write(&one_log_bytes()).await.expect("write 1");

        assert!(
            !sink.last_write_delivered(),
            "a log write that only buffers must report last_write_delivered() == false"
        );
    }

    #[tokio::test]
    async fn create_sink_otlp_grpc_with_invalid_max_buffer_age_returns_err() {
        let config = SinkConfig::OtlpGrpc {
            endpoint: "http://127.0.0.1:1".to_string(),
            signal_type: OtlpSignalType::Metrics,
            batch_size: None,
            max_buffer_age: Some("garbage".to_string()),
            retry: None,
        };
        assert!(
            create_sink(&config, None).await.is_err(),
            "invalid max_buffer_age must cause the factory to fail"
        );
    }

    #[cfg(feature = "config")]
    #[test]
    fn sink_config_otlp_grpc_deserializes_with_max_buffer_age() {
        let yaml = r#"
type: otlp_grpc
endpoint: "http://localhost:4317"
signal_type: metrics
max_buffer_age: 10s
"#;
        let config: SinkConfig = serde_yaml_ng::from_str(yaml).expect("deser ok");
        match config {
            SinkConfig::OtlpGrpc { max_buffer_age, .. } => {
                assert_eq!(max_buffer_age.as_deref(), Some("10s"));
            }
            other => panic!("expected OtlpGrpc, got {other:?}"),
        }
    }

    #[cfg(feature = "config")]
    #[test]
    fn sink_config_otlp_grpc_max_buffer_age_defaults_to_none() {
        let yaml = r#"
type: otlp_grpc
endpoint: "http://localhost:4317"
signal_type: metrics
"#;
        let config: SinkConfig = serde_yaml_ng::from_str(yaml).expect("deser ok");
        match config {
            SinkConfig::OtlpGrpc { max_buffer_age, .. } => {
                assert!(max_buffer_age.is_none(), "max_buffer_age defaults to None");
            }
            other => panic!("expected OtlpGrpc, got {other:?}"),
        }
    }
}
