feat: Unread Count: added api, store refresher, invalidation and events (2/3)[CW-6851] (#14369)

# Pull Request Template

## Description

This is the second PR in a series of PRs for Introducing unread counts
in the sidebar for inboxes and labels.

In this PR:

* added api for unread counts 
* Added the store refresher and invalidation with event listeners
* Added action cable event
* Added specs for the changes

Issue:
https://linear.app/chatwoot/issue/CW-6851/support-unread-conversation-counts

## Type of change

Please delete options that are not relevant.

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality not to work as expected)
- [ ] This change requires a documentation update

## How Has This Been Tested?

Please describe the tests that you ran to verify your changes. Provide
instructions so we can reproduce. Please also list any relevant details
for your test configuration.


## Checklist:

- [x] My code follows the style guidelines of this project
- [x] I have performed a self-review of my code
- [ ] I have commented on my code, particularly in hard-to-understand
areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] Any dependent changes have been merged and published in downstream
modules

---------

Co-authored-by: Sojan Jose <sojan@pepalo.com>

app/controllers/api/v1/accounts/conversations/unread_counts_controller.rb

@@ -0,0 +1,16 @@
+class Api::V1::Accounts::Conversations::UnreadCountsController < Api::V1::Accounts::BaseController
+  before_action :ensure_unread_counts_enabled
+
+  def index
+    counts = ::Conversations::UnreadCounts::Counter.new(account: Current.account, user: Current.user).perform
+    render json: { payload: counts }
+  end
+
+  private
+
+  def ensure_unread_counts_enabled
+    return if Current.account.feature_enabled?('conversation_unread_counts')
+
+    render json: { error: I18n.t('errors.conversations.unread_counts.feature_not_enabled') }, status: :forbidden
+  end
+end

app/controllers/api/v1/accounts/conversations_controller.rb

@@ -162,6 +162,8 @@ class Api::V1::Accounts::ConversationsController < Api::V1::Accounts::BaseContro
     # rubocop:disable Rails/SkipsModelValidations
     @conversation.update_columns(updates)
     # rubocop:enable Rails/SkipsModelValidations
+
+    ::Conversations::UnreadCounts::Notifier.new(@conversation).perform
   end
 
   def should_update_last_seen?

app/dispatchers/async_dispatcher.rb

@@ -17,6 +17,7 @@ class AsyncDispatcher < BaseDispatcher
       InstallationWebhookListener.instance,
       NotificationListener.instance,
       ParticipationListener.instance,
+      Conversations::UnreadCounts::Listener.instance,
       ReportingEventListener.instance,
       WebhookListener.instance
     ]

app/listeners/action_cable_listener.rb

@@ -90,6 +90,15 @@ class ActionCableListener < BaseListener
     broadcast(account, tokens, CONVERSATION_UPDATED, conversation.push_event_data)
   end
 
+  def conversation_unread_count_changed(event)
+    account, inbox_members = ::Conversations::UnreadCounts::BroadcastScope.new(event).perform
+    return if account.blank? || !account.feature_enabled?('conversation_unread_counts')
+
+    tokens = user_tokens(account, inbox_members)
+
+    broadcast(account, tokens, CONVERSATION_UNREAD_COUNT_CHANGED, {})
+  end
+
   def conversation_typing_on(event)
     conversation = event.data[:conversation]
     account = conversation.account

app/models/account.rb

@@ -109,6 +109,7 @@ class Account < ApplicationRecord
 
   before_validation :validate_limit_keys
   after_create_commit :notify_creation
+  after_update_commit :clear_unread_conversation_counts_cache, if: :saved_change_to_feature_conversation_unread_counts?
   after_destroy :remove_account_sequences
 
   def agents
@@ -174,6 +175,10 @@ class Account < ApplicationRecord
     Rails.configuration.dispatcher.dispatch(ACCOUNT_CREATED, Time.zone.now, account: self)
   end
 
+  def clear_unread_conversation_counts_cache
+    ::Conversations::UnreadCounts::Store.clear_account!(id)
+  end
+
   trigger.after(:insert).for_each(:row) do
     "execute format('create sequence IF NOT EXISTS conv_dpid_seq_%s', NEW.id);"
   end

app/models/concerns/cache_keys.rb

@@ -30,6 +30,7 @@ module CacheKeys
       update_cache_key_for_account(id, model.name.underscore)
     end
 
+    ::Conversations::UnreadCounts::Store.clear_account!(id)
     dispatch_cache_update_event
   end
 

app/models/conversation.rb

@@ -121,6 +121,8 @@ class Conversation < ApplicationRecord
   after_update_commit :execute_after_update_commit_callbacks
   after_create_commit :notify_conversation_creation
   after_create_commit :load_attributes_created_by_db_triggers
+  before_destroy :set_unread_count_deletion_data
+  after_destroy_commit :notify_conversation_deletion
 
   delegate :auto_resolve_after, to: :account
 
@@ -270,6 +272,12 @@ class Conversation < ApplicationRecord
     dispatcher_dispatch(CONVERSATION_CREATED)
   end
 
+  def notify_conversation_deletion
+    return if @unread_count_deletion_data.blank?
+
+    Rails.configuration.dispatcher.dispatch(CONVERSATION_DELETED, Time.zone.now, conversation_data: @unread_count_deletion_data)
+  end
+
   def notify_conversation_updation
     return unless previous_changes.keys.present? && allowed_keys?
 
@@ -315,6 +323,17 @@ class Conversation < ApplicationRecord
                                                                        performed_by: Current.executed_by)
   end
 
+  def set_unread_count_deletion_data
+    @unread_count_deletion_data = {
+      id: id,
+      account_id: account_id,
+      inbox_id: inbox_id,
+      assignee_id: assignee_id,
+      team_id: team_id,
+      cached_label_list: cached_label_list
+    }
+  end
+
   def conversation_status_changed_to_open?
     return false unless open?
     # saved_change_to_status? method only works in case of update

app/services/conversations/unread_counts/broadcast_scope.rb

@@ -0,0 +1,36 @@
+class Conversations::UnreadCounts::BroadcastScope
+  attr_reader :event
+
+  def initialize(event)
+    @event = event
+  end
+
+  def perform
+    return [conversation.account, conversation.inbox.members] if conversation.present?
+
+    deleted_conversation_scope
+  end
+
+  private
+
+  def conversation
+    event.data[:conversation]
+  end
+
+  def deleted_conversation_scope
+    conversation_data = event.data[:conversation_data]&.with_indifferent_access
+    return if conversation_data.blank?
+
+    account = Account.find_by(id: conversation_data[:account_id])
+    return if account.blank?
+
+    [account, inbox_members_for(account, conversation_data[:inbox_id])]
+  end
+
+  def inbox_members_for(account, inbox_id)
+    inbox = account.inboxes.find_by(id: inbox_id)
+    return User.none if inbox.blank?
+
+    inbox.members
+  end
+end

