chore: Migrate PubSub Token to contact inbox (#3434)

At present, the websocket pubsub tokens are present at the contact objects in chatwoot. A better approach would be to have these tokens at the contact_inbox object instead. This helps chatwoot to deliver the websocket events targetted to the specific widget connection, stop contact events from leaking into other chat sessions from the same contact.

Fixes #1682
Fixes #1664

Co-authored-by: Pranav Raj S <pranav@chatwoot.com>
Co-authored-by: Muhsin Keloth <muhsinkeramam@gmail.com>

app/actions/contact_merge_action.rb

@@ -48,11 +48,10 @@ class ContactMergeAction
 
     # attributes in base contact are given preference
     merged_attributes = mergee_contact_attributes.deep_merge(base_contact_attributes)
-    # retaining old pubsub token to notify the contacts that are listening
-    mergee_pubsub_token = mergee_contact.pubsub_token
 
     @mergee_contact.destroy!
-    Rails.configuration.dispatcher.dispatch(CONTACT_MERGED, Time.zone.now, contact: @base_contact, tokens: [mergee_pubsub_token])
+    Rails.configuration.dispatcher.dispatch(CONTACT_MERGED, Time.zone.now, contact: @base_contact,
+                                                                           tokens: [@base_contact.contact_inboxes.filter_map(&:pubsub_token)])
     @base_contact.update!(merged_attributes)
   end
 end

app/channels/room_channel.rb

@@ -31,7 +31,7 @@ class RoomChannel < ApplicationCable::Channel
 
   def current_user
     @current_user ||= if params[:user_id].blank?
-                        Contact.find_by!(pubsub_token: @pubsub_token)
+                        ContactInbox.find_by!(pubsub_token: @pubsub_token).contact
                       else
                         User.find_by!(pubsub_token: @pubsub_token, id: params[:user_id])
                       end

app/controllers/widgets_controller.rb

@@ -29,21 +29,21 @@ class WidgetsController < ActionController::Base
   def set_contact
     return if @auth_token_params[:source_id].nil?
 
-    contact_inbox = ::ContactInbox.find_by(
+    @contact_inbox = ::ContactInbox.find_by(
       inbox_id: @web_widget.inbox.id,
       source_id: @auth_token_params[:source_id]
     )
 
-    @contact = contact_inbox ? contact_inbox.contact : nil
+    @contact = @contact_inbox ? @contact_inbox.contact : nil
   end
 
   def build_contact
     return if @contact.present?
 
-    contact_inbox = @web_widget.create_contact_inbox(additional_attributes)
+    @contact_inbox = @web_widget.create_contact_inbox(additional_attributes)
-    @contact = contact_inbox.contact
+    @contact = @contact_inbox.contact
 
-    payload = { source_id: contact_inbox.source_id, inbox_id: @web_widget.inbox.id }
+    payload = { source_id: @contact_inbox.source_id, inbox_id: @web_widget.inbox.id }
     @token = ::Widget::TokenService.new(payload: payload).generate_token
   end
 

app/finders/conversation_finder.rb

@@ -121,7 +121,7 @@ class ConversationFinder
 
   def conversations
     @conversations = @conversations.includes(
-      :taggings, :inbox, { assignee: { avatar_attachment: [:blob] } }, { contact: { avatar_attachment: [:blob] } }, :team
+      :taggings, :inbox, { assignee: { avatar_attachment: [:blob] } }, { contact: { avatar_attachment: [:blob] } }, :team, :contact_inbox
     )
     @conversations.latest.page(current_page)
   end

app/javascript/dashboard/assets/scss/_utility-helpers.scss

@@ -2,6 +2,10 @@
   margin-right: var(--space-small);
 }
 
+.margin-right-smaller {
+  margin-right: var(--space-smaller);
+}
+
 .fs-small {
   font-size: var(--font-size-small);
 }
@@ -42,3 +46,7 @@
 .bg-white {
   background-color: var(--white);
 }
+
+.text-y-800 {
+  color: var(--y-800);
+}

app/javascript/dashboard/components/widgets/conversation/ConversationHeader.vue

@@ -10,7 +10,12 @@
       />
       <div class="user--profile__meta">
         <h3 class="user--name text-truncate">
-          {{ currentContact.name }}
+          <span class="margin-right-smaller">{{ currentContact.name }}</span>
+          <i
+            v-if="!isHMACVerified"
+            v-tooltip="$t('CONVERSATION.UNVERIFIED_SESSION')"
+            class="ion-android-alert text-y-800 fs-default"
+          />
         </h3>
         <div class="conversation--header--actions">
           <inbox-name :inbox="inbox" class="margin-right-small" />
@@ -73,11 +78,15 @@ export default {
       uiFlags: 'inboxAssignableAgents/getUIFlags',
       currentChat: 'getSelectedChat',
     }),
