feat: base layer for unread counts (store, counter and builder) (1/3)[CW-6851] (#14368)

## Description

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

In this PR:

* Added the unread store, counter and builder modules
* Added redis keys for unread count management
* Added specs for all 3 modules, some specs are for testing enterprise
only feature like specific roles and permissions which are added in the
respective enterprise folder itself.

**Note**
None of this changes affect anything else and nothing is wired to
existing modules.

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/services/conversations/unread_counts.rb

@@ -0,0 +1,4 @@
+module Conversations::UnreadCounts
+  READY_TTL = 24.hours.to_i
+  SET_TTL = 25.hours.to_i
+end

app/services/conversations/unread_counts/builder.rb

@@ -0,0 +1,75 @@
+class Conversations::UnreadCounts::Builder
+  BATCH_SIZE = 1000
+
+  attr_reader :account
+
+  def initialize(account)
+    @account = account
+  end
+
+  def build_base!
+    store.clear_account!(account.id)
+    write_memberships(assignment: false)
+    store.mark_base_ready!(account.id)
+  end
+
+  def build_assignment!
+    store.clear_assignment!(account.id)
+    write_memberships(assignment: true)
+    store.mark_assignment_ready!(account.id)
+  end
+
+  def build_all!
+    build_base!
+    build_assignment!
+  end
+
+  private
+
+  def write_memberships(assignment:)
+    unread_conversations.in_batches(of: BATCH_SIZE) do |relation|
+      columns = %i[id inbox_id assignee_id cached_label_list team_id]
+      memberships = relation.pluck(*columns).map do |id, inbox_id, assignee_id, cached_label_list, team_id|
+        {
+          conversation_id: id,
+          inbox_id: inbox_id,
+          assignee_id: assignee_id,
+          team_id: team_id,
+          label_ids: label_ids_for(cached_label_list)
+        }
+      end
+
+      store.add_memberships(account_id: account.id, memberships: memberships, assignment: assignment)
+    end
+  end
+
+  def unread_conversations
+    account.conversations
+           .open
+           .joins(:messages)
+           .merge(Message.incoming.reorder(nil))
+           .where(messages: { account_id: account.id })
+           .where(unread_since_last_seen_condition)
+           .distinct
+  end
+
+  def unread_since_last_seen_condition
+    conversations = Conversation.arel_table
+    messages = Message.arel_table
+
+    conversations[:agent_last_seen_at].eq(nil).or(messages[:created_at].gt(conversations[:agent_last_seen_at]))
+  end
+
+  def label_ids_for(cached_label_list)
+    label_titles = cached_label_list.to_s.split(',').map(&:strip).compact_blank
+    labels_by_title.values_at(*label_titles).compact
+  end
+
+  def labels_by_title
+    @labels_by_title ||= account.labels.pluck(:title, :id).to_h
+  end
+
+  def store
+    ::Conversations::UnreadCounts::Store
+  end
+end

app/services/conversations/unread_counts/counter.rb

@@ -0,0 +1,200 @@
+class Conversations::UnreadCounts::Counter
+  MANAGE_ALL_PERMISSION = 'conversation_manage'.freeze
+  UNASSIGNED_PERMISSION = 'conversation_unassigned_manage'.freeze
+  PARTICIPATING_PERMISSION = 'conversation_participating_manage'.freeze
+  BUILD_LOCK_TTL = 15.minutes.to_i
+  BUILD_WAIT_TIMEOUT = 30.seconds.to_i
+  BUILD_WAIT_INTERVAL = 0.1.seconds
+
+  attr_reader :account, :user
+
+  def initialize(account:, user:)
+    @account = account
+    @user = user
+  end
+
+  def perform
+    return empty_counts if permission_mode == :none
+
+    ensure_base_cache!
+    ensure_assignment_cache! if assignment_mode?
+
+    {
+      inboxes: unread_inbox_counts,
+      labels: unread_label_counts,
+      teams: unread_team_counts
+    }
+  end
+
+  private
+
+  def ensure_base_cache!
+    ensure_cache_ready!(
+      ready: -> { store.base_ready?(account.id) },
+      lock_key: base_build_lock_key
+    ) { ::Conversations::UnreadCounts::Builder.new(account).build_base! }
+  end
+
+  def ensure_assignment_cache!
+    ensure_cache_ready!(
+      ready: -> { store.assignment_ready?(account.id) },
+      lock_key: assignment_build_lock_key
+    ) { ::Conversations::UnreadCounts::Builder.new(account).build_assignment! }
+  end
+
+  def ensure_cache_ready!(ready:, lock_key:)
+    lock_manager = Redis::LockManager.new
+
+    loop do
+      return if ready.call
+
+      return if lock_manager.with_lock(lock_key, BUILD_LOCK_TTL) { yield unless ready.call }
+
+      wait_for_cache_ready(ready)
+    end
+  end
+
+  def wait_for_cache_ready(ready)
+    deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + BUILD_WAIT_TIMEOUT
+    sleep BUILD_WAIT_INTERVAL until ready.call || Process.clock_gettime(Process::CLOCK_MONOTONIC) >= deadline
+  end
+
+  def base_build_lock_key
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_BASE_BUILD_LOCK, account_id: account.id)
+  end
+
+  def assignment_build_lock_key
+    format(Redis::Alfred::UNREAD_CONVERSATIONS_ASSIGNMENT_BUILD_LOCK, account_id: account.id)
+  end
+
+  def unread_inbox_counts
+    counts_for_grouped_keys(visible_inbox_ids.index_with { |inbox_id| inbox_keys_for_mode(inbox_id) })
+  end
+
+  def unread_label_counts
+    keys_by_id = Hash.new { |hash, key| hash[key] = [] }
+    sidebar_label_ids.each do |label_id|
+      visible_inbox_ids.each do |inbox_id|
+        keys_by_id[label_id].concat(label_inbox_keys_for_mode(label_id, inbox_id))
+      end
+    end
+
+    counts_for_grouped_keys(keys_by_id)
+  end
+
+  def unread_team_counts
+    keys_by_id = Hash.new { |hash, key| hash[key] = [] }
+    visible_team_ids.each do |team_id|
+      visible_inbox_ids.each do |inbox_id|
+        keys_by_id[team_id].concat(team_inbox_keys_for_mode(team_id, inbox_id))
+      end
+    end
+
+    counts_for_grouped_keys(keys_by_id)
+  end
+
+  def inbox_keys_for_mode(inbox_id)
+    case permission_mode
+    when :base
+      [store.inbox_key(account.id, inbox_id)]
+    when :unassigned_and_mine
+      [store.inbox_unassigned_key(account.id, inbox_id), store.inbox_assignee_key(account.id, inbox_id, user.id)]
+    when :mine
+      [store.inbox_assignee_key(account.id, inbox_id, user.id)]
+    end
+  end
+
+  def label_inbox_keys_for_mode(label_id, inbox_id)
+    case permission_mode
+    when :base
+      [store.label_inbox_key(account.id, label_id, inbox_id)]
+    when :unassigned_and_mine
+      [
+        store.label_inbox_unassigned_key(account.id, label_id, inbox_id),
+        store.label_inbox_assignee_key(account.id, label_id, inbox_id, user.id)
+      ]
+    when :mine
+      [store.label_inbox_assignee_key(account.id, label_id, inbox_id, user.id)]
+    end
+  end
+
+  def team_inbox_keys_for_mode(team_id, inbox_id)
+    case permission_mode
+    when :base
+      [store.team_inbox_key(account.id, team_id, inbox_id)]
+    when :unassigned_and_mine
+      [
+        store.team_inbox_unassigned_key(account.id, team_id, inbox_id),
+        store.team_inbox_assignee_key(account.id, team_id, inbox_id, user.id)
+      ]
+    when :mine
+      [store.team_inbox_assignee_key(account.id, team_id, inbox_id, user.id)]
+    end
+  end
+
+  def counts_for_grouped_keys(keys_by_id)
+    counts_by_key = store.counts_for_keys(keys_by_id.values.flatten)
+
+    keys_by_id.each_with_object({}) do |(id, keys), result|
+      count = keys.sum { |key| counts_by_key[key].to_i }
+      result[id.to_s] = count if count.positive?
+    end
+  end
+
+  def assignment_mode?
+    %i[unassigned_and_mine mine].include?(permission_mode)
+  end
+
+  def permission_mode
+    @permission_mode ||=
+      if !custom_role_agent? || permissions.include?(MANAGE_ALL_PERMISSION)
+        :base
+      elsif permissions.include?(UNASSIGNED_PERMISSION)
+        :unassigned_and_mine
+      elsif permissions.include?(PARTICIPATING_PERMISSION)
+        :mine
+      else
+        :none
+      end
+  end
+
+  def custom_role_agent?
+    account_user&.agent? && account_user.custom_role_id.present?
+  end
+
+  def permissions
+    account_user&.permissions || []
+  end
+
+  def account_user
+    @account_user ||= account.account_users.find_by(user_id: user.id)
+  end
+
+  def visible_inbox_ids
+    @visible_inbox_ids ||= if account_user&.administrator?
+                             account.inboxes.pluck(:id)
+                           else
+                             user.inboxes.where(account_id: account.id).pluck(:id)
+                           end
+  end
+
+  def sidebar_label_ids
+    @sidebar_label_ids ||= account.labels.where(show_on_sidebar: true).pluck(:id)
+  end
+
+  def visible_team_ids
+    @visible_team_ids ||= if account_user&.administrator?
+                            account.teams.pluck(:id)
+                          else
+                            user.teams.where(account_id: account.id).pluck(:id)
+                          end
+  end
+
+  def empty_counts
+    { inboxes: {}, labels: {}, teams: {} }
+  end
+
+  def store
+    ::Conversations::UnreadCounts::Store
+  end
+end