app/services/conversations/unread_counts/listener.rb

@@ -0,0 +1,96 @@
+class Conversations::UnreadCounts::Listener < BaseListener
+  include Events::Types
+
+  def message_created(event)
+    message, = extract_message_and_account(event)
+    return unless message.incoming?
+    return unless message.account.feature_enabled?('conversation_unread_counts')
+
+    refresh(message.conversation)
+  end
+
+  def conversation_status_changed(event)
+    conversation, = extract_conversation_and_account(event)
+    refresh(conversation, event.data[:changed_attributes])
+  end
+
+  def conversation_updated(event)
+    return unless label_changed?(event.data[:changed_attributes])
+
+    conversation, = extract_conversation_and_account(event)
+    refresh(conversation, event.data[:changed_attributes])
+  end
+
+  def assignee_changed(event)
+    conversation, = extract_conversation_and_account(event)
+    refresh(conversation, event.data[:changed_attributes])
+  end
+
+  def team_changed(event)
+    conversation, = extract_conversation_and_account(event)
+    refresh(conversation, event.data[:changed_attributes])
+  end
+
+  def conversation_deleted(event)
+    conversation_data = event.data[:conversation_data]&.with_indifferent_access
+    return if conversation_data.blank?
+
+    account = Account.find_by(id: conversation_data[:account_id])
+    return unless account&.feature_enabled?('conversation_unread_counts')
+    return unless remove_deleted_conversation(account, conversation_data)
+
+    Rails.configuration.dispatcher.dispatch(CONVERSATION_UNREAD_COUNT_CHANGED, Time.zone.now, conversation_data: conversation_data.to_h)
+  end
+
+  private
+
+  def refresh(conversation, changed_attributes = nil)
+    ::Conversations::UnreadCounts::Notifier.new(conversation, changed_attributes: changed_attributes).perform
+  end
+
+  def remove_deleted_conversation(account, conversation_data)
+    return false unless store.base_ready?(account.id) || store.assignment_ready?(account.id)
+
+    removed = false
+    removed = remove_deleted_base_membership(account, conversation_data) || removed if store.base_ready?(account.id)
+    removed = remove_deleted_assignment_membership(account, conversation_data) || removed if store.assignment_ready?(account.id)
+    removed
+  end
+
+  def remove_deleted_base_membership(account, conversation_data)
+    store.remove_base_membership(
+      account_id: account.id,
+      inbox_ids: [conversation_data[:inbox_id]],
+      label_ids: label_ids_for(account, conversation_data[:cached_label_list]),
+      team_ids: [conversation_data[:team_id]],
+      conversation_id: conversation_data[:id]
+    )
+  end
+
+  def remove_deleted_assignment_membership(account, conversation_data)
+    store.remove_assignment_membership(
+      account_id: account.id,
+      inbox_ids: [conversation_data[:inbox_id]],
+      label_ids: label_ids_for(account, conversation_data[:cached_label_list]),
+      assignee_ids: [conversation_data[:assignee_id]],
+      team_ids: [conversation_data[:team_id]],
+      conversation_id: conversation_data[:id]
+    )
+  end
+
+  def label_ids_for(account, label_list)
+    label_titles = label_list.to_s.split(',').map(&:strip).compact_blank
+    account.labels.pluck(:title, :id).to_h.values_at(*label_titles).compact
+  end
+
+  def label_changed?(changed_attributes)
+    return false if changed_attributes.blank?
+
+    changed_attributes.key?('label_list') || changed_attributes.key?(:label_list) ||
+      changed_attributes.key?('cached_label_list') || changed_attributes.key?(:cached_label_list)
+  end
+
+  def store
+    ::Conversations::UnreadCounts::Store
+  end
+end

app/services/conversations/unread_counts/notifier.rb

@@ -0,0 +1,19 @@
+class Conversations::UnreadCounts::Notifier
+  include Events::Types
+
+  attr_reader :conversation, :changed_attributes
+
+  def initialize(conversation, changed_attributes: nil)
+    @conversation = conversation
+    @changed_attributes = changed_attributes
+  end
+
+  def perform
+    return false unless conversation.account.feature_enabled?('conversation_unread_counts')
+
+    return false unless ::Conversations::UnreadCounts::Refresher.new(conversation, changed_attributes: changed_attributes).perform
+
+    Rails.configuration.dispatcher.dispatch(CONVERSATION_UNREAD_COUNT_CHANGED, Time.zone.now, conversation: conversation)
+    true
+  end
+end

app/services/conversations/unread_counts/refresher.rb

