Channel index fix (#22947)

* WIP

* Fetched ES index for type checking

* WIP

* Moved migration key to model

* CI

* CI

* Fixed existing tests

* Updated einterface mocks

* CI

* WIP

* Reverted test makefile changes

* Fixed error level

* CI

* Not attempting to launch job if not enterprise ready

* CI

* Fixing job trigger condition

* CI

* Updated eemocks

* CI

server/channels/app/enterprise.go

@@ -50,6 +50,12 @@ func RegisterJobsElasticsearchIndexerInterface(f func(*Server) ejobs.IndexerJobI
 	jobsElasticsearchIndexerInterface = f
 }
 
+var jobsElasticsearchFixChannelIndexInterface func(*Server) ejobs.ElasticsearchFixChannelIndexInterface
+
+func RegisterJobsElasticsearchFixChannelIndexInterface(f func(*Server) ejobs.ElasticsearchFixChannelIndexInterface) {
+	jobsElasticsearchFixChannelIndexInterface = f
+}
+
 var jobsLdapSyncInterface func(*App) ejobs.LdapSyncInterface
 
 func RegisterJobsLdapSyncInterface(f func(*App) ejobs.LdapSyncInterface) {

server/channels/app/migrations.go

@@ -559,6 +559,24 @@ func (s *Server) doPostPriorityConfigDefaultTrueMigration() {
 	}
 }
 
+func (s *Server) doElasticsearchFixChannelIndex() {
+	// If the migration is already marked as completed, don't do it again.
+	if _, err := s.Store().System().GetByName(model.MigrationKeyElasticsearchFixChannelIndex); err == nil {
+		return
+	}
+
+	license := s.License()
+	if model.BuildEnterpriseReady != "true" || license == nil || !*license.Features.Elasticsearch {
+		mlog.Info("Skipping triggering Elasticsearch channel index fix job as build is not Enterprise ready")
+		return
+	}
+
+	if _, appErr := s.Jobs.CreateJob(model.JobTypeElasticsearchFixChannelIndex, nil); appErr != nil {
+		mlog.Fatal("failed to start job for fixing Elasticsearch channels index", mlog.Err(appErr))
+		return
+	}
+}
+
 func (a *App) DoAppMigrations() {
 	a.Srv().doAppMigrations()
 }
@@ -580,4 +598,5 @@ func (s *Server) doAppMigrations() {
 	s.doFirstAdminSetupCompleteMigration()
 	s.doRemainingSchemaMigrations()
 	s.doPostPriorityConfigDefaultTrueMigration()
+	s.doElasticsearchFixChannelIndex()
 }

server/channels/app/server.go

@@ -1476,6 +1476,11 @@ func (s *Server) initJobs() {
 		s.Jobs.RegisterJobType(model.JobTypeElasticsearchPostIndexing, builder.MakeWorker(), nil)
 	}
 
+	if jobsElasticsearchFixChannelIndexInterface != nil {
+		builder := jobsElasticsearchFixChannelIndexInterface(s)
+		s.Jobs.RegisterJobType(model.JobTypeElasticsearchFixChannelIndex, builder.MakeWorker(), nil)
+	}
+
 	if jobsLdapSyncInterface != nil {
 		builder := jobsLdapSyncInterface(New(ServerConnector(s.Channels())))
 		s.Jobs.RegisterJobType(model.JobTypeLdapSync, builder.MakeWorker(), builder.MakeScheduler())

server/channels/einterfaces/jobs/elasticsearch.go

@@ -15,3 +15,7 @@ type ElasticsearchAggregatorInterface interface {
 	MakeWorker() model.Worker
 	MakeScheduler() model.Scheduler
 }
+
+type ElasticsearchFixChannelIndexInterface interface {
+	MakeWorker() model.Worker
+}

server/channels/einterfaces/mocks/ElasticsearchFixChannelIndexInterface.go

@@ -0,0 +1,46 @@
+// Code generated by mockery v2.23.2. DO NOT EDIT.
+
+// Regenerate this file using `make einterfaces-mocks`.
+
+package mocks
+
+import (
+	model "github.com/mattermost/mattermost-server/server/v8/model"
+	mock "github.com/stretchr/testify/mock"
+)
+
+// ElasticsearchFixChannelIndexInterface is an autogenerated mock type for the ElasticsearchFixChannelIndexInterface type
+type ElasticsearchFixChannelIndexInterface struct {
+	mock.Mock
+}
+
+// MakeWorker provides a mock function with given fields:
+func (_m *ElasticsearchFixChannelIndexInterface) MakeWorker() model.Worker {
+	ret := _m.Called()
+
+	var r0 model.Worker
+	if rf, ok := ret.Get(0).(func() model.Worker); ok {
+		r0 = rf()
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(model.Worker)
+		}
+	}
+
+	return r0
+}
+
+type mockConstructorTestingTNewElasticsearchFixChannelIndexInterface interface {
+	mock.TestingT
+	Cleanup(func())
+}
+
+// NewElasticsearchFixChannelIndexInterface creates a new instance of ElasticsearchFixChannelIndexInterface. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
+func NewElasticsearchFixChannelIndexInterface(t mockConstructorTestingTNewElasticsearchFixChannelIndexInterface) *ElasticsearchFixChannelIndexInterface {
+	mock := &ElasticsearchFixChannelIndexInterface{}
+	mock.Mock.Test(t)
+
+	t.Cleanup(func() { mock.AssertExpectations(t) })
+
+	return mock
+}

server/channels/testlib/store.go

@@ -72,6 +72,7 @@ func GetMockStoreForSetupFunctions() *mocks.Store {
 	systemStore.On("GetByName", model.MigrationKeyAddCustomUserGroupsPermissionRestore).Return(&model.System{Name: model.MigrationKeyAddCustomUserGroupsPermissionRestore, Value: "true"}, nil)
 	systemStore.On("GetByName", "CustomGroupAdminRoleCreationMigrationComplete").Return(&model.System{Name: model.MigrationKeyAddPlayboosksManageRolesPermissions, Value: "true"}, nil)
 	systemStore.On("GetByName", "products_boards").Return(&model.System{Name: "products_boards", Value: "true"}, nil)
+	systemStore.On("GetByName", "elasticsearch_fix_channel_index_migration_complete").Return(&model.System{Name: "elasticsearch_fix_channel_index_migration_complete", Value: "true"}, nil)
 	systemStore.On("InsertIfExists", mock.AnythingOfType("*model.System")).Return(&model.System{}, nil).Once()
 	systemStore.On("Save", mock.AnythingOfType("*model.System")).Return(nil)
 

server/model/job.go

@@ -13,6 +13,7 @@ const (
 	JobTypeMessageExport                = "message_export"
 	JobTypeElasticsearchPostIndexing    = "elasticsearch_post_indexing"
 	JobTypeElasticsearchPostAggregation = "elasticsearch_post_aggregation"
+	JobTypeElasticsearchFixChannelIndex = "elasticsearch_fix_channel_index"
 	JobTypeBlevePostIndexing            = "bleve_post_indexing"
 	JobTypeLdapSync                     = "ldap_sync"
 	JobTypeMigrations                   = "migrations"

server/model/migration.go

@@ -40,4 +40,5 @@ const (
 	MigrationKeyAddPlayboosksManageRolesPermissions    = "playbooks_manage_roles"
 	MigrationKeyAddProductsBoardsPermissions           = "products_boards"
 	MigrationKeyAddCustomUserGroupsPermissionRestore   = "custom_groups_permission_restore"
+	MigrationKeyElasticsearchFixChannelIndex           = "elasticsearch_fix_channel_index_migration_complete"
 )