cedarling 0.0.8

// This software is available under the Apache-2.0 license.
// See https://www.apache.org/licenses/LICENSE-2.0.txt for full text.
//
// Copyright (c) 2024, Gluu, Inc.

//! gRPC transport implementation for Lock Server communication.

#[cfg(not(target_arch = "wasm32"))]
use std::time::Duration;

use async_trait::async_trait;
use prost_types::Timestamp;
#[cfg(not(target_arch = "wasm32"))]
use tonic::transport::{Channel, ClientTlsConfig};
use tonic::{Request, metadata::MetadataValue};
#[cfg(target_arch = "wasm32")]
use tonic_web_wasm_client::Client;

use crate::{
    lock::{
        LockLogEntry,
        proto::{
            BulkHealthRequest, BulkLogRequest, BulkTelemetryRequest, HealthEntry, LogEntry,
            TelemetryEntry, audit_service_client::AuditServiceClient,
        },
        transport::{
            self, AuditKind, AuditTransport, SerializedAuditEntry, TransportError, TransportResult,
            mapping::{
                CedarlingLogEntry, CedarlingMetricsEntry, LockServerHealthEntry,
                LockServerLogEntry, LockServerMetricsEntry,
            },
        },
    },
    log::{LogWriter, Logger},
};

#[cfg(not(target_arch = "wasm32"))]
/// Duration to wait for each gRPC request
const GRPC_REQUEST_TIMEOUT: Duration = Duration::from_secs(10);

pub(crate) struct GrpcTransport {
    #[cfg(not(target_arch = "wasm32"))]
    client: AuditServiceClient<Channel>,
    #[cfg(target_arch = "wasm32")]
    client: AuditServiceClient<Client>,
    access_token: String,
    logger: Option<Logger>,
}

impl GrpcTransport {
    #[cfg_attr(target_arch = "wasm32", allow(clippy::unnecessary_wraps))]
    /// Constructs a new [`GrpcTransport`]
    pub(crate) fn new(
        endpoint: impl Into<String>,
        access_token: &str,
        logger: Option<Logger>,
    ) -> Result<Self, TransportError> {
        #[cfg(target_arch = "wasm32")]
        let client = AuditServiceClient::new(Client::new(endpoint.into()));

        #[cfg(not(target_arch = "wasm32"))]
        let client = AuditServiceClient::new(
            Channel::from_shared(endpoint.into())
                .map_err(|e| TransportError::InvalidUri(format!("failed to construct URI: {e:?}")))?
                .tls_config(ClientTlsConfig::new().with_native_roots())?
                .timeout(GRPC_REQUEST_TIMEOUT)
                .connect_lazy(),
        );

        Ok(Self {
            client,
            access_token: access_token.into(),
            logger,
        })
    }
}

