feat: Add sidebar unread counts for filters (CW-7262) (#14726)

## Description Extends the conversation unread-count system so the left sidebar can show unread badges for Mentions, Participating, Unattended, and saved conversation folders. Folder badges reuse the existing `custom_filters` conversation filter semantics, store user-scoped Redis sets lazily, and skip unsupported folder filters so invalid saved folders continue to render without a badge. The Unattended badge counts all visible unread open conversations that match the existing unattended conversation scope. Closes - [CW-7262](https://linear.app/chatwoot/issue/CW-7262/unread-counts-for-filters-folders) ## What changed - Added user-scoped unread-count Redis keys and cache builders for mentions, participating conversations, unattended conversations, and saved folder filters. - Reused `Conversations::FilterService` through a relation-returning path so folder counts match the folder conversation list behavior. - Invalidated user filter caches from mention, participant, custom-filter, and relevant conversation update events. - Extended the unread-count endpoint payload and sidebar Vuex/sidebar rendering for the new badge counts, including the Unattended sidebar item. - Added Ruby, Enterprise, request, listener, and frontend store coverage for the new unread-count dimensions. ## Type of change - [x] New feature (non-breaking change which adds functionality) ## How Has This Been Tested? - Created local validation folders for `john@acme.inc` and confirmed the unread-count payload includes open, resolved, and high-priority folder badges while excluding the invalid unsupported folder. - Added coverage for the Unattended badge rule: all visible unread open conversations matching `Conversation.unattended`. - Ran focused unread-count Ruby specs, including service, listener, request, and Enterprise counter coverage. - Ran frontend unread-count store specs. - Ran RuboCop on the touched Ruby files. - Ran ESLint through the project script; it completed with warnings in existing unrelated files and no errors. <img width="369" height="525" alt="Screenshot 2026-06-13 at 10 51 39 PM" src="https://github.com/user-attachments/assets/36b1d2c4-dac1-4f6f-9c0e-7ef5a6cc2975" /> ## Checklist: - [x] My code follows the style guidelines of this project - [x] I have performed a self-review of my code - [x] I have commented on my code, particularly in hard-to-understand areas - [x] Documentation changes are not required for this internal unread-count behavior - [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 - [x] No dependent downstream changes are required --------- Co-authored-by: Muhsin Keloth <muhsinkeramam@gmail.com>
2026-06-16 23:46:49 +05:30
parent 352f120c6a
commit fe6368b42e
54 changed files with 1581 additions and 97 deletions
@@ -9,7 +9,7 @@ RSpec.describe Conversations::UnreadCounts::Builder do
  let(:store) { Conversations::UnreadCounts::Store }

  after do
-    store.clear_account!(account.id)
+    store.clear_all_account!(account.id)
  end

  describe '#build_base!' do
@@ -75,6 +75,173 @@ RSpec.describe Conversations::UnreadCounts::Builder do
    end
  end

+  describe '#build_filters_for!' do
+    before do
+      create(:inbox_member, user: assignee, inbox: inbox)
+    end
+
+    it 'stores unread open conversations by mentions and participating dimensions' do
+      mentioned_conversation = create_unread_conversation(account: account, inbox: inbox)
+      participating_conversation = create_unread_conversation(account: account, inbox: inbox)
+      resolved_mentioned_conversation = create_unread_conversation(account: account, inbox: inbox)
+      inaccessible_conversation = create_unread_conversation(account: account, inbox: create(:inbox, account: account))
+      resolved_mentioned_conversation.update!(status: :resolved)
+
+      create(:mention, account: account, conversation: mentioned_conversation, user: assignee)
+      create(:mention, account: account, conversation: resolved_mentioned_conversation, user: assignee)
+      create(:mention, account: account, conversation: inaccessible_conversation, user: assignee)
+      create(:conversation_participant, account: account, conversation: participating_conversation, user: assignee)
+
+      described_class.new(account).build_filters_for!(assignee)
+
+      expect(store.filters_ready?(account.id, assignee.id)).to be(true)
+      expect(redis_set_members(store.user_mentions_key(account.id, assignee.id))).to contain_exactly(mentioned_conversation.id.to_s)
+      expect(redis_set_members(store.user_participating_key(account.id, assignee.id))).to contain_exactly(participating_conversation.id.to_s)
+    end
+
+    it 'excludes participating conversations that are no longer visible to the user' do
+      participating_conversation = create_unread_conversation(account: account, inbox: inbox)
+      create(:conversation_participant, account: account, conversation: participating_conversation, user: assignee)
+      InboxMember.find_by!(user: assignee, inbox: inbox).destroy!
+
+      described_class.new(account).build_filters_for!(assignee)
+
+      expect(redis_set_members(store.user_participating_key(account.id, assignee.id))).to be_empty
+    end
+
+    it 'stores visible unread open unattended conversations' do
+      no_first_reply_conversation = create_unread_conversation(account: account, inbox: inbox)
+      waiting_conversation = create_unread_conversation(account: account, inbox: inbox)
+      attended_conversation = create_unread_conversation(account: account, inbox: inbox)
+      inaccessible_conversation = create_unread_conversation(account: account, inbox: create(:inbox, account: account))
+      resolved_conversation = create_unread_conversation(account: account, inbox: inbox)
+      create_read_conversation
+
+      waiting_conversation.update!(first_reply_created_at: 5.minutes.ago)
+      attended_conversation.update!(first_reply_created_at: 5.minutes.ago, waiting_since: nil)
+      inaccessible_conversation.update!(first_reply_created_at: nil)
+      resolved_conversation.update!(status: :resolved)
+
+      described_class.new(account).build_filters_for!(assignee)
+
+      expect(redis_set_members(store.user_unattended_key(account.id, assignee.id))).to contain_exactly(
+        no_first_reply_conversation.id.to_s,
+        waiting_conversation.id.to_s
+      )
+    end
+
+    it 'stores folder memberships using the saved filter status conditions' do
+      resolved_conversation = create_unread_conversation(account: account, inbox: inbox)
+      resolved_conversation.update!(status: :resolved)
+      create_unread_conversation(account: account, inbox: inbox, assignee: assignee)
+      custom_filter = create(
+        :custom_filter, account: account, user: assignee, filter_type: :conversation, query: filter_query('status', ['resolved'])
+      )
+
+      described_class.new(account).build_filters_for!(assignee)
+
+      expect(redis_set_members(store.user_folder_key(account.id, assignee.id, custom_filter.id))).to contain_exactly(resolved_conversation.id.to_s)
+    end
+
+    it 'loads folder filters after taking the invalidation version snapshot' do
+      create_unread_conversation(account: account, inbox: inbox)
+      resolved_conversation = create_unread_conversation(account: account, inbox: inbox)
+      resolved_conversation.update!(status: :resolved)
+      custom_filter = create(
+        :custom_filter, account: account, user: assignee, filter_type: :conversation, query: filter_query('status', ['open'])
+      )
+      notifier = instance_double(Conversations::UnreadCounts::UserFilterNotifier, perform: true)
+      allow(Conversations::UnreadCounts::UserFilterNotifier).to receive(:new).and_return(notifier)
+      filter_updated = false
+      allow(store).to receive(:filter_version_snapshot).and_wrap_original do |method, *args|
+        method.call(*args).tap do
+          next if filter_updated
+
+          filter_updated = true
+          custom_filter.update!(query: filter_query('status', ['resolved']))
+        end
+      end
+
+      described_class.new(account).build_filters_for!(assignee)
+
+      expect(redis_set_members(store.user_folder_key(account.id, assignee.id, custom_filter.id))).to contain_exactly(resolved_conversation.id.to_s)
+    end
+
+    it 'expires relative-date folder caches at the next date boundary' do
+      create(
+        :custom_filter,
+        account: account,
+        user: assignee,
+        filter_type: :conversation,
+        query: filter_query('created_at', [7], filter_operator: 'days_before')
+      )
+      allow(store).to receive(:mark_filters_ready_if_current!).and_call_original
+      expected_ttl = nil
+
+      travel_to Time.zone.local(2026, 1, 1, 9, 30, 0) do
+        expected_ttl = (Time.zone.tomorrow.beginning_of_day - Time.current).ceil
+        described_class.new(account).build_filters_for!(assignee)
+      end
+
+      expect(store).to have_received(:mark_filters_ready_if_current!).with(
+        account.id,
+        assignee.id,
+        version_snapshot: kind_of(Hash),
+        expires_in: expected_ttl
+      )
+    end
+
+    it 'does not mark filters ready when user filters are invalidated during the build' do
+      conversation = create_unread_conversation(account: account, inbox: inbox)
+      create(:mention, account: account, conversation: conversation, user: assignee)
+      clear_user_filters_after_membership_write
+
+      described_class.new(account).build_filters_for!(assignee)
+
+      expect(store.filters_ready?(account.id, assignee.id)).to be(false)
+      expect(redis_set_members(store.user_mentions_key(account.id, assignee.id))).to be_empty
+    end
+
+    it 'does not mark filters ready when account filters are invalidated during the build' do
+      conversation = create_unread_conversation(account: account, inbox: inbox)
+      create(:mention, account: account, conversation: conversation, user: assignee)
+      clear_filter_caches_after_membership_write
+
+      described_class.new(account).build_filters_for!(assignee)
+
+      expect(store.filters_ready?(account.id, assignee.id)).to be(false)
+      expect(redis_set_members(store.user_mentions_key(account.id, assignee.id))).to be_empty
+    end
+
+    it 'skips invalid folder filters and still marks the user filter cache ready' do
+      create_unread_conversation(account: account, inbox: inbox)
+      invalid_filter = create(
+        :custom_filter, account: account, user: assignee, filter_type: :conversation, query: filter_query('missing_attribute', ['open'])
+      )
+
+      described_class.new(account).build_filters_for!(assignee)
+
+      expect(store.filters_ready?(account.id, assignee.id)).to be(true)
+      expect(redis_set_members(store.user_folder_key(account.id, assignee.id, invalid_filter.id))).to be_empty
+    end
+
+    it 'skips folder filters that fail when the SQL query is executed' do
+      conversation = create_unread_conversation(account: account, inbox: inbox)
+      invalid_filter = create(
+        :custom_filter,
+        account: account,
+        user: assignee,
+        filter_type: :conversation,
+        query: filter_query('display_id', [conversation.display_id.to_s], filter_operator: 'contains')
+      )
+
+      expect { described_class.new(account).build_filters_for!(assignee) }.not_to raise_error
+
+      expect(store.filters_ready?(account.id, assignee.id)).to be(true)
+      expect(redis_set_members(store.user_folder_key(account.id, assignee.id, invalid_filter.id))).to be_empty
+    end
+  end
+
  def create_read_conversation
    conversation = create(:conversation, account: account, inbox: inbox, agent_last_seen_at: 1.minute.from_now)
    create(:message, account: account, inbox: inbox, conversation: conversation, message_type: :incoming)
@@ -90,4 +257,32 @@ RSpec.describe Conversations::UnreadCounts::Builder do
  def redis_set_members(key)
    Redis::Alfred.pipelined { |pipeline| pipeline.smembers(key) }.first
  end
+
+  def clear_user_filters_after_membership_write
+    allow(store).to receive(:add_filter_memberships).and_wrap_original do |method, *args, **kwargs|
+      method.call(*args, **kwargs)
+      store.clear_user_filters!(account.id, assignee.id)
+    end
+  end
+
+  def clear_filter_caches_after_membership_write
+    allow(store).to receive(:add_filter_memberships).and_wrap_original do |method, *args, **kwargs|
+      method.call(*args, **kwargs)
+      store.clear_filter_caches!(account.id)
+    end
+  end
+
+  def filter_query(attribute_key, values, filter_operator: 'equal_to')
+    {
+      payload: [
+        {
+          attribute_key: attribute_key,
+          filter_operator: filter_operator,
+          values: values,
+          query_operator: nil,
+          custom_attribute_type: ''
+        }
+      ]
+    }
+  end
 end
@@ -17,7 +17,7 @@ RSpec.describe Conversations::UnreadCounts::Counter do
  end

  after do
-    store.clear_account!(account.id)
+    store.clear_all_account!(account.id)
  end

  it 'builds the base cache on demand' do
@@ -32,6 +32,7 @@ RSpec.describe Conversations::UnreadCounts::Counter do
    lock_key = "UNREAD_CONVERSATIONS::V1::ACCOUNT::#{account.id}::BUILD_LOCK::BASE"
    lock_manager = instance_double(Redis::LockManager)
    allow(Redis::LockManager).to receive(:new).and_return(lock_manager)
+    allow(lock_manager).to receive(:with_lock).and_yield.and_return(true)
    allow(lock_manager).to receive(:with_lock).with(lock_key, described_class::BUILD_LOCK_TTL).and_yield.and_return(true)

    create_unread_conversation(account: account, inbox: visible_inbox, labels: [label.title], team: visible_team)
@@ -46,13 +47,33 @@ RSpec.describe Conversations::UnreadCounts::Counter do
    counter = described_class.new(account: account, user: agent)

    allow(Redis::LockManager).to receive(:new).and_return(lock_manager)
-    allow(counter).to receive(:wait_for_cache_ready) { store.mark_base_ready!(account.id) }
+    allow(counter).to receive(:wait_for_cache_ready) do
+      store.mark_base_ready!(account.id)
+      store.mark_filters_ready!(account.id, agent.id)
+    end
    expect(Conversations::UnreadCounts::Builder).not_to receive(:new)

    counter.perform

    expect(counter).to have_received(:wait_for_cache_ready)
    expect(store.base_ready?(account.id)).to be(true)
+    expect(store.filters_ready?(account.id, agent.id)).to be(true)
+  end
+
+  it 'retries when a build finishes without marking the cache ready' do
+    builder = instance_double(Conversations::UnreadCounts::Builder)
+    attempts = 0
+    allow(Conversations::UnreadCounts::Builder).to receive(:new).and_return(builder)
+    allow(builder).to receive(:build_base!) do
+      attempts += 1
+      store.mark_base_ready!(account.id) if attempts == 2
+    end
+    allow(builder).to receive(:build_filters_for!) { store.mark_filters_ready!(account.id, agent.id) }
+
+    described_class.new(account: account, user: agent).perform
+
+    expect(builder).to have_received(:build_base!).twice
+    expect(store.base_ready?(account.id)).to be(true)
  end

  it 'counts unread conversations only across inboxes visible to a normal agent' do
@@ -65,7 +86,11 @@ RSpec.describe Conversations::UnreadCounts::Counter do
      all_count: 1,
      inboxes: { visible_inbox.id.to_s => 1 },
      labels: { label.id.to_s => 1 },
-      teams: { visible_team.id.to_s => 1 }
+      teams: { visible_team.id.to_s => 1 },
+      mentions_count: 0,
+      participating_count: 0,
+      unattended_count: 1,
+      folders: {}
    )
  end

