[MM-37013] Async job to fix CRT channel unreads (#18340)

Summary
The addition of the TotalMsgCountRoot and MsgCountRoot columns to support CRT caused several issues with previously read threads and channels being marked as unread. Previously we attempted to fix this purely in a SQL migration [MM-35345][MM-35494] fixes for incorrect mentions and unreads for threads and channels #17803 but that turned out to be too heavy and it was decided to break up some of the fixes into async jobs.
This PR implements an async job to mark channels as read if there are no user posts since the last time the user viewed the channel. 

Ticket Link
https://mattermost.atlassian.net/browse/MM-37013

app/enterprise.go

@@ -30,6 +30,12 @@ func RegisterComplianceInterface(f func(*Server) einterfaces.ComplianceInterface
 	complianceInterface = f
 }
 
+var fixCRTChannelUnreadsJobInterface func(*Server) tjobs.FixCRTChannelUnreadsJobInterface
+
+func RegisterFixCRTChannelUnreadsJobInterface(f func(*Server) tjobs.FixCRTChannelUnreadsJobInterface) {
+	fixCRTChannelUnreadsJobInterface = f
+}
+
 var dataRetentionInterface func(*Server) einterfaces.DataRetentionInterface
 
 func RegisterDataRetentionInterface(f func(*Server) einterfaces.DataRetentionInterface) {

app/server.go

@@ -1967,6 +1967,10 @@ func (s *Server) initJobs() {
 		s.Jobs.ExtractContent = jobsExtractContentInterface(s)
 	}
 
+	if fixCRTChannelUnreadsJobInterface != nil {
+		s.Jobs.FixCRTChannelUnreads = fixCRTChannelUnreadsJobInterface(s)
+	}
+
 	s.Jobs.InitWorkers()
 	s.Jobs.InitSchedulers()
 }

i18n/en.json

@@ -7523,6 +7523,22 @@
     "id": "extract_content.worker.do_job.file_info",
     "translation": "Failed to get file information for content extraction."
   },
