VAR reversals 2: add football match diffing to generate a synth state change event

football/src/main/scala/com/gu/mobile/notifications/football/Lambda.scala

@@ -8,7 +8,7 @@ import com.amazonaws.services.dynamodbv2.{AmazonDynamoDBAsync, AmazonDynamoDBAsy
 import com.amazonaws.services.s3.{AmazonS3, AmazonS3ClientBuilder}
 import com.gu.contentapi.client.GuardianContentClient
 import com.gu.mobile.liveactivities.event.bus.LiveActivityPusher
-import com.gu.mobile.notifications.football.lib.{ArticleSearcher, DynamoDistinctCheck, DynamoMatchLiveActivity, DynamoMatchNotification, EventConsumer, EventFilter, FootballData, LiveActivityEventConsumer, NotificationHttpProvider, NotificationSender, NotificationsApiClient, PACompetition, PaFootballClient, S3DataStore, SyntheticMatchEventGenerator}
+import com.gu.mobile.notifications.football.lib.{ArticleSearcher, DynamoDistinctCheck, DynamoMatchLiveActivity, DynamoMatchNotification, DynamoPayloadStateCheck, EventConsumer, EventFilter, FootballData, LiveActivityEventConsumer, NotificationHttpProvider, NotificationSender, NotificationsApiClient, PACompetition, PaFootballClient, S3DataStore, SyntheticMatchEventGenerator}
 import com.gu.mobile.notifications.football.notificationbuilders.{MatchStatusLiveActivityPayloadBuilder, MatchStatusNotificationBuilder}
 import play.api.libs.json.Json
 
@@ -53,6 +53,7 @@ object Lambda extends Logging {
 
   lazy val capiClient = GuardianContentClient(configuration.capiApiKey)
 
+  lazy val payloadStateCheck = new DynamoPayloadStateCheck(dynamoDBClient, liveActivitiesTableName)
   lazy val syntheticMatchEventGenerator = new SyntheticMatchEventGenerator(getZonedDateTime)
 
   lazy val notificationHttpProvider = new NotificationHttpProvider()
@@ -66,7 +67,7 @@ object Lambda extends Logging {
 
   lazy val competitionsDataStore = new S3DataStore[PACompetition](s3Client, paDataBucket)
 
-  lazy val footballData = new FootballData(paFootballClient, syntheticMatchEventGenerator, competitionsDataStore, configuration.stage)
+  lazy val footballData = new FootballData(paFootballClient, syntheticMatchEventGenerator, competitionsDataStore, payloadStateCheck, configuration.stage)
 
   lazy val articleSearcher = new ArticleSearcher(capiClient)
 
@@ -130,6 +131,7 @@ object Lambda extends Logging {
   }
 }
 
+// TODO handle match state changes
 class NotificationHandler(configuration: Configuration, apiClient: NotificationsApiClient, dynamoDBClient: AmazonDynamoDBAsync, tableName: String) extends Logging {
 
   lazy val notificationSender = new NotificationSender(apiClient)

football/src/main/scala/com/gu/mobile/notifications/football/lib/DynamoPayloadStateCheck.scala

@@ -0,0 +1,50 @@
+package com.gu.mobile.notifications.football.lib
+
+import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync
+import com.gu.mobile.notifications.client.models.liveActitivites.{FootballMatchContentState, LiveActivityPayload}
+import com.gu.mobile.notifications.football.Logging
+import org.scanamo.{ScanamoAsync, Table}
+import play.api.libs.json.Json
+
+import scala.concurrent.{ExecutionContext, Future}
+
+// This is used to pre-diff the state in FootballData fetcher to determine if state-change synthetic event should be generated.
+// It is not used to determine if a notification should be sent, that is handled by DynamoDistinctCheck.
+// Both classes read the same table data.
+class DynamoPayloadStateCheck(client: AmazonDynamoDBAsync, tableName: String) extends Logging {
+
+  def isMatchStateIdentical(id: String, state: FootballMatchContentState)(implicit ec: ExecutionContext): Future[Boolean] = {
+    import org.scanamo.generic.auto._
+    import org.scanamo.syntax._
+
+    val scanamoAsync: ScanamoAsync = ScanamoAsync(client)
+    val payloadTable = Table[DynamoMatchLiveActivity](tableName)
+    val lastPayloadIndex = payloadTable.index("lastPayload-index") // only live activities payload table has an index.
+
+    val gsiQuery = lastPayloadIndex.descending.limit(1).query("liveActivityID" -> id)
+
+    scanamoAsync.exec(gsiQuery).map { rows =>
+      isIdenticalToLatest(rows.collect { case Right(row) => row }, state)
+    } recover {
+      case e =>
+        logger.error(s"Failure while checking for dynamodb GSI for match state $tableName: ${e.getMessage}.")
+        false // todo should this be true - do we want to generate a synthetic state-change event if we can't check the latest state?
+    }
+  }
+
+  private[lib] def isIdenticalToLatest(rows: List[DynamoMatchLiveActivity], state: FootballMatchContentState): Boolean =
+    rows.headOption
+      .flatMap(row => footballStateFromPayload(row.payload))
+      .forall(latest => ignoringArticleUrl(latest, state))
+
+  // articleUrl is only added to the payload after isMatchStateIdentical is called, so we must ignore it.
+  private def ignoringArticleUrl(a: FootballMatchContentState, b: FootballMatchContentState): Boolean =
+    a.copy(articleUrl = None, currentMinute = None) == b.copy(articleUrl = None, currentMinute = None) // these are calculated outside of state generation
+
+  // The stored `payload` column is the JSON-serialised LiveActivityPayload; the match state we diff
+  // against is its `broadcastContentStateData` subfield.
+  private[lib] def footballStateFromPayload(payloadJson: String): Option[FootballMatchContentState] =
+    Json.parse(payloadJson).as[LiveActivityPayload].broadcastContentStateData.collect {
+      case footballState: FootballMatchContentState => footballState
+    }
+}

football/src/main/scala/com/gu/mobile/notifications/football/lib/EventConsumer.scala

@@ -24,7 +24,6 @@ class EventConsumer(
     /** We have added synthetic events for live activities which we don't want
       * to generate push notifications for, so we filter those out here.
       */
-    // todo can we capture these strings in union type?
     val liveActivityEventTypes =
       List(
         // additional synthetic live activity life cycle events
@@ -35,7 +34,8 @@ class EventConsumer(
         "resumed",
         "abandoned",
         "cancelled",
-        "postponed"
+        "postponed",
+        "state-change" // todo
       )
 
     val filteredMatchData = matchData.copy(allEvents =
@@ -84,6 +84,7 @@ class LiveActivityEventConsumer(
     val duplicatePhaseEventTypes =
       List(
         "full-time", // only send "end-live-activity" synthetic event
+        "state-change" // todo
       )
 
     val filteredMatchData =

football/src/main/scala/com/gu/mobile/notifications/football/lib/FootballData.scala

@@ -1,19 +1,19 @@
 package com.gu.mobile.notifications.football.lib
 
+import com.gu.mobile.notifications.client.models.liveActitivites.LiveActivityPayload
+
 import java.time.{LocalDate, LocalTime, ZonedDateTime}
-import com.gu.mobile.notifications.football.{Configuration, Logging}
-import com.gu.mobile.notifications.football.models.RawMatchData
-import org.joda.time.DateTime
-import pa.MatchDay
+import com.gu.mobile.notifications.football.Logging
+import com.gu.mobile.notifications.football.models.{FootballMatchEvent, RawMatchData}
+import com.gu.mobile.notifications.football.notificationbuilders.MatchStatusLiveActivityPayloadBuilder
+import pa.{MatchDay, MatchEvent}
 import play.api.libs.json.{Format, Json}
 
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.Future
 import scala.util.control.NonFatal
 import scala.util.{Failure, Success, Try}
 
-case class EndedMatch(matchId: String, startTime: DateTime)
-
 case class PACompetition(
     id: String,
     tag: String,
@@ -31,6 +31,7 @@ class FootballData(
     paClient: PaFootballClient,
     syntheticEvents: SyntheticMatchEventGenerator,
     competitionsDataStore: S3DataStore[PACompetition],
+    payloadStateCheck: DynamoPayloadStateCheck,
     stage: String,
 ) extends Logging {
 
@@ -169,15 +170,47 @@ class FootballData(
     }
   }
 
-  private def processMatch(matchDay: MatchDay): Future[Option[RawMatchData]] = {
+  private def appendSyntheticEvents(matchDay: MatchDay, events: List[MatchEvent], stateChange: Boolean)(syntheticMatchEventGenerator: SyntheticMatchEventGenerator): Future[(MatchDay, List[MatchEvent])] = {
+    val eventsWithSyntheticEvents = syntheticMatchEventGenerator.generate(events, matchDay.id, matchDay, stateChange )
+    Future.successful((matchDay, eventsWithSyntheticEvents))
+  }
+
+  // Process individual match
+  private[lib] def processMatch(matchDay: MatchDay): Future[Option[RawMatchData]] = {
     val matchData = for {
-      (matchDay, events) <- paClient.eventsForMatch(matchDay, syntheticEvents)
-    } yield Some(RawMatchData(matchDay, events))
+      // fetch current PA events timeline for match
+      (_, events) <- paClient.eventsForMatch(matchDay)
+      // check if the current match state is identical to the last known payload state sent.
+      stateChange <- isFootballMatchStateIdentical(matchDay, events).map(!_)
+      // add synthetic events to the PA events timeline, if any are generated.
+      // a state change synth even will be generated for every update, but we filter out the superfluous ones in the EventConsumer.
+      (_, eventsWithSyntheticEvents) <- appendSyntheticEvents(matchDay, events, stateChange)(syntheticEvents)
+    } yield Some(RawMatchData(matchDay, eventsWithSyntheticEvents))
 
     matchData.recover { case NonFatal(exception) =>
       logger.error(s"Failed to process match ${matchDay.id}: ${exception.getMessage}", exception)
       None
     }
   }
 
+  private[lib] def isFootballMatchStateIdentical(matchDay: MatchDay, events: List[pa.MatchEvent]): Future[Boolean] = {
+    val payloadBuilder = new MatchStatusLiveActivityPayloadBuilder()
+    val toFootballEvent = FootballMatchEvent.fromPaMatchEvent(matchDay.homeTeam, matchDay.awayTeam) _
+
+    if (events.isEmpty) Future.successful(true) // no stage change before a match has started.
+    else {
+      val footballMatchState = payloadBuilder.buildFootballContentState(
+        currentMinute = None,
+        matchInfo = matchDay,
+        allEvents = events.flatMap(toFootballEvent),
+        articleId = None,
+      )
+      payloadStateCheck
+        .isMatchStateIdentical(matchDay.id, footballMatchState)
+        .recover { case NonFatal(exception) =>
+          logger.error(s"Error checking for for identical football match state ${matchDay.id}: ${exception.getMessage}")
+          true // assume identical // todo confirm this behaviour is desired.
+        }
+    }
+  }
 }

football/src/main/scala/com/gu/mobile/notifications/football/lib/PaFootballClient.scala

@@ -52,10 +52,10 @@ class PaFootballClient(override val apiKey: String, apiBase: String) extends PaC
     Future.reduceLeft(days.map { day => matchDay(day).recover { case _ => List.empty } })(_ ++ _)
   }
 
-  def eventsForMatch(matchDay: MatchDay, syntheticMatchEventGenerator: SyntheticMatchEventGenerator)(implicit ec: ExecutionContext): Future[(MatchDay, List[MatchEvent])] =
+  def eventsForMatch(matchDay: MatchDay)(implicit ec: ExecutionContext): Future[(MatchDay, List[MatchEvent])] =
     for {
       events <- matchEvents(matchDay.id).map(_.toList.flatMap(_.events))
     } yield {
-      (matchDay, syntheticMatchEventGenerator.generate(events, matchDay.id, matchDay))
+      (matchDay, events)
     }
 }