@@ -79,7 +104,11 @@ RSpec.describe Conversations::UnreadCounts::Counter do
      all_count: 2,
      inboxes: { visible_inbox.id.to_s => 1, hidden_inbox.id.to_s => 1 },
      labels: { label.id.to_s => 2 },
-      teams: { visible_team.id.to_s => 2 }
+      teams: { visible_team.id.to_s => 2 },
+      mentions_count: 0,
+      participating_count: 0,
+      unattended_count: 2,
+      folders: {}
    )
  end

@@ -92,7 +121,46 @@ RSpec.describe Conversations::UnreadCounts::Counter do
      all_count: 1,
      inboxes: { visible_inbox.id.to_s => 1 },
      labels: {},
-      teams: { visible_team.id.to_s => 1 }
+      teams: { visible_team.id.to_s => 1 },
+      mentions_count: 0,
+      participating_count: 0,
+      unattended_count: 1,
+      folders: {}
    )
  end
+
+  it 'returns mention, participating, unattended, and valid folder unread counts for the user' do
+    mentioned_conversation = create_unread_conversation(account: account, inbox: visible_inbox)
+    participating_conversation = create_unread_conversation(account: account, inbox: visible_inbox)
+    resolved_conversation = create_unread_conversation(account: account, inbox: visible_inbox)
+    resolved_conversation.update!(status: :resolved)
+    valid_folder = create(:custom_filter, account: account, user: agent, filter_type: :conversation, query: filter_query('status', ['resolved']))
+    invalid_folder = create(:custom_filter, account: account, user: agent, filter_type: :conversation, query: filter_query('unknown', ['open']))
+
+    create(:mention, account: account, conversation: mentioned_conversation, user: agent)
+    create(:conversation_participant, account: account, conversation: participating_conversation, user: agent)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result[:mentions_count]).to eq(1)
+    expect(result[:participating_count]).to eq(1)
+    expect(result[:unattended_count]).to eq(2)
+    expect(result[:folders]).to eq(valid_folder.id.to_s => 1)
+    expect(result[:folders]).not_to have_key(invalid_folder.id.to_s)
+    expect(store.filters_ready?(account.id, agent.id)).to be(true)
+  end
+
+  def filter_query(attribute_key, values)
+    {
+      payload: [
+        {
+          attribute_key: attribute_key,
+          filter_operator: 'equal_to',
+          values: values,
+          query_operator: nil,
+          custom_attribute_type: ''
+        }
+      ]
+    }
+  end
 end