+  {
+    "id": "fix_crt_channel_unreads.worker.do_job.get_bad_channel_memberships",
+    "translation": "Failed to get bad channel memberships"
+  },
+  {
+    "id": "fix_crt_channel_unreads.worker.do_job.get_post_types",
+    "translation": "Failed to get unique post types"
+  },
+  {
+    "id": "fix_crt_channel_unreads.worker.do_job.kill_orphan_jobs",
+    "translation": "Killing orphaned jobs"
+  },
+  {
+    "id": "fix_crt_channel_unreads.worker.do_job.mark_channel_as_read",
+    "translation": "Failed to mark channel as read"
+  },
   {
     "id": "group_not_associated_to_synced_team",
     "translation": "Group cannot be associated to the channel until it is first associated to the parent group-synced team."

imports/placeholder.go

@@ -39,4 +39,7 @@ import (
 
 	// This is a placeholder so this package can be imported in Team Edition when it will be otherwise empty.
 	_ "github.com/mattermost/mattermost-server/v6/jobs/extract_content"
+
+	// This is a placeholder so this package can be imported in Team Edition when it will be otherwise empty.
+	_ "github.com/mattermost/mattermost-server/v6/jobs/fix_crt_channel_unreads"
 )

jobs/fix_crt_channel_unreads/scheduler.go

@@ -0,0 +1,67 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package fix_crt_channel_unreads
+
+import (
+	"time"
+
+	"github.com/mattermost/mattermost-server/v6/app"
+	"github.com/mattermost/mattermost-server/v6/model"
+	"github.com/mattermost/mattermost-server/v6/shared/mlog"
+)
+
+type Scheduler struct {
+	App *app.App
+}
+
+func (i *FixCRTChannelUnreadsJobInterfaceImpl) MakeScheduler() model.Scheduler {
+	return &Scheduler{i.App}
+}
+
+func (s *Scheduler) Name() string {
+	return JobName + "Scheduler"
+}
+
+func (s *Scheduler) JobType() string {
+	return model.JobTypeFixChannelUnreadsForCRT
+}
+
+func (s *Scheduler) Enabled(cfg *model.Config) bool {
+	if _, err := s.App.Srv().Store.System().GetByName(model.MigrationKeyFixCRTChannelUnreads); err == nil {
+		return false
+	}
+	return true
+}
+
+func (s *Scheduler) NextScheduleTime(cfg *model.Config, now time.Time, pendingJobs bool, lastSuccessfulJob *model.Job) *time.Time {
+	nextTime := time.Now().Add(1 * time.Minute)
+
+	runningJobs, err := s.App.Srv().Store.Job().GetCountByStatusAndType(model.JobStatusInProgress, s.JobType())
+	if err != nil {
+		mlog.Error("Failed to get running jobs", mlog.Err(err))
+		runningJobs = 1
+	}
+	// if we have have pending or running jobs then schedule later
+	if pendingJobs || runningJobs > 0 {
+		nextTime = time.Now().Add(30 * time.Minute)
+	}
+	return &nextTime
+}
+
+func (s *Scheduler) ScheduleJob(cfg *model.Config, pendingJobs bool, lastSuccessfulJob *model.Job) (*model.Job, *model.AppError) {
+	// if we have pending or running jobs then don't create a job
+	runningJobs, sErr := s.App.Srv().Store.Job().GetCountByStatusAndType(model.JobStatusInProgress, s.JobType())
+	if sErr != nil {
+		mlog.Error("Failed to get running jobs", mlog.Err(sErr))
+	}
+	if pendingJobs || runningJobs > 0 {
+		return nil, nil
+	}
+
+	job, err := s.App.Srv().Jobs.CreateJob(model.JobTypeFixChannelUnreadsForCRT, map[string]string{})
+	if err != nil {
+		return nil, err
+	}
+	return job, nil
+}

jobs/fix_crt_channel_unreads/worker.go

@@ -0,0 +1,273 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package fix_crt_channel_unreads
+
+import (
+	"database/sql"
+	"net/http"
+	"strconv"
+
+	"github.com/mattermost/mattermost-server/v6/app"
+	"github.com/mattermost/mattermost-server/v6/jobs"
+	tjobs "github.com/mattermost/mattermost-server/v6/jobs/interfaces"
+	"github.com/mattermost/mattermost-server/v6/model"
+	"github.com/mattermost/mattermost-server/v6/shared/mlog"
+)
+
+const (
+	JobName = "FixCRTChannelUnreads"
+)
+
+type FixCRTChannelUnreadsWorker struct {
+	name        string
+	stopChan    chan struct{}
+	stoppedChan chan struct{}
+	jobsChan    chan model.Job
+	jobServer   *jobs.JobServer
+	app         *app.App
+}
+
+type FixCRTChannelUnreadsJobInterfaceImpl struct {
+	App *app.App
+}
+
+func init() {
+	app.RegisterFixCRTChannelUnreadsJobInterface(func(s *app.Server) tjobs.FixCRTChannelUnreadsJobInterface {
+		a := app.New(app.ServerConnector(s.Channels()))
+		return &FixCRTChannelUnreadsJobInterfaceImpl{a}
+	})
+}
+
+func (i *FixCRTChannelUnreadsJobInterfaceImpl) MakeWorker() model.Worker {
+
+	return &FixCRTChannelUnreadsWorker{
+		name:        JobName,
+		stopChan:    make(chan struct{}),
+		stoppedChan: make(chan struct{}),
+		jobsChan:    make(chan model.Job),
+		jobServer:   i.App.Srv().Jobs,
+		app:         i.App,
+	}
+}
+
+func (w *FixCRTChannelUnreadsWorker) JobChannel() chan<- model.Job {
+	return w.jobsChan
+}
+
+func (w *FixCRTChannelUnreadsWorker) Run() {
+	mlog.Debug("Worker started", mlog.String("worker", w.name))
+
+	// kill all in-progress jobs in DB. This can happen if server
+	// was shut down incorrectly
+	olderJobs, err := w.app.Srv().Store.Job().GetAllByStatus(model.JobStatusInProgress)
+	if err == nil {
+		for _, j := range olderJobs {
+			if j.Type == model.JobTypeFixChannelUnreadsForCRT {
+				w.setJobError(
+					j,
+					model.NewAppError(w.name, "fix_crt_channel_unreads.worker.do_job.kill_orphan_jobs",
+						nil,
+						"killing orphan jobs",
+						http.StatusInternalServerError),
+				)
+			}
+		}
+	}
+
+	defer func() {
+		mlog.Debug("Worker finished", mlog.String("worker", w.name))
+		close(w.stoppedChan)
+	}()
+
+	for {
+		select {
+		case <-w.stopChan:
+			mlog.Debug("Worker received stop signal", mlog.String("worker", w.name))
+			return
+		case job := <-w.jobsChan:
+			mlog.Debug("Worker received a new candidate job.", mlog.String("worker", w.name))
+			w.doJob(&job)
+		}
+	}
+}
+
+func (w *FixCRTChannelUnreadsWorker) Stop() {
+	mlog.Debug("Worker stopping", mlog.String("worker", w.name))
+	close(w.stopChan)
+	<-w.stoppedChan
+}
+
+func (w *FixCRTChannelUnreadsWorker) doJob(job *model.Job) {
+	if claimed, err := w.jobServer.ClaimJob(job); err != nil {
+		mlog.Warn("Worker experienced an error while trying to claim job",
+			mlog.String("worker", w.name),
+			mlog.String("job_id", job.Id),
+			mlog.String("error", err.Error()))
+		return
+	} else if !claimed {
+		return
+	}
+	if _, err := w.app.Srv().Store.System().GetByName(model.MigrationKeyFixCRTChannelUnreads); err == nil {
+		mlog.Info("Worker: migration already done", mlog.String("worker", w.name), mlog.String("job_id", job.Id))
+		w.setJobSuccess(job)
+		return
+	}
+
+	fixedBadCM := 0
+	checkedCM := 0
+	migrationDone := false
+	prevErr := false
+	channelID, userID := w.getProgressFromPreviousJobs(job)
+	if userID != "" && channelID != "" {
+		mlog.Info("Restarting from previous job run", mlog.String("ChannelID", channelID), mlog.String("UserID", userID))
+	}
+	shouldStop := false
+	for {
+		select {
+		case <-w.stopChan:
+			shouldStop = true
+		default:
+			shouldStop = false
+		}
+		if shouldStop {
+			break
+		}
+		cms, sErr := w.app.Srv().Store.Channel().GetCRTUnfixedChannelMembershipsAfter(channelID, userID, 100)
+		if sErr != nil {
+			if sErr == sql.ErrNoRows {
+				migrationDone = true
+				break
+			}
+			mlog.Warn("Failed to get bad channel memberships",
+				mlog.String("worker", w.name),
+				mlog.String("job_id", job.Id),
+				mlog.String("error", sErr.Error()))
+			if prevErr {
+				w.updateProgress(job, channelID, userID)
+				w.setJobError(job, model.NewAppError(w.name, "fix_crt_channel_unreads.worker.do_job.get_bad_channel_memberships", nil, sErr.Error(), http.StatusInternalServerError))
+				return
+			}
+			prevErr = true
+			continue
+		}
+		if len(cms) == 0 {
+			migrationDone = true
+			break
+		}
+		lastCM := cms[len(cms)-1]
+		channelID = lastCM.ChannelId
+		userID = lastCM.UserId
+
+		prevErr = false
+		cmToFix := make(map[string][]string)
+		for _, cm := range cms {
+			postTypes, err := w.app.Srv().Store.Post().GetUniquePostTypesSince(cm.ChannelId, cm.LastViewedAt)
+			if err != nil {
+				mlog.Warn("Failed to get unique unread posts",
+					mlog.String("worker", w.name),
+					mlog.String("job_id", job.Id),
+					mlog.String("error", err.Error()))
+				if prevErr {
+					w.updateProgress(job, cm.ChannelId, cm.UserId)
+					w.setJobError(job, model.NewAppError(w.name, "fix_crt_channel_unreads.worker.do_job.get_post_types", nil, err.Error(), http.StatusInternalServerError))
+					return
+				}
+				prevErr = true
+				continue
+			}
+			if containsOnlyJoinLeaveMessages(postTypes) {
+				cmToFix[cm.UserId] = append(cmToFix[cm.UserId], cm.ChannelId)
+			}
+		}
+		checkedCM += len(cms)
+		for uID, cIDs := range cmToFix {
+			_, err := w.app.Srv().Store.Channel().UpdateLastViewedAt(cIDs, uID, false)
+			if err != nil {
+				mlog.Warn("Worker experienced an error while trying to mark channel as read",
+					mlog.String("worker", w.name),
+					mlog.String("job_id", job.Id),
+					mlog.String("error", err.Error()))
+				if prevErr {
+					w.updateProgress(job, channelID, userID)
+					w.setJobError(job, model.NewAppError(w.name, "fix_crt_channel_unreads.worker.do_job.mark_channel_as_read", nil, err.Error(), http.StatusInternalServerError))
+					return
+				}
+				prevErr = true
+				continue
+			}
+			fixedBadCM += len(cIDs)
+		}
+		w.updateProgress(job, channelID, userID)
+		prevErr = false
+	}
+
+	if migrationDone {
+		system := model.System{
+			Name:  model.MigrationKeyFixCRTChannelUnreads,
+			Value: "true",
+		}
+
+		if err := w.app.Srv().Store.System().Save(&system); err != nil {
+			mlog.Critical("Failed to mark crt channel unreads migration job as completed.", mlog.Err(err))
+		}
+	}
+
+	job.Data["BadChannelMembershipsFixed"] = strconv.Itoa(fixedBadCM)
+	job.Data["TotalChannelMembershipsChecked"] = strconv.Itoa(checkedCM)
+	w.updateData(job)
+
+	mlog.Info("Worker: Job is complete", mlog.String("worker", w.name), mlog.String("job_id", job.Id))
+	w.setJobSuccess(job)
+}
+
+func (w *FixCRTChannelUnreadsWorker) setJobSuccess(job *model.Job) {
+	if err := w.app.Srv().Jobs.SetJobSuccess(job); err != nil {
+		mlog.Error("Worker: Failed to set success for job", mlog.String("worker", w.name), mlog.String("job_id", job.Id), mlog.String("error", err.Error()))
+		w.setJobError(job, err)
+	}
+}
+
+func (w *FixCRTChannelUnreadsWorker) setJobError(job *model.Job, appError *model.AppError) {
+	if err := w.app.Srv().Jobs.SetJobError(job, appError); err != nil {
+		mlog.Error("Worker: Failed to set job error", mlog.String("worker", w.name), mlog.String("job_id", job.Id), mlog.String("error", err.Error()))
+	}
+}
+
+func (w *FixCRTChannelUnreadsWorker) updateData(job *model.Job) {
+	if err := w.app.Srv().Jobs.UpdateInProgressJobData(job); err != nil {
+		mlog.Error("Worker: Failed to update job data", mlog.String("worker", w.name), mlog.String("job_id", job.Id), mlog.String("error", err.Error()))
+	}
+}
+
+func (w *FixCRTChannelUnreadsWorker) updateProgress(job *model.Job, channelID, userID string) {
+	job.Data["ChannelID"] = channelID
+	job.Data["UserID"] = userID
+	w.updateData(job)
+}
+
+func (w *FixCRTChannelUnreadsWorker) getProgressFromPreviousJobs(job *model.Job) (string, string) {
+	olderJob, err := w.app.Srv().Store.Job().GetNewestJobByStatusesAndType(
+		[]string{model.JobStatusCanceled, model.JobStatusCancelRequested, model.JobStatusSuccess, model.JobStatusError},
+		job.Type,
+	)
+	if err != nil {
+		return "", ""
+	}
+	return olderJob.Data["ChannelID"], olderJob.Data["UserID"]
+}
+
+func containsOnlyJoinLeaveMessages(postTypes []string) bool {
+	for _, pt := range postTypes {
+		switch pt {
+		case model.PostTypeJoinLeave, model.PostTypeAddRemove,
+			model.PostTypeJoinChannel, model.PostTypeLeaveChannel,
+			model.PostTypeJoinTeam, model.PostTypeLeaveTeam,
+			model.PostTypeAddToChannel, model.PostTypeRemoveFromChannel,
+			model.PostTypeAddToTeam, model.PostTypeRemoveFromTeam:
+		default:
+			return false
+		}
+	}
+	return true
+}

jobs/interfaces/fix_crt_channel_unreads_interface.go

@@ -0,0 +1,13 @@
+// Copyright (c) 2015-present Mattermost, Inc. All Rights Reserved.
+// See LICENSE.txt for license information.
+
+package interfaces
+
+import (
+	"github.com/mattermost/mattermost-server/v6/model"
+)
+
+type FixCRTChannelUnreadsJobInterface interface {
+	MakeWorker() model.Worker
+	MakeScheduler() model.Scheduler
+}

jobs/jobs_watcher.go

@@ -198,6 +198,13 @@ func (watcher *Watcher) PollAndNotify() {
 				default:
 				}
 			}
