chore_: cleanup `hash_ratchet_encrypted_messages` periodically

closes: #5036

protocol/common/message_sender.go

@@ -1278,3 +1278,14 @@ func (s *MessageSender) GetCurrentKeyForGroup(groupID []byte) (*encryption.HashR
 func (s *MessageSender) GetKeysForGroup(groupID []byte) ([]*encryption.HashRatchetKeyCompatibility, error) {
 	return s.protocol.GetKeysForGroup(groupID)
 }
+
+func (s *MessageSender) CleanupHashRatchetEncryptedMessages() error {
+	monthAgo := time.Now().AddDate(0, -1, 0).Unix()
+
+	err := s.persistence.DeleteHashRatchetMessagesOlderThan(monthAgo)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}

protocol/common/raw_messages_persistence.go

@@ -372,6 +372,11 @@ func (db RawMessagesPersistence) DeleteHashRatchetMessages(ids [][]byte) error {
 	return err
 }
 
+func (db *RawMessagesPersistence) DeleteHashRatchetMessagesOlderThan(timestamp int64) error {
+	_, err := db.db.Exec("DELETE FROM hash_ratchet_encrypted_messages WHERE timestamp < ?", timestamp)
+	return err
+}
+
 func (db *RawMessagesPersistence) IsMessageAlreadyCompleted(hash []byte) (bool, error) {
 	var alreadyCompleted int
 	err := db.db.QueryRow("SELECT COUNT(*) FROM message_segments_completed WHERE hash = ?", hash).Scan(&alreadyCompleted)

protocol/messenger.go

@@ -792,6 +792,7 @@ func (m *Messenger) Start() (*MessengerResponse, error) {
 		m.startCuratedCommunitiesUpdateLoop()
 	}
 	m.startMessageSegmentsCleanupLoop()
+	m.startHashRatchetEncryptedMessagesCleanupLoop()
 
 	if err := m.cleanTopics(); err != nil {
 		return nil, err
@@ -6006,8 +6007,8 @@ func (m *Messenger) GetDeleteForMeMessages() ([]*protobuf.SyncDeleteForMeMessage
 	return m.persistence.GetDeleteForMeMessages()
 }
 
-func (m *Messenger) startMessageSegmentsCleanupLoop() {
-	logger := m.logger.Named("messageSegmentsCleanupLoop")
+func (m *Messenger) startCleanupLoop(name string, cleanupFunc func() error) {
+	logger := m.logger.Named(name)
 
 	go func() {
 		// Delay by a few minutes to minimize messenger's startup time
@@ -6018,9 +6019,9 @@ func (m *Messenger) startMessageSegmentsCleanupLoop() {
 				// Set the regular interval after the first execution
 				interval = 1 * time.Hour
 
-				err := m.sender.CleanupSegments()
+				err := cleanupFunc()
 				if err != nil {
-					logger.Error("failed to cleanup segments", zap.Error(err))
+					logger.Error("failed to cleanup", zap.Error(err))
 				}
 
 			case <-m.quit:
@@ -6030,6 +6031,14 @@ func (m *Messenger) startMessageSegmentsCleanupLoop() {
 	}()
 }
 
+func (m *Messenger) startMessageSegmentsCleanupLoop() {
+	m.startCleanupLoop("messageSegmentsCleanupLoop", m.sender.CleanupSegments)
+}
+
+func (m *Messenger) startHashRatchetEncryptedMessagesCleanupLoop() {
+	m.startCleanupLoop("hashRatchetEncryptedMessagesCleanupLoop", m.sender.CleanupHashRatchetEncryptedMessages)
+}
+
 func (m *Messenger) FindStatusMessageIDForBridgeMessageID(bridgeMessageID string) (string, error) {
 	return m.persistence.FindStatusMessageIDForBridgeMessageID(bridgeMessageID)
 }