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()