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 cats.implicits._
23 import akka.Done
24 import akka.actor.typed.scaladsl.AskPattern.schedulerFromActorSystem
25 import akka.stream.scaladsl._
26 import akka.util.Timeout
27 import cats.data.NonEmptyList
28 import com.sksamuel.elastic4s.ElasticDsl._
29 import com.sksamuel.elastic4s.requests.indexes.CreateIndexResponse
30 import com.sksamuel.elastic4s.requests.indexes.admin.AliasActionResponse
31 import eu.timepit.refined.auto._
32 import grizzled.slf4j.Logging
33 import org.make.api.idea._
34 import org.make.api.operation.PersistentOperationOfQuestionServiceComponent
35 import org.make.api.post.PostServiceComponent
36 import org.make.api.technical.job.JobCoordinatorServiceComponent
37 import org.make.api.technical.job.JobActor.Protocol.Response.JobAcceptance
38 import org.make.api.technical.{ActorSystemComponent, ReadJournalComponent, TimeSettings}
39 import org.make.core.elasticsearch.IndexationStatus
40 import org.make.core.technical.Pagination
41 import org.make.core.idea.Idea
42 import org.make.core.job.Job.JobId.{Reindex, ReindexPosts}
43 import org.make.core.operation.indexed.OperationOfQuestionSearchResult
44 import org.make.core.operation.{
45   OperationKind,
46   OperationKindsSearchFilter,
47   OperationOfQuestion,
48   OperationOfQuestionSearchFilters,
49   OperationOfQuestionSearchQuery,
50   StatusSearchFilter
51 }
52 import org.make.core.proposal.ProposalId
53 import org.make.core.user.User
54 
55 import scala.concurrent.ExecutionContext.Implicits.global
56 import scala.concurrent.Future
57 import scala.util.{Failure, Success}
58 
59 trait IndexationComponent {
60   def indexationService: IndexationService
61 }
62 
63 sealed trait EntitiesToIndex
64 case object IndexIdeas extends EntitiesToIndex
65 case object IndexOrganisations extends EntitiesToIndex
66 case object IndexProposals extends EntitiesToIndex
67 case object IndexOperationOfQuestions extends EntitiesToIndex
68 case object IndexPost extends EntitiesToIndex
69 
70 trait IndexationService {
71   def reindexData(
72     forceIdeas: Boolean,
73     forceOrganisations: Boolean,
74     forceProposals: Boolean,
75     forceOperationOfQuestions: Boolean
76   ): Future[JobAcceptance]
77   def reindexPostsData(): Future[JobAcceptance]
78   def indicesToReindex(
79     forceIdeas: Boolean,
80     forceOrganisations: Boolean,
81     forceProposals: Boolean,
82     forceOperationOfQuestions: Boolean
83   ): Future[Set[EntitiesToIndex]]
84 }
85 
86 //TODO: test this component
87 trait DefaultIndexationComponent
88     extends IndexationComponent
89     with IdeaIndexationStream
90     with OrganisationIndexationStream
91     with ProposalIndexationStream
92     with OperationOfQuestionIndexationStream
93     with PostIndexationStream {
94 
95   this: ActorSystemComponent
96     with ElasticsearchConfigurationComponent
97     with ElasticsearchClientComponent
98     with Logging
99     with JobCoordinatorServiceComponent
100     with ReadJournalComponent
101     with PersistentIdeaServiceComponent
102     with PersistentOperationOfQuestionServiceComponent
103     with PostServiceComponent =>
104 
105   override lazy val indexationService: IndexationService = new DefaultIndexationService
106 
107   class DefaultIndexationService extends IndexationService {
108 
109     private lazy val client = elasticsearchClient.client
110 
111     private implicit val timeout: Timeout = TimeSettings.defaultTimeout
112 
113     private def reindexOrganisationsIfNeeded(needsReindex: Boolean): Future[Done] = {
114       if (needsReindex) {
115         logger.info("Reindexing organisations")
116         val organisationIndexName =
117           elasticsearchClient.createIndexName(elasticsearchConfiguration.organisationAliasName)
118         for {
119           _      <- executeCreateIndex(elasticsearchConfiguration.organisationAliasName, organisationIndexName)
120           result <- executeIndexOrganisations(organisationIndexName)
121           _      <- executeSetAlias(elasticsearchConfiguration.organisationAliasName, organisationIndexName)
122         } yield result
123       } else {
124         Future.successful(Done)
125       }
126     }
127 
128     private def reindexIdeasIfNeeded(needsReindex: Boolean): Future[Done] = {
129       if (needsReindex) {
130         logger.info("Reindexing ideas")
131         val ideaIndexName = elasticsearchClient.createIndexName(elasticsearchConfiguration.ideaAliasName)
132         for {
133           _           <- executeCreateIndex(elasticsearchConfiguration.ideaAliasName, ideaIndexName)
134           resultIdeas <- executeIndexIdeas(ideaIndexName)
135           _           <- executeSetAlias(elasticsearchConfiguration.ideaAliasName, ideaIndexName)
136         } yield resultIdeas
137       } else {
138         Future.successful(Done)
139       }
140     }
141 
142     private def reindexProposalsIfNeeded(needsReindex: Boolean): Future[Done] = {
143       if (needsReindex) {
144         logger.info("Reindexing proposals")
145         val proposalIndexName =
146           elasticsearchClient.createIndexName(elasticsearchConfiguration.proposalAliasName)
147         for {
148           _               <- executeCreateIndex(elasticsearchConfiguration.proposalAliasName, proposalIndexName)
149           resultProposals <- executeIndexProposal(proposalIndexName)
150           _               <- executeSetAlias(elasticsearchConfiguration.proposalAliasName, proposalIndexName)
151         } yield resultProposals
152       } else {
153         Future.successful(Done)
154       }
155     }
156 
157     private def reindexOperationOfQuestionsIfNeeded(needsReindex: Boolean): Future[Done] = {
158       if (needsReindex) {
159         logger.info("Reindexing operation of questions")
160         val operationOfQuestionIndexName =
161           elasticsearchClient.createIndexName(elasticsearchConfiguration.operationOfQuestionAliasName)
162         for {
163           _      <- executeCreateIndex(elasticsearchConfiguration.operationOfQuestionAliasName, operationOfQuestionIndexName)
164           result <- executeIndexOperationOfQuestions(operationOfQuestionIndexName)
165           _      <- executeSetAlias(elasticsearchConfiguration.operationOfQuestionAliasName, operationOfQuestionIndexName)
166         } yield result
167       } else {
168         Future.successful(Done)
169       }
170     }
171 
172     override def reindexData(
173       forceIdeas: Boolean,
174       forceOrganisations: Boolean,
175       forceProposals: Boolean,
176       forceOperationOfQuestions: Boolean
177     ): Future[JobAcceptance] =
178       jobCoordinatorService.start(Reindex) { report =>
179         logger.info(s"Elasticsearch Reindexation")
180         indicesToReindex(forceIdeas, forceOrganisations, forceProposals, forceOperationOfQuestions).flatMap {
181           indicesNotUpToDate =>
182             // !mandatory: run question indexation before organisation and not in parallel
183             reindexOperationOfQuestionsIfNeeded(indicesNotUpToDate.contains(IndexOperationOfQuestions))
184               .map(_ => indicesNotUpToDate)
185               .recoverWith {
186                 case e =>
187                   logger.error("Questions indexation failed", e)
188                   Future.successful(indicesNotUpToDate)
189               }
190         }.flatMap { indicesNotUpToDate =>
191           // !mandatory: run organisation indexation before proposals and not in parallel
192           reindexOrganisationsIfNeeded(indicesNotUpToDate.contains(IndexOrganisations))
193             .map(_ => indicesNotUpToDate)
194             .recoverWith {
195               case e =>
196                 logger.error("Organisations indexation failed", e)
197                 Future.successful(indicesNotUpToDate)
198             }
199         }.flatMap { indicesNotUpToDate =>
200           report(25d).map(_ => indicesNotUpToDate)
201         }.flatMap { indicesNotUpToDate =>
202           val futureIdeasIndexation: Future[Done] = reindexIdeasIfNeeded(indicesNotUpToDate.contains(IndexIdeas))
203           val futureProposalsIndexation = reindexProposalsIfNeeded(indicesNotUpToDate.contains(IndexProposals))
204 
205           for {
206             _ <- futureIdeasIndexation
207             _ <- futureProposalsIndexation
208           } yield indicesNotUpToDate
209         }.flatMap { toUpdate =>
210           postReindex(toUpdate.contains(IndexProposals) || toUpdate.contains(IndexOperationOfQuestions))
211         }
212       }
213 
214     def postReindex(updateQuestion: Boolean): Future[Unit] = {
215       if (updateQuestion) {
216         executeUpdateOperationOfQuestions().void
217       } else {
218         Future.unit
219       }
220     }
221 
222     override def reindexPostsData(): Future[JobAcceptance] =
223       jobCoordinatorService.start(ReindexPosts) { report =>
224         logger.info(s"Elasticsearch Posts Reindexation")
225         val postIndexName = elasticsearchClient.createIndexName(elasticsearchConfiguration.postAliasName)
226         for {
227           _ <- executeCreateIndex(elasticsearchConfiguration.postAliasName, postIndexName)
228           _ <- report(33d)
229           _ <- executeIndexPosts(postIndexName)
230           _ <- report(66d)
231           _ <- executeSetAlias(elasticsearchConfiguration.postAliasName, postIndexName)
232         } yield {}
233       }
234 
235     override def indicesToReindex(
236       forceIdeas: Boolean,
237       forceOrganisations: Boolean,
238       forceProposals: Boolean,
239       forceOperationOfQuestions: Boolean
240     ): Future[Set[EntitiesToIndex]] = {
241       val hashes: Map[EntitiesToIndex, String] = Map(
242         IndexIdeas -> elasticsearchClient.hashForAlias(elasticsearchConfiguration.ideaAliasName),
243         IndexOrganisations -> elasticsearchClient.hashForAlias(elasticsearchConfiguration.organisationAliasName),
244         IndexProposals -> elasticsearchClient.hashForAlias(elasticsearchConfiguration.proposalAliasName),
245         IndexOperationOfQuestions -> elasticsearchClient.hashForAlias(
246           elasticsearchConfiguration.operationOfQuestionAliasName
247         )
248       )
249       elasticsearchClient.getCurrentIndicesName.map { currentIndices =>
250         val currentHashes: Seq[String] = currentIndices.map(elasticsearchClient.getHashFromIndex)
251         var result: Set[EntitiesToIndex] = Set.empty
252         if (forceIdeas) {
253           result += IndexIdeas
254         }
255         if (forceOrganisations) {
256           result += IndexOrganisations
257         }
258         if (forceProposals) {
259           result += IndexProposals
260         }
261         if (forceOperationOfQuestions) {
262           result += IndexOperationOfQuestions
263         }
264         result ++= hashes.flatMap {
265           case (entitiesToIndex, hash) if !currentHashes.contains(hash) => Some(entitiesToIndex)
266           case _                                                        => None
267         }.toSet
268 
269         result
270       }
271     }
272 
273     private def addAndRemoveAlias(
274       aliasName: String,
275       newIndexName: String,
276       indexes: Seq[String]
277     ): Future[AliasActionResponse] = {
278       if (indexes.isEmpty) {
279         logger.error("indexes with alias is empty")
280       }
281 
282       client.executeAsFuture {
283         aliases(addAlias(aliasName, newIndexName), indexes.map(removeAlias(aliasName, _)): _*)
284       }
285     }
286 
287     private def executeSetAlias(aliasName: String, indexName: String): Future[Done] = {
288 
289       logger.info(s"calling executeSetAlias for $aliasName - $indexName")
290 
291       client.executeAsFuture {
292         getAliases(Seq.empty, Seq(aliasName))
293       }.flatMap { getAliasResponse =>
294         addAndRemoveAlias(aliasName, indexName, getAliasResponse.mappings.keys.map(_.name).toSeq).map(_ => Done)
295       }.recoverWith {
296         case e =>
297           logger.error("fail to retrieve ES alias", e)
298           Future.successful(Done)
299       }
300     }
301 
302     private def executeCreateIndex(aliasName: String, indexName: String): Future[CreateIndexResponse] = {
303       client.executeAsFuture(createIndex(indexName).source(elasticsearchClient.mappingForAlias(aliasName)))
304     }
305 
306     private def executeIndexProposal(proposalIndexName: String): Future[Done] = {
307       val start = System.currentTimeMillis()
308 
309       val result =
310         proposalJournal
311           .currentPersistenceIds()
312           .map(ProposalId.apply)
313           .via(ProposalStream.flowIndexProposals(proposalIndexName))
314           .runWith(Sink.ignore)
315 
316       result.onComplete {
317         case Success(_) =>
318           val time = System.currentTimeMillis() - start
319           logger.info(s"Proposal indexation success in $time ms")
320         case Failure(e) =>
321           val time = System.currentTimeMillis() - start
322           logger.error(s"Proposal indexation failed in $time ms", e)
323       }
324 
325       result
326     }
327 
328     private def executeIndexIdeas(indexName: String): Future[Done] = {
329       val start = System.currentTimeMillis()
330 
331       val result = persistentIdeaService
332         .findAll(IdeaFiltersRequest.empty)
333         .flatMap { ideas =>
334           logger.info(s"Ideas to index: ${ideas.size}")
335           Source[Idea](ideas)
336             .via(IdeaStream.flowIndexIdeas(indexName))
337             .runWith(Sink.ignore)
338         }
339 
340       result.onComplete {
341         case Success(_) =>
342           val time = System.currentTimeMillis() - start
343           logger.info(s"Idea indexation success in $time ms")
344         case Failure(e) =>
345           val time = System.currentTimeMillis() - start
346           logger.error(s"Idea indexation failed in $time ms", e)
347       }
348 
349       result
350     }
351 
352     private def executeIndexOrganisations(indexName: String): Future[Done] = {
353       val start = System.currentTimeMillis()
354 
355       val result = persistentUserService
356         .findAllOrganisations()
357         .flatMap { organisations =>
358           logger.info(s"Organisations to index: ${organisations.size}")
359           Source[User](organisations)
360             .via(OrganisationStream.flowIndexOrganisations(indexName))
361             .runWith(Sink.ignore)
362         }
363 
364       result.onComplete {
365         case Success(_) =>
366           val time = System.currentTimeMillis() - start
367           logger.info(s"Organisation indexation success in $time ms")
368         case Failure(e) =>
369           val time = System.currentTimeMillis() - start
370           logger.error(s"Organisation indexation failed in $time ms", e)
371       }
372 
373       result
374     }
375 
376     private def executeIndexOperationOfQuestions(indexName: String): Future[Done] = {
377       val start = System.currentTimeMillis()
378 
379       val result = persistentOperationOfQuestionService
380         .find()
381         .flatMap { operationOfQuestions =>
382           logger.info(s"Operation of questions to index: ${operationOfQuestions.size}")
383           Source[OperationOfQuestion](operationOfQuestions)
384             .via(OperationOfQuestionStream.flowIndexOperationOfQuestions(indexName))
385             .runWith(Sink.ignore)
386         }
387 
388       result.onComplete {
389         case Success(_) =>
390           val time = System.currentTimeMillis() - start
391           logger.info(s"Operation of questions indexation success in $time ms")
392         case Failure(e) =>
393           val time = System.currentTimeMillis() - start
394           logger.error(s"Operation of questions indexation failed in $time ms", e)
395       }
396 
397       result
398     }
399 
400     private def executeUpdateOperationOfQuestions(): Future[IndexationStatus] = {
401       val start = System.currentTimeMillis()
402 
403       def query(limit: Pagination.Limit) = OperationOfQuestionSearchQuery(
404         filters = Some(
405           OperationOfQuestionSearchFilters(
406             status = Some(StatusSearchFilter(OperationOfQuestion.Status.Open)),
407             operationKinds = Some(OperationKindsSearchFilter(OperationKind.values))
408           )
409         ),
410         limit = Some(limit)
411       )
412 
413       @SuppressWarnings(Array("org.wartremover.warts.Recursion", "org.wartremover.warts.ThreadSleep"))
414       def countProposals(retries: Int = 3): Future[Int] = {
415         elasticsearchOperationOfQuestionAPI.count(OperationOfQuestionSearchQuery()).flatMap {
416           case 0 if retries > 0 =>
417             // Needed to wait for previous bulk indexation to actually persist in ES
418             Thread.sleep(1000)
419             countProposals(retries - 1)
420           case 0 =>
421             logger.warn("retrying but no retries left")
422             Future.failed(
423               new IllegalStateException(
424                 s"ES index ${elasticsearchConfiguration.operationOfQuestionAliasName} does not yet have indexed values."
425               )
426             )
427           case _ =>
428             elasticsearchOperationOfQuestionAPI.count(query(Pagination.Limit(0))).map(_.toInt)
429         }
430       }
431 
432       def updateQuestions(result: OperationOfQuestionSearchResult): Future[IndexationStatus] = {
433         logger.info(s"Operation of questions to update: ${result.results.size}")
434         result.results.map(_.questionId) match {
435           case Seq() => Future.successful(IndexationStatus.Completed)
436           case head +: tail =>
437             elasticsearchProposalAPI.computeTop20ConsensusThreshold(NonEmptyList(head, tail.toList)).flatMap {
438               thresholds =>
439                 val questions = result.results.map { question =>
440                   question.copy(top20ConsensusThreshold = thresholds.get(question.questionId))
441                 }
442                 elasticsearchOperationOfQuestionAPI.indexOperationOfQuestions(questions, None)
443             }
444         }
445       }
446 
447       val result = for {
448         count  <- countProposals()
449         result <- elasticsearchOperationOfQuestionAPI.searchOperationOfQuestions(query(Pagination.Limit(count)))
450         status <- updateQuestions(result)
451       } yield status
452 
453       result.onComplete {
454         case Success(IndexationStatus.Completed) =>
455           logger.info(s"Operation of questions post reindex update success in ${System.currentTimeMillis() - start} ms")
456         case Success(IndexationStatus.Failed(e)) =>
457           logger.error(
458             s"Operation of questions post reindex update failed in ${System.currentTimeMillis() - start} ms",
459             e
460           )
461         case Failure(e) =>
462           logger.error(
463             s"Operation of questions post reindex update failed in ${System.currentTimeMillis() - start} ms",
464             e
465           )
466       }
467 
468       result
469     }
470 
471     private def executeIndexPosts(indexName: String): Future[Done] = {
472       val start = System.currentTimeMillis()
473 
474       val result = Source
475         .future(postService.fetchPostsForHome())
476         .wireTap { posts =>
477           logger.info(s"Posts to index: ${posts.size}")
478         }
479         .mapConcat(identity)
480         .via(PostStream.flowIndexPosts(indexName))
481         .runWith(Sink.ignore)
482 
483       result.onComplete {
484         case Success(_) =>
485           val time = System.currentTimeMillis() - start
486           logger.info(s"Posts indexation success in $time ms")
487         case Failure(e) =>
488           val time = System.currentTimeMillis() - start
489           logger.error(s"Posts indexation failed in $time ms", e)
490       }
491 
492       result
493     }
494   }
495 }