Struct DbConnection 

pub struct DbConnection<C> { /* private fields */ }

A wrapper for RawDbConnection that houses all XMTP DB operations.

Implementations

impl<C> DbConnection<C>

pub fn new(conn: C) -> Self

Trait Implementations

impl<C: Clone> Clone for DbConnection<C>

fn clone(&self) -> DbConnection<C>

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl<C> ConnectionExt for DbConnection<C>

where C: ConnectionExt,

fn raw_query<T, F>(&self, fun: F) -> Result<T, ConnectionError>

where F: FnOnce(&mut SqliteConnection) -> Result<T, Error>, Self: Sized,

Run a scoped query against the underlying SQLite connection.

fn disconnect(&self) -> Result<(), ConnectionError>

fn reconnect(&self) -> Result<(), ConnectionError>

impl<C> Debug for DbConnection<C>

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

Formats the value using the given formatter.

impl<C: ConnectionExt> Delete<StoredGroupIntent> for DbConnection<C>

type Key = i32

fn delete(&self, key: ID) -> Result<usize, StorageError>

impl<C: ConnectionExt> Delete<StoredKeyStoreEntry> for DbConnection<C>

type Key = Vec<u8>

fn delete(&self, key: Vec<u8>) -> Result<usize, StorageError>

impl<C: ConnectionExt> From<DbConnection<C>> for XmtpOpenMlsProvider<SqlKeyStore<C>>

fn from(db: DbConnection<C>) -> XmtpOpenMlsProvider<SqlKeyStore<C>>

Converts to this type from the input type.

impl<C: ConnectionExt> IntoConnection for DbConnection<C>

type Connection = C

fn into_connection(self) -> Self::Connection

impl<C: ConnectionExt> Pragmas for DbConnection<C>

fn busy_timeout(&self) -> Result<i32, ConnectionError>

Check the busy timeout value

fn set_sqlcipher_log<S: AsRef<str>>( &self, level: S, ) -> Result<(), ConnectionError>

impl<C: ConnectionExt> QueryAssociationStateCache for DbConnection<C>

fn write_to_cache( &self, inbox_id: String, sequence_id: i64, state: AssociationStateProto, ) -> Result<(), StorageError>

fn read_from_cache<A: AsRef<str>>( &self, inbox_id: A, sequence_id: i64, ) -> Result<Option<AssociationStateProto>, StorageError>

fn batch_read_from_cache( &self, identifiers: Vec<(String, i64)>, ) -> Result<Vec<AssociationStateProto>, StorageError>

impl<C: ConnectionExt> QueryConsentRecord for DbConnection<C>

fn get_consent_record( &self, entity: String, entity_type: ConsentType, ) -> Result<Option<StoredConsentRecord>, ConnectionError>

Returns the consent_records for the given entity up

fn insert_or_replace_consent_records( &self, records: &[StoredConsentRecord], ) -> Result<Vec<StoredConsentRecord>, ConnectionError>

Insert consent_records, and replace existing entries, returns records that are new or changed

fn consent_records(&self) -> Result<Vec<StoredConsentRecord>, ConnectionError>

fn consent_records_paged( &self, limit: i64, offset: i64, ) -> Result<Vec<StoredConsentRecord>, ConnectionError>

fn insert_newer_consent_record( &self, record: StoredConsentRecord, ) -> Result<bool, ConnectionError>

Returns true if newer

fn maybe_insert_consent_record_return_existing( &self, record: &StoredConsentRecord, ) -> Result<Option<StoredConsentRecord>, ConnectionError>

fn find_consent_by_dm_id( &self, dm_id: &str, ) -> Result<Vec<StoredConsentRecord>, ConnectionError>

impl<C: ConnectionExt> QueryConversationList for DbConnection<C>

fn fetch_conversation_list<A: AsRef<GroupQueryArgs>>( &self, args: A, ) -> Result<Vec<ConversationListItem>, StorageError>

impl<C: ConnectionExt> QueryDeviceSyncMessages for DbConnection<C>