#[cfg_attr(not(any(target_arch = "wasm32", target_arch = "wasm64")), async_trait)]
#[cfg_attr(any(target_arch = "wasm32", target_arch = "wasm64"), async_trait(?Send))]
impl AuditTransport for GrpcTransport {
    async fn send(
        &self,
        entries: &[SerializedAuditEntry],
        audit_kind: &AuditKind,
    ) -> TransportResult<()> {
        if entries.is_empty() {
            return Ok(());
        }

        let token: MetadataValue<_> = format!("Bearer {}", self.access_token).parse()?;
        let mut client = self.client.clone();
        let warn = |msg| self.logger.log_any(LockLogEntry::warn(msg));

        let response = match audit_kind {
            AuditKind::Log(_) => {
                let entries = transport::deserialize_entries::<
                    LockServerLogEntry,
                    CedarlingLogEntry,
                >(entries, "log", warn)?
                .into_iter()
                .map(log_json_to_proto)
                .collect();

                let mut request = Request::new(BulkLogRequest { entries });
                request
                    .metadata_mut()
                    .insert("authorization", token.clone());

                client.process_bulk_log(request).await?
            },
            AuditKind::Telemetry(_) => {
                let entries = transport::deserialize_entries::<
                    LockServerMetricsEntry,
                    CedarlingMetricsEntry,
                >(entries, "telemetry", warn)?
                .into_iter()
                .map(telemetry_json_to_proto)
                .collect();

                let mut request = Request::new(BulkTelemetryRequest { entries });
                request
                    .metadata_mut()
                    .insert("authorization", token.clone());

                client.process_bulk_telemetry(request).await?
            },
            AuditKind::Health(_) => {
                let entries: Vec<LockServerHealthEntry> = entries
                    .iter()
                    .enumerate()
                    .filter_map(|(idx, v)| match serde_json::from_str(v) {
                        Ok(e) => Some(e),
                        Err(e) => {
                            self.logger.log_any(LockLogEntry::warn(format!(
                                "failed to parse health entry[{idx}]: {e}"
                            )));
                            None
                        },
                    })
                    .collect();

                if entries.is_empty() {
                    return Err(TransportError::Serialization(
                        "all health entries were malformed, nothing to send".to_string(),
                    ));
                }

                let proto_entries: Vec<HealthEntry> =
                    entries.into_iter().map(health_json_to_proto).collect();

                let mut request = Request::new(BulkHealthRequest {
                    entries: proto_entries,
                });
                request.metadata_mut().insert("authorization", token);

                client.process_bulk_health(request).await?
            },
        };

        let inner = response.into_inner();
        if !inner.success {
            return Err(TransportError::GrpcServer(format!(
                "Server reported failure: {}",
                inner.message
            )));
        }

        Ok(())
    }
}

/// Converts a [`LockServerLogEntry`] into a [`LogEntry`]
fn log_json_to_proto(entry: LockServerLogEntry) -> LogEntry {
    LogEntry {
        creation_date: parse_timestamp(&entry.creation_date),
        event_time: parse_timestamp(&entry.event_time),
        service: entry.service.unwrap_or_default(),
        node_name: entry.node_name,
        event_type: entry.event_type,
        severity_level: entry.severity_level.unwrap_or_default(),
        action: entry.action,
        decision_result: entry.decision_result,
        requested_resource: entry.requested_resource,
        principal_id: entry.principal_id.unwrap_or_default(),
        client_id: entry.client_id.unwrap_or_default(),
        jti: String::new(),
        context_information: entry
            .context_information
            .and_then(|v| v.as_object().cloned())
            .unwrap_or_default()
            .into_iter()
            .map(|(k, v)| (k, v.as_str().map_or_else(|| v.to_string(), str::to_owned)))
            .collect(),
    }
}

/// Converts a [`LockServerMetricsEntry`] into a [`TelemetryEntry`]
fn telemetry_json_to_proto(entry: LockServerMetricsEntry) -> TelemetryEntry {
    TelemetryEntry {
        creation_date: parse_timestamp(&entry.creation_date),
        service: entry.service.unwrap_or_default(),
        node_name: entry.node_name,
        status: entry.status,
        policy_stats: entry.policy_stats,
        error_counters: entry.error_counters,
        operational_stats: entry.operational_stats,
        interval_secs: entry.interval_secs,
    }
}

/// Converts a [`LockServerHealthEntry`] into a [`HealthEntry`]
fn health_json_to_proto(entry: LockServerHealthEntry) -> HealthEntry {
    HealthEntry {
        creation_date: parse_timestamp(&entry.creation_date),
        event_time: parse_timestamp(&entry.event_time),
        service: entry.service,
        node_name: entry.node_name,
        status: entry.status,
        engine_status: entry
            .engine_status
            .into_iter()
            .map(|(k, v)| (k, v.to_string()))
            .collect(),
    }
}

// Parse a RFC3339 timestamp string into a protobuf Timestamp
fn parse_timestamp(s: &str) -> Option<Timestamp> {
    chrono::DateTime::parse_from_rfc3339(s)
        .ok()
        .map(|dt| Timestamp {
            seconds: dt.timestamp(),
            nanos: dt.timestamp_subsec_nanos().cast_signed(),
        })
}

