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.http.scaladsl.model.ContentType
23 import akka.stream.scaladsl.Sink
24 import cats.implicits._
25 import com.github.t3hnar.bcrypt._
26 import grizzled.slf4j.Logging
27 import org.make.api.extensions.{MailJetConfigurationComponent, MakeSettingsComponent}
28 import org.make.api.partner.PersistentPartnerServiceComponent
29 import org.make.api.proposal.ProposalServiceComponent
30 import org.make.api.proposal.PublishedProposalEvent.ReindexProposal
31 import org.make.api.question.AuthorRequest
32 import org.make.api.technical.Futures.FutureOfOption
33 import org.make.api.technical._
34 import org.make.api.technical.auth.{
35   MakeDataHandlerComponent,
36   TokenGeneratorComponent,
37   TokenResponse,
38   UserTokenGeneratorComponent
39 }
40 import org.make.api.technical.crm.{CrmServiceComponent, PersistentCrmUserServiceComponent}
41 import org.make.api.technical.job.JobActor.Protocol.Response.JobAcceptance
42 import org.make.api.technical.job.JobCoordinatorServiceComponent
43 import org.make.api.technical.security.SecurityHelper
44 import org.make.api.technical.storage.Content.FileContent
45 import org.make.api.technical.storage.StorageServiceComponent
46 import org.make.api.user.UserExceptions.{EmailAlreadyRegisteredException, EmailNotAllowed}
47 import org.make.api.user.social.models.UserInfo
48 import org.make.api.user.validation.UserRegistrationValidatorComponent
49 import org.make.api.userhistory._
50 import org.make.core._
51 import org.make.core.auth.UserRights
52 import org.make.core.job.Job.JobId.AnonymizeInactiveUsers
53 import org.make.core.profile.Gender.{Female, Male, Other}
54 import org.make.core.profile.{Gender, Profile}
55 import org.make.core.proposal._
56 import org.make.core.question.QuestionId
57 import org.make.core.reference.{Country, Language}
58 import org.make.core.technical.Pagination
59 import org.make.core.user.Role.RoleCitizen
60 import org.make.core.user._
61 import scalaoauth2.provider.AuthInfo
62 
63 import java.io.File
64 import java.nio.file.Files
65 import java.time.{LocalDate, ZonedDateTime}
66 import scala.concurrent.ExecutionContext.Implicits.global
67 import scala.concurrent.Future
68 import scala.util.{Failure, Success}
69 
70 trait DefaultUserServiceComponent extends UserServiceComponent with ShortenedNames with Logging {
71   this: ActorSystemComponent
72     with IdGeneratorComponent
73     with MakeDataHandlerComponent
74     with UserTokenGeneratorComponent
75     with PersistentUserServiceComponent
76     with PersistentCrmUserServiceComponent
77     with ProposalServiceComponent
78     with CrmServiceComponent
79     with EventBusServiceComponent
80     with TokenGeneratorComponent
81     with MakeSettingsComponent
82     with UserRegistrationValidatorComponent
83     with StorageServiceComponent
84     with DownloadServiceComponent
85     with UserHistoryCoordinatorServiceComponent
86     with JobCoordinatorServiceComponent
87     with MailJetConfigurationComponent
88     with DateHelperComponent
89     with PersistentPartnerServiceComponent =>
90 
91   override lazy val userService: UserService = new DefaultUserService
92 
93   class DefaultUserService extends UserService {
94 
95     val validationTokenExpiresIn: Long = makeSettings.validationTokenExpiresIn.toSeconds
96     val resetTokenExpiresIn: Long = makeSettings.resetTokenExpiresIn.toSeconds
97     val resetTokenB2BExpiresIn: Long = makeSettings.resetTokenB2BExpiresIn.toSeconds
98 
99     private lazy val batchSize: Int = mailJetConfiguration.userListBatchSize
100 
101     override def getUser(userId: UserId): Future[Option[User]] = {
102       persistentUserService.get(userId)
103     }
104 
105     override def getPersonality(id: UserId): Future[Option[User]] = {
106       persistentUserService.findByUserIdAndUserType(id, UserType.UserTypePersonality)
107     }
108 
109     override def getUserByEmail(email: String): Future[Option[User]] = {
110       persistentUserService.findByEmail(email.trim)
111     }
112 
113     override def getUserByUserIdAndPassword(userId: UserId, password: Option[String]): Future[Option[User]] = {
114       persistentUserService.findByUserIdAndPassword(userId, password)
115     }
116 
117     override def getUserByEmailAndPassword(email: String, password: String): Future[Option[User]] = {
118       persistentUserService.findByEmailAndPassword(email.trim, password)
119     }
120 
121     override def getUsersByUserIds(ids: Seq[UserId]): Future[Seq[User]] = {
122       persistentUserService.findAllByUserIds(ids)
123     }
124 
125     override def adminFindUsers(
126       offset: Pagination.Offset,
127       end: Option[Pagination.End],
128       sort: Option[String],
129       order: Option[Order],
130       ids: Option[Seq[UserId]],
131       email: Option[String],
132       firstName: Option[String],
133       lastName: Option[String],
134       role: Option[Role],
135       userType: Option[UserType]
136     ): Future[Seq[User]] = {
137       persistentUserService.adminFindUsers(
138         offset,
139         end,
140         sort,
141         order,
142         ids,
143         email.map(_.trim),
144         firstName.map(_.trim),
145         lastName.map(_.trim),
146         role,
147         userType
148       )
149     }
150 
151     private def registerUser(
152       userRegisterData: UserRegisterData,
153       lowerCasedEmail: String,
154       profile: Option[Profile],
155       hashedVerificationToken: String
156     ): Future[User] = {
157       val user = User(
158         userId = idGenerator.nextUserId(),
159         email = lowerCasedEmail,
160         firstName = userRegisterData.firstName,
161         lastName = userRegisterData.lastName,
162         lastIp = userRegisterData.lastIp,
163         hashedPassword = userRegisterData.password.map(_.boundedBcrypt),
164         enabled = true,
165         emailVerified = false,
166         lastConnection = Some(dateHelper.now()),
167         verificationToken = Some(hashedVerificationToken),
168         verificationTokenExpiresAt = Some(dateHelper.now().plusSeconds(validationTokenExpiresIn)),
169         resetToken = None,
170         resetTokenExpiresAt = None,
171         roles = userRegisterData.roles,
172         country = userRegisterData.country,
173         language = userRegisterData.language,
174         profile = profile,
175         availableQuestions = userRegisterData.availableQuestions,
176         availableEvents = userRegisterData.availableEvents,
177         userType = UserType.UserTypeUser,
178         publicProfile = userRegisterData.publicProfile,
179         privacyPolicyApprovalDate = userRegisterData.privacyPolicyApprovalDate
180       )
181 
182       persistentUserService.persist(user)
183     }
184 
185     private def persistPersonality(
186       personalityRegisterData: PersonalityRegisterData,
187       lowerCasedEmail: String,
188       profile: Option[Profile],
189       resetToken: String
190     ): Future[User] = {
191 
192       val user = User(
193         userId = idGenerator.nextUserId(),
194         email = lowerCasedEmail,
195         firstName = personalityRegisterData.firstName,
196         lastName = personalityRegisterData.lastName,
197         lastIp = None,
198         hashedPassword = None,
199         enabled = true,
200         emailVerified = true,
201         lastConnection = None,
202         verificationToken = None,
203         verificationTokenExpiresAt = None,
204         resetToken = Some(resetToken),
205         resetTokenExpiresAt = Some(dateHelper.now().plusSeconds(resetTokenB2BExpiresIn)),
206         roles = Seq(Role.RoleCitizen),
207         country = personalityRegisterData.country,
208         language = personalityRegisterData.language,
209         profile = profile,
210         availableQuestions = Seq.empty,
211         availableEvents = Seq.empty,
212         userType = UserType.UserTypePersonality,
213         publicProfile = true,
214         privacyPolicyApprovalDate = Some(DateHelper.now())
215       )
216 
217       persistentUserService.persist(user)
218     }
219 
220     private def persistVirtualUser(
221       virtualUserRegisterData: VirtualUserRegisterData,
222       lowerCasedEmail: String
223     ): Future[User] = {
224       val dateOfBirth = virtualUserRegisterData.age.map(age => LocalDate.now().minusYears(age).withDayOfYear(1))
225       val profile = Profile.parseProfile(dateOfBirth = dateOfBirth)
226 
227       val user = User(
228         userId = idGenerator.nextUserId(),
229         email = lowerCasedEmail,
230         firstName = virtualUserRegisterData.firstName,
231         lastName = None,
232         lastIp = None,
233         hashedPassword = None,
234         enabled = true,
235         emailVerified = false,
236         lastConnection = None,
237         verificationToken = None,
238         verificationTokenExpiresAt = None,
239         resetToken = None,
240         resetTokenExpiresAt = None,
241         roles = Seq(Role.RoleCitizen),
242         country = virtualUserRegisterData.country,
243         language = virtualUserRegisterData.language,
244         profile = profile,
245         availableQuestions = Seq.empty,
246         availableEvents = Seq.empty,
247         userType = UserType.UserTypeVirtual,
248         privacyPolicyApprovalDate = Some(DateHelper.now())
249       )
250 
251       persistentUserService.persist(user)
252     }
253 
254     private def persistExternalUser(
255       externalUserRegisterData: ExternalUserRegisterData,
256       lowerCasedEmail: String
257     ): Future[User] = {
258       val dateOfBirth = externalUserRegisterData.age.map(age => LocalDate.now().minusYears(age).withDayOfYear(1))
259       val profile = Profile.parseProfile(dateOfBirth = dateOfBirth)
260 
261       val user = User(
262         userId = idGenerator.nextUserId(),
263         email = lowerCasedEmail,
264         firstName = externalUserRegisterData.firstName,
265         lastName = None,
266         lastIp = None,
267         hashedPassword = None,
268         enabled = true,
269         emailVerified = false,
270         lastConnection = None,
271         verificationToken = None,
272         verificationTokenExpiresAt = None,
273         resetToken = None,
274         resetTokenExpiresAt = None,
275         roles = Seq(Role.RoleCitizen),
276         country = externalUserRegisterData.country,
277         language = externalUserRegisterData.language,
278         profile = profile,
279         availableQuestions = Seq.empty,
280         availableEvents = Seq.empty,
281         userType = UserType.UserTypeExternal,
282         privacyPolicyApprovalDate = Some(DateHelper.now())
283       )
284 
285       persistentUserService.persist(user)
286     }
287 
288     private def validateAccountCreation(
289       emailExists: Boolean,
290       canRegister: Boolean,
291       lowerCasedEmail: String
292     ): Future[Unit] = {
293       if (emailExists) {
294         Future.failed(EmailAlreadyRegisteredException(lowerCasedEmail))
295       } else if (!canRegister) {
296         Future.failed(EmailNotAllowed(lowerCasedEmail))
297       } else {
298         Future.unit
299       }
300     }
301 
302     private def generateVerificationToken(): Future[String] = {
303       userTokenGenerator.generateVerificationToken().map {
304         case (_, token) => token
305       }
306     }
307 
308     private def generateResetToken(): Future[String] = {
309       userTokenGenerator.generateResetToken().map {
310         case (_, token) => token
311       }
312     }
313 
314     private def generateReconnectToken(): Future[String] = {
315       userTokenGenerator.generateReconnectToken().map {
316         case (_, token) => token
317       }
318     }
319 
320     override def register(userRegisterData: UserRegisterData, requestContext: RequestContext): Future[User] = {
321       val trimmedRegisterData = userRegisterData.trim
322       val lowerCasedEmail: String = trimmedRegisterData.email.toLowerCase()
323       val profile: Option[Profile] =
324         Profile.parseProfile(
325           dateOfBirth = trimmedRegisterData.dateOfBirth,
326           profession = trimmedRegisterData.profession,
327           postalCode = trimmedRegisterData.postalCode,
328           gender = trimmedRegisterData.gender,
329           crmCountry = trimmedRegisterData.crmCountry,
330           crmLanguage = trimmedRegisterData.crmLanguage,
331           socioProfessionalCategory = trimmedRegisterData.socioProfessionalCategory,
332           registerQuestionId = trimmedRegisterData.questionId,
333           optInNewsletter = trimmedRegisterData.optIn.getOrElse(true),
334           optInPartner = trimmedRegisterData.optInPartner,
335           politicalParty = trimmedRegisterData.politicalParty,
336           website = trimmedRegisterData.website
337         )
338 
339       val result = for {
340         emailExists             <- persistentUserService.emailExists(lowerCasedEmail)
341         canRegister             <- userRegistrationValidator.canRegister(trimmedRegisterData)
342         _                       <- validateAccountCreation(emailExists, canRegister, lowerCasedEmail)
343         hashedVerificationToken <- generateVerificationToken()
344         user                    <- registerUser(trimmedRegisterData, lowerCasedEmail, profile, hashedVerificationToken)
345       } yield user
346 
347       result.map { user =>
348         eventBusService.publish(
349           UserRegisteredEvent(
350             connectedUserId = Some(user.userId),
351             userId = user.userId,
352             requestContext = requestContext,
353             firstName = user.firstName,
354             country = user.country,
355             optInPartner = user.profile.flatMap(_.optInPartner),
356             registerQuestionId = user.profile.flatMap(_.registerQuestionId),
357             eventDate = dateHelper.now(),
358             eventId = Some(idGenerator.nextEventId())
359           )
360         )
361         user
362       }
363     }
364 
365     override def registerPersonality(
366       personalityRegisterData: PersonalityRegisterData,
367       requestContext: RequestContext
368     ): Future[User] = {
369       val trimmedRegisterData = personalityRegisterData.trim
370       val lowerCasedEmail: String = trimmedRegisterData.email.toLowerCase()
371       val profile: Option[Profile] =
372         Profile.parseProfile(
373           gender = trimmedRegisterData.gender,
374           genderName = trimmedRegisterData.genderName,
375           description = trimmedRegisterData.description,
376           avatarUrl = trimmedRegisterData.avatarUrl,
377           optInNewsletter = false,
378           optInPartner = Some(false),
379           politicalParty = trimmedRegisterData.politicalParty,
380           website = trimmedRegisterData.website
381         )
382 
383       val result = for {
384         emailExists <- persistentUserService.emailExists(lowerCasedEmail)
385         _           <- validateAccountCreation(emailExists, canRegister = true, lowerCasedEmail)
386         resetToken  <- generateResetToken()
387         user        <- persistPersonality(trimmedRegisterData, lowerCasedEmail, profile, resetToken)
388       } yield user
389 
390       result.map { user =>
391         eventBusService.publish(
392           PersonalityRegisteredEvent(
393             connectedUserId = Some(user.userId),
394             userId = user.userId,
395             requestContext = requestContext,
396             email = user.email,
397             country = user.country,
398             eventDate = dateHelper.now(),
399             eventId = Some(idGenerator.nextEventId())
400           )
401         )
402         user
403       }
404     }
405 
406     override def registerVirtualUser(
407       virtualUserRegisterData: VirtualUserRegisterData,
408       requestContext: RequestContext
409     ): Future[User] = {
410       val trimmedRegisterData = virtualUserRegisterData.trim
411       val lowerCasedEmail: String = trimmedRegisterData.email.toLowerCase()
412 
413       for {
414         emailExists <- persistentUserService.emailExists(lowerCasedEmail)
415         _           <- validateAccountCreation(emailExists, canRegister = true, lowerCasedEmail)
416         user        <- persistVirtualUser(trimmedRegisterData, lowerCasedEmail)
417       } yield user
418     }
419 
420     override def registerExternalUser(
421       externalUserRegisterData: ExternalUserRegisterData,
422       requestContext: RequestContext
423     ): Future[User] = {
424       val trimmedRegisterData = externalUserRegisterData.trim
425       val lowerCasedEmail: String = trimmedRegisterData.email.toLowerCase()
426 
427       for {
428         emailExists <- persistentUserService.emailExists(lowerCasedEmail)
429         _           <- validateAccountCreation(emailExists, canRegister = true, lowerCasedEmail)
430         user        <- persistExternalUser(trimmedRegisterData, lowerCasedEmail)
431       } yield user
432     }
433 
434     override def createOrUpdateUserFromSocial(
435       userInfo: UserInfo,
436       questionId: Option[QuestionId],
437       country: Country,
438       language: Language,
439       crmCountry: Country,
440       crmLanguage: Language,
441       requestContext: RequestContext,
442       privacyPolicyApprovalDate: Option[ZonedDateTime],
443       optIn: Option[Boolean]
444     ): Future[(User, Boolean)] = {
445 
446       userInfo.email
447         .filter(_ != "")
448         .map(_.trim.toLowerCase())
449         .map { lowerCasedEmail =>
450           persistentUserService.findByEmail(lowerCasedEmail).flatMap {
451             case Some(user) =>
452               updateUserFromSocial(user, userInfo, requestContext.ipAddress, privacyPolicyApprovalDate)
453                 .map((_, false))
454             case None =>
455               createUserFromSocial(
456                 lowerCasedEmail,
457                 requestContext,
458                 userInfo,
459                 country,
460                 language,
461                 crmCountry,
462                 crmLanguage,
463                 questionId,
464                 privacyPolicyApprovalDate,
465                 optIn
466               ).map((_, true))
467           }
468         }
469         .getOrElse {
470           logger.error(
471             s"We couldn't find any email on social login. UserInfo: $userInfo, requestContext: $requestContext"
472           )
473           Future
474             .failed(ValidationFailedError(Seq(ValidationError("email", "missing", Some("No email found for user")))))
475         }
476     }
477 
478     private def createUserFromSocial(
479       lowerCasedEmail: String,
480       requestContext: RequestContext,
481       userInfo: UserInfo,
482       country: Country,
483       language: Language,
484       crmCountry: Country,
485       crmLanguage: Language,
486       questionId: Option[QuestionId],
487       privacyPolicyApprovalDate: Option[ZonedDateTime],
488       optIn: Option[Boolean]
489     ): Future[User] = {
490 
491       val profile: Option[Profile] =
492         Profile.parseProfile(
493           facebookId = userInfo.facebookId,
494           googleId = userInfo.googleId,
495           oidcInfos = userInfo.oidcInfos,
496           avatarUrl = userInfo.picture,
497           gender = userInfo.gender.map {
498             case "male"   => Male
499             case "female" => Female
500             case _        => Other
501           },
502           genderName = userInfo.gender,
503           crmCountry = crmCountry,
504           crmLanguage = crmLanguage,
505           registerQuestionId = questionId,
506           dateOfBirth = userInfo.dateOfBirth,
507           optInNewsletter = optIn.getOrElse(true)
508         )
509 
510       val user = User(
511         userId = idGenerator.nextUserId(),
512         email = lowerCasedEmail,
513         firstName = userInfo.firstName.map(_.trim),
514         lastName = None,
515         lastIp = requestContext.ipAddress,
516         hashedPassword = None,
517         enabled = true,
518         emailVerified = true,
519         lastConnection = Some(dateHelper.now()),
520         verificationToken = None,
521         verificationTokenExpiresAt = None,
522         resetToken = None,
523         resetTokenExpiresAt = None,
524         roles = Seq(Role.RoleCitizen),
525         country = country,
526         language = language,
527         profile = profile,
528         availableQuestions = Seq.empty,
529         availableEvents = Seq.empty,
530         userType = UserType.UserTypeUser,
531         privacyPolicyApprovalDate = privacyPolicyApprovalDate
532       )
533 
534       persistentUserService.persist(user).map { user =>
535         publishCreateEventsFromSocial(user = user, requestContext = requestContext)
536         user
537       }
538     }
539 
540     private def updateUserFromSocial(
541       user: User,
542       userInfo: UserInfo,
543       clientIp: Option[String],
544       privacyPolicyApprovalDate: Option[ZonedDateTime]
545     ): Future[User] = {
546       val hashedPassword = if (!user.emailVerified) None else user.hashedPassword
547 
548       val profile: Option[Profile] = user.profile.orElse(Profile.parseProfile())
549       val updatedProfile: Option[Profile] = profile.map(
550         _.copy(
551           facebookId = userInfo.facebookId.orElse(profile.flatMap(_.facebookId)),
552           googleId = userInfo.googleId.orElse(profile.flatMap(_.googleId)),
553           oidcInfos = Some(userInfo.oidcInfos.foldLeft(profile.flatMap(_.oidcInfos).getOrElse(Map.empty))(_ ++ _))
554             .filter(_.nonEmpty),
555           gender = userInfo.gender
556             .map[Gender] {
557               case "male"   => Male
558               case "female" => Female
559               case _        => Other
560             }
561             .orElse(profile.flatMap(_.gender)),
562           genderName = userInfo.gender.orElse(profile.flatMap(_.genderName)),
563           dateOfBirth = profile.flatMap(_.dateOfBirth).orElse(userInfo.dateOfBirth)
564         )
565       )
566       val updatedUser: User =
567         user.copy(
568           firstName = userInfo.firstName.orElse(user.firstName).map(_.trim),
569           lastIp = clientIp,
570           profile = updatedProfile,
571           hashedPassword = hashedPassword,
572           emailVerified = true,
573           lastConnection = Some(DateHelper.now()),
574           privacyPolicyApprovalDate = privacyPolicyApprovalDate.orElse(user.privacyPolicyApprovalDate)
575         )
576 
577       persistentUserService.updateSocialUser(updatedUser).map { userUpdated =>
578         if (userUpdated) updatedUser else user
579       }
580     }
581 
582     private def publishCreateEventsFromSocial(user: User, requestContext: RequestContext): Unit = {
583       eventBusService.publish(
584         UserRegisteredEvent(
585           connectedUserId = Some(user.userId),
586           userId = user.userId,
587           requestContext = requestContext,
588           firstName = user.firstName,
589           country = user.country,
590           isSocialLogin = true,
591           optInPartner = user.profile.flatMap(_.optInPartner),
592           registerQuestionId = user.profile.flatMap(_.registerQuestionId),
593           eventDate = dateHelper.now(),
594           eventId = Some(idGenerator.nextEventId())
595         )
596       )
597       eventBusService.publish(
598         UserValidatedAccountEvent(
599           userId = user.userId,
600           country = user.country,
601           requestContext = requestContext,
602           isSocialLogin = true,
603           eventDate = dateHelper.now(),
604           eventId = Some(idGenerator.nextEventId())
605         )
606       )
607       user.profile.flatMap(_.avatarUrl).foreach { avatarUrl =>
608         eventBusService.publish(
609           UserUploadAvatarEvent(
610             connectedUserId = Some(user.userId),
611             userId = user.userId,
612             country = user.country,
613             requestContext = requestContext,
614             avatarUrl = avatarUrl,
615             eventDate = dateHelper.now(),
616             eventId = Some(idGenerator.nextEventId())
617           )
618         )
619       }
620     }
621 
622     override def requestPasswordReset(userId: UserId): Future[Boolean] = {
623       for {
624         resetToken <- generateResetToken()
625         result <- persistentUserService.requestResetPassword(
626           userId,
627           resetToken,
628           Some(dateHelper.now().plusSeconds(resetTokenExpiresIn))
629         )
630       } yield result
631     }
632 
633     override def updatePassword(userId: UserId, resetToken: Option[String], password: String): Future[Boolean] = {
634       persistentUserService.updatePassword(userId, resetToken, password.boundedBcrypt)
635     }
636 
637     override def validateEmail(user: User, verificationToken: String): Future[TokenResponse] = {
638 
639       for {
640         emailVerified <- persistentUserService.validateEmail(verificationToken)
641         accessToken <- oauth2DataHandler.createAccessToken(authInfo = AuthInfo(
642           user = UserRights(user.userId, user.roles, user.availableQuestions, emailVerified),
643           clientId = None,
644           scope = None,
645           redirectUri = None
646         )
647         )
648       } yield {
649         TokenResponse.fromAccessToken(accessToken)
650       }
651     }
652 
653     override def updateOptInNewsletter(userId: UserId, optInNewsletter: Boolean): Future[Boolean] = {
654       persistentUserService.updateOptInNewsletter(userId, optInNewsletter).map { result =>
655         if (result) {
656           eventBusService.publish(
657             UserUpdatedOptInNewsletterEvent(
658               connectedUserId = Some(userId),
659               eventDate = dateHelper.now(),
660               userId = userId,
661               requestContext = RequestContext.empty,
662               optInNewsletter = optInNewsletter,
663               eventId = Some(idGenerator.nextEventId())
664             )
665           )
666         }
667         result
668       }
669     }
670 
671     override def updateIsHardBounce(userId: UserId, isHardBounce: Boolean): Future[Boolean] = {
672       persistentUserService.updateIsHardBounce(userId, isHardBounce)
673     }
674 
675     override def updateOptInNewsletter(email: String, optInNewsletter: Boolean): Future[Boolean] = {
676       getUserByEmail(email).flatMap { maybeUser =>
677         persistentUserService.updateOptInNewsletter(email, optInNewsletter).map { result =>
678           if (result) {
679             maybeUser.foreach { user =>
680               val userId = user.userId
681               eventBusService.publish(
682                 UserUpdatedOptInNewsletterEvent(
683                   connectedUserId = Some(userId),
684                   userId = userId,
685                   eventDate = dateHelper.now(),
686                   requestContext = RequestContext.empty,
687                   optInNewsletter = optInNewsletter,
688                   eventId = Some(idGenerator.nextEventId())
689                 )
690               )
691             }
692           }
693           result
694         }
695       }
696     }
697 
698     override def updateIsHardBounce(email: String, isHardBounce: Boolean): Future[Boolean] = {
699       persistentUserService.updateIsHardBounce(email, isHardBounce)
700     }
701 
702     override def updateLastMailingError(email: String, lastMailingError: Option[MailingErrorLog]): Future[Boolean] = {
703       persistentUserService.updateLastMailingError(email, lastMailingError)
704     }
705 
706     override def updateLastMailingError(userId: UserId, lastMailingError: Option[MailingErrorLog]): Future[Boolean] = {
707       persistentUserService.updateLastMailingError(userId, lastMailingError)
708     }
709 
710     override def getUsersWithoutRegisterQuestion: Future[Seq[User]] = {
711       persistentUserService.findUsersWithoutRegisterQuestion
712     }
713 
714     private def updateProposalVotedByOrganisation(user: User): Future[Unit] = {
715       if (user.userType == UserType.UserTypeOrganisation) {
716         proposalService
717           .searchProposalsVotedByUser(
718             userId = user.userId,
719             filterVotes = None,
720             filterQualifications = None,
721             preferredLanguage = None,
722             sort = None,
723             limit = None,
724             offset = None,
725             RequestContext.empty
726           )
727           .map(
728             result =>
729               result.results.foreach(
730                 proposal =>
731                   eventBusService
732                     .publish(
733                       ReindexProposal(
734                         proposal.id,
735                         dateHelper.now(),
736                         RequestContext.empty,
737                         Some(idGenerator.nextEventId())
738                       )
739                     )
740               )
741           )
742       } else {
743         Future.unit
744       }
745     }
746 
747     private def updateProposalsSubmitByUser(user: User, requestContext: RequestContext): Future[Unit] = {
748       proposalService
749         .searchForUser(
750           userId = Some(user.userId),
751           query = SearchQuery(filters = Some(
752             SearchFilters(
753               users = Some(UserSearchFilter(userIds = Seq(user.userId))),
754               status = Some(StatusSearchFilter(ProposalStatus.values))
755             )
756           )
757           ),
758           requestContext = requestContext,
759           preferredLanguage = None,
760           questionDefaultLanguage = None
761         )
762         .map(
763           result =>
764             result.results.foreach(
765               proposal =>
766                 eventBusService
767                   .publish(
768                     ReindexProposal(
769                       proposal.id,
770                       dateHelper.now(),
771                       RequestContext.empty,
772                       Some(idGenerator.nextEventId())
773                     )
774                   )
775             )
776         )
777     }
778 
779     private def handleEmailChangeIfNecessary(user: User, oldEmail: Option[String]): Future[Unit] = {
780       oldEmail match {
781         case Some(oldEmail) if oldEmail != user.email =>
782           for {
783             _ <- persistentUserService.findByEmail(user.email).flatMap {
784               case None => Future.unit
785               case _    => Future.failed(EmailAlreadyRegisteredException(user.email))
786             }
787             _ <- crmService.deleteRecipient(oldEmail)
788           } yield {}
789         case _ => Future.unit
790       }
791     }
792 
793     override def updateConnectionInfos(
794       userId: UserId,
795       lastConnection: Option[ZonedDateTime],
796       connectionAttemptsSinceLastSuccessful: Int,
797       privacyPolicyApprovalDate: Option[ZonedDateTime]
798     ): Future[Unit] = {
799       persistentUserService.updateConnectionInfos(
800         userId,
801         lastConnection,
802         connectionAttemptsSinceLastSuccessful,
803         privacyPolicyApprovalDate
804       )
805     }
806 
807     override def update(user: User, requestContext: RequestContext): Future[User] = {
808       for {
809         previousUser <- persistentUserService.get(user.userId)
810         updatedUser  <- persistentUserService.updateUser(user)
811         _            <- handleEmailChangeIfNecessary(user, previousUser.map(_.email))
812         _            <- updateProposalVotedByOrganisation(updatedUser)
813         _            <- updateProposalsSubmitByUser(updatedUser, requestContext)
814       } yield updatedUser
815     }
816 
817     private def updatePersonalityEmail(
818       personality: User,
819       moderatorId: Option[UserId],
820       newEmail: Option[String],
821       oldEmail: String,
822       requestContext: RequestContext
823     ): Unit =
824       newEmail.foreach(
825         email =>
826           eventBusService.publish(
827             PersonalityEmailChangedEvent(
828               connectedUserId = moderatorId,
829               userId = personality.userId,
830               requestContext = requestContext,
831               country = personality.country,
832               eventDate = dateHelper.now(),
833               oldEmail = oldEmail,
834               newEmail = email,
835               eventId = Some(idGenerator.nextEventId())
836             )
837           )
838       )
839 
840     override def updatePersonality(
841       personality: User,
842       moderatorId: Option[UserId],
843       oldEmail: String,
844       requestContext: RequestContext
845     ): Future[User] = {
846 
847       val newEmail: Option[String] = personality.email match {
848         case email if email.toLowerCase == oldEmail.toLowerCase => None
849         case email                                              => Some(email)
850       }
851 
852       val updateUserWithNewEmailIfNecessary = persistentUserService
853         .updateUser(personality)
854         .map(user => {
855           updatePersonalityEmail(user, moderatorId, newEmail, oldEmail, requestContext)
856           user
857         })
858 
859       for {
860         updatedUser <- updateUserWithNewEmailIfNecessary
861         _           <- updateProposalVotedByOrganisation(updatedUser)
862         _           <- updateProposalsSubmitByUser(updatedUser, requestContext)
863       } yield updatedUser
864     }
865 
866     override def anonymize(
867       user: User,
868       adminId: UserId,
869       requestContext: RequestContext,
870       mode: Anonymization
871     ): Future[Unit] = {
872 
873       val anonymizedUser: User = user.copy(
874         email = s"${user.userId.value.trim}@example.com",
875         firstName = if (mode == Anonymization.Explicit) Some("DELETE_REQUESTED") else user.firstName.map(_.trim),
876         lastName = Some("DELETE_REQUESTED"),
877         lastIp = None,
878         hashedPassword = None,
879         enabled = false,
880         emailVerified = false,
881         lastConnection = None,
882         verificationToken = None,
883         verificationTokenExpiresAt = None,
884         resetToken = None,
885         resetTokenExpiresAt = None,
886         roles = Seq(RoleCitizen),
887         profile = Profile.parseProfile(optInNewsletter = false),
888         isHardBounce = true,
889         lastMailingError = None,
890         organisationName = None,
891         userType = if (mode == Anonymization.Explicit) UserType.UserTypeAnonymous else user.userType,
892         createdAt = Some(dateHelper.now()),
893         publicProfile = false
894       )
895 
896       def unlinkUser(user: User): Future[Unit] = {
897         val unlinkPartners = user.userType match {
898           case UserType.UserTypeOrganisation =>
899             persistentPartnerService
900               .find(
901                 offset = Pagination.Offset.zero,
902                 end = None,
903                 sort = None,
904                 order = None,
905                 questionId = None,
906                 organisationId = Some(user.userId),
907                 partnerKind = None
908               )
909               .flatMap { partners =>
910                 Future.traverse(partners) { partner =>
911                   persistentPartnerService.modify(partner.copy(organisationId = None))
912                 }
913               }
914           case _ => Future.unit
915         }
916         for {
917           _ <- unlinkPartners
918           _ <- persistentUserService.removeAnonymizedUserFromFollowedUserTable(user.userId)
919         } yield {}
920       }
921 
922       val futureDelete: Future[Unit] = for {
923         _ <- persistentUserService.updateUser(anonymizedUser)
924         _ <- unlinkUser(user)
925         _ <- userHistoryCoordinatorService.delete(user.userId)
926         _ <- updateProposalsSubmitByUser(user, requestContext)
927       } yield {}
928       futureDelete.as(
929         eventBusService.publish(
930           UserAnonymizedEvent(
931             connectedUserId = Some(adminId),
932             userId = user.userId,
933             requestContext = requestContext,
934             country = user.country,
935             eventDate = dateHelper.now(),
936             adminId = adminId,
937             mode = mode,
938             eventId = Some(idGenerator.nextEventId())
939           )
940         )
941       )
942     }
943 
944     override def anonymizeInactiveUsers(adminId: UserId, requestContext: RequestContext): Future[JobAcceptance] = {
945       val startTime: Long = System.currentTimeMillis()
946 
947       jobCoordinatorService.start(AnonymizeInactiveUsers) { _ =>
948         val anonymizeUsers = StreamUtils
949           .asyncPageToPageSource(persistentCrmUserService.findInactiveUsers(_, batchSize))
950           .mapAsync(1) { crmUsers =>
951             persistentUserService.findAllByUserIds(crmUsers.map(user => UserId(user.userId)))
952           }
953           .mapConcat(identity)
954           .mapAsync(1) { user =>
955             anonymize(user, adminId, requestContext, Anonymization.Automatic)
956           }
957           .runWith(Sink.ignore)
958           .void
959 
960         anonymizeUsers.onComplete {
961           case Failure(exception) =>
962             logger.error(s"Inactive users anonymization failed:", exception)
963           case Success(_) =>
964             logger.info(s"Inactive users anonymization succeeded in ${System.currentTimeMillis() - startTime}ms")
965         }
966 
967         anonymizeUsers
968       }
969     }
970 
971     override def getFollowedUsers(userId: UserId): Future[Seq[UserId]] = {
972       persistentUserService.getFollowedUsers(userId).map(_.map(UserId(_)))
973     }
974 
975     override def followUser(followedUserId: UserId, userId: UserId, requestContext: RequestContext): Future[UserId] = {
976       persistentUserService.followUser(followedUserId, userId).map(_ => followedUserId).map { value =>
977         eventBusService.publish(
978           UserFollowEvent(
979             connectedUserId = Some(userId),
980             eventDate = dateHelper.now(),
981             userId = userId,
982             requestContext = requestContext,
983             followedUserId = followedUserId,
984             eventId = Some(idGenerator.nextEventId())
985           )
986         )
987         value
988       }
989     }
990 
991     override def unfollowUser(
992       followedUserId: UserId,
993       userId: UserId,
994       requestContext: RequestContext
995     ): Future[UserId] = {
996       persistentUserService.unfollowUser(followedUserId, userId).map(_ => followedUserId).map { value =>
997         eventBusService.publish(
998           UserUnfollowEvent(
999             connectedUserId = Some(userId),
1000             userId = userId,
1001             eventDate = dateHelper.now(),
1002             requestContext = requestContext,
1003             unfollowedUserId = followedUserId,
1004             eventId = Some(idGenerator.nextEventId())
1005           )
1006         )
1007         value
1008       }
1009     }
1010 
1011     override def retrieveOrCreateVirtualUser(userInfo: AuthorRequest, country: Country): Future[User] = {
1012       // Take only 50 chars to avoid having values too large for the column
1013       val fullHash: String = tokenGenerator.tokenToHash(s"$userInfo")
1014       val hash = fullHash.substring(0, Math.min(50, fullHash.length)).toLowerCase()
1015       val email = s"$hash@example.com"
1016       getUserByEmail(email).flatMap {
1017         case Some(user) => Future.successful(user)
1018         case None =>
1019           userService
1020             .registerVirtualUser(
1021               VirtualUserRegisterData(
1022                 email = email,
1023                 firstName = Some(userInfo.firstName),
1024                 age = userInfo.age,
1025                 country = country,
1026                 language = Language("fr")
1027               ),
1028               RequestContext.empty
1029             )
1030       }
1031     }
1032 
1033     override def retrieveOrCreateExternalUser(
1034       email: String,
1035       userInfo: AuthorRequest,
1036       country: Country,
1037       lineNumber: Int
1038     ): Future[User] = {
1039       getUserByEmail(email.trim).flatMap {
1040         case Some(user) if !user.firstName.contains(userInfo.firstName) =>
1041           Future.failed(
1042             ValidationFailedError(
1043               Seq(
1044                 ValidationError(
1045                   "firstName",
1046                   "unexpected_value",
1047                   Some(s"users with the same externalUserId should have the same firstName on line $lineNumber")
1048                 )
1049               )
1050             )
1051           )
1052         case Some(user)
1053             if user.profile
1054               .flatMap(_.dateOfBirth)
1055               .isEmpty && userInfo.age.isEmpty =>
1056           Future.successful(user)
1057         case Some(user)
1058             if !user.profile
1059               .flatMap(_.dateOfBirth)
1060               .exists(userInfo.age.map(age => LocalDate.now().minusYears(age).withDayOfYear(1)).contains) =>
1061           Future.failed(
1062             ValidationFailedError(
1063               Seq(
1064                 ValidationError(
1065                   "age",
1066                   "unexpected_value",
1067                   Some(s"users with the same externalUserId should have the same age on line $lineNumber")
1068                 )
1069               )
1070             )
1071           )
1072         case Some(user) =>
1073           Future.successful(user)
1074         case None =>
1075           userService
1076             .registerExternalUser(
1077               ExternalUserRegisterData(
1078                 email = email.trim,
1079                 firstName = Some(userInfo.firstName),
1080                 age = userInfo.age,
1081                 country = country,
1082                 language = Language("fr")
1083               ),
1084               RequestContext.empty
1085             )
1086       }
1087     }
1088 
1089     override def adminCountUsers(
1090       ids: Option[Seq[UserId]],
1091       email: Option[String],
1092       firstName: Option[String],
1093       lastName: Option[String],
1094       role: Option[Role],
1095       userType: Option[UserType]
1096     ): Future[Int] = {
1097       persistentUserService.adminCountUsers(
1098         ids,
1099         email.map(_.trim),
1100         firstName.map(_.trim),
1101         lastName.map(_.trim),
1102         role,
1103         userType
1104       )
1105     }
1106 
1107     private def getConnectionModes(user: User): Seq[ConnectionMode] = {
1108       Map(
1109         ConnectionMode.Facebook -> user.profile.flatMap(_.facebookId),
1110         ConnectionMode.Google -> user.profile.flatMap(_.googleId),
1111         ConnectionMode.Mail -> user.hashedPassword
1112       ).collect {
1113         case (mode, Some(_)) => mode
1114       }.toSeq
1115     }
1116 
1117     override def reconnectInfo(userId: UserId): Future[Option[ReconnectInfo]] = {
1118       val futureReconnectInfo = for {
1119         user           <- persistentUserService.get(userId)
1120         reconnectToken <- generateReconnectToken()
1121         _              <- persistentUserService.updateReconnectToken(userId, reconnectToken, dateHelper.now())
1122       } yield (user, reconnectToken)
1123 
1124       futureReconnectInfo.map {
1125         case (maybeUser, reconnectToken) =>
1126           maybeUser.map { user =>
1127             val hiddenEmail = SecurityHelper.anonymizeEmail(user.email)
1128             ReconnectInfo(
1129               reconnectToken,
1130               user.firstName,
1131               user.profile.flatMap(_.avatarUrl),
1132               hiddenEmail,
1133               getConnectionModes(user)
1134             )
1135           }
1136       }
1137     }
1138 
1139     override def changeEmailVerificationTokenIfNeeded(userId: UserId): Future[Option[User]] = {
1140       getUser(userId).flatMap {
1141         case Some(user)
1142             // if last verification token was changed more than 10 minutes ago
1143             if user.verificationTokenExpiresAt
1144               .forall(_.minusSeconds(validationTokenExpiresIn).plusMinutes(10).isBefore(dateHelper.now()))
1145               && !user.emailVerified =>
1146           userTokenGenerator.generateVerificationToken().flatMap {
1147             case (_, token) =>
1148               persistentUserService
1149                 .updateUser(
1150                   user.copy(
1151                     verificationToken = Some(token),
1152                     verificationTokenExpiresAt = Some(dateHelper.now().plusSeconds(validationTokenExpiresIn))
1153                   )
1154                 )
1155                 .map(Some(_))
1156           }
1157         case _ => Future.successful(None)
1158       }
1159     }
1160 
1161     def changeAvatarForUser(
1162       userId: UserId,
1163       avatarUrl: String,
1164       requestContext: RequestContext,
1165       eventDate: ZonedDateTime
1166     ): Future[Unit] = {
1167       def extension(contentType: ContentType): String = contentType.mediaType.subType
1168       def destFn(contentType: ContentType): File =
1169         Files.createTempFile("user-upload-avatar", s".${extension(contentType)}").toFile
1170 
1171       downloadService
1172         .downloadImage(avatarUrl, destFn)
1173         .flatMap {
1174           case (contentType, tempFile) =>
1175             tempFile.deleteOnExit()
1176             storageService
1177               .uploadUserAvatar(extension(contentType), contentType.value, FileContent(tempFile))
1178               .map(Option.apply)
1179         }
1180         .recover {
1181           case _: ImageUnavailable => None
1182         }
1183         .flatMap(
1184           path =>
1185             getUser(userId).flatMap {
1186               case Some(user) =>
1187                 val newProfile: Option[Profile] = user.profile match {
1188                   case Some(profile) => Some(profile.copy(avatarUrl = path))
1189                   case None          => Profile.parseProfile(avatarUrl = path)
1190                 }
1191                 update(user.copy(profile = newProfile), requestContext).map(_ => path)
1192               case None =>
1193                 logger.warn(s"Could not find user $userId to update avatar")
1194                 Future.successful(path)
1195             }
1196         )
1197         .as(
1198           userHistoryCoordinatorService.logHistory(
1199             LogUserUploadedAvatarEvent(
1200               userId = userId,
1201               requestContext = requestContext,
1202               action = UserAction(
1203                 date = eventDate,
1204                 actionType = LogUserUploadedAvatarEvent.action,
1205                 arguments = UploadedAvatar(avatarUrl = avatarUrl)
1206               )
1207             )
1208           )
1209         )
1210 
1211     }
1212 
1213     override def adminUpdateUserEmail(user: User, email: String): Future[Unit] = {
1214       persistentUserService.updateUser(user.copy(email = email)).void
1215     }
1216 
1217     private val FailedLoginAttemptsBeforeBan = 5
1218     private val LoginBanDuration = 5 // Minutes
1219 
1220     override def isThrottled(username: String): Future[Boolean] =
1221       getUserByEmail(username)
1222         .flattenOrFail(s"User $username does not exist")
1223         .map(
1224           user =>
1225             user.connectionAttemptsSinceLastSuccessful > FailedLoginAttemptsBeforeBan &&
1226               user.lastFailedConnectionAttempt.exists(ZonedDateTime.now().minusMinutes(LoginBanDuration).isBefore)
1227         )
1228 
1229     override def registerFailedConnectionAttempt(username: String): Future[Unit] =
1230       getUserByEmail(username).flatMap({
1231         case None => Future.unit
1232         case Some(previousUser) =>
1233           val updatedUser = previousUser.copy(
1234             connectionAttemptsSinceLastSuccessful = previousUser.connectionAttemptsSinceLastSuccessful + 1,
1235             lastFailedConnectionAttempt = Some(ZonedDateTime.now())
1236           )
1237           persistentUserService.updateUser(updatedUser).void
1238       })
1239   }
1240 }