fn unprocessed_sync_group_messages( &self, ) -> Result<Vec<StoredGroupMessage>, StorageError>

fn sync_group_messages_paged( &self, offset: i64, limit: i64, ) -> Result<Vec<StoredGroupMessage>, StorageError>

fn mark_device_sync_msg_as_processed( &self, message_id: &[u8], ) -> Result<(), StorageError>

Marks a device sync message as processed.

fn increment_device_sync_msg_attempt( &self, message_id: &[u8], max_attempts: i32, ) -> Result<i32, StorageError>

Increments the attempt count for a device sync message. If the attempt count reaches max_attempts, the state is set to Failed. Returns the new attempt count.

impl<C: ConnectionExt> QueryDms for DbConnection<C>

fn fetch_stitched( &self, key: &GroupId, ) -> Result<Option<StoredGroup>, ConnectionError>

Same behavior as fetched, but will stitch DM groups

fn other_dms( &self, group_id: &GroupId, ) -> Result<Vec<StoredGroup>, ConnectionError>

Load the other DMs that are stitched into this group

fn find_active_dm_group<M>( &self, members: M, ) -> Result<Option<StoredGroup>, ConnectionError>

where M: Display,

impl<C: ConnectionExt> QueryGroup for DbConnection<C>

fn find_groups<A: AsRef<GroupQueryArgs>>( &self, args: A, ) -> Result<Vec<StoredGroup>, ConnectionError>

Return regular Purpose::Conversation groups with additional optional filters

fn update_group_membership<Id: AsRef<[u8]>>( &self, group_id: Id, state: GroupMembershipState, ) -> Result<(), ConnectionError>

Updates group membership state

fn find_group( &self, id: &GroupId, ) -> Result<Option<StoredGroup>, ConnectionError>

Return a single group that matches the given ID

fn find_group_by_sequence_id( &self, cursor: Cursor, ) -> Result<Option<StoredGroup>, ConnectionError>

Return a single group that matches the given welcome ID

fn update_rotated_at_ns(&self, group_id: &GroupId) -> Result<(), StorageError>

Updates the ‘last time checked’ we checked for new installations.

fn update_installations_time_checked( &self, group_id: &GroupId, ) -> Result<(), StorageError>

Updates the ‘last time checked’ we checked for new installations.

fn group_cursors(&self) -> Result<Vec<Cursor>, ConnectionError>

Get all the welcome ids turned into groups

fn get_conversation_ids_for_remote_log_publish( &self, ) -> Result<Vec<StoredGroupCommitLogPublicKey>, ConnectionError>

Get conversation IDs for all conversations that require a remote commit log publish (DMs and groups where user is super admin, excluding sync groups and rejected groups)

fn find_groups_by_id_paged<A: AsRef<GroupQueryArgs>>( &self, args: A, offset: i64, ) -> Result<Vec<StoredGroup>, ConnectionError>

fn all_sync_groups(&self) -> Result<Vec<StoredGroup>, ConnectionError>

fn find_sync_group( &self, id: &GroupId, ) -> Result<Option<StoredGroup>, ConnectionError>

fn primary_sync_group(&self) -> Result<Option<StoredGroup>, ConnectionError>

fn get_rotated_at_ns(&self, group_id: &GroupId) -> Result<i64, StorageError>

fn get_installations_time_checked( &self, group_id: &GroupId, ) -> Result<i64, StorageError>

fn update_message_disappearing_from_ns( &self, group_id: &GroupId, from_ns: Option<i64>, ) -> Result<(), StorageError>

fn update_message_disappearing_in_ns( &self, group_id: &GroupId, in_ns: Option<i64>, ) -> Result<(), StorageError>

fn insert_or_replace_group( &self, group: StoredGroup, ) -> Result<StoredGroup, StorageError>

fn mark_group_as_maybe_forked( &self, group_id: &GroupId, fork_details: String, ) -> Result<(), StorageError>

fn clear_fork_flag_for_group( &self, group_id: &GroupId, ) -> Result<(), ConnectionError>