@@ -21,13 +21,22 @@ RSpec.describe Conversations::UnreadCounts::Listener do
    expect(notifier).to have_received(:perform)
  end

-  it 'ignores outgoing message creation' do
+  it 'clears user filter counts when a non-incoming message updates last activity' do
+    account.enable_features!(:conversation_unread_counts)
    message = create(:message, account: account, inbox: conversation.inbox, conversation: conversation, message_type: :outgoing)
    event = Events::Base.new('message.created', Time.zone.now, message: message)
+    allow(store).to receive(:clear_filter_caches!).and_return(true)
+    allow(Rails.configuration.dispatcher).to receive(:dispatch)

    listener.message_created(event)

    expect(Conversations::UnreadCounts::Notifier).not_to have_received(:new)
+    expect(store).to have_received(:clear_filter_caches!).with(account.id)
+    expect(Rails.configuration.dispatcher).to have_received(:dispatch).with(
+      'conversation.unread_count_changed',
+      kind_of(Time),
+      conversation: conversation
+    )
  end

  it 'ignores incoming message creation when conversation unread counts are disabled' do
@@ -52,6 +61,7 @@ RSpec.describe Conversations::UnreadCounts::Listener do
  end

  it 'refreshes unread counts when labels change' do
+    account.enable_features!(:conversation_unread_counts)
    changed_attributes = { label_list: [%w[old], %w[new]] }
    event = Events::Base.new('conversation.updated', Time.zone.now, conversation: conversation, changed_attributes: changed_attributes)

