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 java.time.format.DateTimeFormatter
23 import cats.implicits._
24 import com.sksamuel.elastic4s.ElasticClient
25 import com.sksamuel.elastic4s.akka.{AkkaHttpClient, AkkaHttpClientSettings}
26 import com.sksamuel.elastic4s.ElasticDsl._
27 import com.sksamuel.elastic4s.requests.indexes.admin.AliasExistsResponse
28 import grizzled.slf4j.Logging
29 import org.make.api.technical.security.SecurityHelper
30 import org.make.api.technical.ActorSystemComponent
31 import org.make.core.DateHelper
32 import akka.actor.typed.scaladsl.adapter._
33 
34 import scala.concurrent.ExecutionContext.Implicits.global
35 import scala.concurrent.Future
36 import scala.io.{Codec, Source}
37 
38 trait ElasticsearchClient {
39   def client: ElasticClient
40   def createIndexName(aliasName: String): String
41   def hashForAlias(aliasName: String): String
42   def getCurrentIndicesName: Future[Seq[String]]
43   def getHashFromIndex(index: String): String
44   def mappingForAlias: String => String
45   def initialize(): Future[Unit]
46 }
47 
48 trait ElasticsearchClientComponent {
49   def elasticsearchClient: ElasticsearchClient
50 }
51 
52 trait DefaultElasticsearchClientComponent extends ElasticsearchClientComponent with Logging {
53   self: ElasticsearchConfigurationComponent with ActorSystemComponent =>
54 
55   override lazy val elasticsearchClient: ElasticsearchClient = new DefaultElasticsearchClient
56 
57   class DefaultElasticsearchClient extends ElasticsearchClient {
58 
59     private lazy val allAliases =
60       Seq(
61         elasticsearchConfiguration.ideaAliasName,
62         elasticsearchConfiguration.proposalAliasName,
63         elasticsearchConfiguration.organisationAliasName,
64         elasticsearchConfiguration.operationOfQuestionAliasName,
65         elasticsearchConfiguration.postAliasName
66       )
67 
68     // create index
69     lazy val elasticsearchIdeaMapping: String =
70       Source.fromResource("elasticsearch-mappings/idea.json")(Codec.UTF8).getLines().mkString("")
71     lazy val elasticsearchProposalMapping: String =
72       Source.fromResource("elasticsearch-mappings/proposal.json")(Codec.UTF8).getLines().mkString("")
73     lazy val elasticsearchOrganisationMapping: String =
74       Source.fromResource("elasticsearch-mappings/organisation.json")(Codec.UTF8).getLines().mkString("")
75     lazy val elasticsearchOperationOfQuestionMapping: String =
76       Source.fromResource("elasticsearch-mappings/operation-of-question.json")(Codec.UTF8).getLines().mkString("")
77     lazy val elasticsearchPostMapping: String =
78       Source.fromResource("elasticsearch-mappings/post.json")(Codec.UTF8).getLines().mkString("")
79 
80     override lazy val client: ElasticClient = ElasticClient(
81       AkkaHttpClient(AkkaHttpClientSettings(elasticsearchConfiguration.connectionString.split(",").toSeq))(
82         actorSystem.toClassic
83       )
84     )
85 
86     private val dateFormatter = DateTimeFormatter.ofPattern("yyyyMMddHHmmss")
87 
88     override def mappingForAlias: String => String = {
89       case alias if alias == elasticsearchConfiguration.ideaAliasName         => elasticsearchIdeaMapping
90       case alias if alias == elasticsearchConfiguration.proposalAliasName     => elasticsearchProposalMapping
91       case alias if alias == elasticsearchConfiguration.organisationAliasName => elasticsearchOrganisationMapping
92       case alias if alias == elasticsearchConfiguration.operationOfQuestionAliasName =>
93         elasticsearchOperationOfQuestionMapping
94       case alias if alias == elasticsearchConfiguration.postAliasName => elasticsearchPostMapping
95     }
96 
97     override def hashForAlias(aliasName: String): String = {
98       val shortHashLength = 12
99       val localMapping: String = mappingForAlias(aliasName)
100       SecurityHelper.defaultHash(localMapping).take(shortHashLength).toLowerCase()
101     }
102 
103     override def createIndexName(aliasName: String): String = {
104       Seq(
105         elasticsearchConfiguration.indexName,
106         aliasName,
107         dateFormatter.format(DateHelper.now()),
108         hashForAlias(aliasName)
109       ).mkString("-")
110     }
111 
112     override def getHashFromIndex(index: String): String =
113       index.split("-").lastOption.getOrElse("")
114 
115     override def getCurrentIndicesName: Future[Seq[String]] = {
116       client
117         .executeAsFuture(getAliases(Seq.empty, allAliases))
118         .map(_.mappings.map { case (index, _) => index.name }.toSeq)
119         .recover {
120           case e: Exception =>
121             logger.error("fail to retrieve ES alias", e)
122             Seq.empty
123         }
124     }
125 
126     private def createInitialIndexAndAlias(aliasName: String): Future[Unit] = {
127       val newIndexName = createIndexName(aliasName)
128       client
129         .executeAsFuture(createIndex(newIndexName).source(mappingForAlias(aliasName)))
130         .flatMap { _ =>
131           logger.info(s"Elasticsearch index $newIndexName created")
132           client
133             .executeAsFuture(aliases(addAlias(aliasName, newIndexName)))
134             .map { _ =>
135               logger.info(s"Elasticsearch alias $aliasName created")
136             }
137         }
138     }
139 
140     override def initialize(): Future[Unit] = {
141       Future
142         .traverse(allAliases) { aliasName =>
143           client.executeAsFuture(aliasExists(aliasName)).flatMap {
144             case AliasExistsResponse(true) =>
145               logger.info(s"Elasticsearch alias $aliasName exist")
146               Future.unit
147             case AliasExistsResponse(false) =>
148               logger.info(s"Elasticsearch alias $aliasName not found")
149               createInitialIndexAndAlias(aliasName)
150           }
151         }
152         .void
153     }
154   }
155 }