fn has_duplicate_dm(&self, group_id: &GroupId) -> Result<bool, ConnectionError>

fn get_conversation_ids_for_remote_log_download( &self, ) -> Result<Vec<StoredGroupCommitLogPublicKey>, ConnectionError>

Get conversations for all conversations that require a remote commit log download (DMs and groups that are not sync groups)

fn get_conversation_ids_for_fork_check( &self, ) -> Result<Vec<Vec<u8>>, ConnectionError>

Get conversation IDs for fork checking (excludes already forked conversations and sync groups)

fn get_conversation_ids_for_requesting_readds( &self, ) -> Result<Vec<StoredGroupForReaddRequest>, ConnectionError>

Get conversation IDs for conversations that are forked and need readd requests

fn get_conversation_ids_for_responding_readds( &self, ) -> Result<Vec<StoredGroupForRespondingReadds>, ConnectionError>

Get conversation IDs for conversations that need to respond to readd requests

fn get_conversation_type( &self, group_id: &GroupId, ) -> Result<ConversationType, ConnectionError>

fn set_group_commit_log_public_key( &self, group_id: &GroupId, public_key: &[u8], ) -> Result<(), StorageError>

Updates the commit log public key for a group

fn set_group_commit_log_forked_status( &self, group_id: &GroupId, is_forked: Option<bool>, ) -> Result<(), StorageError>

Updates the is_commit_log_forked status for a group

fn get_group_commit_log_forked_status( &self, group_id: &GroupId, ) -> Result<Option<bool>, StorageError>

Gets the is_commit_log_forked status for a group

fn set_group_has_pending_leave_request_status( &self, group_id: &GroupId, has_pending_leave_request: Option<bool>, ) -> Result<(), StorageError>

Updates the has_pending_leave_request status for a group

fn get_groups_have_pending_leave_request( &self, ) -> Result<Vec<Vec<u8>>, ConnectionError>

impl<C: ConnectionExt> QueryGroupIntent for DbConnection<C>

fn set_group_intent_superseded(&self, intent_id: ID) -> Result<(), StorageError>

Mark the intent abandoned because its compare-and-swap guard no longer matches. Terminal, and deliberately not Error: the caller needs to tell a stale write apart from a genuine failure.

fn supersede_pending_intents_for_inactive_group( &self, group_id: &[u8], ) -> Result<usize, StorageError>

Removal is terminal for work that has not been accepted by the group. The state is Error, not Superseded: nothing raced this write. A ToPublish intent can never be published now, and a Published one can never be confirmed: its own echo is unreachable behind the inactive boundary, and a later re-add installs fresh state past it. Leaving those intents in place strands them, and a stranded Published state change preempts every later intent on the group.

The prepared attempt is cleared with the state so no stale bytes can be reused against a new membership generation.

fn set_group_intent_error(&self, intent_id: ID) -> Result<(), StorageError>

Set the intent with the given ID to Error

fn find_dependant_commits<P: AsRef<[u8]>>( &self, payload_hashes: &[P], ) -> Result<HashMap<PayloadHash, IntentDependency>, StorageError>

Find the commit message refresh state for each intent by payload hash. Returns a map from payload hash to a vector of group dependencies.

fn insert_group_intent( &self, to_save: NewGroupIntent, ) -> Result<StoredGroupIntent, ConnectionError>

fn find_group_intents<Id: AsRef<[u8]>>( &self, group_id: Id, allowed_states: Option<Vec<IntentState>>, allowed_kinds: Option<Vec<IntentKind>>, ) -> Result<Vec<StoredGroupIntent>, ConnectionError>

fn set_group_intent_published( &self, intent_id: ID, payload_hash: &[u8], post_commit_data: Option<Vec<u8>>, staged_commit: Option<Vec<u8>>, published_in_epoch: i64, ) -> Result<(), StorageError>

fn set_group_intent_committed( &self, intent_id: ID, cursor: Cursor, ) -> Result<(), StorageError>

fn set_group_intent_processed(&self, intent_id: ID) -> Result<(), StorageError>

