Migrate Jobs store to sync by default (#11183)
* Migrate Jobs store to sync by default * Fixing compilation * Fixing compilation * Fixing govet
Этот коммит содержится в:
Jesús Espino
коммит произвёл
GitHub
GitHub
родитель
bbdd6927de
Коммит
f934502a56
@@ -34,267 +34,161 @@ func (jss SqlJobStore) CreateIndexesIfNotExists() {
|
||||
jss.CreateIndexIfNotExists("idx_jobs_type", "Jobs", "Type")
|
||||
}
|
||||
|
||||
func (jss SqlJobStore) Save(job *model.Job) store.StoreChannel {
|
||||
return store.Do(func(result *store.StoreResult) {
|
||||
if err := jss.GetMaster().Insert(job); err != nil {
|
||||
result.Err = model.NewAppError("SqlJobStore.Save", "store.sql_job.save.app_error", nil, "id="+job.Id+", "+err.Error(), http.StatusInternalServerError)
|
||||
} else {
|
||||
result.Data = job
|
||||
}
|
||||
})
|
||||
func (jss SqlJobStore) Save(job *model.Job) (*model.Job, *model.AppError) {
|
||||
if err := jss.GetMaster().Insert(job); err != nil {
|
||||
return nil, model.NewAppError("SqlJobStore.Save", "store.sql_job.save.app_error", nil, "id="+job.Id+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return job, nil
|
||||
}
|
||||
|
||||
func (jss SqlJobStore) UpdateOptimistically(job *model.Job, currentStatus string) store.StoreChannel {
|
||||
return store.Do(func(result *store.StoreResult) {
|
||||
if sqlResult, err := jss.GetMaster().Exec(
|
||||
`UPDATE
|
||||
Jobs
|
||||
SET
|
||||
LastActivityAt = :LastActivityAt,
|
||||
Status = :Status,
|
||||
Progress = :Progress,
|
||||
Data = :Data
|
||||
WHERE
|
||||
Id = :Id
|
||||
AND
|
||||
Status = :OldStatus`,
|
||||
map[string]interface{}{
|
||||
"Id": job.Id,
|
||||
"OldStatus": currentStatus,
|
||||
"LastActivityAt": model.GetMillis(),
|
||||
"Status": job.Status,
|
||||
"Data": job.DataToJson(),
|
||||
"Progress": job.Progress,
|
||||
}); err != nil {
|
||||
result.Err = model.NewAppError("SqlJobStore.UpdateOptimistically", "store.sql_job.update.app_error", nil, "id="+job.Id+", "+err.Error(), http.StatusInternalServerError)
|
||||
} else {
|
||||
rows, err := sqlResult.RowsAffected()
|
||||
func (jss SqlJobStore) UpdateOptimistically(job *model.Job, currentStatus string) (bool, *model.AppError) {
|
||||
query := "UPDATE Jobs SET LastActivityAt = :LastActivityAt, Status = :Status, Progress = :Progress, Data = :Data WHERE Id = :Id AND Status = :OldStatus"
|
||||
params := map[string]interface{}{
|
||||
"Id": job.Id,
|
||||
"OldStatus": currentStatus,
|
||||
"LastActivityAt": model.GetMillis(),
|
||||
"Status": job.Status,
|
||||
"Data": job.DataToJson(),
|
||||
"Progress": job.Progress,
|
||||
}
|
||||
sqlResult, err := jss.GetMaster().Exec(query, params)
|
||||
if err != nil {
|
||||
return false, model.NewAppError("SqlJobStore.UpdateOptimistically", "store.sql_job.update.app_error", nil, "id="+job.Id+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
result.Err = model.NewAppError("SqlJobStore.UpdateStatus", "store.sql_job.update.app_error", nil, "id="+job.Id+", "+err.Error(), http.StatusInternalServerError)
|
||||
} else {
|
||||
if rows == 1 {
|
||||
result.Data = true
|
||||
} else {
|
||||
result.Data = false
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
rows, err := sqlResult.RowsAffected()
|
||||
|
||||
if err != nil {
|
||||
return false, model.NewAppError("SqlJobStore.UpdateStatus", "store.sql_job.update.app_error", nil, "id="+job.Id+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
if rows != 1 {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (jss SqlJobStore) UpdateStatus(id string, status string) store.StoreChannel {
|
||||
return store.Do(func(result *store.StoreResult) {
|
||||
job := &model.Job{
|
||||
Id: id,
|
||||
Status: status,
|
||||
LastActivityAt: model.GetMillis(),
|
||||
}
|
||||
func (jss SqlJobStore) UpdateStatus(id string, status string) (*model.Job, *model.AppError) {
|
||||
job := &model.Job{
|
||||
Id: id,
|
||||
Status: status,
|
||||
LastActivityAt: model.GetMillis(),
|
||||
}
|
||||
|
||||
if _, err := jss.GetMaster().UpdateColumns(func(col *gorp.ColumnMap) bool {
|
||||
return col.ColumnName == "Status" || col.ColumnName == "LastActivityAt"
|
||||
}, job); err != nil {
|
||||
result.Err = model.NewAppError("SqlJobStore.UpdateStatus", "store.sql_job.update.app_error", nil, "id="+id+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
if _, err := jss.GetMaster().UpdateColumns(func(col *gorp.ColumnMap) bool {
|
||||
return col.ColumnName == "Status" || col.ColumnName == "LastActivityAt"
|
||||
}, job); err != nil {
|
||||
return nil, model.NewAppError("SqlJobStore.UpdateStatus", "store.sql_job.update.app_error", nil, "id="+id+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
|
||||
if result.Err == nil {
|
||||
result.Data = job
|
||||
}
|
||||
})
|
||||
return job, nil
|
||||
}
|
||||
|
||||
func (jss SqlJobStore) UpdateStatusOptimistically(id string, currentStatus string, newStatus string) store.StoreChannel {
|
||||
return store.Do(func(result *store.StoreResult) {
|
||||
var startAtClause string
|
||||
if newStatus == model.JOB_STATUS_IN_PROGRESS {
|
||||
startAtClause = `StartAt = :StartAt,`
|
||||
}
|
||||
func (jss SqlJobStore) UpdateStatusOptimistically(id string, currentStatus string, newStatus string) (bool, *model.AppError) {
|
||||
var startAtClause string
|
||||
if newStatus == model.JOB_STATUS_IN_PROGRESS {
|
||||
startAtClause = "StartAt = :StartAt,"
|
||||
}
|
||||
query := "UPDATE Jobs SET " + startAtClause + " Status = :NewStatus, LastActivityAt = :LastActivityAt WHERE Id = :Id AND Status = :OldStatus"
|
||||
params := map[string]interface{}{
|
||||
"Id": id,
|
||||
"OldStatus": currentStatus,
|
||||
"NewStatus": newStatus,
|
||||
"StartAt": model.GetMillis(),
|
||||
"LastActivityAt": model.GetMillis(),
|
||||
}
|
||||
|
||||
if sqlResult, err := jss.GetMaster().Exec(
|
||||
`UPDATE
|
||||
Jobs
|
||||
SET `+startAtClause+`
|
||||
Status = :NewStatus,
|
||||
LastActivityAt = :LastActivityAt
|
||||
WHERE
|
||||
Id = :Id
|
||||
AND
|
||||
Status = :OldStatus`, map[string]interface{}{"Id": id, "OldStatus": currentStatus, "NewStatus": newStatus, "StartAt": model.GetMillis(), "LastActivityAt": model.GetMillis()}); err != nil {
|
||||
result.Err = model.NewAppError("SqlJobStore.UpdateStatus", "store.sql_job.update.app_error", nil, "id="+id+", "+err.Error(), http.StatusInternalServerError)
|
||||
} else {
|
||||
rows, err := sqlResult.RowsAffected()
|
||||
sqlResult, err := jss.GetMaster().Exec(query, params)
|
||||
if err != nil {
|
||||
return false, model.NewAppError("SqlJobStore.UpdateStatus", "store.sql_job.update.app_error", nil, "id="+id+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
rows, err := sqlResult.RowsAffected()
|
||||
if err != nil {
|
||||
return false, model.NewAppError("SqlJobStore.UpdateStatus", "store.sql_job.update.app_error", nil, "id="+id+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
if rows != 1 {
|
||||
return false, nil
|
||||
}
|
||||
|
||||
if err != nil {
|
||||
result.Err = model.NewAppError("SqlJobStore.UpdateStatus", "store.sql_job.update.app_error", nil, "id="+id+", "+err.Error(), http.StatusInternalServerError)
|
||||
} else {
|
||||
if rows == 1 {
|
||||
result.Data = true
|
||||
} else {
|
||||
result.Data = false
|
||||
}
|
||||
}
|
||||
}
|
||||
})
|
||||
return true, nil
|
||||
}
|
||||
|
||||
func (jss SqlJobStore) Get(id string) store.StoreChannel {
|
||||
return store.Do(func(result *store.StoreResult) {
|
||||
var status *model.Job
|
||||
func (jss SqlJobStore) Get(id string) (*model.Job, *model.AppError) {
|
||||
query := "SELECT * FROM Jobs WHERE Id = :Id"
|
||||
|
||||
if err := jss.GetReplica().SelectOne(&status,
|
||||
`SELECT
|
||||
*
|
||||
FROM
|
||||
Jobs
|
||||
WHERE
|
||||
Id = :Id`, map[string]interface{}{"Id": id}); err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
result.Err = model.NewAppError("SqlJobStore.Get", "store.sql_job.get.app_error", nil, "Id="+id+", "+err.Error(), http.StatusNotFound)
|
||||
} else {
|
||||
result.Err = model.NewAppError("SqlJobStore.Get", "store.sql_job.get.app_error", nil, "Id="+id+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
} else {
|
||||
result.Data = status
|
||||
var status *model.Job
|
||||
if err := jss.GetReplica().SelectOne(&status, query, map[string]interface{}{"Id": id}); err != nil {
|
||||
if err == sql.ErrNoRows {
|
||||
return nil, model.NewAppError("SqlJobStore.Get", "store.sql_job.get.app_error", nil, "Id="+id+", "+err.Error(), http.StatusNotFound)
|
||||
}
|
||||
})
|
||||
return nil, model.NewAppError("SqlJobStore.Get", "store.sql_job.get.app_error", nil, "Id="+id+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return status, nil
|
||||
}
|
||||
|
||||
func (jss SqlJobStore) GetAllPage(offset int, limit int) store.StoreChannel {
|
||||
return store.Do(func(result *store.StoreResult) {
|
||||
var statuses []*model.Job
|
||||
func (jss SqlJobStore) GetAllPage(offset int, limit int) ([]*model.Job, *model.AppError) {
|
||||
query := "SELECT * FROM Jobs ORDER BY CreateAt DESC LIMIT :Limit OFFSET :Offset"
|
||||
|
||||
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.NewAppError("SqlJobStore.GetAllPage", "store.sql_job.get_all.app_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
} else {
|
||||
result.Data = statuses
|
||||
}
|
||||
})
|
||||
var statuses []*model.Job
|
||||
if _, err := jss.GetReplica().Select(&statuses, query, map[string]interface{}{"Limit": limit, "Offset": offset}); err != nil {
|
||||
return nil, model.NewAppError("SqlJobStore.GetAllPage", "store.sql_job.get_all.app_error", nil, err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return statuses, nil
|
||||
}
|
||||
|
||||
func (jss SqlJobStore) GetAllByType(jobType string) store.StoreChannel {
|
||||
return store.Do(func(result *store.StoreResult) {
|
||||
var statuses []*model.Job
|
||||
|
||||
if _, err := jss.GetReplica().Select(&statuses,
|
||||
`SELECT
|
||||
*
|
||||
FROM
|
||||
Jobs
|
||||
WHERE
|
||||
Type = :Type
|
||||
ORDER BY
|
||||
CreateAt DESC`, map[string]interface{}{"Type": jobType}); err != nil {
|
||||
result.Err = model.NewAppError("SqlJobStore.GetAllByType", "store.sql_job.get_all.app_error", nil, "Type="+jobType+", "+err.Error(), http.StatusInternalServerError)
|
||||
} else {
|
||||
result.Data = statuses
|
||||
}
|
||||
})
|
||||
func (jss SqlJobStore) GetAllByType(jobType string) ([]*model.Job, *model.AppError) {
|
||||
query := "SELECT * FROM Jobs WHERE Type = :Type ORDER BY CreateAt DESC"
|
||||
var statuses []*model.Job
|
||||
if _, err := jss.GetReplica().Select(&statuses, query, map[string]interface{}{"Type": jobType}); err != nil {
|
||||
return nil, model.NewAppError("SqlJobStore.GetAllByType", "store.sql_job.get_all.app_error", nil, "Type="+jobType+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return statuses, nil
|
||||
}
|
||||
|
||||
func (jss SqlJobStore) GetAllByTypePage(jobType string, offset int, limit int) store.StoreChannel {
|
||||
return store.Do(func(result *store.StoreResult) {
|
||||
var statuses []*model.Job
|
||||
func (jss SqlJobStore) GetAllByTypePage(jobType string, offset int, limit int) ([]*model.Job, *model.AppError) {
|
||||
query := "SELECT * FROM Jobs WHERE Type = :Type ORDER BY CreateAt DESC LIMIT :Limit OFFSET :Offset"
|
||||
|
||||
if _, err := jss.GetReplica().Select(&statuses,
|
||||
`SELECT
|
||||
*
|
||||
FROM
|
||||
Jobs
|
||||
WHERE
|
||||
Type = :Type
|
||||
ORDER BY
|
||||
CreateAt DESC
|
||||
LIMIT
|
||||
:Limit
|
||||
OFFSET
|
||||
:Offset`, map[string]interface{}{"Type": jobType, "Limit": limit, "Offset": offset}); err != nil {
|
||||
result.Err = model.NewAppError("SqlJobStore.GetAllByTypePage", "store.sql_job.get_all.app_error", nil, "Type="+jobType+", "+err.Error(), http.StatusInternalServerError)
|
||||
} else {
|
||||
result.Data = statuses
|
||||
}
|
||||
})
|
||||
var statuses []*model.Job
|
||||
if _, err := jss.GetReplica().Select(&statuses, query, map[string]interface{}{"Type": jobType, "Limit": limit, "Offset": offset}); err != nil {
|
||||
return nil, model.NewAppError("SqlJobStore.GetAllByTypePage", "store.sql_job.get_all.app_error", nil, "Type="+jobType+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return statuses, nil
|
||||
}
|
||||
|
||||
func (jss SqlJobStore) GetAllByStatus(status string) store.StoreChannel {
|
||||
return store.Do(func(result *store.StoreResult) {
|
||||
var statuses []*model.Job
|
||||
func (jss SqlJobStore) GetAllByStatus(status string) ([]*model.Job, *model.AppError) {
|
||||
var statuses []*model.Job
|
||||
query := "SELECT * FROM Jobs WHERE Status = :Status ORDER BY CreateAt ASC"
|
||||
|
||||
if _, err := jss.GetReplica().Select(&statuses,
|
||||
`SELECT
|
||||
*
|
||||
FROM
|
||||
Jobs
|
||||
WHERE
|
||||
Status = :Status
|
||||
ORDER BY
|
||||
CreateAt ASC`, map[string]interface{}{"Status": status}); err != nil {
|
||||
result.Err = model.NewAppError("SqlJobStore.GetAllByStatus", "store.sql_job.get_all.app_error", nil, "Status="+status+", "+err.Error(), http.StatusInternalServerError)
|
||||
} else {
|
||||
result.Data = statuses
|
||||
}
|
||||
})
|
||||
if _, err := jss.GetReplica().Select(&statuses, query, map[string]interface{}{"Status": status}); err != nil {
|
||||
return nil, model.NewAppError("SqlJobStore.GetAllByStatus", "store.sql_job.get_all.app_error", nil, "Status="+status+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return statuses, nil
|
||||
}
|
||||
|
||||
func (jss SqlJobStore) GetNewestJobByStatusAndType(status string, jobType string) store.StoreChannel {
|
||||
return store.Do(func(result *store.StoreResult) {
|
||||
var job *model.Job
|
||||
func (jss SqlJobStore) GetNewestJobByStatusAndType(status string, jobType string) (*model.Job, *model.AppError) {
|
||||
query := "SELECT * FROM Jobs WHERE Status = :Status AND Type = :Type ORDER BY CreateAt DESC LIMIT 1"
|
||||
|
||||
if err := jss.GetReplica().SelectOne(&job,
|
||||
`SELECT
|
||||
*
|
||||
FROM
|
||||
Jobs
|
||||
WHERE
|
||||
Status = :Status
|
||||
AND
|
||||
Type = :Type
|
||||
ORDER BY
|
||||
CreateAt DESC
|
||||
LIMIT 1`, map[string]interface{}{"Status": status, "Type": jobType}); err != nil && err != sql.ErrNoRows {
|
||||
result.Err = model.NewAppError("SqlJobStore.GetNewestJobByStatusAndType", "store.sql_job.get_newest_job_by_status_and_type.app_error", nil, "Status="+status+", "+err.Error(), http.StatusInternalServerError)
|
||||
} else {
|
||||
result.Data = job
|
||||
}
|
||||
})
|
||||
var job *model.Job
|
||||
if err := jss.GetReplica().SelectOne(&job, query, map[string]interface{}{"Status": status, "Type": jobType}); err != nil && err != sql.ErrNoRows {
|
||||
return nil, model.NewAppError("SqlJobStore.GetNewestJobByStatusAndType", "store.sql_job.get_newest_job_by_status_and_type.app_error", nil, "Status="+status+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return job, nil
|
||||
}
|
||||
|
||||
func (jss SqlJobStore) GetCountByStatusAndType(status string, jobType string) store.StoreChannel {
|
||||
return store.Do(func(result *store.StoreResult) {
|
||||
if count, err := jss.GetReplica().SelectInt(`SELECT
|
||||
COUNT(*)
|
||||
FROM
|
||||
Jobs
|
||||
WHERE
|
||||
Status = :Status
|
||||
AND
|
||||
Type = :Type`, map[string]interface{}{"Status": status, "Type": jobType}); err != nil {
|
||||
result.Err = model.NewAppError("SqlJobStore.GetCountByStatusAndType", "store.sql_job.get_count_by_status_and_type.app_error", nil, "Status="+status+", "+err.Error(), http.StatusInternalServerError)
|
||||
} else {
|
||||
result.Data = count
|
||||
}
|
||||
})
|
||||
func (jss SqlJobStore) GetCountByStatusAndType(status string, jobType string) (int64, *model.AppError) {
|
||||
query := "SELECT COUNT(*) FROM Jobs WHERE Status = :Status AND Type = :Type"
|
||||
count, err := jss.GetReplica().SelectInt(query, map[string]interface{}{"Status": status, "Type": jobType})
|
||||
if err != nil {
|
||||
return int64(0), model.NewAppError("SqlJobStore.GetCountByStatusAndType", "store.sql_job.get_count_by_status_and_type.app_error", nil, "Status="+status+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
|
||||
func (jss SqlJobStore) Delete(id string) store.StoreChannel {
|
||||
return store.Do(func(result *store.StoreResult) {
|
||||
if _, err := jss.GetMaster().Exec(
|
||||
`DELETE FROM
|
||||
Jobs
|
||||
WHERE
|
||||
Id = :Id`, map[string]interface{}{"Id": id}); err != nil {
|
||||
result.Err = model.NewAppError("SqlJobStore.DeleteByType", "store.sql_job.delete.app_error", nil, "id="+id+", "+err.Error(), http.StatusInternalServerError)
|
||||
} else {
|
||||
result.Data = id
|
||||
}
|
||||
})
|
||||
func (jss SqlJobStore) Delete(id string) (string, *model.AppError) {
|
||||
query := "DELETE FROM Jobs WHERE Id = :Id"
|
||||
if _, err := jss.GetMaster().Exec(query, map[string]interface{}{"Id": id}); err != nil {
|
||||
return "", model.NewAppError("SqlJobStore.DeleteByType", "store.sql_job.delete.app_error", nil, "id="+id+", "+err.Error(), http.StatusInternalServerError)
|
||||
}
|
||||
return id, nil
|
||||
}
|
||||
|
||||
Ссылка в новой задаче
Block a user