#[cfg(test)]
mod test {
    use std::net::SocketAddr;

    use super::*;
    use serde_json::json;
    use tokio::{net::TcpListener, sync::mpsc};
    use tokio_stream::wrappers::TcpListenerStream;
    use tonic::{Response, Status, transport::Server};

    use crate::lock::{
        health_registry::HealthStatus,
        proto::{
            self, AuditResponse,
            audit_service_server::{AuditService, AuditServiceServer},
        },
    };

    // Mock gRPC server for testing
    #[derive(Debug)]
    struct MockAuditService {
        log_sender: mpsc::UnboundedSender<Vec<LogEntry>>,
        telemetry_sender: mpsc::UnboundedSender<Vec<TelemetryEntry>>,
        health_sender: mpsc::UnboundedSender<Vec<HealthEntry>>,
        should_fail: bool,
    }

    #[tonic::async_trait]
    impl AuditService for MockAuditService {
        async fn process_health(
            &self,
            _: Request<proto::HealthRequest>,
        ) -> Result<Response<AuditResponse>, Status> {
            unimplemented!()
        }

        async fn process_bulk_health(
            &self,
            request: Request<proto::BulkHealthRequest>,
        ) -> Result<Response<AuditResponse>, Status> {
            if self.should_fail {
                return Ok(Response::new(AuditResponse {
                    success: false,
                    message: "Server error".to_string(),
                }));
            }

            let auth = request
                .metadata()
                .get("authorization")
                .and_then(|v| v.to_str().ok());

            assert!(
                matches!(auth, Some(token) if token.starts_with("Bearer ")),
                "expected Bearer token in authorization header, got {auth:?}"
            );

            let entries = request.into_inner().entries;
            self.health_sender.send(entries).unwrap();

            Ok(Response::new(AuditResponse {
                success: true,
                message: "OK".to_string(),
            }))
        }

        async fn process_log(
            &self,
            _: Request<proto::LogRequest>,
        ) -> Result<Response<AuditResponse>, Status> {
            unimplemented!()
        }

        async fn process_bulk_log(
            &self,
            request: Request<BulkLogRequest>,
        ) -> Result<Response<AuditResponse>, Status> {
            if self.should_fail {
                return Ok(Response::new(AuditResponse {
                    success: false,
                    message: "Server error".to_string(),
                }));
            }

            let auth = request
                .metadata()
                .get("authorization")
                .and_then(|v| v.to_str().ok());

            assert!(
                matches!(auth, Some(token) if token.starts_with("Bearer ")),
                "expected Bearer token in authorization header, got {auth:?}"
            );

            let entries = request.into_inner().entries;
            self.log_sender.send(entries).unwrap();

            Ok(Response::new(AuditResponse {
                success: true,
                message: "OK".to_string(),
            }))
        }

        async fn process_telemetry(
            &self,
            _: Request<proto::TelemetryRequest>,
        ) -> Result<Response<AuditResponse>, Status> {
            unimplemented!()
        }

        async fn process_bulk_telemetry(
            &self,
            request: Request<proto::BulkTelemetryRequest>,
        ) -> Result<Response<AuditResponse>, Status> {
            if self.should_fail {
                return Ok(Response::new(AuditResponse {
                    success: false,
                    message: "Server error".to_string(),
                }));
            }

            let auth = request
                .metadata()
                .get("authorization")
                .and_then(|v| v.to_str().ok());

            assert!(
                matches!(auth, Some(token) if token.starts_with("Bearer ")),
                "expected Bearer token in authorization header, got {auth:?}"
            );

            let entries = request.into_inner().entries;
            self.telemetry_sender.send(entries).unwrap();

            Ok(Response::new(AuditResponse {
                success: true,
                message: "OK".to_string(),
            }))
        }
    }

