liminal-server 0.19.0

use serde::Serialize;

use super::cost::{RecordError, ServerMetadata, build_commit};

/// Publication cause carried on the internal call rather than mutable state.
#[derive(Clone, Copy, Debug)]
pub(in crate::server) enum PublishOrigin<'a> {
    /// A connection's publish, by its id (the connection process's pid).
    Connection { id: u64 },
    /// The server's own record, as `liminal.<device>`.
    Server { device: &'a str },
}

impl PublishOrigin<'_> {
    pub(in crate::server) const fn is_server(&self) -> bool {
        matches!(self, Self::Server { .. })
    }
}

/// A fact observed at a server-side operation site.
#[derive(Clone, Copy, Debug)]
pub(in crate::server) enum ServerEvent<'a> {
    Lifecycle {
        conversation: u64,
        generation: u64,
        transition: &'a str,
    },
    Pressure {
        connection: u64,
        response: &'a str,
    },
    Flush {
        store: &'a str,
        batch_rows: Option<u64>,
    },
    ConnectionOpened {
        connection: u64,
    },
    ConnectionClosed {
        connection: u64,
    },
}

impl ServerEvent<'_> {
    pub(in crate::server) fn encode(
        &self,
        metadata: &ServerMetadata,
        server_time_ms: u64,
    ) -> Result<Vec<u8>, RecordError> {
        let details = match *self {
            Self::Lifecycle {
                conversation,
                generation,
                transition,
            } => {
                if transition.trim().is_empty() {
                    return Err(RecordError::EmptyLifecycleTransition);
                }
                EventDetails {
                    conversation: Some(conversation),
                    generation: Some(generation),
                    transition: Some(transition),
                    ..EventDetails::new("lifecycle")
                }
            }
            Self::Pressure {
                connection,
                response,
            } => {
                if response.trim().is_empty() {
                    return Err(RecordError::EmptyPressureResponse);
                }
                EventDetails {
                    connection: Some(connection),
                    response: Some(response),
                    ..EventDetails::new("pressure")
                }
            }
            Self::Flush { store, batch_rows } => {
                if store.trim().is_empty() {
                    return Err(RecordError::EmptyFlushStore);
                }
                EventDetails {
                    store: Some(store),
                    batch_rows,
                    ..EventDetails::new("flush")
                }
            }
            Self::ConnectionOpened { connection } => EventDetails {
                connection: Some(connection),
                ..EventDetails::new("connection_open")
            },
            Self::ConnectionClosed { connection } => EventDetails {
                connection: Some(connection),
                ..EventDetails::new("connection_close")
            },
        };
        let record = EventRecord {
            v: 1,
            build_commit: build_commit(),
            crate_version: env!("CARGO_PKG_VERSION"),
            details,
            device: metadata.device(),
            publisher: metadata.publisher(),
            server_incarnation: metadata.server_incarnation(),
            server_time_ms,
        };
        Ok(serde_json::to_vec(&record)?)
    }
}

#[derive(Serialize)]
struct EventRecord<'a> {
    v: u8,
    build_commit: &'static str,
    crate_version: &'static str,
    details: EventDetails<'a>,
    device: &'a str,
    publisher: &'a str,
    server_incarnation: u64,
    server_time_ms: u64,
}

#[derive(Serialize)]
struct EventDetails<'a> {
    #[serde(skip_serializing_if = "Option::is_none")]
    batch_rows: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    connection: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    conversation: Option<u64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    generation: Option<u64>,
    kind: &'static str,
    #[serde(skip_serializing_if = "Option::is_none")]
    store: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    response: Option<&'a str>,
    #[serde(skip_serializing_if = "Option::is_none")]
    transition: Option<&'a str>,
}

impl EventDetails<'_> {
    const fn new(kind: &'static str) -> Self {
        Self {
            batch_rows: None,
            store: None,
            connection: None,
            conversation: None,
            generation: None,
            kind,
            response: None,
            transition: None,
        }
    }
}