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.Flow
23 import akka.NotUsed
24 import com.sksamuel.elastic4s.Index
25 import grizzled.slf4j.Logging
26 import org.make.api.operation.{OperationOfQuestionSearchEngineComponent, OperationServiceComponent}
27 import org.make.api.question.QuestionServiceComponent
28 import org.make.core.elasticsearch.IndexationStatus
29 import org.make.core.operation.indexed.IndexedOperationOfQuestion
30 import org.make.core.operation.{OperationOfQuestion, SimpleOperation}
31 import org.make.core.question.Question
32 
33 import scala.concurrent.ExecutionContext.Implicits.global
34 import scala.concurrent.Future
35 
36 trait OperationOfQuestionIndexationStream
37     extends IndexationStream
38     with OperationOfQuestionSearchEngineComponent
39     with QuestionServiceComponent
40     with OperationServiceComponent
41     with Logging {
42 
43   object OperationOfQuestionStream {
44 
45     def flowIndexOperationOfQuestions(
46       operationOfQuestionIndexName: String
47     ): Flow[OperationOfQuestion, IndexationStatus, NotUsed] =
48       grouped[OperationOfQuestion].via(runIndexOperationOfQuestion(operationOfQuestionIndexName))
49 
50     def runIndexOperationOfQuestion(
51       operationOfQuestionIndexName: String
52     ): Flow[Seq[OperationOfQuestion], IndexationStatus, NotUsed] = {
53       Flow[Seq[OperationOfQuestion]]
54         .mapAsync(singleAsync)(
55           operationOfQuestions => executeIndexOperationOfQuestions(operationOfQuestions, operationOfQuestionIndexName)
56         )
57     }
58 
59     private def executeIndexOperationOfQuestions(
60       operationOfQuestions: Seq[OperationOfQuestion],
61       operationOfQuestionIndexName: String
62     ): Future[IndexationStatus] = {
63 
64       val futureQuestion: Future[Seq[Question]] =
65         questionService.getQuestions(operationOfQuestions.map(_.questionId))
66       val futureOperation: Future[Seq[SimpleOperation]] =
67         operationService.findSimple()
68 
69       val futureIndexedOperationOfQuestion = for {
70         questions  <- futureQuestion
71         operations <- futureOperation
72       } yield (questions, operations)
73 
74       futureIndexedOperationOfQuestion.map {
75         case (questions, operations) =>
76           operationOfQuestions.map { operationOfQuestion =>
77             for {
78               question  <- questions.find(_.questionId == operationOfQuestion.questionId)
79               operation <- operations.find(_.operationId == operationOfQuestion.operationId)
80             } yield IndexedOperationOfQuestion
81               .createFromOperationOfQuestion(operationOfQuestion, operation, question)
82           }
83       }.map(_.flatten).flatMap { operationOfQuestions =>
84         elasticsearchOperationOfQuestionAPI
85           .indexOperationOfQuestions(operationOfQuestions, Some(Index(operationOfQuestionIndexName)))
86       }
87     }
88   }
89 
90 }