From 0da249c6517f24946f5bc7b6062b98fd3628c712 Mon Sep 17 00:00:00 2001 From: Ashish Bhate Date: Thu, 18 Nov 2021 15:31:18 +0530 Subject: [PATCH] [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 | 6 + app/server.go | 4 + i18n/en.json | 16 + imports/placeholder.go | 3 + jobs/fix_crt_channel_unreads/scheduler.go | 67 +++++ jobs/fix_crt_channel_unreads/worker.go | 273 ++++++++++++++++++ .../fix_crt_channel_unreads_interface.go | 13 + jobs/jobs_watcher.go | 7 + jobs/schedulers.go | 4 + jobs/server.go | 1 + jobs/workers.go | 13 + model/job.go | 3 + model/migration.go | 1 + store/opentracinglayer/opentracinglayer.go | 36 +++ store/retrylayer/retrylayer.go | 42 +++ store/sqlstore/channel_store.go | 33 ++- store/sqlstore/post_store.go | 19 ++ store/store.go | 6 + store/storetest/mocks/ChannelStore.go | 23 ++ store/storetest/mocks/PostStore.go | 23 ++ store/timerlayer/timerlayer.go | 32 ++ 21 files changed, 623 insertions(+), 2 deletions(-) create mode 100644 jobs/fix_crt_channel_unreads/scheduler.go create mode 100644 jobs/fix_crt_channel_unreads/worker.go create mode 100644 jobs/interfaces/fix_crt_channel_unreads_interface.go diff --git a/app/enterprise.go b/app/enterprise.go index c1d2a778d4..5003d0cfec 100644 --- a/app/enterprise.go +++ b/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) { diff --git a/app/server.go b/app/server.go index f2d47c4abd..91777fee01 100644 --- a/app/server.go +++ b/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() } diff --git a/i18n/en.json b/i18n/en.json index 65afc7923b..8a2cc06aeb 100644 --- a/i18n/en.json +++ b/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." diff --git a/imports/placeholder.go b/imports/placeholder.go index 46d29c79da..780f8ef1cf 100644 --- a/imports/placeholder.go +++ b/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" ) diff --git a/jobs/fix_crt_channel_unreads/scheduler.go b/jobs/fix_crt_channel_unreads/scheduler.go new file mode 100644 index 0000000000..15e0ba06c9 --- /dev/null +++ b/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 +} diff --git a/jobs/fix_crt_channel_unreads/worker.go b/jobs/fix_crt_channel_unreads/worker.go new file mode 100644 index 0000000000..c8d1c5b88a --- /dev/null +++ b/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 +} diff --git a/jobs/interfaces/fix_crt_channel_unreads_interface.go b/jobs/interfaces/fix_crt_channel_unreads_interface.go new file mode 100644 index 0000000000..5463b6e73f --- /dev/null +++ b/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 +} diff --git a/jobs/jobs_watcher.go b/jobs/jobs_watcher.go index d519c428e9..dad7ad1150 100644 --- a/jobs/jobs_watcher.go +++ b/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: + } + } } } } diff --git a/jobs/schedulers.go b/jobs/schedulers.go index 774f4c6fc9..6d5a7a519b 100644 --- a/jobs/schedulers.go +++ b/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 diff --git a/jobs/server.go b/jobs/server.go index b34b073927..ef0da40959 100644 --- a/jobs/server.go +++ b/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 diff --git a/jobs/workers.go b/jobs/workers.go index bb34ed4b97..a8d53fe932 100644 --- a/jobs/workers.go +++ b/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") diff --git a/model/job.go b/model/job.go index e892b051b4..0a8cf00ac8 100644 --- a/model/job.go +++ b/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) } diff --git a/model/migration.go b/model/migration.go index 2e0efb46c0..23afe1237c 100644 --- a/model/migration.go +++ b/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" ) diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index dffe769526..0d27fbfea2 100644 --- a/store/opentracinglayer/opentracinglayer.go +++ b/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") diff --git a/store/retrylayer/retrylayer.go b/store/retrylayer/retrylayer.go index 3a096db82b..78e7c2616a 100644 --- a/store/retrylayer/retrylayer.go +++ b/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 diff --git a/store/sqlstore/channel_store.go b/store/sqlstore/channel_store.go index 89cdde8f81..e1a076486e 100644 --- a/store/sqlstore/channel_store.go +++ b/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 +} diff --git a/store/sqlstore/post_store.go b/store/sqlstore/post_store.go index 0364ca3802..31428a7740 100644 --- a/store/sqlstore/post_store.go +++ b/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 +} diff --git a/store/store.go b/store/store.go index 5d85261ae3..267a884937 100644 --- a/store/store.go +++ b/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 { diff --git a/store/storetest/mocks/ChannelStore.go b/store/storetest/mocks/ChannelStore.go index c304742824..77fb8c7e0e 100644 --- a/store/storetest/mocks/ChannelStore.go +++ b/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) diff --git a/store/storetest/mocks/PostStore.go b/store/storetest/mocks/PostStore.go index 5b91492e82..6896a6b74b 100644 --- a/store/storetest/mocks/PostStore.go +++ b/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) diff --git a/store/timerlayer/timerlayer.go b/store/timerlayer/timerlayer.go index d94e287716..3c1b8ee3bb 100644 --- a/store/timerlayer/timerlayer.go +++ b/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()