+		} else if job.Type == model.JobTypeFixChannelUnreadsForCRT {
+			if watcher.workers.FixCRTChannelUnreads != nil {
+				select {
+				case watcher.workers.FixCRTChannelUnreads.JobChannel() <- *job:
+				default:
+				}
+			}
 		}
 	}
 }

jobs/schedulers.go

@@ -101,6 +101,10 @@ func (srv *JobServer) InitSchedulers() error {
 		schedulers.schedulers = append(schedulers.schedulers, exportDeleteInterface.MakeScheduler())
 	}
 
+	if fixCRTChannelUnreadsInterface := srv.FixCRTChannelUnreads; fixCRTChannelUnreadsInterface != nil {
+		schedulers.schedulers = append(schedulers.schedulers, fixCRTChannelUnreadsInterface.MakeScheduler())
+	}
+
 	schedulers.nextRunTimes = make([]*time.Time, len(schedulers.schedulers))
 	srv.schedulers = schedulers
 

jobs/server.go

@@ -37,6 +37,7 @@ type JobServer struct {
 	Cloud                   ejobs.CloudJobInterface
 	ResendInvitationEmails  ejobs.ResendInvitationEmailJobInterface
 	ExtractContent          tjobs.ExtractContentInterface
+	FixCRTChannelUnreads    tjobs.FixCRTChannelUnreadsJobInterface
 
 	// mut is used to protect the following fields from concurrent access.
 	mut        sync.Mutex

jobs/workers.go

@@ -33,6 +33,7 @@ type Workers struct {
 	Cloud                    model.Worker
 	ResendInvitationEmail    model.Worker
 	ExtractContent           model.Worker
+	FixCRTChannelUnreads     model.Worker
 
 	listenerId string
 	running    bool
@@ -129,6 +130,10 @@ func (srv *JobServer) InitWorkers() error {
 		workers.ExtractContent = extractContentInterface.MakeWorker()
 	}
 
+	if fixCRTChannelUnreads := srv.FixCRTChannelUnreads; fixCRTChannelUnreads != nil {
+		workers.FixCRTChannelUnreads = fixCRTChannelUnreads.MakeWorker()
+	}
+
 	srv.workers = workers
 
 	return nil
@@ -211,6 +216,10 @@ func (workers *Workers) Start() {
 		go workers.ExtractContent.Run()
 	}
 
+	if workers.FixCRTChannelUnreads != nil {
+		go workers.FixCRTChannelUnreads.Run()
+	}
+
 	go workers.Watcher.Start()
 
 	workers.listenerId = workers.ConfigService.AddConfigListener(workers.handleConfigChange)
@@ -348,6 +357,10 @@ func (workers *Workers) Stop() {
 		workers.ExtractContent.Stop()
 	}
 
+	if workers.FixCRTChannelUnreads != nil {
+		workers.FixCRTChannelUnreads.Stop()
+	}
+
 	workers.running = false
 
 	mlog.Info("Stopped workers")

model/job.go

@@ -27,6 +27,7 @@ const (
 	JobTypeCloud                        = "cloud"
 	JobTypeResendInvitationEmail        = "resend_invitation_email"
 	JobTypeExtractContent               = "extract_content"
+	JobTypeFixChannelUnreadsForCRT      = "fix_channel_unreads_for_crt"
 
 	JobStatusPending         = "pending"
 	JobStatusInProgress      = "in_progress"
@@ -55,6 +56,7 @@ var AllJobTypes = [...]string{
 	JobTypeExportDelete,
 	JobTypeCloud,
 	JobTypeExtractContent,
+	JobTypeFixChannelUnreadsForCRT,
 }
 
 type Job struct {
@@ -97,6 +99,7 @@ func (j *Job) IsValid() *AppError {
 	case JobTypeCloud:
 	case JobTypeResendInvitationEmail:
 	case JobTypeExtractContent:
+	case JobTypeFixChannelUnreadsForCRT:
 	default:
 		return NewAppError("Job.IsValid", "model.job.is_valid.type.app_error", nil, "id="+j.Id, http.StatusBadRequest)
 	}

model/migration.go

@@ -35,4 +35,5 @@ const (
 	MigrationKeyAddTestEmailAncillaryPermission        = "test_email_ancillary_permission"
 	MigrationKeyAddAboutSubsectionPermissions          = "about_subsection_permissions"
 	MigrationKeyAddIntegrationsSubsectionPermissions   = "integrations_subsection_permissions"
+	MigrationKeyFixCRTChannelUnreads                   = "fix_crt_channel_unreads"
 )

store/opentracinglayer/opentracinglayer.go

@@ -1015,6 +1015,24 @@ func (s *OpenTracingLayerChannelStore) GetByNames(team_id string, names []string
 	return result, err
 }
 
+func (s *OpenTracingLayerChannelStore) GetCRTUnfixedChannelMembershipsAfter(channelID string, userID string, count int) ([]model.ChannelMember, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelStore.GetCRTUnfixedChannelMembershipsAfter")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.ChannelStore.GetCRTUnfixedChannelMembershipsAfter(channelID, userID, count)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerChannelStore) GetChannelCounts(teamID string, userID string) (*model.ChannelCounts, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "ChannelStore.GetChannelCounts")
@@ -5646,6 +5664,24 @@ func (s *OpenTracingLayerPostStore) GetSingle(id string, inclDeleted bool) (*mod
 	return result, err
 }
 
+func (s *OpenTracingLayerPostStore) GetUniquePostTypesSince(channelId string, timestamp int64) ([]string, error) {
+	origCtx := s.Root.Store.Context()
+	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.GetUniquePostTypesSince")
+	s.Root.Store.SetContext(newCtx)
+	defer func() {
+		s.Root.Store.SetContext(origCtx)
+	}()
+
+	defer span.Finish()
+	result, err := s.PostStore.GetUniquePostTypesSince(channelId, timestamp)
+	if err != nil {
+		span.LogFields(spanlog.Error(err))
+		ext.Error.Set(span, true)
+	}
+
+	return result, err
+}
+
 func (s *OpenTracingLayerPostStore) HasAutoResponsePostByUserSince(options model.GetPostsSinceOptions, userId string) (bool, error) {
 	origCtx := s.Root.Store.Context()
 	span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "PostStore.HasAutoResponsePostByUserSince")

store/retrylayer/retrylayer.go

@@ -1129,6 +1129,27 @@ func (s *RetryLayerChannelStore) GetByNames(team_id string, names []string, allo
 
 }
 
+func (s *RetryLayerChannelStore) GetCRTUnfixedChannelMembershipsAfter(channelID string, userID string, count int) ([]model.ChannelMember, error) {
+
+	tries := 0
+	for {
+		result, err := s.ChannelStore.GetCRTUnfixedChannelMembershipsAfter(channelID, userID, count)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+		timepkg.Sleep(100 * timepkg.Millisecond)
+	}
+
+}
+
 func (s *RetryLayerChannelStore) GetChannelCounts(teamID string, userID string) (*model.ChannelCounts, error) {
 
 	tries := 0
@@ -6385,6 +6406,27 @@ func (s *RetryLayerPostStore) GetSingle(id string, inclDeleted bool) (*model.Pos
 
 }
 
+func (s *RetryLayerPostStore) GetUniquePostTypesSince(channelId string, timestamp int64) ([]string, error) {
+
+	tries := 0
+	for {
+		result, err := s.PostStore.GetUniquePostTypesSince(channelId, timestamp)
+		if err == nil {
+			return result, nil
+		}
+		if !isRepeatableError(err) {
+			return result, err
+		}
+		tries++
+		if tries >= 3 {
+			err = errors.Wrap(err, "giving up after 3 consecutive repeatable transaction failures")
+			return result, err
+		}
+		timepkg.Sleep(100 * timepkg.Millisecond)
+	}
+
+}
+
 func (s *RetryLayerPostStore) HasAutoResponsePostByUserSince(options model.GetPostsSinceOptions, userId string) (bool, error) {
 
 	tries := 0

store/sqlstore/channel_store.go

@@ -2128,7 +2128,7 @@ func (s SqlChannelStore) GetMemberCountsByGroup(ctx context.Context, channelID s
 
 	if includeTimezones {
 		if s.DriverName() == model.DatabaseDriverMysql {
-			selectStr += `, 
+			selectStr += `,
 				COUNT(DISTINCT
 				(
 					CASE WHEN Timezone->"$.useAutomaticTimezone" = 'true' AND LENGTH(JSON_UNQUOTE(Timezone->"$.automaticTimezone")) > 0
@@ -2138,7 +2138,7 @@ func (s SqlChannelStore) GetMemberCountsByGroup(ctx context.Context, channelID s
 					END
 				)) AS ChannelMemberTimezonesCount`
 		} else if s.DriverName() == model.DatabaseDriverPostgres {
-			selectStr += `, 
+			selectStr += `,
 				COUNT(DISTINCT
 				(
 					CASE WHEN Timezone->>'useAutomaticTimezone' = 'true' AND length(Timezone->>'automaticTimezone') > 0
@@ -3753,3 +3753,32 @@ func (s SqlChannelStore) GetTeamForChannel(channelID string) (*model.Team, error
 	}
 	return &team, nil
 }
+
+func (s SqlChannelStore) GetCRTUnfixedChannelMembershipsAfter(channelID, userID string, count int) ([]model.ChannelMember, error) {
+	// we want both channelID and userID, or neither of them specified
+	if (userID == "" || channelID == "") && (channelID != userID) {
+		return nil, fmt.Errorf("channelID=%q userID=%q, got one empty param, both need to be empty or specified", channelID, userID)
+	}
+	getUnfixedCMQuery := `
+			SELECT ChannelMembers.*
+			FROM ChannelMembers, Channels
+			WHERE ChannelId = Id AND (ChannelMembers.UserId, ChannelMembers.ChannelId) > (:userId, :channelId) AND Channels.TotalMsgCountRoot > ChannelMembers.MsgCountRoot
+			ORDER BY UserId, ChannelId
+			LIMIT :count;
+	`
+	if userID == "" && channelID == "" {
+		getUnfixedCMQuery = `
+			SELECT ChannelMembers.*
+			FROM ChannelMembers, Channels
+			WHERE ChannelId = Id AND Channels.TotalMsgCountRoot > ChannelMembers.MsgCountRoot
+			ORDER BY UserId, ChannelId
+			LIMIT :count;
+		`
+	}
+	var cms []model.ChannelMember
+
+	if _, err := s.GetReplica().Select(&cms, getUnfixedCMQuery, map[string]interface{}{"channelId": channelID, "userId": userID, "count": count}); err != nil {
+		return nil, errors.Wrapf(err, "failed to %d ChannelMembers after channelId=%q and userId=%q", count, channelID, userID)
+	}
+	return cms, nil
+}

store/sqlstore/post_store.go

@@ -2522,3 +2522,22 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *gorp.Transaction, pos
 	}
 	return nil
 }
+
+// GetUniquePostTypesSince returns the unique post types in a channel after the given timestamp
+func (s *SqlPostStore) GetUniquePostTypesSince(channelId string, timestamp int64) ([]string, error) {
+	query, args, err := s.getQueryBuilder().
+		Select("DISTINCT Type").
+		From("Posts").
+		Where(sq.And{
+			sq.Eq{"ChannelId": channelId},
+			sq.GtOrEq{"CreateAt": timestamp},
+		}).ToSql()
+	if err != nil {
+		return nil, err
+	}
+	var types []string
+	if _, err := s.GetReplica().Select(&types, query, args...); err != nil {
+		return nil, err
+	}
+	return types, nil
+}

store/store.go

@@ -276,6 +276,9 @@ type ChannelStore interface {
 	SetShared(channelId string, shared bool) error
 	// GetTeamForChannel returns the team for a given channelID.
 	GetTeamForChannel(channelID string) (*model.Team, error)
+
+	//GetCRTUnfixedChannelMembershipsAfter gets CRT unfixed channel memberships after the given channelID and userID
+	GetCRTUnfixedChannelMembershipsAfter(channelID string, userID string, count int) ([]model.ChannelMember, error)
 }
 
 type ChannelMemberHistoryStore interface {
@@ -359,6 +362,9 @@ type PostStore interface {
 	GetOldestEntityCreationTime() (int64, error)
 	HasAutoResponsePostByUserSince(options model.GetPostsSinceOptions, userId string) (bool, error)
 	GetPostsSinceForSync(options model.GetPostsSinceForSyncOptions, cursor model.GetPostsSinceForSyncCursor, limit int) ([]*model.Post, model.GetPostsSinceForSyncCursor, error)
+
+	// GetUniquePostTypesSince returns the unique post types in a channel after the given timestamp
+	GetUniquePostTypesSince(channelId string, timestamp int64) ([]string, error)
 }
 
 type UserStore interface {

store/storetest/mocks/ChannelStore.go

@@ -582,6 +582,29 @@ func (_m *ChannelStore) GetByNames(team_id string, names []string, allowFromCach
 	return r0, r1
 }
 
+// GetCRTUnfixedChannelMembershipsAfter provides a mock function with given fields: channelID, userID, count
+func (_m *ChannelStore) GetCRTUnfixedChannelMembershipsAfter(channelID string, userID string, count int) ([]model.ChannelMember, error) {
+	ret := _m.Called(channelID, userID, count)
+
+	var r0 []model.ChannelMember
+	if rf, ok := ret.Get(0).(func(string, string, int) []model.ChannelMember); ok {
+		r0 = rf(channelID, userID, count)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]model.ChannelMember)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string, string, int) error); ok {
+		r1 = rf(channelID, userID, count)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetChannelCounts provides a mock function with given fields: teamID, userID
 func (_m *ChannelStore) GetChannelCounts(teamID string, userID string) (*model.ChannelCounts, error) {
 	ret := _m.Called(teamID, userID)

store/storetest/mocks/PostStore.go

@@ -635,6 +635,29 @@ func (_m *PostStore) GetSingle(id string, inclDeleted bool) (*model.Post, error)
 	return r0, r1
 }
 
+// GetUniquePostTypesSince provides a mock function with given fields: channelId, timestamp
+func (_m *PostStore) GetUniquePostTypesSince(channelId string, timestamp int64) ([]string, error) {
+	ret := _m.Called(channelId, timestamp)
+
+	var r0 []string
+	if rf, ok := ret.Get(0).(func(string, int64) []string); ok {
+		r0 = rf(channelId, timestamp)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).([]string)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(string, int64) error); ok {
+		r1 = rf(channelId, timestamp)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // HasAutoResponsePostByUserSince provides a mock function with given fields: options, userId
 func (_m *PostStore) HasAutoResponsePostByUserSince(options model.GetPostsSinceOptions, userId string) (bool, error) {
 	ret := _m.Called(options, userId)

store/timerlayer/timerlayer.go

@@ -949,6 +949,22 @@ func (s *TimerLayerChannelStore) GetByNames(team_id string, names []string, allo
 	return result, err
 }
 
+func (s *TimerLayerChannelStore) GetCRTUnfixedChannelMembershipsAfter(channelID string, userID string, count int) ([]model.ChannelMember, error) {
+	start := timemodule.Now()
+
+	result, err := s.ChannelStore.GetCRTUnfixedChannelMembershipsAfter(channelID, userID, count)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("ChannelStore.GetCRTUnfixedChannelMembershipsAfter", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerChannelStore) GetChannelCounts(teamID string, userID string) (*model.ChannelCounts, error) {
 	start := timemodule.Now()
 
@@ -5114,6 +5130,22 @@ func (s *TimerLayerPostStore) GetSingle(id string, inclDeleted bool) (*model.Pos
 	return result, err
 }
 
+func (s *TimerLayerPostStore) GetUniquePostTypesSince(channelId string, timestamp int64) ([]string, error) {
+	start := timemodule.Now()
+
+	result, err := s.PostStore.GetUniquePostTypesSince(channelId, timestamp)
+
+	elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second)
+	if s.Root.Metrics != nil {
+		success := "false"
+		if err == nil {
+			success = "true"
+		}
+		s.Root.Metrics.ObserveStoreMethodDuration("PostStore.GetUniquePostTypesSince", success, elapsed)
+	}
+	return result, err
+}
+
 func (s *TimerLayerPostStore) HasAutoResponsePostByUserSince(options model.GetPostsSinceOptions, userId string) (bool, error) {
 	start := timemodule.Now()