@@ -61,8 +71,43 @@ RSpec.describe Conversations::UnreadCounts::Listener do
    expect(notifier).to have_received(:perform)
  end

-  it 'ignores conversation updates unrelated to unread count dimensions' do
+  it 'clears user filter counts when a folder filter dimension changes' do
+    account.enable_features!(:conversation_unread_counts)
    event = Events::Base.new('conversation.updated', Time.zone.now, conversation: conversation, changed_attributes: { priority: [nil, 'high'] })
+    allow(store).to receive(:clear_filter_caches!).and_return(true)
+    allow(Rails.configuration.dispatcher).to receive(:dispatch)
+
+    listener.conversation_updated(event)
+
+    expect(Conversations::UnreadCounts::Notifier).not_to have_received(:new)
+    expect(store).to have_received(:clear_filter_caches!).with(account.id)
+    expect(Rails.configuration.dispatcher).to have_received(:dispatch).with(
+      'conversation.unread_count_changed',
+      kind_of(Time),
+      conversation: conversation
+    )
+  end
+
+  it 'clears user filter counts when the conversation contact changes' do
+    account.enable_features!(:conversation_unread_counts)
+    event = Events::Base.new('conversation.contact_changed', Time.zone.now, conversation: conversation)
+    allow(store).to receive(:clear_filter_caches!).and_return(true)
+    allow(Rails.configuration.dispatcher).to receive(:dispatch)
+
+    listener.conversation_contact_changed(event)
+
+    expect(Conversations::UnreadCounts::Notifier).not_to have_received(:new)
+    expect(store).to have_received(:clear_filter_caches!).with(account.id)
+    expect(Rails.configuration.dispatcher).to have_received(:dispatch).with(
+      'conversation.unread_count_changed',
+      kind_of(Time),
+      conversation: conversation
+    )
+  end
+
+  it 'ignores conversation updates without changed attributes' do
+    account.enable_features!(:conversation_unread_counts)
+    event = Events::Base.new('conversation.updated', Time.zone.now, conversation: conversation, changed_attributes: {})

    listener.conversation_updated(event)

