Struct GenericMessagePersistenceManager 

pub struct GenericMessagePersistenceManager<T: MessagesManagerTrait> {
    messages_manager: Arc<T>,
    persistence_task: Arc<AbortOnDrop<Result<()>>>,
}

High-level message persistence manager that automatically stores messages.

The MessagePersistenceManager bridges the gap between real-time messaging and persistent storage by:

• Automatic Persistence: Stores all message events as they occur
• Relay Sync Tracking: Tracks which messages have been synced to which relays
• Non-blocking Operation: Runs persistence in the background without affecting message flow

This manager operates by subscribing to the message events stream from MessagesManager and automatically persisting all events to the configured storage backend.

Fields

messages_manager: Arc<T>

The messages manager that this persistence manager wraps

persistence_task: Arc<AbortOnDrop<Result<()>>>

Handle to the background persistence task

Implementations

impl<T: MessagesManagerTrait> GenericMessagePersistenceManager<T>

pub fn messages_manager(&self) -> &Arc<T>

Get a reference to the underlying MessagesManager

This provides explicit access to the MessagesManager for cases where method discovery is important or when you need to pass it to other components.

pub async fn load_subscription_state( storage: &dyn MessageStorage, relay_id: &KeyId, ) -> Result<Option<SubscriptionState>>

Load subscription state from storage for a specific relay

pub async fn load_all_subscription_states( storage: &dyn MessageStorage, ) -> Result<HashMap<KeyId, SubscriptionState>>

Load all subscription states from storage

async fn new( storage: Arc<dyn MessageStorage>, messages_manager: Arc<T>, relay_id: KeyId, ) -> Result<Self>

Create a new MessagePersistenceManager with the given components.

This starts the background persistence task immediately.

Arguments

• storage - The storage implementation to persist messages to
• messages_manager - The messages manager to monitor for events
• relay_pubkey - Optional relay public key for sync tracking
• buffer_size - Optional buffer size for the task queue

async fn handle_message_event( storage: &dyn MessageStorage, event: &MessageEvent, relay_id: &KeyId, ) -> Result<()>

Handle a single message event by persisting it

pub fn is_running(&self) -> bool

Check if the persistence task is still running

impl GenericMessagePersistenceManager<MessagesManager>

pub fn builder() -> MessagePersistenceManagerBuilder

Create a new MessagePersistenceManager builder

pub fn inner_messages_manager(&self) -> &MessagesManager

Get a reference to the inner MessagesManager for use with components that require the concrete type (like PqxdhProtocolHandler)

This method is only available on the concrete MessagePersistenceManager type alias, not the generic version.

Trait Implementations

impl<T: Clone + MessagesManagerTrait> Clone for GenericMessagePersistenceManager<T>

fn clone(&self) -> GenericMessagePersistenceManager<T>

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<T: Debug + MessagesManagerTrait> Debug for GenericMessagePersistenceManager<T>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl<T: MessagesManagerTrait + 'static> MessagesManagerTrait for GenericMessagePersistenceManager<T>

fn message_events_stream(&self) -> Receiver<MessageEvent>

Get a stream of all message events for persistence and monitoring

fn get_subscription_state_updates<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Subscriber<SubscriptionState, AsyncLock>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Subscribe to subscription state changes for reactive programming

fn subscribe<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Subscribe to messages with current filters

fn publish<'life0, 'async_trait>( &'life0 self, message: MessageFull, ) -> Pin<Box<dyn Future<Output = Result<PublishResult>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Publish a message

fn ensure_contains_filter<'life0, 'async_trait>( &'life0 self, filter: Filter, ) -> Pin<Box<dyn Future<Output = Result<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Ensure a filter is included in the subscription

fn messages_stream(&self) -> Receiver<StreamMessage>

Get a stream of incoming messages

fn catch_up_stream(&self) -> Receiver<CatchUpResponse>

Get a stream of catch-up responses

fn filtered_messages_stream( &self, filter: Filter, ) -> Pin<Box<dyn Stream<Item = Box<MessageFull>> + Send>>

Get a filtered stream of messages matching the given filter

fn catch_up_and_subscribe<'life0, 'async_trait>( &'life0 self, filter: Filter, since: Option<String>, ) -> Pin<Box<dyn Future<Output = Result<Pin<Box<dyn Stream<Item = Box<MessageFull>> + Send>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Catch up to historical messages and subscribe to new ones for a filter

fn user_data<'life0, 'async_trait>( &'life0 self, author: KeyId, storage_key: StoreKey, ) -> Pin<Box<dyn Future<Output = Result<Option<MessageFull>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Get user data by author and storage key (for PQXDH inbox fetching)

fn check_messages<'life0, 'async_trait>( &'life0 self, message_ids: Vec<MessageId>, ) -> Pin<Box<dyn Future<Output = Result<Vec<Option<String>>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Check which messages the server already has and return their global stream IDs. Returns a vec of Option<String> in the same order as the input, where: