Adds recipe indexer (#337)

* moves capi recipe indexing logic to common

* update test imports

* adds indexer sbt project

* slightly more helpful error message

* Remove RecipeStatusPendingCuration from list of curatedStatuses

build.sbt

@@ -33,7 +33,7 @@ lazy val flywaySettings = Seq(
 )
 
 lazy val root = (project in file("."))
-  .aggregate(ui, common, etl)
+  .aggregate(ui, common, etl, indexer)
 
 def env(key: String): Option[String] = Option(System.getenv(key))
 
@@ -71,7 +71,9 @@ lazy val common = (project in file("common"))
   .settings(Seq(
       libraryDependencies ++= Seq(
           "org.jsoup" % "jsoup" % "1.9.2",
-          "com.gu" %% "content-api-client" % "10.24"
+          "com.gu" %% "content-api-client" % "10.24",
+          "com.typesafe.play" %% "play-ws" % "2.5.4",
+          "com.amazonaws" % "aws-java-sdk-sts" % "1.10.74"
       )
   ))
 
@@ -86,6 +88,13 @@ lazy val etl = (project in file("etl"))
       cancelable in Global := true
   ))
 
+lazy val indexer = (project in file("indexer"))
+  .settings(commonSettings)
+  .dependsOn(common)
+  .settings(Seq(
+      cancelable in Global := true
+  ))
+
 
 initialize := {
   val _ = initialize.value

common/src/main/scala/com/gu/recipeasy/db/DB.scala

@@ -141,6 +141,11 @@ class DB(contextWrapper: ContextWrapper) {
     contextWrapper.dbContext.run(action)
   }
 
+  val curatedStatuses = List(RecipeStatusVerified, RecipeStatusCurated, RecipeStatusFinalised, RecipeStatusPendingVerification, RecipeStatusPendingFinalisation).map(_.name)
+  def getCuratedRecipesIds(): List[String] = {
+    contextWrapper.dbContext.run(quote(query[Recipe]).filter(r => liftQuery(curatedStatuses).contains(r.status)).map(r => r.id))
+  }
+
   // ---------------------------------------------
   // Original Recipes
 

ui/app/com/gu/recipeasy/services/AtomPublisher.scala → common/src/main/scala/com/gu/recipeasy/services/AtomPublisher.scala

@@ -1,4 +1,4 @@
-package services
+package com.gu.recipeasy.services
 
 import com.amazonaws.services.kinesis.model.PutRecordRequest
 import com.amazonaws.services.kinesis.model.PutRecordResult

ui/app/com/gu/recipeasy/services/PublisherConfig.scala → common/src/main/scala/com/gu/recipeasy/services/PublisherConfig.scala

@@ -1,4 +1,4 @@
-package services
+package com.gu.recipeasy.services
 
 import com.amazonaws.auth.{ AWSCredentialsProviderChain, STSAssumeRoleSessionCredentialsProvider }
 import com.amazonaws.auth.profile.ProfileCredentialsProvider

ui/app/com/gu/recipeasy/services/RecipePublisher.scala → common/src/main/scala/com/gu/recipeasy/services/RecipePublisher.scala

@@ -1,4 +1,4 @@
-package services
+package com.gu.recipeasy.services
 
 import java.time.OffsetDateTime
 
@@ -7,10 +7,9 @@ import com.gu.auxiliaryatom.model.auxiliaryatomevent.v1.{ AuxiliaryAtom, Auxilia
 import com.gu.contentatom.thrift._
 import com.gu.recipeasy.db.DB
 import com.gu.recipeasy.models.{ CuratedRecipe, Recipe }
-import com.gu.recipeasy.services.ContentApi
 
 import scala.concurrent.Future
-import scala.util.{ Failure, Success, Try }
+import scala.util.Try
 import scala.concurrent.ExecutionContext.Implicits.global
 
 object RecipePublisher {
@@ -50,4 +49,4 @@ object RecipePublisher {
     AtomPublisher.send(atomEvents)(config)
   }
 
-}
+}

ui/app/com/gu/recipeasy/services/Teleporter.scala → common/src/main/scala/com/gu/recipeasy/services/Teleporter.scala

@@ -1,4 +1,4 @@
-package services
+package com.gu.recipeasy.services
 
 import play.api.libs.ws.WSClient
 import scala.concurrent.ExecutionContext

ui/app/com/gu/recipeasy/services/ThriftSerializer.scala → common/src/main/scala/com/gu/recipeasy/services/ThriftSerializer.scala

@@ -1,4 +1,4 @@
-package services
+package com.gu.recipeasy.services
 
 import java.io.ByteArrayOutputStream
 import java.nio.ByteBuffer

indexer/.gitignore

@@ -0,0 +1,2 @@
+data/
+application.conf

indexer/src/main/resources/application.conf.template

@@ -0,0 +1,16 @@
+aws {
+    atom.auxiliary.stsRoleArn=""
+    atom.content.stsRoleArn=""
+}
+
+capi.key=""
+
+db {
+  ctx.dataSourceClassName=org.postgresql.ds.PGSimpleDataSource
+  ctx.dataSource.user=recipeasy
+  ctx.dataSource.password=""
+  ctx.dataSource.databaseName=recipeasy
+  ctx.dataSource.portNumber=7777
+  ctx.dataSource.serverName=localhost
+  ctx.connectionTimeout=30000
+}

indexer/src/main/resources/logback.xml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
+
+    <root level="INFO">
+        <appender-ref ref="STDOUT"/>
+    </root>
+
+    <logger name="com.ning.http.client" level="WARN"/>
+</configuration>

indexer/src/main/scala/indexer/Indexer.scala

@@ -0,0 +1,53 @@
+package indexer
+
+import akka.actor.ActorSystem
+import akka.stream.ActorMaterializer
+import com.amazonaws.regions.{ Region, Regions }
+import com.amazonaws.services.kinesis.model.PutRecordResult
+import com.gu.contentapi.client.GuardianContentClient
+import com.gu.recipeasy.db.{ ContextWrapper, DB }
+import com.gu.recipeasy.services._
+import com.typesafe.config.ConfigFactory
+import play.api.Configuration
+import play.api.libs.ws.ahc.AhcWSClient
+
+import scala.concurrent.Future
+import scala.util.{ Failure, Success, Try }
+import scala.concurrent.ExecutionContext.Implicits.global
+
+object Indexer extends App {
+
+  if (args.isEmpty) {
+    Console.err.println("Usage: Indexer <STAGE>")
+    sys.exit(1)
+  }
+
+  val stage = args(0)
+  val contextWrapper = new ContextWrapper { val config = ConfigFactory.load() }
+  val config = new Configuration(ConfigFactory.load())
+  val db = new DB(contextWrapper)
+
+  implicit val actorSystem = ActorSystem()
+  implicit val materializer = ActorMaterializer()
+  val teleporter = new Teleporter(AhcWSClient())
+  val region = Region.getRegion(Regions.fromName(Regions.EU_WEST_1.getName))
+
+  val publisherConfig = PublisherConfig(config, region, stage)
+
+  val contentApiClient = new ContentApi(contentApiClient = new GuardianContentClient(publisherConfig.contentAtomConfig.capiKey))
+
+  val curatedRecipeIds: List[String] = db.getCuratedRecipesIds()
+
+  curatedRecipeIds.foreach { r =>
+    val fresult: Future[Try[List[PutRecordResult]]] = RecipePublisher.publishRecipe(r, db, publisherConfig, teleporter, contentApiClient)
+    fresult.map { result =>
+      result match {
+        case Success(_) => println(s"Successfully published $r")
+        case Failure(error) => println(s"Failed to published $r")
+      }
+    }
+
+    Thread.sleep(1000)
+  }
+
+}

reindex.sh

@@ -0,0 +1,30 @@
+#!/bin/bash
+#
+# Connects localhost to PROD database and reindexes
+# Requires application.conf to be correctly populated
+
+aws rds describe-db-instances 1>/dev/null
+if [ "$?" -eq 255 ]
+then
+    echo You need to have valid capi and composer AWS credentials
+    exit
+else
+    rds_host=$(aws rds describe-db-instances --db-instance-identifier recipeasy-rds-primary-prod | jq -r .DBInstances[].Endpoint.Address 2>/dev/null)
+    ec2=$(marauder -s stage=PROD stack=content-api-recipeasy app=recipeasy 2>/dev/null)
+    stage=${1?Stage missing, use CODE or PROD}
+
+    echo Setting up SSH tunnel
+    ssh -o StrictHostKeyChecking=no -N -L 7777:${rds_host}:5432 ubuntu@${ec2} &
+    SSH_PID=$!
+
+    trap finish SIGINT EXIT
+    function finish {
+        kill $SSH_PID
+    }
+
+    sleep 3
+
+    echo Running reindex
+
+    sbt "indexer/run $stage"
+fi

ui/app/com/gu/recipeasy/AppComponents.scala

@@ -5,7 +5,7 @@ import com.amazonaws.regions.{ Region, Regions }
 import com.gu.cm.{ ConfigurationLoader, Identity }
 import com.gu.contentapi.client.GuardianContentClient
 import com.gu.recipeasy.db.{ ContextWrapper, DB }
-import com.gu.recipeasy.services.ContentApi
+import com.gu.recipeasy.services._
 import com.gu.recipeasy.{ KinesisAppenderConfig, LogStash }
 import play.api.ApplicationLoader.Context
 import play.api.i18n.{ DefaultLangs, DefaultMessagesApi, MessagesApi }
@@ -18,8 +18,6 @@ import router.Routes
 import scala.concurrent.Future
 import schedule.DBHouseKeepingScheduler
 import controllers._
-import services._
-import services.PublisherConfig
 
 class AppComponents(context: Context)
     extends BuiltInComponentsFromContext(context)

ui/app/com/gu/recipeasy/controllers/Application.scala

@@ -11,7 +11,7 @@ import com.gu.recipeasy.auth.AuthActions
 import com.gu.recipeasy.db._
 import com.gu.recipeasy.models._
 import com.gu.recipeasy.services.ContentApi
-import services.{ PublisherConfig, RecipePublisher, Teleporter }
+import com.gu.recipeasy.services.{ PublisherConfig, RecipePublisher, Teleporter }
 import com.gu.recipeasy.views
 import models._
 import models.CuratedRecipeForm._

ui/app/com/gu/recipeasy/controllers/Publisher.scala

@@ -1,24 +1,16 @@
 package controllers
 
-import java.time.OffsetDateTime
-
 import com.amazonaws.services.kinesis.model.PutRecordResult
-import com.gu.auxiliaryatom.model.auxiliaryatomevent.v1.{ AuxiliaryAtom, AuxiliaryAtomEvent, EventType => AuxiliaryAtomEventType }
-import com.gu.contentatom.thrift._
 import com.gu.recipeasy.auth.AuthActions
 import com.gu.recipeasy.db.DB
-import com.gu.recipeasy.models.{ CuratedRecipe, Recipe }
-import com.gu.recipeasy.services.ContentApi
+import com.gu.recipeasy.services.{ ContentApi, PublisherConfig, RecipePublisher, Teleporter }
 import com.typesafe.scalalogging.StrictLogging
 import play.api.Configuration
 import play.api.libs.ws.WSClient
 import play.api.mvc.{ Action, Controller }
-import services.{ AtomPublisher, PublisherConfig, Teleporter }
-import services.RecipePublisher
 
 import scala.concurrent.Future
 import scala.util.{ Failure, Success, Try }
-
 import scala.concurrent.ExecutionContext.Implicits.global
 
 class Publisher(override val wsClient: WSClient, override val conf: Configuration, publisherConfig: PublisherConfig, db: DB, teleporter: Teleporter, contentApi: ContentApi) extends Controller with AuthActions with StrictLogging {

ui/test/com/gu/recipeasy/ApplicationSpec.scala

@@ -1,6 +1,4 @@
 import org.scalatest._
-import collection.mutable.Stack
-import controllers._
 import com.gu.recipeasy.models._
 import java.time.OffsetDateTime
 

ui/test/com/gu/recipeasy/services/TeleporterSpec.scala

@@ -1,6 +1,7 @@
 package services
 
-import org.scalatest.{ FlatSpec, Matchers }
+import com.gu.recipeasy.services.Teleporter
+import org.scalatest.{FlatSpec, Matchers}
 import play.api.test.WsTestClient.withClient
 
 import scala.concurrent.Await