@@ -0,0 +1,178 @@
+class Conversations::UnreadCounts::Refresher
+  attr_reader :conversation, :changed_attributes
+
+  def initialize(conversation, changed_attributes: nil)
+    @conversation = conversation
+    @changed_attributes = changed_attributes.is_a?(Hash) ? changed_attributes : {}
+  end
+
+  def perform
+    return false unless base_ready? || assignment_ready?
+
+    before_memberships = store.memberships_for_keys(affected_cache_keys, conversation.id)
+    refresh_base_membership if base_ready?
+    refresh_assignment_membership if assignment_ready?
+    after_memberships = store.memberships_for_keys(affected_cache_keys, conversation.id)
+
+    before_memberships != after_memberships
+  end
+
+  private
+
+  def affected_cache_keys
+    keys = []
+    keys.concat(affected_base_keys) if base_ready?
+    keys.concat(affected_assignment_keys) if assignment_ready?
+    keys.uniq
+  end
+
+  def affected_base_keys
+    affected_inbox_ids.flat_map do |inbox_id|
+      [store.inbox_key(account.id, inbox_id)] +
+        affected_label_ids.map { |label_id| store.label_inbox_key(account.id, label_id, inbox_id) } +
+        affected_team_ids.map { |team_id| store.team_inbox_key(account.id, team_id, inbox_id) }
+    end
+  end
+
+  def affected_assignment_keys
+    affected_inbox_ids.flat_map do |inbox_id|
+      affected_assignee_ids.flat_map do |assignee_id|
+        assignment_keys_for(inbox_id, assignee_id)
+      end
+    end
+  end
+
+  def assignment_keys_for(inbox_id, assignee_id)
+    keys = assignee_id.present? ? assignee_keys_for(inbox_id, assignee_id) : unassigned_keys_for(inbox_id)
+
+    keys + affected_team_ids.map { |team_id| team_assignment_key_for(team_id, inbox_id, assignee_id) }
+  end
+
+  def assignee_keys_for(inbox_id, assignee_id)
+    [store.inbox_assignee_key(account.id, inbox_id, assignee_id)] +
+      affected_label_ids.map { |label_id| store.label_inbox_assignee_key(account.id, label_id, inbox_id, assignee_id) }
+  end
+
+  def unassigned_keys_for(inbox_id)
+    [store.inbox_unassigned_key(account.id, inbox_id)] +
+      affected_label_ids.map { |label_id| store.label_inbox_unassigned_key(account.id, label_id, inbox_id) }
+  end
+
+  def refresh_base_membership
+    store.remove_base_membership(
+      account_id: account.id,
+      inbox_ids: affected_inbox_ids,
+      label_ids: affected_label_ids,
+      team_ids: affected_team_ids,
+      conversation_id: conversation.id
+    )
+    return unless unread?
+
+    store.add_base_membership(
+      account_id: account.id,
+      inbox_id: conversation.inbox_id,
+      label_ids: current_label_ids,
+      team_id: conversation.team_id,
+      conversation_id: conversation.id
+    )
+  end
+
+  def refresh_assignment_membership
+    store.remove_assignment_membership(
+      account_id: account.id,
+      inbox_ids: affected_inbox_ids,
+      label_ids: affected_label_ids,
+      assignee_ids: affected_assignee_ids,
+      team_ids: affected_team_ids,
+      conversation_id: conversation.id
+    )
+    return unless unread?
+
+    store.add_assignment_membership(
+      account_id: account.id,
+      inbox_id: conversation.inbox_id,
+      label_ids: current_label_ids,
+      assignee_id: conversation.assignee_id,
+      team_id: conversation.team_id,
+      conversation_id: conversation.id
+    )
+  end
+
+  def unread?
+    # Sidebar unread counts intentionally track only open conversations.
+    return false unless conversation.open?
+
+    incoming_messages = conversation.messages.incoming.where(account_id: account.id)
+    if conversation.agent_last_seen_at
+      incoming_messages = incoming_messages.where(Message.arel_table[:created_at].gt(conversation.agent_last_seen_at))
+    end
+    incoming_messages.exists?
+  end
+
+  def affected_inbox_ids
+    [previous_value_for(:inbox_id), conversation.inbox_id].compact.uniq
+  end
+
+  def affected_assignee_ids
+    return [conversation.assignee_id].compact unless changed_attribute?(:assignee_id)
+
+    [previous_value_for(:assignee_id), conversation.assignee_id].uniq
+  end
+
+  def affected_label_ids
+    (previous_label_ids + current_label_ids).uniq
+  end
+
+  def affected_team_ids
+    [previous_value_for(:team_id), conversation.team_id].compact.uniq
+  end
+
+  def previous_label_ids
+    label_ids_for(previous_value_for(:label_list) || previous_value_for(:cached_label_list) || conversation.cached_label_list)
+  end
+
+  def current_label_ids
+    @current_label_ids ||= label_ids_for(conversation.cached_label_list)
+  end
+
+  def label_ids_for(label_list)
+    labels = label_list.is_a?(Array) ? label_list : label_list.to_s.split(',')
+    label_titles = labels.map(&:to_s).map(&:strip).compact_blank
+    labels_by_title.values_at(*label_titles).compact
+  end
+
+  def previous_value_for(attribute)
+    change = changed_attributes[attribute.to_s] || changed_attributes[attribute.to_sym]
+    change&.first
+  end
+
+  def changed_attribute?(attribute)
+    changed_attributes.key?(attribute.to_s) || changed_attributes.key?(attribute.to_sym)
+  end
+
+  def team_assignment_key_for(team_id, inbox_id, assignee_id)
+    return store.team_inbox_assignee_key(account.id, team_id, inbox_id, assignee_id) if assignee_id.present?
+
+    store.team_inbox_unassigned_key(account.id, team_id, inbox_id)
+  end
+
+  def labels_by_title
+    @labels_by_title ||= account.labels.pluck(:title, :id).to_h
+  end
+
+  def base_ready?
+    @base_ready ||= store.base_ready?(account.id)
+  end
+
+  def assignment_ready?
+    @assignment_ready ||= store.assignment_ready?(account.id)
+  end
+
+  def account
+    conversation.account
+  end
+
+  def store
+    ::Conversations::UnreadCounts::Store
+  end
+end

app/services/conversations/unread_counts/store.rb

@@ -1,4 +1,6 @@
 class Conversations::UnreadCounts::Store
+  extend ::Conversations::UnreadCounts::StoreKeys
+
   class << self
     def base_ready?(account_id)
       Redis::Alfred.exists?(base_ready_key(account_id))
@@ -78,46 +80,14 @@ class Conversations::UnreadCounts::Store
       keys.zip(counts).to_h
     end
 
-    def inbox_key(account_id, inbox_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_INBOX, account_id: account_id, inbox_id: inbox_id)
-    end
+    def memberships_for_keys(keys, conversation_id)
+      keys = keys.compact_blank
+      return {} if keys.blank?
 
-    def label_inbox_key(account_id, label_id, inbox_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_LABEL_INBOX, account_id: account_id, label_id: label_id, inbox_id: inbox_id)
-    end
-
-    def team_inbox_key(account_id, team_id, inbox_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_TEAM_INBOX, account_id: account_id, team_id: team_id, inbox_id: inbox_id)
-    end
-
-    def inbox_unassigned_key(account_id, inbox_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_INBOX_UNASSIGNED, account_id: account_id, inbox_id: inbox_id)
-    end
-
-    def inbox_assignee_key(account_id, inbox_id, user_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_INBOX_ASSIGNEE, account_id: account_id, inbox_id: inbox_id, user_id: user_id)
-    end
-
-    def label_inbox_unassigned_key(account_id, label_id, inbox_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_LABEL_INBOX_UNASSIGNED, account_id: account_id, label_id: label_id, inbox_id: inbox_id)
-    end
-
-    def label_inbox_assignee_key(account_id, label_id, inbox_id, user_id)
-      format(
-        Redis::Alfred::UNREAD_CONVERSATIONS_LABEL_INBOX_ASSIGNEE,
-        account_id: account_id,
-        label_id: label_id,
-        inbox_id: inbox_id,
-        user_id: user_id
-      )
-    end
-
-    def team_inbox_unassigned_key(account_id, team_id, inbox_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_TEAM_INBOX_UNASSIGNED, account_id: account_id, team_id: team_id, inbox_id: inbox_id)
-    end
-
-    def team_inbox_assignee_key(account_id, team_id, inbox_id, user_id)
-      format(Redis::Alfred::UNREAD_CONVERSATIONS_TEAM_INBOX_ASSIGNEE, account_id: account_id, team_id: team_id, inbox_id: inbox_id, user_id: user_id)
+      memberships = Redis::Alfred.pipelined do |pipeline|
+        keys.each { |key| pipeline.sismember(key, conversation_id) }
+      end
+      keys.zip(memberships.map { |membership| membership == true || membership == 1 }).to_h
     end
 
     private
@@ -183,7 +153,16 @@ class Conversations::UnreadCounts::Store
     end
 
     def remove_from_sets(keys, conversation_id)
