VAR reversals: filter out duplicate StateChange Events

api-models/src/main/scala/com/gu/mobile.notifications.client/models/liveActitivites/LiveActivityEventBridgePayloads.scala

@@ -40,6 +40,11 @@ sealed trait LiveActivityEventType {
 case object CreateChannelEvent extends LiveActivityEventType { val asString = "channel-create" }
 case object StartLiveActivityEvent extends LiveActivityEventType { val asString = "broadcast-start" }
 case object UpdateLiveActivityEvent extends LiveActivityEventType { val asString = "broadcast-update" }
+
+// Required for deduping, but string value allows processing as standard update event once pushed via eventbus to live activities service.
+case object UpdateStateChangeLiveActivityEvent extends LiveActivityEventType {
+  val asString = "broadcast-update"
+}
 case object EndLiveActivityEvent extends LiveActivityEventType { val asString = "broadcast-end" }
 case object DeleteChannelEvent extends LiveActivityEventType { val asString = "channel-delete" }
 
@@ -48,6 +53,7 @@ object LiveActivityEventType {
     CreateChannelEvent,
     StartLiveActivityEvent,
     UpdateLiveActivityEvent,
+    UpdateStateChangeLiveActivityEvent,
     EndLiveActivityEvent,
     DeleteChannelEvent
   )

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

@@ -84,7 +84,6 @@ 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/EventFilter.scala

@@ -65,17 +65,36 @@ class EventFilter[A <: Payload, D](distinctCheck: DynamoDistinctCheck[A, D]) ext
     isolatedEndEvents ++ updateEvents
   }
 
