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.scaladsl.Behaviors
24 import akka.actor.typed.{ActorRef, Behavior}
25 import cats.data.{NonEmptyList => Nel}
26 import org.make.api.sequence.SequenceBehaviour.ConsensusParam
27 import org.make.api.technical.ActorProtocol
28 import org.make.api.technical.sequence.SequenceCacheConfiguration
29 import org.make.core.proposal.indexed.{IndexedProposal, Zone}
30 import org.make.core.question.QuestionId
31 import org.make.core.sequence.{SequenceConfiguration, SequenceKind}
32 import scalacache._
33 import scalacache.guava._
34 import scalacache.modes.scalaFuture._
35 import scalacache.memoization.memoizeF
36 
37 import scala.concurrent.duration.DurationInt
38 import scala.concurrent.Future
39 import scala.concurrent.ExecutionContext.Implicits.global
40 
41 object SequenceCacheManager {
42 
43   type ConfiguredBehaviour = SequenceConfiguration => Future[SequenceBehaviour]
44 
45   def apply(
46     config: SequenceCacheConfiguration,
47     sequenceService: SequenceService,
48     sequenceConfigurationService: SequenceConfigurationService
49   ): Behavior[Protocol] = {
50     Behaviors.withTimers { timers =>
51       timers.startTimerWithFixedDelay(ExpireChildren, config.checkInactivityTimer)
52       handleMessage(Map.empty)(reloadProposals(config, sequenceService, sequenceConfigurationService), config)
53     }
54   }
55 
56   def reloadProposals(
57     sequenceCacheConfiguration: SequenceCacheConfiguration,
58     sequenceService: SequenceService,
59     sequenceConfigurationService: SequenceConfigurationService
60   )(key: CacheKey, behaviour: ConfiguredBehaviour)(): Future[Nel[IndexedProposal]] = {
61     @SuppressWarnings(Array("org.wartremover.warts.Recursion"))
62     def simpleSequence(behaviour: SequenceBehaviour, retries: Int = 3): Future[Nel[IndexedProposal]] = {
63       sequenceService.simpleSequence(Seq.empty, behaviour, Seq.empty, None).flatMap {
64         case Seq() =>
65           if (retries > 0) {
66             simpleSequence(behaviour, retries - 1)
67           } else {
68             Future.failed(
69               new IllegalStateException(
70                 s"Question ${key.questionId.value} might not have any proposals eligible for sequence."
71               )
72             )
73           }
74         case head +: tail => Future.successful(Nel(head, tail.toList))
75       }
76     }
77 
78     sequenceConfigurationService.getSequenceConfigurationByQuestionId(key.questionId).flatMap { config =>
79       val customConfig: SequenceConfiguration = config
80         .copy(mainSequence = config.mainSequence.copy(sequenceSize = sequenceCacheConfiguration.proposalsPoolSize))
81       behaviour(customConfig).flatMap(simpleSequence(_))
82     }
83   }
84 
85   @SuppressWarnings(Array("org.wartremover.warts.Recursion"))
86   def handleMessage(configCache: Map[CacheKey, ActorRef[SequenceCacheActor.Protocol]])(
87     implicit reloadProposals: (CacheKey, ConfiguredBehaviour) => () => Future[Nel[IndexedProposal]],
88     config: SequenceCacheConfiguration
89   ): Behavior[Protocol] = {
90     Behaviors.setup { context =>
91       def getProposal(cmd: GetProposalCmd, replyTo: ActorRef[IndexedProposal]): Behavior[Protocol] = {
92         configCache.get(cmd.key) match {
93           case None =>
94             val cache: ActorRef[SequenceCacheActor.Protocol] =
95               context.spawn(
96                 SequenceCacheActor(reloadProposals(cmd.key, cmd.behaviour), config),
97                 SequenceCacheActor.name(cmd.key)
98               )
99             context.watchWith(cache, ChildTerminated(cmd.key))
100             cache ! SequenceCacheActor.GetProposal(replyTo)
101             handleMessage(configCache ++ Map(cmd.key -> cache))
102           case Some(actorRef) =>
103             actorRef ! SequenceCacheActor.GetProposal(replyTo)
104             Behaviors.same
105         }
106       }
107 
108       Behaviors
109         .receiveMessage[Protocol] {
110           case cmd: GetProposalCmd => getProposal(cmd, cmd.replyTo)
111           case ExpireChildren =>
112             configCache.values.foreach { ref =>
113               ref ! SequenceCacheActor.Expire
114             }
115             Behaviors.same
116           case ChildTerminated(questionId) =>
117             handleMessage(configCache - questionId)
118         }
119     }
120   }
121 
122   implicit val guavaCache: Cache[ConsensusParam] = GuavaCache[ConsensusParam]
123 
124   def getCachedThreshold(getThreshold: () => Future[ConsensusParam]): Future[ConsensusParam] = {
125     memoizeF[Future, ConsensusParam](Some(1.hour)) {
126       getThreshold()
127     }
128   }
129 
130   sealed trait Protocol extends ActorProtocol
131 
132   sealed trait Command extends Protocol
133 
134   sealed trait GetProposalCmd extends Command {
135     val replyTo: ActorRef[IndexedProposal]
136     val key: CacheKey
137     def behaviour: ConfiguredBehaviour
138   }
139 
140   final case class GetProposal(id: QuestionId, replyTo: ActorRef[IndexedProposal]) extends GetProposalCmd {
141     override val key: CacheKey = CacheKey(id, SequenceKind.Standard)
142     override def behaviour: ConfiguredBehaviour = { config =>
143       Future.successful(SequenceBehaviourProvider[Unit].behaviour((), config, key.questionId, None))
144     }
145   }
146 
147   final case class GetConsensusProposal(
148     id: QuestionId,
149     threshold: () => Future[ConsensusParam],
150     replyTo: ActorRef[IndexedProposal]
151   ) extends GetProposalCmd {
152     override val key: CacheKey = CacheKey(id, SequenceKind.Consensus)
153     override def behaviour: ConfiguredBehaviour = { config =>
154       getCachedThreshold(threshold).map { param =>
155         SequenceBehaviourProvider[ConsensusParam].behaviour(param, config, key.questionId, None)
156       }
157     }
158   }
159 
160   final case class GetControversyProposal(id: QuestionId, replyTo: ActorRef[IndexedProposal]) extends GetProposalCmd {
161     override val key: CacheKey = CacheKey(id, SequenceKind.Controversy)
162     override def behaviour: ConfiguredBehaviour = { config =>
163       Future.successful(
164         SequenceBehaviourProvider[Zone.Controversy.type].behaviour(Zone.Controversy, config, key.questionId, None)
165       )
166     }
167   }
168   case object ExpireChildren extends Command
169   final case class ChildTerminated(key: CacheKey) extends Command
170 
171   val name: String = "sequence-cache-manager"
172   val SequenceCacheActorKey: ServiceKey[Protocol] = ServiceKey(name)
173 
174   final case class CacheKey(questionId: QuestionId, kind: SequenceKind)
175 }