From 68382c5fb71abc36a962a8bd4b86d3430fd4b268 Mon Sep 17 00:00:00 2001
From: Agniva De Sarker <agnivade@yahoo.co.in>
Date: Wed, 23 Mar 2022 13:46:10 +0530
Subject: [PATCH] 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                     | 12 ++++++++++++
 store/sqlstore/channel_store.go  | 10 ++++++----
 store/storetest/channel_store.go |  6 ++++++
 3 files changed, 24 insertions(+), 4 deletions(-)

diff --git a/jobs/jobs.go b/jobs/jobs.go
index 897fe87c25..f1a4979d8c 100644
--- a/jobs/jobs.go
+++ b/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()
diff --git a/store/sqlstore/channel_store.go b/store/sqlstore/channel_store.go
index 7faa237c6c..9b929479fa 100644
--- a/store/sqlstore/channel_store.go
+++ b/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})
diff --git a/store/storetest/channel_store.go b/store/storetest/channel_store.go
index 3e7343af75..e74c59ed00 100644
--- a/store/storetest/channel_store.go
+++ b/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) {