[WIP] Remove PushRegistrationWorkerManager and refactor PushRegistrationManager

Author: rfc2822

app/src/androidTest/kotlin/at/bitfire/davdroid/TestModules.kt

@@ -4,27 +4,13 @@
 
 package at.bitfire.davdroid
 
-import at.bitfire.davdroid.push.PushRegistrationWorkerManager
-import at.bitfire.davdroid.repository.DavCollectionRepository
 import at.bitfire.davdroid.startup.StartupPlugin
 import at.bitfire.davdroid.startup.TasksAppWatcher
 import dagger.Module
 import dagger.hilt.components.SingletonComponent
 import dagger.hilt.testing.TestInstallIn
 import dagger.multibindings.Multibinds
 
-// remove PushRegistrationWorkerModule from Android tests
-@Module
-@TestInstallIn(
-    components = [SingletonComponent::class],
-    replaces = [PushRegistrationWorkerManager.PushRegistrationWorkerModule::class]
-)
-abstract class TestPushRegistrationWorkerModule {
-    // provides empty set of listeners
-    @Multibinds
-    abstract fun empty(): Set<DavCollectionRepository.OnChangeListener>
-}
-
 // remove TasksAppWatcherModule from Android tests
 @Module
 @TestInstallIn(

app/src/main/kotlin/at/bitfire/davdroid/db/CollectionDao.kt

@@ -24,7 +24,7 @@ interface CollectionDao {
     fun getFlow(id: Long): Flow<Collection?>
 
     @Query("SELECT * FROM collection WHERE serviceId=:serviceId")
-    fun getByService(serviceId: Long): List<Collection>
+    suspend fun getByService(serviceId: Long): List<Collection>
 
     @Query("SELECT * FROM collection WHERE serviceId=:serviceId AND homeSetId IS :homeSetId")
     fun getByServiceAndHomeset(serviceId: Long, homeSetId: Long?): List<Collection>
@@ -36,7 +36,7 @@ interface CollectionDao {
     fun getSyncableByPushTopic(topic: String): Collection?
 
     @Query("SELECT pushVapidKey FROM collection WHERE serviceId=:serviceId AND pushVapidKey IS NOT NULL LIMIT 1")
-    fun getFirstVapidKey(serviceId: Long): String?
+    suspend fun getFirstVapidKey(serviceId: Long): String?
 
     @Query("SELECT COUNT(*) FROM collection WHERE serviceId=:serviceId AND type=:type")
     suspend fun anyOfType(serviceId: Long, @CollectionType type: String): Boolean
@@ -78,8 +78,11 @@ interface CollectionDao {
     @Query("SELECT * FROM collection WHERE serviceId=:serviceId AND sync AND supportsWebPush AND pushTopic IS NOT NULL")
     fun getPushCapableSyncCollections(serviceId: Long): List<Collection>
 
-    @Query("SELECT * FROM collection WHERE pushSubscription IS NOT NULL AND NOT sync")
-    suspend fun getPushRegisteredAndNotSyncable(): List<Collection>
+    @Query("SELECT * FROM collection WHERE serviceId=:serviceId AND pushSubscription IS NOT NULL")
+    suspend fun getPushRegistered(serviceId: Long): List<Collection>
+
+    @Query("SELECT * FROM collection WHERE serviceId=:serviceId AND pushSubscription IS NOT NULL AND NOT sync")
+    suspend fun getPushRegisteredAndNotSyncable(serviceId: Long): List<Collection>
 
     @Insert(onConflict = OnConflictStrategy.IGNORE)
     fun insert(collection: Collection): Long

app/src/main/kotlin/at/bitfire/davdroid/push/PushRegistrationManager.kt

@@ -5,50 +5,82 @@
 package at.bitfire.davdroid.push
 
 import android.content.Context
+import androidx.work.BackoffPolicy
+import androidx.work.Constraints
+import androidx.work.ExistingPeriodicWorkPolicy
+import androidx.work.NetworkType
+import androidx.work.PeriodicWorkRequest
+import androidx.work.WorkManager
 import at.bitfire.dav4jvm.DavCollection
 import at.bitfire.dav4jvm.DavResource
 import at.bitfire.dav4jvm.HttpUtils
 import at.bitfire.dav4jvm.XmlUtils
 import at.bitfire.dav4jvm.XmlUtils.insertTag
+import at.bitfire.dav4jvm.exception.DavException
 import at.bitfire.dav4jvm.property.push.AuthSecret
 import at.bitfire.dav4jvm.property.push.PushRegister
 import at.bitfire.dav4jvm.property.push.PushResource
 import at.bitfire.dav4jvm.property.push.Subscription
 import at.bitfire.dav4jvm.property.push.SubscriptionPublicKey
 import at.bitfire.dav4jvm.property.push.WebPushSubscription
 import at.bitfire.davdroid.db.Collection
+import at.bitfire.davdroid.db.Service
 import at.bitfire.davdroid.network.HttpClient
 import at.bitfire.davdroid.repository.AccountRepository
 import at.bitfire.davdroid.repository.DavCollectionRepository
 import at.bitfire.davdroid.repository.DavServiceRepository
+import dagger.Lazy
 import dagger.hilt.android.qualifiers.ApplicationContext
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.runInterruptible
+import kotlinx.coroutines.withContext
+import okhttp3.HttpUrl
+import okhttp3.HttpUrl.Companion.toHttpUrlOrNull
 import okhttp3.RequestBody.Companion.toRequestBody
 import org.unifiedpush.android.connector.UnifiedPush
 import org.unifiedpush.android.connector.data.PushEndpoint
 import java.io.StringWriter
 import java.time.Duration
 import java.time.Instant
+import java.util.concurrent.TimeUnit
 import java.util.logging.Level
 import java.util.logging.Logger
 import javax.inject.Inject
 import javax.inject.Provider
 
 class PushRegistrationManager @Inject constructor(
-    private val accountRepository: AccountRepository,
+    private val accountRepository: Lazy<AccountRepository>,
     private val collectionRepository: DavCollectionRepository,
     @ApplicationContext private val context: Context,
     private val httpClientBuilder: Provider<HttpClient.Builder>,
     private val logger: Logger,
     private val serviceRepository: DavServiceRepository
 ) {
 
-    fun update() {
+    /**
+     * Updates all push registrations and subscriptions so that if Push is available, it's up-to-date and
+     * working for all database services.
+     *
+     * Also makes sure that the [PushRegistrationWorker] is enabled if there's a Push-enabled collection.
+     */
+    suspend fun update() = withContext(dispatcher) {
         for (service in serviceRepository.getAll())
-            update(service.id)
+            updateService(service.id)
+
+        updatePeriodicWorker()
     }
 
-    fun update(serviceId: Long) {
-        val service = serviceRepository.get(serviceId) ?: return
+    /**
+     * Same as [update], but for a specific database service.
+     */
+    suspend fun update(serviceId: Long) {
+        updateService(serviceId)
+        updatePeriodicWorker()
+    }
+
+    private suspend fun updateService(serviceId: Long) = withContext(dispatcher) {
+        val service = serviceRepository.get(serviceId) ?: return@withContext
         val vapid = collectionRepository.getVapidKey(serviceId)
 
         if (vapid != null)
@@ -59,39 +91,103 @@ class PushRegistrationManager @Inject constructor(
             }
         else
             UnifiedPush.unregister(context, serviceId.toString())
+
+        // UnifiedPush has now been called. It will do its work and then call back to UnifiedPushService, which
+        // will then call processSubscription or removeSubscription.
     }
 
-    fun registerSubscription(serviceId: Long, endpoint: PushEndpoint) {
-        val service = serviceRepository.get(serviceId) ?: return
 
-        val collectionsToRegister = collectionRepository.getPushCapableAndSyncable(serviceId)
-        if (collectionsToRegister.isEmpty())
+    /**
+     * Called when a subscription (endpoint) is available for the given service.
+     *
+     * Uses the subscription to subscribe to syncable collections, and then unsubscribes from non-syncable collections.
+     */
+    internal suspend fun processSubscription(serviceId: Long, endpoint: PushEndpoint) = withContext(dispatcher) {
+        val service = serviceRepository.get(serviceId) ?: return@withContext
+
+        subscribeSyncable(service, endpoint)
+        unsubscribeNotSyncable(service)
+    }
+
+    private suspend fun subscribeSyncable(service: Service, endpoint: PushEndpoint) {
+        val subscribeTo = collectionRepository.getPushCapableAndSyncable(service.id)
+        if (subscribeTo.isEmpty())
             return
 
-        val account = accountRepository.fromName(service.accountName)
+        val account = accountRepository.get().fromName(service.accountName)
         httpClientBuilder.get()
             .fromAccount(account)
             .build()
             .use { httpClient ->
-                for (collection in collectionsToRegister)
+                for (collection in subscribeTo)
                     try {
                         val expires = collection.pushSubscriptionExpires
                         // calculate next run time, but use the duplicate interval for safety (times are not exact)
-                        val nextRun = Instant.now() + Duration.ofDays(2 * PushRegistrationWorkerManager.INTERVAL_DAYS)
+                        val nextRun = Instant.now() + Duration.ofDays(2 * WORKER_INTERVAL_DAYS)
                         if (expires != null && expires >= nextRun.epochSecond)
                             logger.fine("Push subscription for ${collection.url} is still valid until ${collection.pushSubscriptionExpires}")
                         else {
                             // no existing subscription or expiring soon
                             logger.fine("Registering push subscription for ${collection.url}")
-                            registerSubscription(httpClient, collection, endpoint)
+                            subscribe(httpClient, collection, endpoint)
                         }
                     } catch (e: Exception) {
                         logger.log(Level.WARNING, "Couldn't register subscription at CalDAV/CardDAV server", e)
                     }
             }
     }
 
-    private fun registerSubscription(httpClient: HttpClient, collection: Collection, endpoint: PushEndpoint) {
+    private suspend fun unsubscribeNotSyncable(service: Service) {
+        val unsubscribeFrom = collectionRepository.getPushRegisteredAndNotSyncable(service.id)
+        if (unsubscribeFrom.isEmpty())
+            return
+
+        val account = accountRepository.get().fromName(service.accountName)
+        httpClientBuilder.get()
+            .fromAccount(account)
+            .build()
+            .use { httpClient ->
+                for (collection in unsubscribeFrom)
+                    collection.pushSubscription?.toHttpUrlOrNull()?.let { url ->
+                        logger.info("Unregistering push for ${collection.url}")
+                        unsubscribe(httpClient, collection, url)
+                    }
+            }
+    }
+
+    /**
+     * Called when no subscription is available (anymore) for the given service.
+     *
+     * Unsubscribes from all collections.
+     */
+    internal suspend fun removeSubscription(serviceId: Long) {
+        val service = serviceRepository.get(serviceId) ?: return
+        val unsubscribeFrom = collectionRepository.getPushRegistered(service.id)
+        if (unsubscribeFrom.isEmpty())
+            return
+
+        val account = accountRepository.get().fromName(service.accountName)
+        httpClientBuilder.get()
+            .fromAccount(account)
+            .build()
+            .use { httpClient ->
+                for (collection in unsubscribeFrom)
+                    collection.pushSubscription?.toHttpUrlOrNull()?.let { url ->
+                        logger.info("Unregistering push for ${collection.url}")
+                        unsubscribe(httpClient, collection, url)
+                    }
+            }
+    }
+
+
+    /**
+     * Registers the subscription to a given collection ("subscribe to a collection").
+     *
+     * @param httpClient    HTTP client to use
+     * @param collection    collection to subscribe to
+     * @param endpoint      subscription to register
+     */
+    private suspend fun subscribe(httpClient: HttpClient, collection: Collection, endpoint: PushEndpoint) {
         // requested expiration time: 3 days
         val requestedExpiration = Instant.now() + Duration.ofDays(3)
 
@@ -124,26 +220,89 @@ class PushRegistrationManager @Inject constructor(
         }
         serializer.endDocument()
 
-        val xml = writer.toString().toRequestBody(DavResource.MIME_XML)
-        DavCollection(httpClient.okHttpClient, collection.url).post(xml) { response ->
-            if (response.isSuccessful) {
-                // update subscription URL and expiration in DB
-                val subscriptionUrl = response.header("Location")
-                val expires = response.header("Expires")?.let { expiresDate ->
-                    HttpUtils.parseDate(expiresDate)
-                } ?: requestedExpiration
-                collectionRepository.updatePushSubscription(
-                    id = collection.id,
-                    subscriptionUrl = subscriptionUrl,
-                    expires = expires?.epochSecond
-                )
-            } else
-                logger.warning("Couldn't register push for ${collection.url}: $response")
+        runInterruptible {
+            val xml = writer.toString().toRequestBody(DavResource.MIME_XML)
+            DavCollection(httpClient.okHttpClient, collection.url).post(xml) { response ->
+                if (response.isSuccessful) {
+                    // update subscription URL and expiration in DB
+                    val subscriptionUrl = response.header("Location")
+                    val expires = response.header("Expires")?.let { expiresDate ->
+                        HttpUtils.parseDate(expiresDate)
+                    } ?: requestedExpiration
+                    collectionRepository.updatePushSubscription(
+                        id = collection.id,
+                        subscriptionUrl = subscriptionUrl,
+                        expires = expires?.epochSecond
+                    )
+                } else
+                    logger.warning("Couldn't register push for ${collection.url}: $response")
+            }
+        }
+    }
+
+    private suspend fun unsubscribe(httpClient: HttpClient, collection: Collection, url: HttpUrl) {
+        runInterruptible {
+            try {
+                DavResource(httpClient.okHttpClient, url).delete {
+                    // deleted
+                }
+            } catch (e: DavException) {
+                logger.log(Level.WARNING, "Couldn't unregister push for ${collection.url}", e)
+            }
+
+            // remove registration URL from DB in any case
+            collectionRepository.updatePushSubscription(
+                id = collection.id,
+                subscriptionUrl = null,
+                expires = null
+            )
+        }
+    }
+
+
+    /**
+     * Determines whether there are any push-capable collections and updates the periodic worker accordingly.
+     *
+     * If there are push-capable collections, a unique periodic worker with an initial delay of 5 seconds is enqueued.
+     * A potentially existing worker is replaced, so that the first run should be soon.
+     *
+     * Otherwise, a potentially existing worker is cancelled.
+     */
+    fun updatePeriodicWorker() {
+        val workerNeeded = runBlocking {
+            collectionRepository.anyPushCapable()
+        }
+
+        val workManager = WorkManager.getInstance(context)
+        if (workerNeeded) {
+            logger.info("Enqueuing periodic PushRegistrationWorker")
+            workManager.enqueueUniquePeriodicWork(
+                WORKER_UNIQUE_NAME,
+                ExistingPeriodicWorkPolicy.UPDATE,
+                PeriodicWorkRequest.Builder(PushRegistrationWorker::class, WORKER_INTERVAL_DAYS, TimeUnit.DAYS)
+                    .setInitialDelay(5, TimeUnit.SECONDS)
+                    .setConstraints(
+                        Constraints.Builder()
+                            .setRequiredNetworkType(NetworkType.CONNECTED)
+                            .build()
+                    )
+                    .setBackoffCriteria(BackoffPolicy.EXPONENTIAL, 1, TimeUnit.MINUTES)
+                    .build()
+            )
+        } else {
+            logger.info("Cancelling periodic PushRegistrationWorker")
+            workManager.cancelUniqueWork(WORKER_UNIQUE_NAME)
         }
     }
 
-    fun unregisterSubscription(serviceId: Long) {
-        // TODO
+
+    companion object {
+
+        private const val WORKER_UNIQUE_NAME = "push-registration"
+        const val WORKER_INTERVAL_DAYS = 1L
+
+        val dispatcher = Dispatchers.IO.limitedParallelism(1)
+
     }
 
 }