ruststream-sea-file 0.7.1

//! [`SeaMessage`]: a delivered message, shared by the file and stdio transports.

use std::future::{Future, ready};

use bytes::Bytes;
use ruststream::{AckError, HeaderMap, IncomingMessage, Positioned, Str};
use sea_streamer_types::{Buffer as _, Message as _, SharedMessage};

#[cfg(feature = "testing")]
use crate::in_process::Release;
use crate::wire;

/// Header exposing the message's sequence number within its stream.
pub const SEQUENCE_HEADER: &str = "stream-sequence";

/// A position in a stream file's retained log: where `start_at(..)` opens a subscription, and
/// where [`Seeker::seek`](ruststream::Seeker::seek) moves a running one.
///
/// Captured positions ([`Positioned::position`]) carry the pinned semantics the framework
/// defines: seeking to one redelivers exactly that message (the transport's sequence rewind
/// is inclusive). The other forms keep the transport's own semantics: `Beginning` replays
/// everything retained, `End` skips to the tip, and `Timestamp` resumes at the earliest
/// message strictly later than the instant (milliseconds since the Unix epoch).
///
/// Seek to a position the stream has reached, or to the sequence it writes next, which is the
/// tip: a service that handled every message resumes there. A sequence further ahead, and an
/// instant later than every message it holds, are both refused - the transport reads forward
/// looking for the message and runs out of file - and the subscription ends with that refusal
/// rather than staying where it was. A captured position, the beginning and the tip are always
/// safe; a number a service worked out for itself is what needs the care.
///
/// The
/// [FilePosition](https://powersemmi.github.io/ruststream-sea-file/latest/components/positions/)
/// page shows it in a service.
///
/// # Examples
///
/// A handler reads its delivery's place in the file with `Ctx<Position>` and moves its own
/// subscription with `Ctx<SeekHandle>`:
///
/// ```
/// use ruststream_sea_file::file::prelude::*;
/// use serde::Deserialize;
///
/// #[derive(Debug, Deserialize)]
/// struct Job {
///     id: u64,
///     poisoned: bool,
/// }
///
/// #[subscriber(FileStream::new("jobs"), start_at(FilePosition::beginning()))]
/// async fn work(
///     job: &Job,
///     Ctx(at): Ctx<Position>,
///     Ctx(seeker): Ctx<SeekHandle>,
/// ) -> HandlerOutcome {
///     if job.poisoned && seeker.seek(FilePosition::end()).await.is_err() {
///         return HandlerOutcome::retry();
///     }
///     tracing::info!(job.id, position = ?at, "worked");
///     HandlerOutcome::ack()
/// }
///
/// #[ruststream::app]
/// fn app() -> impl App {
///     RustStream::new(AppInfo::new("jobs", "0.1.0"))
///         .with_broker(FileBroker::new("/tmp/jobs.ss"), |b| {
///             b.include(work);
///         })
/// }
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FilePosition {
    /// Everything retained.
    Beginning,
    /// The tip of the stream.
    End,
    /// A message sequence (inclusive). A stream key is numbered from one, and the numbering
    /// carries on when a later connection appends to the same file.
    Sequence(u64),
    /// Milliseconds since the Unix epoch (exclusive).
    Timestamp(u64),
}

// Constructor forms of the variants: the `start_at(..)` macro clause resolves the position's
// type from a constructor call, which a bare unit-variant path does not provide.
impl FilePosition {
    /// Everything retained: [`FilePosition::Beginning`].
    #[must_use]
    pub const fn beginning() -> Self {
        Self::Beginning
    }

    /// The tip of the stream: [`FilePosition::End`].
    #[must_use]
    pub const fn end() -> Self {
        Self::End
    }

    /// A message sequence, redelivered inclusively: [`FilePosition::Sequence`].
    #[must_use]
    pub const fn sequence(sequence: u64) -> Self {
        Self::Sequence(sequence)
    }

    /// Milliseconds since the Unix epoch, resuming strictly later:
    /// [`FilePosition::Timestamp`].
    #[must_use]
    pub const fn timestamp(millis: u64) -> Self {
        Self::Timestamp(millis)
    }
}

/// A message delivered by one of this crate's subscribers.
///
/// This is the whole delivery on standard input; a stream file wraps it in a
/// [`FileMessage`](crate::FileMessage), which adds the subscription's reposition handle.
///
/// The transport keeps no consumer positions (its resumable mode is unimplemented upstream),
/// so acknowledgement reports [`AckError::Unsupported`] rather than pretending; resume
/// explicitly via the descriptor's start position or a captured [`FilePosition`].
pub struct SeaMessage {
    payload: Bytes,
    headers: HeaderMap,
    stream: String,
    sequence: u64,
    /// Tells the test harness this delivery is done when it is dropped: the in-process transport
    /// counts its deliveries in flight. The field is there only with the `testing` feature.
    #[cfg(feature = "testing")]
    release: Option<Release>,
}

// The zero-cost promise of the in-process mode, held by the compiler: a build without it gives the
// delivery exactly the fields a delivery off a file or a pipe needs.
#[cfg(not(feature = "testing"))]
const _: () = assert!(size_of::<SeaMessage>() == size_of::<(Bytes, HeaderMap, String, u64)>());

impl std::fmt::Debug for SeaMessage {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SeaMessage")
            .field("stream", &self.stream)
            .field("sequence", &self.sequence)
            .field("payload_len", &self.payload.len())
            .finish_non_exhaustive()
    }
}

impl SeaMessage {
    pub(crate) fn new(message: &SharedMessage) -> Self {
        let (mut headers, payload) = wire::decode(message.message().as_bytes());
        let sequence = message.sequence();
        headers.insert(Str::from_static(SEQUENCE_HEADER), sequence.to_string());
        Self {
            payload,
            headers,
            stream: message.stream_key().name().to_owned(),
            sequence,
            #[cfg(feature = "testing")]
            release: None,
        }
    }

    /// The same delivery, counted in flight by the harness until it is dropped.
    #[cfg(feature = "testing")]
    pub(crate) fn released_by(mut self, release: Option<Release>) -> Self {
        self.release = release;
        self
    }

    /// The stream key this message was published to.
    #[must_use]
    pub fn stream(&self) -> &str {
        &self.stream
    }
}

impl Positioned for SeaMessage {
    type Position = FilePosition;

    fn position(&self) -> FilePosition {
        FilePosition::Sequence(self.sequence)
    }
}

impl IncomingMessage for SeaMessage {
    fn payload(&self) -> &[u8] {
        &self.payload
    }

    fn headers(&self) -> &HeaderMap {
        &self.headers
    }

    fn ack(self) -> impl Future<Output = Result<(), AckError>> {
        ready(Err(AckError::Unsupported))
    }

    fn nack(self, _requeue: bool) -> impl Future<Output = Result<(), AckError>> {
        ready(Err(AckError::Unsupported))
    }
}