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.apache.kafka.clients.consumer.ConsumerConfig
26 import org.make.api.technical.{KafkaConsumerBehavior, TimeSettings}
27 import org.make.api.userhistory._
28 
29 import java.util.Properties
30 import scala.concurrent.Future
31 import scala.concurrent.duration.{DurationInt, FiniteDuration}
32 
33 class UserImageConsumerBehavior(userService: UserService) extends KafkaConsumerBehavior[UserEventWrapper] with Logging {
34 
35   override protected val topicKey: String = UserProducerBehavior.topicKey
36   override val groupId = "user-images"
37 
38   override def customProperties: Properties = {
39     val props = new Properties()
40 // Fetch a batch of events only in order to handle some downloads but not all at once
41     props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 4)
42     props
43   }
44   override def handleMessagesTimeout: FiniteDuration = 5.minutes
45 
46   implicit val timeout: Timeout = TimeSettings.defaultTimeout
47 
48   override def handleMessage(message: UserEventWrapper)(system: ActorSystem[_]): Future[Unit] = {
49     message.event match {
50       case event: ResetPasswordEvent              => doNothing(event)
51       case event: UserRegisteredEvent             => doNothing(event)
52       case event: UserConnectedEvent              => doNothing(event)
53       case event: UserUpdatedTagEvent             => doNothing(event)
54       case event: ResendValidationEmailEvent      => doNothing(event)
55       case event: UserValidatedAccountEvent       => doNothing(event)
56       case event: OrganisationRegisteredEvent     => doNothing(event)
57       case event: OrganisationUpdatedEvent        => doNothing(event)
58       case event: OrganisationEmailChangedEvent   => doNothing(event)
59       case event: PersonalityEmailChangedEvent    => doNothing(event)
60       case event: OrganisationInitializationEvent => doNothing(event)
61       case event: UserUpdatedOptInNewsletterEvent => doNothing(event)
62       case event: UserAnonymizedEvent             => doNothing(event)
63       case event: UserFollowEvent                 => doNothing(event)
64       case event: UserUnfollowEvent               => doNothing(event)
65       case event: UserUploadAvatarEvent           => handleUserUploadAvatarEvent(event)
66       case event: PersonalityRegisteredEvent      => doNothing(event)
67       case event: VoteOnlyEvent                   => doNothing(event)
68       case event: VoteOnlyTestEvent               => doNothing(event)
69     }
70   }
71 
72   def handleUserUploadAvatarEvent(event: UserUploadAvatarEvent): Future[Unit] = {
73     userService.changeAvatarForUser(event.userId, event.avatarUrl, event.requestContext, event.eventDate)
74   }
75 }
76 
77 object UserImageConsumerBehavior {
78   def apply(userService: UserService): Behavior[KafkaConsumerBehavior.Protocol] =
79     new UserImageConsumerBehavior(userService).createBehavior(name)
80   val name: String = "user-image-consumer"
81 }