clear queue

cmd/api-service/handler/job.go

@@ -31,6 +31,8 @@ func (jobAPI *JobAPI) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		jobAPI.List(w, r)
 	case r.Method == http.MethodDelete && subPath != "":
 		jobAPI.Delete(w, r, subPath)
+	case r.Method == http.MethodDelete && subPath == "":
+		jobAPI.DeleteAll(w, r)
 	default:
 		Error(w, http.StatusNotFound, "unregistered path", nil, logger)
 	}
@@ -81,3 +83,14 @@ func (jobAPI *JobAPI) Delete(w http.ResponseWriter, r *http.Request, id string)
 
 	w.WriteHeader(http.StatusNoContent)
 }
+
+func (jobAPI *JobAPI) DeleteAll(w http.ResponseWriter, r *http.Request) {
+	logger := jobAPI.logger.With("method", "deleteall")
+
+	if err := jobAPI.jq.DeleteAll(); err != nil {
+		Error(w, http.StatusInternalServerError, "could not delete all jobs", err, logger)
+		return
+	}
+
+	w.WriteHeader(http.StatusNoContent)
+}

cmd/api-service/job/queue.go

@@ -77,7 +77,7 @@ WHERE id=?`, job.ID); err != nil {
 	}
 }
 
-func (jq *JobQueue) MarkDone(id string) {
+func (jq *JobQueue) MarkDone(id int) {
 	logger := jq.logger.With("method", "markdone")
 	if _, err := jq.db.Exec(`
 UPDATE job_queue SET 
@@ -117,3 +117,11 @@ WHERE id=?`, id); err != nil {
 	}
 	return nil
 }
+
+func (jq *JobQueue) DeleteAll() error {
+	if _, err := jq.db.Exec(`
+DELETE FROM job_queue`); err != nil {
+		return err
+	}
+	return nil
+}

cmd/api-service/job/worker.go

@@ -55,6 +55,9 @@ func (w *Worker) RefreshAllReviews(jobID int) {
 			return
 		}
 	}
+
+	logger.Info("refresh all reviews", "count", len(movies))
+	w.jq.MarkDone(jobID)
 }
 
 func (w *Worker) RefreshReviews(jobID int, movieID string) {
@@ -85,4 +88,5 @@ func (w *Worker) RefreshReviews(jobID int, movieID string) {
 	}
 
 	logger.Info("refresh reviews", "count", len(reviews))
+	w.jq.MarkDone(jobID)
 }