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.proposal
21 
22 import cats.data.NonEmptyList
23 import com.sksamuel.elastic4s.circe._
24 import com.sksamuel.elastic4s.ElasticDsl._
25 import com.sksamuel.elastic4s.Index
26 import com.sksamuel.elastic4s.requests.common.RefreshPolicy
27 import com.sksamuel.elastic4s.requests.searches.{SearchRequest, SearchResponse}
28 import com.sksamuel.elastic4s.requests.searches.aggs.responses.bucket.{TermBucket, Terms}
29 import com.sksamuel.elastic4s.requests.searches.term.TermQuery
30 import com.sksamuel.elastic4s.requests.script.Script
31 import com.sksamuel.elastic4s.requests.searches.aggs.{
32   AbstractAggregation,
33   FilterAggregation,
34   GlobalAggregation,
35   MaxAggregation,
36   PercentilesAggregation,
37   TermsAggregation,
38   TopHitsAggregation
39 }
40 import com.sksamuel.elastic4s.requests.searches.aggs.pipeline.BucketSortPipelineAgg
41 import com.sksamuel.elastic4s.requests.searches.aggs.responses.metrics.TopHits
42 import com.sksamuel.elastic4s.requests.searches.queries.{ExistsQuery, Query}
43 import com.sksamuel.elastic4s.requests.searches.queries.compound.BoolQuery
44 import com.sksamuel.elastic4s.requests.searches.sort.{FieldSort, SortOrder}
45 import grizzled.slf4j.Logging
46 import org.make.api.question.{AvatarsAndProposalsCount, PopularTagResponse}
47 import org.make.api.technical.elasticsearch._
48 import org.make.core.DateHelper._
49 import org.make.core.idea.IdeaId
50 import org.make.core.proposal.ProposalStatus.Accepted
51 import org.make.core.proposal._
52 import org.make.core.proposal.indexed.{
53   IndexedProposal,
54   ProposalElasticsearchFieldName,
55   ProposalsSearchResult,
56   SequencePool,
57   Zone
58 }
59 import org.make.core.question.QuestionId
60 import org.make.core.tag.TagId
61 import org.make.core.user.UserId
62 
63 import scala.concurrent.ExecutionContext.Implicits.global
64 import scala.concurrent.Future
65 
66 trait ProposalSearchEngineComponent {
67   def elasticsearchProposalAPI: ProposalSearchEngine
68 }
69 
70 //TODO: add multi-country
71 trait ProposalSearchEngine {
72   def findProposalById(proposalId: ProposalId): Future[Option[IndexedProposal]]
73 
74   def searchProposals(searchQuery: SearchQuery): Future[ProposalsSearchResult]
75 
76   def countProposals(searchQuery: SearchQuery): Future[Long]
77 
78   def countProposalsByQuestion(
79     maybeQuestionIds: Option[Seq[QuestionId]],
80     status: Option[Seq[ProposalStatus]],
81     maybeUserId: Option[UserId],
82     toEnrich: Option[Boolean],
83     minVotesCount: Option[Int],
84     minScore: Option[Double]
85   ): Future[Map[QuestionId, Long]]
86 
87   def countVotedProposals(searchQuery: SearchQuery): Future[Int]
88 
89   def indexProposals(records: Seq[IndexedProposal], mayBeIndex: Option[Index] = None): Future[Seq[IndexedProposal]]
90 
91   def updateProposals(records: Seq[IndexedProposal], mayBeIndex: Option[Index] = None): Future[Seq[IndexedProposal]]
92 
93   def getPopularTagsByProposal(questionId: QuestionId, size: Int): Future[Seq[PopularTagResponse]]
94 
95   def getTopProposals(
96     questionId: QuestionId,
97     size: Int,
98     aggregationField: ProposalElasticsearchFieldName
99   ): Future[Seq[IndexedProposal]]
100 
101   def countProposalsByIdea(ideaIds: Seq[IdeaId]): Future[Map[IdeaId, Long]]
102 
103   def getRandomProposalsByIdeaWithAvatar(ideaIds: Seq[IdeaId], seed: Int): Future[Map[IdeaId, AvatarsAndProposalsCount]]
104 
105   def computeTop20ConsensusThreshold(questionIds: NonEmptyList[QuestionId]): Future[Map[QuestionId, Double]]
106 
107   def getFeaturedProposals(searchQuery: SearchQuery): Future[ProposalsSearchResult]
108 }
109 
110 object ProposalSearchEngine {
111   val proposalIndexName: String = "proposal"
112 }
113 
114 trait DefaultProposalSearchEngineComponent extends ProposalSearchEngineComponent {
115   self: ElasticsearchConfigurationComponent with ElasticsearchClientComponent =>
116 
117   override lazy val elasticsearchProposalAPI: DefaultProposalSearchEngine = new DefaultProposalSearchEngine
118 
119   class DefaultProposalSearchEngine extends ProposalSearchEngine with Logging {
120 
121     private lazy val client = elasticsearchClient.client
122 
123     private val proposalAlias: Index =
124       elasticsearchConfiguration.proposalAliasName
125 
126     override def findProposalById(proposalId: ProposalId): Future[Option[IndexedProposal]] = {
127       client.executeAsFuture(get(proposalAlias, proposalId.value)).map(_.toOpt[IndexedProposal])
128     }
129 
130     override def searchProposals(searchQuery: SearchQuery): Future[ProposalsSearchResult] =
131       constructSearchQuery(searchQuery)
132 
133     override def getFeaturedProposals(searchQuery: SearchQuery): Future[ProposalsSearchResult] =
134       constructSearchQuery(searchQuery.copy(sortAlgorithm = Some(Featured)))
135 
136     private def constructSearchQuery(searchQuery: SearchQuery): Future[ProposalsSearchResult] = {
137       // parse json string to build search query
138       val searchFilters = SearchFilters.getSearchFilters(searchQuery)
139       val excludesFilters = SearchFilters.getExcludeFilters(searchQuery)
140       var request: SearchRequest = search(proposalAlias)
141         .bool(BoolQuery(must = searchFilters, not = excludesFilters))
142         .sortBy(SearchFilters.getSort(searchQuery).toList)
143         .from(SearchFilters.getSkipSearch(searchQuery))
144         .trackTotalHits(true)
145 
146       request = request.size(SearchFilters.getLimitSearch(searchQuery))
147 
148       searchQuery.sortAlgorithm.foreach { sortAlgorithm =>
149         request = sortAlgorithm.sortDefinition(request)
150       }
151 
152       client
153         .executeAsFuture(request)
154         .map(response => ProposalsSearchResult(total = response.totalHits, results = response.to[IndexedProposal]))
155     }
156 
157     override def countProposals(searchQuery: SearchQuery): Future[Long] = {
158       // parse json string to build search query
159       val searchFilters = SearchFilters.getSearchFilters(searchQuery)
160 
161       val request = search(proposalAlias)
162         .bool(BoolQuery(must = searchFilters))
163         .limit(0)
164         .trackTotalHits(true)
165 
166       client.executeAsFuture(request).map { response =>
167         response.totalHits
168       }
169 
170     }
171 
172     override def countProposalsByQuestion(
173       maybeQuestionIds: Option[Seq[QuestionId]],
174       status: Option[Seq[ProposalStatus]],
175       maybeUserId: Option[UserId],
176       toEnrich: Option[Boolean],
177       minVotesCount: Option[Int],
178       minScore: Option[Double]
179     ): Future[Map[QuestionId, Long]] = {
180       val searchQuery: SearchQuery = SearchQuery(filters = Some(
181         SearchFilters(
182           question = maybeQuestionIds.map(QuestionSearchFilter.apply),
183           status = status.map(StatusSearchFilter.apply),
184           users = maybeUserId.map(userId => UserSearchFilter(Seq(userId))),
185           toEnrich = toEnrich.map(ToEnrichSearchFilter.apply),
186           minVotesCount = minVotesCount.map(MinVotesCountSearchFilter.apply),
187           minScore = minScore.map(MinScoreSearchFilter.apply)
188         )
189       )
190       )
191       val searchFilters: Seq[Query] = SearchFilters.getSearchFilters(searchQuery)
192       val request: SearchRequest = search(proposalAlias).bool(BoolQuery(must = searchFilters))
193       val questionAggrSize: Int = maybeQuestionIds.map(_.length + 1).getOrElse(10000)
194 
195       val finalRequest: SearchRequest = request
196         .aggregations(
197           termsAgg(name = "questions", field = ProposalElasticsearchFieldName.questionId.field)
198             .size(size = questionAggrSize)
199             .minDocCount(min = 1)
200         )
201         .limit(0)
202         .trackTotalHits(true)
203 
204       client.executeAsFuture(finalRequest).map { response =>
205         response.aggregations
206           .result[Terms]("questions")
207           .buckets
208           .map(termBucket => QuestionId(termBucket.key) -> termBucket.docCount)
209           .toMap
210       }
211 
212     }
213 
214     override def countVotedProposals(searchQuery: SearchQuery): Future[Int] = {
215       // parse json string to build search query
216       val searchFilters = SearchFilters.getSearchFilters(searchQuery)
217 
218       val request = search(proposalAlias)
219         .bool(BoolQuery(must = searchFilters))
220         .aggregations(sumAgg("total_votes", "votes.count"))
221         .trackTotalHits(true)
222 
223       client.executeAsFuture(request).map { response =>
224         response.aggregations.sum("total_votes").valueOpt.map(_.toInt).getOrElse(0)
225       }
226     }
227 
228     override def indexProposals(
229       proposals: Seq[IndexedProposal],
230       mayBeIndex: Option[Index] = None
231     ): Future[Seq[IndexedProposal]] = {
232       @SuppressWarnings(Array("org.wartremover.warts.IterableOps"))
233       val records = proposals
234         .groupBy(_.id)
235         .map {
236           case (_, duplicatedProposals) =>
237             duplicatedProposals.maxBy(_.updatedAt)
238         }
239       val index = mayBeIndex.getOrElse(proposalAlias)
240       client
241         .executeAsFuture(bulk(records.map { record =>
242           indexInto(index).doc(record).refresh(RefreshPolicy.IMMEDIATE).id(record.id.value)
243         }))
244         .map(_ => records.toSeq)
245     }
246 
247     override def updateProposals(
248       proposals: Seq[IndexedProposal],
249       mayBeIndex: Option[Index] = None
250     ): Future[Seq[IndexedProposal]] = {
251       @SuppressWarnings(Array("org.wartremover.warts.IterableOps"))
252       val records = proposals
253         .groupBy(_.id)
254         .map {
255           case (_, duplicatedProposals) =>
256             duplicatedProposals.maxBy(_.updatedAt)
257         }
258       val index = mayBeIndex.getOrElse(proposalAlias)
259       client
260         .executeAsFuture(bulk(records.map { record =>
261           updateById(index, record.id.value)
262             .doc(record)
263             .refresh(RefreshPolicy.IMMEDIATE)
264         }))
265         .map(_ => records.toSeq)
266     }
267 
268     override def getPopularTagsByProposal(questionId: QuestionId, size: Int): Future[Seq[PopularTagResponse]] = {
269       // parse json string to build search query
270       val searchQuery: SearchQuery =
271         SearchQuery(filters = Some(SearchFilters(question = Some(QuestionSearchFilter(Seq(questionId))))))
272       val searchFilters: Seq[Query] = SearchFilters.getSearchFilters(searchQuery)
273       val request: SearchRequest = search(proposalAlias).bool(BoolQuery(must = searchFilters))
274 
275       val finalRequest: SearchRequest = request
276         .aggregations(
277           TermsAggregation(
278             name = "popularTags",
279             script = Some(
280               Script(
281                 "def tags = params._source['tags'];tags.removeIf(tag -> !tag['display']);String[] idLabel = new String[tags.length]; for (int i = 0; i < tags.length; i++) { idLabel[i] = tags[i]['tagId'] + ',' + tags[i]['label'] } return idLabel"
282               )
283             )
284           ).size(size = size)
285         )
286         .limit(0)
287         .trackTotalHits(true)
288 
289       def popularTagResponseFrombucket(bucket: TermBucket): PopularTagResponse = {
290         val Array(tagId, label) = bucket.key.split(",", 2)
291         PopularTagResponse(TagId(tagId), label, bucket.docCount)
292       }
293 
294       client.executeAsFuture(finalRequest).map { response =>
295         response.aggregations
296           .result[Terms]("popularTags")
297           .buckets
298           .map(popularTagResponseFrombucket)
299       }
300 
301     }
302 
303     override def getTopProposals(
304       questionId: QuestionId,
305       size: Int,
306       aggregationField: ProposalElasticsearchFieldName
307     ): Future[Seq[IndexedProposal]] = {
308       val topHitsAggregationName = "topHits"
309       val termsAggregationName = "termsAgg"
310       val maxAggregationName = "maxTopScore"
311 
312       val searchQuery: SearchQuery =
313         SearchQuery(filters = Some(SearchFilters(question = Some(QuestionSearchFilter(Seq(questionId))))))
314       val searchFilters: Seq[Query] = SearchFilters.getSearchFilters(searchQuery)
315       val request: SearchRequest = search(proposalAlias).bool(BoolQuery(must = searchFilters))
316 
317       // This aggregation create a field "maxTopScore" with the max value of indexedProposal.scores.topScore
318       val maxAggregation: AbstractAggregation =
319         MaxAggregation(name = maxAggregationName, field = Some(ProposalElasticsearchFieldName.scoreLowerBound.field))
320 
321       // This aggregation sort each bucket from the field "maxTopScore"
322       val bucketSortAggregation: AbstractAggregation = BucketSortPipelineAgg(
323         name = "topScoreBucketSort",
324         sort = Seq(FieldSort(field = maxAggregationName, order = SortOrder.DESC))
325       )
326 
327       // This aggregation take the proposal with the highest indexedProposal.scores.topScore on each bucket
328       val topHitsAggregation: AbstractAggregation =
329         TopHitsAggregation(
330           name = topHitsAggregationName,
331           sorts = Seq(FieldSort(field = ProposalElasticsearchFieldName.scoreLowerBound.field, order = SortOrder.DESC)),
332           size = Some(1)
333         )
334 
335       val finalRequest: SearchRequest = request
336         .aggregations(
337           TermsAggregation(name = termsAggregationName, field = Some(aggregationField.field), size = Some(size))
338             .subAggregations(Seq(maxAggregation, bucketSortAggregation, topHitsAggregation)) // Those 3 subAggregation are execute on each bucket created by the parent aggregation
339             .minDocCount(min = 1)
340         )
341         .size(0)
342         .trackTotalHits(true)
343 
344       client.executeAsFuture(finalRequest).map { response =>
345         response.aggregations
346           .result[Terms](termsAggregationName)
347           .buckets
348           .flatMap(_.result[TopHits](topHitsAggregationName).hits.map(_.to[IndexedProposal]))
349       }
350     }
351 
352     override def countProposalsByIdea(ideaIds: Seq[IdeaId]): Future[Map[IdeaId, Long]] = {
353       val searchFilters = SearchFilters.getSearchFilters(
354         SearchQuery(filters = Some(SearchFilters(idea = Some(IdeaSearchFilter(ideaIds)))))
355       )
356 
357       val request = search(proposalAlias)
358         .bool(BoolQuery(must = searchFilters))
359         .aggregations(termsAgg("by_idea", ProposalElasticsearchFieldName.ideaId.field))
360         .size(0)
361         .trackTotalHits(true)
362 
363       client.executeAsFuture(request).map { response =>
364         response.aggregations
365           .result[Terms]("by_idea")
366           .buckets
367           .map(termBucket => IdeaId(termBucket.key) -> termBucket.docCount)
368           .toMap
369       }
370 
371     }
372 
373     override def getRandomProposalsByIdeaWithAvatar(
374       ideaIds: Seq[IdeaId],
375       seed: Int
376     ): Future[Map[IdeaId, AvatarsAndProposalsCount]] = {
377       val avatarsSize = 4
378 
379       // this aggregation count the proposals without taking account of the search filters set for the bool query
380       val globalAggregation = GlobalAggregation(name = "all_proposals")
381         .subAggregations(
382           FilterAggregation(
383             name = "filter_global",
384             query = TermQuery(field = ProposalElasticsearchFieldName.status.field, value = Accepted.value)
385           ).subAggregations(
386             TermsAggregation(
387               name = "by_idea_global",
388               field = Some(ProposalElasticsearchFieldName.ideaId.field),
389               includeExactValues = ideaIds.map(_.value),
390               size = Some(ideaIds.size)
391             )
392           )
393         )
394 
395       val topHitsAggregation =
396         TopHitsAggregation(
397           name = "top_proposals",
398           sorts = Seq(FieldSort(field = "_score", order = SortOrder.DESC)),
399           size = Some(avatarsSize)
400         )
401 
402       var request = search(proposalAlias)
403         .bool(BoolQuery(must = Seq(ExistsQuery(field = ProposalElasticsearchFieldName.authorAvatarUrl.field))))
404         .aggregations(
405           TermsAggregation(
406             name = "by_idea",
407             field = Some(ProposalElasticsearchFieldName.ideaId.field),
408             includeExactValues = ideaIds.map(_.value),
409             size = Some(ideaIds.size)
410           ).subAggregations(topHitsAggregation),
411           globalAggregation
412         )
413         .size(0)
414         .trackTotalHits(true)
415 
416       request = RandomAlgorithm(seed).sortDefinition(request)
417 
418       ideaIds match {
419         case Seq() => Future.successful(Map.empty)
420         case _ =>
421           client.executeAsFuture(request).map { response =>
422             computeAvatarAndProposalsCountResponse(response)
423           }
424       }
425     }
426 
427     private def computeAvatarAndProposalsCountResponse(
428       response: SearchResponse
429     ): Map[IdeaId, AvatarsAndProposalsCount] = {
430       val proposalsCountByIdea: Map[String, Long] = response.aggregations
431         .global("all_proposals")
432         .filter("filter_global")
433         .result[Terms]("by_idea_global")
434         .buckets
435         .map(bucket => bucket.key -> bucket.docCount)
436         .toMap
437       val avatarsByIdea: Map[String, Seq[String]] = response.aggregations
438         .result[Terms]("by_idea")
439         .buckets
440         .map(
441           bucket =>
442             bucket.key -> bucket
443               .result[TopHits]("top_proposals")
444               .hits
445               .map(_.to[IndexedProposal])
446               .map(_.author.avatarUrl.getOrElse(""))
447         )
448         .toMap
449       proposalsCountByIdea.map {
450         case (ideaId, count) =>
451           IdeaId(ideaId) -> AvatarsAndProposalsCount(
452             avatars = avatarsByIdea.getOrElse(ideaId, Seq.empty),
453             proposalsCount = count.toInt
454           )
455       }
456     }
457 
458     override def computeTop20ConsensusThreshold(
459       questionIds: NonEmptyList[QuestionId]
460     ): Future[Map[QuestionId, Double]] = {
461       val searchFilters = SearchFilters.getSearchFilters(
462         SearchQuery(filters = Some(
463           SearchFilters(
464             question = Some(QuestionSearchFilter(questionIds.toList)),
465             sequencePool = Some(SequencePoolSearchFilter(SequencePool.Tested)),
466             zone = Some(ZoneSearchFilter(Zone.Consensus))
467           )
468         )
469         )
470       )
471 
472       val consensusOutlier = PercentilesAggregation(
473         name = "consensus_outlier",
474         field = Some(ProposalElasticsearchFieldName.scoreLowerBound.field),
475         percents = Seq(80d)
476       )
477       val aggByQuestion = TermsAggregation(
478         name = "by_question",
479         field = Some(ProposalElasticsearchFieldName.questionId.field),
480         size = Some(questionIds.size),
481         subaggs = Seq(consensusOutlier)
482       )
483       val request = search(proposalAlias)
484         .bool(BoolQuery(must = searchFilters))
485         .aggregations(aggByQuestion)
486         .limit(0)
487         .trackTotalHits(true)
488 
489       client.executeAsFuture(request).map { response =>
490         response.aggregations
491           .result[Terms](aggByQuestion.name)
492           .buckets
493           .flatMap { termBucket =>
494             termBucket.percentiles(consensusOutlier.name).values.get("80.0").map(QuestionId(termBucket.key) -> _)
495           }
496           .toMap
497       }
498 
499     }
500 
501   }
502 
503 }