test: action cable broadcast job

app/javascript/dashboard/helper/actionCable.js

@@ -9,6 +9,7 @@ class ActionCableConnector extends BaseActionCableConnector {
     const { websocketURL = '' } = window.chatwootConfig || {};
     super(app, pubsubToken, websocketURL);
     this.CancelTyping = [];
+    this.conversationEventTimestamps = new Map();
     this.events = {
       'message.created': this.onMessageCreated,
       'message.updated': this.onMessageUpdated,
@@ -33,6 +34,23 @@ class ActionCableConnector extends BaseActionCableConnector {
     };
   }
 
+  /**
+   * Checks if a conversation event is out of sequence by comparing timestamps
+   * @param {number} conversationId - The ID of the conversation to check
+   * @param {number} eventTimestamp - The timestamp of the current event
+   * @returns {boolean} Returns true if event is out of sequence, false otherwise
+   */
+  eventOutOfSequence(conversationId, eventTimestamp) {
+    if (!eventTimestamp) return false;
+    const lastTimestamp = this.conversationEventTimestamps.get(conversationId);
+    if (!lastTimestamp || eventTimestamp > lastTimestamp) {
+      this.conversationEventTimestamps.set(conversationId, eventTimestamp);
+      return false;
+    }
+
+    return true;
+  }
+
   // eslint-disable-next-line class-methods-use-this
   onReconnect = () => {
     emitter.emit(BUS_EVENTS.WEBSOCKET_RECONNECT);
@@ -68,8 +86,10 @@ class ActionCableConnector extends BaseActionCableConnector {
     }
   };
 
-  onAssigneeChanged = payload => {
+  onAssigneeChanged = (payload, eventTimestamp) => {
     const { id } = payload;
+    if (this.eventOutOfSequence(id, eventTimestamp)) return;
+
     if (id) {
       this.app.$store.dispatch('updateConversation', payload);
     }
@@ -81,7 +101,8 @@ class ActionCableConnector extends BaseActionCableConnector {
     this.fetchConversationStats();
   };
 
-  onConversationRead = data => {
+  onConversationRead = (data, eventTimestamp) => {
+    if (this.eventOutOfSequence(data.id, eventTimestamp)) return;
     this.app.$store.dispatch('updateConversation', data);
   };
 
@@ -104,12 +125,14 @@ class ActionCableConnector extends BaseActionCableConnector {
   // eslint-disable-next-line class-methods-use-this
   onReload = () => window.location.reload();
 
-  onStatusChange = data => {
+  onStatusChange = (data, eventTimestamp) => {
+    if (this.eventOutOfSequence(data.id, eventTimestamp)) return;
     this.app.$store.dispatch('updateConversation', data);
     this.fetchConversationStats();
   };
 
-  onConversationUpdated = data => {
+  onConversationUpdated = (data, eventTimestamp) => {
+    if (this.eventOutOfSequence(data.id, eventTimestamp)) return;
     this.app.$store.dispatch('updateConversation', data);
     this.fetchConversationStats();
   };

app/javascript/shared/helpers/BaseActionCableConnector.js

@@ -79,10 +79,10 @@ class BaseActionCableConnector {
     this.consumer.disconnect();
   }
 
-  onReceived = ({ event, data } = {}) => {
+  onReceived = ({ event, data, event_timestamp } = {}) => {
     if (this.isAValidEvent(data)) {
       if (this.events[event] && typeof this.events[event] === 'function') {
-        this.events[event](data);
+        this.events[event](data, event_timestamp);
       }
     }
   };

app/jobs/action_cable_broadcast_job.rb

@@ -10,11 +10,11 @@ class ActionCableBroadcastJob < ApplicationJob
     CONVERSATION_STATUS_CHANGED
   ].freeze
 
-  def perform(members, event_name, data)
+  def perform(members, event_name, data, event_timestamp)
     return if members.blank?
 
     broadcast_data = prepare_broadcast_data(event_name, data)
-    broadcast_to_members(members, event_name, broadcast_data)
+    broadcast_to_members(members, event_name, broadcast_data, event_timestamp)
   end
 
   private
@@ -30,13 +30,17 @@ class ActionCableBroadcastJob < ApplicationJob
     conversation.push_event_data.merge(account_id: data[:account_id])
   end
 
-  def broadcast_to_members(members, event_name, broadcast_data)
+  # A timestamp is added to the broadcast data, this timestamp is
+  # generated in the main app thread when the job is enqueued.
+  # This is used as a "sequence number" to ensure we ignore out of order events
+  def broadcast_to_members(members, event_name, broadcast_data, event_timestamp)
     members.each do |member|
       ActionCable.server.broadcast(
         member,
         {
           event: event_name,
-          data: broadcast_data
+          data: broadcast_data,
+          event_timestamp: event_timestamp
         }
       )
     end

app/listeners/action_cable_listener.rb

@@ -58,7 +58,8 @@ class ActionCableListener < BaseListener
     conversation, account = extract_conversation_and_account(event)
     tokens = user_tokens(account, conversation.inbox.members) + contact_inbox_tokens(conversation.contact_inbox)
 
-    broadcast(account, tokens, CONVERSATION_CREATED, conversation.push_event_data)
+    # Include the original event timestamp to ensure correct ordering on frontend
+    broadcast(account, tokens, CONVERSATION_CREATED, conversation.push_event_data, event.timestamp.to_f)
   end
 
   def conversation_read(event)
@@ -72,14 +73,16 @@ class ActionCableListener < BaseListener
     conversation, account = extract_conversation_and_account(event)
     tokens = user_tokens(account, conversation.inbox.members) + contact_inbox_tokens(conversation.contact_inbox)
 
-    broadcast(account, tokens, CONVERSATION_STATUS_CHANGED, conversation.push_event_data)
+    # Include the original event timestamp to ensure correct ordering on frontend
+    broadcast(account, tokens, CONVERSATION_STATUS_CHANGED, conversation.push_event_data, event.timestamp.to_f)
   end
 
   def conversation_updated(event)
     conversation, account = extract_conversation_and_account(event)
     tokens = user_tokens(account, conversation.inbox.members) + contact_inbox_tokens(conversation.contact_inbox)
 
-    broadcast(account, tokens, CONVERSATION_UPDATED, conversation.push_event_data)
+    # Include the original event timestamp to ensure correct ordering on frontend
+    broadcast(account, tokens, CONVERSATION_UPDATED, conversation.push_event_data, event.timestamp.to_f)
   end
 
   def conversation_typing_on(event)
@@ -118,14 +121,16 @@ class ActionCableListener < BaseListener
     conversation, account = extract_conversation_and_account(event)
     tokens = user_tokens(account, conversation.inbox.members)
 
-    broadcast(account, tokens, ASSIGNEE_CHANGED, conversation.push_event_data)
+    # Include the original event timestamp to ensure correct ordering on frontend
+    broadcast(account, tokens, ASSIGNEE_CHANGED, conversation.push_event_data, event.timestamp.to_f)
   end
 
   def team_changed(event)
     conversation, account = extract_conversation_and_account(event)
     tokens = user_tokens(account, conversation.inbox.members)
 
-    broadcast(account, tokens, TEAM_CHANGED, conversation.push_event_data)
+    # Include the original event timestamp to ensure correct ordering on frontend
+    broadcast(account, tokens, TEAM_CHANGED, conversation.push_event_data, event.timestamp.to_f)
   end
 
   def conversation_contact_changed(event)
@@ -197,7 +202,7 @@ class ActionCableListener < BaseListener
     contact_inbox.hmac_verified? ? contact.contact_inboxes.where(hmac_verified: true).filter_map(&:pubsub_token) : [contact_inbox.pubsub_token]
   end
 
-  def broadcast(account, tokens, event_name, data)
+  def broadcast(account, tokens, event_name, data, event_timestamp = nil)
     return if tokens.blank?
 
     payload = data.merge(account_id: account.id)
@@ -205,6 +210,6 @@ class ActionCableListener < BaseListener
     # Useful in cases like conversation assignment for generating a notification with assigner name.
     payload[:performer] = Current.user&.push_event_data if Current.user.present?
 
-    ::ActionCableBroadcastJob.perform_later(tokens.uniq, event_name, payload)
+    ::ActionCableBroadcastJob.perform_later(tokens.uniq, event_name, payload, event_timestamp)
   end
 end

spec/jobs/action_cable_broadcast_job_spec.rb

@@ -0,0 +1,105 @@
+require 'rails_helper'
+
+RSpec.describe ActionCableBroadcastJob do
+  include Events::Types
+
+  let(:account) { create(:account) }
+  let(:conversation) { create(:conversation, account: account) }
+  let(:members) { %w[test_member_1 test_member_2] }
+  let(:event_name) { 'test.event' }
+  let(:data) { { account_id: account.id, id: conversation.display_id } }
+  let(:timestamp) { Time.zone.now.to_f }
+
+  describe '#perform' do
+    context 'when members are blank' do
+      it 'returns without broadcasting' do
+        expect(ActionCable.server).not_to receive(:broadcast)
+        described_class.perform_now([], event_name, data, timestamp)
+      end
+    end
+
+    context 'when event is not a conversation update event' do
+      it 'broadcasts original data to all members' do
+        members.each do |member|
+          expect(ActionCable.server).to receive(:broadcast).with(
+            member,
+            {
+              event: event_name,
+              data: data,
+              event_timestamp: timestamp
+            }
+          )
+        end
+
+        described_class.perform_now(members, event_name, data, timestamp)
+      end
+    end
+
+    context 'when event is a conversation update event' do
+      let(:conversation_data) { conversation.push_event_data.merge(account_id: account.id) }
+      let(:conversation_update_events) do
+        [
+          Events::Types::CONVERSATION_READ,
+          Events::Types::CONVERSATION_UPDATED,
+          Events::Types::TEAM_CHANGED,
+          Events::Types::ASSIGNEE_CHANGED,
+          Events::Types::CONVERSATION_STATUS_CHANGED
+        ]
+      end
+
+      it 'refreshes and broadcasts latest conversation data for each update event' do
+        conversation_update_events.each do |update_event|
+          members.each do |member|
+            expect(ActionCable.server).to receive(:broadcast).with(
+              member,
+              {
+                event: update_event,
+                data: conversation_data,
+                event_timestamp: timestamp
+              }
+            )
+          end
+
+          described_class.perform_now(members, update_event, data, timestamp)
+        end
+      end
+
+      context 'when conversation is not found' do
+        let(:data) { { account_id: account.id, id: 0 } }
+
+        it 'raises ActiveRecord::RecordNotFound' do
+          expect do
+            described_class.perform_now(
+              members,
+              Events::Types::CONVERSATION_UPDATED,
+              data,
+              timestamp
+            )
+          end.to raise_error(ActiveRecord::RecordNotFound)
+        end
+      end
+    end
+  end
+
+  describe 'queue' do
+    it 'uses critical queue' do
+      expect(described_class.queue_name).to eq('critical')
+    end
+  end
+
+  describe 'timestamp handling' do
+    it 'includes event_timestamp in broadcast data' do
+      expect(ActionCable.server).to receive(:broadcast).with(
+        members.first,
+        hash_including(event_timestamp: timestamp)
+      ).once
+
+      expect(ActionCable.server).to receive(:broadcast).with(
+        members.last,
+        hash_including(event_timestamp: timestamp)
+      ).once
+
+      described_class.perform_now(members, event_name, data, timestamp)
+    end
+  end
+end

spec/listeners/action_cable_listener_spec.rb

@@ -30,7 +30,8 @@ describe ActionCableListener do
           agent.pubsub_token, admin.pubsub_token, conversation.contact_inbox.pubsub_token
         ),
         'message.created',
-        message.push_event_data.merge(account_id: account.id)
+        message.push_event_data.merge(account_id: account.id),
+        nil
       )
       listener.message_created(event)
     end
@@ -48,7 +49,8 @@ describe ActionCableListener do
           agent.pubsub_token, admin.pubsub_token, conversation.contact_inbox.pubsub_token, verified_contact_inbox.pubsub_token
         ),
         'message.created',
-        message.push_event_data.merge(account_id: account.id)
+        message.push_event_data.merge(account_id: account.id),
+        nil
       )
       listener.message_created(event)
     end
@@ -68,7 +70,8 @@ describe ActionCableListener do
         'conversation.typing_on', { conversation: conversation.push_event_data,
                                     user: agent.push_event_data,
                                     account_id: account.id,
-                                    is_private: false }
+                                    is_private: false },
+        nil
       )
       listener.conversation_typing_on(event)
     end
@@ -88,7 +91,8 @@ describe ActionCableListener do
         'conversation.typing_on', { conversation: conversation.push_event_data,
                                     user: conversation.contact.push_event_data,
                                     account_id: account.id,
-                                    is_private: false }
+                                    is_private: false },
+        nil
       )
       listener.conversation_typing_on(event)
     end
