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
21 
22 import akka.actor.typed.scaladsl.AskPattern.Askable
23 import akka.actor.typed.{ActorSystem, Scheduler}
24 import akka.cluster.typed.Cluster
25 import akka.http.scaladsl.Http
26 import akka.http.scaladsl.Http.ServerBinding
27 import akka.http.scaladsl.server.Route
28 import com.typesafe.config.{Config, ConfigFactory}
29 import grizzled.slf4j.Logging
30 import kamon.Kamon
31 import org.make.api.MakeGuardian.Initialize
32 import org.make.api.extensions.ThreadPoolMonitoringActor.MonitorThreadPool
33 import org.make.api.extensions.{DatabaseConfigurationExtension, MakeSettingsExtension, ThreadPoolMonitoringActor}
34 import org.make.api.technical.{ClusterShardingMonitor, MemoryMonitoringActor}
35 
36 import java.net.InetAddress
37 import java.nio.file.{Files, Paths}
38 import scala.concurrent.duration.DurationInt
39 import scala.concurrent.{Await, ExecutionContextExecutor, Future}
40 import scala.jdk.CollectionConverters._
41 
42 @SuppressWarnings(
43   Array("org.wartremover.warts.While", "org.wartremover.warts.ThreadSleep", "org.wartremover.warts.ScalaApp")
44 )
45 object MakeMain extends App with Logging with MakeApi {
46 
47   Thread.setDefaultUncaughtExceptionHandler { (thread, exception) =>
48     logger.error(s"Error in thread ${thread.getName}:", exception)
49   }
50 
51   val envName: Option[String] = Option(System.getenv("ENV_NAME"))
52 
53   private val resourceName = envName match {
54     case Some(name) if name.nonEmpty => s"$name-application.conf"
55     case _                           => "default-application.conf"
56   }
57 
58   private val configuration: Config = {
59     val defaultConfiguration = ConfigFactory.load(resourceName)
60     val configurationPath = defaultConfiguration.getString("make-api.secrets-configuration-path")
61     val extraConfigPath = Paths.get(configurationPath)
62     val configWithSecrets = if (Files.exists(extraConfigPath) && !Files.isDirectory(extraConfigPath)) {
63       ConfigFactory.parseFile(extraConfigPath.toFile).resolve().withFallback(defaultConfiguration)
64     } else {
65       defaultConfiguration
66     }
67 
68     val bindAddress = configWithSecrets.getString("akka.remote.artery.canonical.hostname")
69     val resolvedAddress: String = InetAddress.getByName(bindAddress).getHostAddress
70 
71     ConfigFactory
72       .parseMap(Map("akka.remote.artery.canonical.hostname" -> resolvedAddress).asJava)
73       .withFallback(configWithSecrets)
74       .resolve()
75   }
76 
77   override implicit val actorSystem: ActorSystem[MakeGuardian.GuardianCommand] =
78     ActorSystem(MakeGuardian.createBehavior(this), "make-api", configuration)
79 
80   implicit private val scheduler: Scheduler = actorSystem.scheduler
81 
82   // Wait until cluster is connected before initializing clustered actors
83   while (Cluster(actorSystem).state.leader.isEmpty) {
84     Thread.sleep(100)
85   }
86 
87   Kamon.init(configuration)
88 
89   private val databaseConfiguration = actorSystem.registerExtension(DatabaseConfigurationExtension)
90   databaseConfiguration.migrateDatabase()
91 
92   Await.result(elasticsearchClient.initialize(), 10.seconds)
93 
94   Await.result(swiftClient.init(), 10.seconds)
95 
96   Await.result(actorSystem ? Initialize, atMost = 5.seconds)
97 
98   actorSystem.systemActorOf(ClusterShardingMonitor(), ClusterShardingMonitor.name)
99   actorSystem.systemActorOf(MemoryMonitoringActor(), MemoryMonitoringActor.name)
100   private val threadPoolMonitor =
101     actorSystem.systemActorOf(ThreadPoolMonitoringActor(), ThreadPoolMonitoringActor.name)
102   threadPoolMonitor ! MonitorThreadPool(databaseConfiguration.readExecutor, "db-read-pool")
103   threadPoolMonitor ! MonitorThreadPool(databaseConfiguration.writeExecutor, "db-write-pool")
104 
105   private val settings = MakeSettingsExtension(actorSystem)
106 
107   // Ensure database stuff is initialized
108   Await.result(userService.getUserByEmail(settings.defaultAdmin.email), atMost = 20.seconds)
109 
110   implicit val ec: ExecutionContextExecutor = actorSystem.executionContext
111 
112   private val host = settings.Http.host
113   private val port = settings.Http.port
114 
115   val bindingFuture: Future[ServerBinding] =
116     Http().newServerAt(interface = host, port = port).bind(Route.toFunction(makeRoutes))
117 
118   bindingFuture.map { serverBinding =>
119     logger.info(s"Make API bound to ${serverBinding.localAddress} ")
120   }.onComplete {
121     case util.Failure(ex) =>
122       logger.error(s"Failed to bind to $host:$port!", ex)
123       actorSystem.terminate()
124     case _ =>
125   }
126 
127 }