fn set_group_intent_to_publish(&self, intent_id: ID) -> Result<(), StorageError>

fn find_group_intent_by_payload_hash( &self, payload_hash: &[u8], ) -> Result<Option<StoredGroupIntent>, StorageError>

fn own_intent_kind_is_unreadable( &self, payload_hash: &[u8], ) -> Result<bool, StorageError>

True when a row with this payload hash exists but carries an IntentKind this build cannot decode.

fn increment_intent_publish_attempt_count( &self, intent_id: ID, ) -> Result<(), StorageError>

fn set_group_intent_error_and_fail_msg( &self, intent: &StoredGroupIntent, msg_id: Option<Vec<u8>>, ) -> Result<(), StorageError>

impl<C: ConnectionExt> QueryGroupMessage for DbConnection<C>

fn get_group_messages( &self, group_id: &GroupId, args: &MsgQueryArgs, ) -> Result<Vec<StoredGroupMessage>, ConnectionError>

Query for group messages

fn count_group_messages( &self, group_id: &GroupId, args: &MsgQueryArgs, ) -> Result<i64, ConnectionError>

Count group messages matching the given criteria

fn get_group_messages_with_reactions( &self, group_id: &GroupId, args: &MsgQueryArgs, ) -> Result<Vec<StoredGroupMessageWithReactions>, ConnectionError>

Query for group messages with their reactions

fn get_group_message<MessageId: AsRef<[u8]>>( &self, id: MessageId, ) -> Result<Option<StoredGroupMessage>, ConnectionError>

Get a particular group message

fn write_conn_get_group_message<MessageId: AsRef<[u8]>>( &self, id: MessageId, ) -> Result<Option<StoredGroupMessage>, ConnectionError>

Get a particular group message using the write connection

fn missing_messages( &self, group_id: &GroupId, sequence_ids: &[u64], ) -> Result<Vec<StoredGroupMessage>, ConnectionError>

Return all Application-kind messages stored locally for group_id whose sequence_id is NOT in the provided list. Used by tools that compare local state against an authoritative set of sequence ids (e.g. xdbg’s healthcheck validator).

fn group_messages_paged( &self, args: &MsgQueryArgs, offset: i64, ) -> Result<Vec<StoredGroupMessage>, ConnectionError>

fn get_inbound_relations( &self, group_id: &GroupId, message_ids: &[&[u8]], relation_query: RelationQuery, ) -> Result<InboundRelations, ConnectionError>

fn get_outbound_relations( &self, group_id: &GroupId, reference_ids: &[&[u8]], ) -> Result<OutboundRelations, ConnectionError>

fn get_inbound_relation_counts( &self, group_id: &GroupId, message_ids: &[&[u8]], relation_query: RelationQuery, ) -> Result<RelationCounts, ConnectionError>

fn get_latest_message_times_by_sender<Id: AsRef<[u8]>>( &self, group_id: Id, allowed_content_types: &[ContentType], ) -> Result<LatestMessageTimeBySender, ConnectionError>

fn get_group_message_by_timestamp<Id: AsRef<[u8]>>( &self, group_id: Id, timestamp: i64, ) -> Result<Option<StoredGroupMessage>, ConnectionError>

fn get_group_message_by_cursor<Id: AsRef<[u8]>>( &self, group_id: Id, cursor: Cursor, ) -> Result<Option<StoredGroupMessage>, ConnectionError>

fn set_delivery_status_to_published<MessageId: AsRef<[u8]>>( &self, msg_id: &MessageId, timestamp: u64, cursor: Cursor, message_expire_at_ns: Option<i64>, ) -> Result<usize, StorageError>

fn set_delivery_status_to_failed<MessageId: AsRef<[u8]>>( &self, msg_id: &MessageId, ) -> Result<usize, ConnectionError>

fn delete_expired_messages( &self, ) -> Result<Vec<StoredGroupMessage>, ConnectionError>

fn min_expire_at_ns(&self) -> Result<Option<i64>, ConnectionError>

