liminal-server 0.25.4

//! The production [`KeySender`]: a session's keys delivered as a publish on
//! `sessions.<id>.keys` through the server's own publish path (ACCESS-003
//! R1).
//!
//! The keys are published only after Lys answered the deliberate ask
//! outright; a held act never reaches this sender. The publish goes through
//! [`ServicePublisher`], whose real implementation is the connection
//! services' embedded-host publish (`publish_embedded`): the same admission,
//! namespace refusal, durability and dedup a connection publish gets, with
//! publisher `liminal.<host>`. The request's operation id is the
//! idempotency key, so a request sent again is delivered at most once. The
//! session's runner subscribes to its own `sessions.<id>.keys` subject; a
//! subject the roster does not carry is refused by the roster, by name.

use super::{KeySender, ServiceEffectError};
use crate::server::connection::LiminalConnectionServices;
use liminal::protocol::{CausalContext, MessageEnvelope, SchemaId};
use std::sync::Arc;

/// The schema bytes the services' messages are published under: the payload
/// is the request's bytes, unchanged, and the schema names which service
/// carried them.
pub const SESSION_KEYS_SCHEMA: &[u8] = b"liminal.services.session-keys/1";

/// Publishes one service message on a subject, once per operation.
pub trait ServicePublisher: Send + Sync + std::fmt::Debug {
    /// Publishes `envelope` on `subject` with `operation` as its idempotency
    /// key.
    ///
    /// # Errors
    /// The publish was refused or failed, by name.
    fn publish(
        &self,
        subject: &str,
        envelope: &MessageEnvelope,
        operation: &str,
    ) -> Result<(), ServiceEffectError>;
}

/// The connection services' embedded-host publish as a [`ServicePublisher`].
#[derive(Debug)]
pub struct ConnectionPublisher {
    services: Arc<LiminalConnectionServices>,
    host: String,
}

impl ConnectionPublisher {
    /// Publishes through `services` as `liminal.<host>`.
    #[must_use]
    pub const fn new(services: Arc<LiminalConnectionServices>, host: String) -> Self {
        Self { services, host }
    }
}

impl ServicePublisher for ConnectionPublisher {
    fn publish(
        &self,
        subject: &str,
        envelope: &MessageEnvelope,
        operation: &str,
    ) -> Result<(), ServiceEffectError> {
        self.services
            .publish_embedded(&self.host, subject, envelope, Some(operation))
            .map(|_outcome| ())
            .map_err(|error| ServiceEffectError(format!("publish on {subject} refused: {error}")))
    }
}

/// The envelope one service message travels in.
#[must_use]
pub fn service_envelope(schema: &[u8], payload: &[u8]) -> MessageEnvelope {
    MessageEnvelope::new(
        SchemaId::from_schema_bytes(schema),
        CausalContext::independent(),
        payload.to_vec(),
    )
}

/// Session keys published on `sessions.<id>.keys`.
#[derive(Debug)]
pub struct ChannelKeySender {
    publisher: Arc<dyn ServicePublisher>,
}

impl ChannelKeySender {
    /// Sends through `publisher`.
    #[must_use]
    pub const fn new(publisher: Arc<dyn ServicePublisher>) -> Self {
        Self { publisher }
    }
}

/// The subject a session's keys are delivered on.
#[must_use]
pub fn keys_subject(session: &str) -> String {
    format!("sessions.{session}.keys")
}

impl KeySender for ChannelKeySender {
    fn send(&self, session: &str, keys: &[u8], operation: &str) -> Result<(), ServiceEffectError> {
        if keys.is_empty() {
            return Err(ServiceEffectError(format!(
                "no keys to send to session {session}"
            )));
        }
        self.publisher.publish(
            &keys_subject(session),
            &service_envelope(SESSION_KEYS_SCHEMA, keys),
            operation,
        )
    }
}