add Messages.delete_all(..., older_than:)

lib/outboxer/messages.rb

@@ -173,7 +173,7 @@ def requeue_by_ids(ids:, time: Time)
       end
     end
 
-    def delete_all(status: nil, batch_size: 100)
+    def delete_all(status: nil, batch_size: 100, older_than: nil)
       deleted_count = 0
 
       loop do
@@ -183,6 +183,8 @@ def delete_all(status: nil, batch_size: 100)
           ActiveRecord::Base.transaction do
             query = Models::Message.all
             query = query.where(status: status) unless status.nil?
+            query = query.where('updated_at < ?', older_than) if older_than
+
             locked_ids = query.order(:updated_at)
               .limit(batch_size)
               .lock('FOR UPDATE SKIP LOCKED')

lib/outboxer/publisher.rb

@@ -34,6 +34,29 @@ def stop
       @publishing = false
     end
 
+    def create_delete_published_messages_thread(
+      cleanup_interval: 60, retention_period: 3600,
+      logger: Logger.new($stdout, level: Logger::INFO), kernel: Kernel, time: Time
+    )
+      Thread.new do
+        loop do
+          begin
+            Messages.delete_all(
+              status: Models::Message::Status::PUBLISHED,
+              batch_size: 100,
+              older_than: time.now.utc - retention_period)
+
+            kernel.sleep(cleanup_interval)
+          rescue StandardError => exception
+            logger.error "#{exception.class}: #{exception.message}"
+            exception.backtrace.each { |frame| logger.error frame }
+
+            kernel.sleep(cleanup_interval)
+          end
+        end
+      end
+    end
+
     def create_publisher_thread(queue:, logger:, &block)
       Thread.new do
         loop do

spec/lib/outboxer/messages/delete_all_spec.rb

@@ -16,14 +16,22 @@ module Outboxer
 
       let!(:message_5) { create(:outboxer_message, :publishing) }
 
+      let!(:message_6) { create(:outboxer_message, :published, updated_at: 3.hours.ago) }
+      let!(:exception_3) { create(:outboxer_exception, message: message_5) }
+      let!(:frame_3) { create(:outboxer_frame, exception: exception_3) }
+
+      let!(:message_7) { create(:outboxer_message, :published, updated_at: 1.hours.ago) }
+      let!(:exception_4) { create(:outboxer_exception, message: message_6) }
+      let!(:frame_4) { create(:outboxer_frame, exception: exception_4) }
+
       context 'when status is failed' do
         before do
           Messages.delete_all(status: Message::Status::FAILED, batch_size: 1)
         end
 
         it 'deletes failed messages' do
-          expect(Models::Message.pluck(:id))
-            .to match_array([message_1.id, message_2.id, message_5.id])
+          expect(Models::Message.pluck(:id)).to match_array([
+            message_1.id, message_2.id, message_5.id, message_6.id, message_7.id])
         end
       end
 
@@ -33,8 +41,8 @@ module Outboxer
         end
 
         it 'deletes queued messages' do
-          expect(Models::Message.all.pluck(:id))
-            .to match_array([message_1.id, message_3.id, message_4.id, message_5.id])
+          expect(Models::Message.all.pluck(:id)).to match_array([
+            message_1.id, message_3.id, message_4.id, message_5.id, message_6.id, message_7.id])
         end
       end
 
@@ -44,8 +52,42 @@ module Outboxer
         end
 
         it 'deletes publishing messages' do
-          expect(Models::Message.pluck(:id))
-            .to match_array([message_1.id, message_2.id, message_3.id, message_4.id])
+          expect(Models::Message.pluck(:id)).to match_array([
+            message_1.id, message_2.id, message_3.id, message_4.id, message_6.id, message_7.id])
+        end
+      end
+
+      context 'when status is published' do
+        context 'when older than nil' do
+          before do
+            Messages.delete_all(status: Message::Status::PUBLISHING, batch_size: 1)
+          end
+
+          it 'deletes publishing messages' do
+            expect(Models::Message.pluck(:id)).to match_array([
+              message_1.id, message_2.id, message_3.id, message_4.id, message_6.id, message_7.id])
+          end
+        end
+
+        context 'when older than provided' do
+          before do
+            Messages.delete_all(
+              status: Message::Status::PUBLISHED,
+              batch_size: 1,
+              older_than: 2.hours.ago)
+          end
+
+          it 'deletes the published messages older than the provided date' do
+            expect(Models::Message.pluck(:id)).to match_array([
+              message_1.id, message_2.id, message_3.id, message_4.id, message_5.id, message_7.id])
+          end
+
+          it 'deletes the correct exceptions and frames' do
+            expect(Models::Exception.pluck(:id)).to match_array([
+              exception_1.id, exception_2.id, exception_3.id])
+
+            expect(Models::Frame.pluck(:id)).to match_array([frame_1.id, frame_2.id, frame_3.id])
+          end
         end
       end