Create messages, search entries and unread state atomically

Apply the Laravel implementation’s MessageWriter architecture to Rails creation: save native Action Text and attachment metadata, insert the search entry and update unread memberships in the message transaction. Keep notification jobs after commit and preserve Room#receive and existing serialized jobs.

Cover native HTML/mentions, attachment-only search, derived-write failures, outer rollback and SQL commit ordering. Updates and destruction retain their current indexing behavior.

app/models/message.rb

@@ -9,7 +9,9 @@ class Message < ApplicationRecord
   has_rich_text :body
 
   before_create -> { self.client_message_id ||= Random.uuid } # Bots don't care
-  after_create_commit -> { room.receive(self) }
+  # Run after Action Text and Active Storage autosave, while creation is still atomic.
+  after_save :record_creation, if: :previously_new_record?
+  after_create_commit -> { room.push_later(self) }
 
   scope :ordered, -> { order(:created_at) }
   scope :with_creator, -> { preload(creator: :avatar_attachment) }
@@ -45,4 +47,10 @@ class Message < ApplicationRecord
       Sound.find_by_name match[:name]
     end
   end
+
+  private
+    def record_creation
+      create_in_index
+      room.unread_memberships(self)
+    end
 end

app/models/message/searchable.rb

@@ -2,7 +2,6 @@ module Message::Searchable
   extend ActiveSupport::Concern
 
   included do
-    after_create_commit  :create_in_index
     before_update        -> { @attachment_replaced = attachment_changes.key?("attachment") }
     after_update_commit  :update_in_index, if: :indexed_text_changed?
     after_destroy_commit :remove_from_index

app/models/room.rb

@@ -78,6 +78,19 @@ class Room < ApplicationRecord
     push_later(message)
   end
 
+  # Rewriting every member on every message is most of what posting to a large room writes,
+  # so members who are unread already stay as they are. Directs keep touching them all: a
+  # direct's sidebar row is cached by membership and shows the room's recency.
+  def unread_memberships(message)
+    recipients = memberships.visible.disconnected.where.not(user: message.creator)
+    recipients = recipients.where(unread_at: nil) unless direct?
+    recipients.update_all(unread_at: message.created_at, updated_at: Time.current)
+  end
+
+  def push_later(message)
+    Room::PushMessageJob.perform_later(self, message)
+  end
+
   def open?
     is_a?(Rooms::Open)
   end
@@ -103,17 +116,4 @@ class Room < ApplicationRecord
         errors.add :type, "can't be changed for a direct room"
       end
     end
-
-    # Rewriting every member on every message is most of what posting to a large room writes,
-    # so members who are unread already stay as they are. Directs keep touching them all: a
-    # direct's sidebar row is cached by membership and shows the room's recency.
-    def unread_memberships(message)
-      recipients = memberships.visible.disconnected.where.not(user: message.creator)
-      recipients = recipients.where(unread_at: nil) unless direct?
-      recipients.update_all(unread_at: message.created_at, updated_at: Time.current)
-    end
-
-    def push_later(message)
-      Room::PushMessageJob.perform_later(self, message)
-    end
 end

test/models/message/creation_test.rb

