Skip to content

Commit ea40bb0

Browse files
AndyScherzingerAndyScherzinger
committed
refactor(chat): move push message catch-up into a dedicated worker
Fetching the pushed room's messages no longer runs synchronously inside NotificationWorker. It now enqueues a ChatMessageCatchUpWorker (network-constrained, exponential backoff) after the notification is displayed, so a slow or failing fetch can never delay the notification and transient failures are retried instead of lost. SyncOutcome gained a syncFailed flag so the worker can tell "no new messages" from a failed fetch. Assisted-by: Claude Code:claude-fable-5 Signed-off-by: Andy Scherzinger <info@andy-scherzinger.de>
1 parent 61ba4e8 commit ea40bb0

5 files changed

Lines changed: 183 additions & 43 deletions

File tree

‎app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt‎

Lines changed: 15 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -87,11 +87,17 @@ class ChatMessageSyncer @Inject constructor(
8787
}
8888
}
8989

90+
/**
91+
* [syncFailed] is true when the sync ended in a transient error (offline, failed request), so
92+
* callers like a background worker can retry later. It stays false for skips that retrying
93+
* would not change, e.g. a missing server capability.
94+
*/
9095
data class SyncOutcome(
9196
val persistedNewMessages: Boolean,
9297
val newestPersistedMessageId: Long?,
9398
val oldestPersistedMessageId: Long? = null,
94-
val persistedMessageCount: Int = 0
99+
val persistedMessageCount: Int = 0,
100+
val syncFailed: Boolean = false
95101
)
96102