@@ -128,7 +173,7 @@ RSpec.describe Conversations::UnreadCounts::Listener do
      conversation_data: conversation_data.stringify_keys
    )
  ensure
-    store.clear_account!(account.id)
+    store.clear_all_account!(account.id)
  end

  def deleted_conversation_data(conversation)
@@ -29,6 +29,18 @@ RSpec.describe Conversations::UnreadCounts::Notifier do

      expect(Rails.configuration.dispatcher).not_to have_received(:dispatch)
    end
+
+    it 'dispatches unread count changed event when user filter caches were cleared' do
+      allow(Conversations::UnreadCounts::Store).to receive(:clear_filter_caches!).and_return(true)
+
+      described_class.new(conversation).perform
+
+      expect(Rails.configuration.dispatcher).to have_received(:dispatch).with(
+        'conversation.unread_count_changed',
+        kind_of(Time),
+        conversation: conversation
+      )
+    end
  end

  context 'when conversation unread counts feature is disabled' do
@@ -12,7 +12,7 @@ RSpec.describe Conversations::UnreadCounts::Refresher do
  let(:store) { Conversations::UnreadCounts::Store }

  after do
-    store.clear_account!(account.id)
+    store.clear_all_account!(account.id)
  end

  it 'does not update redis when unread caches are not ready' do