app/services/conversations/unread_counts/store.rb

@@ -0,0 +1,220 @@
+class Conversations::UnreadCounts::Store
+  class << self
+    def base_ready?(account_id)
+      Redis::Alfred.exists?(base_ready_key(account_id))
+    end
+
+    def assignment_ready?(account_id)
+      Redis::Alfred.exists?(assignment_ready_key(account_id))
+    end
+
+    def mark_base_ready!(account_id)
+      Redis::Alfred.set(base_ready_key(account_id), Time.current.to_i, ex: Conversations::UnreadCounts::READY_TTL)
+    end
+
+    def mark_assignment_ready!(account_id)
+      Redis::Alfred.set(assignment_ready_key(account_id), Time.current.to_i, ex: Conversations::UnreadCounts::READY_TTL)
+    end
+
+    def clear_account!(account_id)
+      delete_matching("#{account_prefix(account_id)}::*")
+    end
+
+    def clear_assignment!(account_id)
+      assignment_key_patterns(account_id).each { |pattern| delete_matching(pattern) }
+    end
+
+    def add_base_membership(account_id:, inbox_id:, label_ids:, conversation_id:, team_id: nil)
+      add_to_sets(base_keys(account_id, inbox_id, label_ids, team_id), conversation_id)
+    end
+
+    def remove_base_membership(account_id:, inbox_ids:, label_ids:, conversation_id:, team_ids: [])
+      keys = Array(inbox_ids).flat_map { |inbox_id| removable_base_keys(account_id, inbox_id, label_ids, team_ids) }
+      remove_from_sets(keys, conversation_id)
+    end
+
+    def add_assignment_membership(account_id:, conversation_id:, **membership)
+      add_to_sets(
+        assignment_keys(account_id, membership[:inbox_id], membership[:label_ids], membership[:assignee_id], membership[:team_id]),
+        conversation_id
+      )
+    end
+
+    def remove_assignment_membership(account_id:, conversation_id:, **membership)
+      keys = Array(membership[:inbox_ids]).flat_map do |inbox_id|
+        Array(membership[:assignee_ids]).flat_map do |assignee_id|
+          removable_assignment_keys(account_id, inbox_id, membership[:label_ids], assignee_id, membership[:team_ids])
+        end
+      end
+      remove_from_sets(keys, conversation_id)
+    end
+
+    def add_memberships(account_id:, memberships:, assignment: false)
+      return if memberships.blank?
+
+      Redis::Alfred.pipelined do |pipeline|
+        memberships.each do |membership|
+          keys = if assignment
+                   assignment_keys(account_id, membership[:inbox_id], membership[:label_ids], membership[:assignee_id], membership[:team_id])
+                 else
+                   base_keys(account_id, membership[:inbox_id], membership[:label_ids], membership[:team_id])
+                 end
+
+          keys.each do |key|
+            pipeline.sadd(key, membership[:conversation_id])
+            pipeline.expire(key, Conversations::UnreadCounts::SET_TTL)
+          end
+        end
+      end
+    end
+
+    def counts_for_keys(keys)
+      keys = keys.compact_blank
+      return {} if keys.blank?
+
+      counts = Redis::Alfred.pipelined do |pipeline|
+        keys.each { |key| pipeline.scard(key) }
+      end
+      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 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
+
+    private
+
+    def base_ready_key(account_id)
+      format(Redis::Alfred::UNREAD_CONVERSATIONS_BASE_READY, account_id: account_id)
+    end
+
+    def assignment_ready_key(account_id)
+      format(Redis::Alfred::UNREAD_CONVERSATIONS_ASSIGNMENT_READY, account_id: account_id)
+    end
+
+    def account_prefix(account_id)
+      format(Redis::Alfred::UNREAD_CONVERSATIONS_ACCOUNT_PREFIX, account_id: account_id)
+    end
+
+    def base_keys(account_id, inbox_id, label_ids, team_id = nil)
+      keys = [inbox_key(account_id, inbox_id)] + Array(label_ids).map { |label_id| label_inbox_key(account_id, label_id, inbox_id) }
+      keys << team_inbox_key(account_id, team_id, inbox_id) if team_id.present?
+      keys
+    end
+
+    def removable_base_keys(account_id, inbox_id, label_ids, team_ids)
+      keys = base_keys(account_id, inbox_id, label_ids)
+      keys.concat(Array(team_ids).compact_blank.map { |team_id| team_inbox_key(account_id, team_id, inbox_id) })
+    end
+
+    def assignment_keys(account_id, inbox_id, label_ids, assignee_id, team_id = nil)
+      keys = assignment_keys_without_team(account_id, inbox_id, label_ids, assignee_id)
+      keys << team_assignment_key(account_id, team_id, inbox_id, assignee_id) if team_id.present?
+      keys
+    end
+
+    def removable_assignment_keys(account_id, inbox_id, label_ids, assignee_id, team_ids)
+      keys = assignment_keys_without_team(account_id, inbox_id, label_ids, assignee_id)
+      keys.concat(Array(team_ids).compact_blank.map { |team_id| team_assignment_key(account_id, team_id, inbox_id, assignee_id) })
+    end
+
+    def assignment_keys_without_team(account_id, inbox_id, label_ids, assignee_id)
+      return assignee_keys(account_id, inbox_id, label_ids, assignee_id) if assignee_id.present?
+
+      unassigned_keys(account_id, inbox_id, label_ids)
+    end
+
+    def assignee_keys(account_id, inbox_id, label_ids, assignee_id)
+      [inbox_assignee_key(account_id, inbox_id, assignee_id)] +
+        Array(label_ids).map { |label_id| label_inbox_assignee_key(account_id, label_id, inbox_id, assignee_id) }
+    end
+
+    def unassigned_keys(account_id, inbox_id, label_ids)
+      [inbox_unassigned_key(account_id, inbox_id)] +
+        Array(label_ids).map { |label_id| label_inbox_unassigned_key(account_id, label_id, inbox_id) }
+    end
+
+    def team_assignment_key(account_id, team_id, inbox_id, assignee_id)
+      return team_inbox_assignee_key(account_id, team_id, inbox_id, assignee_id) if assignee_id.present?
+
+      team_inbox_unassigned_key(account_id, team_id, inbox_id)
+    end
+
+    def add_to_sets(keys, conversation_id)
+      write_to_sets(keys) { |pipeline, key| pipeline.sadd(key, conversation_id) }
+    end
+
+    def remove_from_sets(keys, conversation_id)
+      write_to_sets(keys) { |pipeline, key| pipeline.srem(key, conversation_id) }
+    end
+
+    def write_to_sets(keys)
+      keys = keys.compact_blank
+      return if keys.blank?
+
+      Redis::Alfred.pipelined do |pipeline|
+        keys.each do |key|
+          yield(pipeline, key)
+          pipeline.expire(key, Conversations::UnreadCounts::SET_TTL)
+        end
+      end
+    end
+
+    def delete_matching(pattern)
+      Redis::Alfred.scan_each(match: pattern, count: 1000) do |key|
+        Redis::Alfred.delete(key)
+      end
+    end
+
+    def assignment_key_patterns(account_id)
+      prefix = account_prefix(account_id)
+      [
+        assignment_ready_key(account_id),
+        "#{prefix}::INBOX::*::UNASSIGNED",
+        "#{prefix}::INBOX::*::ASSIGNEE::*",
+        "#{prefix}::LABEL::*::INBOX::*::UNASSIGNED",
+        "#{prefix}::LABEL::*::INBOX::*::ASSIGNEE::*",
+        "#{prefix}::TEAM::*::INBOX::*::UNASSIGNED",
+        "#{prefix}::TEAM::*::INBOX::*::ASSIGNEE::*"
+      ]
+    end
+  end
+end

