Update logic

app/channels/room_channel.rb

@@ -4,7 +4,7 @@ class RoomChannel < ApplicationCable::Channel
     # for now going ahead with guard clauses in update_subscription and broadcast_presence
     current_user
     current_account
-    ensure_stream
+    prepare_stream
     update_subscription
     broadcast_presence
   end
@@ -24,9 +24,18 @@ class RoomChannel < ApplicationCable::Channel
     ActionCable.server.broadcast(pubsub_token, { event: 'presence.update', data: data })
   end
 
-  def ensure_stream
+  def prepare_stream
     stream_from pubsub_token
     stream_from "account_#{@current_account.id}" if @current_account.present? && @current_user.is_a?(User)
+    stream_from_inboxes
+  end
+
+  def stream_from_inboxes
+    return if current_user.blank?
+
+    current_user.assigned_inboxes.each do |inbox|
+      stream_from "inbox_#{inbox.id}"
+    end
   end
 
   def update_subscription

app/listeners/action_cable_listener.rb

@@ -23,136 +23,76 @@ class ActionCableListener < BaseListener
 
   def account_cache_invalidated(event)
     account = event.data[:account]
-    tokens = user_tokens(account, account.agents)
-
-    broadcast(account, tokens, ACCOUNT_CACHE_INVALIDATED, {
-                cache_keys: event.data[:cache_keys]
-              })
+    broadcast(account, account_stream(account), ACCOUNT_CACHE_INVALIDATED, { cache_keys: event.data[:cache_keys] })
   end
 
   def message_created(event)
-    message, account = extract_message_and_account(event)
-    conversation = message.conversation
-    tokens = user_tokens(account, conversation.inbox.members) + contact_tokens(conversation.contact_inbox, message)
-
-    broadcast(account, tokens, MESSAGE_CREATED, message.push_event_data)
+    broadcast_message_event(MESSAGE_CREATED, event)
+    broadcast_message_event_to_contact(MESSAGE_CREATED, event)
   end
 
   def message_updated(event)
-    message, account = extract_message_and_account(event)
-    conversation = message.conversation
-    tokens = user_tokens(account, conversation.inbox.members) + contact_tokens(conversation.contact_inbox, message)
-
-    broadcast(account, tokens, MESSAGE_UPDATED, message.push_event_data.merge(previous_changes: event.data[:previous_changes]))
+    broadcast_message_event(MESSAGE_UPDATED, event)
+    broadcast_message_event_to_contact(MESSAGE_UPDATED, event)
   end
 
   def first_reply_created(event)
-    message, account = extract_message_and_account(event)
-    conversation = message.conversation
-    tokens = user_tokens(account, conversation.inbox.members)
-
-    broadcast(account, tokens, FIRST_REPLY_CREATED, message.push_event_data)
+    broadcast_message_event(FIRST_REPLY_CREATED, event)
   end
 
   def conversation_created(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_CREATED, conversation.push_event_data)
+    broadcast_conversation_event_to_contact(CONVERSATION_CREATED, event)
+    broadcast_conversation_event(CONVERSATION_CREATED, event)
   end
 
   def conversation_read(event)
-    conversation, account = extract_conversation_and_account(event)
-    tokens = user_tokens(account, conversation.inbox.members)
-
-    broadcast(account, tokens, CONVERSATION_READ, conversation.push_event_data)
+    broadcast_conversation_event(CONVERSATION_READ, event)
   end
 
   def conversation_status_changed(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_STATUS_CHANGED, conversation.push_event_data)
+    broadcast_conversation_event_to_contact(CONVERSATION_STATUS_CHANGED, event)
+    broadcast_conversation_event(CONVERSATION_STATUS_CHANGED, event)
   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)
+    broadcast_conversation_event_to_contact(CONVERSATION_UPDATED, event)
+    broadcast_conversation_event(CONVERSATION_UPDATED, event)
   end
 
   def conversation_typing_on(event)
-    conversation = event.data[:conversation]
-    account = conversation.account
-    user = event.data[:user]
-    tokens = typing_event_listener_tokens(account, conversation, user)
-
-    broadcast(
-      account,
-      tokens,
-      CONVERSATION_TYPING_ON,
-      conversation: conversation.push_event_data,
-      user: user.push_event_data,
-      is_private: event.data[:is_private] || false
-    )
+    broadcast_typing_event(CONVERSATION_TYPING_ON, event.data)
   end
 
   def conversation_typing_off(event)
