Skip to content

Commit 05910ab

Browse files
marjisoundmarjisound
authored
Merge pull request #1787 from guardian/complete-live-activity-life-cycle
Complete live activity life cycle
2 parents f4736a3 + 7b3b0b9 commit 05910ab

11 files changed

Lines changed: 135 additions & 92 deletions

File tree

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

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,15 +13,18 @@ import com.gu.mobile.notifications.client.models.Payload
1313

1414
sealed trait EventSource
1515
case object FootballLambda extends EventSource
16+
case object ChannelManagerLambda extends EventSource
1617

1718
object EventSource {
1819
implicit val format: Format[EventSource] = new Format[EventSource] {
1920
def writes(source: EventSource): JsValue = source match {
2021
case FootballLambda => JsString("football-lambda")
22+
case ChannelManagerLambda => JsString("channel-manager-lambda")
2123
}
2224

2325
def reads(json: JsValue): JsResult[EventSource] = json match {
2426
case JsString("football-lambda") => JsSuccess(FootballLambda)
27+
case JsString("channel-manager-lambda") => JsSuccess(ChannelManagerLambda)
2528
case JsString(other) => JsError(s"Invalid EventSource: $other")
2629
case _ => JsError("EventSource must be a string")
2730
}

‎build.sbt‎

Lines changed: 38 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -248,33 +248,33 @@ lazy val apiModels = {
248248

249249
def lambda(projectName: String, directoryName: String, mainClassName: Option[String] = None): Project =
250250
Project(projectName, file(directoryName))
251-
.enablePlugins(AssemblyPlugin)
252-
.settings(
253-
organization := "com.gu",
254-
resolvers ++= Seq(
255-
"Guardian GitHub Releases" at "https://guardian.github.com/maven/repo-releases",
256-
"snapshots" at "https://oss.sonatype.org/content/repositories/snapshots"
257-
),
258-
libraryDependencies ++= Seq(
259-
"com.amazonaws" % "aws-lambda-java-core" % "1.4.0",
260-
"org.slf4j" % "slf4j-api" % slf4jVersion,
261-
"com.gu" %% "simple-configuration-core" % simpleConfigurationVersion,
262-
"com.gu" %% "simple-configuration-ssm" % simpleConfigurationVersion,
263-
"ch.qos.logback" % "logback-classic" % logbackVersion,
264-
"net.logstash.logback" % "logstash-logback-encoder" % "8.1",
265-
specs2 % Test
266-
),
267-
assemblyJarName := s"$projectName.jar",
268-
assembly / assemblyMergeStrategy := {
269-
case "META-INF/MANIFEST.MF" => MergeStrategy.discard
270-
case _ => MergeStrategy.first
271-
},
272-
Test / run / fork := true,
273-
scalacOptions := compilerOptions,
274-
mainClass := mainClassName,
275-
// Workaround Mockito causes deadlock on SBT classloaders: https://github.com/sbt/sbt/issues/3022
276-
Test / parallelExecution := false
277-
)
251+
.enablePlugins(AssemblyPlugin)
252+
.settings(
253+
organization := "com.gu",
254+
resolvers ++= Seq(
255+
"Guardian GitHub Releases" at "https://guardian.github.com/maven/repo-releases",
256+
"snapshots" at "https://oss.sonatype.org/content/repositories/snapshots"
257+
),
258+
libraryDependencies ++= Seq(
259+
"com.amazonaws" % "aws-lambda-java-core" % "1.4.0",
260+
"org.slf4j" % "slf4j-api" % slf4jVersion,
261+
"com.gu" %% "simple-configuration-core" % simpleConfigurationVersion,
262+
"com.gu" %% "simple-configuration-ssm" % simpleConfigurationVersion,
263+
"ch.qos.logback" % "logback-classic" % logbackVersion,
264+
"net.logstash.logback" % "logstash-logback-encoder" % "8.1",
265+
specs2 % Test
266+
),
267+
assemblyJarName := s"$projectName.jar",
268+
assembly / assemblyMergeStrategy := {
269+
case "META-INF/MANIFEST.MF" => MergeStrategy.discard
270+
case _ => MergeStrategy.first
271+
},
272+
Test / run / fork := true,
273+
scalacOptions := compilerOptions,
274+
mainClass := mainClassName,
275+
// Workaround Mockito causes deadlock on SBT classloaders: https://github.com/sbt/sbt/issues/3022
276+
Test / parallelExecution := false
277+
)
278278

279279
lazy val schedulelambda = lambda("schedule", "schedulelambda")
280280
.dependsOn(commonscheduledynamodb)
@@ -301,8 +301,16 @@ lazy val schedulelambda = lambda("schedule", "schedulelambda")
301301
)
302302
}
303303

304-
lazy val football = lambda("football", "football")
304+
lazy val commonEventBusPusher = project
305305
.dependsOn(apiModels % "test->test", apiModels % "compile->compile")
306+
.settings(Seq(
307+
libraryDependencies ++= Seq(
308+
"software.amazon.awssdk" % "eventbridge" % "2.20.162"
309+
),
310+
))
311+
312+
lazy val football = lambda("football", "football")
313+
.dependsOn(commonEventBusPusher)
306314
.settings(
307315
resolvers += "Guardian GitHub Releases" at "https://guardian.github.com/maven/repo-releases",
308316
libraryDependencies ++= Seq(
@@ -318,7 +326,6 @@ lazy val football = lambda("football", "football")
318326
"io.netty" % "netty-codec-http" % nettyVersion,
319327
"io.netty" % "netty-codec-http2" % nettyVersion,
320328
"io.netty" % "netty-common" % nettyVersion,
321-
"software.amazon.awssdk" % "eventbridge" % "2.20.162"
322329
),
323330
excludeDependencies ++= Seq(
324331
ExclusionRule("org.playframework", "play-ahc-ws_2.13"),
@@ -432,7 +439,7 @@ lazy val notificationworkerlambda = lambda("notificationworkerlambda", "notifica
432439
// Hopefully this workaround can be removed once play-json-extensions either updates to Play 3.0 or is merged into play-json
433440
ExclusionRule(organization = "com.typesafe.play")
434441
),
435-
)
442+
)
436443

437444
lazy val fakebreakingnewslambda = lambda("fakebreakingnewslambda", "fakebreakingnewslambda", Some("fakebreakingnews.LocalRun"))
438445
.dependsOn(common)
@@ -462,7 +469,7 @@ lazy val reportExtractor = lambda("reportextractor", "reportextractor", Some("co
462469

463470
lazy val liveactivities = lambda("liveactivities", "liveactivities", Some("com.gu.liveactivities.LambdaLocalRun"))
464471
.dependsOn(common)
465-
.dependsOn(apiModels % "test->test", apiModels % "compile->compile")
472+
.dependsOn(commonEventBusPusher)
466473
.settings(LocalDynamoDBLiveActivities.settings)
467474
.settings(
468475
libraryDependencies ++= Seq(

‎cdk/lib/__snapshots__/liveactivities.test.ts.snap‎

Lines changed: 20 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -834,6 +834,26 @@ exports[`The LiveActivities stack matches the snapshot for CODE 1`] = `
834834
],
835835
},
836836
},
837+
{
838+
"Action": "events:PutEvents",
839+
"Effect": "Allow",
840+
"Resource": {
841+
"Fn::Join": [
842+
"",
843+
[
844+
"arn:aws:events:",
845+
{
846+
"Ref": "AWS::Region",
847+
},
848+
":",
849+
{
850+
"Ref": "AWS::AccountId",
851+
},
852+
":event-bus/liveactivities-eventbus-CODE",
853+
],
854+
],
855+
},
856+
},
837857
],
838858
"Version": "2012-10-17",
839859
},

‎cdk/lib/liveactivities.ts‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -78,6 +78,15 @@ export class LiveActivities extends GuStack {
7878
}),
7979
);
8080

81+
channelLambda.addToRolePolicy(
82+
new PolicyStatement({
83+
actions: ['events:PutEvents'],
84+
resources: [
85+
`arn:aws:events:${region}:${account}:event-bus/${app}-eventbus-${stage}`,
86+
],
87+
}),
88+
);
89+
8190
const broadcastDlq = new Queue(this, 'BroadcastDlq', {
8291
queueName: `${app}-broadcast-dlq-${stage}`,
8392
visibilityTimeout: Duration.minutes(4),

football/src/main/scala/com/gu/mobile/notifications/football/lib/LiveActivityPusher.scala renamed to commonEventBusPusher/src/main/scala/com/gu/mobile/liveactivities/event/bus/LiveActivityPusher.scala

Lines changed: 17 additions & 36 deletions
Original file line numberDiff line numberDiff line change
@@ -1,28 +1,15 @@
1-
package com.gu.mobile.notifications.football.lib
1+
package com.gu.mobile.liveactivities.event.bus
22

33
import com.gu.mobile.notifications.client.models.liveActitivites._
4-
import com.gu.mobile.notifications.football.Logging
5-
import com.gu.mobile.notifications.football.models.MatchDataWithArticle
4+
import org.slf4j.Logger
65
import play.api.libs.json.Json
76
import software.amazon.awssdk.services.eventbridge.EventBridgeClient
8-
import software.amazon.awssdk.services.eventbridge.model.{
9-
PutEventsRequest,
10-
PutEventsRequestEntry,
11-
PutEventsResponse
12-
}
13-
import scala.util.Try
14-
15-
// Temporary json formatters to test pushing match data - will be replaced once we know what we want to send.
16-
import play.api.libs.json.Writes
17-
import scala.concurrent.Future
18-
import scala.util.Success
19-
import scala.util.Failure
20-
import scala.concurrent.ExecutionContext
7+
import software.amazon.awssdk.services.eventbridge.model.{PutEventsRequest, PutEventsRequestEntry, PutEventsResponse}
218

22-
class LiveActivityPusher extends Logging {
9+
import scala.concurrent.{ExecutionContext, Future}
10+
import scala.util.{Failure, Success, Try}
2311

24-
private val eventBusName =
25-
"liveactivities-eventbus-CODE"
12+
class LiveActivityPusher(eventBusName: String, logger: Logger) {
2613
private val eventBridgeClient =
2714
EventBridgeClient
2815
.builder()
@@ -41,13 +28,6 @@ class LiveActivityPusher extends Logging {
4128
logger.info(
4229
s"Eventbus pusher: Processing event with id ${payload.id}"
4330
)
44-
val jsonDetail = Json.toJson(payload).toString()
45-
if (jsonDetail.isEmpty || jsonDetail == "{}") {
46-
logger.info(
47-
s"Eventbus pusher: Skipping empty event for ${payload.id}"
48-
)
49-
Future.successful(())
50-
} else {
5131
val result = Try {
5232
val entry = PutEventsRequestEntry
5333
.builder()
@@ -64,26 +44,27 @@ class LiveActivityPusher extends Logging {
6444

6545
val response: PutEventsResponse = eventBridgeClient.putEvents(request)
6646

67-
// todo - alert for failed entry counts?
68-
logger.info(
69-
s"Eventbus pusher: Event published with event ${payload.eventType} is ${payload.id}. Failed entry count: ${response.failedEntryCount()}"
70-
)
47+
if (response.failedEntryCount() > 0) {
48+
val error = response.entries().get(0).errorMessage()
49+
throw new RuntimeException(s"EventBridge rejected entry: $error")
50+
} else {
51+
logger.info(
52+
s"Eventbus pusher: Event published with event ${payload.eventType} is ${payload.id}"
53+
)
54+
}
7155
}
7256

7357
result match {
74-
case Success(_) => {
58+
case Success(_) =>
7559
logger.info(
7660
s"Eventbus pusher: Successfully processed live activities event ${payload.eventType} with id ${payload.id}"
7761
)
7862
Future.successful(())
79-
}
80-
case Failure(e) => {
63+
case Failure(e) =>
8164
logger.error(
8265
s"Eventbus pusher: Failed to publish live activities event ${payload.eventType} with id ${payload.id}: ${e.getMessage}"
8366
)
8467
Future.failed(e)
85-
}
8668
}
87-
}
8869
}
89-
}
70+
}

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

Lines changed: 7 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,12 +1,13 @@
11
package com.gu.mobile.notifications.football
22

3-
import java.net.{URI, URL}
3+
import java.net.URI
44
import java.time.ZonedDateTime
55
import java.util.concurrent.TimeUnit
66
import com.amazonaws.regions.Regions
77
import com.amazonaws.services.dynamodbv2.{AmazonDynamoDBAsync, AmazonDynamoDBAsyncClientBuilder}
88
import com.gu.contentapi.client.GuardianContentClient
9-
import com.gu.mobile.notifications.football.lib.{ArticleSearcher, DynamoDistinctCheck, DynamoMatchLiveActivity, DynamoMatchNotification, EventConsumer, EventFilter, FootballData, LiveActivityEventConsumer, LiveActivityPusher, NotificationHttpProvider, NotificationSender, NotificationsApiClient, PaFootballClient, SyntheticMatchEventGenerator}
9+
import com.gu.mobile.liveactivities.event.bus.LiveActivityPusher
10+
import com.gu.mobile.notifications.football.lib.{ArticleSearcher, DynamoDistinctCheck, DynamoMatchLiveActivity, DynamoMatchNotification, EventConsumer, EventFilter, FootballData, LiveActivityEventConsumer, NotificationHttpProvider, NotificationSender, NotificationsApiClient, PaFootballClient, SyntheticMatchEventGenerator}
1011
import com.gu.mobile.notifications.football.notificationbuilders.{MatchStatusLiveActivityPayloadBuilder, MatchStatusNotificationBuilder}
1112
import play.api.libs.json.Json
1213

@@ -147,7 +148,10 @@ class NotificationHandler(configuration: Configuration, apiClient: Notifications
147148

148149
class LiveActivityHandler(configuration: Configuration, dynamoDBClient: AmazonDynamoDBAsync, tableName: String) extends Logging {
149150

150-
lazy val liveActivityPusher = new LiveActivityPusher()
151+
private val eventBusName =
152+
"liveactivities-eventbus-CODE"
153+
154+
lazy val liveActivityPusher = new LiveActivityPusher(eventBusName, logger)
151155

152156
lazy val matchStatusLiveActivityPayloadBuilder = new MatchStatusLiveActivityPayloadBuilder(configuration.mapiHost)
153157

‎liveactivities/src/main/scala/com/gu/liveactivities/BroadcastLambda.scala‎

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,10 @@ object BroadcastLambda extends RequestStreamHandler with Lambda with Logging {
3434
case JsSuccess(request, _) => {
3535

3636
// If we see these errors there is a misconfiguration in the eventbridge routing rules.
37-
if (request.`detail-type` != UpdateLiveActivityEvent && request.`detail-type` != EndLiveActivityEvent) {
37+
if (request.`detail-type` != UpdateLiveActivityEvent &&
38+
request.`detail-type` != EndLiveActivityEvent &&
39+
request.`detail-type` != StartLiveActivityEvent
40+
) {
3841
logger.error(s"Unexpected eventbridge event type: ${request.`detail-type`}")
3942
throw new Exception(s"Unexpected event type: ${request.`detail-type`}")
4043
}
@@ -83,11 +86,6 @@ object BroadcastLambda extends RequestStreamHandler with Lambda with Logging {
8386
Future.failed(new LiveActivityInvalidStateException(matchId, "Channel not active"))
8487
} else Future.successful(())
8588

86-
_ <- if (!mapping.isLive) {
87-
logger.error(s"Event not live for match ID $matchId")
88-
Future.failed(new LiveActivityInvalidStateException(matchId, "Event not live"))
89-
} else Future.successful(())
90-
9189
_ <- if (mapping.lastEventId.contains(eventId)) {
9290
logger.warn(s"Duplicate event ID $eventId for match ID $matchId")
9391
Future.failed(new LiveActivityInvalidStateException(matchId, "Duplicate event ID"))
@@ -103,9 +101,9 @@ object BroadcastLambda extends RequestStreamHandler with Lambda with Logging {
103101
_ = logger.info(s"Sending broadcast for match ID $matchId to channel ID ${mapping.channelId}")
104102
broadcastPayload = BroadcastBody(contentState, shouldEndBroadcast)
105103
_ <- broadcastApiClient.sendToChannel(mapping.channelId, None, None, broadcastPayload)
106-
_ = logger.info(s"Broadcast ${if(shouldEndBroadcast)"END"} sent successfully for match ID $matchId to channel ID ${mapping.channelId}")
104+
_ = logger.info(s"Broadcast ${requestPayload.eventType.asString} sent successfully for match ID $matchId to channel ID ${mapping.channelId}")
107105

108-
_ <- repository.updateMappingLastEvent(matchId, Some(eventId), Some(eventTime))
106+
_ <- repository.updateMappingLiveAndLastEvent(matchId, isLive = !shouldEndBroadcast, Some(eventId), Some(eventTime))
109107
_ = logger.info(s"Record updated successfully for match ID $matchId")
110108
} yield mapping.channelId
111109

‎liveactivities/src/main/scala/com/gu/liveactivities/ChannelManagerLambda.scala‎

Lines changed: 22 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -4,25 +4,20 @@ import com.amazonaws.services.lambda.runtime.{Context, RequestStreamHandler}
44
import com.gu.liveactivities.models.LiveActivityData
55
import com.gu.liveactivities.service.ChannelApiClient
66
import com.gu.liveactivities.util.Logging
7-
import com.gu.mobile.notifications.client.models.liveActitivites.{CreateChannelEvent, DeleteChannelEvent, EventBridgeEvent, LiveActivityPayload}
8-
import play.api.libs.json.{Format, JsError, JsSuccess, Json}
7+
import com.gu.mobile.notifications.client.models.liveActitivites._
8+
import play.api.libs.json.{JsError, JsSuccess, Json}
99

1010
import java.io.{InputStream, OutputStream}
11+
import java.util.{Date, UUID}
1112
import scala.concurrent.duration.DurationInt
1213
import scala.concurrent.{Await, Future}
1314
import scala.util.{Failure, Success, Try}
1415

15-
case class ChannelRequest(matchId: String, competitionId: Option[String], eventData: Option[LiveActivityData], toCreate: Boolean)
16-
17-
object ChannelRequest {
18-
implicit val jf: Format[ChannelRequest] = Json.format[ChannelRequest]
19-
}
20-
2116
object ChannelManagerLambda extends RequestStreamHandler with Lambda with Logging {
2217

2318
private val channelApiClient = new ChannelApiClient(authentication, config.bundleId, config.sendingToProdServer)
2419

25-
private def processCreateChannelRequest(matchId: String, eventData: Option[LiveActivityData]): Future[String] = {
20+
private def processCreateChannelRequest(matchId: String, eventData: Option[LiveActivityData], broadcastContentStateData: Option[ContentState]): Future[String] = {
2621
logger.info(s"Received request to create channel for match ID $matchId")
2722

2823
val maybeChannelId = repository.containMapping(matchId)
@@ -36,6 +31,8 @@ object ChannelManagerLambda extends RequestStreamHandler with Lambda with Loggin
3631
channelId <- channelApiClient.createChannel()
3732
_ <- repository.createMapping(matchId, channelId, eventData)
3833
_ = logger.info(s"Channel created with channel ID $channelId for match ID $matchId")
34+
_ <- pushUpdateLiveActivityEvent(matchId, broadcastContentStateData)
35+
_ = logger.info(s"Initial broadcast update pushed to event bus for channel ID $channelId and match ID $matchId")
3936
} yield channelId
4037
}
4138
}
@@ -76,7 +73,7 @@ object ChannelManagerLambda extends RequestStreamHandler with Lambda with Loggin
7673
val channelFuture = request.eventType match {
7774
case CreateChannelEvent =>
7875
val eventData = request.broadcastContentStateData.map(LiveActivityData.toLiveActivityData)
79-
processCreateChannelRequest(request.liveActivityID, eventData)
76+
processCreateChannelRequest(request.liveActivityID, eventData, request.broadcastContentStateData)
8077
case DeleteChannelEvent =>
8178
processCloseChannelRequest(request.liveActivityID)
8279
case other =>
@@ -95,4 +92,19 @@ object ChannelManagerLambda extends RequestStreamHandler with Lambda with Loggin
9592
throw exception
9693
}
9794
}
95+
96+
private def pushUpdateLiveActivityEvent(matchId: String, broadcastContentStateData: Option[ContentState]): Future[Unit] = {
97+
val triggeringEventId = s"football-match-status/$matchId/live-activity-initial-data"
98+
liveActivityPusher.pushToEventbus(
99+
LiveActivityPayload(
100+
id = UUID.nameUUIDFromBytes(triggeringEventId.getBytes),
101+
eventType = UpdateLiveActivityEvent,
102+
liveActivityType = FootballLiveActivity,
103+
liveActivityID = matchId,
104+
dynamoStoreData = None,
105+
broadcastContentStateData = broadcastContentStateData,
106+
eventTimestamp = new Date().getTime,
107+
)
108+
)
109+
}
98110
}

0 commit comments

Comments
 (0)