lib/redis/alfred.rb

@@ -40,6 +40,11 @@ module Redis::Alfred
       $alfred.with { |conn| conn.expire(key, seconds) }
     end
 
+    # get expiry of a key in seconds
+    def ttl(key)
+      $alfred.with { |conn| conn.ttl(key) }
+    end
+
     # scan keys matching a pattern
     def scan_each(match: nil, count: 100, &)
       $alfred.with do |conn|
@@ -80,6 +85,10 @@ module Redis::Alfred
       $alfred.with { |conn| conn.lrem(key, count, value) }
     end
 
+    def pipelined(&)
+      $alfred.with { |conn| conn.pipelined(&) }
+    end
+
     # hash operations
 
     # add a key value to redis hash

lib/redis/lock_manager.rb

@@ -49,6 +49,18 @@ class Redis::LockManager
     true
   end
 
+  def with_lock(key, timeout = LOCK_TIMEOUT)
+    return false unless lock(key, timeout)
+
+    begin
+      yield
+    ensure
+      unlock(key)
+    end
+
+    true
+  end
+
   # Checks if the given key is currently locked.
   #
   # === Parameters

lib/redis/redis_keys.rb

@@ -9,6 +9,28 @@ module Redis::RedisKeys
   # Whether a conversation is muted ?
   CONVERSATION_MUTE_KEY = 'CONVERSATION::%<id>d::MUTED'.freeze
   CONVERSATION_DRAFT_MESSAGE = 'CONVERSATION::%<id>d::DRAFT_MESSAGE'.freeze
