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.user
21 
22 import akka.actor.typed.{ActorSystem, Behavior}
23 import akka.util.Timeout
24 import grizzled.slf4j.Logging
25 import org.make.api.technical.KafkaConsumerBehavior.Protocol
26 import org.make.api.technical.{KafkaConsumerBehavior, TimeSettings}
27 import org.make.api.userhistory._
28 
29 import scala.concurrent.ExecutionContext.Implicits.global
30 import scala.concurrent.Future
31 
32 class UserHistoryConsumerBehavior(
33   userHistoryCoordinatorService: UserHistoryCoordinatorService,
34   userService: UserService
35 ) extends KafkaConsumerBehavior[UserEventWrapper]
36     with Logging {
37 
38   override protected val topicKey: String = UserProducerBehavior.topicKey
39   override val groupId = "user-history"
40 
41   implicit val timeout: Timeout = TimeSettings.defaultTimeout
42 
43   override def handleMessage(message: UserEventWrapper)(system: ActorSystem[_]): Future[Unit] = {
44     message.event match {
45       case event: ResetPasswordEvent              => doNothing(event)
46       case event: UserRegisteredEvent             => handleUserRegisteredEvent(event)
47       case event: UserConnectedEvent              => handleUserConnectedEvent(event)
48       case event: UserUpdatedTagEvent             => doNothing(event)
49       case event: ResendValidationEmailEvent      => doNothing(event)
50       case event: UserValidatedAccountEvent       => doNothing(event)
51       case event: OrganisationRegisteredEvent     => doNothing(event)
52       case event: OrganisationUpdatedEvent        => doNothing(event)
53       case event: OrganisationEmailChangedEvent   => handleOrganisationEmailChangedEvent(event)
54       case event: PersonalityEmailChangedEvent    => handlePersonalityEmailChangedEvent(event)
55       case event: OrganisationInitializationEvent => doNothing(event)
56       case event: UserUpdatedOptInNewsletterEvent => handleUserUpdatedOptInNewsletterEvent(event)
57       case event: UserAnonymizedEvent             => handleUserAnonymizedEvent(event)
58       case event: UserFollowEvent                 => doNothing(event)
59       case event: UserUnfollowEvent               => doNothing(event)
60       case event: UserUploadAvatarEvent           => doNothing(event)
61       case event: PersonalityRegisteredEvent      => doNothing(event)
62       case event: VoteOnlyEvent                   => doNothing(event)
63       case event: VoteOnlyTestEvent               => doNothing(event)
64     }
65   }
66 
67   def handleUserRegisteredEvent(event: UserRegisteredEvent): Future[Unit] = {
68     userHistoryCoordinatorService.logHistory(
69       LogRegisterCitizenEvent(
70         userId = event.userId,
71         requestContext = event.requestContext,
72         action = UserAction(
73           date = event.eventDate,
74           actionType = LogRegisterCitizenEvent.action,
75           arguments = UserRegistered(firstName = event.firstName, country = event.country)
76         )
77       )
78     )
79 
80     Future.unit
81   }
82 
83   def handleUserConnectedEvent(event: UserConnectedEvent): Future[Unit] = {
84     userService.getUser(event.userId).map {
85       case Some(_) =>
86         userHistoryCoordinatorService.logHistory(
87           LogUserConnectedEvent(
88             userId = event.userId,
89             requestContext = event.requestContext,
90             action = UserAction(
91               date = event.eventDate,
92               actionType = LogUserConnectedEvent.action,
93               arguments = UserHasConnected()
94             )
95           )
96         )
97       case None =>
98         warn(s"User not found after UserConnectedEvent: $event")
99     }
100   }
101 
102   def handleUserUpdatedOptInNewsletterEvent(event: UserUpdatedOptInNewsletterEvent): Future[Unit] = {
103     if (event.optInNewsletter) {
104       userHistoryCoordinatorService.logHistory(
105         LogUserOptInNewsletterEvent(
106           userId = event.userId,
107           requestContext = event.requestContext,
108           action = UserAction(
109             date = event.eventDate,
110             actionType = LogUserOptInNewsletterEvent.action,
111             arguments = UserUpdatedOptIn(newOptIn = event.optInNewsletter)
112           )
113         )
114       )
115     } else {
116       userHistoryCoordinatorService.logHistory(
117         LogUserOptOutNewsletterEvent(
118           userId = event.userId,
119           requestContext = event.requestContext,
120           action = UserAction(
121             date = event.eventDate,
122             actionType = LogUserOptOutNewsletterEvent.action,
123             arguments = UserUpdatedOptIn(newOptIn = event.optInNewsletter)
124           )
125         )
126       )
127     }
128 
129     Future.unit
130   }
131 
132   def handleUserAnonymizedEvent(event: UserAnonymizedEvent): Future[Unit] = {
133     userHistoryCoordinatorService.logHistory(
134       LogUserAnonymizedEvent(
135         userId = event.userId,
136         requestContext = event.requestContext,
137         action = UserAction(
138           date = event.eventDate,
139           actionType = LogUserAnonymizedEvent.action,
140           arguments = UserAnonymized(userId = event.userId, adminId = event.adminId, mode = event.mode)
141         )
142       )
143     )
144 
145     Future.unit
146   }
147 
148   def handleOrganisationEmailChangedEvent(event: OrganisationEmailChangedEvent): Future[Unit] = {
149     userHistoryCoordinatorService.logHistory(
150       LogOrganisationEmailChangedEvent(
151         userId = event.userId,
152         requestContext = event.requestContext,
153         action = UserAction(
154           date = event.eventDate,
155           actionType = LogOrganisationEmailChangedEvent.action,
156           arguments = OrganisationEmailChanged(oldEmail = event.oldEmail, newEmail = event.newEmail)
157         )
158       )
159     )
160 
161     Future.unit
162   }
163 
164   def handlePersonalityEmailChangedEvent(event: PersonalityEmailChangedEvent): Future[Unit] = {
165     userHistoryCoordinatorService.logHistory(
166       LogPersonalityEmailChangedEvent(
167         userId = event.userId,
168         requestContext = event.requestContext,
169         action = UserAction(
170           date = event.eventDate,
171           actionType = LogPersonalityEmailChangedEvent.action,
172           arguments = PersonalityEmailChanged(oldEmail = event.oldEmail, newEmail = event.newEmail)
173         )
174       )
175     )
176 
177     Future.unit
178   }
179 }
180 
181 object UserHistoryConsumerBehavior {
182   def apply(
183     userHistoryCoordinatorService: UserHistoryCoordinatorService,
184     userService: UserService
185   ): Behavior[Protocol] =
186     new UserHistoryConsumerBehavior(userHistoryCoordinatorService, userService)
187       .createBehavior(name)
188   val name: String = "user-history-consumer"
189 }