    async fn start_mock_server(
        should_fail: bool,
    ) -> (
        SocketAddr,
        mpsc::UnboundedReceiver<Vec<LogEntry>>,
        mpsc::UnboundedReceiver<Vec<TelemetryEntry>>,
        mpsc::UnboundedReceiver<Vec<HealthEntry>>,
    ) {
        let (tx, rx) = mpsc::unbounded_channel();
        let (telemetry_tx, telemetry_rx) = mpsc::unbounded_channel();
        let (health_tx, health_rx) = mpsc::unbounded_channel();
        let service = MockAuditService {
            log_sender: tx,
            telemetry_sender: telemetry_tx,
            health_sender: health_tx,
            should_fail,
        };

        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        tokio::spawn(async move {
            Server::builder()
                .add_service(AuditServiceServer::new(service))
                .serve_with_incoming(TcpListenerStream::new(listener))
                .await
                .unwrap();
        });

        (addr, rx, telemetry_rx, health_rx)
    }

    #[test]
    fn test_json_to_proto_conversion() {
        use test_utils::assert_eq;

        let json_str = r#"{
            "timestamp": "2026-03-23T11:50:37.504Z",
            "log_kind": "Decision",
            "level": "INFO",
            "action": "Jans::Action::\"Read\"",
            "decision": "ALLOW",
            "principal": ["Jans::User::\"some_user\""],
            "resource": "Jans::Issue::\"random_id\"",
            "application_id": "test_app",
            "pdp_id": "12a8a3be-6593-4215-a42b-bbf5c4f5defa",
            "lock_client_id": "client-456"
        }"#;

        let json_entry: CedarlingLogEntry = serde_json::from_str(json_str).unwrap();
        let lock_entry = LockServerLogEntry::try_from(json_entry).unwrap();
        let proto_entry = log_json_to_proto(lock_entry);