The soonest expire_at_ns among published Application messages that have an expiry set, or None if no disappearing messages exist. Note this can return a timestamp already in the past (an expiry that elapsed while the worker was asleep) — the caller clamps the resulting sleep to >= 0 and deletes on the next wake. Same filters as delete_expired_messages without its expire_at_ns <= now bound.

fn delete_message_by_id<MessageId: AsRef<[u8]>>( &self, message_id: MessageId, ) -> Result<usize, ConnectionError>

fn messages_newer_than( &self, cursors_by_group: &HashMap<Vec<u8>, Cursor>, ) -> Result<Vec<(GroupId, Cursor)>, ConnectionError>

Stored messages above each group’s cursor, attributed to their group. The attribution matters: sequence ids are not scoped per group, so a caller folding these into per-group state must never mix groups.

fn clear_messages( &self, group_ids: Option<&[GroupId]>, retention_days: Option<u32>, ) -> Result<usize, ConnectionError>

Clear messages from the database with optional filtering.

impl<C: ConnectionExt> QueryGroupVersion for DbConnection<C>

fn set_group_paused( &self, group_id: &GroupId, min_version: &str, ) -> Result<(), StorageError>

fn unpause_group(&self, group_id: &GroupId) -> Result<(), StorageError>

fn get_group_paused_version( &self, group_id: &GroupId, ) -> Result<Option<String>, StorageError>

fn get_paused_groups_with_versions( &self, ) -> Result<Vec<(GroupId, String)>, StorageError>

Return every group currently flagged as paused, with the paused_for_version floor it’s pinned to. Used by the startup/sweep recovery path to re-evaluate paused groups against the now-current pkg_version without having to sync each group individually.

impl<C: ConnectionExt> QueryIdentity for DbConnection<C>

fn queue_key_package_rotation(&self) -> Result<(), StorageError>

fn queue_key_rotation_with_nudge( &self, rotation_task_hash: &TaskDataHash, rotation_seed: NewTask, ) -> Result<(), StorageError>

Atomically lower/initialize the rotation column (5s debounce) AND enqueue a PullInDeadline task targeting rotation_task_hash at the resulting column value — one transaction, so neither write can land without the other. rotation_seed is insert-or-ignored first so the pull-in always has a live target (commit-target-first), even if startup seeding never ran. Callers wake the TaskWorker AFTER this returns (never inside a tx).

fn reset_key_package_rotation_queue( &self, rotation_interval_ns: i64, ) -> Result<(), StorageError>

fn is_identity_needs_rotation(&self) -> Result<bool, StorageError>

fn next_key_package_rotation_ns(&self) -> Result<Option<i64>, StorageError>

The identity’s absolute rotation deadline (next_key_package_rotation_ns). None if NULL or if no identity row exists yet (indistinguishable to callers; treat as “no scheduled deadline”).

impl<C: ConnectionExt> QueryIdentityCache for DbConnection<C>

fn fetch_cached_inbox_ids( &self, identifiers: &[(String, StoredIdentityKind)], ) -> Result<HashMap<String, String>, StorageError>

Returns a HashMap of WalletAddress -> InboxId

fn cache_inbox_id<S: ToString>( &self, kind: StoredIdentityKind, identity: String, inbox_id: S, ) -> Result<(), StorageError>

impl<C: ConnectionExt> QueryIdentityUpdates for DbConnection<C>

fn get_identity_updates<InboxId: AsRef<str>>( &self, inbox_id: InboxId, from_sequence_id: Option<i64>, to_sequence_id: Option<i64>, ) -> Result<Vec<StoredIdentityUpdate>, ConnectionError>

Returns all identity updates for the given inbox ID up to the provided sequence_id. Returns updates greater than from_sequence_id and less than or equal to to_sequence_id

fn insert_or_ignore_identity_updates( &self, updates: &[StoredIdentityUpdate], ) -> Result<(), ConnectionError>

Batch insert identity updates, ignoring duplicates.

fn get_latest_sequence_id( &self, inbox_ids: &[&str], ) -> Result<HashMap<String, i64>, ConnectionError>

