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.elasticsearch
21 
22 import akka.stream.scaladsl.GraphDSL.Implicits._
23 import akka.stream.scaladsl.{Flow, GraphDSL, Merge, Partition, Sink, Source}
24 import akka.stream.{FlowShape, Materializer}
25 import akka.{Done, NotUsed}
26 import cats.data.OptionT
27 import cats.implicits._
28 import com.sksamuel.elastic4s.Index
29 import grizzled.slf4j.Logging
30 import org.make.api.operation.{OperationOfQuestionServiceComponent, OperationServiceComponent}
31 import org.make.api.organisation.OrganisationServiceComponent
32 import org.make.api.proposal.ProposalScorer.VotesCounter
33 import org.make.api.proposal.{ProposalCoordinatorServiceComponent, ProposalScorer, ProposalSearchEngineComponent}
34 import org.make.api.question.QuestionServiceComponent
35 import org.make.api.segment.SegmentServiceComponent
36 import org.make.api.sequence.SequenceConfigurationComponent
37 import org.make.api.tag.TagServiceComponent
38 import org.make.api.tagtype.TagTypeServiceComponent
39 import org.make.api.technical.RetryableFuture
40 import org.make.api.user.UserServiceComponent
41 import org.make.core.SlugHelper
42 import org.make.core.operation.{OperationOfQuestion, SimpleOperation}
43 import org.make.core.proposal._
44 import org.make.core.proposal.indexed.{
45   IndexedAuthor,
46   IndexedOrganisationInfo,
47   IndexedProposal,
48   IndexedProposalKeyword,
49   IndexedProposalQuestion,
50   IndexedScores,
51   IndexedTag,
52   SequencePool,
53   IndexedContext => ProposalContext
54 }
55 import org.make.core.question.{Question, QuestionId}
56 import org.make.core.sequence.SequenceConfiguration
57 import org.make.core.tag.{Tag, TagType}
58 import org.make.core.technical.Multilingual
59 import org.make.core.user.User
60 
61 import java.time.temporal.ChronoUnit
62 import java.time.{LocalDate, ZonedDateTime}
63 import scala.concurrent.ExecutionContext.Implicits.global
64 import scala.concurrent.Future
65 import scala.concurrent.duration.DurationInt
66 
67 trait ProposalIndexationStream
68     extends IndexationStream
69     with ProposalCoordinatorServiceComponent
70     with UserServiceComponent
71     with OrganisationServiceComponent
72     with OperationOfQuestionServiceComponent
73     with OperationServiceComponent
74     with QuestionServiceComponent
75     with TagServiceComponent
76     with TagTypeServiceComponent
77     with ProposalSearchEngineComponent
78     with SequenceConfigurationComponent
79     with Logging
80     with SegmentServiceComponent {
81 
82   object ProposalStream {
83 
84     private def maybeIndexedProposal(implicit mat: Materializer): Flow[ProposalId, Option[IndexedProposal], NotUsed] =
85       Flow[ProposalId]
86         .mapAsync(parallelism) { proposalId =>
87           RetryableFuture.retryOnAskTimeout(() => getIndexedProposal(proposalId))
88         }
89 
90     private def runIndexProposals(proposalIndexName: String): Flow[Seq[IndexedProposal], Done, NotUsed] =
91       Flow[Seq[IndexedProposal]].mapAsync(parallelism)(proposals => executeIndexProposals(proposals, proposalIndexName))
92 
93     private val findOrElseIndexedProposal: Flow[IndexedProposal, ProposalFlow, NotUsed] =
94       Flow[IndexedProposal]
95         .mapAsync(parallelism) { proposal =>
96           elasticsearchProposalAPI.findProposalById(proposal.id).map {
97             case Some(_) => UpdateProposalFlow(proposal)
98             case _       => IndexProposalFlow(proposal)
99           }
100         }
101 
102     private val indexProposals: Flow[Seq[IndexedProposal], Seq[IndexedProposal], NotUsed] =
103       Flow[Seq[IndexedProposal]].mapAsync(singleAsync) { proposals =>
104         elasticsearchProposalAPI.indexProposals(proposals)
105       }
106 
107     private val updateProposals: Flow[Seq[IndexedProposal], Seq[IndexedProposal], NotUsed] =
108       Flow[Seq[IndexedProposal]].mapAsync(singleAsync) { proposals =>
109         elasticsearchProposalAPI.updateProposals(proposals)
110       }
111 
112     def flowIndexProposals(proposalIndexName: String)(implicit mat: Materializer): Flow[ProposalId, Done, NotUsed] =
113       maybeIndexedProposal
114         .via(filterIsDefined[IndexedProposal])
115         .groupedWithin(100, 500.milliseconds)
116         .via(runIndexProposals(proposalIndexName))
117 
118     def indexOrUpdateFlow(implicit mat: Materializer): Flow[ProposalId, Seq[IndexedProposal], NotUsed] =
119       Flow.fromGraph[ProposalId, Seq[IndexedProposal], NotUsed](GraphDSL.create() {
120         implicit builder: GraphDSL.Builder[NotUsed] =>
121           val source = builder.add(maybeIndexedProposal)
122           val partition = builder.add(Partition[ProposalFlow](outputPorts = 2, partitioner = {
123             case IndexProposalFlow(_)  => 0
124             case UpdateProposalFlow(_) => 1
125           }))
126           val merge = builder.add(Merge[Seq[IndexedProposal]](2))
127 
128           val filterIndex: Flow[ProposalFlow, IndexedProposal, NotUsed] =
129             Flow[ProposalFlow].filter {
130               case IndexProposalFlow(_)  => true
131               case UpdateProposalFlow(_) => false
132             }.map(_.proposal)
133 
134           val filterUpdate: Flow[ProposalFlow, IndexedProposal, NotUsed] =
135             Flow[ProposalFlow].filter {
136               case IndexProposalFlow(_)  => false
137               case UpdateProposalFlow(_) => true
138             }.map(_.proposal)
139 
140           source.out ~> filterIsDefined[IndexedProposal] ~> findOrElseIndexedProposal ~> partition.in
141 
142           partition.out(0) ~> filterIndex  ~> grouped[IndexedProposal] ~> indexProposals  ~> merge
143           partition.out(1) ~> filterUpdate ~> grouped[IndexedProposal] ~> updateProposals ~> merge
144 
145           FlowShape(source.in, merge.out)
146       })
147   }
148 
149   private def getQuestion: Option[QuestionId] => Future[Option[Question]] = {
150     case Some(questionId) => questionService.getQuestion(questionId)
151     case None             => Future.successful[Option[Question]](None)
152   }
153 
154   private def getSelectedStakeTag(tags: Seq[Tag], tagTypes: Seq[TagType])(
155     implicit mat: Materializer
156   ): Future[Option[IndexedTag]] = {
157     tagTypes.find(_.label.toLowerCase == "stake") match {
158       case None => Future.failed(new IllegalStateException("Unable to find stake tag types"))
159       case Some(stakeTypeTag) =>
160         val stakeTags: Seq[Tag] = tags.filter(_.tagTypeId.value == stakeTypeTag.tagTypeId.value)
161         tagService.retrieveIndexedTags(stakeTags, Seq(stakeTypeTag)).filter(_.display) match {
162           case Seq()         => Future.successful(None)
163           case Seq(stakeTag) => Future.successful(Some(stakeTag))
164           case indexedTags =>
165             Source(indexedTags)
166               .mapAsync(5) { tag =>
167                 elasticsearchProposalAPI
168                   .countProposals(
169                     SearchQuery(filters = Some(SearchFilters(tags = Some(TagsSearchFilter(Seq(tag.tagId))))))
170                   )
171                   .map(tag -> _)
172               }
173               .runWith(Sink.seq)
174               .map {
175                 _.sortBy {
176                   case (tag, count) => (count * -1, tag.label)
177                 }.collectFirst {
178                   case (tag, _) => tag
179                 }
180               }
181         }
182     }
183   }
184 
185   def getIndexedProposal(proposalId: ProposalId)(implicit mat: Materializer): Future[Option[IndexedProposal]] = {
186 
187     val maybeResult: OptionT[Future, IndexedProposal] = for {
188       proposal         <- OptionT(proposalCoordinatorService.getProposal(proposalId))
189       user             <- OptionT(userService.getUser(proposal.author))
190       tags             <- OptionT(tagService.findByTagIds(proposal.tags).map(Option.apply))
191       tagTypes         <- OptionT(tagTypeService.findAll().map(Option.apply))
192       selectedStakeTag <- OptionT(getSelectedStakeTag(tags, tagTypes).map(Option.apply))
193       organisationInfos <- OptionT(
194         Source(proposal.organisationIds)
195           .mapAsync(5)(organisationService.getOrganisation)
196           .runWith(Sink.seq)
197           .map(organisations => Option(organisations.flatten))
198       )
199       question            <- OptionT(getQuestion(proposal.questionId))
200       operationOfQuestion <- OptionT(operationOfQuestionService.findByQuestionId(question.questionId))
201       sequenceConfiguration <- OptionT(
202         sequenceConfigurationService.getSequenceConfigurationByQuestionId(question.questionId).map(Option.apply)
203       )
204       operation <- OptionT(operationService.findOneSimple(operationOfQuestion.operationId))
205       // in order to insert this in this for-comprehension correctly we need to transform the Future[Option[String]]
206       // into a Future[Option[Option[String]]] since we want to keep an Option[String] in the end.
207       segment <- OptionT(segmentService.resolveSegment(proposal.creationContext).map(Option(_)))
208     } yield {
209       createIndexedProposal(
210         proposal,
211         segment,
212         sequenceConfiguration,
213         user,
214         organisationInfos,
215         tagService.retrieveIndexedTags(tags, tagTypes),
216         Proposal.needsEnrichment(proposal.status, tagTypes, tags.map(_.tagTypeId)),
217         selectedStakeTag,
218         question,
219         operationOfQuestion,
220         operation
221       )
222     }
223 
224     maybeResult.value
225   }
226 
227   @SuppressWarnings(Array("org.wartremover.warts.Throw"))
228   private def createIndexedProposal(
229     proposal: Proposal,
230     segment: Option[String],
231     sequenceConfiguration: SequenceConfiguration,
232     user: User,
233     organisationInfos: Seq[User],
234     tags: Seq[IndexedTag],
235     needsEnrichment: Boolean,
236     selectedStakeTag: Option[IndexedTag],
237     question: Question,
238     operationOfQuestion: OperationOfQuestion,
239     operation: SimpleOperation
240   ): IndexedProposal = {
241 
242     val isBeforeContextSourceFeature: Boolean =
243       proposal.createdAt.exists(_.isBefore(ZonedDateTime.parse("2018-09-01T00:00:00Z")))
244     val votingOptions = proposal.votingOptions.getOrElse(VotingOptions.empty)
245     val sequenceScore =
246       ProposalScorer(votingOptions, VotesCounter.SequenceVotesCounter, sequenceConfiguration.nonSequenceVotesWeight)
247 
248     // If the proposal is not segmented, the scores should all be at 0
249     val segmentScore = segment.as(
250       ProposalScorer(votingOptions, VotesCounter.SegmentVotesCounter, sequenceConfiguration.nonSequenceVotesWeight)
251     )
252 
253     val mainLanguage = proposal.submittedAsLanguage.getOrElse(question.defaultLanguage)
254     val contentVersions =
255       proposal.contentTranslations
256         .getOrElse(Multilingual.empty)
257         .addTranslation(mainLanguage, proposal.content)
258 
259     IndexedProposal(
260       id = proposal.proposalId,
261       submittedAsLanguage = proposal.submittedAsLanguage,
262       content = contentVersions,
263       contentGeneral = proposal.content,
264       contentKeyword = proposal.content,
265       slug = proposal.slug,
266       status = proposal.status,
267       createdAt = proposal.createdAt match {
268         case Some(date) => date
269         case _          => throw new IllegalStateException("created at is required")
270       },
271       updatedAt = proposal.updatedAt,
272       validatedAt = proposal.validatedAt,
273       postponedAt = proposal.postponedAt,
274       votes = votingOptions,
275       votesCount = votingOptions.counts.totalVotes,
276       votesVerifiedCount = votingOptions.verifiedCounts.totalVotes,
277       votesSequenceCount = votingOptions.sequenceCounts.totalVotes,
278       votesSegmentCount = votingOptions.segmentCounts.totalVotes,
279       toEnrich = needsEnrichment,
280       scores = IndexedScores(sequenceScore, sequenceConfiguration.mainSequence.sequenceThresholds),
281       segmentScores =
282         segmentScore.fold(IndexedScores.empty)(ss => IndexedScores(ss, sequenceConfiguration.mainSequence.sequenceThresholds)),
283       agreementRate = BaseVote.rate(votingOptions, VoteKey.Agree),
284       context = Some(ProposalContext(proposal.creationContext, isBeforeContextSourceFeature)),
285       isAnonymous = proposal.isAnonymous,
286       author = IndexedAuthor(
287         userId = user.userId,
288         firstName = user.firstName,
289         displayName = user.displayName,
290         organisationName = user.organisationName,
291         organisationSlug = user.organisationName.map(name => SlugHelper(name)),
292         postalCode = user.profile.flatMap(_.postalCode),
293         age = user.profile
294           .flatMap(_.dateOfBirth)
295           .map(date => ChronoUnit.YEARS.between(date, LocalDate.now()).toInt),
296         avatarUrl = user.profile.flatMap(_.avatarUrl),
297         userType = user.userType,
298         profession = user.profile.flatMap(_.profession)
299       ),
300       organisations = organisationInfos
301         .map(
302           organisation =>
303             IndexedOrganisationInfo(
304               organisation.userId,
305               organisation.organisationName,
306               organisation.organisationName.map(name => SlugHelper(name))
307             )
308         ),
309       tags = tags,
310       selectedStakeTag = selectedStakeTag,
311       ideaId = proposal.idea,
312       operationId = proposal.operation,
313       question = proposal.questionId.map(
314         questionId =>
315           IndexedProposalQuestion(
316             questionId = questionId,
317             slug = question.slug,
318             titles = operationOfQuestion.operationTitles,
319             questions = question.questions,
320             countries = question.countries,
321             languages = question.languages,
322             startDate = operationOfQuestion.startDate,
323             endDate = operationOfQuestion.endDate,
324             isOpen = operationOfQuestion.status == OperationOfQuestion.Status.Open
325           )
326       ),
327       sequencePool = sequenceScore
328         .pool(sequenceConfiguration, proposal.status),
329       sequenceSegmentPool =
330         segmentScore.map(_.pool(sequenceConfiguration, proposal.status)).getOrElse(SequencePool.Excluded),
331       initialProposal = proposal.initialProposal,
332       proposalType = proposal.proposalType,
333       refusalReason = proposal.refusalReason,
334       operationKind = Option(operation.operationKind),
335       segment = segment,
336       keywords = proposal.keywords.map(IndexedProposalKeyword.apply)
337     )
338   }
339 
340   private def executeIndexProposals(proposals: Seq[IndexedProposal], indexName: String): Future[Done] = {
341     elasticsearchProposalAPI
342       .indexProposals(proposals, Some(Index(indexName)))
343       .map(_ => Done)
344       .recoverWith {
345         case e =>
346           logger.error("Indexing proposals in proposal index failed", e)
347           Future.successful(Done)
348       }
349   }
350 
351 }
352 
353 sealed trait ProposalFlow {
354   val proposal: IndexedProposal
355 }
356 
357 final case class IndexProposalFlow(override val proposal: IndexedProposal) extends ProposalFlow
358 
359 final case class UpdateProposalFlow(override val proposal: IndexedProposal) extends ProposalFlow