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.Materializer
23 import akka.stream.scaladsl.{Flow, Sink, Source}
24 import akka.{Done, NotUsed}
25 import com.sksamuel.elastic4s.Index
26 import grizzled.slf4j.Logging
27 import org.make.api.organisation.OrganisationSearchEngineComponent
28 import org.make.api.proposal.ProposalSearchEngineComponent
29 import org.make.api.user.PersistentUserServiceComponent
30 import org.make.api.userhistory.{RequestUserVotedProposals, UserHistoryCoordinatorServiceComponent}
31 import org.make.core.history.HistoryActions.VoteAndQualifications
32 import org.make.core.proposal._
33 import org.make.core.proposal.indexed.{IndexedProposal, ProposalsSearchResult}
34 import org.make.core.question.QuestionId
35 import org.make.core.user.indexed.{IndexedOrganisation, ProposalsAndVotesCountsByQuestion}
36 import org.make.core.user.{User, UserId}
37 
38 import scala.concurrent.ExecutionContext.Implicits.global
39 import scala.concurrent.Future
40 import scala.concurrent.duration.DurationDouble
41 
42 trait OrganisationIndexationStream
43     extends IndexationStream
44     with OrganisationSearchEngineComponent
45     with PersistentUserServiceComponent
46     with UserHistoryCoordinatorServiceComponent
47     with ProposalSearchEngineComponent
48     with Logging {
49   object OrganisationStream {
50     def runIndexOrganisations(
51       organisationIndexName: String
52     )(implicit mat: Materializer): Flow[Seq[User], Done, NotUsed] =
53       Flow[Seq[User]]
54         .mapAsync(singleAsync)(organisations => executeIndexOrganisations(organisations, organisationIndexName))
55 
56     def flowIndexOrganisations(organisationIndexName: String)(implicit mat: Materializer): Flow[User, Done, NotUsed] =
57       groupedOrganisations.via(runIndexOrganisations(organisationIndexName))
58   }
59 
60   // Use custom `grouped` to reduce the size of every batch. Purpose: avoid a heavy load of actor commands.
61   private def groupedOrganisations: Flow[User, Seq[User], NotUsed] = Flow[User].groupedWithin(20, 500.milliseconds)
62 
63   private def executeIndexOrganisations(organisations: Seq[User], organisationIndexName: String)(
64     implicit mat: Materializer
65   ): Future[Done] = {
66     def futureVotedProposals(organisationId: UserId): Future[Seq[ProposalId]] =
67       userHistoryCoordinatorService.retrieveVotedProposals(RequestUserVotedProposals(organisationId))
68 
69     def futureFindProposals(proposalIds: Seq[ProposalId]): Future[ProposalsSearchResult] =
70       elasticsearchProposalAPI.searchProposals(
71         SearchQuery(filters = Some(SearchFilters(proposal = Some(ProposalSearchFilter(proposalIds)))))
72       )
73 
74     def futureVotes(
75       proposalIds: Seq[ProposalId],
76       organisationId: UserId
77     ): Future[Map[ProposalId, VoteAndQualifications]] =
78       userHistoryCoordinatorService.retrieveVoteAndQualifications(organisationId, proposalIds)
79 
80     def futureProposalsCountByQuestion(organisationId: UserId): Future[Map[QuestionId, Long]] =
81       elasticsearchProposalAPI.countProposalsByQuestion(
82         maybeQuestionIds = None,
83         status = Some(Seq(ProposalStatus.Accepted)),
84         maybeUserId = Some(organisationId),
85         toEnrich = None,
86         minVotesCount = None,
87         minScore = None
88       )
89 
90     def generateProposalsAndVotesCountsByQuestion(
91       proposals: ProposalsSearchResult,
92       votesByProposals: Map[ProposalId, VoteAndQualifications],
93       proposalsCountByQuestion: Map[QuestionId, Long]
94     ): Seq[ProposalsAndVotesCountsByQuestion] = {
95       val questionIds: Set[QuestionId] = proposalsCountByQuestion.keySet ++ proposals.results.collect {
96         case proposal if votesByProposals.keys.toSet.contains(proposal.id) => proposal.question.map(_.questionId)
97       }.flatten
98 
99       questionIds.map { questionId =>
100         val votedProposalsOfQuestion: Seq[IndexedProposal] =
101           proposals.results.filter(_.question.exists(_.questionId == questionId))
102         val votesCount: Int = votesByProposals.collect {
103           case (proposalId, votes) if votedProposalsOfQuestion.exists(_.id == proposalId) => votes
104         }.size
105         val proposalsCount: Int = proposalsCountByQuestion.getOrElse(questionId, 0L).toInt
106         ProposalsAndVotesCountsByQuestion(questionId, proposalsCount, votesCount)
107       }.toSeq
108     }
109 
110     Source(organisations)
111       .mapAsync(parallelism) { organisation =>
112         val futureCountsByQuestion: Future[Seq[ProposalsAndVotesCountsByQuestion]] = for {
113           proposalIds              <- futureVotedProposals(organisation.userId)
114           proposals                <- futureFindProposals(proposalIds)
115           votesByProposals         <- futureVotes(proposalIds, organisation.userId)
116           proposalsCountByQuestion <- futureProposalsCountByQuestion(organisation.userId)
117         } yield generateProposalsAndVotesCountsByQuestion(proposals, votesByProposals, proposalsCountByQuestion)
118 
119         futureCountsByQuestion.map { countsByQuestion =>
120           IndexedOrganisation.createFromOrganisation(organisation, countsByQuestion)
121         }
122       }
123       .groupedWithin(20, 500.milliseconds)
124       .mapAsync(singleAsync) { organisations =>
125         elasticsearchOrganisationAPI
126           .indexOrganisations(organisations, Some(Index(organisationIndexName)))
127           .recoverWith {
128             case e =>
129               logger.error("Indexing of one organisation chunk failed", e)
130               Future.successful(Done)
131           }
132       }
133       .runWith(Sink.ignore)
134   }
135 }