feat: include event timestamp in the payload

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