diff --git a/app/models/content_participation.rb b/app/models/content_participation.rb
index 3f5c6fd49c0..ae370f3beaf 100644
--- a/app/models/content_participation.rb
+++ b/app/models/content_participation.rb
@@ -21,21 +21,25 @@
 class ContentParticipation < ActiveRecord::Base
   include Workflow
 
-  ACCESSIBLE_ATTRIBUTES = %i[content user workflow_state].freeze
+  ACCESSIBLE_ATTRIBUTES = %i[content user workflow_state content_item].freeze
+  CONTENT_ITEMS = %w[grade comment rubric].freeze
 
   belongs_to :content, polymorphic: [:submission]
   belongs_to :user
 
   before_create :set_root_account_id
-  after_save :update_participation_count
+  after_save :update_participation_count, if: :call_update_participation_count?
 
-  validates :content_type, :content_id, :user_id, :workflow_state, presence: true
+  validates :content_type, :content_id, :user_id, :workflow_state, :content_item, presence: true
+  validates :content_item, inclusion: { in: CONTENT_ITEMS }
 
   workflow do
     state :unread
     state :read
   end
 
+  attr_accessor :unread_count_offset
+
   def self.create_or_update(opts = {})
     opts = opts.with_indifferent_access
     content = opts.delete(:content)
@@ -52,18 +56,138 @@ class ContentParticipation < ActiveRecord::Base
     participant
   end
 
+  def call_update_participation_count?
+    return true unless Account.site_admin.feature_enabled?(:visibility_feedback_student_grades_page)
+
+    content.posted?
+  end
+
   def update_participation_count
     return unless saved_change_to_workflow_state?
 
+    offset = if Account.site_admin.feature_enabled?(:visibility_feedback_student_grades_page)
+               unread_count_offset
+             else
+               ((workflow_state == "unread") ? 1 : -1)
+             end
+
     ContentParticipationCount.create_or_update({
                                                  context: content.context,
                                                  user: user,
                                                  content_type: content_type,
-                                                 offset: (workflow_state == "unread" ? 1 : -1),
+                                                 offset: offset,
                                                })
   end
 
   def set_root_account_id
     self.root_account_id = content.assignment.root_account_id
   end
+
+  def self.participate(content:, user:, workflow_state: "unread", content_item: "grade")
+    return unless Account.site_admin.feature_enabled?(:visibility_feedback_student_grades_page)
+
+    raise "cannot read user and content" unless content.is_a?(Submission) && user.is_a?(User)
+
+    participant = nil
+
+    unique_constraint_retry do
+      participations = content.content_participations.where(user_id: user)
+
+      participant = create_first_participation_item(participations, content, user, workflow_state, content_item)
+
+      participant ||= update_existing_participation_item(participations, workflow_state, content_item)
+
+      participant ||= add_participation_item(participations, content, user, workflow_state, content_item)
+
+      participant.save! if participant.new_record? || participant.changed?
+    end
+
+    participant
+  end
+
+  def self.create_first_participation_item(participations, content, user, workflow_state, content_item)
+    return if participations.any?
+
+    participant = build_item(content, user, workflow_state, content_item)
+    participant.unread_count_offset = workflow_state == "unread" ? 1 : 0
+
+    participant
+  end
+  private_class_method :create_first_participation_item
+
+  def self.update_existing_participation_item(participations, workflow_state, content_item)
+    participant = participations.find { |p| p.content_item == content_item }
+
+    return participant if participant.nil? || same_workflow_state?(participant, workflow_state)
+
+    participations -= [participant]
+
+    participant.unread_count_offset = if participations.empty? || all_read?(participations)
+                                        workflow_state == "unread" ? 1 : -1
+                                      else
+                                        0
+                                      end
+    participant.workflow_state = workflow_state
+    participant
+  end
+  private_class_method :update_existing_participation_item
+
+  def self.add_participation_item(participations, content, user, workflow_state, content_item)
+    participant = build_item(content, user, workflow_state, content_item)
+    participant.unread_count_offset = (all_read?(participations) && workflow_state == "unread") ? 1 : 0
+    participant
+  end
+  private_class_method :add_participation_item
+
+  def self.all_read?(items)
+    items.all? { |participant| participant.workflow_state == "read" }
+  end
+  private_class_method :all_read?
+
+  def self.build_item(content, user, workflow_state, content_item)
+    content.content_participations.build(user: user, workflow_state: workflow_state, content_item: content_item)
+  end
+  private_class_method :build_item
+
+  def self.same_workflow_state?(participant, workflow_state)
+    participant.present? && participant.workflow_state == workflow_state
+  end
+  private_class_method :same_workflow_state?
+
+  def self.submission_read?(content:, user:)
+    submission_read_state(content, user) != "unread"
+  end
+
+  def self.submission_item_read?(content:, user:, content_item:)
+    submission_read_state(content, user, content_item) != "unread"
+  end
+
+  def self.submission_read_state(content, user, content_item = nil)
+    raise "content is not a Submission" unless content.is_a?(Submission)
+    raise "#{content_item} is invalid" if content_item.present? && !CONTENT_ITEMS.include?(content_item)
+
+    states = if content_item.present?
+               ContentParticipation.where(content: content, user: user, content_item: content_item).pluck(:workflow_state)
+             else
+               ContentParticipation.where(content: content, user: user).pluck(:workflow_state)
+             end
+
+    return nil if states.empty?
+    return "unread" if states.any?("unread")
+
+    "read"
+  end
+
+  def self.already_read_count(ids = [], user)
+    ContentParticipation
+      .group(:content_id)
+      .where(
+        content_type: "Submission",
+        content_id: ids,
+        user_id: user
+      )
+      .having("sum(case workflow_state when 'unread' then 1 else 0 end) = 0")
+      .pluck(:content_id)
+      .count
+  end
 end