-      write_to_sets(keys) { |pipeline, key| pipeline.srem(key, conversation_id) }
+      keys = keys.compact_blank
+      return false if keys.blank?
+
+      results = Redis::Alfred.pipelined do |pipeline|
+        keys.each do |key|
+          pipeline.srem(key, conversation_id)
+          pipeline.expire(key, Conversations::UnreadCounts::SET_TTL)
+        end
+      end
+      results.each_slice(2).any? { |removed, _| removed == true || (removed.respond_to?(:to_i) && removed.to_i.positive?) }
     end
 
     def write_to_sets(keys)

app/services/conversations/unread_counts/store_keys.rb

@@ -0,0 +1,43 @@
+module Conversations::UnreadCounts::StoreKeys
+  def inbox_key(account_id, inbox_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_INBOX, account_id: account_id, inbox_id: inbox_id)
+  end
+
+  def label_inbox_key(account_id, label_id, inbox_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_LABEL_INBOX, account_id: account_id, label_id: label_id, inbox_id: inbox_id)
+  end
+
+  def team_inbox_key(account_id, team_id, inbox_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_TEAM_INBOX, account_id: account_id, team_id: team_id, inbox_id: inbox_id)
+  end
+
+  def inbox_unassigned_key(account_id, inbox_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_INBOX_UNASSIGNED, account_id: account_id, inbox_id: inbox_id)
+  end
+
+  def inbox_assignee_key(account_id, inbox_id, user_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_INBOX_ASSIGNEE, account_id: account_id, inbox_id: inbox_id, user_id: user_id)
+  end
+
+  def label_inbox_unassigned_key(account_id, label_id, inbox_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_LABEL_INBOX_UNASSIGNED, account_id: account_id, label_id: label_id, inbox_id: inbox_id)
+  end
+
+  def label_inbox_assignee_key(account_id, label_id, inbox_id, user_id)
+    format(
+      Redis::Alfred::UNREAD_CONVERSATIONS_LABEL_INBOX_ASSIGNEE,
+      account_id: account_id,
+      label_id: label_id,
+      inbox_id: inbox_id,
+      user_id: user_id
+    )
+  end
+
+  def team_inbox_unassigned_key(account_id, team_id, inbox_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_TEAM_INBOX_UNASSIGNED, account_id: account_id, team_id: team_id, inbox_id: inbox_id)
+  end
+
+  def team_inbox_assignee_key(account_id, team_id, inbox_id, user_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_TEAM_INBOX_ASSIGNEE, account_id: account_id, team_id: team_id, inbox_id: inbox_id, user_id: user_id)
+  end
+end

config/features.yml

@@ -17,10 +17,10 @@
   display_name: Facebook Channel
   enabled: true
   help_url: https://chwt.app/hc/fb
-- name: channel_twitter
-  display_name: Twitter Channel
-  enabled: true
-  deprecated: true
+- name: conversation_unread_counts
+  display_name: Conversation Unread Counts
+  enabled: false
+  chatwoot_internal: true
 - name: ip_lookup
   display_name: IP Lookup
   enabled: false

config/locales/en.yml

@@ -81,6 +81,9 @@ en:
     saml:
       feature_not_enabled: SAML feature not enabled for this account
       sso_not_enabled: SAML SSO is not enabled for this installation
+    conversations:
+      unread_counts:
+        feature_not_enabled: Conversation unread counts feature not enabled for this account
     data_import:
       data_type:
         invalid: Invalid data type

config/routes.rb

@@ -133,6 +133,7 @@ Rails.application.routes.draw do
             collection do
               get :meta
               get :search
+              get :unread_counts, to: 'conversations/unread_counts#index'
               post :filter
             end
             scope module: :conversations do

db/migrate/20260508000000_repurpose_channel_twitter_flag_for_conversation_unread_counts.rb

@@ -0,0 +1,20 @@
+class RepurposeChannelTwitterFlagForConversationUnreadCounts < ActiveRecord::Migration[7.1]
+  def up
+    # The channel_twitter flag (deprecated) has been renamed to conversation_unread_counts.
+    # Disable it on any accounts that had channel_twitter enabled so the repurposed
+    # flag starts in its intended default-off state.
+    Account.feature_conversation_unread_counts.find_each(batch_size: 100) do |account|
+      account.disable_features(:conversation_unread_counts)
+      account.save!(validate: false)
+    end
+
+    # Remove the stale channel_twitter entry from ACCOUNT_LEVEL_FEATURE_DEFAULTS.
+    # ConfigLoader only adds new flags; it never removes renamed ones.
+    config = InstallationConfig.find_by(name: 'ACCOUNT_LEVEL_FEATURE_DEFAULTS')
+    return if config&.value.blank?
+
+    config.value = config.value.reject { |feature| feature['name'] == 'channel_twitter' }
+    config.save!
+    GlobalConfig.clear_cache
+  end
+end

lib/events/types.rb

@@ -16,6 +16,7 @@ module Events::Types
   # conversation events
   CONVERSATION_CREATED = 'conversation.created'
   CONVERSATION_UPDATED = 'conversation.updated'
+  CONVERSATION_DELETED = 'conversation.deleted'
   CONVERSATION_READ = 'conversation.read'
   CONVERSATION_BOT_HANDOFF = 'conversation.bot_handoff'
   # FIXME: deprecate the opened and resolved events in future in favor of status changed event.
@@ -26,6 +27,7 @@ module Events::Types
 
   CONVERSATION_STATUS_CHANGED = 'conversation.status_changed'
   CONVERSATION_CONTACT_CHANGED = 'conversation.contact_changed'
+  CONVERSATION_UNREAD_COUNT_CHANGED = 'conversation.unread_count_changed'
   ASSIGNEE_CHANGED = 'assignee.changed'
   TEAM_CHANGED = 'team.changed'
   CONVERSATION_TYPING_ON = 'conversation.typing_on'

spec/controllers/api/v1/accounts/conversations_controller_spec.rb

@@ -101,6 +101,76 @@ RSpec.describe 'Conversations API', type: :request do
     end
   end
 