Given a list of inbox_ids return a HashMap of each inbox ID -> highest known sequence ID

fn get_latest_sequence_id_for_inbox( &self, inbox_id: &str, ) -> Result<i64, ConnectionError>

fn count_inbox_updates( &self, inbox_ids: &[&str], ) -> Result<HashMap<String, i64>, ConnectionError>

Returns the count of identity updates for inbox_ids

impl<C: ConnectionExt> QueryKeyPackageHistory for DbConnection<C>

fn record_key_package_publication( &self, history_id: i32, sequence: Cursor, ) -> Result<(), StorageError>

Retire keys by confirmed publication order, not local creation order. The latest published key stays usable; duplicate receipts keep the first retirement deadline.

fn store_key_package_history_entry( &self, key_package_hash_ref: Vec<u8>, post_quantum_public_key: Option<Vec<u8>>, ) -> Result<StoredKeyPackageHistoryEntry, StorageError>

fn find_key_package_history_entry_by_hash_ref( &self, hash_ref: Vec<u8>, ) -> Result<StoredKeyPackageHistoryEntry, StorageError>

fn find_key_package_history_entries_before_id( &self, id: i32, ) -> Result<Vec<StoredKeyPackageHistoryEntry>, StorageError>

fn get_expired_key_packages( &self, ) -> Result<Vec<StoredKeyPackageHistoryEntry>, StorageError>

fn min_key_package_delete_at_ns(&self) -> Result<Option<i64>, StorageError>

Soonest pending delete_at_ns across all key packages marked for deletion, or None if none are marked. The KpDeletion task’s reschedule source.

fn delete_key_package_entry_with_id(&self, id: i32) -> Result<(), StorageError>

impl<C: ConnectionExt> QueryKeyStoreEntry for DbConnection<C>

fn insert_or_update_key_store_entry( &self, key: Vec<u8>, value: Vec<u8>, ) -> Result<(), StorageError>

impl<C: ConnectionExt> QueryLocalCommitLog for DbConnection<C>

fn get_group_logs( &self, group_id: &GroupId, ) -> Result<Vec<LocalCommitLog>, ConnectionError>

fn get_local_commit_log_after_cursor( &self, group_id: &GroupId, after_cursor: i64, order: LocalCommitLogOrder, ) -> Result<Vec<LocalCommitLog>, ConnectionError>

fn get_latest_log_for_group( &self, group_id: &GroupId, ) -> Result<Option<LocalCommitLog>, ConnectionError>

fn get_local_commit_log_cursor( &self, group_id: &GroupId, ) -> Result<Option<i32>, ConnectionError>

fn get_latest_chain_start_rowid( &self, group_id: &GroupId, ) -> Result<Option<i32>, ConnectionError>

Rowid of the most recent chain-start entry for this group, if any. Chain-start entries have commit_sequence_id == 0 (Welcome / GroupCreation / BackupRestore) and mark the beginning of the member’s current membership session.

impl<C: ConnectionExt> QueryMessageDeletion for DbConnection<C>

fn get_message_deletion( &self, id: &[u8], ) -> Result<Option<StoredMessageDeletion>, ConnectionError>

Get a deletion record by the DeleteMessage ID

fn get_deletion_by_deleted_message_id( &self, deleted_message_id: &[u8], ) -> Result<Option<StoredMessageDeletion>, ConnectionError>

Get deletion record for a specific deleted message

fn get_deletions_for_messages( &self, message_ids: Vec<Vec<u8>>, ) -> Result<Vec<StoredMessageDeletion>, ConnectionError>

Get all deletions for a list of message IDs

fn get_group_deletions( &self, group_id: &GroupId, ) -> Result<Vec<StoredMessageDeletion>, ConnectionError>

Get all deletions in a group

fn is_message_deleted(&self, message_id: &[u8]) -> Result<bool, ConnectionError>

Check if a message has been deleted

impl<C: ConnectionExt> QueryMigrations for DbConnection<C>

fn applied_migrations(&self) -> Result<Vec<String>, ConnectionError>

Returns a list of all applied migration versions, most recent first.