@@ -0,0 +1,132 @@
+require "test_helper"
+
+class Message::CreationTest < ActiveSupport::TestCase
+  include ActionDispatch::TestProcess
+
+  setup do
+    @room = rooms(:designers)
+    @creator = users(:david)
+  end
+
+  test "creation indexes persisted native HTML and mentions before notifying" do
+    html = "<p>Fish &amp; chips 🌍 #{mention_attachment_for(:jason)}</p>"
+    message = @room.messages.new(creator: @creator, body: html)
+    original = message.method(:create_in_index)
+    message.define_singleton_method(:create_in_index) do
+      raise "rich text not autosaved" unless ActionText::RichText.exists?(record: self)
+      original.call
+    end
+
+    message.save!
+
+    assert_equal message.body.to_plain_text, indexed_body(message)
+    assert_equal message.reload.body.to_plain_text, indexed_body(message)
+    assert_includes message.mentionees, users(:jason)
+    assert_equal [ message ], @room.messages.search("Fish chips")
+    assert_no_match(/<p>|&amp;/, indexed_body(message))
+  end
+
+  test "an attachment-only message is attached before its filename is indexed" do
+    message = @room.messages.new(creator: @creator, attachment: fixture_file_upload("moon.jpg", "image/jpeg"))
+    original = message.method(:create_in_index)
+    message.define_singleton_method(:create_in_index) do
+      raise "attachment not autosaved" unless ActiveStorage::Attachment.exists?(record: self, name: "attachment")
+      original.call
+    end
+
+    message.save!
+
+    assert_equal "moon.jpg", indexed_body(message)
+    assert_equal [ message ], @room.messages.search("moon")
+  end
+
+  test "a failed index insert rolls back the message and its associations without notifications" do
+    message = @room.messages.new(creator: @creator, body: "<p>Failed &amp; indexed</p>",
+      attachment: fixture_file_upload("moon.jpg", "image/jpeg"))
+    message.define_singleton_method(:create_in_index) { raise "index unavailable" }
+
+    assert_rolled_back_creation(message) do
+      assert_raises(RuntimeError) { message.save! }
+    end
+  end
+
+  test "a failed unread update rolls back the already inserted index and associations" do
+    message = @room.messages.new(creator: @creator, body: "<p>Unread &amp; rollback</p>",
+      attachment: fixture_file_upload("moon.jpg", "image/jpeg"))
+    original = @room.method(:unread_memberships)
+    @room.define_singleton_method(:unread_memberships) do |created|
+      original.call(created)
+      raise "unread unavailable"
+    end
+
+    assert_rolled_back_creation(message) do
+      assert_raises(RuntimeError) { message.save! }
+    end
+  end
+
+  test "an outer rollback restores the index unread state and counter and queues no jobs" do
+    message = @room.messages.new(creator: @creator, body: "<p>Outer &amp; rollback</p>")
+
+    assert_rolled_back_creation(message) do
+      Message.transaction(requires_new: true) do
+        message.save!
+        assert_equal "Outer & rollback", indexed_body(message)
+        raise ActiveRecord::Rollback
+      end
+    end
+  end
+
+  test "the existing public receive operation still marks unread and queues a push" do
+    message = messages(:first)
+    room = message.room
+    room.expects(:unread_memberships).with(message)
+    assert_enqueued_with(job: Room::PushMessageJob, args: [ room, message ]) { room.receive(message) }
+  end
+
+  private
+    def indexed_body(message)
+      Message.connection.select_value(Message.sanitize_sql([ "SELECT body FROM message_search_index WHERE rowid = ?", message.id ]))
+    end
+
+    def assert_rolled_back_creation(message)
+      models = [ Message, ActionText::RichText, ActiveStorage::Attachment, ActiveStorage::Blob ]
+      counts = models.map(&:count)
+      memberships = @room.memberships.order(:id).map(&:attributes)
+      room_state = @room.reload.attributes.slice("messages_count", "updated_at")
+      indexed = Message.connection.select_value("SELECT COUNT(*) FROM message_search_index")
+      clear_enqueued_jobs
+
+      yield
+
+      assert_equal counts, models.map(&:count)
+      assert_equal indexed, Message.connection.select_value("SELECT COUNT(*) FROM message_search_index")
+      assert_equal memberships, @room.memberships.order(:id).map(&:attributes)
+      assert_equal room_state, @room.reload.attributes.slice("messages_count", "updated_at")
+      assert_enqueued_jobs 0
+    end
+end
+
+class MessageCreationCommitTest < ActiveSupport::TestCase
+  self.use_transactional_tests = false
+
+  test "creation commits its index and unread writes once before enqueuing push" do
+    room = rooms(:designers)
+    sql = []
+    collect = ->(*, payload) { sql << payload[:sql] }
+    Room::PushMessageJob.expects(:perform_later).with do |queued_room, message|
+      assert_not Message.connection.transaction_open?
+      assert_equal room, queued_room
+      assert_equal "One native commit", Message.connection.select_value("SELECT body FROM message_search_index WHERE rowid = #{message.id}")
+      true
+    end
+
+    ActiveSupport::Notifications.subscribed(collect, "sql.active_record") do
+      room.messages.create!(creator: users(:david), body: "<p>One native commit</p>")
+    end
+
+    assert_equal 1, sql.count { |statement| statement.match?(/\ACOMMIT\b/i) }
+    commit = sql.index { |statement| statement.match?(/\ACOMMIT\b/i) }
+    assert_operator sql.index { |statement| statement.match?(/insert into message_search_index/i) }, :<, commit
+    assert_operator sql.index { |statement| statement.match?(/UPDATE "memberships"/i) }, :<, commit
+  end
+end