+  describe 'GET /api/v1/accounts/{account.id}/conversations/unread_counts' do
+    context 'when it is an unauthenticated user' do
+      it 'returns unauthorized' do
+        get "/api/v1/accounts/#{account.id}/conversations/unread_counts"
+
+        expect(response).to have_http_status(:unauthorized)
+      end
+    end
+
+    context 'when it is an authenticated user' do
+      let(:agent) { create(:user, account: account, role: :agent) }
+      let(:visible_inbox) { create(:inbox, account: account) }
+      let(:hidden_inbox) { create(:inbox, account: account) }
+      let(:label) { create(:label, account: account, title: 'billing', show_on_sidebar: true) }
+      let(:team) { create(:team, account: account, allow_auto_assign: false) }
+
+      before do
+        create(:inbox_member, user: agent, inbox: visible_inbox)
+        create(:team_member, user: agent, team: team)
+      end
+
+      after do
+        Conversations::UnreadCounts::Store.clear_account!(account.id)
+      end
+
+      context 'when conversation unread counts feature is enabled' do
+        before do
+          account.enable_features!(:conversation_unread_counts)
+        end
+
+        it 'returns unread conversation counts scoped to the signed-in user' do
+          create_unread_conversation(account: account, inbox: visible_inbox, labels: [label.title])
+          create_unread_conversation(account: account, inbox: hidden_inbox, labels: [label.title])
+
+          get "/api/v1/accounts/#{account.id}/conversations/unread_counts",
+              headers: agent.create_new_auth_token,
+              as: :json
+
+          expect(response).to have_http_status(:success)
+          expect(response.parsed_body['payload']).to eq(
+            'inboxes' => { visible_inbox.id.to_s => 1 },
+            'labels' => { label.id.to_s => 1 },
+            'teams' => {}
+          )
+        end
+
+        it 'returns unread team conversation counts scoped to the signed-in user' do
+          create_unread_conversation(account: account, inbox: visible_inbox, team: team)
+          create_unread_conversation(account: account, inbox: hidden_inbox, team: team)
+
+          get "/api/v1/accounts/#{account.id}/conversations/unread_counts",
+              headers: agent.create_new_auth_token,
+              as: :json
+
+          expect(response).to have_http_status(:success)
+          expect(response.parsed_body['payload']['teams']).to eq(team.id.to_s => 1)
+        end
+      end
+
+      it 'returns forbidden when conversation unread counts feature is disabled' do
+        get "/api/v1/accounts/#{account.id}/conversations/unread_counts",
+            headers: agent.create_new_auth_token,
+            as: :json
+
+        expect(response).to have_http_status(:forbidden)
+        expect(response.parsed_body['error']).to eq('Conversation unread counts feature not enabled for this account')
+      end
+    end
+  end
+
   describe 'GET /api/v1/accounts/{account.id}/conversations/search' do
     context 'when it is an unauthenticated user' do
       it 'returns unauthorized' do
@@ -777,6 +847,23 @@ RSpec.describe 'Conversations API', type: :request do
         expect(conversation.reload.agent_last_seen_at).to be > initial_last_seen
       end
 
+      it 'refreshes unread count cache when conversation is marked read' do
+        account.enable_features!(:conversation_unread_counts)
+        conversation.update!(agent_last_seen_at: 1.hour.ago)
+        create(:message, account: account, inbox: conversation.inbox, conversation: conversation, message_type: :incoming, created_at: 5.minutes.ago)
+        Conversations::UnreadCounts::Builder.new(account).build_base!
+
+        post "/api/v1/accounts/#{account.id}/conversations/#{conversation.display_id}/update_last_seen",
+             headers: agent.create_new_auth_token,
+             as: :json
+
+        inbox_key = Conversations::UnreadCounts::Store.inbox_key(account.id, conversation.inbox_id)
+        expect(response).to have_http_status(:success)
+        expect(Conversations::UnreadCounts::Store.counts_for_keys([inbox_key])).to eq(inbox_key => 0)
+      ensure
+        Conversations::UnreadCounts::Store.clear_account!(account.id)
+      end
+
       it 'updates both if one timestamp is old even when the other is recent' do
         conversation.update!(assignee_id: agent.id, agent_last_seen_at: 2.hours.ago, assignee_last_seen_at: 30.minutes.ago)
         # Ensure all messages are older than assignee_last_seen_at (no unread messages)
@@ -847,6 +934,22 @@ RSpec.describe 'Conversations API', type: :request do
         expect(conversation.reload.agent_last_seen_at).to eq(last_seen_at)
         expect(conversation.reload.assignee_last_seen_at).to eq(last_seen_at)
       end
+
+      it 'refreshes unread count cache when conversation is marked unread' do
+        account.enable_features!(:conversation_unread_counts)
+        conversation.update!(agent_last_seen_at: 1.minute.from_now, assignee_last_seen_at: 1.minute.from_now)
+        Conversations::UnreadCounts::Builder.new(account).build_base!
+
+        post "/api/v1/accounts/#{account.id}/conversations/#{conversation.display_id}/unread",
+             headers: agent.create_new_auth_token,
+             as: :json
+
+        inbox_key = Conversations::UnreadCounts::Store.inbox_key(account.id, conversation.inbox_id)
+        expect(response).to have_http_status(:success)
+        expect(Conversations::UnreadCounts::Store.counts_for_keys([inbox_key])).to eq(inbox_key => 1)
+      ensure
+        Conversations::UnreadCounts::Store.clear_account!(account.id)
+      end
     end
   end
 

spec/controllers/super_admin/accounts_controller_spec.rb

@@ -32,6 +32,10 @@ RSpec.describe 'Super Admin accounts API', type: :request do
       create(:team, account: account)
     end
 
+    after do
+      Conversations::UnreadCounts::Store.clear_account!(account.id)
+    end
+
     context 'when it is an unauthenticated user' do
       it 'returns unauthorized' do
         post "/super_admin/accounts/#{account.id}/reset_cache"
@@ -52,6 +56,21 @@ RSpec.describe 'Super Admin accounts API', type: :request do
         range = now_timestamp..(now_timestamp + 10)
         expect(account.reload.cache_keys.values.all? { |v| range.cover?(v.to_i) }).to be(true)
       end
+
+      it 'clears conversation unread count cache' do
+        inbox = account.inboxes.first
+        store = Conversations::UnreadCounts::Store
+        inbox_key = store.inbox_key(account.id, inbox.id)
+        store.mark_base_ready!(account.id)
+        store.add_base_membership(account_id: account.id, inbox_id: inbox.id, label_ids: [], conversation_id: 1)
+
+        sign_in(super_admin, scope: :super_admin)
+        post "/super_admin/accounts/#{account.id}/reset_cache"
+
+        expect(response).to have_http_status(:redirect)
+        expect(store.base_ready?(account.id)).to be(false)
+        expect(store.counts_for_keys([inbox_key])).to eq(inbox_key => 0)
+      end
     end
   end
 

spec/listeners/action_cable_listener_spec.rb

@@ -231,4 +231,68 @@ describe ActionCableListener do
       listener.conversation_updated(event)
     end
   end