fn available_migrations(&self) -> Result<Vec<String>, ConnectionError>

Returns a list of all available (embedded) migration names.

fn rollback_to_version( &self, version: &str, ) -> Result<Vec<String>, ConnectionError>

Rollback all migrations after and including the specified version.

fn run_migration(&self, name: &str) -> Result<(), ConnectionError>

Run a specific migration by name.

fn revert_migration(&self, name: &str) -> Result<(), ConnectionError>

Revert a specific migration by name.

fn run_pending_migrations(&self) -> Result<Vec<String>, ConnectionError>

Run all pending migrations.

impl<C: ConnectionExt> QueryNotifications for DbConnection<C>

fn notification_record(&self) -> Result<StoredNotification, StorageError>

fn save_notification_record( &self, record: &StoredNotification, ) -> Result<(), StorageError>

fn uploaded_topics(&self) -> Result<Vec<UploadedTopic>, StorageError>

fn disable_notifications( &self, ) -> Result<(StoredNotification, Vec<UploadedTopic>), StorageError>

Atomically disable notifications and return the cleared uploads. Keep the recipient identity, conversation overrides, and task retry state.

fn confirm_uploaded_topics( &self, adds: &[UploadedTopic], removes: &[Vec<u8>], ) -> Result<(), StorageError>

fn clear_uploaded_topics(&self) -> Result<(), StorageError>

fn mark_uploaded_topics_stale(&self) -> Result<(), StorageError>

fn notification_groups(&self) -> Result<Vec<StoredGroup>, StorageError>

fn set_notification_override( &self, group: &GroupId, value: Option<i32>, ) -> Result<(), StorageError>

impl<C: ConnectionExt> QueryPendingRemove for DbConnection<C>

fn get_pending_remove_users( &self, group_id: &GroupId, ) -> Result<Vec<String>, ConnectionError>

fn get_user_pending_remove_status( &self, group_id: &GroupId, inbox_id: &str, ) -> Result<bool, ConnectionError>

fn delete_pending_remove_users( &self, group_id: &GroupId, inbox_ids: Vec<String>, ) -> Result<usize, ConnectionError>

impl<C: ConnectionExt> QueryReaddStatus for DbConnection<C>

fn get_readd_status( &self, group_id: &GroupId, installation_id: &[u8], ) -> Result<Option<ReaddStatus>, ConnectionError>

fn is_awaiting_readd( &self, group_id: &GroupId, installation_id: &[u8], ) -> Result<bool, ConnectionError>

fn update_requested_at_sequence_id( &self, group_id: &GroupId, installation_id: &[u8], sequence_id: i64, ) -> Result<(), ConnectionError>

Update the requested_at_sequence_id for a given group_id and installation_id, provided it is higher than the current value. Inserts the row if it doesn’t exist.

fn update_responded_at_sequence_id( &self, group_id: &GroupId, installation_id: &[u8], sequence_id: i64, ) -> Result<(), ConnectionError>

Update the responded_at_sequence_id for a given group_id and installation_id, provided it is higher than the current value. Inserts the row if it doesn’t exist.

fn delete_other_readd_statuses( &self, group_id: &GroupId, self_installation_id: &[u8], ) -> Result<(), ConnectionError>

fn delete_readd_statuses( &self, group_id: &GroupId, installation_ids: HashSet<Vec<u8>>, ) -> Result<(), ConnectionError>

fn get_readds_awaiting_response( &self, group_id: &GroupId, self_installation_id: &[u8], ) -> Result<Vec<ReaddStatus>, ConnectionError>

impl<C: ConnectionExt> QueryRefreshState for DbConnection<C>

fn get_refresh_state<Id: AsRef<[u8]>>( &self, entity_id: Id, entity_kind: EntityKind, ) -> Result<Option<RefreshState>, StorageError>

fn get_last_cursor<Id: AsRef<[u8]>>( &self, id: Id, entity_kind: EntityKind, ) -> Result<Cursor, StorageError>

Read one ledger position. Create a zero position when it is absent.

