1 /*
2  *  Make.org Core API
3  *  Copyright (C) 2018 Make.org
4  *
5  * This program is free software: you can redistribute it and/or modify
6  *  it under the terms of the GNU Affero General Public License as
7  *  published by the Free Software Foundation, either version 3 of the
8  *  License, or (at your option) any later version.
9  *
10  *  This program is distributed in the hope that it will be useful,
11  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
12  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
13  *  GNU Affero General Public License for more details.
14  *
15  *  You should have received a copy of the GNU Affero General Public License
16  *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
17  *
18  */
19 
20 package org.make.core
21 package job
22 
23 import java.time.ZonedDateTime
24 import java.time.{Duration => JavaDuration}
25 
26 import eu.timepit.refined.api.Refined
27 import eu.timepit.refined.numeric.Interval
28 import eu.timepit.refined.{refineV, W}
29 import io.circe.{Decoder, Encoder, Json}
30 import org.make.core.SprayJsonFormatters._
31 import org.make.core.job.Job.JobStatus.Running
32 import org.make.core.job.Job.{missableHeartbeats, JobId, JobStatus}
33 import spray.json.DefaultJsonProtocol._
34 import spray.json._
35 
36 import scala.concurrent.duration.{Duration, DurationInt, FiniteDuration}
37 
38 final case class Job(id: JobId, status: JobStatus, createdAt: Option[ZonedDateTime], updatedAt: Option[ZonedDateTime])
39     extends Timestamped {
40 
41   /** Check whether the job is stuck, given the expected heart rate.
42     *
43     * A job is stuck if it has missed three heartbeats (and did not update otherwise), to allow some leniency if it does
44     * not report itself often and last heartbeats have not yet been received or handled (because e.g. of some network
45     * delay, high load…).
46     */
47   def isStuck(heartRate: Duration): Boolean = {
48     status match {
49       case Running(_) =>
50         updatedAt.forall(JavaDuration.between(_, DateHelper.now()).toMillis > (heartRate.toMillis * missableHeartbeats))
51       case _ => false
52     }
53   }
54 
55 }
56 
57 object Job {
58 
59   val defaultHeartRate: FiniteDuration = 10.seconds
60   val missableHeartbeats: Int = 3
61 
62   type ProgressRefinement = Interval.Closed[W.`0D`.T, W.`100D`.T]
63   type Progress = Double Refined ProgressRefinement
64 
65   implicit val progressJsonFormat: JsonFormat[Progress] = new JsonFormat[Progress] {
66     override def write(obj: Progress): JsValue = JsNumber(obj.value)
67 
68     @SuppressWarnings(Array("org.wartremover.warts.Throw"))
69     override def read(json: JsValue): Progress = json match {
70       case n @ JsNumber(value) =>
71         refineV[ProgressRefinement](value.toDouble) match {
72           case Right(progress) => progress
73           case Left(error)     => throw new IllegalArgumentException(s"Unable to convert $n: $error")
74         }
75       case other => throw new IllegalArgumentException(s"Unable to convert $other")
76     }
77   }
78 
79   implicit val jobJsonFormat: RootJsonFormat[Job] = DefaultJsonProtocol.jsonFormat4(Job.apply)
80 
81   final case class JobId(value: String) extends StringValue
82 
83   object JobId {
84 
85     val Reindex: JobId = JobId("Reindex")
86     val ReindexPosts: JobId = JobId("ReindexPosts")
87     val SyncCrmData: JobId = JobId("SyncCrmData")
88     val AnonymizeInactiveUsers: JobId = JobId("AnonymizeInactiveUsers")
89     val SubmittedAsLanguagePatch: JobId = JobId("SubmittedAsLanguagePatch")
90 
91     final val swaggerAllowableValues = "Reindex,ReindexPosts,SyncCrmData,AnonymizeInactiveUsers"
92 
93     implicit val jobIdFormatter: JsonFormat[JobId] = SprayJsonFormatters.forStringValue(JobId.apply)
94 
95     implicit lazy val jobIdEncoder: Encoder[JobId] = (a: JobId) => Json.fromString(a.value)
96     implicit lazy val jobIdDecoder: Decoder[JobId] = Decoder.decodeString.map(JobId(_))
97   }
98 
99   sealed abstract class JobStatus extends Product with Serializable
100 
101   object JobStatus {
102 
103     final case class Running(progress: Progress) extends JobStatus
104 
105     final case class Finished(outcome: Option[String]) extends JobStatus
106 
107     implicit val statusJsonFormat: JsonFormat[JobStatus] = new JsonFormat[JobStatus] {
108 
109       import SprayJsonFormatters.syntax._
110 
111       private implicit val runningFormat: JsonFormat[Running] = DefaultJsonProtocol.jsonFormat1(Running.apply)
112       private implicit val finishedFormat: JsonFormat[Finished] = DefaultJsonProtocol.jsonFormat1(Finished.apply)
113 
114       override def read(json: JsValue): JobStatus = {
115         json.asJsObject.getFields("kind") match {
116           case Seq(JsString("running"))  => json.as[Running]
117           case Seq(JsString("finished")) => json.as[Finished]
118         }
119       }
120 
121       override def write(obj: JobStatus): JsValue = obj match {
122         case Running(progress) => JsObject("kind" := "running", "progress" := progress)
123         case Finished(outcome) => JsObject("kind" := "finished", "outcome" := outcome)
124       }
125 
126     }
127 
128   }
129 
130 }