+
+  describe '#conversation_unread_count_changed' do
+    let(:event_name) { :'conversation.unread_count_changed' }
+    let!(:agent_without_inbox_access) { create(:user, account: account, role: :agent) }
+    let!(:event) { Events::Base.new(event_name, Time.zone.now, conversation: conversation) }
+
+    before do
+      account.enable_features!(:conversation_unread_counts)
+    end
+
+    it 'sends a lightweight refresh event to inbox agents and admins' do
+      expect(conversation.inbox.reload.inbox_members.count).to eq(1)
+
+      expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+        a_collection_containing_exactly(agent.pubsub_token, admin.pubsub_token),
+        'conversation.unread_count_changed',
+        {
+          account_id: account.id
+        }
+      )
+
+      listener.conversation_unread_count_changed(event)
+    end
+
+    it 'does not broadcast unread count refresh to agents outside the inbox' do
+      expect(ActionCableBroadcastJob).not_to receive(:perform_later).with(
+        array_including(agent_without_inbox_access.pubsub_token),
+        anything,
+        anything
+      )
+
+      listener.conversation_unread_count_changed(event)
+    end
+
+    it 'does not broadcast when conversation unread counts feature is disabled' do
+      account.disable_features!(:conversation_unread_counts)
+
+      expect(ActionCableBroadcastJob).not_to receive(:perform_later)
+
+      listener.conversation_unread_count_changed(event)
+    end
+
+    it 'supports deleted conversation data' do
+      event = Events::Base.new(
+        event_name,
+        Time.zone.now,
+        conversation_data: {
+          id: conversation.id,
+          account_id: account.id,
+          inbox_id: conversation.inbox_id
+        }
+      )
+
+      expect(ActionCableBroadcastJob).to receive(:perform_later).with(
+        a_collection_containing_exactly(agent.pubsub_token, admin.pubsub_token),
+        'conversation.unread_count_changed',
+        {
+          account_id: account.id
+        }
+      )
+
+      listener.conversation_unread_count_changed(event)
+    end
+  end
 end

spec/models/account_spec.rb

@@ -50,6 +50,44 @@ RSpec.describe Account do
     end
   end
 
+  describe 'conversation unread counts feature flag' do
+    let(:account) { create(:account) }
+    let(:inbox) { create(:inbox, account: account) }
+    let(:store) { Conversations::UnreadCounts::Store }
+    let(:inbox_key) { store.inbox_key(account.id, inbox.id) }
+
+    after do
+      store.clear_account!(account.id)
+    end
+
+    it 'clears unread count cache when the feature is enabled' do
+      build_unread_count_cache
+
+      account.enable_features!(:conversation_unread_counts)
+
+      expect(store.base_ready?(account.id)).to be(false)
+      expect(store.assignment_ready?(account.id)).to be(false)
+      expect(store.counts_for_keys([inbox_key])).to eq(inbox_key => 0)
+    end
+
+    it 'clears unread count cache when the feature is disabled' do
+      account.enable_features!(:conversation_unread_counts)
+      build_unread_count_cache
+
+      account.disable_features!(:conversation_unread_counts)
+
+      expect(store.base_ready?(account.id)).to be(false)
+      expect(store.assignment_ready?(account.id)).to be(false)
+      expect(store.counts_for_keys([inbox_key])).to eq(inbox_key => 0)
+    end
+
+    def build_unread_count_cache
+      store.mark_base_ready!(account.id)
+      store.mark_assignment_ready!(account.id)
+      store.add_base_membership(account_id: account.id, inbox_id: inbox.id, label_ids: [], conversation_id: 1)
+    end
+  end
+
   describe 'inbound_email_domain' do
     let(:account) { create(:account) }
 

spec/models/conversation_spec.rb

@@ -717,6 +717,25 @@ RSpec.describe Conversation do
 
       expect { notification.reload }.to raise_error ActiveRecord::RecordNotFound
     end
+
+    it 'dispatches conversation deleted event with unread count cache data' do
+      allow(Rails.configuration.dispatcher).to receive(:dispatch)
+
+      conversation.destroy!
+
+      expect(Rails.configuration.dispatcher).to have_received(:dispatch).with(
+        'conversation.deleted',
+        kind_of(Time),
+        conversation_data: {
+          id: conversation.id,
+          account_id: conversation.account_id,
+          inbox_id: conversation.inbox_id,
+          assignee_id: conversation.assignee_id,
+          team_id: conversation.team_id,
+          cached_label_list: conversation.cached_label_list
+        }
+      )
+    end
   end
 
   describe 'validate invalid referer url' do

spec/services/conversations/unread_counts/listener_spec.rb