-    conversation = event.data[:conversation]
-    account = conversation.account
-    user = event.data[:user]
-    tokens = typing_event_listener_tokens(account, conversation, user)
-
-    broadcast(
-      account,
-      tokens,
-      CONVERSATION_TYPING_OFF,
-      conversation: conversation.push_event_data,
-      user: user.push_event_data,
-      is_private: event.data[:is_private] || false
-    )
+    broadcast_typing_event(CONVERSATION_TYPING_OFF, event.data)
   end
 
   def assignee_changed(event)
-    conversation, account = extract_conversation_and_account(event)
-    tokens = user_tokens(account, conversation.inbox.members)
-
-    broadcast(account, tokens, ASSIGNEE_CHANGED, conversation.push_event_data)
+    broadcast_conversation_event(ASSIGNEE_CHANGED, event)
   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)
+    broadcast_conversation_event(TEAM_CHANGED, event)
   end
 
   def conversation_contact_changed(event)
-    conversation, account = extract_conversation_and_account(event)
-    tokens = user_tokens(account, conversation.inbox.members)
-
-    broadcast(account, tokens, CONVERSATION_CONTACT_CHANGED, conversation.push_event_data)
+    broadcast_conversation_event(CONVERSATION_CONTACT_CHANGED, event)
   end
 
   def contact_created(event)
-    contact, account = extract_contact_and_account(event)
-    broadcast(account, [account_token(account)], CONTACT_CREATED, contact.push_event_data)
+    broadcast_contact_event(CONTACT_CREATED, event)
   end
 
   def contact_updated(event)
-    contact, account = extract_contact_and_account(event)
-    broadcast(account, [account_token(account)], CONTACT_UPDATED, contact.push_event_data)
+    broadcast_contact_event(CONTACT_UPDATED, event)
   end
 
   def contact_merged(event)
-    contact, account = extract_contact_and_account(event)
-    broadcast(account, [account_token(account)], CONTACT_MERGED, contact.push_event_data)
+    broadcast_contact_event(CONTACT_MERGED, event)
   end
 
   def contact_deleted(event)
-    contact, account = extract_contact_and_account(event)
-    broadcast(account, [account_token(account)], CONTACT_DELETED, contact.push_event_data)
+    broadcast_contact_event(CONTACT_DELETED, event)
   end
 
   def conversation_mentioned(event)
@@ -164,19 +104,59 @@ class ActionCableListener < BaseListener
 
   private
 
-  def account_token(account)
-    "account_#{account.id}"
+  def account_stream(account)
+    ["account_#{account.id}"]
   end
 
-  def typing_event_listener_tokens(account, conversation, user)
-    current_user_token = user.is_a?(Contact) ? conversation.contact_inbox.pubsub_token : user.pubsub_token
-    (user_tokens(account, conversation.inbox.members) + [conversation.contact_inbox.pubsub_token]) - [current_user_token]
+  def inbox_stream(inbox)
+    ["inbox_#{inbox.id}"]
   end
 
-  def user_tokens(account, agents)
-    agent_tokens = agents.pluck(:pubsub_token)
-    admin_tokens = account.administrators.pluck(:pubsub_token)
-    (agent_tokens + admin_tokens).uniq
+  def broadcast_contact_event(event_name, event_data)
+    contact, account = extract_contact_and_account(event_data)
+    broadcast(account, account_stream(account), event_name, contact.push_event_data)
+  end
+
+  def broadcast_conversation_event(event_name, event_data)
+    conversation, account = extract_conversation_and_account(event_data)
+    broadcast(account, inbox_stream(conversation.inbox), event_name, conversation.push_event_data)
+  end
+
+  def broadcast_message_event(event_name, event_data)
+    message, account = extract_message_and_account(event_data)
+    broadcast(account, inbox_stream(message.inbox), event_name, message.push_event_data)
+  end
+
+  def broadcast_conversation_event_to_contact(event_name, event_data)
+    conversation, account = extract_conversation_and_account(event_data)
+    tokens = contact_inbox_tokens(conversation.contact_inbox)
+    broadcast(account, tokens, event_name, conversation.push_event_data)
+  end
+
+  def broadcast_message_event_to_contact(event_name, event_data)
+    message, account = extract_message_and_account(event_data)
+    tokens = contact_tokens(message.conversation.contact_inbox, message)
+    broadcast(account, tokens, event_name, message.push_event_data)
+  end
+
+  def broadcast_typing_event(event_name, event_data)
+    conversation = event_data[:conversation]
+    account = conversation.account
+    user = event_data[:user]
+    tokens = typing_event_listener_tokens(conversation)
+
+    broadcast(
+      account,
+      tokens,
+      event_name,
+      conversation: conversation.push_event_data,
+      user: user.push_event_data,
+      is_private: event_data[:is_private] || false
+    )
+  end
+
+  def typing_event_listener_tokens(conversation)
+    ["inbox_#{conversation.inbox.id}"] + [conversation.contact_inbox.pubsub_token]
   end
 
   def contact_tokens(contact_inbox, message)

