saddle-observability 0.2.0

use std::time::{Instant, SystemTime, UNIX_EPOCH};

use saddle_admission::RequestMemory;

use super::{
    FileStream,
    codec::{Field, JsonlRecord, Scalar},
    fixed_core::{FixedEncodingLease, FixedFileSink},
};
use crate::file::{CompletionResult, FixedCompletion, FixedFailure, SubmitError};

pub const CORRELATION_SCHEMA_VERSION: u16 = 1;
const CORRELATION_SINK_IDENTITY: [u8; 32] = [0xc8; 32];

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CorrelationCallKind {
    ExternalRoute,
    Service,
    Database,
    Transaction,
}

impl CorrelationCallKind {
    const fn value(self) -> &'static str {
        match self {
            Self::ExternalRoute => "external_route",
            Self::Service => "service",
            Self::Database => "database",
            Self::Transaction => "transaction",
        }
    }

    const fn stream(self) -> FileStream {
        match self {
            Self::ExternalRoute => FileStream::Access,
            Self::Service | Self::Database | Self::Transaction => FileStream::Trace,
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CorrelationRecordPhase {
    Started,
    Finished,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CorrelationCallOutcome {
    Success,
    Failure,
    Cancelled,
    Abandoned,
}

impl CorrelationCallOutcome {
    const fn value(self) -> &'static str {
        match self {
            Self::Success => "success",
            Self::Failure => "failure",
            Self::Cancelled => "cancelled",
            Self::Abandoned => "abandoned",
        }
    }
}

/// Fixed-shape correlation data. Every textual identity is borrowed from a
/// verified component proof; this type owns no growable string or field map.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct CorrelationRecord<'a> {
    kind: CorrelationCallKind,
    phase: CorrelationRecordPhase,
    trace_id: &'a str,
    span_id: &'a str,
    parent_span_id: Option<&'a str>,
    route: &'a str,
    service: &'a str,
    operation: &'a str,
    outcome: Option<CorrelationCallOutcome>,
    elapsed_us: Option<u64>,
}

impl<'a> CorrelationRecord<'a> {
    #[doc(hidden)]
    pub const fn started(
        kind: CorrelationCallKind,
        trace_id: &'a str,
        span_id: &'a str,
        parent_span_id: Option<&'a str>,
        route: &'a str,
        service: &'a str,
        operation: &'a str,
    ) -> Self {
        Self {
            kind,
            phase: CorrelationRecordPhase::Started,
            trace_id,
            span_id,
            parent_span_id,
            route,
            service,
            operation,
            outcome: None,
            elapsed_us: None,
        }
    }

    pub const fn phase(&self) -> CorrelationRecordPhase {
        self.phase
    }

    pub const fn kind(&self) -> CorrelationCallKind {
        self.kind
    }

    fn finished(self, outcome: CorrelationCallOutcome, elapsed_us: u64) -> Self {
        Self {
            phase: CorrelationRecordPhase::Finished,
            outcome: Some(outcome),
            elapsed_us: Some(elapsed_us),
            ..self
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct CorrelationSinkIdentity {
    schema_version: u16,
    sink_identity: [u8; 32],
}

impl CorrelationSinkIdentity {
    pub const fn schema_version(self) -> u16 {
        self.schema_version
    }

    pub const fn sink_identity(self) -> [u8; 32] {
        self.sink_identity
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CorrelationHealth {
    Accepting,
    ShuttingDown,
    Failed(FixedFailure),
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum CorrelationSubmitError {
    Backpressured,
    Encoding,
    ShuttingDown,
    Unhealthy(FixedFailure),
    Admission,
}

impl From<SubmitError> for CorrelationSubmitError {
    fn from(error: SubmitError) -> Self {
        match error {
            SubmitError::Backpressured | SubmitError::CompletionBusy => Self::Backpressured,
            SubmitError::Encoding => Self::Encoding,
            SubmitError::ShuttingDown | SubmitError::OutstandingEncoding => Self::ShuttingDown,
            SubmitError::Unhealthy(failure) => Self::Unhealthy(failure),
            SubmitError::Admission(_) => Self::Admission,
        }
    }
}

/// Private-field, non-Clone owner of the only fixed correlation sink.
///
/// ```compile_fail
/// use saddle_observability::file::CorrelationSinkOwner;
/// fn duplicate<const B: usize, const Y: usize, const C: usize>(
///     owner: &CorrelationSinkOwner<B, Y, C>,
/// ) -> CorrelationSinkOwner<B, Y, C> {
///     owner.clone()
/// }
/// ```
///
/// ```compile_fail
/// use saddle_observability::file::CorrelationSinkOwner;
/// let _forged: CorrelationSinkOwner<1, 1, 1> = CorrelationSinkOwner { sink: todo!() };
/// ```
#[doc(hidden)]
pub struct CorrelationSinkOwner<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> {
    sink: FixedFileSink<BLOCKS, BYTES, COMMANDS>,
}

/// Public read-only verification surface. It contains no sink or submit path.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct VerifiedCorrelationSink {
    identity: CorrelationSinkIdentity,
    health: CorrelationHealth,
}

/// Linear flush/shutdown barrier bound to the same versioned sink identity.
#[doc(hidden)]
pub struct CorrelationBarrier<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> {
    identity: CorrelationSinkIdentity,
    pub(super) completion: FixedCompletion<BLOCKS, BYTES, COMMANDS>,
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>
    CorrelationBarrier<BLOCKS, BYTES, COMMANDS>
{
    pub const fn identity(&self) -> CorrelationSinkIdentity {
        self.identity
    }

    pub fn result(&self) -> CompletionResult<FixedFailure> {
        self.completion.result()
    }

    pub fn recycle(self) -> Result<(), CorrelationSubmitError> {
        self.completion.recycle().map_err(Into::into)
    }

    pub fn cancel(self) -> Result<CompletionResult<FixedFailure>, CorrelationSubmitError> {
        self.completion.cancel().map_err(Into::into)
    }

    #[doc(hidden)]
    pub fn into_completion(self) -> FixedCompletion<BLOCKS, BYTES, COMMANDS> {
        self.completion
    }
}

impl VerifiedCorrelationSink {
    pub const fn identity(self) -> CorrelationSinkIdentity {
        self.identity
    }

    pub const fn health(self) -> CorrelationHealth {
        self.health
    }
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>
    CorrelationSinkOwner<BLOCKS, BYTES, COMMANDS>
{
    #[doc(hidden)]
    pub fn from_fixed_sink(sink: FixedFileSink<BLOCKS, BYTES, COMMANDS>) -> Self {
        Self { sink }
    }

    pub const fn identity(&self) -> CorrelationSinkIdentity {
        CorrelationSinkIdentity {
            schema_version: CORRELATION_SCHEMA_VERSION,
            sink_identity: CORRELATION_SINK_IDENTITY,
        }
    }

    pub fn verify(&self) -> VerifiedCorrelationSink {
        let health = match self.sink.health() {
            Ok(true) => CorrelationHealth::Accepting,
            Ok(false) => CorrelationHealth::ShuttingDown,
            Err(failure) => CorrelationHealth::Failed(failure),
        };
        VerifiedCorrelationSink {
            identity: self.identity(),
            health,
        }
    }

    #[doc(hidden)]
    pub fn begin_call<'a>(
        &self,
        memory: &RequestMemory,
        started: CorrelationRecord<'a>,
    ) -> Result<ActiveCorrelationCall<'a, BLOCKS, BYTES, COMMANDS>, CorrelationSubmitError> {
        if started.phase != CorrelationRecordPhase::Started {
            return Err(CorrelationSubmitError::Encoding);
        }
        let start = self.sink.begin_record(memory)?;
        let finish = self.sink.begin_record(memory)?;
        encode(start, started)?;
        Ok(ActiveCorrelationCall {
            finish: Some(finish),
            started,
            clock: Instant::now(),
        })
    }

    #[doc(hidden)]
    pub fn try_flush(
        &self,
    ) -> Result<CorrelationBarrier<BLOCKS, BYTES, COMMANDS>, CorrelationSubmitError> {
        Ok(CorrelationBarrier {
            identity: self.identity(),
            completion: self.sink.try_flush()?,
        })
    }

    #[doc(hidden)]
    pub fn try_shutdown(
        &self,
    ) -> Result<CorrelationBarrier<BLOCKS, BYTES, COMMANDS>, CorrelationSubmitError> {
        Ok(CorrelationBarrier {
            identity: self.identity(),
            completion: self.sink.try_shutdown()?,
        })
    }
}

#[doc(hidden)]
pub struct ActiveCorrelationCall<'a, const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>
{
    finish: Option<FixedEncodingLease<BLOCKS, BYTES, COMMANDS>>,
    started: CorrelationRecord<'a>,
    clock: Instant,
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>
    ActiveCorrelationCall<'_, BLOCKS, BYTES, COMMANDS>
{
    pub fn finish(mut self, outcome: CorrelationCallOutcome) -> Result<(), CorrelationSubmitError> {
        self.commit_finish(outcome)
    }

    fn commit_finish(
        &mut self,
        outcome: CorrelationCallOutcome,
    ) -> Result<(), CorrelationSubmitError> {
        let elapsed = u64::try_from(self.clock.elapsed().as_micros()).unwrap_or(u64::MAX);
        let lease = self.finish.take().ok_or(CorrelationSubmitError::Encoding)?;
        encode(lease, self.started.finished(outcome, elapsed))
    }
}

impl<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize> Drop
    for ActiveCorrelationCall<'_, BLOCKS, BYTES, COMMANDS>
{
    fn drop(&mut self) {
        if self.finish.is_some() {
            // The finish capacity was reserved before the call began. Failure
            // here is sticky in the underlying sink and is never retried or
            // redirected to stdout/stderr.
            let _ = self.commit_finish(CorrelationCallOutcome::Abandoned);
        }
    }
}

fn encode<const BLOCKS: usize, const BYTES: usize, const COMMANDS: usize>(
    lease: FixedEncodingLease<BLOCKS, BYTES, COMMANDS>,
    record: CorrelationRecord<'_>,
) -> Result<(), CorrelationSubmitError> {
    let schema = u64::from(CORRELATION_SCHEMA_VERSION);
    let fields = [
        Field {
            name: "schema_version",
            value: Scalar::Unsigned(schema),
        },
        Field {
            name: "call_kind",
            value: Scalar::String(record.kind.value()),
        },
        Field {
            name: "phase",
            value: Scalar::String(match record.phase {
                CorrelationRecordPhase::Started => "started",
                CorrelationRecordPhase::Finished => "finished",
            }),
        },
        Field {
            name: "route",
            value: Scalar::String(record.route),
        },
        Field {
            name: "service",
            value: Scalar::String(record.service),
        },
        Field {
            name: "operation",
            value: Scalar::String(record.operation),
        },
        Field {
            name: "outcome",
            value: record
                .outcome
                .map_or(Scalar::Null, |value| Scalar::String(value.value())),
        },
        Field {
            name: "elapsed_us",
            value: record.elapsed_us.map_or(Scalar::Null, Scalar::Unsigned),
        },
    ];
    lease
        .encode_and_commit(
            record.kind.stream(),
            JsonlRecord {
                timestamp_unix_ms: SystemTime::now()
                    .duration_since(UNIX_EPOCH)
                    .unwrap_or_default()
                    .as_millis(),
                level: "info",
                event: match record.phase {
                    CorrelationRecordPhase::Started => "framework.call.started",
                    CorrelationRecordPhase::Finished => "framework.call.finished",
                },
                trace_id: Some(record.trace_id),
                span_id: Some(record.span_id),
                parent_span_id: record.parent_span_id,
                fields: &fields,
                dropped_events: 0,
            },
        )
        .map_err(Into::into)
}