+  UNREAD_CONVERSATIONS_ACCOUNT_PREFIX = 'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d'.freeze
+  UNREAD_CONVERSATIONS_BASE_READY = 'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::READY::BASE'.freeze
+  UNREAD_CONVERSATIONS_ASSIGNMENT_READY = 'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::READY::ASSIGNMENT'.freeze
+  UNREAD_CONVERSATIONS_BASE_BUILD_LOCK = 'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::BUILD_LOCK::BASE'.freeze
+  UNREAD_CONVERSATIONS_ASSIGNMENT_BUILD_LOCK = 'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::BUILD_LOCK::ASSIGNMENT'.freeze
+  UNREAD_CONVERSATIONS_INBOX = 'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::INBOX::%<inbox_id>d'.freeze
+  UNREAD_CONVERSATIONS_LABEL_INBOX =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::LABEL::%<label_id>d::INBOX::%<inbox_id>d'.freeze
+  UNREAD_CONVERSATIONS_TEAM_INBOX =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::TEAM::%<team_id>d::INBOX::%<inbox_id>d'.freeze
+  UNREAD_CONVERSATIONS_INBOX_UNASSIGNED =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::INBOX::%<inbox_id>d::UNASSIGNED'.freeze
+  UNREAD_CONVERSATIONS_INBOX_ASSIGNEE =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::INBOX::%<inbox_id>d::ASSIGNEE::%<user_id>d'.freeze
+  UNREAD_CONVERSATIONS_LABEL_INBOX_UNASSIGNED =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::LABEL::%<label_id>d::INBOX::%<inbox_id>d::UNASSIGNED'.freeze
+  UNREAD_CONVERSATIONS_LABEL_INBOX_ASSIGNEE =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::LABEL::%<label_id>d::INBOX::%<inbox_id>d::ASSIGNEE::%<user_id>d'.freeze
+  UNREAD_CONVERSATIONS_TEAM_INBOX_UNASSIGNED =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::TEAM::%<team_id>d::INBOX::%<inbox_id>d::UNASSIGNED'.freeze
+  UNREAD_CONVERSATIONS_TEAM_INBOX_ASSIGNEE =
+    'UNREAD_CONVERSATIONS::V1::ACCOUNT::%<account_id>d::TEAM::%<team_id>d::INBOX::%<inbox_id>d::ASSIGNEE::%<user_id>d'.freeze
 
   ## User Keys
   # SSO Auth Tokens