spec/listeners/action_cable_listener_spec.rb

@@ -2,7 +2,7 @@ require 'rails_helper'
 describe ActionCableListener do
   let(:listener) { described_class.instance }
   let!(:account) { create(:account) }
-  let!(:admin) { create(:user, account: account, role: :administrator) }
+  let(:admin) { create(:user, account: account, role: :administrator) }
   let!(:inbox) { create(:inbox, account: account) }
   let!(:agent) { create(:user, account: account, role: :agent) }
   let!(:conversation) { create(:conversation, account: account, inbox: inbox, assignee: agent) }
@@ -21,32 +21,33 @@ describe ActionCableListener do
     end
     let!(:event) { Events::Base.new(event_name, Time.zone.now, message: message) }
 
-    it 'sends message to account admins, inbox agents and the contact' do
-      # HACK: to reload conversation inbox members
-      expect(conversation.inbox.reload.inbox_members.count).to eq(1)
-
+    it 'sends message to inbox stream' do
       expect(ActionCableBroadcastJob).to receive(:perform_later).with(
-        a_collection_containing_exactly(
-          agent.pubsub_token, admin.pubsub_token, conversation.contact_inbox.pubsub_token
-        ),
+        ["inbox_#{message.inbox.id}"],
+        'message.created',
+        message.push_event_data.merge(account_id: account.id)
+      )
+      expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+        [conversation.contact_inbox.pubsub_token],
         'message.created',
         message.push_event_data.merge(account_id: account.id)
       )
       listener.message_created(event)
     end
 
-    it 'sends message to all hmac verified contact inboxes' do
-      # HACK: to reload conversation inbox members
-      expect(conversation.inbox.reload.inbox_members.count).to eq(1)
+    it 'sends message to all hmac verified contact inboxes and inbox stream' do
       conversation.contact_inbox.update(hmac_verified: true)
       # creating a non verified contact inbox to ensure the events are not sent to it
       create(:contact_inbox, contact: conversation.contact, inbox: inbox)
       verified_contact_inbox = create(:contact_inbox, contact: conversation.contact, inbox: inbox, hmac_verified: true)
 
       expect(ActionCableBroadcastJob).to receive(:perform_later).with(
-        a_collection_containing_exactly(
-          agent.pubsub_token, admin.pubsub_token, conversation.contact_inbox.pubsub_token, verified_contact_inbox.pubsub_token
-        ),
+        [conversation.contact_inbox.pubsub_token, verified_contact_inbox.pubsub_token],
+        'message.created',
+        message.push_event_data.merge(account_id: account.id)
+      )
+      expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+        ["inbox_#{inbox.id}"],
         'message.created',
         message.push_event_data.merge(account_id: account.id)
       )
@@ -58,13 +59,9 @@ describe ActionCableListener do
     let(:event_name) { :'conversation.typing_on' }
     let!(:event) { Events::Base.new(event_name, Time.zone.now, conversation: conversation, user: agent, is_private: false) }
 
