Skip to content

Commit 673d5f6

Browse files
vlbeevlbee
authored
Add liveActivityEventPusher to football lambda. (#1761)
* Add liveActivityEventPusher to football lambda. * Remove competition id * Remove unneccessary logging * Disable pushing in PROD
1 parent c7719ac commit 673d5f6

6 files changed

Lines changed: 312 additions & 28 deletions

File tree

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

Lines changed: 69 additions & 17 deletions
Original file line numberDiff line numberDiff line change
@@ -6,15 +6,19 @@ 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, EventConsumer, EventFilter, FootballData, NotificationHttpProvider, NotificationSender, NotificationsApiClient, PaFootballClient, SyntheticMatchEventGenerator}
10-
import com.gu.mobile.notifications.football.notificationbuilders.MatchStatusNotificationBuilder
9+
import com.gu.mobile.notifications.football.lib.{ArticleSearcher, DynamoDistinctCheck, EventConsumer, LiveActivityEventConsumer, EventFilter, FootballData, LiveActivityPusher, NotificationHttpProvider, NotificationSender, NotificationsApiClient, PaFootballClient, SyntheticMatchEventGenerator}
10+
import com.gu.mobile.notifications.football.notificationbuilders.{MatchStatusLiveActivityPayloadBuilder, MatchStatusNotificationBuilder}
1111
import play.api.libs.json.Json
1212

1313
import scala.concurrent.ExecutionContext.Implicits.global
1414
import scala.concurrent.duration.Duration
1515
import scala.concurrent.{Await, TimeoutException}
1616
import scala.io.Source
1717
import scala.util.Failure
18+
import com.gu.mobile.notifications.client.models.NotificationPayload
19+
import com.gu.mobile.notifications.client.models.liveActitivites.LiveActivityPayload
20+
import com.gu.mobile.notifications.football.models.MatchDataWithArticle
21+
import scala.concurrent.Future
1822