+
+  //
+  private def filterOutStateChangeEventsNotReceivedInIsolation(events: List[LiveActivityPayload]): List[LiveActivityPayload] = {
+    val (stateChangeEvents, updateEvents) = events.partition(_.eventType == com.gu.mobile.notifications.client.models.liveActitivites.UpdateStateChangeLiveActivityEvent)
+    val updateEventMatchIds = updateEvents.map(_.liveActivityID).toSet
+    val (isolatedStateChangeEvents, superfluousStateChangeEvents) =
+      stateChangeEvents.partition(event => !updateEventMatchIds.contains(event.liveActivityID))
+
+    // todo clean this up and change partition to filter.
+    // logging to verify hypothesis of state change event for every cycle there is an update including end.
+    if (superfluousStateChangeEvents.nonEmpty)
+      logger.debug("Superfluous state-change event(s) suppressed for match ids: " + superfluousStateChangeEvents.map(_.liveActivityID))
+
+    isolatedStateChangeEvents ++ updateEvents
+  }
+
   // we need to be able to access liveActivityId so the Payload trait must be narrowed. "An instance of A <:< B witnesses that A is a subtype of B."
   def filterDynamoEventsForLiveActivities(
       dynamoEvents: List[LiveActivityPayload],
   )(implicit ec: ExecutionContext, ev: LiveActivityPayload <:< A): Future[List[LiveActivityPayload]] = {
     for {
+      // Filter out any payloads that have already been sent (either in the local cache or in the dynamo table) to avoid duplicate processing.
       newEvents <- filterAsync(dynamoEvents)(item => isUniqueRecord(ev(item)))
-      /** Because we poll once a minute, we might end up with a triggering update event (eg. very late goal) along with
-        * an end event in the same polling cycle, but we only ever want to process the end event alone after all updates
-        * have been processed (dispatched via eventbridge). This only affects Live Activities.
-        */
-      eventsToProcess = filterOutEndEventsNotReceivedInIsolation(newEvents)
+      // In order to capture VAR goal reversals, we need to calculate payload state change events, but remove the superfluous
+      // ones with state change already captured by normal update events
+      eventsWithoutStateChangeEvents = filterOutStateChangeEventsNotReceivedInIsolation(newEvents)
+      // Because we poll once a minute, we might end up with a triggering update event (eg. very late goal)
+      // along with an end event in the same polling cycle, but we only ever want to process the end event
+      // alone after all updates have been processed (dispatched via eventbridge). This only affects Live Activities.
+      eventsToProcess = filterOutEndEventsNotReceivedInIsolation(eventsWithoutStateChangeEvents)
       processedEvents <- Future.traverse(eventsToProcess) { item =>
         distinctCheck.insertEvent(ev(item)).map {
           case Distinct => {
@@ -88,6 +107,7 @@ class EventFilter[A <: Payload, D](distinctCheck: DynamoDistinctCheck[A, D]) ext
     } yield processedEvents.flatten
   }
 
+  // TODO filter out statechange events for push notifications as well.
   def filterDynamoEvents(dynamoEvents: List[A])(implicit ec: ExecutionContext): Future[List[A]] = {
     Future.traverse(dynamoEvents)(filterDynamoEvent).map(_.flatten)
   }

football/src/main/scala/com/gu/mobile/notifications/football/notificationbuilders/MatchStatusLiveActivityPayloadBuilder.scala

@@ -1,6 +1,6 @@
 package com.gu.mobile.notifications.football.notificationbuilders
 
-import com.gu.mobile.notifications.client.models.liveActitivites.{Competition, CreateChannelEvent, EndLiveActivityEvent, FootballLiveActivity, FootballMatchContentState, LiveActivityPayload, MatchStatus, TeamState, UpdateLiveActivityEvent}
+import com.gu.mobile.notifications.client.models.liveActitivites.{Competition, CreateChannelEvent, EndLiveActivityEvent, FootballLiveActivity, FootballMatchContentState, LiveActivityPayload, MatchStatus, TeamState, UpdateLiveActivityEvent, UpdateStateChangeLiveActivityEvent}
 import com.gu.mobile.notifications.football.models._
 import pa.MatchDay
 
@@ -11,9 +11,9 @@ class MatchStatusLiveActivityPayloadBuilder {
 
   def buildFootballContentState(
                                  currentMinute: Option[Int],
-      matchInfo: MatchDay,
+                                 matchInfo: MatchDay,
                                  allEvents: List[FootballMatchEvent],
-      articleId: Option[String]
+                                 articleId: Option[String]
   ): FootballMatchContentState = {
 
     val goals = allEvents.collect { case g: Goal => g }
@@ -86,6 +86,7 @@ class MatchStatusLiveActivityPayloadBuilder {
       case Cancelled(_) => EndLiveActivityEvent
       case CreateChannel(_) => CreateChannelEvent
       case EndLiveActivity(_) => EndLiveActivityEvent
+      case MatchStateChangeEvent(_) => UpdateStateChangeLiveActivityEvent // Json writer should encode this as "broadcast-update" for APNS compatibility
       case _ => UpdateLiveActivityEvent
     }
 

football/src/main/scala/com/gu/mobile/notifications/football/notificationbuilders/MatchStatusNotificationBuilder.scala

@@ -5,7 +5,7 @@ import java.util.UUID
 import com.gu.mobile.notifications.client.models.Importance.{Importance, Major, Minor}
 import com.gu.mobile.notifications.client.models._
 import com.gu.mobile.notifications.client.models.liveActitivites.MatchStatus
-import com.gu.mobile.notifications.football.models.{Dismissal, ExtraTimeFirstHalf, ExtraTimeHalfTime, ExtraTimeSecondHalf, ExtraTimeToBePlayed, FootballMatchEvent, FullTime, Goal, HalfTime, KickOff, Penalties, PenaltiesToBePlayed, PenaltyShootoutKick, PenaltyShootoutScore, PreMatch, RedCards, Score, SecondHalf, StartLiveActivity}
+import com.gu.mobile.notifications.football.models.{Dismissal, ExtraTimeFirstHalf, ExtraTimeHalfTime, ExtraTimeSecondHalf, ExtraTimeToBePlayed, FootballMatchEvent, FullTime, Goal, HalfTime, KickOff, MatchStateChangeEvent, Penalties, PenaltiesToBePlayed, PenaltyShootoutKick, PenaltyShootoutScore, PreMatch, RedCards, Score, SecondHalf, StartLiveActivity}
 import pa.{MatchDay, MatchDayTeam}
 
 import scala.PartialFunction.condOpt
@@ -207,6 +207,7 @@ class MatchStatusNotificationBuilder(mapiHost: String) {
     case _: ExtraTimeSecondHalf => "Extra time: second half"
     case _: PenaltiesToBePlayed => "Penalties to follow"
     case _: Penalties => "Penalties"
+    case _: MatchStateChangeEvent => "Match update"
     case _ => "The Guardian"
   }
 

football/src/test/scala/com/gu/mobile/notifications/football/lib/EventConsumerSpec.scala

@@ -407,6 +407,26 @@ class EventConsumerSpec(implicit ev: ExecutionEnv)
         payload.eventType == UpdateLiveActivityEvent
       )
     }
+
+    "generate a state-change UPDATE payload when a state change has occurred" in new MatchEventsContext {
+      override def stateChange: Boolean = true
+
+      val result: List[LiveActivityPayload] =
+        eventConsumerLiveActivities.eventsToLiveActivityPayload(matchDataLA)
+      result must contain((payload: LiveActivityPayload) =>
+        payload.eventType == UpdateStateChangeLiveActivityEvent
+      )
+    }
+
+    "NOT generate a state-change payload when no state change has occurred" in new MatchEventsContext {
+      override def stateChange: Boolean = false
+
+      val result: List[LiveActivityPayload] =
+        eventConsumerLiveActivities.eventsToLiveActivityPayload(matchDataLA)
+      result must not(contain((payload: LiveActivityPayload) =>
+        payload.eventType == UpdateStateChangeLiveActivityEvent
+      ))
+    }
   }
 
   "A LiveActivity EventConsumer handles synthetic events for match phases and" should {
@@ -609,7 +629,6 @@ class EventConsumerSpec(implicit ev: ExecutionEnv)
       )
     }
 
-    // todo - check end broadcast states are shown
     "generate a abandoned live activity END payload" in new MatchEventsContext {
       override def matchDayLA: MatchDay =
         super.matchDay.copy(matchStatus = "Abandoned", result = false)
@@ -703,12 +722,15 @@ class EventConsumerSpec(implicit ev: ExecutionEnv)
 
     def matchDay: MatchDay = Parser.parseMatchDay(loadFile("20170811.xml")).head
 
+    // Whether the synthetic event generator should emit a state-change event (VAR overturn etc.).
+    def stateChange: Boolean = false
+
     def events: List[MatchEvent] =
       new SyntheticMatchEventGenerator(() => ZonedDateTime.now()).generate(
         rawEvents,
         "4011135",
         matchDay,
-        false
+        stateChange
       )
 
     def matchData = MatchDataWithArticle(
@@ -730,7 +752,7 @@ class EventConsumerSpec(implicit ev: ExecutionEnv)
 
     def eventsLA: List[MatchEvent] =
       new SyntheticMatchEventGenerator(() => ZonedDateTime.now())
-        .generate(rawEventsLA, "4484328", matchDayLA, false)
+        .generate(rawEventsLA, "4484328", matchDayLA, stateChange)
 
     def matchDataLA = MatchDataWithArticle(
       matchDayLA,

football/src/test/scala/com/gu/mobile/notifications/football/lib/EventFilterSpec.scala

@@ -112,5 +112,62 @@ class EventFilterSpec(implicit ev: ExecutionEnv) extends Specification with Mock
       // isDuplicate should only have been called once (from the first invocation)
       there was one(distinctCheck).isDuplicate(event1)
     }
+
+    "suppress a state-change event when an update event exists for the same match" in new FilterScopeLiveActivities {
+      val update = makePayload(UpdateLiveActivityEvent, "match-1")
+      val stateChange = makePayload(UpdateStateChangeLiveActivityEvent, "match-1")
+
+      distinctCheck.isDuplicate(update) returns Future.successful(false)
+      distinctCheck.isDuplicate(stateChange) returns Future.successful(false)
+      distinctCheck.insertEvent(update) returns Future.successful(Distinct)
+
+      eventFilter.filterDynamoEventsForLiveActivities(List(update, stateChange)) must contain(exactly(update)).await
+    }
+
+    "allow a state-change event through when received in isolation (no update for same match)" in new FilterScopeLiveActivities {
+      val stateChange = makePayload(UpdateStateChangeLiveActivityEvent, "match-1")
+
+      distinctCheck.isDuplicate(stateChange) returns Future.successful(false)
+      distinctCheck.insertEvent(stateChange) returns Future.successful(Distinct)
+
+      eventFilter.filterDynamoEventsForLiveActivities(List(stateChange)) must contain(exactly(stateChange)).await
+    }
+
+    "allow a state-change event through when the update is for a different match" in new FilterScopeLiveActivities {
+      val update = makePayload(UpdateLiveActivityEvent, "match-2")
+      val stateChange = makePayload(UpdateStateChangeLiveActivityEvent, "match-1")
+
+      distinctCheck.isDuplicate(update) returns Future.successful(false)
+      distinctCheck.isDuplicate(stateChange) returns Future.successful(false)
+      distinctCheck.insertEvent(update) returns Future.successful(Distinct)
+      distinctCheck.insertEvent(stateChange) returns Future.successful(Distinct)
+
+      eventFilter.filterDynamoEventsForLiveActivities(List(update, stateChange)) must contain(exactly(update, stateChange)).await
+    }
+
+    "drop an event when the final insertEvent does not return Distinct (Duplicate/Unknown)" in new FilterScopeLiveActivities {
+      val event = makePayload(UpdateLiveActivityEvent, "match-1")
+
+      distinctCheck.isDuplicate(event) returns Future.successful(false)
+      distinctCheck.insertEvent(event) returns Future.successful(Duplicate) // or DynamoDistinctCheck.Unknown
+
+      eventFilter.filterDynamoEventsForLiveActivities(List(event)) must beEmpty[List[LiveActivityPayload]].await
+    }
+
+    "not cache an end event that was suppressed as an early end event" in new FilterScopeLiveActivities {
+      val update = makePayload(UpdateLiveActivityEvent, "match-1")
+      val end = makePayload(EndLiveActivityEvent, "match-1")
+
+      distinctCheck.isDuplicate(update) returns Future.successful(false)
+      distinctCheck.isDuplicate(end) returns Future.successful(false)
+      distinctCheck.insertEvent(update) returns Future.successful(Distinct)
+      distinctCheck.insertEvent(end) returns Future.successful(Distinct)
+
+      // cycle 1: end suppressed (update present) -> only update dispatched
+      eventFilter.filterDynamoEventsForLiveActivities(List(update, end)) must contain(exactly(update)).await
+      // cycle 2: end now arrives in isolation and should still be dispatchable (not cached away)
+      eventFilter.filterDynamoEventsForLiveActivities(List(end)) must contain(exactly(end)).await
+    }
+
   }
 }

football/src/test/scala/com/gu/mobile/notifications/football/notificationbuilders/MatchStatusLiveActivityPayloadBuilderSpec.scala

@@ -1,8 +1,8 @@
 package com.gu.mobile.notifications.football.notificationbuilders
 
 import com.gu.mobile.notifications.client.models.{DefaultGoalType, ScoredShootoutResult}
-import com.gu.mobile.notifications.client.models.liveActitivites.{CreateChannelEvent, EndLiveActivityEvent, FirstHalf, FootballLiveActivity, FootballMatchContentState, LiveActivityPayload, TeamState, UpdateLiveActivityEvent, Competition => LACompetition}
-import com.gu.mobile.notifications.football.models.{Abandoned, Cancelled, CreateChannel, Dismissal, EndLiveActivity, Goal, HalfTime, PenaltyShootoutKick}
+import com.gu.mobile.notifications.client.models.liveActitivites.{CreateChannelEvent, EndLiveActivityEvent, FirstHalf, FootballLiveActivity, FootballMatchContentState, LiveActivityPayload, TeamState, UpdateLiveActivityEvent, UpdateStateChangeLiveActivityEvent, Competition => LACompetition}
+import com.gu.mobile.notifications.football.models.{Abandoned, Cancelled, CreateChannel, Dismissal, EndLiveActivity, Goal, HalfTime, MatchStateChangeEvent, PenaltyShootoutKick}
 import org.specs2.mutable.Specification
 import org.specs2.specification.Scope
 import pa.{Competition, MatchDay, MatchDayTeam, Round, Stage, Venue}
@@ -71,6 +71,12 @@ class MatchStatusLiveActivityPayloadBuilderSpec extends Specification {
       builder.build(EndLiveActivity("e1"), matchInfo, List.empty, None).eventType mustEqual EndLiveActivityEvent
     }
 
+    "map a MatchStateChangeEvent trigger to an UpdateStateChangeLiveActivityEvent" in new MatchEventsContext {
+      val stateChangePayload = builder.build(MatchStateChangeEvent("e1"), matchInfo, List.empty, None)
+      stateChangePayload.eventType mustEqual UpdateStateChangeLiveActivityEvent
+      stateChangePayload.eventType.asString mustEqual "broadcast-update"
+    }
+
     "map any other trigger (e.g. a goal) to an UpdateLiveActivityEvent" in new MatchEventsContext {
       builder.build(baseGoal, matchInfo, List.empty, None).eventType mustEqual UpdateLiveActivityEvent
     }