diff --git a/app/models/content_participation_count.rb b/app/models/content_participation_count.rb
index 0f5274990be..d6db55af6fe 100644
--- a/app/models/content_participation_count.rb
+++ b/app/models/content_participation_count.rb
@@ -46,19 +46,27 @@ class ContentParticipationCount < ActiveRecord::Base
         end
         participant.attributes = opts.slice(*ACCESSIBLE_ATTRIBUTES)
 
-        # if the participant was just created, the count will already be correct
-        if opts[:offset].present? && !participant.new_record?
-          current_unread_count = participant.unread_count(refresh: false)
-          unless current_unread_count == 0 && opts[:offset] == -1
-            participant.unread_count = current_unread_count + opts[:offset]
-          end
-        end
+        set_unread_count(participant, opts)
+
         participant.save if participant.new_record? || participant.changed?
       end
     end
     participant
   end
 
+  def self.set_unread_count(participant, opts = {})
+    offset = opts.delete(:offset)
+    unread_count = opts.delete(:unread_count)
+
+    # allow setting the unread_count and when not present increment|decrement using an offset
+    unless unread_count.is_a?(Integer)
+      unread_count = participant.unread_count(refresh: false) + offset.to_i
+    end
+
+    participant.unread_count = unread_count > 0 ? unread_count : 0
+  end
+  private_class_method :set_unread_count
+
   def self.unread_count_for(type, context, user)
     return 0 unless user.present? && context.present?
 
@@ -100,17 +108,22 @@ class ContentParticipationCount < ActiveRecord::Base
                                        SQL
         (subs_with_grades + subs_with_comments).uniq
       end
-      already_read_count = if potential_ids.any?
-                             ContentParticipation.where(
-                               content_type: "Submission",
-                               content_id: potential_ids,
-                               user_id: user,
-                               workflow_state: "read"
-                             ).count
-                           else
-                             0
-                           end
-      potential_ids.size - already_read_count
+      potential_ids.size - already_read_count(potential_ids, user)
+    end
+  end
+
+  def self.already_read_count(ids = [], user)
+    return 0 if ids.empty?
+
+    if Account.site_admin.feature_enabled?(:visibility_feedback_student_grades_page)
+      ContentParticipation.already_read_count(ids, user)
+    else
+      ContentParticipation.where(
+        content_type: "Submission",
+        content_id: ids,
+        user_id: user,
+        workflow_state: "read"
+      ).count
     end
   end
 