@@ -0,0 +1,164 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Listener do
+  let(:listener) { described_class.instance }
+  let(:account) { create(:account) }
+  let(:conversation) { create(:conversation, account: account) }
+  let(:notifier) { instance_double(Conversations::UnreadCounts::Notifier, perform: true) }
+
+  before do
+    allow(Conversations::UnreadCounts::Notifier).to receive(:new).and_return(notifier)
+  end
+
+  it 'refreshes unread counts when an incoming message is created' do
+    account.enable_features!(:conversation_unread_counts)
+    message = create(:message, account: account, inbox: conversation.inbox, conversation: conversation, message_type: :incoming)
+    event = Events::Base.new('message.created', Time.zone.now, message: message)
+
+    listener.message_created(event)
+
+    expect(Conversations::UnreadCounts::Notifier).to have_received(:new).with(conversation, changed_attributes: nil)
+    expect(notifier).to have_received(:perform)
+  end
+
+  it 'ignores outgoing message creation' do
+    message = create(:message, account: account, inbox: conversation.inbox, conversation: conversation, message_type: :outgoing)
+    event = Events::Base.new('message.created', Time.zone.now, message: message)
+
+    listener.message_created(event)
+
+    expect(Conversations::UnreadCounts::Notifier).not_to have_received(:new)
+  end
+
+  it 'ignores incoming message creation when conversation unread counts are disabled' do
+    message = create(:message, account: account, inbox: conversation.inbox, conversation: conversation, message_type: :incoming)
+    event = Events::Base.new('message.created', Time.zone.now, message: message)
+
+    expect(message).not_to receive(:conversation)
+
+    listener.message_created(event)
+
+    expect(Conversations::UnreadCounts::Notifier).not_to have_received(:new)
+  end
+
+  it 'refreshes unread counts when conversation status changes' do
+    changed_attributes = { 'status' => %w[open resolved] }
+    event = Events::Base.new('conversation.status_changed', Time.zone.now, conversation: conversation, changed_attributes: changed_attributes)
+
+    listener.conversation_status_changed(event)
+
+    expect(Conversations::UnreadCounts::Notifier).to have_received(:new).with(conversation, changed_attributes: changed_attributes)
+    expect(notifier).to have_received(:perform)
+  end
+
+  it 'refreshes unread counts when labels change' do
+    changed_attributes = { label_list: [%w[old], %w[new]] }
+    event = Events::Base.new('conversation.updated', Time.zone.now, conversation: conversation, changed_attributes: changed_attributes)
+
+    listener.conversation_updated(event)
+
+    expect(Conversations::UnreadCounts::Notifier).to have_received(:new).with(conversation, changed_attributes: changed_attributes)
+    expect(notifier).to have_received(:perform)
+  end
+
+  it 'ignores conversation updates unrelated to unread count dimensions' do
+    event = Events::Base.new('conversation.updated', Time.zone.now, conversation: conversation, changed_attributes: { priority: [nil, 'high'] })
+
+    listener.conversation_updated(event)
+
+    expect(Conversations::UnreadCounts::Notifier).not_to have_received(:new)
+  end
+
+  it 'refreshes unread counts when assignee changes' do
+    changed_attributes = { assignee_id: [nil, 1] }
+    event = Events::Base.new('assignee.changed', Time.zone.now, conversation: conversation, changed_attributes: changed_attributes)
+
+    listener.assignee_changed(event)
+
+    expect(Conversations::UnreadCounts::Notifier).to have_received(:new).with(conversation, changed_attributes: changed_attributes)
+    expect(notifier).to have_received(:perform)
+  end
+
+  it 'refreshes unread counts when team changes' do
+    changed_attributes = { team_id: [nil, 1] }
+    event = Events::Base.new('team.changed', Time.zone.now, conversation: conversation, changed_attributes: changed_attributes)
+
+    listener.team_changed(event)
+
+    expect(Conversations::UnreadCounts::Notifier).to have_received(:new).with(conversation, changed_attributes: changed_attributes)
+    expect(notifier).to have_received(:perform)
+  end
+
+  it 'removes unread count memberships when a conversation is deleted' do
+    account.enable_features!(:conversation_unread_counts)
+    label = create(:label, account: account)
+    team = create(:team, account: account)
+    assignee = create(:user, account: account)
+    create(:team_member, team: team, user: assignee)
+    conversation.update!(assignee_id: assignee.id, team: team)
+    conversation.update_labels([label.title])
+    conversation.reload
+    conversation_data = deleted_conversation_data(conversation)
+    store.mark_base_ready!(account.id)
+    store.mark_assignment_ready!(account.id)
+    store.add_base_membership(
+      account_id: account.id,
+      inbox_id: conversation.inbox_id,
+      label_ids: [label.id],
+      team_id: team.id,
+      conversation_id: conversation.id
+    )
+    store.add_assignment_membership(
+      account_id: account.id,
+      inbox_id: conversation.inbox_id,
+      label_ids: [label.id],
+      assignee_id: assignee.id,
+      team_id: team.id,
+      conversation_id: conversation.id
+    )
+    allow(Rails.configuration.dispatcher).to receive(:dispatch)
+
+    listener.conversation_deleted(Events::Base.new('conversation.deleted', Time.zone.now, conversation_data: conversation_data))
+
+    expect(store.counts_for_keys(deleted_base_keys(conversation, label, team)).values).to all(eq(0))
+    expect(store.counts_for_keys(deleted_assignment_keys(conversation, label, team, assignee)).values).to all(eq(0))
+    expect(Rails.configuration.dispatcher).to have_received(:dispatch).with(
+      'conversation.unread_count_changed',
+      kind_of(Time),
+      conversation_data: conversation_data.stringify_keys
+    )
+  ensure
+    store.clear_account!(account.id)
+  end
+
+  def deleted_conversation_data(conversation)
+    {
+      id: conversation.id,
+      account_id: conversation.account_id,
+      inbox_id: conversation.inbox_id,
+      assignee_id: conversation.assignee_id,
+      team_id: conversation.team_id,
+      cached_label_list: conversation.cached_label_list
+    }
+  end
+
+  def deleted_base_keys(conversation, label, team)
+    [
+      store.inbox_key(account.id, conversation.inbox_id),
+      store.label_inbox_key(account.id, label.id, conversation.inbox_id),
+      store.team_inbox_key(account.id, team.id, conversation.inbox_id)
+    ]
+  end
+
+  def deleted_assignment_keys(conversation, label, team, assignee)
+    [
+      store.inbox_assignee_key(account.id, conversation.inbox_id, assignee.id),
+      store.label_inbox_assignee_key(account.id, label.id, conversation.inbox_id, assignee.id),
+      store.team_inbox_assignee_key(account.id, team.id, conversation.inbox_id, assignee.id)
+    ]
+  end
+
+  def store
+    Conversations::UnreadCounts::Store
+  end
+end

spec/services/conversations/unread_counts/notifier_spec.rb

@@ -0,0 +1,48 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Notifier do
+  let!(:conversation) { create(:conversation) }
+  let(:refresher) { instance_double(Conversations::UnreadCounts::Refresher, perform: refresh_result) }
+  let(:refresh_result) { true }
+
+  before do
+    conversation.account.enable_features!(:conversation_unread_counts)
+    allow(Conversations::UnreadCounts::Refresher).to receive(:new).and_return(refresher)
+    allow(Rails.configuration.dispatcher).to receive(:dispatch)
+  end
+
+  it 'dispatches unread count changed event after a successful refresh' do
+    described_class.new(conversation).perform
+
+    expect(Rails.configuration.dispatcher).to have_received(:dispatch).with(
+      'conversation.unread_count_changed',
+      kind_of(Time),
+      conversation: conversation
+    )
+  end
+
+  context 'when refresh does not change unread count memberships' do
+    let(:refresh_result) { false }
+
+    it 'does not dispatch unread count changed event' do
+      described_class.new(conversation).perform
+
+      expect(Rails.configuration.dispatcher).not_to have_received(:dispatch)
+    end
+  end
+
+  context 'when conversation unread counts feature is disabled' do
+    before do
+      conversation.account.disable_features!(:conversation_unread_counts)
+      allow(Conversations::UnreadCounts::Store).to receive(:clear_account!)
+    end
+
+    it 'does not refresh, clear cache, or dispatch unread count changed event' do
+      described_class.new(conversation).perform
+
+      expect(Conversations::UnreadCounts::Refresher).not_to have_received(:new)
+      expect(Conversations::UnreadCounts::Store).not_to have_received(:clear_account!)
+      expect(Rails.configuration.dispatcher).not_to have_received(:dispatch)
+    end
+  end
+end

spec/services/conversations/unread_counts/refresher_spec.rb

