diff --git a/app/app_iface.go b/app/app_iface.go index afbeddd89d..72f99abd56 100644 --- a/app/app_iface.go +++ b/app/app_iface.go @@ -817,6 +817,7 @@ type AppIface interface { ImportPermissions(jsonl io.Reader) error InitPlugins(c *request.Context, pluginDir, webappPluginDir string) InvalidateAllEmailInvites() *model.AppError + InvalidateAllResendInviteEmailJobs() *model.AppError InvalidateCacheForUser(userID string) InviteGuestsToChannels(teamID string, guestsInvite *model.GuestsInvite, senderId string) *model.AppError InviteGuestsToChannelsGracefully(teamID string, guestsInvite *model.GuestsInvite, senderId string) ([]*model.EmailInviteWithError, *model.AppError) diff --git a/app/opentracing/opentracing_layer.go b/app/opentracing/opentracing_layer.go index 67a39f8e5b..bbd0d778af 100644 --- a/app/opentracing/opentracing_layer.go +++ b/app/opentracing/opentracing_layer.go @@ -10912,6 +10912,28 @@ func (a *OpenTracingAppLayer) InvalidateAllEmailInvites() *model.AppError { return resultVar0 } +func (a *OpenTracingAppLayer) InvalidateAllResendInviteEmailJobs() *model.AppError { + origCtx := a.ctx + span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.InvalidateAllResendInviteEmailJobs") + + a.ctx = newCtx + a.app.Srv().Store.SetContext(newCtx) + defer func() { + a.app.Srv().Store.SetContext(origCtx) + a.ctx = origCtx + }() + + defer span.Finish() + resultVar0 := a.app.InvalidateAllResendInviteEmailJobs() + + if resultVar0 != nil { + span.LogFields(spanlog.Error(resultVar0)) + ext.Error.Set(span, true) + } + + return resultVar0 +} + func (a *OpenTracingAppLayer) InvalidateCacheForUser(userID string) { origCtx := a.ctx span, newCtx := tracing.StartSpanWithParentByContext(a.ctx, "app.InvalidateCacheForUser") diff --git a/app/team.go b/app/team.go index 34966f0426..bcb5b04006 100644 --- a/app/team.go +++ b/app/team.go @@ -1901,11 +1901,29 @@ func (a *App) RemoveTeamIcon(teamID string) *model.AppError { func (a *App) InvalidateAllEmailInvites() *model.AppError { if err := a.Srv().Store.Token().RemoveAllTokensByType(TokenTypeTeamInvitation); err != nil { - return model.NewAppError("InvalidateAllEmailInvites", "api.team.invalidate_all_email_invites.app_error", nil, err.Error(), http.StatusBadRequest) + return model.NewAppError("InvalidateAllEmailInvites", "api.team.invalidate_all_email_invites.app_error", nil, err.Error(), http.StatusInternalServerError) } if err := a.Srv().Store.Token().RemoveAllTokensByType(TokenTypeGuestInvitation); err != nil { - return model.NewAppError("InvalidateAllEmailInvites", "api.team.invalidate_all_email_invites.app_error", nil, err.Error(), http.StatusBadRequest) + return model.NewAppError("InvalidateAllEmailInvites", "api.team.invalidate_all_email_invites.app_error", nil, err.Error(), http.StatusInternalServerError) } + if err := a.InvalidateAllResendInviteEmailJobs(); err != nil { + return model.NewAppError("InvalidateAllEmailInvites", "api.team.invalidate_all_email_invites.app_error", nil, err.Error(), http.StatusInternalServerError) + } + return nil +} + +func (a *App) InvalidateAllResendInviteEmailJobs() *model.AppError { + jobs, appErr := a.Srv().Jobs.GetJobsByTypeAndStatus(model.JobTypeResendInvitationEmail, model.JobStatusPending) + if appErr != nil { + return appErr + } + + for _, j := range jobs { + a.Srv().Jobs.SetJobCanceled(j) + // clean up any system values this job was using + a.Srv().Store.System().PermanentDeleteByName(j.Id) + } + return nil } diff --git a/app/team_test.go b/app/team_test.go index 1699c47a3b..bc79122ac4 100644 --- a/app/team_test.go +++ b/app/team_test.go @@ -19,6 +19,7 @@ import ( "github.com/mattermost/mattermost-server/v6/app/teams" "github.com/mattermost/mattermost-server/v6/app/users" "github.com/mattermost/mattermost-server/v6/model" + "github.com/mattermost/mattermost-server/v6/store" "github.com/mattermost/mattermost-server/v6/store/sqlstore" "github.com/mattermost/mattermost-server/v6/store/storetest/mocks" ) @@ -1152,6 +1153,29 @@ func TestUpdateTeamMemberRolesChangingGuest(t *testing.T) { }) } +func TestInvalidateAllResendInviteEmailJobs(t *testing.T) { + th := Setup(t) + defer th.TearDown() + + job, err := th.App.Srv().Jobs.CreateJob(model.JobTypeResendInvitationEmail, map[string]string{}) + require.Nil(t, err) + + sysVar := &model.System{Name: job.Id, Value: "0"} + e := th.App.Srv().Store.System().SaveOrUpdate(sysVar) + require.NoError(t, e) + + appErr := th.App.InvalidateAllResendInviteEmailJobs() + require.Nil(t, appErr) + + j, e := th.App.Srv().Store.Job().Get(job.Id) + require.NoError(t, e) + require.Equal(t, j.Status, model.JobStatusCanceled) + + _, sysValErr := th.App.Srv().Store.System().GetByName(job.Id) + var errNotFound *store.ErrNotFound + require.ErrorAs(t, sysValErr, &errNotFound) +} + func TestInvalidateAllEmailInvites(t *testing.T) { th := Setup(t) defer th.TearDown() diff --git a/i18n/en.json b/i18n/en.json index a0be67dcf8..c942c5594f 100644 --- a/i18n/en.json +++ b/i18n/en.json @@ -5539,6 +5539,10 @@ "id": "app.job.get_all.app_error", "translation": "Unable to get the jobs." }, + { + "id": "app.job.get_all_jobs_by_type_and_status.app_error", + "translation": "Unable to get the all jobs by type and status." + }, { "id": "app.job.get_count_by_status_and_type.app_error", "translation": "Unable to get the job count by status and type." diff --git a/jobs/jobs.go b/jobs/jobs.go index bf0dc4a59f..5e739b3d9d 100644 --- a/jobs/jobs.go +++ b/jobs/jobs.go @@ -230,6 +230,15 @@ func (srv *JobServer) CheckForPendingJobsByType(jobType string) (bool, *model.Ap return count > 0, nil } +func (srv *JobServer) GetJobsByTypeAndStatus(jobType string, status string) ([]*model.Job, *model.AppError) { + jobs, err := srv.Store.Job().GetAllByTypeAndStatus(jobType, status) + if err != nil { + return nil, model.NewAppError("GetJobsByTypeAndStatus", "app.job.get_all_jobs_by_type_and_status.app_error", nil, err.Error(), http.StatusInternalServerError) + } + + return jobs, nil +} + func (srv *JobServer) GetLastSuccessfulJobByType(jobType string) (*model.Job, *model.AppError) { statuses := []string{model.JobStatusSuccess} if jobType == model.JobTypeMessageExport { diff --git a/store/opentracinglayer/opentracinglayer.go b/store/opentracinglayer/opentracinglayer.go index ad7f352f2a..fc63e95e37 100644 --- a/store/opentracinglayer/opentracinglayer.go +++ b/store/opentracinglayer/opentracinglayer.go @@ -4501,6 +4501,24 @@ func (s *OpenTracingLayerJobStore) GetAllByType(jobType string) ([]*model.Job, e return result, err } +func (s *OpenTracingLayerJobStore) GetAllByTypeAndStatus(jobType string, status string) ([]*model.Job, error) { + origCtx := s.Root.Store.Context() + span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "JobStore.GetAllByTypeAndStatus") + s.Root.Store.SetContext(newCtx) + defer func() { + s.Root.Store.SetContext(origCtx) + }() + + defer span.Finish() + result, err := s.JobStore.GetAllByTypeAndStatus(jobType, status) + if err != nil { + span.LogFields(spanlog.Error(err)) + ext.Error.Set(span, true) + } + + return result, err +} + func (s *OpenTracingLayerJobStore) GetAllByTypePage(jobType string, offset int, limit int) ([]*model.Job, error) { origCtx := s.Root.Store.Context() span, newCtx := tracing.StartSpanWithParentByContext(s.Root.Store.Context(), "JobStore.GetAllByTypePage") diff --git a/store/retrylayer/retrylayer.go b/store/retrylayer/retrylayer.go index db408ce154..08ce43db17 100644 --- a/store/retrylayer/retrylayer.go +++ b/store/retrylayer/retrylayer.go @@ -5086,6 +5086,27 @@ func (s *RetryLayerJobStore) GetAllByType(jobType string) ([]*model.Job, error) } +func (s *RetryLayerJobStore) GetAllByTypeAndStatus(jobType string, status string) ([]*model.Job, error) { + + tries := 0 + for { + result, err := s.JobStore.GetAllByTypeAndStatus(jobType, status) + 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 *RetryLayerJobStore) GetAllByTypePage(jobType string, offset int, limit int) ([]*model.Job, error) { tries := 0 diff --git a/store/sqlstore/job_store.go b/store/sqlstore/job_store.go index 99152d15ca..3ba7c9e085 100644 --- a/store/sqlstore/job_store.go +++ b/store/sqlstore/job_store.go @@ -201,6 +201,22 @@ func (jss SqlJobStore) GetAllByType(jobType string) ([]*model.Job, error) { return statuses, nil } +func (jss SqlJobStore) GetAllByTypeAndStatus(jobType string, status string) ([]*model.Job, error) { + query, args, err := jss.getQueryBuilder(). + Select("*"). + From("Jobs"). + Where(sq.Eq{"Type": jobType, "Status": status}). + OrderBy("CreateAt DESC").ToSql() + if err != nil { + return nil, errors.Wrap(err, "job_tosql") + } + jobs := []*model.Job{} + if err = jss.GetReplicaX().Select(&jobs, query, args...); err != nil { + return nil, errors.Wrapf(err, "failed to find Jobs with type=%s", jobType) + } + return jobs, nil +} + func (jss SqlJobStore) GetAllByTypePage(jobType string, offset int, limit int) ([]*model.Job, error) { query, args, err := jss.getQueryBuilder(). Select("*"). diff --git a/store/store.go b/store/store.go index 37c5a9b193..ec26fa24bf 100644 --- a/store/store.go +++ b/store/store.go @@ -689,6 +689,7 @@ type JobStore interface { Get(id string) (*model.Job, error) GetAllPage(offset int, limit int) ([]*model.Job, error) GetAllByType(jobType string) ([]*model.Job, error) + GetAllByTypeAndStatus(jobType string, status string) ([]*model.Job, error) GetAllByTypePage(jobType string, offset int, limit int) ([]*model.Job, error) GetAllByTypesPage(jobTypes []string, offset int, limit int) ([]*model.Job, error) GetAllByStatus(status string) ([]*model.Job, error) diff --git a/store/storetest/job_store.go b/store/storetest/job_store.go index 5ac9d4051d..e5c88453b5 100644 --- a/store/storetest/job_store.go +++ b/store/storetest/job_store.go @@ -19,6 +19,7 @@ import ( func TestJobStore(t *testing.T, ss store.Store) { t.Run("JobSaveGet", func(t *testing.T) { testJobSaveGet(t, ss) }) t.Run("JobGetAllByType", func(t *testing.T) { testJobGetAllByType(t, ss) }) + t.Run("JobGetAllByTypeAndStatus", func(t *testing.T) { testJobGetAllByTypeAndStatus(t, ss) }) t.Run("JobGetAllByTypePage", func(t *testing.T) { testJobGetAllByTypePage(t, ss) }) t.Run("JobGetAllByTypesPage", func(t *testing.T) { testJobGetAllByTypesPage(t, ss) }) t.Run("JobGetAllPage", func(t *testing.T) { testJobGetAllPage(t, ss) }) @@ -85,6 +86,34 @@ func testJobGetAllByType(t *testing.T, ss store.Store) { require.ElementsMatch(t, []string{jobs[0].Id, jobs[1].Id}, []string{received[0].Id, received[1].Id}) } +func testJobGetAllByTypeAndStatus(t *testing.T, ss store.Store) { + jobType := model.NewId() + + jobs := []*model.Job{ + { + Id: model.NewId(), + Type: jobType, + Status: model.JobStatusPending, + }, + { + Id: model.NewId(), + Type: jobType, + Status: model.JobStatusPending, + }, + } + + for _, job := range jobs { + _, err := ss.Job().Save(job) + require.NoError(t, err) + defer ss.Job().Delete(job.Id) + } + + received, err := ss.Job().GetAllByTypeAndStatus(jobType, model.JobStatusPending) + require.NoError(t, err) + require.Len(t, received, 2) + require.ElementsMatch(t, []string{jobs[0].Id, jobs[1].Id}, []string{received[0].Id, received[1].Id}) +} + func testJobGetAllByTypePage(t *testing.T, ss store.Store) { jobType := model.NewId() diff --git a/store/storetest/mocks/JobStore.go b/store/storetest/mocks/JobStore.go index 341841da49..86041b40d7 100644 --- a/store/storetest/mocks/JobStore.go +++ b/store/storetest/mocks/JobStore.go @@ -118,6 +118,29 @@ func (_m *JobStore) GetAllByType(jobType string) ([]*model.Job, error) { return r0, r1 } +// GetAllByTypeAndStatus provides a mock function with given fields: jobType, status +func (_m *JobStore) GetAllByTypeAndStatus(jobType string, status string) ([]*model.Job, error) { + ret := _m.Called(jobType, status) + + var r0 []*model.Job + if rf, ok := ret.Get(0).(func(string, string) []*model.Job); ok { + r0 = rf(jobType, status) + } else { + if ret.Get(0) != nil { + r0 = ret.Get(0).([]*model.Job) + } + } + + var r1 error + if rf, ok := ret.Get(1).(func(string, string) error); ok { + r1 = rf(jobType, status) + } else { + r1 = ret.Error(1) + } + + return r0, r1 +} + // GetAllByTypePage provides a mock function with given fields: jobType, offset, limit func (_m *JobStore) GetAllByTypePage(jobType string, offset int, limit int) ([]*model.Job, error) { ret := _m.Called(jobType, offset, limit) diff --git a/store/timerlayer/timerlayer.go b/store/timerlayer/timerlayer.go index fa37d086fa..9950e0ec86 100644 --- a/store/timerlayer/timerlayer.go +++ b/store/timerlayer/timerlayer.go @@ -4091,6 +4091,22 @@ func (s *TimerLayerJobStore) GetAllByType(jobType string) ([]*model.Job, error) return result, err } +func (s *TimerLayerJobStore) GetAllByTypeAndStatus(jobType string, status string) ([]*model.Job, error) { + start := timemodule.Now() + + result, err := s.JobStore.GetAllByTypeAndStatus(jobType, status) + + elapsed := float64(timemodule.Since(start)) / float64(timemodule.Second) + if s.Root.Metrics != nil { + success := "false" + if err == nil { + success = "true" + } + s.Root.Metrics.ObserveStoreMethodDuration("JobStore.GetAllByTypeAndStatus", success, elapsed) + } + return result, err +} + func (s *TimerLayerJobStore) GetAllByTypePage(jobType string, offset int, limit int) ([]*model.Job, error) { start := timemodule.Now()