diff --git a/db/migrate/20220922152827_add_content_item_to_content_participations.rb b/db/migrate/20220922152827_add_content_item_to_content_participations.rb
new file mode 100644
index 00000000000..fff23e57af9
--- /dev/null
+++ b/db/migrate/20220922152827_add_content_item_to_content_participations.rb
@@ -0,0 +1,42 @@
+# frozen_string_literal: true
+
+#
+# Copyright (C) 2022 - present Instructure, Inc.
+#
+# This file is part of Canvas.
+#
+# Canvas is free software: you can redistribute it and/or modify it under
+# the terms of the GNU Affero General Public License as published by the Free
+# Software Foundation, version 3 of the License.
+#
+# Canvas is distributed in the hope that it will be useful, but WITHOUT ANY
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE. See the GNU Affero General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Affero General Public License along
+# with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+
+class AddContentItemToContentParticipations < ActiveRecord::Migration[6.1]
+  tag :predeploy
+
+  disable_ddl_transaction!
+
+  def change
+    add_column :content_participations, :content_item, :string, null: false,
+                                                                default: "grade",
+                                                                if_not_exists: true
+
+    add_index :content_participations, %w[content_id content_type user_id content_item],
+              name: "index_content_participations_by_type_uniquely",
+              unique: true,
+              algorithm: :concurrently,
+              if_not_exists: true
+
+    remove_index :content_participations, column: %w[content_id content_type user_id],
+                                          name: "index_content_participations_uniquely",
+                                          algorithm: :concurrently,
+                                          if_exists: true
+  end
+end
diff --git a/spec/models/content_participation_spec.rb b/spec/models/content_participation_spec.rb
index 5da36d88c77..726d3a27618 100644
--- a/spec/models/content_participation_spec.rb
+++ b/spec/models/content_participation_spec.rb
@@ -57,6 +57,152 @@ describe ContentParticipation do
       cp = ContentParticipation.where(user_id: @student).first
       expect(cp.workflow_state).to eq "unread"
     end
+
+    context "when feeback visibility ff is on" do
+      before do
+        Account.site_admin.enable_feature!(:visibility_feedback_student_grades_page)
+      end
+
+      it "creates 'grade' participation if content_item not given" do
+        expect do
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread")
+        end.to change(ContentParticipation, :count).by 1
+
+        cp = ContentParticipation.where(user_id: @student).first
+        expect(cp.content_item).to eq "grade"
+      end
+
+      it "create a participation item as read" do
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+
+        cp = ContentParticipation.where(user_id: @student).first
+        expect(cp.workflow_state).to eq "read"
+      end
+
+      it "creates all possible content items for a submission" do
+        expect do
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread")
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread", content_item: "comment")
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread", content_item: "rubric")
+        end.to change(ContentParticipation, :count).by 3
+      end
+
+      it "doesn't allow invalid content_item" do
+        expect do
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread", content_item: "_ruby")
+        end.to raise_error(ActiveRecord::RecordInvalid)
+      end
+
+      it "doesn't duplicate content_item if submission is the same" do
+        expect do
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread")
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread")
+        end.to change(ContentParticipation, :count).by 1
+      end
+
+      it "changes a content_item state from unread to read" do
+        expect do
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread")
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+        end.to change(ContentParticipation, :count).by 1
+
+        cp = ContentParticipation.where(user_id: @student).first
+        expect(cp.workflow_state).to eq "read"
+      end
+
+      it "changes a content_item state from read to unread" do
+        expect do
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread")
+        end.to change(ContentParticipation, :count).by 1
+
+        cp = ContentParticipation.where(user_id: @student).first
+        expect(cp.workflow_state).to eq "unread"
+      end
+
+      context "when multiple submissions exist" do
+        before do
+          temp_assignment = @assignment
+          @assignment2 = assignment_model(course: @course)
+          @content2 = @assignment2.submit_homework(@student)
+          @assignment = temp_assignment
+        end
+
+        it "create two participation if same item but different submissions" do
+          expect do
+            ContentParticipation.participate(content: @content, user: @student)
+            ContentParticipation.participate(content: @content2, user: @student)
+          end.to change(ContentParticipation, :count).by 2
+        end
+      end
+    end
+  end
+
+  describe "submission_read?" do
+    context "when feeback visibility ff is on" do
+      before do
+        Account.site_admin.enable_feature!(:visibility_feedback_student_grades_page)
+      end
+
+      it "is read if no participation exist" do
+        expect(ContentParticipation.count).to eq 0
+        expect(ContentParticipation.submission_read?(content: @content, user: @student)).to be_truthy
+      end
+
+      it "is not read if existing item is unread" do
+        ContentParticipation.participate(content: @content, user: @student)
+        expect(ContentParticipation.submission_read?(content: @content, user: @student)).to be_falsey
+      end
+
+      it "is read if one content_item is present and state changes to read" do
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread")
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+        expect(ContentParticipation.submission_read?(content: @content, user: @student)).to be_truthy
+      end
+
+      it "is read when all items are read" do
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+        ContentParticipation.participate(content: @content, user: @student, content_item: "comment", workflow_state: "read")
+        ContentParticipation.participate(content: @content, user: @student, content_item: "rubric", workflow_state: "read")
+        expect(ContentParticipation.submission_read?(content: @content, user: @student)).to be_truthy
+      end
+
+      it "is unread if at least one item is unread" do
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+        ContentParticipation.participate(content: @content, user: @student, content_item: "comment", workflow_state: "unread")
+        ContentParticipation.participate(content: @content, user: @student, content_item: "rubric", workflow_state: "read")
+        expect(ContentParticipation.submission_read?(content: @content, user: @student)).to be_falsey
+      end
+    end
+  end
+
+  describe "submission_item_read?" do
+    context "when feeback visibility ff is on" do
+      before do
+        Account.site_admin.enable_feature!(:visibility_feedback_student_grades_page)
+      end
+
+      it "grade is unread if workflow_state is not given" do
+        ContentParticipation.participate(content: @content, user: @student)
+
+        expect(ContentParticipation.submission_item_read?(
+                 content: @content,
+                 user: @student,
+                 content_item: "grade"
+               )).to be_falsey
+      end
+
+      it "changes submisison grade from unread to read" do
+        ContentParticipation.participate(content: @content, user: @student)
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+
+        expect(ContentParticipation.submission_item_read?(
+                 content: @content,
+                 user: @student,
+                 content_item: "grade"
+               )).to be_truthy
+      end
+    end
   end
 
   describe "update_participation_count" do