spec/enterprise/services/conversations/unread_counts/counter_spec.rb

@@ -0,0 +1,72 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Counter do
+  let(:account) { create(:account) }
+  let(:agent) { create(:user, account: account, role: :agent) }
+  let(:other_agent) { create(:user, account: account, role: :agent) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:label) { create(:label, account: account, title: 'support', show_on_sidebar: true) }
+  let(:team) { create(:team, account: account, allow_auto_assign: false) }
+  let(:account_user) { account.account_users.find_by(user: agent) }
+  let(:store) { Conversations::UnreadCounts::Store }
+
+  before do
+    create(:inbox_member, user: agent, inbox: inbox)
+    create(:team_member, user: agent, team: team)
+  end
+
+  after do
+    store.clear_account!(account.id)
+  end
+
+  it 'uses base counts for custom roles with conversation_manage permission' do
+    account_user.update!(custom_role: create(:custom_role, account: account, permissions: ['conversation_manage']))
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: other_agent, team: team)
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], team: team)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result[:inboxes]).to eq(inbox.id.to_s => 2)
+    expect(result[:labels]).to eq(label.id.to_s => 2)
+    expect(result[:teams]).to eq(team.id.to_s => 2)
+    expect(store.assignment_ready?(account.id)).to be(false)
+  end
+
+  it 'counts assigned and unassigned conversations for conversation_unassigned_manage permission' do
+    account_user.update!(custom_role: create(:custom_role, account: account, permissions: ['conversation_unassigned_manage']))
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: agent, team: team)
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], team: team)
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: other_agent, team: team)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result[:inboxes]).to eq(inbox.id.to_s => 2)
+    expect(result[:labels]).to eq(label.id.to_s => 2)
+    expect(result[:teams]).to eq(team.id.to_s => 2)
+    expect(store.assignment_ready?(account.id)).to be(true)
+  end
+
+  it 'counts only assigned conversations for conversation_participating_manage permission' do
+    account_user.update!(custom_role: create(:custom_role, account: account, permissions: ['conversation_participating_manage']))
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: agent, team: team)
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], team: team)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result[:inboxes]).to eq(inbox.id.to_s => 1)
+    expect(result[:labels]).to eq(label.id.to_s => 1)
+    expect(result[:teams]).to eq(team.id.to_s => 1)
+    expect(store.assignment_ready?(account.id)).to be(true)
+  end
+
+  it 'returns zero for custom roles without conversation permissions' do
+    account_user.update!(custom_role: create(:custom_role, account: account, permissions: []))
+    create_unread_conversation(account: account, inbox: inbox, labels: [label.title], assignee: agent, team: team)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result).to eq(inboxes: {}, labels: {}, teams: {})
+    expect(store.base_ready?(account.id)).to be(false)
+    expect(store.assignment_ready?(account.id)).to be(false)
+  end
+end

