danube-core 0.15.0

use crate::message::StreamMessage;
use async_trait::async_trait;
use futures_core::Stream;
use serde::{Deserialize, Serialize};
use std::pin::Pin;
use thiserror::Error;

/// Start position for a subscription/reader.
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
pub enum StartPosition {
    /// Start from the current tail (latest offset + 1)
    Latest,
    /// Start from a concrete offset (inclusive)
    Offset(u64),
}

#[derive(Debug, Error)]
pub enum PersistentStorageError {
    #[error("I/O error: {0}")]
    Io(String),

    #[error("WAL error: {0}")]
    Wal(String),

    #[error("Cloud storage error: {0}")]
    Cloud(String),

    #[error("Metadata store error: {0}")]
    Metadata(String),

    #[error("Unsupported: {0}")]
    Unsupported(String),

    #[error("Other: {0}")]
    Other(String),
}

/// A boxed async stream of messages for a topic.
pub type TopicStream =
    Pin<Box<dyn Stream<Item = Result<StreamMessage, PersistentStorageError>> + Send>>;

/// A persistent storage interface for a topic.
#[async_trait]
pub trait PersistentStorage: Send + Sync + std::fmt::Debug + 'static {
    /// Append a message to a topic and return the assigned offset.
    async fn append_message(
        &self,
        topic_name: &str,
        msg: StreamMessage,
    ) -> Result<u64, PersistentStorageError>;

    /// Append a batch of messages atomically and return `(first_offset, last_offset)`.
    ///
    /// The default implementation calls `append_message` in a loop. Concrete
    /// implementations (e.g. `WalStorage`) may override this with an optimized
    /// single-lock batch path.
    async fn append_batch(
        &self,
        topic_name: &str,
        messages: &[StreamMessage],
    ) -> Result<(u64, u64), PersistentStorageError> {
        if messages.is_empty() {
            return Err(PersistentStorageError::Other("empty batch".to_string()));
        }
        let mut first = 0u64;
        let mut last = 0u64;
        for (i, msg) in messages.iter().enumerate() {
            let offset = self.append_message(topic_name, msg.clone()).await?;
            if i == 0 {
                first = offset;
            }
            last = offset;
        }
        Ok((first, last))
    }

    /// Create a streaming reader starting from the provided position.
    async fn create_reader(
        &self,
        topic_name: &str,
        start: StartPosition,
    ) -> Result<TopicStream, PersistentStorageError>;

    /// Acknowledge internal checkpoints (used by background uploader/compactor).
    async fn ack_checkpoint(
        &self,
        topic_name: &str,
        up_to_offset: u64,
    ) -> Result<(), PersistentStorageError>;

    /// Optionally force a flush of buffered data.
    async fn flush(&self, topic_name: &str) -> Result<(), PersistentStorageError>;

    /// Return the next offset that will be assigned to the next appended message.
    fn current_offset(&self) -> u64;
}