Merge pull request #43 from Hub-of-all-Things/v2.6.1

V2.6.1 all checks passed

Author: AndriusA

hat/app/org/hatdex/hat/api/controllers/Applications.scala

@@ -72,7 +72,8 @@ class Applications @Inject() (
   }
 
   def applicationStatus(id: String): Action[AnyContent] = SecuredAction(ContainsApplicationRole(Owner(), ApplicationManage(id)) || WithRole(Owner())).async { implicit request =>
-    val bustCache = request.headers.get("Cache-Control").exists(_ == "no-cache")
+    val bustCache = request.headers.get("Cache-Control").contains("no-cache")
+    logger.info(s"Getting app $id status (bust cache: $bustCache)")
     applicationsService.applicationStatus(id, bustCache).map { maybeStatus ⇒
       maybeStatus map { status ⇒
         Ok(Json.toJson(status))

hat/app/org/hatdex/hat/api/controllers/RichData.scala

@@ -85,13 +85,15 @@ class RichData @Inject() (
       val dataEndpoint = s"$namespace/$endpoint"
       val response = request.body match {
         case array: JsArray =>
-          val values = array.value.map(EndpointData(dataEndpoint, None, _, None))
+          // TODO: extract unique ID and timestamp
+          val values = array.value.map(EndpointData(dataEndpoint, None, None, None, _, None))
           dataService.saveData(request.identity.userId, values, skipErrors.getOrElse(false))
             .andThen(dataEventDispatcher.dispatchEventDataCreated(s"saved batch for $dataEndpoint"))
             .map(saved => Created(Json.toJson(saved)))
 
         case value: JsValue =>
-          val values = Seq(EndpointData(dataEndpoint, None, value, None))
+          // TODO: extract unique ID and timestamp
+          val values = Seq(EndpointData(dataEndpoint, None, None, None, value, None))
           dataService.saveData(request.identity.userId, values)
             .andThen(dataEventDispatcher.dispatchEventDataCreated(s"saved data for $dataEndpoint"))
             .map(saved => Created(Json.toJson(saved.head)))

hat/app/org/hatdex/hat/api/service/MigrationService.scala

@@ -25,12 +25,12 @@
 package org.hatdex.hat.api.service
 
 import java.util.UUID
-import javax.inject.Inject
 
 import akka.NotUsed
 import akka.stream.Materializer
 import akka.stream.scaladsl.{ Flow, Keep, RunnableGraph, Sink, Source }
 import com.mohiva.play.silhouette.api.actions.SecuredRequest
+import javax.inject.Inject
 import org.hatdex.hat.api.json.HatJsonFormats
 import org.hatdex.hat.api.models.{ ApiDataRecord, _ }
 import org.hatdex.hat.api.service.monitoring.HatDataEventDispatcher
@@ -84,7 +84,7 @@ class MigrationService @Inject() (
     record.tables.map { tables =>
       val savedRecords = tables.map { table =>
         convertRecordJson(record, includeTimestamp = true, Some(table.name)) map { data =>
-          richDataService.saveData(userId, Seq(EndpointData(s"${table.source}/${table.name}", None, data, None)))
+          richDataService.saveData(userId, Seq(EndpointData(s"${table.source}/${table.name}", None, None, None, data, None)))
             .andThen(dataEventDispatcher.dispatchEventDataCreated(s"migrated record for ${table.source}/${table.name}"))
             .map(_ => 1)
         } getOrElse Future.successful(0)
@@ -116,7 +116,7 @@ class MigrationService @Inject() (
       } via {
         Flow[JsResult[JsObject]].map(_.get) // Unwrap Json Objects
       } via Flow[JsObject].mapAsync(parallelMigrations) { oldJson =>
-        richDataService.saveData(userId, Seq(EndpointData(endpoint, None, oldJson, None)))
+        richDataService.saveData(userId, Seq(EndpointData(endpoint, None, None, None, oldJson, None)))
           .andThen(dataEventDispatcher.dispatchEventDataCreated(s"migrated data to $endpoint"))
           .map(_ => 1L)
           .recover {

hat/app/org/hatdex/hat/api/service/UsersService.scala

@@ -174,14 +174,14 @@ class UsersService @Inject() (cache: AsyncCacheApi)(implicit ec: DalExecutionCon
   }
 
   def previousLogin(user: HatUser)(implicit server: HatServer): Future[Option[HatAccessLog]] = {
-    logger.error(s"Getting previous login for $user")
+    logger.debug(s"Getting previous login for $user@${server.domain}")
     val query = for {
       access <- UserAccessLog.filter(l => l.userId === user.userId).sortBy(_.date.desc).take(2).drop(1)
       user <- access.userUserFk
     } yield (access, user)
     server.db.run(query.result)
       .andThen {
-        case Success(h) ⇒ logger.error(s"Got previous logins $h")
+        case Success(h) ⇒ logger.error(s"Got previous logins $h @${server.domain}")
       }
       .map(_.headOption)
       .map(_.map(au => ModelTranslation.fromDbModel(au._1, ModelTranslation.fromDbModel(au._2))))

hat/app/org/hatdex/hat/api/service/applications/ApplicationsService.scala

@@ -24,8 +24,9 @@
 package org.hatdex.hat.api.service.applications
 
 import akka.Done
-import javax.inject.Inject
+import akka.actor.ActorSystem
 import com.mohiva.play.silhouette.api.Silhouette
+import javax.inject.Inject
 import org.hatdex.hat.api.models.applications.{ Application, ApplicationStatus, HatApplication, Version }
 import org.hatdex.hat.api.models.{ AccessToken, DataDebit, EndpointQuery }
 import org.hatdex.hat.api.service.richData.{ DataDebitService, RichDataDuplicateDebitException, RichDataService }
@@ -35,7 +36,7 @@ import org.hatdex.hat.authentication.models.HatUser
 import org.hatdex.hat.dal.Tables
 import org.hatdex.hat.dal.Tables.ApplicationStatusRow
 import org.hatdex.hat.resourceManagement.HatServer
-import org.hatdex.hat.utils.FutureTransformations
+import org.hatdex.hat.utils.{ FutureTransformations, Utils }
 import org.hatdex.libs.dal.HATPostgresProfile.api._
 import org.joda.time.DateTime
 import play.api.Logger
@@ -46,6 +47,7 @@ import play.api.mvc.RequestHeader
 
 import scala.concurrent.Future
 import scala.concurrent.duration._
+import scala.util.Success
 
 class ApplicationStatusCheckService @Inject() (wsClient: WSClient)(implicit val rec: RemoteExecutionContext) {
 
@@ -56,11 +58,13 @@ class ApplicationStatusCheckService @Inject() (wsClient: WSClient)(implicit val
     }
   }
 
-  protected def status(statusCheck: ApplicationStatus.External, token: String): Future[Boolean] =
+  protected def status(statusCheck: ApplicationStatus.External, token: String): Future[Boolean] = {
     wsClient.url(statusCheck.statusUrl)
       .withHttpHeaders("x-auth-token" → token)
+      .withRequestTimeout(5000.millis)
       .get()
       .map(_.status == statusCheck.expectedStatus)
+  }
 }
 
 class ApplicationsService @Inject() (
@@ -69,15 +73,20 @@ class ApplicationsService @Inject() (
     dataDebitService: DataDebitService,
     statusCheckService: ApplicationStatusCheckService,
     trustedApplicationProvider: TrustedApplicationProvider,
-    silhouette: Silhouette[HatApiAuthEnvironment])(implicit val ec: DalExecutionContext) {
+    silhouette: Silhouette[HatApiAuthEnvironment],
+    system: ActorSystem)(implicit val ec: DalExecutionContext) {
 
   private val logger = Logger(this.getClass)
   private val applicationsCacheDuration: FiniteDuration = 30.minutes
 
   def applicationStatus(id: String, bustCache: Boolean = false)(implicit hat: HatServer, user: HatUser, requestHeader: RequestHeader): Future[Option[HatApplication]] = {
+    val eventuallyCleanedCache = if (bustCache) {
+      cache.remove(s"apps:${hat.domain}")
+      cache.remove(appCacheKey(id))
+    }
+    else { Future.successful(Done) }
     for {
-      _ ← if (bustCache) { cache.remove(appCacheKey(id)) } else { Future.successful(Done) }
-      _ ← if (bustCache) { cache.remove(s"apps:${hat.domain}") } else { Future.successful(Done) }
+      _ ← eventuallyCleanedCache
       application ← cache.get[HatApplication](appCacheKey(id))
         .flatMap {
           case Some(application) ⇒ Future.successful(Some(application))
@@ -86,24 +95,33 @@ class ApplicationsService @Inject() (
             for {
               maybeApp <- trustedApplicationProvider.application(id)
               setup <- applicationSetupStatus(id)(hat.db)
-              status <- FutureTransformations.transform(maybeApp.map(collectStatus(_, setup)))
-              _ ← status.map(cache.set(appCacheKey(id), _, applicationsCacheDuration)).getOrElse(Future.successful(Done))
-            } yield status
+              status <- FutureTransformations.transform(maybeApp.map(refetchApplicationsStatus(_, Seq(setup).flatten)))
+              _ ← status.map(s ⇒ cache.set(appCacheKey(id), s._1, applicationsCacheDuration)).getOrElse(Future.successful(Done))
+            } yield status.map(_._1)
         }
     } yield application
   }
 
   def applicationStatus()(implicit hat: HatServer, user: HatUser, requestHeader: RequestHeader): Future[Seq[HatApplication]] = {
-    cache.getOrElseUpdate(s"apps:${hat.domain}", applicationsCacheDuration) {
-      for {
-        apps <- trustedApplicationProvider.applications // potentially caching
-        setup <- applicationSetupStatus()(hat.db) // database
-        statuses <- Future.sequence(apps
-          .map(a => (a, setup.find(_.id == a.id)))
-          .map(as => collectStatus(as._1, as._2)))
-        _ ← Future.sequence(statuses.map(app ⇒ cache.set(appCacheKey(app.application.id), app, applicationsCacheDuration))) // reinject all fetched items as individual cached items
-      } yield statuses
-    }
+    cache.get[Seq[HatApplication]](s"apps:${hat.domain}")
+      .flatMap({
+        case Some(applications) ⇒ Future.successful(applications)
+        case None ⇒
+          for {
+            apps ← trustedApplicationProvider.applications // potentially caching
+            setup ← applicationSetupStatus()(hat.db) // database
+            statuses ← Future.sequence(apps.map(refetchApplicationsStatus(_, setup)))
+            _ ← if (statuses.forall(_._2)) { cache.set(s"apps:${hat.domain}", statuses.unzip._1, applicationsCacheDuration) } else { Future.successful(Done) }
+          } yield statuses.unzip._1
+      })
+  }
+
+  private def refetchApplicationsStatus(app: Application, setup: Seq[Tables.ApplicationStatusRow])(implicit hat: HatServer, user: HatUser, requestHeader: RequestHeader) = {
+    val aSetup = setup.find(_.id == app.id)
+    Utils.timeFuture(s"${hat.domain} ${app.id} status", logger)(collectStatus(app, aSetup))
+      .andThen {
+        case Success((a, true)) ⇒ cache.set(appCacheKey(a.application.id), a, applicationsCacheDuration)
+      }
   }
 
   def setup(application: HatApplication)(implicit hat: HatServer, user: HatUser, requestHeader: RequestHeader): Future[HatApplication] = {
@@ -197,26 +215,33 @@ class ApplicationsService @Inject() (
     }
   }
 
+  private def fastOrDefault[T](timeout: FiniteDuration, default: T)(block: => Future[T]): Future[(T, Boolean)] = {
+    val fallback = akka.pattern.after(timeout, using = system.scheduler)(Future.successful((default, false)))
+    Future.firstCompletedOf(Seq(block.map((_, true)), fallback))
+  }
+
   private def collectStatus(app: Application, setup: Option[ApplicationStatusRow])(
     implicit
-    hat: HatServer, user: HatUser, requestHeader: RequestHeader): Future[HatApplication] = {
+    hat: HatServer, user: HatUser, requestHeader: RequestHeader): Future[(HatApplication, Boolean)] = { // return status as well as flag indicating if it is successfully generated
 
     setup match {
       case Some(ApplicationStatusRow(_, version, true)) =>
+        val eventualStatus = fastOrDefault(5.seconds, (false, ""))(checkStatus(app))
+        val eventualMostRecentData = fastOrDefault(5.seconds, Option[DateTime](null))(mostRecentDataTime(app))
         for {
-          (status, _) <- checkStatus(app)
-          mostRecentData <- mostRecentDataTime(app)
+          ((status, _), canCacheStatus) <- eventualStatus
+          (mostRecentData, canCacheData) <- eventualMostRecentData
         } yield {
           logger.debug(s"Check compatibility between $version and new ${app.status}: ${Version(version).greaterThan(app.status.compatibility)}")
-          HatApplication(app, setup = true, enabled = true, active = status,
+          (HatApplication(app, setup = true, enabled = true, active = status,
             Some(app.status.compatibility.greaterThan(Version(version))), // Needs updating if setup version beyond compatible
-            mostRecentData)
+            mostRecentData), canCacheStatus && canCacheData)
         }
       case Some(ApplicationStatusRow(_, _, false)) =>
         // If application has been disabled, reflect in status
-        Future.successful(HatApplication(app, setup = true, enabled = false, active = false, needsUpdating = None, mostRecentData = None))
+        Future.successful((HatApplication(app, setup = true, enabled = false, active = false, needsUpdating = None, mostRecentData = None), true))
       case None =>
-        Future.successful(HatApplication(app, setup = false, enabled = false, active = false, needsUpdating = None, mostRecentData = None))
+        Future.successful((HatApplication(app, setup = false, enabled = false, active = false, needsUpdating = None, mostRecentData = None), true))
     }
   }
 

hat/app/org/hatdex/hat/api/service/monitoring/HatDataEventBus.scala

@@ -86,7 +86,7 @@ class HatDataEventDispatcher @Inject() (dataEventBus: HatDataEventBus) {
   protected val logger: Logger = Logger(this.getClass)
 
   def dispatchEventDataCreated(message: String)(implicit request: SecuredRequest[HatApiAuthEnvironment, _]): PartialFunction[Try[Seq[EndpointData]], Unit] = {
-    case Success(saved) =>
+    case Success(saved) if saved.nonEmpty =>
       logger.debug(s"Dispatch data created event: $message")
       dataEventBus.publish(HatDataEventBus.DataCreatedEvent(
         request.dynamicEnvironment.domain,

hat/app/org/hatdex/hat/api/service/richData/DataSentintel.scala

@@ -0,0 +1,136 @@
+/*
+ * Copyright (C) 2017 HAT Data Exchange Ltd
+ * SPDX-License-Identifier: AGPL-3.0
+ *
+ * This file is part of the Hub of All Things project (HAT).
+ *
+ * HAT is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License
+ * as published by the Free Software Foundation, version 3 of
+ * the License.
+ *
+ * HAT is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See
+ * the GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General
+ * Public License along with this program. If not, see
+ * <http://www.gnu.org/licenses/>.
+ *
+ * Written by Andrius Aucinas <andrius.aucinas@hatdex.org>
+ * 7 / 2018
+ */
+
+package org.hatdex.hat.api.service.richData
+
+import akka.Done
+import akka.actor.ActorSystem
+import akka.stream.ActorMaterializer
+import akka.stream.scaladsl.{ Flow, Sink, Source }
+import javax.inject.Inject
+import org.hatdex.hat.api.service.DalExecutionContext
+import org.hatdex.hat.dal.Tables._
+import org.hatdex.libs.dal.HATPostgresProfile.api._
+import org.joda.time.DateTime
+import org.joda.time.format.{ DateTimeFormat, DateTimeFormatter, ISODateTimeFormat }
+import play.api.libs.json._
+
+import scala.concurrent.Future
+
+class DataSentintel @Inject() (implicit ec: DalExecutionContext, actorSystem: ActorSystem) {
+
+  protected implicit val materializer: ActorMaterializer = ActorMaterializer()
+  protected val updateBatchSize = 500
+
+  //  def validateDataStructure(data: JsValue, configuration: EndpointConfiguration): (JsValue, Option[String], Option[DateTime]) = {
+  //    throw new RuntimeException("Not Implemented")
+  //  }
+
+  def ensureUniquenessKey(source: String, key: String)(implicit db: Database): Future[Done] = {
+    import com.github.tminglei.slickpg.window.PgWindowFuncSupport.WindowFunctions._
+    val config = EndpointConfiguration(Some(key), None, None)
+
+    val dbJsonPath = key.split('.').toList
+
+    val clashingRecords = DataJson
+      .filter(_.source === source) // only records for this source
+      .filterNot(_.data.#>>(dbJsonPath) === "") // and only those that do have the key defined
+      .map(r ⇒ (r.recordId, rowNumber().over.partitionBy(r.source, r.data #>> dbJsonPath).sortBy(r.date.desc))) // number the rows starting from most recent
+      .subquery // subquery used to force generating the query before (incorrectly) Slick tries to use the partition windowing function within where clause
+      .filter { case (_, rank) ⇒ rank > 1L } // skip the newest row
+      .map(_._1) // get record ID for each remaining row
+
+    val deleteQuery = DataJson.filter(_.recordId in clashingRecords).delete
+
+    val updatingStream = Source.fromPublisher(db.stream(DataJson.filter(_.source === source).result.transactionally.withStatementParameters(fetchSize = updateBatchSize)))
+      .via(Flow[DataJsonRow].grouped(updateBatchSize))
+      .mapAsync(1)({ batch ⇒
+        db.run(DBIO.sequence(
+          batch
+            .map(r ⇒ r.copy(sourceUniqueId = config.getKey(r.data))) // take the value at key as either string or number
+            .map(DataJson.insertOrUpdate)).transactionally) // update the row with sourceUniqueId inserted
+      })
+
+    db.run(deleteQuery)
+      .flatMap(_ ⇒ {
+        updatingStream.runWith(Sink.ignore) // the result is not important as long as it succeeds
+      })
+      .map(_ ⇒ Done)
+  }
+
+  def updateSourceTimestamp(source: String, key: String, format: String = "")(implicit db: Database): Future[Done] = {
+    val config = EndpointConfiguration(None, Some(key), Some(format))
+
+    val updatingStream = Source.fromPublisher(db.stream(DataJson.filter(_.source === source).result.transactionally.withStatementParameters(fetchSize = updateBatchSize)))
+      .via(Flow[DataJsonRow].grouped(updateBatchSize))
+      .mapAsync(parallelism = 1)({ batch ⇒
+        db.run(DBIO.sequence(
+          batch
+            .map(r ⇒ r.copy(sourceTimestamp = config.getTimestamp(r.data))) // take the value at key as DateTime
+            .map(DataJson.insertOrUpdate)).transactionally) // update the row with sourceUniqueId inserted
+      })
+
+    updatingStream
+      .runWith(Sink.ignore) // the result is not important as long as it succeeds
+      .map(_ ⇒ Done)
+  }
+}
+
+case class EndpointConfiguration(
+    keyField: Option[String],
+    timestampField: Option[String],
+    timestampFormat: Option[String] // JODA DateTime format or "'epoch'" literal to denote that it is in seconds since epoch, defaults to ISO8601
+) {
+  lazy val keyPath: Option[JsPath] = keyField.map(JsonDataTransformer.parseJsPath)
+  lazy val timestampPath: Option[JsPath] = timestampField.map(JsonDataTransformer.parseJsPath)
+  private implicit val timestampJsonFormat = jodaDateReads(timestampFormat.getOrElse(""))
+
+  def getKey(d: JsValue): Option[String] = keyPath.flatMap(k ⇒ k.asSingleJson(d).asOpt[String]
+    .orElse(k.asSingleJson(d).asOpt[Long].map(_.toString)))
+
+  def getTimestamp(d: JsValue): Option[DateTime] = timestampPath.flatMap(_.asSingleJson(d).asOpt[DateTime])
+
+  protected def jodaDateReads(pattern: String): Reads[DateTime] = new Reads[DateTime] {
+
+    private val (df, corrector, numberCorrector): (DateTimeFormatter, String ⇒ String, Long ⇒ Long) =
+      pattern match {
+        case ""        ⇒ (ISODateTimeFormat.dateOptionalTimeParser, identity[String], identity[Long])
+        case "'epoch'" ⇒ (ISODateTimeFormat.dateOptionalTimeParser, identity[String], { x: Long ⇒ x * 1000L })
+        case _         ⇒ (DateTimeFormat.forPattern(pattern), identity[String], identity[Long])
+      }
+
+    def reads(json: JsValue): JsResult[DateTime] = json match {
+      case JsNumber(d) => JsSuccess(new DateTime(numberCorrector(d.toLong)))
+      case JsString(s) => parseDate(corrector(s)) match {
+        case Some(d) => JsSuccess(d)
+        case _       => JsError(Seq(JsPath() -> Seq(JsonValidationError("error.expected.jodadate.format", pattern))))
+      }
+      case _ => JsError(Seq(JsPath() -> Seq(JsonValidationError("error.expected.date"))))
+    }
+
+    private def parseDate(input: String): Option[DateTime] =
+      scala.util.control.Exception.nonFatalCatch[DateTime] opt (DateTime.parse(input, df))
+  }
+}
+