fn get_last_cursor_for_ids<Id: AsRef<[u8]>>( &self, ids: &[Id], entities: &[EntityKind], ) -> Result<HashMap<Vec<u8>, Cursor>, StorageError>

Return the minimum position across the requested kinds for each stored id. An absent kind has position zero. Ids with no rows are absent from the map.

fn update_cursor<Id: AsRef<[u8]>>( &self, entity_id: Id, entity_kind: EntityKind, cursor: Cursor, ) -> Result<bool, StorageError>

Advance a ledger position only when the new position is greater.

fn latest_cursor_for_id<Id: AsRef<[u8]>>( &self, entity_id: Id, entities: &[EntityKind], ) -> Result<Cursor, StorageError>

fn get_remote_log_cursors( &self, conversation_ids: &[&[u8]], ) -> Result<HashMap<Vec<u8>, Cursor>, StorageError>

impl<C: ConnectionExt> QueryRemoteCommitLog for DbConnection<C>

fn get_latest_remote_log_for_group( &self, group_id: &GroupId, ) -> Result<Option<RemoteCommitLog>, ConnectionError>

fn get_remote_commit_log_after_cursor( &self, group_id: &GroupId, after_cursor: i64, order: RemoteCommitLogOrder, ) -> Result<Vec<RemoteCommitLog>, ConnectionError>

impl<C: ConnectionExt> QueryServerConfiguration for DbConnection<C>

fn server_configuration( &self, ) -> Result<Option<StoredServerConfiguration>, StorageError>

The stored copy, or None when this database has never held one.

fn store_server_configuration( &self, identifier: &str, backend_url: &str, response: &[u8], fetched_at_ns: i64, ) -> Result<(), StorageError>

Write the copy whole: identifier, URL, response, and fetch time. Never touches conflicting_identifier, so a matching refresh cannot erase a recorded conflict (CFG-053).

fn record_server_configuration_conflict( &self, conflicting_identifier: &str, ) -> Result<(), StorageError>

Record that the deployment answered with a different identifier (CFG-051). Does nothing when no row exists yet.

impl<C: ConnectionExt> QueryTasks for DbConnection<C>

fn create_task(&self, task: NewTask) -> Result<Task, StorageError>

fn create_or_ignore_task(&self, task: NewTask) -> Result<(), StorageError>

Idempotent enqueue: a payload-identical duplicate is a no-op (the existing row wins; OR IGNORE swallows any constraint hit, not just data_hash UNIQUE).

fn pull_in_task_deadline( &self, target_data_hash: &TaskDataHash, at_ns: i64, ) -> Result<bool, StorageError>

Lower a task’s next_attempt_at_ns to MIN(current, at_ns) — never raises. Returns whether a row matched; a missing target is a no-op (false). TaskWorker dispatch thread only (sole rescheduler).

fn get_tasks(&self) -> Result<Vec<Task>, StorageError>

fn get_next_task(&self) -> Result<Option<Task>, StorageError>

fn upsert_pending_self_remove_task( &self, group_id: &GroupId, task: NewTask, ) -> Result<(), StorageError>

Ensure exactly one live ProcessPendingSelfRemove task exists for group_id. Clears only dead rows (expired / attempts-exhausted) then insert-or-ignores, so a live retrying task keeps its backoff and is never deleted out from under the TaskRunner, while a stale dead row can’t block a fresh retry via the data_hash unique constraint.

fn update_task( &self, id: i32, attempts: i32, last_attempted_at_ns: i64, next_attempt_at_ns: i64, ) -> Result<Task, StorageError>

fn delete_task(&self, id: i32) -> Result<bool, StorageError>

impl<C: ConnectionExt> ReadOnly for DbConnection<C>

fn enable_readonly(&self) -> Result<(), StorageError>

fn disable_readonly(&self) -> Result<(), StorageError>

impl<T, C> Store<DbConnection<C>> for Vec<T>

where T: Store<DbConnection<C>>,

type Output = ()

fn store(&self, into: &DbConnection<C>) -> Result<Self::Output, StorageError>