@@ -113,6 +259,164 @@ describe ContentParticipation do
       cpc = ContentParticipationCount.where(user_id: @student).first
       expect(cpc.unread_count).to eq 0
     end
+
+    context "when feeback visibility ff is on" do
+      before do
+        Account.site_admin.enable_feature!(:visibility_feedback_student_grades_page)
+        @content.update_columns(posted_at: Time.now.utc, workflow_state: "graded", score: 10)
+      end
+
+      it "unread_count is 1 when at least one submission participation item is unread" do
+        expect do
+          ContentParticipation.participate(content: @content, user: @student)
+          ContentParticipation.participate(content: @content, user: @student, content_item: "comment", workflow_state: "read")
+          ContentParticipation.participate(content: @content, user: @student, content_item: "rubric", workflow_state: "read")
+        end.to change(ContentParticipationCount, :count).by 1
+
+        cpc = ContentParticipationCount.where(user_id: @student).first
+        expect(cpc.unread_count).to eq 1
+      end
+
+      it "unread_count is 0 when all submission participation items are read" do
+        expect do
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "read", content_item: "comment")
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "read", content_item: "rubric")
+        end.to change(ContentParticipationCount, :count).by 1
+
+        cpc = ContentParticipationCount.where(user_id: @student).first
+        expect(cpc.unread_count).to eq 0
+      end
+
+      it "unread_count is 1 when all items are unread and one is set to read" do
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+        ContentParticipation.participate(content: @content, user: @student, content_item: "comment")
+        ContentParticipation.participate(content: @content, user: @student, content_item: "rubric")
+
+        expect do
+          ContentParticipation.participate(content: @content, user: @student, content_item: "rubric", workflow_state: "read")
+        end.to change(ContentParticipationCount, :count).by 0
+
+        cpc = ContentParticipationCount.where(user_id: @student).first
+        expect(cpc.unread_count).to eq 1
+      end
+
+      it "unread_count is 1 when all items are read and one is set to unread" do
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read", content_item: "comment")
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read", content_item: "rubric")
+
+        expect(ContentParticipationCount.count).to eq 1
+
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread", content_item: "comment")
+
+        cpc = ContentParticipationCount.where(user_id: @student).first
+        expect(cpc.reload.unread_count).to eq 1
+      end
+
+      it "unread_count is 0 when the last unread item is set to read" do
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread", content_item: "comment")
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read", content_item: "rubric")
+
+        cpc = ContentParticipationCount.where(user_id: @student).first
+        expect(cpc.unread_count).to eq 1
+
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read", content_item: "comment")
+        expect(cpc.reload.unread_count).to eq 0
+      end
+
+      it "unread_count is 1 the only existing item is changed to unread" do
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+
+        cpc = ContentParticipationCount.where(user_id: @student).first
+        expect(cpc.unread_count).to eq 0
+
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "unread")
+        expect(cpc.reload.unread_count).to eq 1
+      end
+
+      it "unread_count is 0 the only existing item is changed to read" do
+        ContentParticipation.participate(content: @content, user: @student)
+
+        cpc = ContentParticipationCount.where(user_id: @student).first
+        expect(cpc.unread_count).to eq 1
+
+        ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+        expect(cpc.reload.unread_count).to eq 0
+      end
+
+      context "when multiple submissions exist" do
+        before do
+          temp_assignment = @assignment
+          @assignment2 = assignment_model(course: @course)
+          @content2 = @assignment2.submit_homework(@student)
+          @assignment = temp_assignment
+          @content2.update_columns(posted_at: Time.now.utc, workflow_state: "graded", score: 10)
+        end
+
+        it "unread_count is 2 when submissions are unread" do
+          ContentParticipation.participate(content: @content, user: @student)
+          ContentParticipation.participate(content: @content2, user: @student)
+
+          cpc = ContentParticipationCount.where(user_id: @student).first
+          expect(cpc.unread_count).to eq 2
+        end
+
+        it "unread_count is 0 when submissions are read" do
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+          ContentParticipation.participate(content: @content2, user: @student, workflow_state: "read")
+
+          cpc = ContentParticipationCount.where(user_id: @student).first
+          expect(cpc.unread_count).to eq 0
+        end
+
+        it "unread_count changes from 2 to 0 after submissions are read" do
+          ContentParticipation.participate(content: @content, user: @student)
+          ContentParticipation.participate(content: @content2, user: @student)
+
+          cpc = ContentParticipationCount.where(user_id: @student).first
+          expect(cpc.unread_count).to eq 2
+
+          ContentParticipation.participate(content: @content, user: @student, workflow_state: "read")
+          ContentParticipation.participate(content: @content2, user: @student, workflow_state: "read")
+
+          expect(cpc.reload.unread_count).to eq 0
+        end
+
+        it "unread_count is 1 when one submission is read and another is unread" do
+          ContentParticipation.participate(content: @content, user: @student)
+          ContentParticipation.participate(content: @content2, user: @student, workflow_state: "read")
+
+          cpc = ContentParticipationCount.where(user_id: @student).first
+          expect(cpc.unread_count).to eq 1
+        end
+      end
+
+      context "when multiple courses exist" do
+        before do
+          @course2 = Course.create!
+          @course2_student = student_in_course(course: @course2, active_all: true).user
+          @course2_assignment = assignment_model(course: @course2, due_at: 2.days, points_possible: 10)
+          @course2_content = @course2_assignment.submit_homework(@course2_student)
+
+          @course2_content.update_columns(posted_at: Time.now.utc, workflow_state: "graded", score: 10)
+        end
+
+        it "unread_count is 1 for each course" do
+          expect do
+            ContentParticipation.participate(content: @content, user: @student)
+            ContentParticipation.participate(content: @course2_content, user: @course2_student)
+          end.to change(ContentParticipationCount, :count).by 2
+
+          cpc1 = @course.content_participation_counts.where(user_id: @student).first
+          expect(cpc1.unread_count).to eq 1
+
+          cpc2 = @course2.content_participation_counts.where(user_id: @course2_student).first
+          expect(cpc2.unread_count).to eq 1
+        end
+      end
+    end
   end
 
   describe "create" do