-    it 'sends message to account admins, inbox agents and the contact' do
-      # HACK: to reload conversation inbox members
-      expect(conversation.inbox.reload.inbox_members.count).to eq(1)
+    it 'sends message to inbox stream and contact' do
       expect(ActionCableBroadcastJob).to receive(:perform_later).with(
-        a_collection_containing_exactly(
-          admin.pubsub_token, conversation.contact_inbox.pubsub_token
-        ),
+        ["inbox_#{conversation.inbox.id}", conversation.contact_inbox.pubsub_token],
         'conversation.typing_on', { conversation: conversation.push_event_data,
                                     user: agent.push_event_data,
                                     account_id: account.id,
@@ -78,13 +75,11 @@ describe ActionCableListener do
     let(:event_name) { :'conversation.typing_on' }
     let!(:event) { Events::Base.new(event_name, Time.zone.now, conversation: conversation, user: conversation.contact, is_private: false) }
 
-    it 'sends message to account admins, inbox agents and the contact' do
-      # HACK: to reload conversation inbox members
-      expect(conversation.inbox.reload.inbox_members.count).to eq(1)
+    it 'sends message to inbox stream and contact' do
       expect(ActionCableBroadcastJob).to receive(:perform_later).with(
-        a_collection_containing_exactly(
-          admin.pubsub_token, agent.pubsub_token
-        ),
+        [
+          "inbox_#{conversation.inbox.id}", conversation.contact_inbox.pubsub_token
+        ],
         'conversation.typing_on', { conversation: conversation.push_event_data,
                                     user: conversation.contact.push_event_data,
                                     account_id: account.id,
@@ -98,13 +93,9 @@ describe ActionCableListener do
     let(:event_name) { :'conversation.typing_off' }
     let!(:event) { Events::Base.new(event_name, Time.zone.now, conversation: conversation, user: agent, is_private: false) }
 
-    it 'sends message to account admins, inbox agents and the contact' do
-      # HACK: to reload conversation inbox members
-      expect(conversation.inbox.reload.inbox_members.count).to eq(1)
+    it 'sends message to inbox stream and contact' do
       expect(ActionCableBroadcastJob).to receive(:perform_later).with(
-        a_collection_containing_exactly(
-          admin.pubsub_token, conversation.contact_inbox.pubsub_token
-        ),
+        ["inbox_#{conversation.inbox.id}", conversation.contact_inbox.pubsub_token],
         'conversation.typing_off', { conversation: conversation.push_event_data,
                                      user: agent.push_event_data,
                                      account_id: account.id,
@@ -119,7 +110,7 @@ describe ActionCableListener do
     let!(:contact) { create(:contact, account: account) }
     let!(:event) { Events::Base.new(event_name, Time.zone.now, contact: contact) }
 
-    it 'sends message to account admins, inbox agents' do
+    it 'sends message to account stream' do
       expect(ActionCableBroadcastJob).to receive(:perform_later).with(
         ["account_#{account.id}"],
         'contact.deleted',
@@ -134,7 +125,7 @@ describe ActionCableListener do
     let!(:notification) { create(:notification, account: account, user: agent) }
     let!(:event) { Events::Base.new(event_name, Time.zone.now, notification: notification) }
 
-    it 'sends message to account admins, inbox agents' do
+    it 'sends message to agent pubsub stream' do
       expect(ActionCableBroadcastJob).to receive(:perform_later).with(
         [agent.pubsub_token],
         'notification.deleted',
@@ -157,7 +148,7 @@ describe ActionCableListener do
     let!(:notification) { create(:notification, account: account, user: agent) }
     let!(:event) { Events::Base.new(event_name, Time.zone.now, notification: notification) }
 
-    it 'sends notification to account admins, inbox agents' do
+    it 'sends notification to agent pubsub stream' do
       expect(ActionCableBroadcastJob).to receive(:perform_later).with(
         [agent.pubsub_token],
         'notification.updated',
@@ -181,11 +172,14 @@ describe ActionCableListener do
       conversation.add_labels(['support'])
     end
 
-    it 'sends update to inbox members' do
-      expect(conversation.inbox.reload.inbox_members.count).to eq(1)
-
+    it 'sends update to contact and inbox' do
       expect(ActionCableBroadcastJob).to receive(:perform_later).with(
-        [agent.pubsub_token, admin.pubsub_token, conversation.contact_inbox.pubsub_token],
+        [conversation.contact_inbox.pubsub_token],
+        'conversation.updated',
+        conversation.push_event_data.merge(account_id: account.id)
+      )
+      expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+        ["inbox_#{inbox.id}"],
         'conversation.updated',
         conversation.push_event_data.merge(account_id: account.id)
       )
@@ -196,7 +190,12 @@ describe ActionCableListener do
       expect(conversation.reload.push_event_data[:labels]).to eq(conversation.labels.pluck(:name))
 
       expect(ActionCableBroadcastJob).to receive(:perform_later).with(
-        [agent.pubsub_token, admin.pubsub_token, conversation.contact_inbox.pubsub_token],
+        ["inbox_#{inbox.id}"],
+        'conversation.updated',
+        conversation.push_event_data.merge(account_id: account.id)
+      )
+      expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+        [conversation.contact_inbox.pubsub_token],
         'conversation.updated',
         conversation.push_event_data.merge(account_id: account.id)
       )