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.{ActorRef, Behavior}
23 import akka.actor.typed.scaladsl.{ActorContext, Behaviors, StashBuffer}
24 import cats.data.{NonEmptyList => Nel}
25 import eu.timepit.refined.auto.autoUnwrap
26 import org.make.api.sequence.SequenceCacheActor._
27 import org.make.api.sequence.SequenceCacheManager.CacheKey
28 import org.make.api.technical.ActorProtocol
29 import org.make.api.technical.sequence.SequenceCacheConfiguration
30 import org.make.core.proposal.indexed.IndexedProposal
31 
32 import scala.util.{Failure, Success}
33 import scala.concurrent.Future
34 import scala.concurrent.duration.Deadline
35 
36 object SequenceCacheActor {
37 
38   def apply(
39     reloadProposals: () => Future[Nel[IndexedProposal]],
40     config: SequenceCacheConfiguration
41   ): Behavior[Protocol] = {
42     Behaviors.withStash(1000) { buffer =>
43       Behaviors.setup { context =>
44         fetchProposals(reloadProposals, context)()
45         new SequenceCacheActor(context, buffer, config, fetchProposals(reloadProposals, context)).init()
46       }
47     }
48   }
49 
50   def fetchProposals(reloadProposals: () => Future[Nel[IndexedProposal]], context: ActorContext[Protocol]): () => Unit =
51     () => {
52       context.pipeToSelf(reloadProposals()) {
53         case Success(proposals) => SetProposalsPoolSuccess(proposals)
54         case Failure(e)         => SetProposalsPoolFailure(e)
55       }
56     }
57 
58   sealed trait Protocol extends ActorProtocol
59 
60   sealed trait Command extends Protocol
61   sealed trait Response extends Protocol
62 
63   final case class GetProposal(replyTo: ActorRef[IndexedProposal]) extends Command
64   final case class SetProposalsPoolSuccess(proposalsPool: Nel[IndexedProposal]) extends Command
65   final case class SetProposalsPoolFailure(error: Throwable) extends Command
66   case object Expire extends Command
67 
68   def name(key: CacheKey): String = s"${key.questionId.value}-${key.kind}"
69 }
70 
71 class SequenceCacheActor private (
72   context: ActorContext[Protocol],
73   buffer: StashBuffer[Protocol],
74   config: SequenceCacheConfiguration,
75   fetchProposals: () => Unit
76 ) {
77 
78   private val untilRefresh: Int = config.proposalsPoolSize * config.cacheRefreshCycles
79 
80   private def newDeadline: Deadline = Deadline.now + config.inactivityTimeout
81 
82   def init(): Behavior[Protocol] = {
83     Behaviors.receiveMessage {
84       case SetProposalsPoolSuccess(proposalsPool) =>
85         buffer.unstashAll(cache(untilRefresh, proposalsPool, proposalsPool.iterator, newDeadline))
86       case SetProposalsPoolFailure(e) =>
87         context.log.error("Refreshing cache failed", e)
88         Behaviors.stopped
89       case cmd =>
90         buffer.stash(cmd)
91         Behaviors.same
92     }
93   }
94 
95   @SuppressWarnings(Array("org.wartremover.warts.Recursion"))
96   def cache(
97     counter: Int,
98     proposalsPool: Nel[IndexedProposal],
99     iterator: Iterator[IndexedProposal],
100     deadline: Deadline
101   ): Behavior[Protocol] = {
102     Behaviors.receiveMessage {
103       case GetProposal(replyTo) =>
104         if (counter == 0) {
105           fetchProposals()
106         }
107         if (iterator.hasNext) {
108           replyTo ! iterator.next()
109           cache(counter - 1, proposalsPool, iterator, newDeadline)
110         } else {
111           val it = proposalsPool.iterator
112           replyTo ! it.next()
113           cache(counter - 1, proposalsPool, it, newDeadline)
114         }
115       case Expire =>
116         if (deadline.isOverdue()) {
117           Behaviors.stopped
118         } else {
119           Behaviors.same
120         }
121       case SetProposalsPoolSuccess(proposalsPool) =>
122         cache(untilRefresh, proposalsPool, proposalsPool.iterator, newDeadline)
123       case SetProposalsPoolFailure(e) =>
124         context.log.error("Refreshing cache failed", e)
125         Behaviors.stopped
126     }
127   }
128 }