diff --git a/app/enterprise.go b/app/enterprise.go index b8e0ed0d70..7d912278a2 100644 --- a/app/enterprise.go +++ b/app/enterprise.go @@ -30,12 +30,6 @@ func RegisterComplianceInterface(f func(*App) einterfaces.ComplianceInterface) { complianceInterface = f } -var fixCRTChannelUnreadsJobInterface func(*Server) tjobs.FixCRTChannelUnreadsJobInterface - -func RegisterFixCRTChannelUnreadsJobInterface(f func(*Server) tjobs.FixCRTChannelUnreadsJobInterface) { - fixCRTChannelUnreadsJobInterface = f -} - var dataRetentionInterface func(*App) einterfaces.DataRetentionInterface func RegisterDataRetentionInterface(f func(*App) einterfaces.DataRetentionInterface) { diff --git a/app/server.go b/app/server.go index 6ff93fae91..8776a1922e 100644 --- a/app/server.go +++ b/app/server.go @@ -1923,10 +1923,6 @@ 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 79d1b3599f..2a57301985 100644 --- a/i18n/en.json +++ b/i18n/en.json @@ -7539,22 +7539,6 @@ "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 780f8ef1cf..46d29c79da 100644 --- a/imports/placeholder.go +++ b/imports/placeholder.go @@ -39,7 +39,4 @@ 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 deleted file mode 100644 index 15e0ba06c9..0000000000 --- a/jobs/fix_crt_channel_unreads/scheduler.go +++ /dev/null @@ -1,67 +0,0 @@ -// 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 deleted file mode 100644 index c8d1c5b88a..0000000000 --- a/jobs/fix_crt_channel_unreads/worker.go +++ /dev/null @@ -1,273 +0,0 @@ -// 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 deleted file mode 100644 index 5463b6e73f..0000000000 --- a/jobs/interfaces/fix_crt_channel_unreads_interface.go +++ /dev/null @@ -1,13 +0,0 @@ -// 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 dad7ad1150..d519c428e9 100644 --- a/jobs/jobs_watcher.go +++ b/jobs/jobs_watcher.go @@ -198,13 +198,6 @@ 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 6d5a7a519b..774f4c6fc9 100644 --- a/jobs/schedulers.go +++ b/jobs/schedulers.go @@ -101,10 +101,6 @@ 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 ef0da40959..b34b073927 100644 --- a/jobs/server.go +++ b/jobs/server.go @@ -37,7 +37,6 @@ 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 a8d53fe932..bb34ed4b97 100644 --- a/jobs/workers.go +++ b/jobs/workers.go @@ -33,7 +33,6 @@ type Workers struct { Cloud model.Worker ResendInvitationEmail model.Worker ExtractContent model.Worker - FixCRTChannelUnreads model.Worker listenerId string running bool @@ -130,10 +129,6 @@ func (srv *JobServer) InitWorkers() error { workers.ExtractContent = extractContentInterface.MakeWorker() } - if fixCRTChannelUnreads := srv.FixCRTChannelUnreads; fixCRTChannelUnreads != nil { - workers.FixCRTChannelUnreads = fixCRTChannelUnreads.MakeWorker() - } - srv.workers = workers return nil @@ -216,10 +211,6 @@ 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) @@ -357,10 +348,6 @@ 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 0a8cf00ac8..e892b051b4 100644 --- a/model/job.go +++ b/model/job.go @@ -27,7 +27,6 @@ const ( JobTypeCloud = "cloud" JobTypeResendInvitationEmail = "resend_invitation_email" JobTypeExtractContent = "extract_content" - JobTypeFixChannelUnreadsForCRT = "fix_channel_unreads_for_crt" JobStatusPending = "pending" JobStatusInProgress = "in_progress" @@ -56,7 +55,6 @@ var AllJobTypes = [...]string{ JobTypeExportDelete, JobTypeCloud, JobTypeExtractContent, - JobTypeFixChannelUnreadsForCRT, } type Job struct { @@ -99,7 +97,6 @@ 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 361eca69a9..629189bbd7 100644 --- a/model/migration.go +++ b/model/migration.go @@ -35,6 +35,5 @@ const ( MigrationKeyAddTestEmailAncillaryPermission = "test_email_ancillary_permission" MigrationKeyAddAboutSubsectionPermissions = "about_subsection_permissions" MigrationKeyAddIntegrationsSubsectionPermissions = "integrations_subsection_permissions" - MigrationKeyFixCRTChannelUnreads = "fix_crt_channel_unreads" MigrationKeyAddPlaybooksPermissions = "playbooks_permissions" ) diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index 8a2ad3c23b..202e1fbafc 100644 --- a/store/opentracinglayer/opentracinglayer.go +++ b/store/opentracinglayer/opentracinglayer.go @@ -1015,24 +1015,6 @@ 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,24 +5628,6 @@ 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 69f356460c..99941f6c21 100644 --- a/store/retrylayer/retrylayer.go +++ b/store/retrylayer/retrylayer.go @@ -1129,27 +1129,6 @@ 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,27 +6364,6 @@ 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 5f4f6b878a..aba1134758 100644 --- a/store/sqlstore/channel_store.go +++ b/store/sqlstore/channel_store.go @@ -3704,32 +3704,3 @@ 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 dbMembers channelMemberWithSchemeRolesList - - if _, err := s.GetReplica().Select(&dbMembers, getUnfixedCMQuery, map[string]interface{}{"channelId": channelID, "userId": userID, "count": count}); err != nil { - return nil, errors.Wrapf(err, "failed to get %d ChannelMembers after channelId=%q and userId=%q", count, channelID, userID) - } - return dbMembers.ToModel(), nil -} diff --git a/store/sqlstore/post_store.go b/store/sqlstore/post_store.go index 481bc8cfd1..92e689d612 100644 --- a/store/sqlstore/post_store.go +++ b/store/sqlstore/post_store.go @@ -2734,23 +2734,3 @@ func (s *SqlPostStore) updateThreadsFromPosts(transaction *sqlxTxWrapper, posts } 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}, - sq.Eq{"DeleteAt": 0}, - }).ToSql() - if err != nil { - return nil, err - } - types := []string{} - if err := s.GetReplicaX().Select(&types, query, args...); err != nil { - return nil, err - } - return types, nil -} diff --git a/store/store.go b/store/store.go index be6af2ebd6..c56cec4f8b 100644 --- a/store/store.go +++ b/store/store.go @@ -275,9 +275,6 @@ 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 { @@ -361,9 +358,6 @@ 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 25c6a6ec7d..70233def60 100644 --- a/store/storetest/mocks/ChannelStore.go +++ b/store/storetest/mocks/ChannelStore.go @@ -582,29 +582,6 @@ 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 6896a6b74b..5b91492e82 100644 --- a/store/storetest/mocks/PostStore.go +++ b/store/storetest/mocks/PostStore.go @@ -635,29 +635,6 @@ 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 02e1b18ebe..08f53ee1d8 100644 --- a/store/timerlayer/timerlayer.go +++ b/store/timerlayer/timerlayer.go @@ -949,22 +949,6 @@ 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,22 +5098,6 @@ 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()