S28 3665: Cron Batching Cleanup Live Event Task

.env.local

@@ -4,6 +4,8 @@ TESTING_SUPPORT_ENDPOINTS_ENABLED=false
 
 # CRONS
 TASK_START_LIVE_EVENTS_BATCH_SIZE=20
+TASK_CLEANUP_LIVE_EVENTS_BATCH_SIZE=3
+TASK_CLEANUP_LIVE_EVENTS_COOLDOWN=60000
 
 # MEDIA SERVICES
 MEDIA_SERVICE=

charts/pre-api/values.yaml

@@ -59,6 +59,8 @@ java: &javavalues #This is an anchor to reuse settings in Job mode.
     CRON_USER_EMAIL: "dts-pre-app-{{ .Values.global.environment }}@hmcts.net"
     ISSUER: https://sts.windows.net/531ff96d-0ae9-462a-8d2d-bec7c0b42082/
     TASK_START_LIVE_EVENTS_BATCH_SIZE: 20
+    TASK_CLEANUP_LIVE_EVENTS_BATCH_SIZE: 3
+    TASK_CLEANUP_LIVE_EVENTS_COOLDOWN: 60000
 job:
   #DO NOT DELETE, Below line is aliasing all values from Java settings above to job to avoid duplication.
   <<: *javavalues

src/main/java/uk/gov/hmcts/reform/preapi/entities/Booking.java

@@ -24,7 +24,7 @@
 @Entity
 @Table(name = "bookings")
 public class Booking extends CreatedModifiedAtEntity implements ISoftDeletable {
-    @ManyToOne(fetch = FetchType.LAZY)
+    @ManyToOne(fetch = FetchType.EAGER)
     @JoinColumn(name = "case_id", referencedColumnName = "id")
     private Case caseId;
 

src/main/java/uk/gov/hmcts/reform/preapi/entities/Participant.java

@@ -25,7 +25,7 @@
 @Entity
 @Table(name = "participants")
 public class Participant extends CreatedModifiedAtEntity {
-    @ManyToOne(fetch = FetchType.LAZY)
+    @ManyToOne(fetch = FetchType.EAGER)
     @JoinColumn(name = "case_id", referencedColumnName = "id")
     private Case caseId;
 

src/main/java/uk/gov/hmcts/reform/preapi/media/IMediaService.java

@@ -33,4 +33,12 @@ public interface IMediaService {
     void cleanupStoppedLiveEvent(String liveEventId);
 
     void deleteAllStreamingLocatorsAndContentKeyPolicies();
+
+    String triggerProcessingStep1(CaptureSessionDTO captureSession, String captureSessionNoHyphen, UUID recordingId);
+
+    String triggerProcessingStep2(UUID recordingId);
+
+    RecordingStatus verifyFinalAssetExists(UUID recordingId);
+
+    RecordingStatus hasJobCompleted(String transformName, String jobName);
 }

src/main/java/uk/gov/hmcts/reform/preapi/tasks/StartLiveEvents.java

@@ -130,7 +130,13 @@ private void awaitIngestAddresses(List<UUID> captureSessionIds) {
                 Thread.sleep(2000);
                 var liveEvents = mediaService.getLiveEvents();
                 startingCaptureSessions = startingCaptureSessions.stream()
-                    .filter(id -> !tryGetIngestAddress(id, liveEvents))
+                    .filter(id -> {
+                        var result = tryGetIngestAddress(id, liveEvents);
+                        if (result) {
+                            log.info("Ingest address obtained for capture session {}", id);
+                        }
+                        return !result;
+                    })
                     .toList();
             } while (!startingCaptureSessions.isEmpty());
         } catch (Exception e) {

src/main/resources/application.yaml

@@ -107,6 +107,10 @@ flow:
 tasks:
   start-live-event:
     batch-size: ${TASK_START_LIVE_EVENTS_BATCH_SIZE:20}
+  cleanup-live-events:
+    batch-size: ${TASK_CLEANUP_LIVE_EVENTS_BATCH_SIZE:3}
+    cooldown: ${TASK_CLEANUP_LIVE_EVENTS_COOLDOWN:60000} # one minute
+    job-poll-interval: 10000 # 10 seconds
 
 #logging:
 #  level:

src/test/java/uk/gov/hmcts/reform/preapi/media/MediaKindTest.java

@@ -519,11 +519,11 @@ void stopLiveEventRecordingAvailable() throws InterruptedException {
         var jobArgument3 = ArgumentCaptor.forClass(String.class);
         var jobArgument4 = ArgumentCaptor.forClass(String.class);
         verify(mockClient, times(1)).putJob(eq(ENCODE_FROM_INGEST_TRANSFORM), jobArgument.capture(), any(MkJob.class));
-        verify(mockClient, times(2)).getJob(eq(ENCODE_FROM_INGEST_TRANSFORM), jobArgument2.capture());
+        verify(mockClient, times(3)).getJob(eq(ENCODE_FROM_INGEST_TRANSFORM), jobArgument2.capture());
         assertThat(jobArgument.getValue()).startsWith(liveEventName);
         assertThat(jobArgument2.getValue()).startsWith(liveEventName);
         verify(mockClient, times(1)).putJob(eq(ENCODE_FROM_MP4_TRANSFORM), jobArgument3.capture(), any(MkJob.class));
-        verify(mockClient, times(2)).getJob(eq(ENCODE_FROM_MP4_TRANSFORM), jobArgument4.capture());
+        verify(mockClient, times(3)).getJob(eq(ENCODE_FROM_MP4_TRANSFORM), jobArgument4.capture());
         assertThat(jobArgument3.getValue()).startsWith(tempName);
         assertThat(jobArgument4.getValue()).startsWith(tempName);
     }
@@ -610,7 +610,7 @@ void stopLiveEventRecordingFoundRunEncodeTransform() throws InterruptedException
         var jobArgument = ArgumentCaptor.forClass(String.class);
         var jobArgument2 = ArgumentCaptor.forClass(String.class);
         verify(mockClient, times(1)).putJob(eq(ENCODE_FROM_INGEST_TRANSFORM), jobArgument.capture(), any(MkJob.class));
-        verify(mockClient, times(2)).getJob(eq(ENCODE_FROM_INGEST_TRANSFORM), jobArgument2.capture());
+        verify(mockClient, times(3)).getJob(eq(ENCODE_FROM_INGEST_TRANSFORM), jobArgument2.capture());
         assertThat(jobArgument.getValue()).startsWith(liveEventName);
         assertThat(jobArgument2.getValue()).startsWith(liveEventName);
         verify(mockClient, times(1)).stopLiveEvent(liveEventName);
@@ -656,7 +656,7 @@ void stopLiveEventRecordingFoundRunEncodeTransform2() throws InterruptedExceptio
         var jobArgument = ArgumentCaptor.forClass(String.class);
         var jobArgument2 = ArgumentCaptor.forClass(String.class);
         verify(mockClient, times(1)).putJob(eq(ENCODE_FROM_INGEST_TRANSFORM), jobArgument.capture(), any(MkJob.class));
-        verify(mockClient, times(2)).getJob(eq(ENCODE_FROM_INGEST_TRANSFORM), jobArgument2.capture());
+        verify(mockClient, times(3)).getJob(eq(ENCODE_FROM_INGEST_TRANSFORM), jobArgument2.capture());
         assertThat(jobArgument.getValue()).startsWith(liveEventName);
         assertThat(jobArgument2.getValue()).startsWith(liveEventName);
         verify(mockClient, times(1)).stopLiveEvent(liveEventName);