1923
object Lambda extends Logging {
2024

@@ -48,19 +52,18 @@ object Lambda extends Logging {
4852

4953
val apiClient = new NotificationsApiClient(configuration)
5054

51-
lazy val notificationSender = new NotificationSender(apiClient)
52-
53-
lazy val matchStatusNotificationBuilder = new MatchStatusNotificationBuilder(configuration.mapiHost)
54-
55-
lazy val eventConsumer = new EventConsumer(matchStatusNotificationBuilder)
55+
lazy val footballData = new FootballData(paFootballClient, syntheticMatchEventGenerator)
5656

57-
lazy val distinctCheck = new DynamoDistinctCheck(dynamoDBClient, tableName)
57+
lazy val articleSearcher = new ArticleSearcher(capiClient)
5858

59-
lazy val eventFilter = new EventFilter(distinctCheck)
59+
lazy val notificationHandler = new NotificationHandler(configuration, apiClient, dynamoDBClient, tableName)
6060

61-
lazy val footballData = new FootballData(paFootballClient, syntheticMatchEventGenerator)
61+
lazy val liveActivityHandler = new LiveActivityHandler(configuration, dynamoDBClient, tableName)
6262

63-
lazy val articleSearcher = new ArticleSearcher(capiClient)
63+
// live activities //
64+
lazy val liveActivityPusher = new LiveActivityPusher()
65+
lazy val matchStatusLiveActivityPayloadBuilder = new MatchStatusLiveActivityPayloadBuilder(configuration.mapiHost)
66+
lazy val liveActivityEventConsumer = new LiveActivityEventConsumer(matchStatusLiveActivityPayloadBuilder)
6467

6568
def getZonedDateTime(): ZonedDateTime = {
6669
val zonedDateTime = if (configuration.stage == "CODE") {
@@ -87,14 +90,16 @@ object Lambda extends Logging {
8790

8891
logContainer()
8992

90-
val processing = footballData.pollFootballData(getZonedDateTime())
91-
.flatMap(articleSearcher.tryToMatchWithCapiArticle)
92-
.map(_.flatMap(eventConsumer.eventsToNotifications))
93-
.flatMap(eventFilter.filterNotifications)
94-
.flatMap(notificationSender.sendNotifications)
93+
val articlesMatches = for {
94+
footballDataResult <- footballData.pollFootballData(getZonedDateTime())
95+
articleMatches <- articleSearcher.tryToMatchWithCapiArticle(footballDataResult)
96+
} yield articleMatches
97+
98+
val notificationsProcessing = articlesMatches.flatMap(notificationHandler.process)
99+
val liveActivitiesProcessing = articlesMatches.flatMap(liveActivityHandler.process)
95100

96101
// we're in a lambda so we do need to block the main thread until processing is finished
97-
val result = Await.ready(processing, Duration(40, TimeUnit.SECONDS))
102+
val result = Await.ready(Future.sequence(List(notificationsProcessing, liveActivitiesProcessing)), Duration(40, TimeUnit.SECONDS))
98103

99104
result.value match {
100105
case Some(Failure(e: TimeoutException)) => logger.error("Task timed out", e)
@@ -116,3 +121,50 @@ object Lambda extends Logging {
116121
}
117122

118123
}
124+
125+
class NotificationHandler(configuration: Configuration, apiClient: NotificationsApiClient, dynamoDBClient: AmazonDynamoDBAsync, tableName: String) extends Logging {
126+
127+
lazy val notificationSender = new NotificationSender(apiClient)
128+
129+
lazy val matchStatusNotificationBuilder = new MatchStatusNotificationBuilder(configuration.mapiHost)
130+
131+
lazy val eventConsumer = new EventConsumer(matchStatusNotificationBuilder)
132+
133+
lazy val distinctCheck = new DynamoDistinctCheck[NotificationPayload](dynamoDBClient, tableName)
134+
135+
lazy val eventFilter = new EventFilter[NotificationPayload](distinctCheck)
136+
137+
def process(rawEvents: List[MatchDataWithArticle]): Future[Unit] = {
138+
val notifications = rawEvents.flatMap(eventConsumer.eventsToNotifications)
139+
for {
140+
filteredNotifications <- eventFilter.filterNotifications(notifications)
141+
result <- notificationSender.sendNotifications(filteredNotifications)
142+
} yield result
143+
}
144+
}
145+
146+
class LiveActivityHandler(configuration: Configuration, dynamoDBClient: AmazonDynamoDBAsync, tableName: String) extends Logging {
147+
148+
lazy val liveActivityPusher = new LiveActivityPusher()
149+
150+
lazy val matchStatusLiveActivityPayloadBuilder = new MatchStatusLiveActivityPayloadBuilder(configuration.mapiHost)
151+
152+
lazy val liveActivityEventConsumer = new LiveActivityEventConsumer(matchStatusLiveActivityPayloadBuilder)
153+
154+
lazy val distinctCheck = new DynamoDistinctCheck[LiveActivityPayload](dynamoDBClient, tableName)
155+
156+
lazy val eventFilter = new EventFilter[LiveActivityPayload](distinctCheck)
157+
158+
def process(rawEvents: List[MatchDataWithArticle]): Future[Unit] = {
159+
val liveActivities = rawEvents.flatMap(liveActivityEventConsumer.eventsToLiveActivityPayload)
160+
161+
for {
162+
// filteredLiveActivities <- eventFilter.filterNotifications(liveActivities)
163+
result <- if (configuration.stage != "PROD") {
164+
liveActivityPusher.pushEvents(liveActivities)
165+
} else {
166+
Future.successful(())
167+
}
168+
} yield result
169+
}
170+
}

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

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@ package com.gu.mobile.notifications.football.lib
22

33
import scala.concurrent.{ExecutionContext, Future}
44
import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync
5-
import com.gu.mobile.notifications.client.models.{FootballMatchStatusPayload, NotificationPayload}
5+
import com.gu.mobile.notifications.client.models.{FootballMatchStatusPayload, Payload}
66
import org.scanamo.{ScanamoAsync, Table}
77
import DynamoDistinctCheck.{Distinct, DistinctStatus, Duplicate, Unknown}
88
import com.gu.mobile.notifications.football.Logging
@@ -16,7 +16,7 @@ case class DynamoMatchNotification(
1616
)
1717

1818
object DynamoMatchNotification {
19-
def apply(notification: NotificationPayload): DynamoMatchNotification = {
19+
def apply[A <: Payload](notification: A)(implicit writes: play.api.libs.json.Writes[A]): DynamoMatchNotification = {
2020
val matchId = PartialFunction.condOpt(notification) {
2121
case p: FootballMatchStatusPayload => p.matchId
2222
}
@@ -37,8 +37,8 @@ object DynamoDistinctCheck {
3737
case object Unknown extends DistinctStatus
3838
}
3939

40-
class DynamoDistinctCheck(client: AmazonDynamoDBAsync, tableName: String) extends Logging {
41-
def insertNotification(notification: NotificationPayload)(implicit ec: ExecutionContext): Future[DistinctStatus] = {
40+
class DynamoDistinctCheck[A <: Payload](client: AmazonDynamoDBAsync, tableName: String) extends Logging {
41+
def insertNotification(notification: A)(implicit ec: ExecutionContext, writes: play.api.libs.json.Writes[A]): Future[DistinctStatus] = {
4242
import org.scanamo.syntax._
4343
import org.scanamo.generic.auto._
4444

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

Lines changed: 51 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,13 @@
11
package com.gu.mobile.notifications.football.lib
22

33
import com.gu.mobile.notifications.client.models.NotificationPayload
4+
import com.gu.mobile.notifications.client.models.liveActitivites.LiveActivityPayload
45
import com.gu.mobile.notifications.football.Logging
56
import com.gu.mobile.notifications.football.models.{FootballMatchEvent, MatchDataWithArticle}
6-
import com.gu.mobile.notifications.football.notificationbuilders.MatchStatusNotificationBuilder
7+
import com.gu.mobile.notifications.football.notificationbuilders.{
8+
MatchStatusLiveActivityPayloadBuilder,
9+
MatchStatusNotificationBuilder
10+
}
711
import pa.{MatchDay, MatchEvent}
812

913
class EventConsumer(
@@ -15,7 +19,6 @@ class EventConsumer(
1519
receiveEvent(matchData.matchDay, matchData.allEvents, event, matchData.articleId)
1620
}
1721
}
18-
1922
def receiveEvent(matchDay: MatchDay, events: List[MatchEvent], event: MatchEvent, articleId: Option[String]): List[NotificationPayload] = {
2023
logger.debug(s"Processing event $event for match ${matchDay.id}")
2124
val previousEvents = events.takeWhile(_ != event)
@@ -29,3 +32,49 @@ class EventConsumer(
2932
} getOrElse Nil
3033
}
3134
}
35+
36+
class LiveActivityEventConsumer(matchStatusLiveActivityPayloadBuilder: MatchStatusLiveActivityPayloadBuilder) extends Logging {
37+
38+
def eventsToLiveActivityPayload(
39+
matchData: MatchDataWithArticle
40+
): List[LiveActivityPayload] = {
41+
matchData.allEvents.flatMap { event =>
42+
processForLiveActivities(
43+
matchData.matchDay,
44+
matchData.allEvents,
45+
event,
46+
matchData.articleId
47+
)
48+
}
49+
}
50+
51+
def processForLiveActivities(
52+
matchDay: MatchDay,
53+
events: List[MatchEvent],
54+
event: MatchEvent,
55+
articleId: Option[String]
56+
): List[LiveActivityPayload] = {
57+
logger.debug(
58+
s"Processing live activity event $event for match ${matchDay.id}"
59+
)
60+
61+
val previousEvents = events.takeWhile(_ != event)
62+
63+
FootballMatchEvent.fromPaMatchEvent(matchDay.homeTeam, matchDay.awayTeam)(
64+
event
65+
) map { ev =>
66+
// Build live activity payload
67+
List(
68+
matchStatusLiveActivityPayloadBuilder.build(
69+
triggeringEvent = ev,
70+
matchInfo = matchDay,
71+
previousEvents = previousEvents.flatMap(
72+
FootballMatchEvent
73+
.fromPaMatchEvent(matchDay.homeTeam, matchDay.awayTeam)(_)
74+
),
75+
articleId = articleId
76+
)
77+
)
78+
} getOrElse Nil
79+
}
80+
}

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

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -4,13 +4,13 @@ import java.util.UUID
44
import java.util.concurrent.atomic.AtomicReference
55
import java.util.function.UnaryOperator
66

7-
import com.gu.mobile.notifications.client.models.NotificationPayload
7+
import com.gu.mobile.notifications.client.models.Payload
88
import com.gu.mobile.notifications.football.lib.DynamoDistinctCheck.{Distinct, Duplicate}
99
import com.gu.mobile.notifications.football.Logging
1010

1111
import scala.concurrent.{ExecutionContext, Future}
1212

13-
class EventFilter(distinctCheck: DynamoDistinctCheck) extends Logging {
13+
class EventFilter[A <: Payload](distinctCheck: DynamoDistinctCheck[A]) extends Logging {
1414
private val processedEvents = new AtomicReference[Set[UUID]](Set.empty)
1515

1616
private def cache(eventId: UUID): Unit = {
@@ -19,7 +19,7 @@ class EventFilter(distinctCheck: DynamoDistinctCheck) extends Logging {
1919
})
2020
}
2121

22-
private def filterNotification(notification: NotificationPayload)(implicit ec: ExecutionContext): Future[Option[NotificationPayload]] = {
22+
private def filterNotification(notification: A)(implicit ec: ExecutionContext, writes: play.api.libs.json.Writes[A]): Future[Option[A]] = {
2323
if (!processedEvents.get.contains(notification.id)) {
2424
distinctCheck.insertNotification(notification).map {
2525
case Distinct =>
@@ -36,7 +36,7 @@ class EventFilter(distinctCheck: DynamoDistinctCheck) extends Logging {
3636
}
3737
}
3838

39-
def filterNotifications(notifications: List[NotificationPayload])(implicit ec: ExecutionContext): Future[List[NotificationPayload]] = {
39+
def filterNotifications(notifications: List[A])(implicit ec: ExecutionContext, writes: play.api.libs.json.Writes[A]): Future[List[A]] = {
4040
Future.traverse(notifications)(filterNotification).map(_.flatten)
4141
}
42-
}
42+
}

‎football/src/main/scala/com/gu/mobile/notifications/football/lib/LiveActivityPusher.scala‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
package com.gu.mobile.notifications.football.lib
2+
3+
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
6+
import play.api.libs.json.Json
7+
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
21+
22+
class LiveActivityPusher extends Logging {
23+
24+
private val eventBusName =
25+
"liveactivities-eventbus-CODE"
26+
private val eventBridgeClient =
27+
EventBridgeClient
28+
.builder()
29+
.build()
30+
31+
def pushEvents(events: List[LiveActivityPayload])(implicit ec: ExecutionContext): Future[Unit] = {
32+
33+
logger.info(
34+
"Eventbus pusher: number of events to push: " + events.size
35+
)
36+
Future.traverse(events)(pushToEventbus).map(_ => ())
37+
}
38+
39+
def pushToEventbus(payload: LiveActivityPayload)(implicit ec: ExecutionContext): Future[Unit] = {
40+
41+
logger.info(
42+
s"Eventbus pusher: Processing event with id ${payload.id}"
43+
)
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 {
51+
52+
// TODO for the eventbus to direct the event to the right place
53+
val activityType = payload.eventType match {
54+
case CreateChannel => "channel-create"
55+
case StartLiveActivity => "broadcast-start"
56+
case UpdateLiveActivity => "broadcast-update"
57+
case EndLiveActivity => "broadcast-end"
58+
case DeleteChannel => "channel-delete"
59+
case _ => "unknown"
60+
}
61+
62+
val result = Try {
63+
val entry = PutEventsRequestEntry
64+
.builder()
65+
.source("football-lambda")
66+
.detailType(activityType)
67+
.detail(Json.toJson(payload).toString())
68+
.eventBusName(eventBusName)
69+
.build()
70+
71+
val request = PutEventsRequest
72+
.builder()
73+
.entries(entry)
74+
.build()
75+
76+
val response: PutEventsResponse = eventBridgeClient.putEvents(request)
77+
78+
// todo - alert for failed entry counts?
79+
logger.info(
80+
s"Eventbus pusher: Event published with event is ${payload.id}. Failed entry count: ${response.failedEntryCount()}"
81+
)
82+
}
83+
84+
result match {
85+
case Success(_) => {
86+
logger.info(
87+
s"Eventbus pusher: Successfully processed event with id ${payload.id}"
88+
)
89+
Future.successful(())
90+
}
91+
case Failure(e) => {
92+
logger.error(
93+
s"Eventbus pusher: Failed to publish event with id ${payload.id}: ${e.getMessage}"
94+
)
95+
Future.failed(e)
96+
}
97+
}
98+
}
99+
}
100+
}

0 commit comments

Comments
 (0)