@@ -108,7 +112,8 @@ describe ActionCableListener do
         'conversation.typing_off', { conversation: conversation.push_event_data,
                                      user: agent.push_event_data,
                                      account_id: account.id,
-                                     is_private: false }
+                                     is_private: false },
+        nil
       )
       listener.conversation_typing_off(event)
     end
@@ -125,7 +130,8 @@ describe ActionCableListener do
           agent.pubsub_token, admin.pubsub_token
         ),
         'contact.deleted',
-        contact.push_event_data.merge(account_id: account.id)
+        contact.push_event_data.merge(account_id: account.id),
+        nil
       )
       listener.contact_deleted(event)
     end
@@ -147,7 +153,8 @@ describe ActionCableListener do
           },
           unread_count: 1,
           count: 1
-        }
+        },
+        nil
       )
 
       listener.notification_deleted(event)
@@ -168,7 +175,8 @@ describe ActionCableListener do
           notification: notification.push_event_data,
           unread_count: 1,
           count: 1
-        }
+        },
+        nil
       )
 
       listener.notification_updated(event)
@@ -177,7 +185,8 @@ describe ActionCableListener do
 
   describe '#conversation_updated' do
     let(:event_name) { :'conversation.updated' }
-    let!(:event) { Events::Base.new(event_name, Time.zone.now, conversation: conversation, user: agent, is_private: false) }
+    let(:timestamp) { Time.zone.now }
+    let!(:event) { Events::Base.new(event_name, timestamp, conversation: conversation, user: agent, is_private: false) }
 
     before do
       conversation.add_labels(['support'])
