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.technical.crm
21 
22 import akka.actor.typed.{ActorSystem, Behavior}
23 import org.make.api.technical.KafkaConsumerBehavior
24 import org.make.api.user.UserService
25 import org.make.core.user.MailingErrorLog
26 
27 import java.time.ZonedDateTime
28 import scala.concurrent.ExecutionContext.Implicits.global
29 import scala.concurrent.Future
30 
31 class MailJetEventConsumerBehavior(userService: UserService) extends KafkaConsumerBehavior[MailJetEventWrapper] {
32 
33   override protected val topicKey: String = MailJetEventProducerBehavior.topicKey
34   override val groupId: String = "mailJet-event-consumer"
35 
36   override def handleMessage(message: MailJetEventWrapper)(system: ActorSystem[_]): Future[_] = {
37     message.event match {
38       case event: MailJetBounceEvent      => handleBounceEvent(event, message.date)
39       case event: MailJetBlockedEvent     => doNothing(event)
40       case event: MailJetSpamEvent        => handleSpamEvent(event)
41       case event: MailJetUnsubscribeEvent => handleUnsubscribeEvent(event)
42       case event                          => doNothing(event)
43     }
44   }
45 
46   def handleBounceEvent(event: MailJetBounceEvent, date: ZonedDateTime): Future[Boolean] = {
47     for {
48       resultUpdateBounce <- userService.updateIsHardBounce(email = event.email, isHardBounce = event.hardBounce)
49       resultUpdateError  <- registerMailingError(email = event.email, maybeError = event.error, date = date)
50     } yield resultUpdateBounce && resultUpdateError
51   }
52 
53   def handleSpamEvent(event: MailJetSpamEvent): Future[Boolean] = {
54     userService.updateOptInNewsletter(email = event.email, optInNewsletter = false)
55   }
56 
57   def handleUnsubscribeEvent(event: MailJetUnsubscribeEvent): Future[Boolean] = {
58     userService.updateOptInNewsletter(email = event.email, optInNewsletter = false)
59   }
60 
61   private def registerMailingError(
62     email: String,
63     maybeError: Option[MailJetError],
64     date: ZonedDateTime
65   ): Future[Boolean] = {
66     maybeError match {
67       case None => Future.successful(false)
68       case Some(error) =>
69         userService.updateLastMailingError(
70           email = email,
71           lastMailingError = Some(MailingErrorLog(error = error.name, date = date))
72         )
73     }
74   }
75 }
76 
77 object MailJetEventConsumerBehavior {
78   def apply(userService: UserService): Behavior[KafkaConsumerBehavior.Protocol] =
79     new MailJetEventConsumerBehavior(userService).createBehavior(name)
80   val name: String = "mailJet-event-consumer"
81 }