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.organisation
21 
22 import akka.Done
23 import akka.actor.typed.{ActorSystem, Behavior}
24 import akka.util.Timeout
25 import grizzled.slf4j.Logging
26 import org.make.api.technical.KafkaConsumerBehavior
27 import org.make.api.technical.KafkaConsumerBehavior.Protocol
28 import org.make.api.user.UserProducerBehavior
29 import org.make.api.userhistory.{OrganisationRegisteredEvent, OrganisationUpdatedEvent, UserEvent, UserEventWrapper}
30 import org.make.core.user.UserId
31 import org.make.core.user.indexed.IndexedOrganisation
32 
33 import scala.concurrent.ExecutionContext.Implicits.global
34 import scala.concurrent.Future
35 import scala.concurrent.duration.DurationInt
36 
37 class OrganisationConsumerBehavior(
38   organisationService: OrganisationService,
39   elasticsearchOrganisationAPI: OrganisationSearchEngine
40 ) extends KafkaConsumerBehavior[UserEventWrapper]
41     with Logging {
42 
43   override protected val topicKey: String = UserProducerBehavior.topicKey
44 
45   implicit val timeout: Timeout = Timeout(5.seconds)
46 
47   override def handleMessage(message: UserEventWrapper)(system: ActorSystem[_]): Future[_] = {
48     message.event match {
49       case event: OrganisationRegisteredEvent => onCreateOrUpdate(event)
50       case event: OrganisationUpdatedEvent    => onCreateOrUpdate(event)
51       case event                              => doNothing(event)
52     }
53   }
54 
55   def onCreateOrUpdate(event: UserEvent): Future[Done] = {
56     retrieveAndShapeOrganisation(event.userId).flatMap(indexOrUpdate)
57   }
58 
59   def indexOrUpdate(organisation: IndexedOrganisation): Future[Done] = {
60     debug(s"Indexing $organisation")
61     elasticsearchOrganisationAPI
62       .findOrganisationById(organisation.organisationId)
63       .flatMap {
64         case None => elasticsearchOrganisationAPI.indexOrganisation(organisation)
65         case Some(found) =>
66           elasticsearchOrganisationAPI.updateOrganisation(organisation.copy(countsByQuestion = found.countsByQuestion))
67       }
68   }
69 
70   private def retrieveAndShapeOrganisation(id: UserId): Future[IndexedOrganisation] = {
71     organisationService.getOrganisation(id).flatMap {
72       case None               => Future.failed(new IllegalArgumentException(s"Organisation ${id.value} doesn't exist"))
73       case Some(organisation) => Future.successful(IndexedOrganisation.createFromOrganisation(organisation))
74     }
75   }
76 
77   override val groupId = "organisation-consumer"
78 }
79 
80 object OrganisationConsumerBehavior {
81   def apply(
82     organisationService: OrganisationService,
83     elasticsearchOrganisationAPI: OrganisationSearchEngine
84   ): Behavior[Protocol] =
85     new OrganisationConsumerBehavior(organisationService, elasticsearchOrganisationAPI)
86       .createBehavior(name)
87 
88   val name: String = "organisation-consumer"
89 }