saddle-db 0.3.8

use saddle_core::{CallContext, OperationId};
use saddle_observability::{ActiveStage, DatabaseDisposition, EventContext, Observer, Stage};

/// One Database operation projected into the framework's typed event chain.
///
/// The operation name comes exclusively from the sealed generated operation
/// declaration. It never contains mapped physical names, SQL, parameters,
/// results, connection details, or deployment secrets.
pub(crate) struct DatabaseOperationObservation {
    observer: Observer,
    event_context: EventContext,
    stage: ActiveStage,
}

impl DatabaseOperationObservation {
    pub(crate) fn start(
        observer: &Observer,
        parent: &CallContext,
        event_context: EventContext,
        logical_operation: &'static str,
    ) -> Self {
        let logical_parent = CallContext::new(
            parent.application().clone(),
            parent.module().clone(),
            parent.service().clone(),
            OperationId::from(logical_operation),
            parent.trace_id(),
            parent.span_id(),
        )
        .with_trace_correlation_id(parent.trace_correlation_id().clone());
        let stage = observer.start_stage(&logical_parent, Stage::Database, event_context.clone());
        Self {
            observer: observer.clone(),
            event_context,
            stage,
        }
    }

    pub(crate) fn finish(self, disposition: saddle_core::DbPhysicalDisposition) {
        let disposition = match disposition {
            saddle_core::DbPhysicalDisposition::Returned => DatabaseDisposition::Returned,
            saddle_core::DbPhysicalDisposition::Discarded => DatabaseDisposition::Discarded,
        };
        self.observer.record_database_disposition(
            self.stage.context(),
            &self.event_context,
            disposition,
        );
        self.stage.succeed();
    }
}

#[cfg(test)]
mod tests {
    use std::{
        io,
        sync::{Arc, Mutex},
    };

    use saddle_core::{ApplicationId, ModuleId, ServiceId, SpanId, TraceId};
    use saddle_observability::{ObserverConfig, RequestIdentity, RouteIdentity};
    use serde_json::Value;

    use super::*;

    #[derive(Clone, Default)]
    struct Capture(Arc<Mutex<Vec<u8>>>);

    impl io::Write for Capture {
        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
            self.0.lock().unwrap().extend_from_slice(bytes);
            Ok(bytes.len())
        }

        fn flush(&mut self) -> io::Result<()> {
            Ok(())
        }
    }

    fn parent() -> CallContext {
        CallContext::new(
            ApplicationId::from("app"),
            ModuleId::from("module"),
            ServiceId::from("service"),
            OperationId::from("request"),
            TraceId::from_u128(7),
            SpanId::from_u64(9),
        )
    }

    fn event() -> EventContext {
        EventContext::new(
            RequestIdentity::new("request-7").unwrap(),
            RouteIdentity::new("orders.create").unwrap(),
            1,
        )
        .unwrap()
    }

    #[tokio::test]
    async fn returned_and_deadline_discard_are_correlated_and_payload_free() {
        let capture = Capture::default();
        let observer = Observer::with_writer(ObserverConfig::default(), capture.clone()).unwrap();
        DatabaseOperationObservation::start(&observer, &parent(), event(), "orders.logical_lookup")
            .finish(saddle_core::DbPhysicalDisposition::Returned);
        DatabaseOperationObservation::start(&observer, &parent(), event(), "orders.logical_lookup")
            .finish(saddle_core::DbPhysicalDisposition::Discarded);
        observer.flush().await.unwrap();

        let records: Vec<Value> = String::from_utf8(capture.0.lock().unwrap().clone())
            .unwrap()
            .lines()
            .map(|line| serde_json::from_str(line).unwrap())
            .collect();
        let dispositions: Vec<_> = records
            .iter()
            .filter(|record| record["event"] == "framework.database.disposition")
            .collect();
        assert_eq!(dispositions.len(), 2);
        assert_eq!(dispositions[0]["db_disposition"], "returned");
        assert_eq!(dispositions[1]["db_disposition"], "discarded");
        for record in records {
            assert_eq!(record["trace_id"], "00000000000000000000000000000007");
            assert_eq!(record["request_identity"], "request-7");
            assert_eq!(record["request"], "orders.logical_lookup");
            assert!(record.get("elapsed_ms").is_some());
            let encoded = serde_json::to_string(&record).unwrap();
            for forbidden in [
                "SELECT ",
                "INSERT ",
                "mysql://",
                "physical_",
                "password",
                "secret",
            ] {
                assert!(!encoded.contains(forbidden));
            }
        }
    }
}