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 java.nio.charset.StandardCharsets
23 import java.nio.file.{Files, Path}
24 import java.time.format.DateTimeFormatter
25 import java.time.temporal.ChronoUnit
26 import java.time.{LocalDate, ZoneOffset, ZonedDateTime}
27 import java.util.concurrent.atomic.AtomicInteger
28 import java.util.concurrent.{Executors, ThreadFactory}
29 import java.util.stream.Collectors
30 import cats.implicits._
31 import akka.actor.typed.scaladsl.AskPattern._
32 import akka.http.scaladsl.model.StatusCodes
33 import akka.persistence.query.EventEnvelope
34 import akka.stream._
35 import akka.stream.alpakka.file.scaladsl.LogRotatorSink
36 import akka.stream.scaladsl.{Sink, Source}
37 import akka.util.{ByteString, Timeout}
38 import akka.{Done, NotUsed}
39 import grizzled.slf4j.Logging
40 import de.heikoseeberger.akkahttpcirce.ErrorAccumulatingCirceSupport
41 import eu.timepit.refined.auto._
42 import org.make.api.extensions.MailJetConfigurationComponent
43 import org.make.api.operation.OperationServiceComponent
44 import org.make.api.proposal.ProposalServiceComponent
45 import org.make.api.question.{QuestionServiceComponent, SearchQuestionRequest}
46 import org.make.api.technical.ExecutorServiceHelper._
47 import org.make.api.technical.job.JobActor.Protocol.Response.JobAcceptance
48 import org.make.api.technical.job.JobCoordinatorServiceComponent
49 import org.make.api.technical._
50 import org.make.api.technical.crm.CrmClient.{Account, Marketing, Transactional}
51 import org.make.api.user.PersistentCrmSynchroUserService.CrmSynchroUser
52 import org.make.api.user.PersistentCrmSynchroUserServiceComponent
53 import org.make.api.userhistory._
54 import org.make.core.DateHelper.isLast30daysDate
55 import org.make.core.technical.Pagination
56 import org.make.core.job.Job.JobId.SyncCrmData
57 import org.make.core.question.Question
58 import org.make.core.session.SessionId
59 import org.make.core.user.{HasUserType, UserId}
60 import org.make.core.user.UserType._
61 import org.make.core.{DateHelper, Order}
62 
63 import java.util.UUID
64 import scala.concurrent.duration.DurationInt
65 import scala.concurrent.{ExecutionContext, Future, Promise}
66 import scala.jdk.CollectionConverters._
67 import scala.math.Ordering.Implicits.infixOrderingOps
68 import scala.util.{Failure, Success, Try}
69 
70 trait DefaultCrmServiceComponent extends CrmServiceComponent with Logging with ErrorAccumulatingCirceSupport {
71   self: MailJetConfigurationComponent
72     with ActorSystemComponent
73     with PersistentCrmSynchroUserServiceComponent
74     with OperationServiceComponent
75     with QuestionServiceComponent
76     with UserHistoryCoordinatorServiceComponent
77     with ReadJournalComponent
78     with ProposalServiceComponent
79     with PersistentCrmUserServiceComponent
80     with CrmClientComponent
81     with EventBusServiceComponent
82     with ErrorAccumulatingCirceSupport
83     with JobCoordinatorServiceComponent
84     with SpawnActorServiceComponent =>
85 
86   override lazy val crmService: DefaultCrmService = new DefaultCrmService
87 
88   class DefaultCrmService extends CrmService {
89 
90     private lazy val batchSize: Int = mailJetConfiguration.userListBatchSize
91     private lazy val baseCsvDirectory: String = mailJetConfiguration.csvDirectory
92     private lazy val csvSize: Int = mailJetConfiguration.csvSize
93     private val retrievePropertiesParallelism = 10
94     private val persistCrmUsersParallelism = 5
95 
96     private val poolSize: Int = 10
97     implicit private val executionContext: ExecutionContext =
98       Executors
99         .newFixedThreadPool(
100           poolSize,
101           new ThreadFactory {
102             val counter = new AtomicInteger()
103 
104             override def newThread(runnable: Runnable): Thread =
105               new Thread(runnable, "crm-batchs-" + counter.getAndIncrement())
106           }
107         )
108         .instrument("crm-batchs")
109         .toExecutionContext
110 
111     implicit val crmSynchroUserUserType: HasUserType[CrmSynchroUser] = _.userType
112 
113     private val localDateFormatter: DateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd 00:00:00")
114     private val dateFormatter: DateTimeFormatter =
115       DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss").withZone(ZoneOffset.UTC)
116     private val dayDateFormatter: DateTimeFormatter = DateTimeFormatter.ofPattern("ddMMyyyy").withZone(ZoneOffset.UTC)
117 
118     private implicit val timeout: Timeout = TimeSettings.defaultTimeout
119 
120     override def sendEmail(messages: SendMessages): Future[SendEmailResponse] = {
121       val result = crmClient.sendEmail(message = messages)
122 
123       result.onComplete {
124         case Success(response) =>
125           logger.info(s"Sent email ${messages.toString} with reponse ${response.toString}")
126         case Failure(e) =>
127           logger.error(s"Sent email $messages failed", e)
128           e match {
129             case CrmClientException.RequestException.SendEmailException(StatusCodes.BadRequest, _) =>
130             case _ =>
131               actorSystem.scheduler
132                 .scheduleOnce(mailJetConfiguration.delayBeforeResend, () => eventBusService.publish(messages))
133           }
134       }
135 
136       result
137     }
138 
139     override def getUsersMailFromList(
140       listId: Option[String] = None,
141       sort: Option[String] = None,
142       order: Option[Order] = None,
143       countOnly: Option[Boolean] = None,
144       limit: Pagination.Limit,
145       offset: Pagination.Offset = Pagination.Offset(0)
146     ): Future[GetUsersMail] = {
147       crmClient
148         .getUsersInformationMailFromList(listId, sort, order, countOnly, limit, offset)
149         .map { response =>
150           GetUsersMail(
151             count = response.count,
152             total = response.total,
153             data = response.data.map(data => ContactMail(contactId = data.id, email = data.email))
154           )
155         }
156         .recoverWith {
157           case e: CrmClientException =>
158             logger.error(e.message)
159             Future.successful(GetUsersMail(0, 0, Seq.empty))
160           case e => Future.failed(e)
161         }
162     }
163 
164     override def deleteRecipient(email: String): Future[Unit] =
165       for {
166         _ <- crmClient.deleteContactByEmail(email, Marketing)
167         _ <- crmClient.deleteContactByEmail(email, Transactional)
168       } yield ()
169 
170     override def createCrmUsers(): Future[Unit] = {
171       for {
172         _        <- persistentCrmUserService.truncateCrmUsers()
173         resolver <- createQuestionResolver
174         _        <- computeAndPersistCrmUsers(resolver)
175       } yield ()
176     }
177 
178     private def resetDirectory(directory: Path): Unit = {
179       if (Files.exists(directory)) {
180         Files.list(directory).forEach { file =>
181           Files.deleteIfExists(file)
182           ()
183         }
184       } else {
185         Files.createDirectories(directory)
186       }
187       ()
188     }
189 
190     def initializeDirectories(): Future[Unit] = {
191       Future[Unit] {
192         resetDirectory(CrmList.HardBounce.targetDirectory(baseCsvDirectory))
193         resetDirectory(CrmList.OptOut.targetDirectory(baseCsvDirectory))
194         resetDirectory(CrmList.OptIn.targetDirectory(baseCsvDirectory))
195       }
196     }
197 
198     override def synchronizeContactsWithCrm(): Future[JobAcceptance] = {
199       val startTime: Long = System.currentTimeMillis()
200       val synchronizationTime = DateHelper.now().format(dateFormatter)
201 
202       jobCoordinatorService.start(SyncCrmData) { report =>
203         val crmSynchronization =
204           for {
205             _ <- createCrmUsers()
206             _ <- report(49d)
207             _ <- initializeDirectories()
208             _ <- report(50d)
209             _ <- synchronizeList(
210               formattedDate = synchronizationTime,
211               list = CrmList.OptIn,
212               csvDirectory = CrmList.OptIn.targetDirectory(baseCsvDirectory)
213             )
214             _ <- report(65d)
215             _ <- synchronizeList(
216               formattedDate = synchronizationTime,
217               list = CrmList.OptOut,
218               csvDirectory = CrmList.OptOut.targetDirectory(baseCsvDirectory)
219             )
220             _ <- report(80d)
221             _ <- synchronizeList(
222               formattedDate = synchronizationTime,
223               list = CrmList.HardBounce,
224               csvDirectory = CrmList.HardBounce.targetDirectory(baseCsvDirectory)
225             )
226             _ <- report(95d)
227             _ <- anonymize()
228           } yield {}
229 
230         crmSynchronization.onComplete {
231           case Failure(exception) =>
232             logger.error(s"Mailjet synchro failed:", exception)
233           case Success(_) =>
234             logger.info(s"Mailjet synchro succeeded in ${System.currentTimeMillis() - startTime}ms")
235         }
236 
237         crmSynchronization
238       }
239 
240     }
241 
242     private def fileSizeTriggerCreator(csvDirectory: Path): () => ByteString => Option[Path] = () => {
243       val max = csvSize
244       var size: Long = max
245       element: ByteString =>
246         if (size + element.size > max) {
247           val path = Files.createFile(csvDirectory.resolve(s"${UUID.randomUUID().toString}.csv"))
248           size = element.size
249           Some(path)
250         } else {
251           size += element.size
252           None
253         }
254     }
255 
256     def createCsv(formattedDate: String, list: CrmList, csvDirectory: Path): Future[Seq[Path]] = {
257 
258       def toContactProperties(user: PersistentCrmUser): ContactProperties = {
259         ContactProperties(
260           userId = Some(UserId(user.userId)),
261           firstName = Some(user.firstname),
262           postalCode = user.zipcode,
263           dateOfBirth = user.dateOfBirth,
264           emailValidationStatus = Some(user.emailValidationStatus),
265           emailHardBounceValue = Some(user.emailHardbounceStatus),
266           unsubscribeStatus = Some(user.unsubscribeStatus),
267           accountCreationDate = user.accountCreationDate,
268           accountCreationSource = user.accountCreationSource,
269           accountCreationOrigin = user.accountCreationOrigin,
270           accountCreationSlug = user.accountCreationOperation,
271           accountCreationLocation = user.accountCreationLocation,
272           favoriteCountry = user.favoriteCountry,
273           favoriteLanguage = user.favoriteLanguage,
274           totalProposals = user.totalNumberProposals,
275           totalVotes = user.totalNumberVotes,
276           firstContributionDate = user.firstContributionDate,
277           lastContributionDate = user.lastContributionDate,
278           operationActivity = user.operationActivity,
279           sourceActivity = user.sourceActivity,
280           daysOfActivity = user.daysOfActivity,
281           daysOfActivity30 = user.daysOfActivity30d,
282           userType = user.userType,
283           accountType = user.accountType,
284           updatedAt = Some(formattedDate),
285           daysBeforeDeletion = user.daysBeforeDeletion,
286           lastActivityDate = user.lastActivityDate,
287           sessionsCount = user.sessionsCount,
288           eventsCount = user.eventsCount
289         )
290       }
291 
292       StreamUtils
293         .asyncPageToPageSource(persistentCrmUserService.list(list.unsubscribed, list.hardBounced, _, batchSize))
294         .mapConcat(identity)
295         .map { crmUser =>
296           Contact(email = crmUser.email, name = Some(crmUser.fullName), properties = Some(toContactProperties(crmUser))).toStringCsv
297         }
298         .map(ByteString.apply(_, StandardCharsets.UTF_8))
299         .runWith(LogRotatorSink(fileSizeTriggerCreator(csvDirectory)))
300         .map(_ => Files.list(csvDirectory).collect(Collectors.toList[Path]).asScala.toSeq)
301     }
302 
303     @SuppressWarnings(Array("org.wartremover.warts.IterableOps"))
304     private def sendCsvToHardBounceList(csv: Path, list: CrmList): Future[Long] = {
305       for {
306         csvId <- crmClient.sendCsv(mailJetConfiguration.hardBounceListId, csv)
307         response <- crmClient.manageContactListWithCsv(
308           CsvImport(
309             mailJetConfiguration.hardBounceListId,
310             csvId.csvId.toString,
311             list.actionOnHardBounce,
312             ImportOptions("yyyy-mm-dd hh:nn:ss").toString
313           )
314         )
315       } yield {
316         response.data.head.jobId
317       }
318     }
319 
320     @SuppressWarnings(Array("org.wartremover.warts.IterableOps"))
321     private def sendCsvToOptInList(csv: Path, list: CrmList): Future[Long] = {
322       for {
323         csvId <- crmClient.sendCsv(mailJetConfiguration.optInListId, csv)
324         response <- crmClient.manageContactListWithCsv(
325           CsvImport(
326             mailJetConfiguration.optInListId,
327             csvId.csvId.toString,
328             list.actionOnOptIn,
329             ImportOptions("yyyy-mm-dd hh:nn:ss").toString
330           )
331         )
332       } yield {
333         response.data.head.jobId
334       }
335     }
336 
337     @SuppressWarnings(Array("org.wartremover.warts.IterableOps"))
338     private def sendCsvToUnsubscribeList(csv: Path, list: CrmList): Future[Long] = {
339       for {
340         csvId <- crmClient.sendCsv(mailJetConfiguration.unsubscribeListId, csv)
341         response <- crmClient.manageContactListWithCsv(
342           CsvImport(
343             mailJetConfiguration.unsubscribeListId,
344             csvId.csvId.toString,
345             list.actionOnOptOut,
346             ImportOptions("yyyy-mm-dd hh:nn:ss").toString
347           )
348         )
349       } yield {
350         response.data.head.jobId
351       }
352     }
353 
354     override def synchronizeList(formattedDate: String, list: CrmList, csvDirectory: Path): Future[Done] = {
355       Source
356         .future(createCsv(formattedDate, list, csvDirectory))
357         .mapConcat(identity)
358         .filter(Files.size(_) > 0)
359         .mapAsync(1) { csv =>
360           for {
361             responseHardBounce  <- sendCsvToHardBounceList(csv, list)
362             responseOptIn       <- sendCsvToOptInList(csv, list)
363             responseUnsubscribe <- sendCsvToUnsubscribeList(csv, list)
364             result              <- verifyJobCompletion(responseHardBounce, responseOptIn, responseUnsubscribe)
365           } yield result
366         }
367         .runWith(Sink.ignore)
368     }
369 
370     private def verifyJobCompletion(
371       responseHardBounce: Long,
372       responseOptIn: Long,
373       responseUnsubscribe: Long
374     ): Future[Unit] = {
375       val jobIds = Seq(responseHardBounce, responseOptIn, responseUnsubscribe)
376       val promise = Promise[Unit]()
377       spawnActorService
378         .spawn(
379           behavior = CrmSynchroCsvMonitor(crmClient, promise, mailJetConfiguration.tickInterval)(jobIds),
380           name = "CrmSynchroCsvMonitor"
381         )
382         .productR(promise.future)
383     }
384 
385     private def computeAndPersistCrmUsers(questionResolver: QuestionResolver): Future[Unit] = {
386       val start = System.currentTimeMillis()
387       StreamUtils
388         .asyncPageToPageSource(
389           page =>
390             persistentCrmSynchroUserService
391               .findUsersForCrmSynchro(None, None, Pagination.Offset(page), Pagination.Limit(batchSize))
392         )
393         .mapConcat(identity)
394         .mapAsync(retrievePropertiesParallelism) { user =>
395           getPropertiesFromUser(user, questionResolver).map { properties =>
396             (user.email, user.fullName.getOrElse(user.email), properties)
397           }
398         }
399         .groupedWithin(batchSize, 5.seconds)
400         .map { contacts =>
401           contacts.map {
402             case (email, fullName, properties) => properties.toPersistentCrmUser(email, fullName)
403           }
404         }
405         .mapAsync(persistCrmUsersParallelism) { crmUsers =>
406           persistentCrmUserService.persist(crmUsers)
407         }
408         .runWith(Sink.ignore)
409         .map(_ => logger.info(s"Crm users creation completed in ${System.currentTimeMillis() - start} ms"))
410     }
411 
412     private final def deleteAnonymizedContacts(contactIds: Seq[Long], account: Account): Future[Done] = {
413       Source(contactIds)
414         .throttle(1, 1.second)
415         .mapAsync(1) { contactId =>
416           crmClient.deleteContactById(contactId.toString, account)
417         }
418         .runWith(Sink.ignore)
419     }
420 
421     override def anonymize(): Future[Unit] = {
422       for {
423         foundAccounts <- findInactiveAccounts
424         _             <- deleteAnonymizedContacts(foundAccounts, Marketing)
425         _             <- deleteAnonymizedContacts(foundAccounts, Transactional)
426       } yield ()
427     }
428 
429     private def findInactiveAccounts: Future[Seq[Long]] = {
430       val checkDate: ZonedDateTime = ZonedDateTime.now(ZoneOffset.UTC).minusHours(24)
431       def isBefore(updatedAt: String): Boolean =
432         Try(ZonedDateTime.parse(updatedAt)).toOption.forall(_.isBefore(checkDate))
433 
434       StreamUtils
435         .asyncPageToPageSource(crmClient.getContactsProperties(_, Pagination.Limit(batchSize)).map(_.data))
436         .mapConcat(identity)
437         .filter { contact =>
438           contact.properties.find(_.name == "updated_at").exists(updatedAt => isBefore(updatedAt.value))
439         }
440         .map(_.contactId)
441         .runWith(Sink.seq)
442         .map { accounts =>
443           logger.info(s"Synchro CRM : Found ${accounts.size} to anonymize")
444           accounts
445         }
446     }
447 
448     private def createQuestionResolver: Future[QuestionResolver] = {
449       val operationsAsMap = operationService
450         .findSimple()
451         .map(_.map(operation => operation.slug -> operation.operationId).toMap)
452       for {
453         questions  <- questionService.searchQuestion(SearchQuestionRequest())
454         operations <- operationsAsMap
455       } yield new QuestionResolver(questions, operations)
456     }
457 
458     final def getPropertiesFromUser(user: CrmSynchroUser, resolver: QuestionResolver): Future[ContactProperties] = {
459 
460       val decider: Supervision.Decider = { e =>
461         logger.error(
462           s"Error in stream getPropertiesFromUser for user ${user.uuid.value}. Stream resumed by dropping this element: ",
463           e
464         )
465         Supervision.Resume
466       }
467 
468       val events: Source[EventEnvelope, NotUsed] =
469         userJournal
470           .currentEventsByPersistenceId(
471             persistenceId = user.uuid.value,
472             fromSequenceNr = 0,
473             toSequenceNr = Long.MaxValue
474           )
475           .withAttributes(ActorAttributes.supervisionStrategy(decider))
476 
477       val initialProperties = userPropertiesFromUser(user, resolver)
478 
479       val userProperties: Future[UserProperties] = events
480         .runFoldAsync(initialProperties) { (accumulator: UserProperties, envelope: EventEnvelope) =>
481           accumulateEvent(accumulator, envelope, resolver)
482         }
483 
484       userProperties.map(properties => contactPropertiesFromUserProperties(properties.normalize()))
485     }
486 
487     private def userPropertiesFromUser(user: CrmSynchroUser, questionResolver: QuestionResolver): UserProperties = {
488       val question =
489         questionResolver.findQuestionWithOperation { question =>
490           user.registerQuestionId.contains(question.questionId)
491         }
492 
493       UserProperties(
494         userId = user.uuid,
495         firstname = user.firstName.getOrElse(""),
496         zipCode = user.postalCode,
497         dateOfBirth = user.dateOfBirth,
498         emailValidationStatus = user.emailVerified,
499         emailHardBounceStatus = user.isHardBounce,
500         unsubscribeStatus = !user.optInNewsletter,
501         accountCreationDate = user.createdAt,
502         userB2B = user.isB2B,
503         updatedAt = Some(DateHelper.now()),
504         crmCountry = Some(user.crmCountry.value),
505         crmLanguage = Some(user.crmLanguage.value),
506         accountCreationSlug = question.map(_.slug),
507         accountType = Some(user.userType.value),
508         lastActivityDate = user.lastConnection,
509         sessionsIds = Set.empty,
510         eventsCount = 0
511       )
512     }
513 
514     private def getDaysBeforeDeletionFromLastActivityDate(properties: UserProperties): Option[Int] = {
515       properties.lastActivityDate.map { date =>
516         val deletionDate = date.plusYears(if (properties.userB2B) 4 else 2).plusMonths(11)
517         ChronoUnit.DAYS.between(ZonedDateTime.now(), deletionDate).toInt
518       }
519     }
520 
521     private def contactPropertiesFromUserProperties(userProperty: UserProperties): ContactProperties = {
522       ContactProperties(
523         userId = Some(userProperty.userId),
524         firstName = Some(userProperty.firstname),
525         postalCode = userProperty.zipCode,
526         dateOfBirth = userProperty.dateOfBirth.map(_.format(localDateFormatter)),
527         emailValidationStatus = Some(userProperty.emailValidationStatus),
528         emailHardBounceValue = Some(userProperty.emailHardBounceStatus),
529         unsubscribeStatus = Some(userProperty.unsubscribeStatus),
530         accountCreationDate = Some(userProperty.accountCreationDate.format(dateFormatter)),
531         accountCreationSource = userProperty.accountCreationSource,
532         accountCreationOrigin = userProperty.accountCreationOrigin,
533         accountCreationSlug = userProperty.accountCreationSlug,
534         accountCreationLocation = userProperty.accountCreationLocation,
535         favoriteCountry = userProperty.crmCountry,
536         favoriteLanguage = userProperty.crmLanguage,
537         totalProposals = Some(userProperty.totalNumberProposals.getOrElse(0)),
538         totalVotes = Some(userProperty.totalNumbervotes.getOrElse(0)),
539         firstContributionDate = userProperty.firstContributionDate.map(_.format(dateFormatter)),
540         lastContributionDate = userProperty.lastContributionDate.map(_.format(dateFormatter)),
541         operationActivity = Some(userProperty.questionActivity.distinct.mkString(",")),
542         sourceActivity = Some(userProperty.sourceActivity.distinct.mkString(",")),
543         daysOfActivity = Some(userProperty.daysOfActivity.distinct.length),
544         daysOfActivity30 = Some(userProperty.daysOfActivity30d.distinct.length),
545         userType = if (userProperty.userB2B) {
546           Some("B2B")
547         } else {
548           Some("B2C")
549         },
550         accountType = userProperty.accountType,
551         updatedAt = userProperty.updatedAt.map(_.format(dateFormatter)),
552         daysBeforeDeletion = getDaysBeforeDeletionFromLastActivityDate(userProperty),
553         lastActivityDate = userProperty.lastActivityDate.map(_.format(dateFormatter)),
554         sessionsCount = Some(userProperty.sessionsIds.size),
555         eventsCount = Some(userProperty.eventsCount)
556       )
557     }
558 
559     private def accumulateEvent(
560       accumulator: UserProperties,
561       envelope: EventEnvelope,
562       questionResolver: QuestionResolver
563     ): Future[UserProperties] = {
564       envelope.event match {
565         case event: LogUserConnectedEvent =>
566           Future.successful(accumulateLogUserConnectedEvent(accumulator, event))
567         case event: LogRegisterCitizenEvent =>
568           Future.successful(accumulateLogRegisterCitizenEvent(accumulator, event, questionResolver))
569         case event: LogUserProposalEvent =>
570           accumulateLogUserProposalEvent(accumulator, event, questionResolver)
571         case event: LogUserVoteEvent =>
572           accumulateLogUserVoteEvent(accumulator, event, questionResolver)
573         case event: LogUserUnvoteEvent =>
574           accumulateLogUserUnvoteEvent(accumulator, event, questionResolver)
575         case event: LogUserQualificationEvent =>
576           accumulateLogUserQualificationEvent(accumulator, event, questionResolver)
577         case event: LogUserUnqualificationEvent =>
578           accumulateLogUserUnqualificationEvent(accumulator, event, questionResolver)
579         case event: LogUserStartSequenceEvent =>
580           Future.successful(accumulateLogUserStartSequenceEvent(accumulator, event))
581         case _ => Future.successful(accumulator)
582       }
583     }
584 
585     private def accumulateLogUserConnectedEvent(
586       accumulator: UserProperties,
587       event: LogUserConnectedEvent
588     ): UserProperties = {
589 
590       accumulator.copy(
591         lastActivityDate = accumulator.lastActivityDate.max(Some(event.action.date)),
592         sessionsIds = accumulator.sessionsIds ++ Set(event.requestContext.sessionId),
593         eventsCount = accumulator.eventsCount + 1
594       )
595     }
596 
597     private def accumulateLogUserUnqualificationEvent(
598       accumulator: UserProperties,
599       event: LogUserUnqualificationEvent,
600       questionResolver: QuestionResolver
601     ): Future[UserProperties] = {
602       val futureQuestion: Future[Option[Question]] =
603         proposalService.resolveQuestionFromVoteEvent(
604           questionResolver,
605           event.requestContext,
606           event.action.arguments.proposalId
607         )
608 
609       futureQuestion.map { maybeQuestion =>
610         accumulator.copy(
611           lastContributionDate = Some(event.action.date),
612           questionActivity = accumulator.questionActivity ++ maybeQuestion.map(_.slug).toList,
613           sourceActivity = accumulator.sourceActivity ++ event.requestContext.source,
614           daysOfActivity = accumulator.daysOfActivity ++ Some(event.action.date.format(dayDateFormatter)),
615           daysOfActivity30d = if (isLast30daysDate(event.action.date)) {
616             accumulator.daysOfActivity30d ++ Some(event.action.date.format(dayDateFormatter))
617           } else {
618             accumulator.daysOfActivity
619           },
620           lastActivityDate = accumulator.lastActivityDate.max(Some(event.action.date)),
621           sessionsIds = accumulator.sessionsIds ++ Set(event.requestContext.sessionId),
622           eventsCount = accumulator.eventsCount + 1
623         )
624       }
625     }
626 
627     private def accumulateLogUserQualificationEvent(
628       accumulator: UserProperties,
629       event: LogUserQualificationEvent,
630       questionResolver: QuestionResolver
631     ): Future[UserProperties] = {
632 
633       val futureQuestion: Future[Option[Question]] =
634         proposalService.resolveQuestionFromVoteEvent(
635           questionResolver,
636           event.requestContext,
637           event.action.arguments.proposalId
638         )
639 
640       futureQuestion.map { maybeQuestion =>
641         accumulator.copy(
642           lastContributionDate = Some(event.action.date),
643           questionActivity = accumulator.questionActivity ++ maybeQuestion.map(_.slug).toList,
644           sourceActivity = accumulator.sourceActivity ++ event.requestContext.source,
645           daysOfActivity = accumulator.daysOfActivity ++ Some(event.action.date.format(dayDateFormatter)),
646           daysOfActivity30d = if (isLast30daysDate(event.action.date)) {
647             accumulator.daysOfActivity30d ++ Some(event.action.date.format(dayDateFormatter))
648           } else {
649             accumulator.daysOfActivity30d
650           },
651           lastActivityDate = accumulator.lastActivityDate.max(Some(event.action.date)),
652           sessionsIds = accumulator.sessionsIds ++ Set(event.requestContext.sessionId),
653           eventsCount = accumulator.eventsCount + 1
654         )
655       }
656     }
657 
658     private def accumulateLogUserUnvoteEvent(
659       accumulator: UserProperties,
660       event: LogUserUnvoteEvent,
661       questionResolver: QuestionResolver
662     ): Future[UserProperties] = {
663       val futureQuestion: Future[Option[Question]] =
664         proposalService.resolveQuestionFromVoteEvent(
665           questionResolver,
666           event.requestContext,
667           event.action.arguments.proposalId
668         )
669 
670       futureQuestion.map { maybeQuestion =>
671         accumulator.copy(
672           totalNumbervotes = accumulator.totalNumbervotes.map(_ - 1).orElse(Some(-1)),
673           lastContributionDate = Some(event.action.date),
674           questionActivity = accumulator.questionActivity ++ maybeQuestion.map(_.slug).toList,
675           sourceActivity = accumulator.sourceActivity ++ event.requestContext.source,
676           daysOfActivity = accumulator.daysOfActivity ++ Some(event.action.date.format(dayDateFormatter)),
677           daysOfActivity30d = if (isLast30daysDate(event.action.date)) {
678             accumulator.daysOfActivity30d ++ Some(event.action.date.format(dayDateFormatter))
679           } else {
680             accumulator.daysOfActivity30d
681           },
682           lastActivityDate = accumulator.lastActivityDate.max(Some(event.action.date)),
683           sessionsIds = accumulator.sessionsIds ++ Set(event.requestContext.sessionId),
684           eventsCount = accumulator.eventsCount + 1
685         )
686       }
687     }
688 
689     private def accumulateLogUserVoteEvent(
690       accumulator: UserProperties,
691       event: LogUserVoteEvent,
692       questionResolver: QuestionResolver
693     ): Future[UserProperties] = {
694       val futureQuestion: Future[Option[Question]] =
695         proposalService.resolveQuestionFromVoteEvent(
696           questionResolver,
697           event.requestContext,
698           event.action.arguments.proposalId
699         )
700 
701       futureQuestion.map { maybeQuestion =>
702         accumulator.copy(
703           totalNumbervotes = accumulator.totalNumbervotes.map(_ + 1).orElse(Some(1)),
704           questionActivity = accumulator.questionActivity ++ maybeQuestion.map(_.slug).toList,
705           sourceActivity = accumulator.sourceActivity ++ event.requestContext.source,
706           firstContributionDate = accumulator.firstContributionDate.orElse(Option(event.action.date)),
707           lastContributionDate = Some(event.action.date),
708           daysOfActivity = accumulator.daysOfActivity ++ Some(event.action.date.format(dayDateFormatter)),
709           daysOfActivity30d = if (isLast30daysDate(event.action.date)) {
710             accumulator.daysOfActivity30d ++ Some(event.action.date.format(dayDateFormatter))
711           } else {
712             accumulator.daysOfActivity30d
713           },
714           lastActivityDate = accumulator.lastActivityDate.max(Some(event.action.date)),
715           sessionsIds = accumulator.sessionsIds ++ Set(event.requestContext.sessionId),
716           eventsCount = accumulator.eventsCount + 1
717         )
718       }
719     }
720 
721     private def accumulateLogUserProposalEvent(
722       accumulator: UserProperties,
723       event: LogUserProposalEvent,
724       questionResolver: QuestionResolver
725     ): Future[UserProperties] = {
726       val futureMaybeQuestion: Future[Option[Question]] =
727         proposalService.resolveQuestionFromUserProposal(
728           questionResolver,
729           event.requestContext,
730           accumulator.userId,
731           event.action.date
732         )
733 
734       futureMaybeQuestion.map { maybeQuestion =>
735         accumulator.copy(
736           totalNumberProposals = accumulator.totalNumberProposals.map(_ + 1).orElse(Some(1)),
737           questionActivity = accumulator.questionActivity ++ maybeQuestion.map(_.slug).toList,
738           sourceActivity = accumulator.sourceActivity ++ event.requestContext.source.toList,
739           firstContributionDate = accumulator.firstContributionDate.orElse(Option(event.action.date)),
740           lastContributionDate = Some(event.action.date),
741           daysOfActivity = accumulator.daysOfActivity ++ Some(event.action.date.format(dayDateFormatter)),
742           daysOfActivity30d = if (isLast30daysDate(event.action.date)) {
743             accumulator.daysOfActivity30d ++ Some(event.action.date.format(dayDateFormatter))
744           } else {
745             accumulator.daysOfActivity30d
746           },
747           lastActivityDate = accumulator.lastActivityDate.max(Some(event.action.date)),
748           sessionsIds = accumulator.sessionsIds ++ Set(event.requestContext.sessionId),
749           eventsCount = accumulator.eventsCount + 1
750         )
751       }
752     }
753 
754     private def accumulateLogRegisterCitizenEvent(
755       accumulator: UserProperties,
756       event: LogRegisterCitizenEvent,
757       questionResolver: QuestionResolver
758     ): UserProperties = {
759 
760       val maybeQuestion = questionResolver
761         .extractQuestionWithOperationFromRequestContext(event.requestContext)
762 
763       accumulator.copy(
764         accountCreationSource = event.requestContext.source,
765         accountCreationOrigin = event.requestContext.getParameters.map { parameters =>
766           parameters.getOrElse("utm_source", "unknown")
767         },
768         accountCreationSlug = accumulator.accountCreationSlug.orElse(maybeQuestion.map(_.slug)),
769         accountCreationLocation = event.requestContext.location,
770         questionActivity = accumulator.questionActivity ++ maybeQuestion.map(_.slug).toList,
771         sourceActivity = accumulator.sourceActivity ++ event.requestContext.source,
772         lastActivityDate = accumulator.lastActivityDate.max(Some(event.action.date)),
773         sessionsIds = accumulator.sessionsIds ++ Set(event.requestContext.sessionId),
774         eventsCount = accumulator.eventsCount + 1
775       )
776     }
777 
778     private def accumulateLogUserStartSequenceEvent(
779       accumulator: UserProperties,
780       event: LogUserStartSequenceEvent
781     ): UserProperties = {
782       accumulator.copy(
783         lastActivityDate = accumulator.lastActivityDate.max(Some(event.action.date)),
784         sessionsIds = accumulator.sessionsIds ++ Set(event.requestContext.sessionId),
785         eventsCount = accumulator.eventsCount + 1
786       )
787     }
788   }
789 }
790 
791 final case class UserProperties(
792   userId: UserId,
793   firstname: String,
794   zipCode: Option[String],
795   dateOfBirth: Option[LocalDate],
796   emailValidationStatus: Boolean,
797   emailHardBounceStatus: Boolean,
798   unsubscribeStatus: Boolean,
799   accountCreationDate: ZonedDateTime,
800   userB2B: Boolean,
801   accountCreationSource: Option[String] = None,
802   accountCreationOrigin: Option[String] = None,
803   accountCreationSlug: Option[String] = None,
804   accountCreationLocation: Option[String] = None,
805   crmCountry: Option[String] = None,
806   crmLanguage: Option[String] = None,
807   totalNumberProposals: Option[Int] = None,
808   totalNumbervotes: Option[Int] = None,
809   firstContributionDate: Option[ZonedDateTime] = None,
810   lastContributionDate: Option[ZonedDateTime] = None,
811   questionActivity: Seq[String] = Seq.empty,
812   sourceActivity: Seq[String] = Seq.empty,
813   daysOfActivity: Seq[String] = Seq.empty,
814   daysOfActivity30d: Seq[String] = Seq.empty,
815   accountType: Option[String] = None,
816   updatedAt: Option[ZonedDateTime],
817   lastActivityDate: Option[ZonedDateTime],
818   sessionsIds: Set[SessionId],
819   eventsCount: Int
820 ) {
821 
822   def normalize(): UserProperties = {
823     normalizeUserPropertiesWhenNoRegisterEvent()
824   }
825 
826   /*
827    * Fix properties for user with no register event (previous bug that has been resolved)
828    */
829   private def normalizeUserPropertiesWhenNoRegisterEvent(): UserProperties = {
830     val sourceFixDate: ZonedDateTime = ZonedDateTime.parse("2018-09-01T00:00:00Z")
831     (accountCreationSource, accountCreationDate) match {
832       case (None, date) if date.isBefore(sourceFixDate) =>
833         this.copy(accountCreationSource = Some("core"), sourceActivity = (sourceActivity ++ Some("core")).distinct)
834       case _ => this
835     }
836   }
837 }