MM-40272: Ability to resume elasticsearch indexing job (#19812)

Due to the way our community deployment is done. The job server
is restarted every day. This means that whenever there is a job
that takes more than 24 hours, it will always get cancelled
when the server restarts and therefore will never finish.

This PR adds ability to resume any stopped jobs, by storing
intermediate progress in the job metadata and setting
the job to pending instead of cancelled when everything is
shut down.

The user can still cancel a job explicitly by clicking on
the cross button in the system console. That functionality
hasn't changed. Only server stop or stopping/starting
job server via config will pause/resume jobs.

```release-note
The elasticsearch indexing job is resumable now. Stopping a
server while the job is running will put the job in pending status
and will resume the job when the server starts.

The job can still be explicitly cancelled via the system console UI.
```

jobs/jobs.go

@@ -155,6 +155,18 @@ func (srv *JobServer) SetJobCanceled(job *model.Job) *model.AppError {
 	return nil
 }
 
+func (srv *JobServer) SetJobPending(job *model.Job) *model.AppError {
+	if _, err := srv.Store.Job().UpdateStatus(job.Id, model.JobStatusPending); err != nil {
+		return model.NewAppError("SetJobPending", "app.job.update.app_error", nil, err.Error(), http.StatusInternalServerError)
+	}
+
+	if srv.metrics != nil {
+		srv.metrics.DecrementJobActive(job.Type)
+	}
+
+	return nil
+}
+
 func (srv *JobServer) UpdateInProgressJobData(job *model.Job) *model.AppError {
 	job.Status = model.JobStatusInProgress
 	job.LastActivityAt = model.GetMillis()

store/sqlstore/channel_store.go

@@ -2712,11 +2712,13 @@ func (s SqlChannelStore) GetForPost(postId string) (*model.Channel, error) {
 }
 
 func (s SqlChannelStore) AnalyticsTypeCount(teamId string, channelType model.ChannelType) (int64, error) {
-
 	query := s.getQueryBuilder().
-		Select("COUNT(Id) AS Value").
-		From("Channels").
-		Where(sq.Eq{"Type": channelType})
+		Select("COUNT(*) AS Value").
+		From("Channels")
+
+	if channelType != "" {
+		query = query.Where(sq.Eq{"Type": channelType})
+	}
 
 	if teamId != "" {
 		query = query.Where(sq.Eq{"TeamId": teamId})

store/storetest/channel_store.go

@@ -3979,6 +3979,12 @@ func testChannelStoreGetMoreChannels(t *testing.T, ss store.Store) {
 		require.NoError(t, err)
 		require.EqualValues(t, 2, count)
 	})
+
+	t.Run("verify analytics for all channels", func(t *testing.T) {
+		count, err := ss.Channel().AnalyticsTypeCount(teamId, "")
+		require.NoError(t, err)
+		require.EqualValues(t, 6, count)
+	})
 }
 
 func testChannelStoreGetPrivateChannelsForTeam(t *testing.T, ss store.Store) {