-
     chatMetadata() {
       return this.chat.meta;
     },
-
+    isHMACVerified() {
+      if (!this.isAWebWidgetInbox) {
+        return true;
+      }
+      return this.chatMetadata.hmac_verified;
+    },
     currentContact() {
       return this.$store.getters['contacts/getContact'](
         this.chat.meta.sender.id

app/javascript/dashboard/i18n/locale/en/conversation.json

@@ -1,6 +1,7 @@
 {
   "CONVERSATION": {
     "404": "Please select a conversation from left pane",
+    "UNVERIFIED_SESSION": "The identity of this user is not verified",
     "NO_MESSAGE_1": "Uh oh! Looks like there are no messages from customers in your inbox.",
     "NO_MESSAGE_2": " to send a message to your page!",
     "NO_INBOX_1": "Hola! Looks like you haven't added any inboxes yet.",

app/javascript/dashboard/routes/dashboard/conversation/contact/ConversationForm.vue

@@ -41,17 +41,17 @@
           </label>
         </div>
       </div>
-      <div class="row" v-if="isAnEmailInbox">
+      <div v-if="isAnEmailInbox" class="row">
         <div class="columns">
-          <label :class="{ error: $v.message.$error }">
+          <label :class="{ error: $v.subject.$error }">
             {{ $t('NEW_CONVERSATION.FORM.SUBJECT.LABEL') }}
             <input
               v-model="subject"
               type="text"
               :placeholder="$t('NEW_CONVERSATION.FORM.SUBJECT.PLACEHOLDER')"
-              @input="$v.message.$touch"
+              @input="$v.subject.$touch"
             />
-            <span v-if="$v.message.$error" class="message">
+            <span v-if="$v.subject.$error" class="message">
               {{ $t('NEW_CONVERSATION.FORM.SUBJECT.ERROR') }}
             </span>
           </label>
@@ -93,7 +93,7 @@ import Thumbnail from 'dashboard/components/widgets/Thumbnail';
 import alertMixin from 'shared/mixins/alertMixin';
 import { INBOX_TYPES } from 'shared/mixins/inboxMixin';
 import { ExceptionWithMessage } from 'shared/helpers/CustomErrors';
-import { required } from 'vuelidate/lib/validators';
+import { required, requiredIf } from 'vuelidate/lib/validators';
 
 export default {
   components: {
@@ -120,7 +120,7 @@ export default {
   },
   validations: {
     subject: {
-      required,
+      required: requiredIf('isAnEmailInbox'),
     },
     message: {
       required,

app/javascript/widget/helpers/actionCable.js

@@ -15,18 +15,6 @@ class ActionCableConnector extends BaseActionCableConnector {
     };
   }
 
-  static refreshConnector = pubsubToken => {
-    if (!pubsubToken || window.chatwootPubsubToken === pubsubToken) {
-      return;
-    }
-    window.chatwootPubsubToken = pubsubToken;
-    window.actionCable.disconnect();
-    window.actionCable = new ActionCableConnector(
-      window.WOOT_WIDGET,
-      window.chatwootPubsubToken
-    );
-  };
-
   onStatusChange = data => {
     this.app.$store.dispatch('conversationAttributes/update', data);
   };
@@ -57,7 +45,7 @@ class ActionCableConnector extends BaseActionCableConnector {
 
   onTypingOn = data => {
     if (data.is_private) {
-      return
+      return;
     }
     this.clearTimer();
     this.app.$store.dispatch('conversation/toggleAgentTyping', {
@@ -88,7 +76,4 @@ class ActionCableConnector extends BaseActionCableConnector {
   };
 }
 
-export const refreshActionCableConnector =
-  ActionCableConnector.refreshConnector;
-
 export default ActionCableConnector;

app/javascript/widget/helpers/campaignHelper.js

@@ -21,10 +21,16 @@ export const filterCampaigns = ({
   currentURL,
   isInBusinessHours,
 }) => {
-  return campaigns.filter(item =>
+  return campaigns.filter(campaign => {
-    item.triggerOnlyDuringBusinessHours
+    const hasMatchingURL =
-      ? isInBusinessHours
+      stripTrailingSlash({ URL: campaign.url }) ===
-      : stripTrailingSlash({ URL: item.url }) ===
+      stripTrailingSlash({ URL: currentURL });
-        stripTrailingSlash({ URL: currentURL })
+    if (!hasMatchingURL) {
-  );
+      return false;
+    }
+    if (campaign.triggerOnlyDuringBusinessHours) {
+      return isInBusinessHours;
+    }
+    return true;
+  });
 };

app/javascript/widget/helpers/specs/campaignHelper.spec.js

@@ -44,11 +44,13 @@ describe('#Campaigns Helper', () => {
               id: 1,
               timeOnPage: 3,
               url: 'https://www.chatwoot.com/pricing',
+              triggerOnlyDuringBusinessHours: false,
             },
             {
               id: 2,
               timeOnPage: 6,
               url: 'https://www.chatwoot.com/about',
+              triggerOnlyDuringBusinessHours: false,
             },
           ],
           currentURL: 'https://www.chatwoot.com/about/',
@@ -58,8 +60,60 @@ describe('#Campaigns Helper', () => {
           id: 2,
           timeOnPage: 6,
           url: 'https://www.chatwoot.com/about',
+          triggerOnlyDuringBusinessHours: false,
         },
       ]);
     });
+    it('should return filtered campaigns if formatted campaigns are passed and business hours enabled', () => {
+      expect(
+        filterCampaigns({
+          campaigns: [
+            {
+              id: 1,
+              timeOnPage: 3,
+              url: 'https://www.chatwoot.com/pricing',
+              triggerOnlyDuringBusinessHours: false,
+            },
+            {
+              id: 2,
+              timeOnPage: 6,
+              url: 'https://www.chatwoot.com/about',
+              triggerOnlyDuringBusinessHours: true,
+            },
+          ],
+          currentURL: 'https://www.chatwoot.com/about/',
+          isInBusinessHours: true,
+        })
+      ).toStrictEqual([
+        {
+          id: 2,
+          timeOnPage: 6,
+          url: 'https://www.chatwoot.com/about',
+          triggerOnlyDuringBusinessHours: true,
+        },
+      ]);
+    });
+    it('should return empty campaigns if formatted campaigns are passed and business hours disabled', () => {
+      expect(
+        filterCampaigns({
+          campaigns: [
+            {
+              id: 1,
+              timeOnPage: 3,
+              url: 'https://www.chatwoot.com/pricing',
+              triggerOnlyDuringBusinessHours: true,
+            },
+            {
+              id: 2,
+              timeOnPage: 6,
+              url: 'https://www.chatwoot.com/about',
+              triggerOnlyDuringBusinessHours: true,
+            },
+          ],
+          currentURL: 'https://www.chatwoot.com/about/',
+          isInBusinessHours: false,
+        })
+      ).toStrictEqual([]);
+    });
   });
 });

app/javascript/widget/store/modules/contacts.js

@@ -1,5 +1,4 @@
 import ContactsAPI from '../../api/contacts';
-import { refreshActionCableConnector } from '../../helpers/actionCable';
 
 const state = {
   currentUser: {},
@@ -31,17 +30,13 @@ export const actions = {
         identifier_hash: userObject.identifier_hash,
         phone_number: userObject.phone_number,
       };
-      const {
+      await ContactsAPI.update(identifier, user);
-        data: { pubsub_token: pubsubToken },
-      } = await ContactsAPI.update(identifier, user);
 
       dispatch('get');
       if (userObject.identifier_hash) {
         dispatch('conversation/clearConversations', {}, { root: true });
         dispatch('conversation/fetchOldConversations', {}, { root: true });
       }
-
-      refreshActionCableConnector(pubsubToken);
     } catch (error) {
       // Ignore error
     }

app/javascript/widget/store/modules/conversation/actions.js

@@ -6,7 +6,6 @@ import {
   toggleTyping,
   setUserLastSeenAt,
 } from 'widget/api/conversation';
-import { refreshActionCableConnector } from '../../../helpers/actionCable';
 
 import { createTemporaryMessage, getNonDeletedMessages } from './helpers';
 
@@ -15,13 +14,9 @@ export const actions = {
     commit('setConversationUIFlag', { isCreating: true });
     try {
       const { data } = await createConversationAPI(params);
-      const {
+      const { messages } = data;
-        contact: { pubsub_token: pubsubToken },
-        messages,
-      } = data;
       const [message = {}] = messages;
       commit('pushMessageToConversation', message);
-      refreshActionCableConnector(pubsubToken);
       dispatch('conversationAttributes/getAttributes', {}, { root: true });
     } catch (error) {
       // Ignore error

app/javascript/widget/store/modules/message.js

@@ -1,5 +1,4 @@
 import MessageAPI from '../../api/message';
-import { refreshActionCableConnector } from '../../helpers/actionCable';
 
 const state = {
   uiFlags: {
@@ -18,9 +17,7 @@ export const actions = {
   ) => {
     commit('toggleUpdateStatus', true);
     try {
-      const {
+      await MessageAPI.update({
-        data: { contact: { pubsub_token: pubsubToken } = {} },
-      } = await MessageAPI.update({
         email,
         messageId,
         values: submittedValues,
@@ -37,7 +34,6 @@ export const actions = {
         { root: true }
       );
       dispatch('contacts/get', {}, { root: true });
-      refreshActionCableConnector(pubsubToken);
     } catch (error) {
       // Ignore error
     }

app/listeners/action_cable_listener.rb

@@ -18,7 +18,7 @@ class ActionCableListener < BaseListener
   def message_created(event)
     message, account = extract_message_and_account(event)
     conversation = message.conversation
-    tokens = user_tokens(account, conversation.inbox.members) + contact_token(conversation.contact, message)
+    tokens = user_tokens(account, conversation.inbox.members) + contact_tokens(conversation.contact_inbox, message)
 
     broadcast(account, tokens, MESSAGE_CREATED, message.push_event_data)
   end
@@ -27,7 +27,7 @@ class ActionCableListener < BaseListener
     message, account = extract_message_and_account(event)
     conversation = message.conversation
     contact = conversation.contact
-    tokens = user_tokens(account, conversation.inbox.members) + contact_token(conversation.contact, message)
+    tokens = user_tokens(account, conversation.inbox.members) + contact_tokens(conversation.contact_inbox, message)
 
     broadcast(account, tokens, MESSAGE_UPDATED, message.push_event_data)
   end
@@ -132,12 +132,14 @@ class ActionCableListener < BaseListener
     (agent_tokens + admin_tokens).uniq
   end
 
-  def contact_token(contact, message)
+  def contact_tokens(contact_inbox, message)
     return [] if message.private?
     return [] if message.activity?
-    return [] if contact.nil?
+    return [] if contact_inbox.nil?
 
-    [contact.pubsub_token]
+    contact = contact_inbox.contact
+
+    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)

app/models/concerns/pubsubable.rb

@@ -7,4 +7,10 @@ module Pubsubable
     # Used by the actionCable/PubSub Service we use for real time communications
     has_secure_token :pubsub_token
   end
+
+  def pubsub_token
+    # backfills tokens for existing records
+    regenerate_pubsub_token if self[:pubsub_token].blank?
+    self[:pubsub_token]
+  end
 end

app/models/contact.rb

@@ -25,7 +25,7 @@
 #
 
 class Contact < ApplicationRecord
-  include Pubsubable
+  # TODO: remove the pubsub_token attribute from this model in future.
   include Avatarable
   include AvailabilityStatusable
   include Labelable

app/models/contact_inbox.rb

@@ -4,6 +4,7 @@
 #
 #  id            :bigint           not null, primary key
 #  hmac_verified :boolean          default(FALSE)
+#  pubsub_token  :string
 #  created_at    :datetime         not null
 #  updated_at    :datetime         not null
 #  contact_id    :bigint
@@ -15,6 +16,7 @@
 #  index_contact_inboxes_on_contact_id              (contact_id)
 #  index_contact_inboxes_on_inbox_id                (inbox_id)
 #  index_contact_inboxes_on_inbox_id_and_source_id  (inbox_id,source_id) UNIQUE
+#  index_contact_inboxes_on_pubsub_token            (pubsub_token) UNIQUE
 #  index_contact_inboxes_on_source_id               (source_id)
 #
 # Foreign Keys
@@ -24,6 +26,7 @@
 #
 
 class ContactInbox < ApplicationRecord
+  include Pubsubable
   validates :inbox_id, presence: true
   validates :contact_id, presence: true
   validates :source_id, presence: true

app/presenters/conversations/event_data_presenter.rb

@@ -23,7 +23,11 @@ class Conversations::EventDataPresenter < SimpleDelegator
   end
 
   def push_meta
-    { sender: contact.push_event_data, assignee: assignee&.push_event_data }
+    {
+      sender: contact.push_event_data,
+      assignee: assignee&.push_event_data,
+      hmac_verified: contact_inbox&.hmac_verified
+    }
   end
 
   def push_timestamps

app/views/api/v1/conversations/partials/_conversation.json.jbuilder

@@ -13,6 +13,7 @@ json.meta do
       json.partial! 'api/v1/models/team.json.jbuilder', resource: conversation.team
     end
   end
+  json.hmac_verified conversation.contact_inbox&.hmac_verified
 end
 
 json.id conversation.display_id

app/views/api/v1/widget/configs/create.json.jbuilder

@@ -22,7 +22,7 @@ json.chatwoot_widget_defaults do
   json.use_inbox_avatar_for_bot ActiveModel::Type::Boolean.new.cast(ENV.fetch('USE_INBOX_AVATAR_FOR_BOT', false))
 end
 json.contact do
-  json.pubsub_token @contact.pubsub_token
+  json.pubsub_token @contact_inbox.pubsub_token
 end
 json.auth_token @token
 json.global_config @global_config

app/views/public/api/v1/inboxes/contacts/create.json.jbuilder

@@ -1,2 +1,3 @@
 json.source_id @contact_inbox.source_id
+json.pubsub_token @contact_inbox.pubsub_token
 json.partial! 'public/api/v1/models/contact.json.jbuilder', resource: @contact_inbox.contact

app/views/public/api/v1/inboxes/contacts/show.json.jbuilder

@@ -1,2 +1,3 @@
 json.source_id @contact_inbox.source_id
+json.pubsub_token @contact_inbox.pubsub_token
 json.partial! 'public/api/v1/models/contact.json.jbuilder', resource: @contact_inbox.contact

app/views/public/api/v1/inboxes/contacts/update.json.jbuilder

@@ -1,2 +1,3 @@
 json.source_id @contact_inbox.source_id
+json.pubsub_token @contact_inbox.pubsub_token
 json.partial! 'public/api/v1/models/contact.json.jbuilder', resource: @contact_inbox.contact

app/views/public/api/v1/models/_contact.json.jbuilder

@@ -1,4 +1,3 @@
 json.id resource.id
 json.name resource.name
 json.email resource.email
-json.pubsub_token resource.pubsub_token

app/views/widgets/show.html.erb

@@ -28,7 +28,7 @@
       window.chatwootWidgetDefaults = {
         useInboxAvatarForBot: <%= ActiveModel::Type::Boolean.new.cast(ENV.fetch('USE_INBOX_AVATAR_FOR_BOT', false)) %>,
       }
-      window.chatwootPubsubToken = '<%= @contact.pubsub_token %>'
+      window.chatwootPubsubToken = '<%= @contact_inbox.pubsub_token %>'
       window.authToken = '<%= @token %>'
       window.globalConfig = <%= raw @global_config.to_json %>
     </script>

config/app.yml

@@ -1,5 +1,5 @@
 shared: &shared
-  version: '1.22.0'
+  version: '1.22.1'
 
 development:
   <<: *shared

db/migrate/20211122061012_add_pub_sub_token_to_contact_inbox.rb

@@ -0,0 +1,6 @@
+class AddPubSubTokenToContactInbox < ActiveRecord::Migration[6.1]
+  def change
+    add_column :contact_inboxes, :pubsub_token, :string
+    add_index :contact_inboxes, :pubsub_token, unique: true
+  end
+end

db/schema.rb

@@ -10,7 +10,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema.define(version: 2021_09_29_150415) do
+ActiveRecord::Schema.define(version: 2021_11_22_061012) do
 
   # These are extensions that must be enabled in order to support this database
   enable_extension "pg_stat_statements"
@@ -273,9 +273,11 @@ ActiveRecord::Schema.define(version: 2021_09_29_150415) do
     t.datetime "created_at", precision: 6, null: false
     t.datetime "updated_at", precision: 6, null: false
     t.boolean "hmac_verified", default: false
+    t.string "pubsub_token"
     t.index ["contact_id"], name: "index_contact_inboxes_on_contact_id"
     t.index ["inbox_id", "source_id"], name: "index_contact_inboxes_on_inbox_id_and_source_id", unique: true
     t.index ["inbox_id"], name: "index_contact_inboxes_on_inbox_id"
+    t.index ["pubsub_token"], name: "index_contact_inboxes_on_pubsub_token", unique: true
     t.index ["source_id"], name: "index_contact_inboxes_on_source_id"
   end
 

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@chatwoot/chatwoot",
-  "version": "1.22.0",
+  "version": "1.22.1",
   "license": "MIT",
   "scripts": {
     "eslint": "eslint app/javascript --fix",

spec/channels/room_channel_spec.rb

@@ -1,15 +1,15 @@
 require 'rails_helper'
 
 RSpec.describe RoomChannel, type: :channel do
-  let!(:contact) { create(:contact) }
+  let!(:contact_inbox) { create(:contact_inbox) }
 
   before do
     stub_connection
   end
 
   it 'subscribes to a stream when pubsub_token is provided' do
-    subscribe(pubsub_token: contact.pubsub_token)
+    subscribe(pubsub_token: contact_inbox.pubsub_token)
     expect(subscription).to be_confirmed
-    expect(subscription).to have_stream_for(contact.pubsub_token)
+    expect(subscription).to have_stream_for(contact_inbox.pubsub_token)
   end
 end

spec/controllers/api/v1/widget/configs_controller_spec.rb

@@ -45,7 +45,7 @@ RSpec.describe '/api/v1/widget/config', type: :request do
         expect(response).to have_http_status(:success)
         response_data = JSON.parse(response.body)
         expect(response_data.keys).to include(*response_keys)
-        expect(response_data['contact']['pubsub_token']).to eq(contact.pubsub_token)
+        expect(response_data['contact']['pubsub_token']).to eq(contact_inbox.pubsub_token)
       end
     end
 

spec/controllers/public/api/v1/inbox/contacts_controller_spec.rb

@@ -23,7 +23,7 @@ RSpec.describe 'Public Inbox Contacts API', type: :request do
       expect(response).to have_http_status(:success)
       data = JSON.parse(response.body)
       expect(data['source_id']).to eq contact_inbox.source_id
-      expect(data['pubsub_token']).to eq contact.pubsub_token
+      expect(data['pubsub_token']).to eq contact_inbox.pubsub_token
     end
   end
 

spec/listeners/action_cable_listener_spec.rb

@@ -24,7 +24,23 @@ describe ActionCableListener 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.pubsub_token],
+        [agent.pubsub_token, admin.pubsub_token, 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)
+      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(
+        [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)
       )

spec/models/contact_inbox_spec.rb

@@ -0,0 +1,40 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe ContactInbox do
+  describe 'pubsub_token' do
+    let(:contact_inbox) { create(:contact_inbox) }
+
+    it 'gets created on object create' do
+      obj = contact_inbox
+      expect(obj.pubsub_token).not_to eq(nil)
+    end
+
+    it 'does not get updated on object update' do
+      obj = contact_inbox
+      old_token = obj.pubsub_token
+      obj.update(source_id: '234234323')
+      expect(obj.pubsub_token).to eq(old_token)
+    end
+
+    it 'backfills pubsub_token on call for older objects' do
+      obj = create(:contact_inbox)
+      # to replicate an object with out pubsub_token
+      # rubocop:disable Rails/SkipsModelValidations
+      obj.update_column(:pubsub_token, nil)
+      # rubocop:enable Rails/SkipsModelValidations
+
+      obj.reload
+
+      # ensure the column is nil in database
+      results = ActiveRecord::Base.connection.execute('Select * from contact_inboxes;')
+      expect(results.first['pubsub_token']).to eq(nil)
+
+      new_token = obj.pubsub_token
+      obj.update(source_id: '234234323')
+      # the generated token shoul be persisted in db
+      expect(obj.pubsub_token).to eq(new_token)
+    end
+  end
+end

spec/models/contact_spec.rb

@@ -11,20 +11,4 @@ RSpec.describe Contact do
     it { is_expected.to belong_to(:account) }
     it { is_expected.to have_many(:conversations).dependent(:destroy) }
   end
-
-  describe 'pubsub_token' do
-    let(:user) { create(:user) }
-
-    it 'gets created on object create' do
-      obj = user
-      expect(obj.pubsub_token).not_to eq(nil)
-    end
-
-    it 'does not get updated on object update' do
-      obj = user
-      old_token = obj.pubsub_token
-      obj.update(name: 'test')
-      expect(obj.pubsub_token).to eq(old_token)
-    end
-  end
 end

spec/models/conversation_spec.rb

@@ -375,7 +375,8 @@ RSpec.describe Conversation, type: :model do
         additional_attributes: {},
         meta: {
           sender: conversation.contact.push_event_data,
-          assignee: conversation.assignee
+          assignee: conversation.assignee,
+          hmac_verified: conversation.contact_inbox.hmac_verified
         },
         id: conversation.display_id,
         messages: [],

spec/presenters/conversations/event_data_presenter_spec.rb

@@ -12,7 +12,8 @@ RSpec.describe Conversations::EventDataPresenter do
         additional_attributes: {},
         meta: {
           sender: conversation.contact.push_event_data,
-          assignee: conversation.assignee
+          assignee: conversation.assignee,
+          hmac_verified: conversation.contact_inbox.hmac_verified
         },
         id: conversation.display_id,
         messages: [],