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
21 
22 import akka.actor.typed.receptionist.{Receptionist, ServiceKey}
23 import akka.actor.typed.scaladsl.{ActorContext, Behaviors}
24 import akka.actor.typed.{ActorRef, Behavior, Props, SpawnProtocol}
25 import org.make.api.idea.{IdeaConsumerBehavior, IdeaProducerBehavior}
26 import org.make.api.kafka.kafkaDispatcher
27 import org.make.api.proposal.ProposalSupervisor
28 import org.make.api.sequence.{SequenceCacheManager, SequenceConfigurationActor}
29 import org.make.api.sequence.SequenceConfigurationActor.SequenceConfigurationActorProtocol
30 import org.make.api.sessionhistory.SessionHistoryCoordinator
31 import org.make.api.technical.crm._
32 import org.make.api.technical.job.JobCoordinator
33 import org.make.api.technical.tracking.{
34   DemographicsProducerBehavior,
35   PanoramicProducerBehavior,
36   TrackingProducerBehavior
37 }
38 import org.make.api.technical.{ActorProtocol, ActorSystemHelper, DeadLettersListenerActor, MakeDowningActor}
39 import org.make.api.user.UserSupervisor
40 import org.make.api.userhistory.{UserHistoryCommand, UserHistoryCoordinator}
41 import org.make.core.job.Job
42 
43 object MakeGuardian {
44   val name: String = "make-api"
45 
46   sealed trait GuardianProtocol extends ActorProtocol
47 
48   sealed trait GuardianCommand extends GuardianProtocol
49   final case class Initialize(replyTo: ActorRef[Initialized.type]) extends GuardianCommand
50 
51   sealed trait GuardianResponse extends GuardianProtocol
52   case object Initialized extends GuardianResponse
53 
54   val SpawnActorKey: ServiceKey[SpawnProtocol.Command] = ServiceKey("spawn-actor")
55 
56   def createBehavior(dependencies: MakeApi): Behavior[GuardianCommand] = {
57     Behaviors.setup { implicit context =>
58       Behaviors.receiveMessagePartial {
59         case Initialize(sender) =>
60           createTechnicalActors(dependencies)
61           createFunctionalActors(dependencies)
62           createProducersAndConsumers(dependencies)
63           sender ! Initialized
64           initialized()
65       }
66     }
67   }
68 
69   private def initialized(): Behavior[GuardianCommand] = {
70     Behaviors.receiveMessagePartial {
71       case Initialize(sender) =>
72         sender ! Initialized
73         Behaviors.same
74     }
75   }
76 
77   private def createTechnicalActors(dependencies: MakeApi)(implicit context: ActorContext[_]): Unit = {
78     context.watch(context.spawn(MakeDowningActor(), MakeDowningActor.name))
79     context.watch(context.spawn(DeadLettersListenerActor(), DeadLettersListenerActor.name))
80     val spawnActorRef = context.spawn(SpawnProtocol(), MakeGuardian.SpawnActorKey.id)
81     context.watch(spawnActorRef)
82     context.system.receptionist ! Receptionist.Register(MakeGuardian.SpawnActorKey, spawnActorRef)
83 
84     val sequenceConfigurationCacheActor: ActorRef[SequenceConfigurationActorProtocol] = context.spawn(
85       ActorSystemHelper.superviseWithBackoff(
86         SequenceConfigurationActor(dependencies.persistentSequenceConfigurationService)
87       ),
88       SequenceConfigurationActor.name
89     )
90     context.watch(sequenceConfigurationCacheActor)
91     context.system.receptionist ! Receptionist.Register(
92       SequenceConfigurationActor.SequenceConfigurationCacheActorKey,
93       sequenceConfigurationCacheActor
94     )
95     val sequenceCacheManager: ActorRef[SequenceCacheManager.Protocol] = context.spawn(
96       ActorSystemHelper.superviseWithBackoff(
97         SequenceCacheManager(
98           dependencies.sequenceCacheConfiguration,
99           dependencies.sequenceService,
100           dependencies.sequenceConfigurationService
101         )
102       ),
103       SequenceCacheManager.name
104     )
105     context.watch(sequenceCacheManager)
106     context.system.receptionist ! Receptionist.Register(
107       SequenceCacheManager.SequenceCacheActorKey,
108       sequenceCacheManager
109     )
110   }
111 
112   private def createFunctionalActors(dependencies: MakeApi)(implicit context: ActorContext[_]): Unit = {
113     val userHistoryCoordinator: ActorRef[UserHistoryCommand] = UserHistoryCoordinator(system = context.system)
114     context.watch(userHistoryCoordinator)
115     context.system.receptionist ! Receptionist.Register(UserHistoryCoordinator.Key, userHistoryCoordinator)
116 
117     val sessionHistoryCoordinator =
118       SessionHistoryCoordinator(
119         context.system,
120         userHistoryCoordinator,
121         dependencies.idGenerator,
122         dependencies.makeSettings
123       )
124     context.watch(sessionHistoryCoordinator)
125     context.system.receptionist ! Receptionist.Register(SessionHistoryCoordinator.Key, sessionHistoryCoordinator)
126 
127     context.watch(
128       context.spawn[Nothing](
129         ProposalSupervisor(dependencies, dependencies.makeSettings.lockDuration),
130         ProposalSupervisor.name
131       )
132     )
133 
134     context.watch(context.spawn[Nothing](UserSupervisor(dependencies), UserSupervisor.name))
135 
136     val jobCoordinatorRef = JobCoordinator(context.system, Job.defaultHeartRate)
137 
138     context.watch(jobCoordinatorRef)
139     context.system.receptionist ! Receptionist.Register(JobCoordinator.Key, jobCoordinatorRef)
140   }
141 
142   private def createProducersAndConsumers(dependencies: MakeApi)(implicit context: ActorContext[_]): Unit = {
143     spawnKafkaActorWithBackoff(DemographicsProducerBehavior(), DemographicsProducerBehavior.name)
144     spawnKafkaActorWithBackoff(IdeaProducerBehavior(), IdeaProducerBehavior.name)
145     spawnKafkaActorWithBackoff(
146       IdeaConsumerBehavior(dependencies.ideaService, dependencies.elasticsearchIdeaAPI),
147       IdeaConsumerBehavior.name
148     )
149     spawnKafkaActorWithBackoff(MailJetProducerBehavior(), MailJetProducerBehavior.name)
150     spawnKafkaActorWithBackoff(MailJetConsumerBehavior(dependencies.crmService), MailJetConsumerBehavior.name)
151     spawnKafkaActorWithBackoff(MailJetEventProducerBehavior(), MailJetEventProducerBehavior.name)
152     spawnKafkaActorWithBackoff(
153       MailJetEventConsumerBehavior(dependencies.userService),
154       MailJetEventConsumerBehavior.name
155     )
156     spawnKafkaActorWithBackoff(TrackingProducerBehavior(), TrackingProducerBehavior.name)
157     spawnKafkaActorWithBackoff(PanoramicProducerBehavior(), PanoramicProducerBehavior.name)
158   }
159 
160   private def spawnKafkaActorWithBackoff(behavior: Behavior[_], name: String)(
161     implicit context: ActorContext[_]
162   ): Unit = {
163     context.watch(
164       context.spawn(
165         ActorSystemHelper.superviseWithBackoff(behavior),
166         name,
167         Props.empty.withDispatcherFromConfig(kafkaDispatcher)
168       )
169     )
170   }
171 }