97103
/**
@@ -211,7 +217,7 @@ class ChatMessageSyncer @Inject constructor(
211217
when {
212218
!networkMonitor.isOnline.value -> {
213219
Log.d(TAG, "Device is offline, skipping catch-up for ${target.internalConversationId}")
214-
NOTHING_SYNCED
220+
SYNC_FAILED
215221
}
216222

217223
!target.user.hasSpreedFeatureCapability(SpreedFeatures.CHAT_KEEP_NOTIFICATIONS.value) -> {
@@ -396,7 +402,8 @@ class ChatMessageSyncer @Inject constructor(
396402
persistedNewMessages = totalCount > 0,
397403
newestPersistedMessageId = newestPersisted,
398404
oldestPersistedMessageId = oldestPersisted,
399-
persistedMessageCount = totalCount
405+
persistedMessageCount = totalCount,
406+
syncFailed = roundOutcome.syncFailed
400407
)
401408
}
402409
anchor = nextAnchor
@@ -426,7 +433,8 @@ class ChatMessageSyncer @Inject constructor(
426433
persistedNewMessages = totalCount > 0 || fallbackOutcome.persistedNewMessages,
427434
newestPersistedMessageId = fallbackOutcome.newestPersistedMessageId,
428435
oldestPersistedMessageId = fallbackOutcome.oldestPersistedMessageId,
429-
persistedMessageCount = fallbackOutcome.persistedMessageCount
436+
persistedMessageCount = fallbackOutcome.persistedMessageCount,
437+
syncFailed = fallbackOutcome.syncFailed
430438
)
431439
}
432440

@@ -507,7 +515,7 @@ class ChatMessageSyncer @Inject constructor(
507515

508516
is ChatPullResult.Error -> {
509517
Log.e(TAG, "Error pulling messages from server", result.throwable)
510-
NOTHING_SYNCED
518+
SYNC_FAILED
511519
}
512520
}
513521
} finally {
@@ -869,6 +877,8 @@ class ChatMessageSyncer @Inject constructor(
869877
val NO_EVENTS: Events = object : Events {}
870878

871879
private val NOTHING_SYNCED = SyncOutcome(persistedNewMessages = false, newestPersistedMessageId = null)
880+
private val SYNC_FAILED =
881+
SyncOutcome(persistedNewMessages = false, newestPersistedMessageId = null, syncFailed = true)
872882

873883
private const val DEFAULT_MESSAGES_LIMIT = 100
874884
private const val MILLIS_PER_SECOND = 1000L

‎app/src/main/java/com/nextcloud/talk/jobs/ChatMessageCatchUpWorker.kt‎

Lines changed: 134 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,134 @@
1+
/*
2+
* Nextcloud Talk - Android Client
3+
*
4+
* SPDX-FileCopyrightText: 2026 Andy Scherzinger <andy.scherzinger@nextcloud.com>
5+
* SPDX-License-Identifier: GPL-3.0-or-later
6+
*/
7+
package com.nextcloud.talk.jobs
8+
9+
import android.content.Context
10+
import android.os.PowerManager
11+
import android.util.Log
12+
import androidx.work.BackoffPolicy
13+
import androidx.work.Constraints
14+
import androidx.work.CoroutineWorker
15+
import androidx.work.Data
16+
import androidx.work.NetworkType
17+
import androidx.work.OneTimeWorkRequest
18+
import androidx.work.WorkManager
19+
import androidx.work.WorkRequest
20+
import androidx.work.WorkerParameters
21+
import autodagger.AutoInjector
22+
import com.nextcloud.talk.application.NextcloudTalkApplication
23+
import com.nextcloud.talk.application.NextcloudTalkApplication.Companion.sharedApplication
24+
import com.nextcloud.talk.chat.data.network.ChatMessageSyncer
25+
import com.nextcloud.talk.users.UserManager
26+
import com.nextcloud.talk.utils.ApiUtils
27+
import com.nextcloud.talk.utils.bundle.BundleKeys.KEY_INTERNAL_USER_ID
28+
import com.nextcloud.talk.utils.bundle.BundleKeys.KEY_ROOM_TOKEN
29+
import com.nextcloud.talk.utils.bundle.BundleKeys.KEY_THREAD_ID
30+
import java.util.concurrent.TimeUnit
31+
import javax.inject.Inject
32+
33+
/**
34+
* Prefetches a pushed room's messages into the local database so they are instantly visible when
35+
* the chat is opened from the notification (or later). Enqueued by [NotificationWorker] after the
36+
* notification is displayed, so the fetch never delays or suppresses the notification and
37+
* transient failures are retried with backoff instead of being lost with the notification worker.
38+
*
39+
* Push bursts for the same room enqueue one worker each, but the per-room coalescing of the
40+
* [ChatMessageSyncer] singleton collapses overlapping catch-ups into few actual fetches. Skipped
41+
* in battery saver mode; the chat-keep-notifications capability gate and the offline check are
42+
* handled inside [ChatMessageSyncer.catchUpRoom].
43+
*/
44+
@AutoInjector(NextcloudTalkApplication::class)
45+
class ChatMessageCatchUpWorker(context: Context, workerParams: WorkerParameters) :
46+
CoroutineWorker(context, workerParams) {
47+
48+
@Inject
49+
lateinit var userManager: UserManager
50+
51+
@Inject
52+
lateinit var chatMessageSyncer: ChatMessageSyncer
53+
54+
override suspend fun doWork(): Result {
55+
sharedApplication!!.componentApplication.inject(this)
56+
57+
val userId = inputData.getLong(KEY_INTERNAL_USER_ID, -1)
58+
val roomToken = inputData.getString(KEY_ROOM_TOKEN)
59+
val threadId = inputData.getLong(KEY_THREAD_ID, NO_THREAD).takeIf { it != NO_THREAD }
60+
61+
return when {
62+
userId < 0 || roomToken.isNullOrEmpty() -> {
63+
Log.e(TAG, "Missing user id or room token, dropping message catch-up")
64+
Result.failure()
65+
}
66+
67+
isPowerSaveMode() -> {
68+
Log.d(TAG, "Battery saver is active, skipping message catch-up for room $roomToken")
69+
Result.success()
70+
}
71+
72+
else -> catchUpRoom(userId, roomToken, threadId)
73+
}
74+
}
75+
76+
private suspend fun catchUpRoom(userId: Long, roomToken: String, threadId: Long?): Result {
77+
val user = userManager.getUserWithId(userId).blockingGet()
78+
val credentials = user?.let { ApiUtils.getCredentials(it.username, it.token) }
79+
if (user == null || credentials == null) {
80+
Log.e(TAG, "No user or credentials found for user id $userId, dropping message catch-up")
81+
return Result.failure()
82+
}
83+
84+
val target = ChatMessageSyncer.SyncTarget(
85+
user = user,
86+
roomToken = roomToken,
87+
threadId = threadId,
88+
credentials = credentials,
89+
urlForChatting = ApiUtils.getUrlForChat(CHAT_API_VERSION, user.baseUrl!!, roomToken)
90+
)
91+
92+
val outcome = runCatching { chatMessageSyncer.catchUpRoom(target) }.getOrElse { throwable ->
93+
Log.e(TAG, "Message catch-up failed for room $roomToken", throwable)
94+
null
95+
}
96+
97+
return if (outcome == null || outcome.syncFailed) {
98+
Log.w(TAG, "Message catch-up for room $roomToken did not complete (attempt ${runAttemptCount + 1})")
99+
retryOrFail()
100+
} else {
101+
Result.success()
102+
}
103+
}
104+
105+
private fun retryOrFail(): Result = if (runAttemptCount < MAX_RUN_ATTEMPTS - 1) Result.retry() else Result.failure()
106+
107+
private fun isPowerSaveMode(): Boolean {
108+
val powerManager = applicationContext.getSystemService(Context.POWER_SERVICE) as PowerManager
109+
return powerManager.isPowerSaveMode
110+
}
111+
112+
companion object {
113+
private val TAG: String = ChatMessageCatchUpWorker::class.java.simpleName
114+
private const val CHAT_API_VERSION = 1
115+
private const val NO_THREAD = -1L
116+
private const val MAX_RUN_ATTEMPTS = 3
117+
118+
fun enqueue(context: Context, userId: Long, roomToken: String, threadId: Long?) {
119+
val data = Data.Builder()
120+
.putLong(KEY_INTERNAL_USER_ID, userId)
121+
.putString(KEY_ROOM_TOKEN, roomToken)
122+
.apply { threadId?.let { putLong(KEY_THREAD_ID, it) } }
123+
.build()
124+
125+
val catchUpWork = OneTimeWorkRequest.Builder(ChatMessageCatchUpWorker::class.java)
126+
.setInputData(data)
127+
.setConstraints(Constraints.Builder().setRequiredNetworkType(NetworkType.CONNECTED).build())
128+
.setBackoffCriteria(BackoffPolicy.EXPONENTIAL, WorkRequest.MIN_BACKOFF_MILLIS, TimeUnit.MILLISECONDS)
129+
.build()
130+
131+
WorkManager.getInstance(context).enqueue(catchUpWork)
132+
}
133+
}
134+
}

