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.api.technical.job
21 
22 import akka.actor.typed.scaladsl.Behaviors
23 import akka.actor.typed.{ActorRef, Behavior}
24 import akka.persistence.typed.PersistenceId
25 import akka.persistence.typed.scaladsl.{Effect, EventSourcedBehavior, RetentionCriteria}
26 import eu.timepit.refined.auto._
27 import org.make.api.technical.job.JobActor.Protocol.Command._
28 import org.make.api.technical.job.JobActor.Protocol.Response._
29 import org.make.api.technical.job.JobEvent._
30 import org.make.api.technical.{ActorCommand, ActorProtocol}
31 import org.make.core.job.Job
32 import org.make.core.job.Job.{JobId, JobStatus, Progress}
33 import org.make.core.{DateHelper, MakeSerializable}
34 import spray.json.{DefaultJsonProtocol, JsString, JsValue, RootJsonFormat}
35 
36 import scala.concurrent.duration.Duration
37 
38 object JobActor {
39 
40   sealed trait JobState extends MakeSerializable {
41     def isStopped: Boolean
42     def isStuck(heartRate: Duration): Boolean
43     def handleCommand(heartRate: Duration): JobActor.Protocol.Command => Effect[JobEvent, JobState]
44     def handleEvent: JobEvent                                         => JobState
45   }
46 
47   case object EmptyJob extends JobState {
48     override def isStopped: Boolean = true
49     override def isStuck(heartRate: Duration): Boolean = false
50 
51     override def handleCommand(heartRate: Duration): JobActor.Protocol.Command => Effect[JobEvent, JobState] = {
52       case Start(id, replyTo: ActorRef[JobAcceptance]) =>
53         Effect.persist(Started(id, DateHelper.now())).thenReply(replyTo)(_ => JobAcceptance(true))
54       case command: StatusCommand => Effect.reply(command.replyTo)(NotRunning)
55       case Get(_, replyTo)        => Effect.reply(replyTo)(State(None))
56       case Kill(_)                => Effect.stop().thenStop()
57     }
58 
59     override def handleEvent: JobEvent => JobState = {
60       case Started(id, date) => DefinedJob(Job(id, JobStatus.Running(0d), Some(date), Some(date)))
61       case _                 => this
62     }
63 
64     implicit val emptyJobJsonFormat: RootJsonFormat[EmptyJob.type] = new RootJsonFormat[EmptyJob.type] {
65       private val Key = "empty"
66 
67       @SuppressWarnings(Array("org.wartremover.warts.Throw"))
68       override def read(json: JsValue): EmptyJob.type = json match {
69         case JsString(Key) => EmptyJob
70         case other         => throw new IllegalStateException(s"$other is not an EmptyJob.")
71       }
72 
73       override def write(obj: EmptyJob.type): JsValue = JsString(Key)
74     }
75   }
76 
77   final case class DefinedJob(job: Job) extends JobState {
78     private def persistIfRunning(replyTo: ActorRef[Process], event: JobEvent): Effect[JobEvent, JobState] = {
79       if (isStopped) {
80         Effect.reply(replyTo)(NotRunning)
81       } else {
82         Effect.persist(event).thenReply(replyTo)(_ => Ack)
83       }
84     }
85 
86     override def isStopped: Boolean = {
87       job.status match {
88         case JobStatus.Running(_) => false
89         case _                    => true
90       }
91     }
92 
93     override def isStuck(heartRate: Duration): Boolean = job.isStuck(heartRate)
94 
95     override def handleCommand(heartRate: Duration): JobActor.Protocol.Command => Effect[JobEvent, JobState] = {
96       case Start(id, replyTo: ActorRef[JobAcceptance]) =>
97         if (isStopped || isStuck(heartRate)) {
98           Effect.persist(Started(id, DateHelper.now())).thenReply(replyTo)(_ => JobAcceptance(true))
99         } else {
100           Effect.reply(replyTo)(JobAcceptance(false))
101         }
102       case Heartbeat(id, replyTo) => persistIfRunning(replyTo, HeartbeatReceived(id, DateHelper.now()))
103       case Report(id, progress, replyTo) =>
104         persistIfRunning(replyTo, Progressed(id, DateHelper.now(), progress))
105       case Finish(id, outcome, replyTo) =>
106         persistIfRunning(replyTo, Finished(id, DateHelper.now(), outcome.flatMap(e => Option(e.getMessage))))
107       case Get(_, replyTo) => Effect.reply(replyTo)(State(Some(job)))
108       case Kill(_)         => Effect.stop().thenStop()
109     }
110 
111     override def handleEvent: JobEvent => JobState = {
112       case Started(id, date) => DefinedJob(Job(id, JobStatus.Running(0d), Some(date), Some(date)))
113       case HeartbeatReceived(_, date) =>
114         DefinedJob(job.copy(updatedAt = Some(date)))
115       case Progressed(_, date, progress) =>
116         DefinedJob(job.copy(status = JobStatus.Running(progress), updatedAt = Some(date)))
117       case Finished(_, date, outcome) =>
118         DefinedJob(job.copy(status = JobStatus.Finished(outcome), updatedAt = Some(date)))
119     }
120   }
121 
122   object DefinedJob {
123     implicit val jsonFormat: RootJsonFormat[DefinedJob] = DefaultJsonProtocol.jsonFormat1(DefinedJob.apply)
124   }
125 
126   sealed abstract class Protocol extends ActorProtocol
127 
128   object Protocol {
129     sealed abstract class Command extends Protocol with ActorCommand[JobId]
130 
131     object Command {
132       sealed trait StatusCommand {
133         def replyTo: ActorRef[Process]
134       }
135 
136       final case class Start(id: JobId, replyTo: ActorRef[JobAcceptance]) extends Command
137       final case class Heartbeat(id: JobId, replyTo: ActorRef[Process]) extends Command with StatusCommand
138       final case class Report(id: JobId, progress: Progress, replyTo: ActorRef[Process])
139           extends Command
140           with StatusCommand
141       final case class Finish(id: JobId, failure: Option[Throwable], replyTo: ActorRef[Process])
142           extends Command
143           with StatusCommand
144       final case class Get(id: JobId, replyTo: ActorRef[State]) extends Command
145       final case class Kill(id: JobId) extends Command
146     }
147 
148     sealed abstract class Response extends Protocol
149 
150     object Response {
151       final case class JobAcceptance(isAccepted: Boolean) extends Response
152       final case class State(value: Option[Job]) extends Response
153 
154       sealed trait Process extends Response
155       final case object Ack extends Process
156       final case object NotRunning extends Process
157     }
158   }
159 
160   val JournalPluginId: String = "make-api.event-sourcing.jobs.journal"
161   val SnapshotPluginId: String = "make-api.event-sourcing.jobs.snapshot"
162 
163   def apply(heartRate: Duration): Behavior[Protocol.Command] = {
164     Behaviors.setup { context =>
165       val id: JobId = JobId(context.self.path.name)
166       val persistenceId: PersistenceId = PersistenceId.ofUniqueId(id.value)
167       EventSourcedBehavior[Protocol.Command, JobEvent, JobState](
168         persistenceId,
169         emptyState = EmptyJob,
170         commandHandler(heartRate),
171         eventHandler
172       ).withJournalPluginId(JournalPluginId)
173         .withSnapshotPluginId(SnapshotPluginId)
174         .withRetention(RetentionCriteria.snapshotEvery(numberOfEvents = 10, keepNSnapshots = 50))
175     }
176   }
177 
178   def commandHandler(heartRate: Duration)(state: JobState, command: Protocol.Command): Effect[JobEvent, JobState] = {
179     state.handleCommand(heartRate)(command)
180   }
181 
182   def eventHandler(state: JobState, event: JobEvent): JobState = state.handleEvent(event)
183 
184 }