spec/lib/redis/lock_manager_spec.rb

@@ -35,6 +35,28 @@ RSpec.describe Redis::LockManager do
     end
   end
 
+  describe '#with_lock' do
+    it 'yields when the lock is acquired and releases the lock' do
+      yielded = false
+
+      expect(lock_manager.with_lock(lock_key) { yielded = true }).to be true
+      expect(yielded).to be true
+      expect(lock_manager.locked?(lock_key)).to be false
+    end
+
+    it 'returns false without yielding when the lock is already acquired' do
+      lock_manager.lock(lock_key)
+
+      expect { |block| lock_manager.with_lock(lock_key, &block) }.not_to yield_control
+      expect(lock_manager.with_lock(lock_key) { raise 'should not run' }).to be false
+    end
+
+    it 'releases the lock when the block raises' do
+      expect { lock_manager.with_lock(lock_key) { raise 'boom' } }.to raise_error('boom')
+      expect(lock_manager.locked?(lock_key)).to be false
+    end
+  end
+
   describe '#locked?' do
     it 'returns true if a key is locked' do
       lock_manager.lock(lock_key)

spec/rails_helper.rb

@@ -71,6 +71,7 @@ RSpec.configure do |config|
   config.include FileUploadHelpers
   config.include CsvSpecHelpers
   config.include InstagramSpecHelpers
+  config.include ConversationsUnreadCountsHelpers
   config.include Devise::Test::IntegrationHelpers, type: :request
   config.include ActiveSupport::Testing::TimeHelpers
   config.include ActionCable::TestHelper

spec/services/conversations/unread_counts/builder_spec.rb

@@ -0,0 +1,93 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Builder do
+  let(:account) { create(:account) }
+  let(:inbox) { create(:inbox, account: account) }
+  let(:label) { create(:label, account: account, title: 'urgent', show_on_sidebar: true) }
+  let(:assignee) { create(:user, account: account, role: :agent) }
+  let(:team) { create(:team, account: account, allow_auto_assign: false) }
+  let(:store) { Conversations::UnreadCounts::Store }
+
+  after do
+    store.clear_account!(account.id)
+  end
+
+  describe '#build_base!' do
+    it 'stores unread open conversations by inbox and label inbox' do
+      unread_conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title], team: team)
+      create_read_conversation
+      create_resolved_unread_conversation
+
+      described_class.new(account).build_base!
+
+      expect(store.base_ready?(account.id)).to be(true)
+      expect(redis_set_members(store.inbox_key(account.id, inbox.id))).to contain_exactly(unread_conversation.id.to_s)
+      expect(redis_set_members(store.label_inbox_key(account.id, label.id, inbox.id))).to contain_exactly(unread_conversation.id.to_s)
+      expect(redis_set_members(store.team_inbox_key(account.id, team.id, inbox.id))).to contain_exactly(unread_conversation.id.to_s)
+    end
+
+    it 'clears assignment-aware cache data before rebuilding base data' do
+      assigned_conversation = create_unread_conversation(
+        account: account,
+        inbox: inbox,
+        labels: [label.title],
+        assignee: assignee,
+        team: team
+      )
+
+      described_class.new(account).build_assignment!
+      described_class.new(account).build_base!
+
+      expect(store.assignment_ready?(account.id)).to be(false)
+      expect(redis_set_members(store.inbox_assignee_key(account.id, inbox.id, assignee.id))).to be_empty
+      expect(redis_set_members(store.inbox_key(account.id, inbox.id))).to contain_exactly(assigned_conversation.id.to_s)
+    end
+  end
+
+  describe '#build_assignment!' do
+    it 'stores unread open conversations by unassigned and assignee dimensions' do
+      assigned_conversation = create_unread_conversation(
+        account: account,
+        inbox: inbox,
+        labels: [label.title],
+        assignee: assignee,
+        team: team
+      )
+      unassigned_conversation = create_unread_conversation(account: account, inbox: inbox, labels: [label.title], team: team)
+
+      described_class.new(account).build_assignment!
+
+      expect(store.assignment_ready?(account.id)).to be(true)
+      expect(redis_set_members(store.inbox_assignee_key(account.id, inbox.id, assignee.id))).to contain_exactly(assigned_conversation.id.to_s)
+      expect(redis_set_members(store.label_inbox_assignee_key(account.id, label.id, inbox.id, assignee.id))).to contain_exactly(
+        assigned_conversation.id.to_s
+      )
+      expect(redis_set_members(store.team_inbox_assignee_key(account.id, team.id, inbox.id, assignee.id))).to contain_exactly(
+        assigned_conversation.id.to_s
+      )
+      expect(redis_set_members(store.inbox_unassigned_key(account.id, inbox.id))).to contain_exactly(unassigned_conversation.id.to_s)
+      expect(redis_set_members(store.label_inbox_unassigned_key(account.id, label.id, inbox.id))).to contain_exactly(
+        unassigned_conversation.id.to_s
+      )
+      expect(redis_set_members(store.team_inbox_unassigned_key(account.id, team.id, inbox.id))).to contain_exactly(
+        unassigned_conversation.id.to_s
+      )
+    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)
+    conversation
+  end
+
+  def create_resolved_unread_conversation
+    conversation = create_unread_conversation(account: account, inbox: inbox)
+    conversation.update!(status: :resolved)
+    conversation
+  end
+
+  def redis_set_members(key)
+    Redis::Alfred.pipelined { |pipeline| pipeline.smembers(key) }.first
+  end
+end

