feat: use engage-sdk for continue watching

ReferenceAppKotlin/app/build.gradle

@@ -122,6 +122,9 @@ dependencies {
     implementation("com.squareup.moshi:moshi-kotlin:1.13.0")
     kapt("com.squareup.moshi:moshi-kotlin-codegen:1.13.0")
 
+    // Engage SDK
+    implementation "com.google.android.engage:engage-core:1.4.0"
+
     // Retrofit for HTTP requests
     implementation 'com.squareup.retrofit2:retrofit:2.9.0'
     implementation 'com.squareup.retrofit2:converter-moshi:2.9.0'

ReferenceAppKotlin/app/src/main/java/com/android/tv/reference/MainActivity.kt

@@ -30,6 +30,7 @@ import com.android.tv.reference.deeplink.DeepLinkViewModel
 import com.android.tv.reference.deeplink.DeepLinkViewModelFactory
 import com.android.tv.reference.shared.datamodel.Video
 import com.android.tv.reference.shared.util.Result
+import com.android.tv.reference.watchnext.Publisher
 import timber.log.Timber
 
 /**
@@ -146,4 +147,9 @@ class MainActivity : FragmentActivity() {
             )
         )
     }
+
+    override fun onStop() {
+        super.onStop()
+        Publisher.publishPeriodically(applicationContext)
+    }
 }

ReferenceAppKotlin/app/src/main/java/com/android/tv/reference/watchnext/ClusterRequestFactory.kt

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.android.tv.reference.watchnext
+
+import android.content.Context
+import com.android.tv.reference.shared.datamodel.VideoType
+import com.google.android.engage.common.datamodel.ContinuationCluster
+import com.google.android.engage.service.PublishContinuationClusterRequest
+
+/**
+ * Class in charge of constructing the publishing requests and sending them to their respective
+ * publishers
+ */
+class ClusterRequestFactory(context: Context) {
+
+    private val engageWatchNextService = EngageWatchNextService.getInstance(context)
+
+    /**
+     * [constructContinuationClusterRequest] returns a [PublishContinuationClusterRequest] to be used
+     * by the [EngageServiceWorker] to publish Continuations clusters
+     *
+     * @return PublishContinuationClusterRequest
+     */
+    fun constructContinuationClusterRequest(): PublishContinuationClusterRequest {
+        val continuationList = engageWatchNextService.getAllWatchNextVideos()
+        val continuationCluster = ContinuationCluster.Builder()
+        for (watchNextVideo in continuationList) {
+            val videoType = watchNextVideo.video.videoType
+            if (videoType == VideoType.EPISODE || videoType == VideoType.MOVIE) {
+                continuationCluster.addEntity(
+                    VideoToEngageEntityConverter.convertVideo(
+                        watchNextVideo
+                    )
+                )
+            }
+        }
+        return PublishContinuationClusterRequest.Builder()
+            .setContinuationCluster(continuationCluster.build())
+            .build()
+    }
+}

ReferenceAppKotlin/app/src/main/java/com/android/tv/reference/watchnext/Constants.kt

@@ -0,0 +1,28 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.android.tv.reference.watchnext
+
+/** Constants that are to be used as reference for publishing guidelines */
+object Constants {
+    const val MAX_PUBLISHING_ATTEMPTS: Int = 5
+
+    const val WORKER_NAME_CONTINUATION: String = "Upload Continuation"
+
+    const val PERIODIC_WORKER_NAME_CONTINUATION: String = "Periodically Upload Continuation"
+
+    const val PUBLISH_TYPE: String = "PUBLISH_TYPE"
+    const val PUBLISH_TYPE_CONTINUATION = "PUBLISH_CONTINUATION"
+}

ReferenceAppKotlin/app/src/main/java/com/android/tv/reference/watchnext/EngageServiceBroadcastReceiver.kt

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.android.tv.reference.watchnext
+
+import android.content.BroadcastReceiver
+import android.content.Context
+import android.content.Intent
+import android.util.Log
+import com.android.tv.reference.watchnext.Publisher.publishContinuationClusters
+import com.google.android.engage.service.Intents.ACTION_PUBLISH_CONTINUATION
+
+/** Broadcast Receiver to trigger a one time publish of clusters. */
+class EngageServiceBroadcastReceiver : BroadcastReceiver() {
+
+    override fun onReceive(context: Context?, intent: Intent?) {
+        if (intent == null || context == null) {
+            return
+        }
+        when (intent.action) {
+            ACTION_PUBLISH_CONTINUATION -> publishContinuationClusters(context)
+            else -> Log.e(TAG, "onReceive: Received unrecognized intent: $intent")
+        }
+    }
+
+    private companion object {
+        const val TAG = "EngageBroadcastReceiver"
+    }
+}

ReferenceAppKotlin/app/src/main/java/com/android/tv/reference/watchnext/EngageServiceWorker.kt

@@ -0,0 +1,162 @@
+/*
+ * Copyright 2024 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.android.tv.reference.watchnext
+
+import android.content.Context
+import android.util.Log
+import androidx.work.Worker
+import androidx.work.WorkerParameters
+import com.android.tv.reference.watchnext.Constants.PUBLISH_TYPE
+import com.android.tv.reference.watchnext.Constants.PUBLISH_TYPE_CONTINUATION
+import com.google.android.engage.service.AppEngageException
+import com.google.android.engage.service.AppEngagePublishClient
+import com.google.android.engage.service.AppEngagePublishStatusCode
+import com.google.android.engage.service.PublishStatusRequest
+import com.google.android.gms.tasks.Task
+import com.google.android.gms.tasks.Tasks
+import com.google.common.annotations.VisibleForTesting
+import timber.log.Timber
+
+/**
+ * [EngageServiceWorker] is a [Worker] class that is tasked with publishing cluster
+ * requests to Engage Service
+ */
+class EngageServiceWorker(
+    context: Context,
+    workerParams: WorkerParameters,
+) : Worker(context, workerParams) {
+
+    @VisibleForTesting
+    constructor(
+        context: Context,
+        workerParams: WorkerParameters,
+        client: AppEngagePublishClient,
+    ) : this(context, workerParams) {
+        this.client = client
+    }
+
+    val TAG = "ENGAGE_SERVICE_WORKER"
+    private var client = AppEngagePublishClient(context)
+    private val clusterRequestFactory = ClusterRequestFactory(context)
+
+    /**
+     * [doWork] is the entry point for the [EngageServiceWorker], and differentiates between
+     * publishing tasks of each cluster
+     */
+    override fun doWork(): Result {
+        // If too many publishing attempts have failed, do not attempt to publish again.
+        if (runAttemptCount > Constants.MAX_PUBLISHING_ATTEMPTS) {
+            return Result.failure()
+        }
+
+        Timber.i("Checking for Engage Service availability")
+
+        // Check if engage service is available before publishing.
+        val isAvailable = Tasks.await(client.isServiceAvailable)
+
+        // If the service is not available, do not attempt to publish and indicate failure.
+        if (!isAvailable) {
+            Timber.e("Engage service is not available")
+            return Result.failure()
+        }
+
+        Timber.i("Engage Service is available. Proceeding to publish clusters")
+
+        // The cluster to publish must be passed into the worker through the input data.
+        // This value must be one of the predefined values indicated a valid cluster to publish. Instead
+        // of using one worker with flags to determine what cluster to publish, you may also choose to
+        // your separate workers to publish different clusters; use whichever approach better fits your
+        // app architecture.
+        return when (inputData.getString(PUBLISH_TYPE)) {
+            PUBLISH_TYPE_CONTINUATION -> publishContinuation()
+            else -> throw IllegalArgumentException("Bad publish type")
+        }
+    }
+
+    /**
+     * [publishContinuation] publishes continuation clusters and returns the result of the attempt to
+     * publish the continuation clusters if the user is signed in. If the user is signed out it
+     * instead publishes a request to delete the continuation cluster that had previously been
+     * published.
+     *
+     * @return result Result of publishing a continuation cluster, or continuation cluster deletion
+     */
+    private fun publishContinuation(): Result {
+        val publishTask: Task<Void> = client.publishContinuationCluster(
+            clusterRequestFactory.constructContinuationClusterRequest()
+        )
+        val statusCode: Int = AppEngagePublishStatusCode.PUBLISHED
+        return publishAndProvideResult(publishTask, statusCode)
+    }
+
+    /**
+     * [publishAndProvideResult] is a method that is in charge of publishing a given task
+     *
+     * @param publishTask A task to publish some cluster or delete some cluster
+     * @param publishStatusCode Publish status code to set through Engage.
+     * @return publishResult Result of [publishTask]
+     */
+    private fun publishAndProvideResult(
+        publishTask: Task<Void>,
+        publishStatusCode: Int
+    ): Result {
+        setPublishStatusCode(publishStatusCode)
+
+        // Result initialized to success, it is changed to retry or failure if an exception occurs.
+        var result: Result = Result.success()
+        try {
+            // An AppEngageException may occur while publishing, so we may not be able to await the
+            // result.
+            Tasks.await(publishTask)
+        } catch (publishException: AppEngageException) {
+            Publisher.logPublishing(publishException)
+            // Some errors are recoverable, such as a threading issue, some are unrecoverable
+            // such as a cluster not containing all necessary fields. If an error is recoverable, we
+            // should attempt to publish again. Setting the  result to retry means WorkManager will
+            // attempt to run the worker again, thus attempting to publish again.
+            result =
+                if (Publisher.isErrorRecoverable(publishException))
+                    Result.retry()
+                else
+                    Result.failure()
+        } catch (exception: Exception) {
+            exception.printStackTrace()
+            result = Result.failure()
+        }
+        // This result is returned back to doWork.
+        return result
+    }
+
+    /**
+     * [setPublishStatusCode] method is in charge of updating the publish status code, which monitors
+     * the health of the integration with EngageSDK
+     *
+     * @param statusCode PublishStatus code to be set through Engage.
+     */
+    private fun setPublishStatusCode(statusCode: Int) {
+        client
+            .updatePublishStatus(PublishStatusRequest.Builder().setStatusCode(statusCode).build())
+            .addOnSuccessListener {
+                Log.i(TAG, "Successfully updated publish status code to $statusCode")
+            }
+            .addOnFailureListener { exception ->
+                Log.e(
+                    TAG,
+                    "Failed to update publish status code to $statusCode\n${exception.stackTrace}"
+                )
+            }
+    }
+}