@@ -7,9 +7,10 @@ RSpec.describe Conversations::UnreadCounts::Store do
  let(:user_id) { 4 }
  let(:conversation_id) { 5 }
  let(:team_id) { 6 }
+  let(:other_user_id) { 8 }

  after do
-    described_class.clear_account!(account_id)
+    described_class.clear_all_account!(account_id)
  end

  describe 'key builders' do
@@ -45,20 +46,69 @@ RSpec.describe Conversations::UnreadCounts::Store do
        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::TEAM::6::INBOX::2::ASSIGNEE::4'
      )
    end
+
+    it 'builds user filter keys using the Redis key naming convention' do
+      expect(described_class.user_mentions_key(account_id, user_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::USER::4::MENTIONS'
+      )
+      expect(described_class.user_participating_key(account_id, user_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::USER::4::PARTICIPATING'
+      )
+      expect(described_class.user_unattended_key(account_id, user_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::USER::4::UNATTENDED'
+      )
+      expect(described_class.user_folder_key(account_id, user_id, 7)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::USER::4::FOLDER::7'
+      )
+    end
  end

  describe 'ready markers' do
-    it 'tracks base and assignment readiness independently' do
+    it 'starts with all ready markers missing' do
      expect(described_class.base_ready?(account_id)).to be(false)
      expect(described_class.assignment_ready?(account_id)).to be(false)
+      expect(described_class.filters_ready?(account_id, user_id)).to be(false)
+    end

+    it 'tracks base, assignment, and user filter readiness independently' do
      described_class.mark_base_ready!(account_id)
      described_class.mark_assignment_ready!(account_id)
+      described_class.mark_filters_ready!(account_id, user_id)

      expect(described_class.base_ready?(account_id)).to be(true)
      expect(described_class.assignment_ready?(account_id)).to be(true)
+      expect(described_class.filters_ready?(account_id, user_id)).to be(true)
      expect(ttl_for('UNREAD_CONVERSATIONS::V1::ACCOUNT::1::READY::BASE')).to be_within(5).of(Conversations::UnreadCounts::READY_TTL)
      expect(ttl_for('UNREAD_CONVERSATIONS::V1::ACCOUNT::1::READY::ASSIGNMENT')).to be_within(5).of(Conversations::UnreadCounts::READY_TTL)
+      expect(ttl_for('UNREAD_CONVERSATIONS::V1::ACCOUNT::1::USER::4::READY::FILTERS')).to be_within(5).of(
+        Conversations::UnreadCounts::READY_TTL
+      )
+    end
+
+    it 'tracks filter invalidation versions independently' do
+      expect(described_class.filter_version_snapshot(account_id, user_id)).to eq(account: 0, user: 0)
+
+      expect(described_class.clear_user_filters!(account_id, user_id)).to be(false)
+
+      expect(described_class.filter_version_snapshot(account_id, user_id)).to eq(account: 0, user: 1)
+      expect(ttl_for(user_filter_version_key)).to be_within(5).of(Conversations::UnreadCounts::SET_TTL)
+
+      expect(described_class.clear_filter_caches!(account_id)).to be(false)
+
+      expect(described_class.filter_version_snapshot(account_id, user_id)).to eq(account: 1, user: 1)
+      expect(ttl_for(account_filter_version_key)).to be_within(5).of(Conversations::UnreadCounts::SET_TTL)
+    end
+
+    it 'marks filter caches ready only when the invalidation version is current' do
+      version_snapshot = described_class.filter_version_snapshot(account_id, user_id)
+
+      expect(described_class.mark_filters_ready_if_current!(account_id, user_id, version_snapshot: version_snapshot)).to be_truthy
+      expect(described_class.filters_ready?(account_id, user_id)).to be(true)
+
+      described_class.clear_user_filters!(account_id, user_id)
+
+      expect(described_class.mark_filters_ready_if_current!(account_id, user_id, version_snapshot: version_snapshot)).to be(false)
+      expect(described_class.filters_ready?(account_id, user_id)).to be(false)
    end
  end

@@ -150,9 +200,94 @@ RSpec.describe Conversations::UnreadCounts::Store do
      expect(base_keys.map { |key| ttl_for(key) }).to all(be_within(5).of(Conversations::UnreadCounts::SET_TTL))
    end

-    it 'clears all account memberships' do
+    it 'adds, counts, and clears user filter memberships' do
+      described_class.add_filter_memberships(
+        account_id: account_id,
+        user_id: user_id,
+        filters: {
+          mentions: [conversation_id],
+          participating: [conversation_id],
+          unattended: [conversation_id]
+        },
+        folders: { 7 => [conversation_id] }
+      )
+      described_class.mark_filters_ready!(account_id, user_id)
+
+      expect(described_class.counts_for_keys(user_filter_keys)).to eq(
+        described_class.user_mentions_key(account_id, user_id) => 1,
+        described_class.user_participating_key(account_id, user_id) => 1,
+        described_class.user_unattended_key(account_id, user_id) => 1,
+        described_class.user_folder_key(account_id, user_id, 7) => 1
+      )
+      expect(user_filter_keys.map { |key| ttl_for(key) }).to all(be_within(5).of(Conversations::UnreadCounts::SET_TTL))
+
+      expect(described_class.clear_user_filters!(account_id, user_id)).to be(true)
+
+      expect(described_class.filters_ready?(account_id, user_id)).to be(false)
+      expect(described_class.counts_for_keys(user_filter_keys).values).to all(eq(0))
+    end
+
+    it 'preserves the user filter build lock when clearing one user filter cache' do
+      described_class.add_filter_memberships(
+        account_id: account_id,
+        user_id: user_id,
+        filters: {
+          mentions: [conversation_id],
+          participating: [conversation_id],
+          unattended: [conversation_id]
+        },
+        folders: { 7 => [conversation_id] }
+      )
+      described_class.mark_filters_ready!(account_id, user_id)
+      Redis::Alfred.set(user_filter_build_lock_key, 'locked')
+
+      expect(described_class.clear_user_filters!(account_id, user_id)).to be(true)
+
+      expect(Redis::Alfred.exists?(user_filter_build_lock_key)).to be(true)
+      expect(described_class.filters_ready?(account_id, user_id)).to be(false)
+      expect(described_class.counts_for_keys(user_filter_keys).values).to all(eq(0))
+    end
+
+    it 'preserves user filter build locks when clearing all account filter caches' do
+      described_class.add_filter_memberships(
+        account_id: account_id,
+        user_id: user_id,
+        filters: {
+          mentions: [conversation_id],
+          participating: [],
+          unattended: []
+        },
+        folders: {}
+      )
+      described_class.add_filter_memberships(
+        account_id: account_id,
+        user_id: other_user_id,
+        filters: {
+          mentions: [conversation_id],
+          participating: [],
+          unattended: []
+        },
+        folders: {}
+      )
+      described_class.mark_filters_ready!(account_id, user_id)
+      described_class.mark_filters_ready!(account_id, other_user_id)
+      Redis::Alfred.set(user_filter_build_lock_key, 'locked')
+      Redis::Alfred.set(user_filter_build_lock_key(other_user_id), 'locked')
+
+      expect(described_class.clear_filter_caches!(account_id)).to be(true)
+
+      expect(Redis::Alfred.exists?(user_filter_build_lock_key)).to be(true)
+      expect(Redis::Alfred.exists?(user_filter_build_lock_key(other_user_id))).to be(true)
+      expect(described_class.filters_ready?(account_id, user_id)).to be(false)
+      expect(described_class.filters_ready?(account_id, other_user_id)).to be(false)
+      expect(described_class.counts_for_keys([described_class.user_mentions_key(account_id, user_id)]).values).to all(eq(0))
+      expect(described_class.counts_for_keys([described_class.user_mentions_key(account_id, other_user_id)]).values).to all(eq(0))
+    end
+
+    it 'clears account memberships without clearing user filter memberships' do
      described_class.mark_base_ready!(account_id)
      described_class.mark_assignment_ready!(account_id)
+      described_class.mark_filters_ready!(account_id, user_id)
      described_class.add_base_membership(
        account_id: account_id,
        inbox_id: inbox_id,
@@ -168,13 +303,69 @@ RSpec.describe Conversations::UnreadCounts::Store do
        team_id: team_id,
        conversation_id: conversation_id
      )
+      described_class.add_filter_memberships(
+        account_id: account_id,
+        user_id: user_id,
+        filters: {
+          mentions: [conversation_id],
+          participating: [],
+          unattended: []
+        },
+        folders: {}
+      )
+      Redis::Alfred.set(user_filter_build_lock_key, 'locked')

      described_class.clear_account!(account_id)

      expect(described_class.base_ready?(account_id)).to be(false)
      expect(described_class.assignment_ready?(account_id)).to be(false)
+      expect(described_class.filters_ready?(account_id, user_id)).to be(true)
      expect(described_class.counts_for_keys(base_keys).values).to all(eq(0))
      expect(described_class.counts_for_keys(assignment_keys).values).to all(eq(0))
+      expect(described_class.counts_for_keys([described_class.user_mentions_key(account_id, user_id)]).values).to all(eq(1))
+      expect(Redis::Alfred.exists?(user_filter_build_lock_key)).to be(true)
+    end
+
+    it 'clears all account unread count keys' do
+      described_class.mark_base_ready!(account_id)
+      described_class.mark_assignment_ready!(account_id)
+      described_class.mark_filters_ready!(account_id, user_id)
+      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
+      )
+      described_class.add_assignment_membership(
+        account_id: account_id,
+        inbox_id: inbox_id,
+        label_ids: [label_id],
+        assignee_id: user_id,
+        team_id: team_id,
+        conversation_id: conversation_id
+      )
+      described_class.add_filter_memberships(
+        account_id: account_id,
+        user_id: user_id,
+        filters: {
+          mentions: [conversation_id],
+          participating: [],
+          unattended: []
+        },
+        folders: {}
+      )
+      Redis::Alfred.set(user_filter_build_lock_key, 'locked')
+
+      described_class.clear_all_account!(account_id)
+
+      expect(described_class.base_ready?(account_id)).to be(false)
+      expect(described_class.assignment_ready?(account_id)).to be(false)
+      expect(described_class.filters_ready?(account_id, user_id)).to be(false)
+      expect(described_class.counts_for_keys(base_keys).values).to all(eq(0))
+      expect(described_class.counts_for_keys(assignment_keys).values).to all(eq(0))
+      expect(described_class.counts_for_keys([described_class.user_mentions_key(account_id, user_id)]).values).to all(eq(0))
+      expect(Redis::Alfred.exists?(user_filter_build_lock_key)).to be(false)
    end
  end

@@ -194,6 +385,27 @@ RSpec.describe Conversations::UnreadCounts::Store do
    ]
  end

+  def user_filter_keys(filter_user_id = user_id)
+    [
+      described_class.user_mentions_key(account_id, filter_user_id),
+      described_class.user_participating_key(account_id, filter_user_id),
+      described_class.user_unattended_key(account_id, filter_user_id),
+      described_class.user_folder_key(account_id, filter_user_id, 7)
+    ]
+  end
+
+  def user_filter_build_lock_key(filter_user_id = user_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_USER_FILTERS_BUILD_LOCK, account_id: account_id, user_id: filter_user_id)
+  end
+
+  def account_filter_version_key
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_FILTERS_VERSION, account_id: account_id)
+  end
+
+  def user_filter_version_key(filter_user_id = user_id)
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_USER_FILTERS_VERSION, account_id: account_id, user_id: filter_user_id)
+  end
+
  def ttl_for(key)
    Redis::Alfred.ttl(key)
  end