‎app/src/main/java/com/nextcloud/talk/jobs/NotificationWorker.kt‎

Lines changed: 8 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,6 @@ import android.os.Build
2020
import android.os.Bundle
2121
import android.os.Handler
2222
import android.os.Looper
23-
import android.os.PowerManager
2423
import android.os.SystemClock
2524
import android.service.notification.StatusBarNotification
2625
import android.text.TextUtils
@@ -53,7 +52,6 @@ import com.nextcloud.talk.application.NextcloudTalkApplication
5352
import com.nextcloud.talk.application.NextcloudTalkApplication.Companion.sharedApplication
5453
import com.nextcloud.talk.arbitrarystorage.ArbitraryStorageManager
5554
import com.nextcloud.talk.callnotification.CallNotificationActivity
56-
import com.nextcloud.talk.chat.data.network.ChatMessageSyncer
5755
import com.nextcloud.talk.chat.data.network.ChatNetworkDataSource
5856
import com.nextcloud.talk.conversationlist.DirectShareHelper
5957
import com.nextcloud.talk.data.user.model.User
@@ -138,9 +136,6 @@ class NotificationWorker(context: Context, workerParams: WorkerParameters) : Wor
138136
var chatNetworkDataSource: ChatNetworkDataSource? = null
139137
@Inject set
140138

141-
var chatMessageSyncer: ChatMessageSyncer? = null
142-
@Inject set
143-
144139
@Inject
145140
lateinit var userManager: UserManager
146141

@@ -219,42 +214,19 @@ class NotificationWorker(context: Context, workerParams: WorkerParameters) : Wor
219214
}
220215