@@ -189,7 +198,8 @@ describe ActionCableListener do
       expect(ActionCableBroadcastJob).to receive(:perform_later).with(
         [agent.pubsub_token, admin.pubsub_token, conversation.contact_inbox.pubsub_token],
         'conversation.updated',
-        conversation.push_event_data.merge(account_id: account.id)
+        conversation.push_event_data.merge(account_id: account.id),
+        timestamp.to_f
       )
       listener.conversation_updated(event)
     end
@@ -200,9 +210,96 @@ describe ActionCableListener do
       expect(ActionCableBroadcastJob).to receive(:perform_later).with(
         [agent.pubsub_token, admin.pubsub_token, conversation.contact_inbox.pubsub_token],
         'conversation.updated',
-        conversation.push_event_data.merge(account_id: account.id)
+        conversation.push_event_data.merge(account_id: account.id),
+        timestamp.to_f
       )
       listener.conversation_updated(event)
     end
   end
+
+  context 'with event timestamps' do
+    let(:timestamp) { Time.zone.now }
+
+    describe '#conversation_created' do
+      let(:event_name) { :'conversation.created' }
+      let!(:event) { Events::Base.new(event_name, timestamp, conversation: conversation) }
+
+      it 'sends event with timestamp' do
+        expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+          [agent.pubsub_token, admin.pubsub_token, conversation.contact_inbox.pubsub_token],
+          'conversation.created',
+          conversation.push_event_data.merge(account_id: account.id),
+          timestamp.to_f
+        )
+        listener.conversation_created(event)
+      end
+    end
+
+    describe '#conversation_status_changed' do
+      let(:event_name) { :'conversation.status_changed' }
+      let!(:event) { Events::Base.new(event_name, timestamp, conversation: conversation) }
+
+      it 'sends event with timestamp' do
+        expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+          [agent.pubsub_token, admin.pubsub_token, conversation.contact_inbox.pubsub_token],
+          'conversation.status_changed',
+          conversation.push_event_data.merge(account_id: account.id),
+          timestamp.to_f
+        )
+        listener.conversation_status_changed(event)
+      end
+    end
+
+    describe '#assignee_changed' do
+      let(:event_name) { :'conversation.assignee_changed' }
+      let!(:event) { Events::Base.new(event_name, timestamp, conversation: conversation) }
+
+      it 'sends event with timestamp' do
+        expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+          [agent.pubsub_token, admin.pubsub_token],
+          'assignee.changed',
+          conversation.push_event_data.merge(account_id: account.id),
+          timestamp.to_f
+        )
+        listener.assignee_changed(event)
+      end
+    end
+
+    describe '#team_changed' do
+      let(:event_name) { :'conversation.team_changed' }
+      let!(:event) { Events::Base.new(event_name, timestamp, conversation: conversation) }
+
+      it 'sends event with timestamp' do
+        expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+          [agent.pubsub_token, admin.pubsub_token],
+          'team.changed',
+          conversation.push_event_data.merge(account_id: account.id),
+          timestamp.to_f
+        )
+        listener.team_changed(event)
+      end
+    end
+
+    # Update the existing conversation_updated test to include timestamp verification
+    describe '#conversation_updated' do
+      let(:event_name) { :'conversation.updated' }
+      let!(:event) { Events::Base.new(event_name, timestamp, conversation: conversation, user: agent, is_private: false) }
+
+      before do
+        conversation.add_labels(['support'])
+      end
+
+      it 'sends update to inbox members with timestamp' do
+        expect(conversation.inbox.reload.inbox_members.count).to eq(1)
+
+        expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+          [agent.pubsub_token, admin.pubsub_token, conversation.contact_inbox.pubsub_token],
+          'conversation.updated',
+          conversation.push_event_data.merge(account_id: account.id),
+          timestamp.to_f
+        )
+        listener.conversation_updated(event)
+      end
+    end
+  end
 end