@@ -0,0 +1,145 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Refresher do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:label) { create(:label, account: account, title: 'urgent', show_on_sidebar: true) }
+  let(:new_label) { create(:label, account: account, title: 'billing', show_on_sidebar: true) }
+  let(:team) { create(:team, account: account, allow_auto_assign: false) }
+  let(:new_team) { create(:team, account: account, allow_auto_assign: false) }
+  let(:assignee) { create(:user, account: account, role: :agent) }
+  let(:other_assignee) { create(:user, account: account, role: :agent) }
+  let(:store) { Conversations::UnreadCounts::Store }
+
+  after do
+    store.clear_account!(account.id)
+  end
+
+  it 'does not update redis when unread caches are not ready' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title])
+
+    expect(described_class.new(conversation).perform).to be(false)
+    expect(store.counts_for_keys([store.inbox_key(account.id, inbox.id)])).to eq(store.inbox_key(account.id, inbox.id) => 0)
+  end
+
+  it 'adds an unread conversation to base cache' do
+    conversation = create(:conversation, account: account, inbox: inbox, agent_last_seen_at: 1.hour.ago)
+    conversation.update_labels([label.title])
+    store.mark_base_ready!(account.id)
+
+    create(:message, account: account, inbox: inbox, conversation: conversation, message_type: :incoming, created_at: 5.minutes.ago)
+
+    expect(described_class.new(conversation.reload).perform).to be(true)
+
+    expect(store.counts_for_keys(base_keys)).to eq(
+      store.inbox_key(account.id, inbox.id) => 1,
+      store.label_inbox_key(account.id, label.id, inbox.id) => 1
+    )
+  end
+
+  it 'returns false when refresh does not change unread counts' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title])
+    Conversations::UnreadCounts::Builder.new(account).build_base!
+
+    create(:message, account: account, inbox: inbox, conversation: conversation, message_type: :incoming)
+
+    expect(described_class.new(conversation.reload).perform).to be(false)
+    expect(store.counts_for_keys(base_keys)).to eq(
+      store.inbox_key(account.id, inbox.id) => 1,
+      store.label_inbox_key(account.id, label.id, inbox.id) => 1
+    )
+  end
+
+  it 'removes a conversation from base cache when it becomes read' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title])
+    Conversations::UnreadCounts::Builder.new(account).build_base!
+
+    conversation.update!(agent_last_seen_at: 1.minute.from_now)
+    expect(described_class.new(conversation.reload).perform).to be(true)
+
+    expect(store.counts_for_keys(base_keys).values).to all(eq(0))
+  end
+
+  it 'moves base label membership when labels change' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title])
+    Conversations::UnreadCounts::Builder.new(account).build_base!
+
+    conversation.update_labels([new_label.title])
+    expect(described_class.new(conversation.reload, changed_attributes: { label_list: [[label.title], [new_label.title]] }).perform).to be(true)
+
+    expect(store.counts_for_keys([
+                                   store.label_inbox_key(account.id, label.id, inbox.id),
+                                   store.label_inbox_key(account.id, new_label.id, inbox.id)
+                                 ])).to eq(
+                                   store.label_inbox_key(account.id, label.id, inbox.id) => 0,
+                                   store.label_inbox_key(account.id, new_label.id, inbox.id) => 1
+                                 )
+  end
+
+  it 'moves base team membership when team changes' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, team: team)
+    Conversations::UnreadCounts::Builder.new(account).build_base!
+
+    conversation.update!(team: new_team)
+    expect(described_class.new(conversation.reload, changed_attributes: { team_id: [team.id, new_team.id] }).perform).to be(true)
+
+    expect(store.counts_for_keys([
+                                   store.team_inbox_key(account.id, team.id, inbox.id),
+                                   store.team_inbox_key(account.id, new_team.id, inbox.id)
+                                 ])).to eq(
+                                   store.team_inbox_key(account.id, team.id, inbox.id) => 0,
+                                   store.team_inbox_key(account.id, new_team.id, inbox.id) => 1
+                                 )
+  end
+
+  it 'moves assignment-aware membership when assignee changes' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: assignee)
+    Conversations::UnreadCounts::Builder.new(account).build_assignment!
+
+    conversation.update!(assignee: other_assignee)
+    described_class.new(conversation.reload, changed_attributes: { assignee_id: [assignee.id, other_assignee.id] }).perform
+
+    expect(store.counts_for_keys([
+                                   store.inbox_assignee_key(account.id, inbox.id, assignee.id),
+                                   store.inbox_assignee_key(account.id, inbox.id, other_assignee.id)
+                                 ])).to eq(
+                                   store.inbox_assignee_key(account.id, inbox.id, assignee.id) => 0,
+                                   store.inbox_assignee_key(account.id, inbox.id, other_assignee.id) => 1
+                                 )
+  end
+
+  it 'does not remove unassigned membership when assignee did not change' do
+    conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: assignee)
+    Conversations::UnreadCounts::Builder.new(account).build_assignment!
+    allow(store).to receive(:remove_assignment_membership).and_call_original
+
+    conversation.update_labels([new_label.title])
+    described_class.new(conversation.reload, changed_attributes: { label_list: [[label.title], [new_label.title]] }).perform
+
+    expect(store).to have_received(:remove_assignment_membership).with(hash_including(assignee_ids: [assignee.id]))
+  end
+
+  it 'moves assignment-aware team membership when team changes' do
+    create(:team_member, user: assignee, team: new_team)
+    conversation = create_unread_conversation(account: account, inbox: inbox, assignee: assignee, team: team)
+    Conversations::UnreadCounts::Builder.new(account).build_assignment!
+
+    conversation.update!(team: new_team)
+    described_class.new(conversation.reload, changed_attributes: { team_id: [team.id, new_team.id] }).perform
+
+    expect(store.counts_for_keys([
+                                   store.team_inbox_assignee_key(account.id, team.id, inbox.id, assignee.id),
+                                   store.team_inbox_assignee_key(account.id, new_team.id, inbox.id, assignee.id)
+                                 ])).to eq(
+                                   store.team_inbox_assignee_key(account.id, team.id, inbox.id, assignee.id) => 0,
+                                   store.team_inbox_assignee_key(account.id, new_team.id, inbox.id, assignee.id) => 1
+                                 )
+  end
+
+  def base_keys
+    [
+      store.inbox_key(account.id, inbox.id),
+      store.label_inbox_key(account.id, label.id, inbox.id)
+    ]
+  end
+end

spec/services/conversations/unread_counts/store_spec.rb

@@ -90,6 +90,23 @@ RSpec.describe Conversations::UnreadCounts::Store do
       expect(described_class.counts_for_keys(base_keys).values).to all(eq(0))
     end
 
+    it 'checks memberships for a conversation across keys' do
+      described_class.add_base_membership(
+        account_id: account_id,
+        inbox_id: inbox_id,
+        label_ids: [label_id],
+        team_id: team_id,
+        conversation_id: conversation_id
+      )
+
+      expect(described_class.memberships_for_keys(base_keys, conversation_id)).to eq(
+        described_class.inbox_key(account_id, inbox_id) => true,
+        described_class.label_inbox_key(account_id, label_id, inbox_id) => true,
+        described_class.team_inbox_key(account_id, team_id, inbox_id) => true
+      )
+      expect(described_class.memberships_for_keys(base_keys, 999).values).to all(be(false))
+    end
+
     it 'adds, counts, and removes assignment-aware memberships' do
       described_class.add_assignment_membership(
         account_id: account_id,