spec/services/conversations/unread_counts/counter_spec.rb

@@ -0,0 +1,95 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Counter do
+  let(:account) { create(:account) }
+  let(:agent) { create(:user, account: account, role: :agent) }
+  let(:admin) { create(:user, account: account, role: :administrator) }
+  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(:hidden_label) { create(:label, account: account, title: 'internal', show_on_sidebar: false) }
+  let(:visible_team) { create(:team, account: account, allow_auto_assign: false) }
+  let(:store) { Conversations::UnreadCounts::Store }
+
+  before do
+    create(:inbox_member, user: agent, inbox: visible_inbox)
+    create(:team_member, user: agent, team: visible_team)
+  end
+
+  after do
+    store.clear_account!(account.id)
+  end
+
+  it 'builds the base cache on demand' do
+    create_unread_conversation(account: account, inbox: visible_inbox, labels: [label.title], team: visible_team)
+
+    described_class.new(account: account, user: agent).perform
+
+    expect(store.base_ready?(account.id)).to be(true)
+  end
+
+  it 'uses a Redis lock while building the base cache on demand' 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).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)
+
+    described_class.new(account: account, user: agent).perform
+
+    expect(lock_manager).to have_received(:with_lock).with(lock_key, described_class::BUILD_LOCK_TTL)
+  end
+
+  it 'waits instead of rebuilding when another process owns the base build lock' do
+    lock_manager = instance_double(Redis::LockManager, with_lock: false)
+    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) }
+    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)
+  end
+
+  it 'counts unread conversations only across inboxes visible to a normal agent' do
+    create_unread_conversation(account: account, inbox: visible_inbox, labels: [label.title], team: visible_team)
+    create_unread_conversation(account: account, inbox: hidden_inbox, labels: [label.title], team: visible_team)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result).to eq(
+      inboxes: { visible_inbox.id.to_s => 1 },
+      labels: { label.id.to_s => 1 },
+      teams: { visible_team.id.to_s => 1 }
+    )
+  end
+
+  it 'counts unread conversations across all account inboxes for admins' do
+    create_unread_conversation(account: account, inbox: visible_inbox, labels: [label.title], team: visible_team)
+    create_unread_conversation(account: account, inbox: hidden_inbox, labels: [label.title], team: visible_team)
+
+    result = described_class.new(account: account, user: admin).perform
+
+    expect(result).to eq(
+      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 }
+    )
+  end
+
+  it 'does not return zero counts or labels hidden from the sidebar' do
+    create_unread_conversation(account: account, inbox: visible_inbox, labels: [hidden_label.title], team: visible_team)
+
+    result = described_class.new(account: account, user: agent).perform
+
+    expect(result).to eq(
+      inboxes: { visible_inbox.id.to_s => 1 },
+      labels: {},
+      teams: { visible_team.id.to_s => 1 }
+    )
+  end
+end

spec/services/conversations/unread_counts/store_spec.rb