221216
/**
222-
* Prefetches the pushed room's messages into the local database so they are instantly visible
223-
* when the chat is opened from the notification (or later). For messages in a thread,
224-
* [threadId] targets the thread so its chat block is extended. Best effort only: failures are
225-
* logged and never delay or suppress the notification, which is displayed beforehand.
226-
* Skipped in battery saver mode; the chat-keep-notifications capability gate and the offline
227-
* check are handled inside [ChatMessageSyncer.catchUpRoom].
217+
* Enqueues a [ChatMessageCatchUpWorker] that prefetches the pushed room's messages into the
218+
* local database so they are instantly visible when the chat is opened from the notification
219+
* (or later). For messages in a thread, [threadId] targets the thread so its chat block is
220+
* extended. The worker runs independently of this one: the notification is displayed
221+
* beforehand and a slow or failing fetch (retried there with backoff) can never delay it.
228222
*/
229223
private fun catchUpPushedRoom(threadId: Long?) {
230224
val roomToken = pushMessage.id
231-
val syncer = chatMessageSyncer
232-
if (pushMessage.type != TYPE_CHAT || roomToken == null || syncer == null || isPowerSaveMode()) {
233-
logger.d(TAG, "Skipping message catch-up for pushed room (not a chat push or battery saver active)")
225+
if (pushMessage.type != TYPE_CHAT || roomToken == null) {
226+
logger.d(TAG, "Skipping message catch-up for pushed room (not a chat push)")
234227
return
235228
}
236-
237-
// the user from the push signature verification may carry stale capabilities, so resolve
238-
// the current state before the capability check in catchUpRoom
239-
val currentUser = userManager.getUserWithId(user.id!!).blockingGet() ?: return
240-
241-
val target = ChatMessageSyncer.SyncTarget(
242-
user = currentUser,
243-
roomToken = roomToken,
244-
threadId = threadId,
245-
credentials = credentials,
246-
urlForChatting = ApiUtils.getUrlForChat(CHAT_API_VERSION, currentUser.baseUrl!!, roomToken)
247-
)
248-
runCatching {
249-
runBlocking { syncer.catchUpRoom(target) }
250-
}.onFailure {
251-
Log.e(TAG, "Message catch-up after push failed for room $roomToken", it)
252-
}
253-
}
254-
255-
private fun isPowerSaveMode(): Boolean {
256-
val powerManager = applicationContext.getSystemService(Context.POWER_SERVICE) as PowerManager
257-
return powerManager.isPowerSaveMode
229+
ChatMessageCatchUpWorker.enqueue(applicationContext, user.id!!, roomToken, threadId)
258230
}
259231

260232
private fun handleRemoteTalkSharePushMessage() {
@@ -1250,7 +1222,6 @@ class NotificationWorker(context: Context, workerParams: WorkerParameters) : Wor
12501222
companion object {
12511223
val TAG: String = NotificationWorker::class.java.simpleName
12521224
private const val TYPE_CHAT = "chat"
1253-
private const val CHAT_API_VERSION = 1
12541225
private const val TYPE_ROOM = "room"
12551226
private const val TYPE_CALL = "call"
12561227
private const val TYPE_RECORDING = "recording"

‎app/src/test/java/android/util/Log.kt‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,13 @@ object Log {
3030
return 1
3131
}
3232

33+
@JvmStatic
34+
fun e(tag: String, msg: String, tr: Throwable): Int {
35+
println("ERROR: $tag: $msg: $tr")
36+
37+
return 1
38+
}
39+
3340
@JvmStatic
3441
fun i(tag: String, msg: String): Int {
3542
println("INFO: $tag: $msg")

‎app/src/test/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncerTest.kt‎

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -95,13 +95,14 @@ class ChatMessageSyncerTest {
9595
}
9696

9797
@Test
98-
fun `catchUpRoom skips when offline`() =
98+
fun `catchUpRoom skips when offline and marks the sync as failed`() =
9999
runTest {
100100
whenever(networkMonitor.isOnline).thenReturn(MutableStateFlow(false))
101101

102102
val outcome = syncer.catchUpRoom(target())
103103

104104
assertFalse(outcome.persistedNewMessages)
105+
assertTrue(outcome.syncFailed)
105106
verifyNoInteractions(network)
106107
}
107108

@@ -111,9 +112,25 @@ class ChatMessageSyncerTest {
111112
val outcome = syncer.catchUpRoom(target(user(withKeepNotificationsCapability = false)))
112113

113114
assertFalse(outcome.persistedNewMessages)
115+
// a missing capability is not retryable, so the skip must not count as failure
116+
assertFalse(outcome.syncFailed)
114117
verifyNoInteractions(network)
115118
}
116119

120+
@Test
121+
fun `catchUpRoom marks the sync as failed when the server request errors`() =
122+
runTest {
123+
whenever(chatBlocksDao.getNewestMessageIdFromChatBlocks(INTERNAL_CONVERSATION_ID, null))
124+
.thenReturn(42L)
125+
wheneverBlocking { network.pullChatMessages(any(), any(), any()) }
126+
.thenReturn(Response.error(HTTP_INTERNAL_SERVER_ERROR, "".toResponseBody()))
127+
128+
val outcome = syncer.catchUpRoom(target())
129+
130+
assertFalse(outcome.persistedNewMessages)
131+
assertTrue(outcome.syncFailed)
132+
}
133+
117134
@Test
118135
fun `catchUpRoom delta-fetches from the newest cached message when a chat block exists`() =
119136
runTest {
@@ -499,5 +516,6 @@ class ChatMessageSyncerTest {
499516
private const val CREDENTIALS = "credentials"
500517
private const val CHAT_URL = "https://server.example.com/ocs/v2.php/apps/spreed/api/v1/chat/$ROOM_TOKEN"
501518
private const val HTTP_NOT_MODIFIED = 304
519+
private const val HTTP_INTERNAL_SERVER_ERROR = 500
502520
}
503521
}

0 commit comments

Comments
 (0)