[DRAFT] Handle VAR reversals

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

@@ -40,17 +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" }
+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" }
 
 object LiveActivityEventType {
-  val values: Seq[LiveActivityEventType] = Seq(
-    CreateChannelEvent,
-    StartLiveActivityEvent,
-    UpdateLiveActivityEvent,
-    EndLiveActivityEvent,
-    DeleteChannelEvent
-  )
 
   implicit val format: Format[LiveActivityEventType] = new Format[LiveActivityEventType] {
     override def reads(json: JsValue): JsResult[LiveActivityEventType] = json match {

cdk/lib/__snapshots__/footballnotificationslambda.test.ts.snap

@@ -151,8 +151,34 @@ exports[`The FootballNotificationsLambda stack matches the snapshot for CODE 1`]
             "AttributeName": "id",
             "AttributeType": "S",
           },
+          {
+            "AttributeName": "liveActivityID",
+            "AttributeType": "S",
+          },
+          {
+            "AttributeName": "ttl",
+            "AttributeType": "N",
+          },
         ],
         "BillingMode": "PAY_PER_REQUEST",
+        "GlobalSecondaryIndexes": [
+          {
+            "IndexName": "lastPayload-index",
+            "KeySchema": [
+              {
+                "AttributeName": "liveActivityID",
+                "KeyType": "HASH",
+              },
+              {
+                "AttributeName": "ttl",
+                "KeyType": "RANGE",
+              },
+            ],
+            "Projection": {
+              "ProjectionType": "ALL",
+            },
+          },
+        ],
         "KeySchema": [
           {
             "AttributeName": "id",
@@ -967,8 +993,34 @@ exports[`The FootballNotificationsLambda stack matches the snapshot for PROD 1`]
             "AttributeName": "id",
             "AttributeType": "S",
           },
+          {
+            "AttributeName": "liveActivityID",
+            "AttributeType": "S",
+          },
+          {
+            "AttributeName": "ttl",
+            "AttributeType": "N",
+          },
         ],
         "BillingMode": "PAY_PER_REQUEST",
+        "GlobalSecondaryIndexes": [
+          {
+            "IndexName": "lastPayload-index",
+            "KeySchema": [
+              {
+                "AttributeName": "liveActivityID",
+                "KeyType": "HASH",
+              },
+              {
+                "AttributeName": "ttl",
+                "KeyType": "RANGE",
+              },
+            ],
+            "Projection": {
+              "ProjectionType": "ALL",
+            },
+          },
+        ],
         "KeySchema": [
           {
             "AttributeName": "id",

cdk/lib/footballnotificationslambda.ts

@@ -10,7 +10,12 @@ import {
 	TreatMissingData,
 	Unit,
 } from 'aws-cdk-lib/aws-cloudwatch';
-import { AttributeType, BillingMode, Table } from 'aws-cdk-lib/aws-dynamodb';
+import {
+	AttributeType,
+	BillingMode,
+	ProjectionType,
+	Table,
+} from 'aws-cdk-lib/aws-dynamodb';
 import { Schedule } from 'aws-cdk-lib/aws-events';
 import { Effect, PolicyStatement } from 'aws-cdk-lib/aws-iam';
 import { LoggingFormat, Runtime } from 'aws-cdk-lib/aws-lambda';
@@ -71,8 +76,8 @@ export class FootballNotificationsLambda extends GuStack {
 			}),
 		);
 
+		// DynamoDB Push Notifications Table
 		const dynamoTableName = `${stack}-football-notifications-${stage}`;
-
 		const dynamoTable = new Table(this, 'DynamoTable', {
 			tableName: dynamoTableName,
 			partitionKey: { name: 'notificationId', type: AttributeType.STRING },
@@ -85,13 +90,11 @@ export class FootballNotificationsLambda extends GuStack {
 			logicalId: 'DynamoTable',
 			reason: 'Retaining a stateful resource previously defined in YAML',
 		});
-
 		Tags.of(dynamoTable).add('devx-backup-enabled', 'true');
 
-		// Live Activities DynamoDB Table
+		// DynamoDB Live Activities Payloads Table
 		const liveActivitiesAppName = 'liveactivities';
 		const liveActivitiesDynamoTableName = `${stack}-${liveActivitiesAppName}-payload-${stage}`;
-
 		const liveActivitiesDynamoTable = new Table(
 			this,
 			'LiveActivitiesDynamoTable',
@@ -104,6 +107,14 @@ export class FootballNotificationsLambda extends GuStack {
 		);
 		Tags.of(liveActivitiesDynamoTable).add('devx-backup-enabled', 'true');
 
+		// GSI is used to determine if an untracked state change event has occurred, ie. goal was overruled.
+		liveActivitiesDynamoTable.addGlobalSecondaryIndex({
+			indexName: 'lastPayload-index',
+			partitionKey: { name: 'liveActivityID', type: AttributeType.STRING },
+			sortKey: { name: 'ttl', type: AttributeType.NUMBER },
+			projectionType: ProjectionType.ALL,
+		});
+
 		footballnotificationslambda.addToRolePolicy(
 			new PolicyStatement({
 				actions: [

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[LiveActivityPayload, DynamoMatchLiveActivity](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)
 
@@ -133,9 +134,7 @@ object Lambda extends Logging {
 class NotificationHandler(configuration: Configuration, apiClient: NotificationsApiClient, dynamoDBClient: AmazonDynamoDBAsync, tableName: String) extends Logging {
 
   lazy val notificationSender = new NotificationSender(apiClient)
-
   lazy val matchStatusNotificationBuilder = new MatchStatusNotificationBuilder(configuration.mapiHost)
-
   lazy val eventConsumer = new EventConsumer(matchStatusNotificationBuilder)
 
   lazy val distinctCheck = new DynamoDistinctCheck[NotificationPayload, DynamoMatchNotification](
@@ -144,7 +143,7 @@ class NotificationHandler(configuration: Configuration, apiClient: Notifications
     partitionKeyName = "notificationId",
     toDynamoModel = payload => DynamoMatchNotification(payload)
   )
-  lazy val eventFilter = new EventFilter[NotificationPayload, DynamoMatchNotification](distinctCheck)
+  lazy val eventFilter = new EventFilter[NotificationPayload, DynamoMatchNotification](distinctCheck) // TODO handle match state changes
 
   def process(rawEvents: List[MatchDataWithArticle]): Future[Unit] = {
     val notifications = rawEvents.flatMap(eventConsumer.eventsToNotifications)
@@ -161,9 +160,7 @@ class LiveActivityHandler(configuration: Configuration, dynamoDBClient: AmazonDy
     s"liveactivities-eventbus-${configuration.stage}"
 
   lazy val liveActivityPusher = new LiveActivityPusher(eventBusName, logger)
-
   lazy val matchStatusLiveActivityPayloadBuilder = new MatchStatusLiveActivityPayloadBuilder()
-
   lazy val liveActivityEventConsumer = new LiveActivityEventConsumer(matchStatusLiveActivityPayloadBuilder)
 
   lazy val liveActivityDistinctCheck = new DynamoDistinctCheck[LiveActivityPayload, DynamoMatchLiveActivity](
@@ -176,6 +173,7 @@ class LiveActivityHandler(configuration: Configuration, dynamoDBClient: AmazonDy
 
   def process(rawEvents: List[MatchDataWithArticle]): Future[Unit] = {
     val liveActivities = rawEvents.flatMap(liveActivityEventConsumer.eventsToLiveActivityPayload)
+    // todo payload is marked as statechange payload now
 
     for {
       filteredLiveActivities <- liveActivityEventFilter.filterDynamoEventsForLiveActivities(liveActivities)

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

@@ -5,7 +5,8 @@ import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync
 import com.gu.mobile.notifications.client.models.{FootballMatchStatusPayload, Payload}
 import org.scanamo.{DynamoFormat, ScanamoAsync, Table}
 import DynamoDistinctCheck.{Distinct, DistinctStatus, Duplicate, Unknown}
-import com.gu.mobile.notifications.client.models.liveActitivites.LiveActivityPayload
+import com.gu.mobile.notifications.client.models.liveActitivites.{FootballMatchContentState, LiveActivityPayload}
+import com.gu.mobile.notifications.football.Lambda.tableName
 import com.gu.mobile.notifications.football.Logging
 import play.api.libs.json.Json
 
@@ -49,6 +50,54 @@ object DynamoMatchLiveActivity {
   }
 }
 
+// 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[A, D](client: AmazonDynamoDBAsync, tableName: String) extends Logging {
+  def isMatchStateIdentical(id: String, state: FootballMatchContentState)(implicit ec: ExecutionContext): Future[Boolean] = {
+    import org.scanamo.syntax._
+    import org.scanamo.generic.auto._
+
+    lazy val scanamoAsync: ScanamoAsync = ScanamoAsync(client)
+    lazy val payloadTable = Table[DynamoMatchLiveActivity](tableName)
+    lazy val lastPayloadIndex = payloadTable.index("lastPayload-index") // only live activities payload table has an index.
+
+    val gsiQuery = lastPayloadIndex.query("liveActivityID" -> id)
+
+    scanamoAsync.exec(gsiQuery).map { rows =>
+      val latestPayloadState = rows
+        .collect { case Right(row) => row }
+        .sortBy(-_.ttl) // most recent first
+        .flatMap(row => footballStateFromPayload(row.payload))
+        .headOption
+
+      // TODO double check waht scanamo supports.
+//      val gsiQuery = lastPayloadIndex.descending.limit(1).query("liveActivityID" -> id)
+//
+//      scanamoAsync.exec(gsiQuery).map { rows =>
+//        rows
+//          .collect { case Right(row) => row }
+//          .flatMap(row => footballStateFromPayload(row.payload))
+//          .headOption
+//          .contains(state)
+//      }
+//
+      // TODO check json serialisation diff works
+      latestPayloadState.contains(state) // TODO check the whole state or just score/penalties?
+    } recover {
+      case e =>
+        logger.error(s"Failure while checking for dynamodb GSI for match state $tableName: ${e.getMessage}.")
+        false
+    }
+  }
+  // The stored `payload` column is the JSON-serialised LiveActivityPayload; the match state we diff
+  // against is its `broadcastContentStateData` subfield (when it is a football content state).
+  private def footballStateFromPayload(payloadJson: String): Option[FootballMatchContentState] =
+    Json.parse(payloadJson).as[LiveActivityPayload].broadcastContentStateData.collect {
+      case footballState: FootballMatchContentState => footballState
+    }
+}
+
+
 object DynamoDistinctCheck {
   sealed trait DistinctStatus
   case object Distinct extends DistinctStatus

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

@@ -35,7 +35,8 @@ class EventConsumer(
         "resumed",
         "abandoned",
         "cancelled",
-        "postponed"
+        "postponed",
+        "state-changed" // remove in order push state change caused by VAR reversal.
       )
 
     val filteredMatchData = matchData.copy(allEvents =

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

@@ -4,13 +4,16 @@ import java.util.UUID
 import java.util.concurrent.atomic.AtomicReference
 import java.util.function.UnaryOperator
 import com.gu.mobile.notifications.client.models.Payload
-import com.gu.mobile.notifications.client.models.liveActitivites.LiveActivityPayload
+import com.gu.mobile.notifications.client.models.liveActitivites.{FootballMatchContentState, LiveActivityPayload, UpdateStateChangeLiveActivityEvent}
 import com.gu.mobile.notifications.football.lib.DynamoDistinctCheck.{Distinct, Duplicate}
 import com.gu.mobile.notifications.football.Logging
 
 import scala.concurrent.{ExecutionContext, Future}
+import scala.util.control.NonFatal
 
-class EventFilter[A <: Payload, D](distinctCheck: DynamoDistinctCheck[A, D]) extends Logging {
+class EventFilter[A <: Payload, D](
+    distinctCheck: DynamoDistinctCheck[A, D]
+) extends Logging {
   private val processedEvents = new AtomicReference[Set[UUID]](Set.empty)
 
   private def cache(eventId: UUID): Unit = {
@@ -55,6 +58,7 @@ class EventFilter[A <: Payload, D](distinctCheck: DynamoDistinctCheck[A, D]) ext
   }
 
 
+  // todo state change will always be here.
   private def filterOutEndEventsNotReceivedInIsolation(events: List[LiveActivityPayload]): List[LiveActivityPayload] = {
     val (endEvents, updateEvents) = events.partition(_.isEndPayload)
     val updateEventMatchIds = updateEvents.map(_.liveActivityID).toSet
@@ -65,18 +69,34 @@ class EventFilter[A <: Payload, D](distinctCheck: DynamoDistinctCheck[A, D]) ext
     isolatedEndEvents ++ updateEvents
   }
 
+  private def filterOutStateChangeEventsNotReceivedInIsolation(events: List[LiveActivityPayload]): List[LiveActivityPayload] = {
+    val (stateChangeEvents, otherEvents) = events.partition(_.eventType == com.gu.mobile.notifications.client.models.liveActitivites.UpdateStateChangeLiveActivityEvent)
+    val otherEventMatchIds = otherEvents.map(_.liveActivityID).toSet
+    val (isolatedStateChangeEvents, superfluousStateChangeEvents) =
+      stateChangeEvents.partition(event => !otherEventMatchIds.contains(event.liveActivityID))
+
+    // todo clean this up and change partition to filter. Hypothesis is we will have a 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 ++ otherEvents
+  }
+
   // 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 {
       newEvents <- filterAsync(dynamoEvents)(item => isUniqueRecord(ev(item)))
+      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(newEvents)
-      processedEvents <- Future.traverse(eventsToProcess) { item =>
+      eventsWithoutEndEvents = filterOutEndEventsNotReceivedInIsolation(eventsWithoutStateChangeEvents)
+
+      processedEvents <- Future.traverse(eventsWithoutEndEvents) { item =>
         distinctCheck.insertEvent(ev(item)).map {
           case Distinct => {
             cache(item.id)