        assert_eq!(
            proto_entry.creation_date,
            Some(Timestamp {
                seconds: 1_774_266_637,
                nanos: 504_000_000
            })
        );
        assert_eq!(
            proto_entry.event_time,
            Some(Timestamp {
                seconds: 1_774_266_637,
                nanos: 504_000_000,
            })
        );
        assert_eq!(proto_entry.service, "test_app");
        assert_eq!(
            proto_entry.node_name,
            "12a8a3be-6593-4215-a42b-bbf5c4f5defa"
        );
        assert_eq!(proto_entry.event_type, "Decision");
        assert_eq!(proto_entry.decision_result, "ALLOW");
        assert_eq!(proto_entry.severity_level, "INFO");
        assert_eq!(proto_entry.action, "Jans::Action::\"Read\"");
        assert_eq!(proto_entry.requested_resource, "Jans::Issue::\"random_id\"");
        assert_eq!(proto_entry.principal_id, "Jans::User::\"some_user\"");
        assert_eq!(proto_entry.client_id, "client-456");
        assert_eq!(proto_entry.jti, "");
    }

    #[test]
    fn test_partial_json_entry() {
        let json_str = r#"{
            "timestamp": "2026-03-23T11:50:37.504Z",
            "application_id": "minimal-service",
            "pdp_id": "node-1",
            "log_kind": "Decision",
            "decision": "ALLOW",
            "action": "Test::Action",
            "resource": "Test::Resource"
        }"#;

        let json_entry: CedarlingLogEntry = serde_json::from_str(json_str).unwrap();
        let lock_entry = LockServerLogEntry::try_from(json_entry).unwrap();
        let proto_entry = log_json_to_proto(lock_entry);

        assert_eq!(proto_entry.service, "minimal-service");
        assert_eq!(proto_entry.event_type, "Decision");
        assert_eq!(proto_entry.node_name, "node-1");
        assert!(proto_entry.creation_date.is_some());
    }

    #[tokio::test]
    async fn test_send_logs_success() {
        let (addr, mut rx, _, _) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries = vec![
            r#"{
                "timestamp": "2026-03-23T11:50:37.504Z",
                "log_kind": "Decision",
                "level": "INFO",
                "action": "Jans::Action::\"Read\"",
                "decision": "ALLOW",
                "principal": ["Jans::User::\"some_user\""],
                "resource": "Jans::Issue::\"random_id\"",
                "application_id": "test_app",
                "pdp_id": "12a8a3be-6593-4215-a42b-bbf5c4f5defa"
            }"#
            .to_string()
            .into_boxed_str(),
        ];

        transport
            .send(
                &entries,
                &AuditKind::Log(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect("logs should be sent successfully");

        // Verify the server received the logs
        let received = rx.try_recv().unwrap();
        assert_eq!(received.len(), 1);
        assert_eq!(received[0].service, "test_app");
        assert_eq!(
            received[0].node_name,
            "12a8a3be-6593-4215-a42b-bbf5c4f5defa"
        );
    }

    #[tokio::test]
    async fn test_send_logs_empty() {
        let (addr, mut rx, _, _) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        transport
            .send(
                &[],
                &AuditKind::Log(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect("logs should be sent successfully");

        // Should not have sent anything
        assert!(rx.try_recv().is_err());
    }

    #[tokio::test]
    async fn test_send_logs_malformed_json() {
        let (addr, _, _, _) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries = vec!["not valid json".to_string().into_boxed_str()];

        let error = transport
            .send(
                &entries,
                &AuditKind::Log(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect_err("this should cause a serialization error");
        assert!(
            matches!(error, TransportError::Serialization(_)),
            "expected serialization error, got {error:?}"
        );
    }

    #[tokio::test]
    async fn test_send_logs_server_error() {
        let (addr, _, _, _) = start_mock_server(true).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries = vec![
            r#"{
                "timestamp": "2026-03-23T11:50:37.504Z",
                "application_id": "test",
                "pdp_id": "node",
                "log_kind": "Decision",
                "decision": "ALLOW",
                "action": "Test::Action",
                "resource": "Test::Resource"
            }"#
            .to_string()
            .into_boxed_str(),
        ];

        let error = transport
            .send(
                &entries,
                &AuditKind::Log(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect_err("this should cause a server error");

        assert!(
            matches!(error, TransportError::GrpcServer(_)),
            "expected server error, got {error:?}",
        );
    }

    #[tokio::test]
    async fn test_send_logs_partial_malformed() {
        let (addr, mut rx, _, _) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries = vec![
            r#"{
                "timestamp": "2026-03-23T11:50:37.504Z",
                "application_id": "valid-service-1",
                "pdp_id": "node-1",
                "log_kind": "Decision",
                "decision": "ALLOW",
                "action": "Test::Action",
                "resource": "Test::Resource"
            }"#
            .to_string()
            .into_boxed_str(),
            "not valid json".to_string().into_boxed_str(),
            r#"{
                "timestamp": "2026-03-23T11:50:37.506Z",
                "application_id": "valid-service-2",
                "pdp_id": "node-2",
                "log_kind": "Decision",
                "decision": "ALLOW",
                "action": "Test::Action",
                "resource": "Test::Resource"
            }"#
            .to_string()
            .into_boxed_str(),
        ];

        transport
            .send(
                &entries,
                &AuditKind::Log(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect("logs should be sent successfully despite malformed entries");

        let received = rx.try_recv().unwrap();
        assert_eq!(received.len(), 2, "only valid entries should be forwarded");
        assert_eq!(received[0].service, "valid-service-1");
        assert_eq!(received[1].service, "valid-service-2");
    }

    #[tokio::test]
    async fn test_send_logs_multiple_entries() {
        let (addr, mut rx, _, _) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries = vec![
            r#"{
                "timestamp": "2026-03-23T11:50:37.504Z",
                "application_id": "service-1",
                "pdp_id": "node-1",
                "log_kind": "Decision",
                "decision": "ALLOW",
                "action": "Test::Action",
                "resource": "Test::Resource"
            }"#
            .to_string()
            .into_boxed_str(),
            r#"{
                "timestamp": "2026-03-23T11:50:37.505Z",
                "application_id": "service-2",
                "pdp_id": "node-2",
                "log_kind": "Decision",
                "decision": "ALLOW",
                "action": "Test::Action",
                "resource": "Test::Resource"
            }"#
            .to_string()
            .into_boxed_str(),
            r#"{
                "timestamp": "2026-03-23T11:50:37.506Z",
                "application_id": "service-3",
                "pdp_id": "node-3",
                "log_kind": "Decision",
                "decision": "ALLOW",
                "action": "Test::Action",
                "resource": "Test::Resource"
            }"#
            .to_string()
            .into_boxed_str(),
        ];

        transport
            .send(
                &entries,
                &AuditKind::Log(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect("logs should be sent successfully");

        let received = rx.try_recv().unwrap();
        assert_eq!(received.len(), 3);
        assert_eq!(received[0].service, "service-1");
        assert_eq!(received[1].service, "service-2");
        assert_eq!(received[2].service, "service-3");
    }

    #[tokio::test]
    async fn test_send_logs_with_all_fields() {
        let (addr, mut rx, _, _) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries = vec![
            r#"{
                "timestamp": "2026-03-23T11:50:37.504Z",
                "log_kind": "Decision",
                "level": "ERROR",
                "action": "Jans::Action::\"Deny\"",
                "decision": "DENY",
                "principal": ["Jans::User::\"admin\""],
                "resource": "Jans::Issue::\"secret\"",
                "application_id": "full-service",
                "pdp_id": "full-node",
                "lock_client_id": "admin-client",
                "extra_field": "extra_value"
            }"#
            .to_string()
            .into_boxed_str(),
        ];

        transport
            .send(
                &entries,
                &AuditKind::Log(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect("logs should be sent successfully");

        let received = rx.try_recv().unwrap();
        assert_eq!(received.len(), 1);
        let entry = &received[0];

        assert_eq!(entry.service, "full-service");
        assert_eq!(entry.node_name, "full-node");
        assert_eq!(entry.event_type, "Decision");
        assert_eq!(entry.severity_level, "ERROR");
        assert_eq!(entry.action, "Jans::Action::\"Deny\"");
        assert_eq!(entry.decision_result, "DENY");
        assert_eq!(entry.requested_resource, "Jans::Issue::\"secret\"");
        assert_eq!(entry.principal_id, "Jans::User::\"admin\"");
        assert_eq!(entry.client_id, "admin-client");
        assert_eq!(entry.jti, "");
    }

    #[tokio::test]
    async fn test_send_logs_with_missing_optional_fields() {
        let (addr, mut rx, _, _) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries = vec![
            r#"{
            "timestamp": "2026-03-23T11:50:37.504Z",
            "application_id": "minimal",
            "pdp_id": "test-pdp",
            "log_kind": "Decision",
            "decision": "ALLOW",
            "action": "Test::Action",
            "resource": "Test::Resource"
        }"#
            .to_string()
            .into_boxed_str(),
        ];

        transport
            .send(
                &entries,
                &AuditKind::Log(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect("logs should be sent successfully");

        let received = rx.try_recv().unwrap();
        assert_eq!(received.len(), 1);
        assert_eq!(received[0].service, "minimal");
        assert_eq!(received[0].node_name, "test-pdp");
    }

    #[tokio::test]
    async fn test_send_logs_large_batch() {
        let (addr, mut rx, _, _) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries: Vec<_> = (0..100)
            .map(|i| {
                json!({
                    "timestamp": "2026-03-23T11:50:37.504Z",
                    "log_kind": "System",
                    "level": "INFO",
                    "action": "Test",
                    "decision": "ALLOW",
                    "principal": ["Jans::User"],
                    "resource": "Jans::Issue",
                    "application_id": format!("service-{i}"),
                    "pdp_id": "node"
                })
                .to_string()
                .into_boxed_str()
            })
            .collect();

        transport
            .send(
                &entries,
                &AuditKind::Log(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect("logs should be sent successfully");

        let received = rx.try_recv().unwrap();
        assert_eq!(received.len(), 100);
    }

    #[tokio::test]
    async fn test_send_telemetry_success() {
        let (addr, _, mut telemetry_rx, _) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries = vec![
            json!({
                "id": "f3c80a24-4608-45b8-adc3-a74f2841e156",
                "request_id": "019d6842-7577-7e43-adfd-46e2bb275405",
                "timestamp": "2026-04-07T17:04:39.162Z",
                "log_kind": "Metric",
                "policy_stats": {
                    "555da5d85403f35ea76519ed1a18a33989f855bf1cf8_allow": 6,
                    "555da5d85403f35ea76519ed1a18a33989f855bf1cf8": 7,
                    "555da5d85403f35ea76519ed1a18a33989f855bf1cf8_deny": 1
                },
                "error_counters": {
                    "parse_error": 0,
                    "validation_error": 0
                },
                "operational_stats": {
                    "evaluation_requests": 100,
                    "memory_usage": 240
                },
                "interval_secs": 60,
                "application_id": "test_app",
                "pdp_id": "node-1"
            })
            .to_string()
            .into_boxed_str(),
        ];

        transport
            .send(
                &entries,
                &AuditKind::Telemetry(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect("telemetry should be sent successfully");

        let received = telemetry_rx.try_recv().unwrap();
        assert_eq!(received.len(), 1);
        assert_eq!(received[0].service, "test_app");
        assert_eq!(received[0].node_name, "node-1");
        assert_eq!(
            received[0]
                .policy_stats
                .get("555da5d85403f35ea76519ed1a18a33989f855bf1cf8_allow"),
            Some(&6)
        );
        assert_eq!(
            received[0]
                .policy_stats
                .get("555da5d85403f35ea76519ed1a18a33989f855bf1cf8_deny"),
            Some(&1)
        );
    }

    #[test]
    fn test_negative_timestamp_handling() {
        let json_str = r#"{
            "timestamp": "1969-12-31T00:00:00Z",
            "log_kind": "System",
            "level": "INFO",
            "action": "Test",
            "decision": "ALLOW",
            "principal": [],
            "resource": "Jans::Issue",
            "application_id": "test-service",
            "pdp_id": "node-1"
        }"#;

        let json_entry: CedarlingLogEntry = serde_json::from_str(json_str).unwrap();
        let lock_entry = LockServerLogEntry::try_from(json_entry).unwrap();
        let proto_entry = log_json_to_proto(lock_entry);

        assert_eq!(
            proto_entry.creation_date,
            Some(Timestamp {
                seconds: -86400,
                nanos: 0
            })
        );
        assert_eq!(
            proto_entry.event_time,
            Some(Timestamp {
                seconds: -86400,
                nanos: 0,
            })
        );
    }

    #[test]
    fn test_large_negative_timestamp() {
        let json_str = r#"{
            "timestamp": "1900-01-01T00:00:00Z",
            "log_kind": "Decision",
            "action": "Test",
            "decision": "ALLOW",
            "resource": "Test::Resource",
            "application_id": "historical-service",
            "pdp_id": "node-1"
        }"#;

        let json_entry: CedarlingLogEntry = serde_json::from_str(json_str).unwrap();
        let lock_entry = LockServerLogEntry::try_from(json_entry).unwrap();
        let proto_entry = log_json_to_proto(lock_entry);

        assert_eq!(
            proto_entry.creation_date.as_ref().unwrap().seconds,
            -2_208_988_800
        );
        assert_eq!(
            proto_entry.event_time.as_ref().unwrap().seconds,
            -2_208_988_800
        );
    }

    #[tokio::test]
    async fn test_send_health_success() {
        let (addr, _, _, mut health_rx) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries = vec![
            json!({
                "creation_date": "2026-03-23T11:50:37.504Z",
                "event_time": "2026-03-23T11:50:37.504Z",
                "service": "test_app",
                "node_name": "test-pdp",
                "status": "running",
                "engine_status": {
                    "core": "success"
                }
            })
            .to_string()
            .into_boxed_str(),
        ];

        transport
            .send(
                &entries,
                &AuditKind::Health(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect("health check should be sent successfully");

        let received = health_rx.try_recv().unwrap();
        assert_eq!(received.len(), 1);
        assert_eq!(received[0].service, "test_app");
        assert_eq!(received[0].node_name, "test-pdp");
        assert_eq!(received[0].status, "running");
        assert_eq!(received[0].engine_status.get("core").unwrap(), "success");
    }

    #[tokio::test]
    async fn test_send_health_empty() {
        let (addr, _, _, mut health_rx) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        transport
            .send(
                &[],
                &AuditKind::Health(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect("empty health check should succeed");

        assert!(health_rx.try_recv().is_err());
    }

    #[tokio::test]
    async fn test_send_health_malformed_json() {
        let (addr, _, _, _) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries = vec!["not valid json".to_string().into_boxed_str()];

        let error = transport
            .send(
                &entries,
                &AuditKind::Health(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect_err("this should cause a serialization error");
        assert!(
            matches!(error, TransportError::Serialization(_)),
            "expected serialization error, got {error:?}"
        );
    }

    #[tokio::test]
    async fn test_send_health_server_error() {
        let (addr, _, _, _) = start_mock_server(true).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries = vec![
            json!({
                "creation_date": "2026-03-23T11:50:37.504Z",
                "event_time": "2026-03-23T11:50:37.504Z",
                "service": "test_app",
                "node_name": "test-pdp",
                "status": "running",
                "engine_status": {
                    "core": "success"
                }
            })
            .to_string()
            .into_boxed_str(),
        ];

        let error = transport
            .send(
                &entries,
                &AuditKind::Health(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect_err("this should cause a server error");

        assert!(
            matches!(error, TransportError::GrpcServer(_)),
            "expected server error, got {error:?}",
        );
    }

    #[tokio::test]
    async fn test_send_health_large_batch() {
        let (addr, _, _, mut health_rx) = start_mock_server(false).await;
        let transport = GrpcTransport::new(format!("http://{addr}"), "test-token", None).unwrap();

        let entries: Vec<_> = (0..100)
            .map(|i| {
                json!({
                    "creation_date": "2026-03-23T11:50:37.504Z",
                    "event_time": "2026-03-23T11:50:37.504Z",
                    "service": format!("service-{i}"),
                    "node_name": "node-1",
                    "status": "running",
                    "engine_status": {
                        "core": "success"
                    }
                })
                .to_string()
                .into_boxed_str()
            })
            .collect();

        transport
            .send(
                &entries,
                &AuditKind::Health(format!("http://{addr}").parse().unwrap()),
            )
            .await
            .expect("health checks should be sent successfully");

        let received = health_rx.try_recv().unwrap();
        assert_eq!(received.len(), 100);
        assert_eq!(received[0].service, "service-0");
        assert_eq!(received[99].service, "service-99");
    }

    #[test]
    fn test_health_json_to_proto_conversion() {
        let entry = LockServerHealthEntry {
            creation_date: "2026-03-23T11:50:37.504Z".to_string(),
            event_time: "2026-03-23T11:50:37.504Z".to_string(),
            service: "test_app".to_string(),
            node_name: "test-pdp".to_string(),
            status: "running".to_string(),
            engine_status: [("core".to_string(), HealthStatus::Success)]
                .into_iter()
                .collect(),
        };

        let proto = health_json_to_proto(entry);

        assert_eq!(
            proto.creation_date,
            Some(Timestamp {
                seconds: 1_774_266_637,
                nanos: 504_000_000
            })
        );
        assert_eq!(
            proto.event_time,
            Some(Timestamp {
                seconds: 1_774_266_637,
                nanos: 504_000_000,
            })
        );
        assert_eq!(proto.service, "test_app");
        assert_eq!(proto.node_name, "test-pdp");
        assert_eq!(proto.status, "running");
        assert_eq!(proto.engine_status.get("core").unwrap(), "success");
    }
}