@@ -0,0 +1,183 @@
+require 'rails_helper'
+
+RSpec.describe Conversations::UnreadCounts::Store do
+  let(:account_id) { 1 }
+  let(:inbox_id) { 2 }
+  let(:label_id) { 3 }
+  let(:user_id) { 4 }
+  let(:conversation_id) { 5 }
+  let(:team_id) { 6 }
+
+  after do
+    described_class.clear_account!(account_id)
+  end
+
+  describe 'key builders' do
+    it 'builds base keys using the Redis key naming convention' do
+      expect(described_class.inbox_key(account_id, inbox_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::INBOX::2'
+      )
+      expect(described_class.label_inbox_key(account_id, label_id, inbox_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::LABEL::3::INBOX::2'
+      )
+      expect(described_class.team_inbox_key(account_id, team_id, inbox_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::TEAM::6::INBOX::2'
+      )
+    end
+
+    it 'builds assignment-aware keys using the Redis key naming convention' do
+      expect(described_class.inbox_unassigned_key(account_id, inbox_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::INBOX::2::UNASSIGNED'
+      )
+      expect(described_class.inbox_assignee_key(account_id, inbox_id, user_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::INBOX::2::ASSIGNEE::4'
+      )
+      expect(described_class.label_inbox_unassigned_key(account_id, label_id, inbox_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::LABEL::3::INBOX::2::UNASSIGNED'
+      )
+      expect(described_class.label_inbox_assignee_key(account_id, label_id, inbox_id, user_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::LABEL::3::INBOX::2::ASSIGNEE::4'
+      )
+      expect(described_class.team_inbox_unassigned_key(account_id, team_id, inbox_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::TEAM::6::INBOX::2::UNASSIGNED'
+      )
+      expect(described_class.team_inbox_assignee_key(account_id, team_id, inbox_id, user_id)).to eq(
+        'UNREAD_CONVERSATIONS::V1::ACCOUNT::1::TEAM::6::INBOX::2::ASSIGNEE::4'
+      )
+    end
+  end
+
+  describe 'ready markers' do
+    it 'tracks base and assignment readiness independently' do
+      expect(described_class.base_ready?(account_id)).to be(false)
+      expect(described_class.assignment_ready?(account_id)).to be(false)
+
+      described_class.mark_base_ready!(account_id)
+      described_class.mark_assignment_ready!(account_id)
+
+      expect(described_class.base_ready?(account_id)).to be(true)
+      expect(described_class.assignment_ready?(account_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)
+    end
+  end
+
+  describe 'set operations' do
+    it 'adds, counts, and removes base memberships' 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.counts_for_keys(base_keys)).to eq(
+        described_class.inbox_key(account_id, inbox_id) => 1,
+        described_class.label_inbox_key(account_id, label_id, inbox_id) => 1,
+        described_class.team_inbox_key(account_id, team_id, inbox_id) => 1
+      )
+      expect(base_keys.map { |key| ttl_for(key) }).to all(be_within(5).of(Conversations::UnreadCounts::SET_TTL))
+
+      described_class.remove_base_membership(
+        account_id: account_id,
+        inbox_ids: [inbox_id],
+        label_ids: [label_id],
+        team_ids: [team_id],
+        conversation_id: conversation_id
+      )
+
+      expect(described_class.counts_for_keys(base_keys).values).to all(eq(0))
+    end
+
+    it 'adds, counts, and removes assignment-aware memberships' do
+      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
+      )
+
+      expect(described_class.counts_for_keys(assignment_keys)).to eq(
+        described_class.inbox_assignee_key(account_id, inbox_id, user_id) => 1,
+        described_class.label_inbox_assignee_key(account_id, label_id, inbox_id, user_id) => 1,
+        described_class.team_inbox_assignee_key(account_id, team_id, inbox_id, user_id) => 1
+      )
+      expect(assignment_keys.map { |key| ttl_for(key) }).to all(be_within(5).of(Conversations::UnreadCounts::SET_TTL))
+
+      described_class.remove_assignment_membership(
+        account_id: account_id,
+        inbox_ids: [inbox_id],
+        label_ids: [label_id],
+        assignee_ids: [user_id],
+        team_ids: [team_id],
+        conversation_id: conversation_id
+      )
+
+      expect(described_class.counts_for_keys(assignment_keys).values).to all(eq(0))
+    end
+
+    it 'sets expiry on bulk membership writes' do
+      described_class.add_memberships(
+        account_id: account_id,
+        memberships: [{
+          inbox_id: inbox_id,
+          label_ids: [label_id],
+          team_id: team_id,
+          conversation_id: conversation_id
+        }]
+      )
+
+      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
+      described_class.mark_base_ready!(account_id)
+      described_class.mark_assignment_ready!(account_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.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.counts_for_keys(base_keys).values).to all(eq(0))
+      expect(described_class.counts_for_keys(assignment_keys).values).to all(eq(0))
+    end
+  end
+
+  def base_keys
+    [
+      described_class.inbox_key(account_id, inbox_id),
+      described_class.label_inbox_key(account_id, label_id, inbox_id),
+      described_class.team_inbox_key(account_id, team_id, inbox_id)
+    ]
+  end
+
+  def assignment_keys
+    [
+      described_class.inbox_assignee_key(account_id, inbox_id, user_id),
+      described_class.label_inbox_assignee_key(account_id, label_id, inbox_id, user_id),
+      described_class.team_inbox_assignee_key(account_id, team_id, inbox_id, user_id)
+    ]
+  end
+
+  def ttl_for(key)
+    Redis::Alfred.ttl(key)
+  end
+end

spec/support/conversations_unread_counts_helpers.rb

@@ -0,0 +1,11 @@
+module ConversationsUnreadCountsHelpers
+  def create_unread_conversation(account:, inbox:, labels: [], assignee: nil, team: nil)
+    create(:team_member, user: assignee, team: team) if assignee.present? && team.present? && !team.members.exists?(assignee.id)
+
+    conversation = create(:conversation, account: account, inbox: inbox, assignee: assignee, team: team, agent_last_seen_at: 1.hour.ago)
+    conversation.update_labels(labels) if labels.present?
+
+    create(:message, account: account, inbox: inbox, conversation: conversation, message_type: :incoming, created_at: 5.minutes.ago)
+    conversation
+  end
+end