@@ -0,0 +1,37 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::UserFilterNotifier do
+  let(:account) { create(:account) }
+  let(:user) { create(:user, account: account) }
+  let(:store) { Conversations::UnreadCounts::Store }
+
+  after do
+    store.clear_all_account!(account.id)
+  end
+
+  it 'clears the user filter cache and dispatches an unread count refresh event' do
+    account.enable_features!(:conversation_unread_counts)
+    store.mark_filters_ready!(account.id, user.id)
+    allow(Rails.configuration.dispatcher).to receive(:dispatch)
+
+    described_class.new(account: account, user: user).perform
+
+    expect(store.filters_ready?(account.id, user.id)).to be(false)
+    expect(Rails.configuration.dispatcher).to have_received(:dispatch).with(
+      'conversation.unread_count_changed',
+      kind_of(Time),
+      account: account,
+      user: user
+    )
+  end
+
+  it 'does nothing when conversation unread counts are disabled' do
+    store.mark_filters_ready!(account.id, user.id)
+    allow(Rails.configuration.dispatcher).to receive(:dispatch)
+
+    described_class.new(account: account, user: user).perform
+
+    expect(store.filters_ready?(account.id, user.id)).to be(true)
+    expect(Rails.configuration.dispatcher).not_to have_received(:dispatch)
+  end
+end