mirrors/mostlymatter
1
0
Форкнуть 0
Код Задачи Запросы на слияние Действия Пакеты Проекты Релизы Вики Активность

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

Просмотр исходного кода 
This reverts commit 0da249c651.
Этот коммит содержится в:
Ashish Bhate
2022-01-12 18:51:47 +05:30
коммит произвёл GitHub
родитель d0724d8b00
Коммит a55bd001b4
21 изменённых файлов: 0 добавлений и 622 удалений
Скачать Patch файл Скачать Diff файл Показать все файлы Свернуть все файлы

6
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) {

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

16
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."

3
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"
)

67
jobs/fix_crt_channel_unreads/scheduler.go
Просмотреть файл

@@ -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
}

273
jobs/fix_crt_channel_unreads/worker.go
Просмотреть файл

@@ -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
}

13
jobs/interfaces/fix_crt_channel_unreads_interface.go
Просмотреть файл

@@ -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
}

7
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:
}
}
}
}
}

4
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

1
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

13
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")

3
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)
}

1
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"
)

36
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")

42
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

29
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
}

20
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
}

6
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 {

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

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

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