PLT-6595-Server: Job Management APIs. (#6931)

* PLT-6595-Server: Job Management APIs.

* MANAGE_JOBS Permission

* Fix test.

api4/apitestlib.go

@@ -24,6 +24,7 @@ import (
 	"github.com/mattermost/platform/wsapi"
 
 	s3 "github.com/minio/minio-go"
+	"github.com/mattermost/platform/jobs"
 )
 
 type TestHelper struct {
@@ -68,6 +69,10 @@ func SetupEnterprise() *TestHelper {
 		*utils.Cfg.TeamSettings.EnableOpenServer = true
 	}
 
+	if jobs.Srv.Store == nil {
+		jobs.Srv.Store = app.Srv.Store
+	}
+
 	th := &TestHelper{}
 	th.Client = th.CreateClient()
 	th.SystemAdminClient = th.CreateClient()
@@ -99,6 +104,10 @@ func Setup() *TestHelper {
 		*utils.Cfg.TeamSettings.EnableOpenServer = true
 	}
 
+	if jobs.Srv.Store == nil {
+		jobs.Srv.Store = app.Srv.Store
+	}
+
 	th := &TestHelper{}
 	th.Client = th.CreateClient()
 	th.SystemAdminClient = th.CreateClient()

api4/job.go

@@ -14,8 +14,11 @@ import (
 func InitJob() {
 	l4g.Info("Initializing job API routes")
 
-	BaseRoutes.Jobs.Handle("/type/{job_type:[A-Za-z0-9_-]+}/statuses", ApiSessionRequired(getJobsByType)).Methods("GET")
+	BaseRoutes.Jobs.Handle("", ApiSessionRequired(getJobs)).Methods("GET")
-	BaseRoutes.Jobs.Handle("/{job_id:[A-Za-z0-9]+}/status", ApiSessionRequired(getJob)).Methods("GET")
+	BaseRoutes.Jobs.Handle("", ApiSessionRequired(createJob)).Methods("POST")
+	BaseRoutes.Jobs.Handle("/{job_id:[A-Za-z0-9]+}", ApiSessionRequired(getJob)).Methods("GET")
+	BaseRoutes.Jobs.Handle("/{job_id:[A-Za-z0-9]+}/cancel", ApiSessionRequired(cancelJob)).Methods("POST")
+	BaseRoutes.Jobs.Handle("/type/{job_type:[A-Za-z0-9_-]+}", ApiSessionRequired(getJobsByType)).Methods("GET")
 }
 
 func getJob(c *Context, w http.ResponseWriter, r *http.Request) {
@@ -24,16 +27,55 @@ func getJob(c *Context, w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if !app.SessionHasPermissionTo(c.Session, model.PERMISSION_MANAGE_SYSTEM) {
+	if !app.SessionHasPermissionTo(c.Session, model.PERMISSION_MANAGE_JOBS) {
-		c.SetPermissionError(model.PERMISSION_MANAGE_SYSTEM)
+		c.SetPermissionError(model.PERMISSION_MANAGE_JOBS)
 		return
 	}
 
-	if status, err := app.GetJob(c.Params.JobId); err != nil {
+	if job, err := app.GetJob(c.Params.JobId); err != nil {
 		c.Err = err
 		return
 	} else {
-		w.Write([]byte(status.ToJson()))
+		w.Write([]byte(job.ToJson()))
+	}
+}
+
+func createJob(c *Context, w http.ResponseWriter, r *http.Request) {
+	job := model.JobFromJson(r.Body)
+	if job == nil {
+		c.SetInvalidParam("job")
+		return
+	}
+
+	if !app.SessionHasPermissionTo(c.Session, model.PERMISSION_MANAGE_JOBS) {
+		c.SetPermissionError(model.PERMISSION_MANAGE_JOBS)
+		return
+	}
+
+	if job, err := app.CreateJob(job); err != nil {
+		c.Err = err
+		return
+	} else {
+		w.WriteHeader(http.StatusCreated)
+		w.Write([]byte(job.ToJson()))
+	}
+}
+
+func getJobs(c *Context, w http.ResponseWriter, r *http.Request) {
+	if c.Err != nil {
+		return
+	}
+
+	if !app.SessionHasPermissionTo(c.Session, model.PERMISSION_MANAGE_JOBS) {
+		c.SetPermissionError(model.PERMISSION_MANAGE_JOBS)
+		return
+	}
+
+	if jobs, err := app.GetJobsPage(c.Params.Page, c.Params.PerPage); err != nil {
+		c.Err = err
+		return
+	} else {
+		w.Write([]byte(model.JobsToJson(jobs)))
 	}
 }
 
@@ -43,15 +85,34 @@ func getJobsByType(c *Context, w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	if !app.SessionHasPermissionTo(c.Session, model.PERMISSION_MANAGE_SYSTEM) {
+	if !app.SessionHasPermissionTo(c.Session, model.PERMISSION_MANAGE_JOBS) {
-		c.SetPermissionError(model.PERMISSION_MANAGE_SYSTEM)
+		c.SetPermissionError(model.PERMISSION_MANAGE_JOBS)
 		return
 	}
 
-	if statuses, err := app.GetJobsByTypePage(c.Params.JobType, c.Params.Page, c.Params.PerPage); err != nil {
+	if jobs, err := app.GetJobsByTypePage(c.Params.JobType, c.Params.Page, c.Params.PerPage); err != nil {
 		c.Err = err
 		return
 	} else {
-		w.Write([]byte(model.JobsToJson(statuses)))
+		w.Write([]byte(model.JobsToJson(jobs)))
 	}
 }
+
+func cancelJob(c *Context, w http.ResponseWriter, r *http.Request) {
+	c.RequireJobId()
+	if c.Err != nil {
+		return
+	}
+
+	if !app.SessionHasPermissionTo(c.Session, model.PERMISSION_MANAGE_JOBS) {
+		c.SetPermissionError(model.PERMISSION_MANAGE_JOBS)
+		return
+	}
+
+	if err := app.CancelJob(c.Params.JobId); err != nil {
+		c.Err = err
+		return
+	}
+
+	ReturnStatusOK(w)
+}

api4/job_test.go

@@ -12,74 +12,157 @@ import (
 	"github.com/mattermost/platform/store"
 )
 
-func TestGetJobStatus(t *testing.T) {
+func TestCreateJob(t *testing.T) {
 	th := Setup().InitBasic().InitSystemAdmin()
 	defer TearDown()
 
-	status := &model.Job{
+	job := &model.Job{
-		Id:     model.NewId(),
+		Type: model.JOB_TYPE_DATA_RETENTION,
-		Status: model.NewId(),
+		Data: map[string]interface{}{
+			"thing": "stuff",
+		},
 	}
-	if result := <-app.Srv.Store.Job().Save(status); result.Err != nil {
+
+	received, resp := th.SystemAdminClient.CreateJob(job)
+	CheckNoError(t, resp)
+
+	defer app.Srv.Store.Job().Delete(received.Id)
+
+	job = &model.Job{
+		Type: model.NewId(),
+	}
+
+	_, resp = th.SystemAdminClient.CreateJob(job)
+	CheckBadRequestStatus(t, resp)
+
+	_, resp = th.Client.CreateJob(job)
+	CheckForbiddenStatus(t, resp)
+}
+
+func TestGetJob(t *testing.T) {
+	th := Setup().InitBasic().InitSystemAdmin()
+	defer TearDown()
+
+	job := &model.Job{
+		Id:     model.NewId(),
+		Status: model.JOB_STATUS_PENDING,
+	}
+	if result := <-app.Srv.Store.Job().Save(job); result.Err != nil {
 		t.Fatal(result.Err)
 	}
 
-	defer app.Srv.Store.Job().Delete(status.Id)
+	defer app.Srv.Store.Job().Delete(job.Id)
 
-	received, resp := th.SystemAdminClient.GetJob(status.Id)
+	received, resp := th.SystemAdminClient.GetJob(job.Id)
 	CheckNoError(t, resp)
 
-	if received.Id != status.Id || received.Status != status.Status {
+	if received.Id != job.Id || received.Status != job.Status {
-		t.Fatal("incorrect job status received")
+		t.Fatal("incorrect job received")
 	}
 
 	_, resp = th.SystemAdminClient.GetJob("1234")
 	CheckBadRequestStatus(t, resp)
 
-	_, resp = th.Client.GetJob(status.Id)
+	_, resp = th.Client.GetJob(job.Id)
 	CheckForbiddenStatus(t, resp)
 
 	_, resp = th.SystemAdminClient.GetJob(model.NewId())
 	CheckNotFoundStatus(t, resp)
 }
 
-func TestGetJobStatusesByType(t *testing.T) {
+func TestGetJobs(t *testing.T) {
 	th := Setup().InitBasic().InitSystemAdmin()
 	defer TearDown()
 
 	jobType := model.NewId()
 
-	statuses := []*model.Job{
+	jobs := []*model.Job{
 		{
-			Id:      model.NewId(),
+			Id:       model.NewId(),
-			Type:    jobType,
+			Type:     jobType,
-			StartAt: 1000,
+			CreateAt: model.GetMillis() + 1,
 		},
 		{
-			Id:      model.NewId(),
+			Id:       model.NewId(),
-			Type:    jobType,
+			Type:     jobType,
-			StartAt: 999,
+			CreateAt: model.GetMillis(),
 		},
 		{
-			Id:      model.NewId(),
+			Id:       model.NewId(),
-			Type:    jobType,
+			Type:     jobType,
-			StartAt: 1001,
+			CreateAt: model.GetMillis() + 2,
 		},
 	}
 
-	for _, status := range statuses {
+	for _, job := range jobs {
-		store.Must(app.Srv.Store.Job().Save(status))
+		store.Must(app.Srv.Store.Job().Save(job))
-		defer app.Srv.Store.Job().Delete(status.Id)
+		defer app.Srv.Store.Job().Delete(job.Id)
+	}
+
+	received, resp := th.SystemAdminClient.GetJobs(0, 2)
+	CheckNoError(t, resp)
+
+	if len(received) != 2 {
+		t.Fatal("received wrong number of jobs")
+	} else if received[0].Id != jobs[2].Id {
+		t.Fatal("should've received newest job first")
+	} else if received[1].Id != jobs[0].Id {
+		t.Fatal("should've received second newest job second")
+	}
+
+	received, resp = th.SystemAdminClient.GetJobs(1, 2)
+	CheckNoError(t, resp)
+
+	if received[0].Id != jobs[1].Id {
+		t.Fatal("should've received oldest job last")
+	}
+
+	_, resp = th.Client.GetJobs(0, 60)
+	CheckForbiddenStatus(t, resp)
+}
+
+func TestGetJobsByType(t *testing.T) {
+	th := Setup().InitBasic().InitSystemAdmin()
+	defer TearDown()
+
+	jobType := model.NewId()
+
+	jobs := []*model.Job{
+		{
+			Id:       model.NewId(),
+			Type:     jobType,
+			CreateAt: 1000,
+		},
+		{
+			Id:       model.NewId(),
+			Type:     jobType,
+			CreateAt: 999,
+		},
+		{
+			Id:       model.NewId(),
+			Type:     jobType,
+			CreateAt: 1001,
+		},
+		{
+			Id:       model.NewId(),
+			Type:     model.NewId(),
+			CreateAt: 1002,
+		},
+	}
+
+	for _, job := range jobs {
+		store.Must(app.Srv.Store.Job().Save(job))
+		defer app.Srv.Store.Job().Delete(job.Id)
 	}
 
 	received, resp := th.SystemAdminClient.GetJobsByType(jobType, 0, 2)
 	CheckNoError(t, resp)
 
 	if len(received) != 2 {
-		t.Fatal("received wrong number of statuses")
+		t.Fatal("received wrong number of jobs")
-	} else if received[0].Id != statuses[1].Id {
+	} else if received[0].Id != jobs[2].Id {
 		t.Fatal("should've received newest job first")
-	} else if received[1].Id != statuses[0].Id {
+	} else if received[1].Id != jobs[0].Id {
 		t.Fatal("should've received second newest job second")
 	}
 
@@ -87,8 +170,8 @@ func TestGetJobStatusesByType(t *testing.T) {
 	CheckNoError(t, resp)
 
 	if len(received) != 1 {
-		t.Fatal("received wrong number of statuses")
+		t.Fatal("received wrong number of jobs")
-	} else if received[0].Id != statuses[2].Id {
+	} else if received[0].Id != jobs[1].Id {
 		t.Fatal("should've received oldest job last")
 	}
 
@@ -101,3 +184,46 @@ func TestGetJobStatusesByType(t *testing.T) {
 	_, resp = th.Client.GetJobsByType(jobType, 0, 60)
 	CheckForbiddenStatus(t, resp)
 }
+
+func TestCancelJob(t *testing.T) {
+	th := Setup().InitBasic().InitSystemAdmin()
+	defer TearDown()
+
+	jobs := []*model.Job{
+		{
+			Id:     model.NewId(),
+			Type:   model.NewId(),
+			Status: model.JOB_STATUS_PENDING,
+		},
+		{
+			Id:     model.NewId(),
+			Type:   model.NewId(),
+			Status: model.JOB_STATUS_IN_PROGRESS,
+		},
+		{
+			Id:     model.NewId(),
+			Type:   model.NewId(),
+			Status: model.JOB_STATUS_SUCCESS,
+		},
+	}
+
+	for _, job := range jobs {
+		store.Must(app.Srv.Store.Job().Save(job))
+		defer app.Srv.Store.Job().Delete(job.Id)
+	}
+
+	_, resp := th.Client.CancelJob(jobs[0].Id)
+	CheckForbiddenStatus(t, resp)
+
+	_, resp = th.SystemAdminClient.CancelJob(jobs[0].Id)
+	CheckNoError(t, resp)
+
+	_, resp = th.SystemAdminClient.CancelJob(jobs[1].Id)
+	CheckNoError(t, resp)
+
+	_, resp = th.SystemAdminClient.CancelJob(jobs[2].Id)
+	CheckInternalErrorStatus(t, resp)
+
+	_, resp = th.SystemAdminClient.CancelJob(model.NewId())
+	CheckInternalErrorStatus(t, resp)
+}

app/job.go

@@ -5,6 +5,7 @@ package app
 
 import (
 	"github.com/mattermost/platform/model"
+	"github.com/mattermost/platform/jobs"
 )
 
 func GetJob(id string) (*model.Job, *model.AppError) {
@@ -15,6 +16,18 @@ func GetJob(id string) (*model.Job, *model.AppError) {
 	}
 }
 
+func GetJobsPage(page int, perPage int) ([]*model.Job, *model.AppError) {
+	return GetJobs(page*perPage, perPage)
+}
+
+func GetJobs(offset int, limit int) ([]*model.Job, *model.AppError) {
+	if result := <-Srv.Store.Job().GetAllPage(offset, limit); result.Err != nil {
+		return nil, result.Err
+	} else {
+		return result.Data.([]*model.Job), nil
+	}
+}
+
 func GetJobsByTypePage(jobType string, page int, perPage int) ([]*model.Job, *model.AppError) {
 	return GetJobsByType(jobType, page*perPage, perPage)
 }
@@ -26,3 +39,11 @@ func GetJobsByType(jobType string, offset int, limit int) ([]*model.Job, *model.
 		return result.Data.([]*model.Job), nil
 	}
 }
+
+func CreateJob(job *model.Job) (*model.Job, *model.AppError) {
+	return jobs.CreateJob(job.Type, job.Data)
+}
+
+func CancelJob(jobId string) *model.AppError {
+	return jobs.RequestCancellation(jobId)
+}

app/job_test.go

@@ -10,7 +10,7 @@ import (
 	"github.com/mattermost/platform/store"
 )
 
-func TestGetJobStatus(t *testing.T) {
+func TestGetJob(t *testing.T) {
 	Setup()
 
 	status := &model.Job{
@@ -30,7 +30,7 @@ func TestGetJobStatus(t *testing.T) {
 	}
 }
 
-func TestGetJobStatusesByType(t *testing.T) {
+func TestGetJobByType(t *testing.T) {
 	Setup()
 
 	jobType := model.NewId()
@@ -39,17 +39,17 @@ func TestGetJobStatusesByType(t *testing.T) {
 		{
 			Id:      model.NewId(),
 			Type:    jobType,
-			StartAt: 1000,
+			CreateAt: 1000,
 		},
 		{
 			Id:      model.NewId(),
 			Type:    jobType,
-			StartAt: 999,
+			CreateAt: 999,
 		},
 		{
 			Id:      model.NewId(),
 			Type:    jobType,
-			StartAt: 1001,
+			CreateAt: 1001,
 		},
 	}
 
@@ -62,7 +62,7 @@ func TestGetJobStatusesByType(t *testing.T) {
 		t.Fatal(err)
 	} else if len(received) != 2 {
 		t.Fatal("received wrong number of statuses")
-	} else if received[0].Id != statuses[1].Id {
+	} else if received[0].Id != statuses[2].Id {
 		t.Fatal("should've received newest job first")
 	} else if received[1].Id != statuses[0].Id {
 		t.Fatal("should've received second newest job second")
@@ -72,7 +72,7 @@ func TestGetJobStatusesByType(t *testing.T) {
 		t.Fatal(err)
 	} else if len(received) != 1 {
 		t.Fatal("received wrong number of statuses")
-	} else if received[0].Id != statuses[2].Id {
+	} else if received[0].Id != statuses[1].Id {
 		t.Fatal("should've received oldest job last")
 	}
 }

i18n/en.json

@@ -3371,6 +3371,14 @@
     "id": "authentication.permissions.create_team_roles.name",
     "translation": "Create Teams"
   },
+  {
+    "id": "authentication.permissions.manage_jobs.description",
+    "translation": "Ability to manage jobs"
+  },
+  {
+    "id": "authentication.permissions.manage_jobs.name",
+    "translation": "Manage Jobs"
+  },
   {
     "id": "authentication.permissions.manage_team_roles.description",
     "translation": "Ability to change the roles of a team member"
@@ -4471,6 +4479,22 @@
     "id": "model.incoming_hook.user_id.app_error",
     "translation": "Invalid user id"
   },
+  {
+    "id": "model.job.is_valid.id.app_error",
+    "translation": "Invalid job Id"
+  },
+  {
+    "id": "model.job.is_valid.create_at.app_error",
+    "translation": "Create at must be a valid time"
+  },
+  {
+    "id": "model.job.is_valid.type.app_error",
+    "translation": "Invalid job type"
+  },
+  {
+    "id": "model.job.is_valid.status.app_error",
+    "translation": "Invalid job status"
+  },
   {
     "id": "model.oauth.is_valid.app_id.app_error",
     "translation": "Invalid app id"

jobs/jobs.go

@@ -25,6 +25,10 @@ func CreateJob(jobType string, jobData map[string]interface{}) (*model.Job, *mod
 		Data:     jobData,
 	}
 
+	if err := job.IsValid(); err != nil {
+		return nil, err
+	}
+
 	if result := <-Srv.Store.Job().Save(&job); result.Err != nil {
 		return nil, result.Err
 	}
@@ -41,7 +45,7 @@ func ClaimJob(job *model.Job) (bool, *model.AppError) {
 	}
 }
 
-func SetJobProgress(job *model.Job, progress int64) (*model.AppError) {
+func SetJobProgress(job *model.Job, progress int64) *model.AppError {
 	job.Status = model.JOB_STATUS_IN_PROGRESS
 	job.Progress = progress
 
@@ -78,7 +82,7 @@ func SetJobError(job *model.Job, jobError *model.AppError) *model.AppError {
 				return result.Err
 			} else {
 				if !result.Data.(bool) {
-					return model.NewAppError("Jobs.SetJobError", "jobs.set_job_error.update.error", nil, "id=" + job.Id, http.StatusInternalServerError)
+					return model.NewAppError("Jobs.SetJobError", "jobs.set_job_error.update.error", nil, "id="+job.Id, http.StatusInternalServerError)
 				}
 			}
 		}
@@ -92,20 +96,20 @@ func SetJobCanceled(job *model.Job) *model.AppError {
 	return result.Err
 }
 
-func RequestCancellation(job *model.Job) *model.AppError {
+func RequestCancellation(jobId string) *model.AppError {
-	if result := <-Srv.Store.Job().UpdateStatusOptimistically(job.Id, model.JOB_STATUS_PENDING, model.JOB_STATUS_CANCELED); result.Err != nil {
+	if result := <-Srv.Store.Job().UpdateStatusOptimistically(jobId, model.JOB_STATUS_PENDING, model.JOB_STATUS_CANCELED); result.Err != nil {
 		return result.Err
 	} else if result.Data.(bool) {
 		return nil
 	}
 
-	if result := <-Srv.Store.Job().UpdateStatusOptimistically(job.Id, model.JOB_STATUS_IN_PROGRESS, model.JOB_STATUS_CANCEL_REQUESTED); result.Err != nil {
+	if result := <-Srv.Store.Job().UpdateStatusOptimistically(jobId, model.JOB_STATUS_IN_PROGRESS, model.JOB_STATUS_CANCEL_REQUESTED); result.Err != nil {
 		return result.Err
 	} else if result.Data.(bool) {
 		return nil
 	}
 
-	return model.NewAppError("Jobs.RequestCancellation", "jobs.request_cancellation.status.error", nil, "id=" + job.Id, http.StatusInternalServerError)
+	return model.NewAppError("Jobs.RequestCancellation", "jobs.request_cancellation.status.error", nil, "id="+jobId, http.StatusInternalServerError)
 }
 
 func CancellationWatcher(ctx context.Context, jobId string, cancelChan chan interface{}) {

model/authorization.go

@@ -58,6 +58,7 @@ var PERMISSION_MANAGE_TEAM *Permission
 var PERMISSION_IMPORT_TEAM *Permission
 var PERMISSION_VIEW_TEAM *Permission
 var PERMISSION_LIST_USERS_WITHOUT_TEAM *Permission
+var PERMISSION_MANAGE_JOBS *Permission
 
 // General permission that encompases all system admin functions
 // in the future this could be broken up to allow access to some
@@ -292,6 +293,11 @@ func InitalizePermissions() {
 		"authentication.permisssions.list_users_without_team.name",
 		"authentication.permisssions.list_users_without_team.description",
 	}
+	PERMISSION_MANAGE_JOBS = &Permission{
+		"manage_jobs",
+		"authentication.permisssions.manage_jobs.name",
+		"authentication.permisssions.manage_jobs.description",
+	}
 }
 
 func InitalizeRoles() {
@@ -405,6 +411,7 @@ func InitalizeRoles() {
 							PERMISSION_CREATE_TEAM.Id,
 							PERMISSION_ADD_USER_TO_TEAM.Id,
 							PERMISSION_LIST_USERS_WITHOUT_TEAM.Id,
+							PERMISSION_MANAGE_JOBS.Id,
 						},
 						ROLE_TEAM_USER.Permissions...,
 					),

model/client4.go

@@ -2800,7 +2800,7 @@ func (c *Client4) OpenGraph(url string) (map[string]string, *Response) {
 
 // GetJob gets a single job.
 func (c *Client4) GetJob(id string) (*Job, *Response) {
-	if r, err := c.DoApiGet(c.GetJobsRoute()+fmt.Sprintf("/%v/status", id), ""); err != nil {
+	if r, err := c.DoApiGet(c.GetJobsRoute()+fmt.Sprintf("/%v", id), ""); err != nil {
 		return nil, BuildErrorResponse(r, err)
 	} else {
 		defer closeBody(r)
@@ -2808,12 +2808,42 @@ func (c *Client4) GetJob(id string) (*Job, *Response) {
 	}
 }
 
-// GetJobsByType gets all jobs of a given type, sorted with the job that most recently started first.
+// Get all jobs, sorted with the job that was created most recently first.
-func (c *Client4) GetJobsByType(jobType string, page int, perPage int) ([]*Job, *Response) {
+func (c *Client4) GetJobs(page int, perPage int) ([]*Job, *Response) {
-	if r, err := c.DoApiGet(c.GetJobsRoute()+fmt.Sprintf("/type/%v/statuses?page=%v&per_page=%v", jobType, page, perPage), ""); err != nil {
+	if r, err := c.DoApiGet(c.GetJobsRoute()+fmt.Sprintf("?page=%v&per_page=%v", page, perPage), ""); err != nil {
 		return nil, BuildErrorResponse(r, err)
 	} else {
 		defer closeBody(r)
 		return JobsFromJson(r.Body), BuildResponse(r)
 	}
 }
+
+// GetJobsByType gets all jobs of a given type, sorted with the job that was created most recently first.
+func (c *Client4) GetJobsByType(jobType string, page int, perPage int) ([]*Job, *Response) {
+	if r, err := c.DoApiGet(c.GetJobsRoute()+fmt.Sprintf("/type/%v?page=%v&per_page=%v", jobType, page, perPage), ""); err != nil {
+		return nil, BuildErrorResponse(r, err)
+	} else {
+		defer closeBody(r)
+		return JobsFromJson(r.Body), BuildResponse(r)
+	}
+}
+
+// CreateJob creates a job based on the provided job struct.
+func (c *Client4) CreateJob(job *Job) (*Job, *Response) {
+	if r, err := c.DoApiPost(c.GetJobsRoute(), job.ToJson()); err != nil {
+		return nil, BuildErrorResponse(r, err)
+	} else {
+		defer closeBody(r)
+		return JobFromJson(r.Body), BuildResponse(r)
+	}
+}
+
+// CancelJob requests the cancellation of the job with the provided Id.
+func (c *Client4) CancelJob(jobId string) (bool, *Response) {
+	if r, err := c.DoApiPost(c.GetJobsRoute()+fmt.Sprintf("/%v/cancel", jobId), ""); err != nil {
+		return false, BuildErrorResponse(r, err)
+	} else {
+		defer closeBody(r)
+		return CheckStatusOK(r), BuildResponse(r)
+	}
+}

model/job.go

@@ -6,6 +6,7 @@ package model
 import (
 	"encoding/json"
 	"io"
+	"net/http"
 )
 
 const (
@@ -32,6 +33,36 @@ type Job struct {
 	Data           map[string]interface{} `json:"data"`
 }
 
+func (j *Job) IsValid() *AppError {
+	if len(j.Id) != 26 {
+		return NewAppError("Job.IsValid", "model.job.is_valid.id.app_error", nil, "id="+j.Id, http.StatusBadRequest)
+	}
+
+	if j.CreateAt == 0 {
+		return NewAppError("Job.IsValid", "model.job.is_valid.create_at.app_error", nil, "id="+j.Id, http.StatusBadRequest)
+	}
+
+	switch j.Type {
+	case JOB_TYPE_DATA_RETENTION:
+	case JOB_TYPE_SEARCH_INDEXING:
+	default:
+		return NewAppError("Job.IsValid", "model.job.is_valid.type.app_error", nil, "id="+j.Id, http.StatusBadRequest)
+	}
+
+	switch j.Status {
+	case JOB_STATUS_PENDING:
+	case JOB_STATUS_IN_PROGRESS:
+	case JOB_STATUS_SUCCESS:
+	case JOB_STATUS_ERROR:
+	case JOB_STATUS_CANCEL_REQUESTED:
+	case JOB_STATUS_CANCELED:
+	default:
+		return NewAppError("Job.IsValid", "model.job.is_valid.status.app_error", nil, "id="+j.Id, http.StatusBadRequest)
+	}
+
+	return nil
+}
+
 func (js *Job) ToJson() string {
 	if b, err := json.Marshal(js); err != nil {
 		return ""
@@ -41,9 +72,9 @@ func (js *Job) ToJson() string {
 }
 
 func JobFromJson(data io.Reader) *Job {
-	var status Job
+	var job Job
-	if err := json.NewDecoder(data).Decode(&status); err == nil {
+	if err := json.NewDecoder(data).Decode(&job); err == nil {
-		return &status
+		return &job
 	} else {
 		return nil
 	}

store/sql_job_store.go

@@ -210,6 +210,38 @@ func (jss SqlJobStore) Get(id string) StoreChannel {
 	return storeChannel
 }
 
+func (jss SqlJobStore) GetAllPage(offset int, limit int) StoreChannel {
+	storeChannel := make(StoreChannel, 1)
+
+	go func() {
+		result := StoreResult{}
+
+		var statuses []*model.Job
+
+		if _, err := jss.GetReplica().Select(&statuses,
+			`SELECT
+				*
+			FROM
+				Jobs
+			ORDER BY
+				CreateAt DESC
+			LIMIT
+				:Limit
+			OFFSET
+				:Offset`, map[string]interface{}{"Limit": limit, "Offset": offset}); err != nil {
+			result.Err = model.NewLocAppError("SqlJobStore.GetAllPage",
+				"store.sql_job.get_all.app_error", nil, err.Error())
+		} else {
+			result.Data = statuses
+		}
+
+		storeChannel <- result
+		close(storeChannel)
+	}()
+
+	return storeChannel
+}
+
 func (jss SqlJobStore) GetAllByType(jobType string) StoreChannel {
 	storeChannel := make(StoreChannel, 1)
 
@@ -224,7 +256,9 @@ func (jss SqlJobStore) GetAllByType(jobType string) StoreChannel {
 			FROM
 				Jobs
 			WHERE
-				Type = :Type`, map[string]interface{}{"Type": jobType}); err != nil {
+				Type = :Type
+			ORDER BY
+				CreateAt DESC`, map[string]interface{}{"Type": jobType}); err != nil {
 			result.Err = model.NewLocAppError("SqlJobStore.GetAllByType",
 				"store.sql_job.get_all.app_error", nil, "Type="+jobType+", "+err.Error())
 		} else {
@@ -254,7 +288,7 @@ func (jss SqlJobStore) GetAllByTypePage(jobType string, offset int, limit int) S
 			WHERE
 				Type = :Type
 			ORDER BY
-				StartAt ASC
+				CreateAt DESC
 			LIMIT
 				:Limit
 			OFFSET

store/sql_job_store_test.go

@@ -82,19 +82,24 @@ func TestJobGetAllByTypePage(t *testing.T) {
 
 	jobs := []*model.Job{
 		{
-			Id:      model.NewId(),
+			Id:       model.NewId(),
-			Type:    jobType,
+			Type:     jobType,
-			StartAt: 1000,
+			CreateAt: 1000,
 		},
 		{
-			Id:      model.NewId(),
+			Id:       model.NewId(),
-			Type:    jobType,
+			Type:     jobType,
-			StartAt: 999,
+			CreateAt: 999,
 		},
 		{
-			Id:      model.NewId(),
+			Id:       model.NewId(),
-			Type:    jobType,
+			Type:     jobType,
-			StartAt: 1001,
+			CreateAt: 1001,
+		},
+		{
+			Id:       model.NewId(),
+			Type:     model.NewId(),
+			CreateAt: 1002,
 		},
 	}
 
@@ -107,7 +112,7 @@ func TestJobGetAllByTypePage(t *testing.T) {
 		t.Fatal(result.Err)
 	} else if received := result.Data.([]*model.Job); len(received) != 2 {
 		t.Fatal("received wrong number of jobs")
-	} else if received[0].Id != jobs[1].Id {
+	} else if received[0].Id != jobs[2].Id {
 		t.Fatal("should've received newest job first")
 	} else if received[1].Id != jobs[0].Id {
 		t.Fatal("should've received second newest job second")
@@ -117,7 +122,54 @@ func TestJobGetAllByTypePage(t *testing.T) {
 		t.Fatal(result.Err)
 	} else if received := result.Data.([]*model.Job); len(received) != 1 {
 		t.Fatal("received wrong number of jobs")
+	} else if received[0].Id != jobs[1].Id {
+		t.Fatal("should've received oldest job last")
+	}
+}
+
+func TestJobGetAllPage(t *testing.T) {
+	Setup()
+
+	jobType := model.NewId()
+
+	jobs := []*model.Job{
+		{
+			Id:       model.NewId(),
+			Type:     jobType,
+			CreateAt: model.GetMillis() + 1,
+		},
+		{
+			Id:       model.NewId(),
+			Type:     jobType,
+			CreateAt: model.GetMillis(),
+		},
+		{
+			Id:       model.NewId(),
+			Type:     jobType,
+			CreateAt: model.GetMillis() + 2,
+		},
+	}
+
+	for _, job := range jobs {
+		Must(store.Job().Save(job))
+		defer store.Job().Delete(job.Id)
+	}
+
+	if result := <-store.Job().GetAllPage(0, 2); result.Err != nil {
+		t.Fatal(result.Err)
+	} else if received := result.Data.([]*model.Job); len(received) != 2 {
+		t.Fatal("received wrong number of jobs")
 	} else if received[0].Id != jobs[2].Id {
+		t.Fatal("should've received newest job first")
+	} else if received[1].Id != jobs[0].Id {
+		t.Fatal("should've received second newest job second")
+	}
+
+	if result := <-store.Job().GetAllPage(2, 2); result.Err != nil {
+		t.Fatal(result.Err)
+	} else if received := result.Data.([]*model.Job); len(received) < 1 {
+		t.Fatal("received wrong number of jobs")
+	} else if received[0].Id != jobs[1].Id {
 		t.Fatal("should've received oldest job last")
 	}
 }
@@ -331,11 +383,11 @@ func TestJobUpdateStatusUpdateStatusOptimistically(t *testing.T) {
 func TestJobDelete(t *testing.T) {
 	Setup()
 
-	status := Must(store.Job().Save(&model.Job{
+	job := Must(store.Job().Save(&model.Job{
 		Id: model.NewId(),
 	})).(*model.Job)
 
-	if result := <-store.Job().Delete(status.Id); result.Err != nil {
+	if result := <-store.Job().Delete(job.Id); result.Err != nil {
 		t.Fatal(result.Err)
 	}
 }

store/store.go

@@ -391,6 +391,7 @@ type JobStore interface {
 	UpdateStatus(id string, status string) StoreChannel
 	UpdateStatusOptimistically(id string, currentStatus string, newStatus string) StoreChannel
 	Get(id string) StoreChannel
+	GetAllPage(offset int, limit int) StoreChannel
 	GetAllByType(jobType string) StoreChannel
 	GetAllByTypePage(jobType string, offset int, limit int) StoreChannel
 	GetAllByStatus(status string) StoreChannel