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.{Done, NotUsed}
24 import com.sksamuel.elastic4s.Index
25 import grizzled.slf4j.Logging
26 import org.make.api.idea.IdeaSearchEngineComponent
27 import org.make.api.proposal.ProposalSearchEngineComponent
28 import org.make.core.idea.indexed.IndexedIdea
29 import org.make.core.idea.{Idea, IdeaId}
30 
31 import scala.concurrent.ExecutionContext.Implicits.global
32 import scala.concurrent.Future
33 
34 trait IdeaIndexationStream
35     extends IndexationStream
36     with IdeaSearchEngineComponent
37     with ProposalSearchEngineComponent
38     with Logging {
39   object IdeaStream {
40     def runIndexIdeas(ideaIndexName: String): Flow[Seq[Idea], Done, NotUsed] =
41       Flow[Seq[Idea]].mapAsync(parallelism)(ideas => executeIndexIdeas(ideas, ideaIndexName))
42 
43     def flowIndexIdeas(ideaIndexName: String): Flow[Idea, Done, NotUsed] =
44       grouped[Idea].via(runIndexIdeas(ideaIndexName))
45 
46   }
47 
48   private def executeIndexIdeas(ideas: Seq[Idea], ideaIndexName: String): Future[Done] = {
49     elasticsearchProposalAPI
50       .countProposalsByIdea(ideas.map(_.ideaId))
51       .flatMap { countProposalsByIdea =>
52         def proposalsCount(ideaId: IdeaId): Int = countProposalsByIdea.getOrElse(ideaId, 0L).toInt
53         elasticsearchIdeaAPI
54           .indexIdeas(
55             ideas.map(idea => IndexedIdea.createFromIdea(idea, proposalsCount(idea.ideaId))),
56             Some(Index(ideaIndexName))
57           )
58       }
59       .recoverWith {
60         case e =>
61           logger.error("Indexing ideas failed", e)
62           Future.successful(Done)
63       }
64   }
65 }