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.sequence
21 
22 import akka.actor.typed.receptionist.ServiceKey
23 import akka.actor.typed.{ActorRef, Behavior}
24 import akka.actor.typed.scaladsl.Behaviors
25 import org.make.api.sequence.SequenceConfigurationActor.SequenceConfigurationActorProtocol
26 import org.make.core.question.QuestionId
27 import org.make.core.sequence.SequenceConfiguration
28 
29 import scala.concurrent.duration.DurationInt
30 import scala.util.{Failure, Success}
31 
32 object SequenceConfigurationActor {
33   def apply(
34     persistentSequenceConfigurationService: PersistentSequenceConfigurationService
35   ): Behavior[SequenceConfigurationActorProtocol] = {
36     Behaviors.withTimers { timers =>
37       timers.startTimerWithFixedDelay(ReloadSequenceConfiguration, 5.minutes)
38       handleMessages(Map.empty)(persistentSequenceConfigurationService)
39     }
40   }
41 
42   @SuppressWarnings(Array("org.wartremover.warts.Recursion"))
43   private def handleMessages(configCache: Map[QuestionId, SequenceConfiguration])(
44     implicit persistentSequenceConfigurationService: PersistentSequenceConfigurationService
45   ): Behavior[SequenceConfigurationActorProtocol] = {
46     Behaviors.setup { context =>
47       Behaviors.receiveMessage {
48         case ReloadSequenceConfiguration =>
49           context.pipeToSelf(persistentSequenceConfigurationService.findAll()) {
50             case Success(configurations) => UpdateSequenceConfiguration(configurations)
51             case Failure(e)              => UpdateSequenceConfigurationsFailed(e)
52           }
53           Behaviors.same
54         case UpdateSequenceConfiguration(configurations) =>
55           val configCache = configurations.map { configuration =>
56             configuration.questionId -> configuration
57           }.toMap
58           handleMessages(configCache)
59         case UpdateSequenceConfigurationsFailed(e) =>
60           context.log.error("Unable to reload sequence configurations:", e)
61           Behaviors.same
62         case GetSequenceConfiguration(questionId, replyTo) =>
63           replyTo ! CachedSequenceConfiguration(configCache.getOrElse(questionId, SequenceConfiguration.default))
64           Behaviors.same
65       }
66     }
67   }
68 
69   sealed trait SequenceConfigurationActorProtocol
70   case object ReloadSequenceConfiguration extends SequenceConfigurationActorProtocol
71   private final case class UpdateSequenceConfiguration(configurations: Seq[SequenceConfiguration])
72       extends SequenceConfigurationActorProtocol
73   private final case class UpdateSequenceConfigurationsFailed(e: Throwable) extends SequenceConfigurationActorProtocol
74 
75   final case class GetSequenceConfiguration(questionId: QuestionId, replyTo: ActorRef[CachedSequenceConfiguration])
76       extends SequenceConfigurationActorProtocol
77 
78   final case class CachedSequenceConfiguration(sequenceConfiguration: SequenceConfiguration)
79   val name: String = "sequence-configuration-cache"
80   val SequenceConfigurationCacheActorKey: ServiceKey[SequenceConfigurationActorProtocol] = ServiceKey(name)
81 
82 }
83 
84 trait SequenceConfigurationActorComponent {
85   def sequenceConfigurationActor: ActorRef[SequenceConfigurationActorProtocol]
86 }