Add global search backfill worker (#7860)

* Add search backfill worker

This will run on background at most once an hour while the device has internet connectivity and the device is charging and start back-paginating through the user's rooms to make sure the search results are populated.

It'll run for the maximum time the default workers allow, 10 minutes.

* Create `RustMatrixClient.Factory` AssistedFactory to help with DI and add `SearchBackfillService`

* Start search backfill when entering the search screen, just once. This should help getting initial search results if none or few are found at first
This commit is contained in:
Jorge Martin Espinosa
2026-10-07 12:41:19 +02:00
committed by GitHub
parent 9805422a8a
commit 3fe0e09e3c
27 changed files with 1104 additions and 73 deletions
@@ -10,12 +10,14 @@ package io.element.android.features.home.impl.search
import androidx.compose.foundation.text.input.clearText
import androidx.compose.foundation.text.input.rememberTextFieldState
import androidx.compose.runtime.Composable
import androidx.compose.runtime.DisposableEffect
import androidx.compose.runtime.LaunchedEffect
import androidx.compose.runtime.getValue
import androidx.compose.runtime.mutableStateOf
import androidx.compose.runtime.produceState
import androidx.compose.runtime.remember
import androidx.compose.runtime.rememberCoroutineScope
import androidx.compose.runtime.saveable.rememberSaveable
import androidx.compose.runtime.setValue
import dev.zacsweers.metro.Inject
import io.element.android.features.home.impl.search.history.SearchHistoryResult
@@ -38,6 +40,7 @@ import io.element.android.libraries.matrix.api.search.MessageSearch
import io.element.android.libraries.matrix.api.search.MessageSearchPaginationState
import io.element.android.libraries.matrix.api.search.MessageSearchResult
import io.element.android.libraries.matrix.api.search.MessageSearchService
import io.element.android.libraries.matrix.api.search.SearchBackfillStrategy
import io.element.android.libraries.matrix.api.timeline.item.event.AudioMessageType
import io.element.android.libraries.matrix.api.timeline.item.event.FileMessageType
import io.element.android.libraries.matrix.api.timeline.item.event.ImageMessageType
@@ -81,6 +84,7 @@ class GlobalSearchPresenter(
val isEnabled by produceState(false) {
featureFlagService.isFeatureEnabledFlow(FeatureFlags.MessageSearch).collectLatest { value = it }
}
val roomListSearchDataSource = remember { roomListSearchDataSourceFactory.create(coroutineScope = coroutineScope) }
val queryState = rememberTextFieldState()
var isSearchActive by remember { mutableStateOf(false) }
@@ -88,6 +92,8 @@ class GlobalSearchPresenter(
var currentTarget: GlobalSearchTarget by remember { mutableStateOf(GlobalSearchTarget.ROOMS) }
val currentMessageSearch: MessageSearch = remember { messageSearchService.createMessageSearch(scope = coroutineScope) }
StartSearchBackfillWhenVisible(isEnabled = isEnabled, isSearchActive = isSearchActive)
val searchHistory by produceState<AsyncData<ImmutableList<SearchHistoryListItem>>>(initialValue = AsyncData.Uninitialized) {
searchHistoryStore.history.collectLatest { history ->
val mappedHistoryItems = history.mapNotNull { result ->
@@ -297,6 +303,35 @@ class GlobalSearchPresenter(
)
}
@Composable
private fun StartSearchBackfillWhenVisible(
isEnabled: Boolean,
isSearchActive: Boolean,
) {
val coroutineScope = rememberCoroutineScope()
var hasBackfillRun by rememberSaveable { mutableStateOf(false) }
if (isEnabled && isSearchActive && !hasBackfillRun) {
DisposableEffect(Unit) {
// When we open the message search, we want to backfill the search results so that we can show the user more results if they scroll down.
val handle = matrixClient.searchBackfillService.startSearchBackfill(SearchBackfillStrategy.FOREGROUND).getOrNull()
?: return@DisposableEffect onDispose {
Timber.w("Could not start search backfill, it may already be running.")
}
val job = coroutineScope.launch(coroutineDispatchers.io) {
while (handle.isRunning()) {
delay(100.milliseconds)
}
hasBackfillRun = true
}
onDispose {
job.cancel()
handle.close()
}
}
}
}
private fun mapMediaContent(
result: MessageSearchResult,
roomInfo: RoomInfo,
@@ -33,13 +33,16 @@ import io.element.android.libraries.matrix.api.roomlist.RoomListService
import io.element.android.libraries.matrix.api.search.MessageSearchService
import io.element.android.libraries.matrix.test.A_ROOM_ID
import io.element.android.libraries.matrix.test.FakeMatrixClient
import io.element.android.libraries.matrix.test.FakeSdkPendingTask
import io.element.android.libraries.matrix.test.permalink.FakePermalinkParser
import io.element.android.libraries.matrix.test.room.aRoomSummary
import io.element.android.libraries.matrix.test.roomlist.FakeDynamicRoomList
import io.element.android.libraries.matrix.test.roomlist.FakeRoomListService
import io.element.android.libraries.matrix.test.search.FakeMessageSearch
import io.element.android.libraries.matrix.test.search.FakeMessageSearchService
import io.element.android.libraries.matrix.test.search.FakeSearchBackfillService
import io.element.android.tests.testutils.consumeItemsUntilPredicate
import io.element.android.tests.testutils.lambda.lambdaRecorder
import io.element.android.tests.testutils.test
import io.element.android.tests.testutils.testCoroutineDispatchers
import kotlinx.collections.immutable.persistentListOf
@@ -205,7 +208,9 @@ class GlobalSearchPresenterTest {
@Test
fun `present - message search results are mapped when the message search emits`() = runTest {
val messageSearch = FakeMessageSearch()
val matrixClient = FakeMatrixClient().apply {
val matrixClient = FakeMatrixClient(
searchBackfillService = FakeSearchBackfillService(startSearchBackfillLambda = { Result.success(FakeSdkPendingTask()) }),
).apply {
getRoomInfoLambda = { Result.success(aRoomInfo()) }
}
val presenter = createGlobalSearchPresenter(
@@ -232,7 +237,9 @@ class GlobalSearchPresenterTest {
@Test
fun `present - message search displays the uninitialized state after removing the query`() = runTest {
val messageSearch = FakeMessageSearch()
val matrixClient = FakeMatrixClient().apply {
val matrixClient = FakeMatrixClient(
searchBackfillService = FakeSearchBackfillService(startSearchBackfillLambda = { Result.success(FakeSdkPendingTask()) }),
).apply {
getRoomInfoLambda = { Result.success(aRoomInfo()) }
}
val presenter = createGlobalSearchPresenter(
@@ -267,7 +274,9 @@ class GlobalSearchPresenterTest {
@Test
fun `present - UpdateVisibleRange triggers pagination for messages when near the end`() = runTest {
val messageSearch = FakeMessageSearch()
val matrixClient = FakeMatrixClient().apply {
val matrixClient = FakeMatrixClient(
searchBackfillService = FakeSearchBackfillService(startSearchBackfillLambda = { Result.success(FakeSdkPendingTask()) }),
).apply {
getRoomInfoLambda = { Result.success(aRoomInfo()) }
}
val presenter = createGlobalSearchPresenter(
@@ -432,6 +441,44 @@ class GlobalSearchPresenterTest {
assertThat(awaitItem().results.isLoading()).isTrue()
}
}
@Test
fun `present - will automatically start search backfill and it will be cancelled when the presenter is closed`() = runTest {
var isRunning = false
val closeLambda = lambdaRecorder<Unit> { isRunning = false }
val matrixClient = FakeMatrixClient(
searchBackfillService = FakeSearchBackfillService(
startSearchBackfillLambda = {
isRunning = true
Result.success(FakeSdkPendingTask(isRunningLambda = { isRunning }, closeLambda = closeLambda))
},
),
)
val presenter = createGlobalSearchPresenter(
matrixClient = matrixClient,
featureFlagService = FakeFeatureFlagService(
initialState = mapOf(FeatureFlags.MessageSearch.key to true),
),
)
presenter.test {
// Await the initial state, with the feature enabled
val initialState = consumeItemsUntilPredicate { it.isEnabled }.last()
assertThat(isRunning).isFalse()
// Opening the search starts the backfill
initialState.eventSink(GlobalSearchEvent.ToggleSearchVisibility)
consumeItemsUntilPredicate { it.isSearchActive }
// The backfill polling loop never ends while the task is running, so don't wait until the scheduler is idle
testScheduler.runCurrent()
assertThat(isRunning).isTrue()
closeLambda.assertions().isNeverCalled()
// Closing the presenter closes the task
cancelAndIgnoreRemainingEvents()
closeLambda.assertions().isCalledOnce()
assertThat(isRunning).isFalse()
}
}
}
private fun TestScope.createGlobalSearchPresenter(
@@ -442,7 +489,8 @@ private fun TestScope.createGlobalSearchPresenter(
dateFormatter: DateFormatter = FakeDateFormatter(),
fileSizeFormatter: FileSizeFormatter = FakeFileSizeFormatter(),
permalinkParser: PermalinkParser = FakePermalinkParser(),
matrixClient: MatrixClient = FakeMatrixClient(),
matrixClient: MatrixClient =
FakeMatrixClient(searchBackfillService = FakeSearchBackfillService(startSearchBackfillLambda = { Result.success(FakeSdkPendingTask()) })),
searchHistoryStore: SearchHistoryStore = FakeSearchHistoryStore(),
): GlobalSearchPresenter {
return GlobalSearchPresenter(
@@ -40,6 +40,7 @@ import io.element.android.libraries.matrix.api.roomdirectory.RoomDirectoryServic
import io.element.android.libraries.matrix.api.roomlist.RoomListService
import io.element.android.libraries.matrix.api.scanner.ContentScanner
import io.element.android.libraries.matrix.api.search.MessageSearchService
import io.element.android.libraries.matrix.api.search.SearchBackfillService
import io.element.android.libraries.matrix.api.spaces.SpaceService
import io.element.android.libraries.matrix.api.sync.SlidingSyncVersion
import io.element.android.libraries.matrix.api.sync.SyncService
@@ -133,6 +134,9 @@ interface MatrixClient : ClientUrlContentFetcher {
/** Scans media content before it is displayed, or `null` when no content scanner is configured for this session. */
val contentScanner: ContentScanner?
/** Backfills messages from the homeserver for searching for messages in a room. */
val searchBackfillService: SearchBackfillService
/** Whether the session is in the process of shutting down, either because it is being logged out or its cache is being cleared. */
val isShuttingDown: Boolean
@@ -0,0 +1,16 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.api
/**
* Represents a pending task in the SDK that can be canceled and closed.
*/
interface SdkPendingTask : AutoCloseable {
/** Checks if the task is still running. */
fun isRunning(): Boolean
}
@@ -0,0 +1,50 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.api.search
import io.element.android.libraries.matrix.api.SdkPendingTask
/**
* Service interface for managing search backfill tasks in the Matrix SDK.
* This service allows starting, checking the status of, and canceling search backfill tasks.
*/
interface SearchBackfillService {
/**
* Enqueues a periodic search backfill task based on the provided strategy.
* The strategy parameter determines how the backfill should be performed (e.g., in the foreground or background).
*/
suspend fun schedulePeriodicSearchBackfill(strategy: SearchBackfillStrategy): Result<Unit>
/**
* Cancels any currently scheduled periodic search backfill tasks.
*/
fun cancelPeriodicSearchBackfill()
/**
* Starts a background task to backfill messages for rooms that have been joined but not fully backfilled yet.
* The strategy parameter determines how the backfill should be performed (e.g., in the foreground or background).
*
* @param strategy the strategy to use for backfilling messages.
* @return a [Result] indicating the success or failure of enqueuing the backfill task.
*/
fun startSearchBackfill(strategy: SearchBackfillStrategy): Result<SdkPendingTask>
/**
* Checks if a search backfill task is currently running.
*
* @return `true` if a search backfill task is running, `false` otherwise.
*/
fun isSearchBackfillRunning(): Boolean
/**
* Cancels the currently running search backfill task, if any.
*
* @return a [Result] indicating the success or failure of the cancellation operation.
*/
fun cancelSearchBackfill(): Result<Unit>
}
@@ -0,0 +1,22 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.api.search
enum class SearchBackfillStrategy {
/**
* The app is in the foreground: pause between paginations so this doesn't
* compete with interactive traffic.
*/
FOREGROUND,
/**
* A time-boxed background task (e.g. iOS `BGAppRefreshTask`) where there's
* no interactive traffic to protect.
*/
BACKGROUND
}
+2
View File
@@ -63,6 +63,7 @@ dependencies {
implementation(projects.appconfig)
implementation(projects.features.enterprise.api)
implementation(projects.features.networkmonitor.api)
implementation(projects.libraries.androidutils)
implementation(projects.libraries.architecture)
implementation(projects.libraries.di)
@@ -80,6 +81,7 @@ dependencies {
testCommonDependencies(libs)
testImplementation(projects.features.enterprise.test)
testImplementation(projects.features.networkmonitor.test)
testImplementation(projects.libraries.featureflag.test)
testImplementation(projects.libraries.matrix.test)
testImplementation(projects.libraries.previewutils)
@@ -8,6 +8,9 @@
package io.element.android.libraries.matrix.impl
import dev.zacsweers.metro.Assisted
import dev.zacsweers.metro.AssistedFactory
import dev.zacsweers.metro.AssistedInject
import io.element.android.libraries.androidutils.file.getSizeOfFiles
import io.element.android.libraries.core.bool.orFalse
import io.element.android.libraries.core.coroutine.CoroutineDispatchers
@@ -16,7 +19,9 @@ import io.element.android.libraries.core.data.bytes
import io.element.android.libraries.core.data.tryOrNull
import io.element.android.libraries.core.extensions.mapFailure
import io.element.android.libraries.core.extensions.runCatchingExceptions
import io.element.android.libraries.di.annotations.AppCoroutineScope
import io.element.android.libraries.featureflag.api.FeatureFlagService
import io.element.android.libraries.featureflag.api.FeatureFlags
import io.element.android.libraries.matrix.api.HomeserverCapabilitiesProvider
import io.element.android.libraries.matrix.api.MatrixClient
import io.element.android.libraries.matrix.api.analytics.SdkStoreSizes
@@ -47,6 +52,7 @@ import io.element.android.libraries.matrix.api.room.join.JoinRule
import io.element.android.libraries.matrix.api.roomdirectory.RoomVisibility
import io.element.android.libraries.matrix.api.roomlist.RoomListService
import io.element.android.libraries.matrix.api.scanner.ContentScanner
import io.element.android.libraries.matrix.api.search.SearchBackfillStrategy
import io.element.android.libraries.matrix.api.spaces.SpaceService
import io.element.android.libraries.matrix.api.sync.SlidingSyncVersion
import io.element.android.libraries.matrix.api.sync.SyncState
@@ -81,6 +87,8 @@ import io.element.android.libraries.matrix.impl.roomlist.RoomListFactory
import io.element.android.libraries.matrix.impl.roomlist.RustRoomListService
import io.element.android.libraries.matrix.impl.roomlist.roomOrNull
import io.element.android.libraries.matrix.impl.search.RustMessageSearchService
import io.element.android.libraries.matrix.impl.search.RustSearchBackfillService
import io.element.android.libraries.matrix.impl.search.map
import io.element.android.libraries.matrix.impl.spaces.RustSpaceService
import io.element.android.libraries.matrix.impl.sync.RustSyncService
import io.element.android.libraries.matrix.impl.sync.map
@@ -110,8 +118,10 @@ import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.buffer
import kotlinx.coroutines.flow.distinctUntilChanged
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.map
import kotlinx.coroutines.flow.mapNotNull
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.flow.stateIn
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking
@@ -149,23 +159,37 @@ import org.matrix.rustcomponents.sdk.SyncService as ClientSyncService
private const val AVATAR_THUMBNAIL_SIZE_IN_PIXEL = 240L
@Suppress("LargeClass")
@AssistedInject
class RustMatrixClient(
override val sessionPaths: SessionPaths,
val innerClient: Client,
@Assisted override val sessionPaths: SessionPaths,
@Assisted val innerClient: Client,
@Assisted private val innerSyncService: ClientSyncService,
@Assisted baseCacheDirectory: File,
@Assisted override val contentScanner: ContentScanner?,
@Assisted override val isMessageSearchAvailable: Boolean,
@Assisted private val sessionDelegate: RustClientSessionDelegate,
private val sessionStore: SessionStore,
private val sessionDelegate: RustClientSessionDelegate,
private val innerSyncService: ClientSyncService,
appCoroutineScope: CoroutineScope,
@AppCoroutineScope appCoroutineScope: CoroutineScope,
dispatchers: CoroutineDispatchers,
baseCacheDirectory: File,
clock: SystemClock,
timelineEventFilterFactory: TimelineEventFilterFactory,
private val featureFlagService: FeatureFlagService,
private val analyticsService: AnalyticsService,
private val workManagerScheduler: WorkManagerScheduler,
override val contentScanner: ContentScanner?,
override val isMessageSearchAvailable: Boolean,
private val searchBackfillServiceFactory: RustSearchBackfillService.Factory,
) : MatrixClient {
@AssistedFactory
interface Factory {
fun create(
sessionPaths: SessionPaths,
innerClient: Client,
innerSyncService: ClientSyncService,
baseCacheDirectory: File,
contentScanner: ContentScanner?,
isMessageSearchAvailable: Boolean,
sessionDelegate: RustClientSessionDelegate,
): RustMatrixClient
}
override val sessionId: UserId = UserId(innerClient.userId())
override val deviceId: DeviceId = DeviceId(innerClient.deviceId())
override val server: String? = innerClient.server()
@@ -280,6 +304,8 @@ class RustMatrixClient(
private var clientDelegateTaskHandle: TaskHandle? = innerClient.setDelegate(sessionDelegate)
override val searchBackfillService = searchBackfillServiceFactory.create(innerClient)
private val _userProfile: MutableStateFlow<MatrixUser> = MutableStateFlow(
MatrixUser(
userId = sessionId,
@@ -332,6 +358,9 @@ class RustMatrixClient(
// Schedule regular database vacuuming to ensure DB performance remains optimal
scheduleDatabaseVacuum()
// Schedule background search backfill if needed
scheduleBackgroundSearchBackfill()
}
private suspend fun setupUserProfile() {
@@ -687,6 +716,7 @@ class RustMatrixClient(
sessionCoroutineScope.cancel()
clientDelegateTaskHandle?.cancelAndDestroy()
ownProfileTaskHandle?.cancelAndDestroy()
searchBackfillService.cancelSearchBackfill()
sessionVerificationService.destroy()
sessionDelegate.clearCurrentClient()
@@ -992,6 +1022,19 @@ class RustMatrixClient(
override fun homeserverCapabilities(): HomeserverCapabilitiesProvider {
return RustHomeserverCapabilitiesProvider(innerClient.homeserverCapabilities())
}
private fun scheduleBackgroundSearchBackfill() {
featureFlagService.isFeatureEnabledFlow(FeatureFlags.MessageSearch)
.distinctUntilChanged()
.onEach { isEnabled ->
if (isEnabled) {
searchBackfillService.schedulePeriodicSearchBackfill(SearchBackfillStrategy.BACKGROUND)
} else {
searchBackfillService.cancelPeriodicSearchBackfill()
}
}
.launchIn(sessionCoroutineScope)
}
}
private fun defaultRoomCreationPowerLevels(isPublic: Boolean, isSpace: Boolean) = PowerLevels(
@@ -23,16 +23,13 @@ import io.element.android.libraries.matrix.api.paths.SessionPaths
import io.element.android.libraries.matrix.impl.analytics.UtdTracker
import io.element.android.libraries.matrix.impl.paths.getSessionPaths
import io.element.android.libraries.matrix.impl.proxy.ProxyProvider
import io.element.android.libraries.matrix.impl.room.TimelineEventFilterFactory
import io.element.android.libraries.matrix.impl.scanner.RustContentScanner
import io.element.android.libraries.matrix.impl.storage.SqliteStoreBuilderProvider
import io.element.android.libraries.matrix.impl.util.anonymizedTokens
import io.element.android.libraries.network.useragent.UserAgentProvider
import io.element.android.libraries.sessionstorage.api.SessionData
import io.element.android.libraries.sessionstorage.api.SessionStore
import io.element.android.libraries.workmanager.api.WorkManagerScheduler
import io.element.android.services.analytics.api.AnalyticsService
import io.element.android.services.toolbox.api.systemclock.SystemClock
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.withContext
import org.matrix.rustcomponents.sdk.ClientBuilder
@@ -61,14 +58,12 @@ class RustMatrixClientFactory(
private val sessionStore: SessionStore,
private val userAgentProvider: UserAgentProvider,
private val proxyProvider: ProxyProvider,
private val clock: SystemClock,
private val analyticsService: AnalyticsService,
private val featureFlagService: FeatureFlagService,
private val timelineEventFilterFactory: TimelineEventFilterFactory,
private val clientBuilderProvider: ClientBuilderProvider,
private val sqliteStoreBuilderProvider: SqliteStoreBuilderProvider,
private val workManagerScheduler: WorkManagerScheduler,
private val clientBuilderEnterpriseHook: ClientBuilderEnterpriseHook,
private val innerMatrixClientFactory: RustMatrixClient.Factory,
) {
private val sessionDelegate = RustClientSessionDelegate(
sessionStore = sessionStore,
@@ -131,22 +126,14 @@ class RustMatrixClientFactory(
.withOfflineMode()
.finish()
RustMatrixClient(
innerMatrixClientFactory.create(
sessionPaths = sessionData.getSessionPaths(),
innerClient = client,
sessionStore = sessionStore,
appCoroutineScope = appCoroutineScope,
sessionDelegate = sessionDelegate,
innerSyncService = syncService,
dispatchers = coroutineDispatchers,
baseCacheDirectory = cacheDirectory,
clock = clock,
timelineEventFilterFactory = timelineEventFilterFactory,
featureFlagService = featureFlagService,
analyticsService = analyticsService,
workManagerScheduler = workManagerScheduler,
contentScanner = client.contentScanner()?.let { RustContentScanner(client, it) },
isMessageSearchAvailable = isMessageSearchAvailable,
sessionDelegate = sessionDelegate,
).also {
Timber.tag("RustMatrixClient").i("Creating Client with access token '$anonymizedAccessToken' and refresh token '$anonymizedRefreshToken'")
}
@@ -0,0 +1,27 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.impl
import io.element.android.libraries.matrix.api.SdkPendingTask
import io.element.android.libraries.matrix.impl.util.cancelAndDestroy
import org.matrix.rustcomponents.sdk.TaskHandle
import java.util.concurrent.atomic.AtomicBoolean
class RustSdkPendingTask(
private val taskHandle: TaskHandle,
) : SdkPendingTask {
private val closed = AtomicBoolean(false)
override fun isRunning(): Boolean = !closed.get() && !taskHandle.isFinished()
override fun close() {
if (!closed.getAndSet(true)) {
taskHandle.cancelAndDestroy()
}
}
}
@@ -0,0 +1,71 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.impl.search
import dev.zacsweers.metro.Assisted
import dev.zacsweers.metro.AssistedFactory
import dev.zacsweers.metro.AssistedInject
import io.element.android.libraries.core.extensions.runCatchingExceptions
import io.element.android.libraries.matrix.api.SdkPendingTask
import io.element.android.libraries.matrix.api.core.SessionId
import io.element.android.libraries.matrix.api.search.SearchBackfillService
import io.element.android.libraries.matrix.api.search.SearchBackfillStrategy
import io.element.android.libraries.matrix.impl.RustSdkPendingTask
import io.element.android.libraries.matrix.impl.search.workmanager.SearchBackfillRequestBuilder
import io.element.android.libraries.workmanager.api.WorkManagerRequestType
import io.element.android.libraries.workmanager.api.WorkManagerScheduler
import org.matrix.rustcomponents.sdk.Client
import timber.log.Timber
import java.util.concurrent.atomic.AtomicReference
@AssistedInject
class RustSearchBackfillService(
@Assisted private val innerClient: Client,
private val searchBackfillRequestBuilderFactory: SearchBackfillRequestBuilder.Factory,
private val workManagerScheduler: WorkManagerScheduler,
) : SearchBackfillService {
@AssistedFactory
interface Factory {
fun create(innerClient: Client): RustSearchBackfillService
}
private val currentSearchBackfillTaskHandle: AtomicReference<SdkPendingTask?> = AtomicReference(null)
override suspend fun schedulePeriodicSearchBackfill(strategy: SearchBackfillStrategy): Result<Unit> = runCatchingExceptions {
val sessionId = SessionId(innerClient.userId())
if (workManagerScheduler.hasPendingWork(sessionId, WorkManagerRequestType.SEARCH_BACKFILL)) {
Timber.d("Background search backfill already scheduled for session $sessionId")
return@runCatchingExceptions
}
Timber.d("Scheduling background search backfill for session $sessionId")
val builder = searchBackfillRequestBuilderFactory.create(sessionId)
workManagerScheduler.submit(builder)
}
override fun cancelPeriodicSearchBackfill() {
val sessionId = SessionId(innerClient.userId())
Timber.d("Cancelling background search backfill for session $sessionId")
workManagerScheduler.cancel(sessionId, WorkManagerRequestType.SEARCH_BACKFILL)
}
override fun startSearchBackfill(strategy: SearchBackfillStrategy): Result<SdkPendingTask> = runCatchingExceptions {
if (currentSearchBackfillTaskHandle.get()?.isRunning() == true) {
error("Search backfill is already running")
}
RustSdkPendingTask(innerClient.runSearchBackfill(strategy.map()))
.also { currentSearchBackfillTaskHandle.set(it) }
}
override fun isSearchBackfillRunning(): Boolean {
return currentSearchBackfillTaskHandle.get()?.isRunning() ?: false
}
override fun cancelSearchBackfill(): Result<Unit> = runCatchingExceptions {
currentSearchBackfillTaskHandle.getAndSet(null)?.close()
}
}
@@ -0,0 +1,21 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.impl.search
import io.element.android.libraries.matrix.api.search.SearchBackfillStrategy
import uniffi.matrix_sdk.SearchBackfillStrategy as RustSearchBackfillStrategy
fun RustSearchBackfillStrategy.map(): SearchBackfillStrategy = when (this) {
RustSearchBackfillStrategy.FOREGROUND -> SearchBackfillStrategy.FOREGROUND
RustSearchBackfillStrategy.BACKGROUND -> SearchBackfillStrategy.BACKGROUND
}
fun SearchBackfillStrategy.map(): RustSearchBackfillStrategy = when (this) {
SearchBackfillStrategy.FOREGROUND -> RustSearchBackfillStrategy.FOREGROUND
SearchBackfillStrategy.BACKGROUND -> RustSearchBackfillStrategy.BACKGROUND
}
@@ -0,0 +1,88 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.impl.search.workmanager
import android.net.NetworkCapabilities
import android.net.NetworkRequest
import androidx.work.Constraints
import androidx.work.NetworkType
import androidx.work.PeriodicWorkRequestBuilder
import androidx.work.workDataOf
import dev.zacsweers.metro.AppScope
import dev.zacsweers.metro.Assisted
import dev.zacsweers.metro.AssistedFactory
import dev.zacsweers.metro.AssistedInject
import dev.zacsweers.metro.ContributesBinding
import io.element.android.features.networkmonitor.api.NetworkMonitor
import io.element.android.libraries.matrix.api.core.SessionId
import io.element.android.libraries.matrix.impl.search.workmanager.SearchBackfillWorker.Companion.SESSION_ID_PARAM
import io.element.android.libraries.workmanager.api.WorkManagerRequestBuilder
import io.element.android.libraries.workmanager.api.WorkManagerRequestType
import io.element.android.libraries.workmanager.api.WorkManagerRequestWrapper
import io.element.android.libraries.workmanager.api.WorkManagerWorkerType
import io.element.android.libraries.workmanager.api.workManagerTag
import kotlinx.coroutines.flow.first
import timber.log.Timber
import kotlin.time.Duration.Companion.hours
import kotlin.time.Duration.Companion.minutes
import kotlin.time.toJavaDuration
interface SearchBackfillRequestBuilder : WorkManagerRequestBuilder {
fun interface Factory {
fun create(sessionId: SessionId): SearchBackfillRequestBuilder
}
}
/**
* Builds a work request that runs the search backfill (back paginations) only when there is network connectivity and the device is charging.
*/
@AssistedInject
class DefaultSearchBackfillRequestBuilder(
@Assisted private val sessionId: SessionId,
private val networkMonitor: NetworkMonitor,
) : SearchBackfillRequestBuilder {
@AssistedFactory
@ContributesBinding(AppScope::class)
interface Factory : SearchBackfillRequestBuilder.Factory {
override fun create(sessionId: SessionId): DefaultSearchBackfillRequestBuilder
}
override suspend fun build(): Result<List<WorkManagerRequestWrapper>> {
val tag = workManagerTag(sessionId, WorkManagerRequestType.SEARCH_BACKFILL)
val type = WorkManagerWorkerType.Default
val networkRequestBuilder = NetworkRequest.Builder()
// Only allow unmetered networks to avoid using mobile data for backfilling.
.addCapability(NetworkCapabilities.NET_CAPABILITY_NOT_METERED)
// Some customers use a VPN to connect to their homeserver, so VPN networks must be allowed.
.removeCapability(NetworkCapabilities.NET_CAPABILITY_NOT_VPN)
// In an air-gapped environment, validating internet connectivity would fail and the worker would never run.
if (networkMonitor.isInAirGappedEnvironment.first()) {
Timber.d("In an air-gapped environment, not adding NET_CAPABILITY_VALIDATED to the network request")
networkRequestBuilder.removeCapability(NetworkCapabilities.NET_CAPABILITY_VALIDATED)
} else {
networkRequestBuilder.addCapability(NetworkCapabilities.NET_CAPABILITY_VALIDATED)
}
val constraints = Constraints.Builder()
.setRequiredNetworkRequest(networkRequestBuilder.build(), NetworkType.UNMETERED)
.setRequiresCharging(true)
.build()
val request = PeriodicWorkRequestBuilder<SearchBackfillWorker>(1.hours.toJavaDuration())
// Add a delay before the first execution to avoid running the backfill immediately after the app starts
.setInitialDelay(5.minutes.toJavaDuration())
.setInputData(workDataOf(SESSION_ID_PARAM to sessionId.value))
.addTag(tag)
.setConstraints(constraints)
.build()
return Result.success(listOf(WorkManagerRequestWrapper(request, type)))
}
}
@@ -0,0 +1,81 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.impl.search.workmanager
import android.content.Context
import androidx.work.CoroutineWorker
import androidx.work.WorkerParameters
import dev.zacsweers.metro.AppScope
import dev.zacsweers.metro.Assisted
import dev.zacsweers.metro.AssistedFactory
import dev.zacsweers.metro.AssistedInject
import dev.zacsweers.metro.ContributesIntoMap
import dev.zacsweers.metro.binding
import io.element.android.libraries.di.annotations.ApplicationContext
import io.element.android.libraries.matrix.api.MatrixClientProvider
import io.element.android.libraries.matrix.api.core.SessionId
import io.element.android.libraries.matrix.api.search.SearchBackfillStrategy
import io.element.android.libraries.workmanager.api.di.MetroWorkerFactory
import io.element.android.libraries.workmanager.api.di.WorkerKey
import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.delay
import kotlinx.coroutines.withTimeout
import timber.log.Timber
import kotlin.time.Duration.Companion.minutes
import kotlin.time.Duration.Companion.seconds
@AssistedInject
class SearchBackfillWorker(
@Assisted params: WorkerParameters,
@ApplicationContext context: Context,
private val matrixClientProvider: MatrixClientProvider,
) : CoroutineWorker(context, params) {
companion object {
const val SESSION_ID_PARAM = "session_id"
private val TIMEOUT = 10.minutes
private val POLL_INTERVAL = 5.seconds
}
override suspend fun doWork(): Result {
Timber.d("Starting search backfill. This will run up to $TIMEOUT")
val sessionId = inputData.getString(SESSION_ID_PARAM)?.let(::SessionId) ?: return Result.failure()
val client = matrixClientProvider.getOrRestore(sessionId).getOrNull() ?: return Result.failure()
if (client.searchBackfillService.isSearchBackfillRunning()) {
Timber.d("Search backfill is already running for session $sessionId, nothing to do")
return Result.success()
}
Timber.d("Starting search backfill for session $sessionId")
val task = client.searchBackfillService.startSearchBackfill(SearchBackfillStrategy.BACKGROUND).getOrElse {
Timber.e(it, "Failed to start search backfill")
return Result.failure()
}
return task.use {
try {
withTimeout(TIMEOUT) {
while (task.isRunning()) {
delay(POLL_INTERVAL)
}
}
Timber.d("Search backfill finished")
Result.success()
} catch (_: TimeoutCancellationException) {
Timber.d("Search backfill timed out after $TIMEOUT, cancelling it")
// Run again later to continue backfilling.
Result.retry()
}
}
}
@ContributesIntoMap(AppScope::class, binding = binding<MetroWorkerFactory.WorkerInstanceFactory<*>>())
@WorkerKey(SearchBackfillWorker::class)
@AssistedFactory
interface Factory : MetroWorkerFactory.WorkerInstanceFactory<SearchBackfillWorker>
}
@@ -12,8 +12,12 @@ import com.google.common.truth.Truth.assertThat
import io.element.android.features.enterprise.test.FakeClientBuilderEnterpriseHook
import io.element.android.libraries.featureflag.test.FakeFeatureFlagService
import io.element.android.libraries.matrix.api.core.SessionId
import io.element.android.libraries.matrix.api.paths.SessionPaths
import io.element.android.libraries.matrix.api.scanner.ContentScanner
import io.element.android.libraries.matrix.impl.auth.FakeProxyProvider
import io.element.android.libraries.matrix.impl.room.FakeTimelineEventFilterFactory
import io.element.android.libraries.matrix.impl.search.RustSearchBackfillService
import io.element.android.libraries.matrix.impl.search.workmanager.SearchBackfillRequestBuilder
import io.element.android.libraries.matrix.impl.storage.FakeSqliteStoreBuilderProvider
import io.element.android.libraries.matrix.impl.storage.SqliteStoreBuilderProvider
import io.element.android.libraries.network.useragent.SimpleUserAgentProvider
@@ -21,6 +25,7 @@ import io.element.android.libraries.sessionstorage.api.SessionStore
import io.element.android.libraries.sessionstorage.test.InMemorySessionStore
import io.element.android.libraries.sessionstorage.test.aSessionData
import io.element.android.libraries.workmanager.api.WorkManagerRequestBuilder
import io.element.android.libraries.workmanager.api.WorkManagerRequestWrapper
import io.element.android.libraries.workmanager.test.FakeWorkManagerScheduler
import io.element.android.services.analytics.test.FakeAnalyticsService
import io.element.android.services.toolbox.test.systemclock.FakeSystemClock
@@ -29,13 +34,15 @@ import io.element.android.tests.testutils.testCoroutineDispatchers
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.runTest
import org.junit.Test
import org.matrix.rustcomponents.sdk.Client
import java.io.File
import org.matrix.rustcomponents.sdk.SyncService as ClientSyncService
class RustMatrixClientFactoryTest {
@Test
fun test() = runTest {
val scheduleVacuumLambda = lambdaRecorder<WorkManagerRequestBuilder, Unit> {}
val workManagerScheduler = FakeWorkManagerScheduler(submitLambda = scheduleVacuumLambda)
val workManagerScheduler = FakeWorkManagerScheduler(submitLambda = scheduleVacuumLambda, cancelLambda = { _, _ -> })
val sut = createRustMatrixClientFactory(workManagerScheduler = workManagerScheduler)
val result = sut.create(aSessionData())
@@ -52,7 +59,7 @@ fun TestScope.createRustMatrixClientFactory(
updateUserProfileResult = { _, _, _, _ -> },
),
clientBuilderProvider: ClientBuilderProvider = FakeClientBuilderProvider(),
workManagerScheduler: FakeWorkManagerScheduler = FakeWorkManagerScheduler(),
workManagerScheduler: FakeWorkManagerScheduler = FakeWorkManagerScheduler(cancelLambda = { _, _ -> }),
sqliteStoreBuilderProvider: SqliteStoreBuilderProvider = FakeSqliteStoreBuilderProvider(),
) = RustMatrixClientFactory(
cacheDirectory = cacheDirectory,
@@ -61,12 +68,52 @@ fun TestScope.createRustMatrixClientFactory(
sessionStore = sessionStore,
userAgentProvider = SimpleUserAgentProvider(),
proxyProvider = FakeProxyProvider(),
clock = FakeSystemClock(),
analyticsService = FakeAnalyticsService(),
featureFlagService = FakeFeatureFlagService(),
timelineEventFilterFactory = FakeTimelineEventFilterFactory(),
clientBuilderProvider = clientBuilderProvider,
sqliteStoreBuilderProvider = sqliteStoreBuilderProvider,
workManagerScheduler = workManagerScheduler,
clientBuilderEnterpriseHook = FakeClientBuilderEnterpriseHook(),
innerMatrixClientFactory = createRustMatrixClientInnerFactory(sessionStore, workManagerScheduler),
)
private fun TestScope.createRustMatrixClientInnerFactory(
sessionStore: SessionStore,
workManagerScheduler: FakeWorkManagerScheduler,
) = object : RustMatrixClient.Factory {
override fun create(
sessionPaths: SessionPaths,
innerClient: Client,
innerSyncService: ClientSyncService,
baseCacheDirectory: File,
contentScanner: ContentScanner?,
isMessageSearchAvailable: Boolean,
sessionDelegate: RustClientSessionDelegate,
) = RustMatrixClient(
sessionPaths = sessionPaths,
innerClient = innerClient,
innerSyncService = innerSyncService,
baseCacheDirectory = baseCacheDirectory,
contentScanner = contentScanner,
isMessageSearchAvailable = isMessageSearchAvailable,
sessionDelegate = sessionDelegate,
sessionStore = sessionStore,
appCoroutineScope = backgroundScope,
dispatchers = testCoroutineDispatchers(),
clock = FakeSystemClock(),
timelineEventFilterFactory = FakeTimelineEventFilterFactory(),
featureFlagService = FakeFeatureFlagService(),
analyticsService = FakeAnalyticsService(),
workManagerScheduler = workManagerScheduler,
searchBackfillServiceFactory = object : RustSearchBackfillService.Factory {
override fun create(innerClient: Client) = RustSearchBackfillService(
innerClient = innerClient,
searchBackfillRequestBuilderFactory = {
object : SearchBackfillRequestBuilder {
override suspend fun build(): Result<List<WorkManagerRequestWrapper>> = Result.success(emptyList())
}
},
workManagerScheduler = workManagerScheduler,
)
},
)
}
@@ -18,6 +18,8 @@ import io.element.android.libraries.matrix.impl.fixtures.factories.aRustUserProf
import io.element.android.libraries.matrix.impl.fixtures.fakes.FakeFfiClient
import io.element.android.libraries.matrix.impl.fixtures.fakes.FakeFfiSyncService
import io.element.android.libraries.matrix.impl.room.FakeTimelineEventFilterFactory
import io.element.android.libraries.matrix.impl.search.RustSearchBackfillService
import io.element.android.libraries.matrix.impl.search.workmanager.SearchBackfillRequestBuilder
import io.element.android.libraries.matrix.test.AN_AVATAR_URL
import io.element.android.libraries.matrix.test.A_DEVICE_ID
import io.element.android.libraries.matrix.test.A_ROOM_ID
@@ -27,6 +29,7 @@ import io.element.android.libraries.matrix.test.scanner.FakeContentScanner
import io.element.android.libraries.sessionstorage.api.SessionStore
import io.element.android.libraries.sessionstorage.test.InMemorySessionStore
import io.element.android.libraries.sessionstorage.test.aSessionData
import io.element.android.libraries.workmanager.api.WorkManagerRequestWrapper
import io.element.android.libraries.workmanager.test.FakeWorkManagerScheduler
import io.element.android.services.analytics.test.FakeAnalyticsService
import io.element.android.services.toolbox.test.systemclock.FakeSystemClock
@@ -325,8 +328,19 @@ class RustMatrixClientTest {
timelineEventFilterFactory = FakeTimelineEventFilterFactory(),
featureFlagService = FakeFeatureFlagService(),
analyticsService = FakeAnalyticsService(),
workManagerScheduler = FakeWorkManagerScheduler(submitLambda = {}),
workManagerScheduler = FakeWorkManagerScheduler(submitLambda = {}, cancelLambda = { _, _ -> }),
contentScanner = FakeContentScanner(),
isMessageSearchAvailable = false,
searchBackfillServiceFactory = object : RustSearchBackfillService.Factory {
override fun create(innerClient: Client) = RustSearchBackfillService(
innerClient = innerClient,
searchBackfillRequestBuilderFactory = {
object : SearchBackfillRequestBuilder {
override suspend fun build(): Result<List<WorkManagerRequestWrapper>> = Result.success(emptyList())
}
},
workManagerScheduler = FakeWorkManagerScheduler(submitLambda = {}, cancelLambda = { _, _ -> }),
)
},
)
}
@@ -8,19 +8,7 @@
package io.element.android.libraries.matrix.impl
import com.google.common.truth.Truth.assertThat
import io.element.android.features.enterprise.test.FakeClientBuilderEnterpriseHook
import io.element.android.libraries.featureflag.test.FakeFeatureFlagService
import io.element.android.libraries.matrix.impl.auth.FakeProxyProvider
import io.element.android.libraries.matrix.impl.paths.SessionPathsFactory
import io.element.android.libraries.matrix.impl.room.FakeTimelineEventFilterFactory
import io.element.android.libraries.matrix.impl.storage.FakeSqliteStoreBuilderProvider
import io.element.android.libraries.network.useragent.SimpleUserAgentProvider
import io.element.android.libraries.sessionstorage.api.SessionStore
import io.element.android.libraries.sessionstorage.test.InMemorySessionStore
import io.element.android.libraries.workmanager.test.FakeWorkManagerScheduler
import io.element.android.services.analytics.test.FakeAnalyticsService
import io.element.android.services.toolbox.test.systemclock.FakeSystemClock
import io.element.android.tests.testutils.testCoroutineDispatchers
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.runTest
import org.junit.Test
@@ -56,28 +44,4 @@ class RustTemporaryMatrixClientFactoryTest {
rustMatrixClientFactory = rustMatrixClientFactory,
)
}
private fun TestScope.createRustMatrixClientFactory(
cacheDirectory: File = File("/cache"),
sessionStore: SessionStore = InMemorySessionStore(
updateUserProfileResult = { _, _, _, _ -> },
),
clientBuilderProvider: ClientBuilderProvider = FakeClientBuilderProvider(),
workManagerScheduler: FakeWorkManagerScheduler = FakeWorkManagerScheduler(),
) = RustMatrixClientFactory(
cacheDirectory = cacheDirectory,
appCoroutineScope = backgroundScope,
coroutineDispatchers = testCoroutineDispatchers(),
sessionStore = sessionStore,
userAgentProvider = SimpleUserAgentProvider(),
proxyProvider = FakeProxyProvider(),
clock = FakeSystemClock(),
analyticsService = FakeAnalyticsService(),
featureFlagService = FakeFeatureFlagService(),
timelineEventFilterFactory = FakeTimelineEventFilterFactory(),
clientBuilderProvider = clientBuilderProvider,
sqliteStoreBuilderProvider = FakeSqliteStoreBuilderProvider(),
workManagerScheduler = workManagerScheduler,
clientBuilderEnterpriseHook = FakeClientBuilderEnterpriseHook(),
)
}
@@ -265,7 +265,7 @@ class RustMatrixAuthenticationServiceTest {
cacheDirectory = File("/cache"),
sessionStore = sessionStore,
clientBuilderProvider = clientBuilderProvider,
workManagerScheduler = FakeWorkManagerScheduler(submitLambda = {}),
workManagerScheduler = FakeWorkManagerScheduler(submitLambda = {}, cancelLambda = { _, _ -> }),
sqliteStoreBuilderProvider = sqliteStoreBuilderProvider,
)
return RustMatrixAuthenticationService(
@@ -45,6 +45,7 @@ import org.matrix.rustcomponents.sdk.SyncServiceBuilder
import org.matrix.rustcomponents.sdk.TaskHandle
import org.matrix.rustcomponents.sdk.UnableToDecryptDelegate
import org.matrix.rustcomponents.sdk.UserProfile
import uniffi.matrix_sdk.SearchBackfillStrategy
import uniffi.matrix_sdk_base.MediaRetentionPolicy
class FakeFfiClient(
@@ -73,6 +74,7 @@ class FakeFfiClient(
private val getUrlResult: (String) -> ByteArray = { lambdaError() },
private val getMediaThumbnailResult: (MediaSource, ULong, ULong) -> ByteArray = { _, _, _ -> byteArrayOf() },
private val contentScannerResult: () -> ContentScanner = { FakeFfiContentScanner() },
private val runSearchBackfillResult: (SearchBackfillStrategy) -> TaskHandle = { lambdaError() },
private val closeResult: () -> Unit = {},
) : Client(NoHandle) {
override fun userId(): String = userId
@@ -170,5 +172,7 @@ class FakeFfiClient(
contentScannerResult()
}
override fun runSearchBackfill(strategy: SearchBackfillStrategy): TaskHandle = runSearchBackfillResult(strategy)
override fun close() = closeResult()
}
@@ -11,7 +11,11 @@ package io.element.android.libraries.matrix.impl.fixtures.fakes
import org.matrix.rustcomponents.sdk.NoHandle
import org.matrix.rustcomponents.sdk.TaskHandle
class FakeFfiTaskHandle : TaskHandle(NoHandle) {
override fun cancel() = Unit
class FakeFfiTaskHandle(
private val isFinishedResult: () -> Boolean = { false },
private val cancelResult: () -> Unit = {},
) : TaskHandle(NoHandle) {
override fun isFinished(): Boolean = isFinishedResult()
override fun cancel() = cancelResult()
override fun destroy() = Unit
}
@@ -0,0 +1,198 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.impl.search
import com.google.common.truth.Truth.assertThat
import io.element.android.libraries.matrix.api.core.SessionId
import io.element.android.libraries.matrix.api.search.SearchBackfillStrategy
import io.element.android.libraries.matrix.impl.fixtures.fakes.FakeFfiClient
import io.element.android.libraries.matrix.impl.fixtures.fakes.FakeFfiTaskHandle
import io.element.android.libraries.matrix.impl.search.workmanager.SearchBackfillRequestBuilder
import io.element.android.libraries.matrix.test.A_SESSION_ID
import io.element.android.libraries.workmanager.api.WorkManagerRequestBuilder
import io.element.android.libraries.workmanager.api.WorkManagerRequestType
import io.element.android.libraries.workmanager.api.WorkManagerRequestWrapper
import io.element.android.libraries.workmanager.test.FakeWorkManagerScheduler
import io.element.android.tests.testutils.lambda.lambdaRecorder
import io.element.android.tests.testutils.lambda.value
import kotlinx.coroutines.test.runTest
import org.junit.Test
import org.matrix.rustcomponents.sdk.TaskHandle
import uniffi.matrix_sdk.SearchBackfillStrategy as RustSearchBackfillStrategy
class RustSearchBackfillServiceTest {
@Test
fun `schedulePeriodicSearchBackfill submits a request when nothing is pending`() = runTest {
val builder = aRequestBuilder()
val createBuilderLambda = lambdaRecorder<SessionId, SearchBackfillRequestBuilder> { builder }
val submitLambda = lambdaRecorder<WorkManagerRequestBuilder, Unit> { }
val service = createService(
requestBuilderFactory = SearchBackfillRequestBuilder.Factory(createBuilderLambda),
workManagerScheduler = FakeWorkManagerScheduler(
submitLambda = submitLambda,
hasPendingWorkLambda = { _, _ -> false },
),
)
val result = service.schedulePeriodicSearchBackfill(SearchBackfillStrategy.BACKGROUND)
assertThat(result.isSuccess).isTrue()
createBuilderLambda.assertions().isCalledOnce().with(value(A_SESSION_ID))
submitLambda.assertions().isCalledOnce().with(value(builder))
}
@Test
fun `schedulePeriodicSearchBackfill does nothing when work is already pending`() = runTest {
val hasPendingWorkLambda = lambdaRecorder<SessionId, WorkManagerRequestType, Boolean> { _, _ -> true }
val createBuilderLambda = lambdaRecorder<SessionId, SearchBackfillRequestBuilder> { error("Should not be called") }
val submitLambda = lambdaRecorder<WorkManagerRequestBuilder, Unit> { error("Should not be called") }
val service = createService(
requestBuilderFactory = SearchBackfillRequestBuilder.Factory(createBuilderLambda),
workManagerScheduler = FakeWorkManagerScheduler(
submitLambda = submitLambda,
hasPendingWorkLambda = hasPendingWorkLambda,
),
)
val result = service.schedulePeriodicSearchBackfill(SearchBackfillStrategy.BACKGROUND)
assertThat(result.isSuccess).isTrue()
hasPendingWorkLambda.assertions().isCalledOnce().with(value(A_SESSION_ID), value(WorkManagerRequestType.SEARCH_BACKFILL))
createBuilderLambda.assertions().isNeverCalled()
submitLambda.assertions().isNeverCalled()
}
@Test
fun `schedulePeriodicSearchBackfill returns a failure when submitting fails`() = runTest {
val service = createService(
requestBuilderFactory = SearchBackfillRequestBuilder.Factory { aRequestBuilder() },
workManagerScheduler = FakeWorkManagerScheduler(submitLambda = { throw IllegalStateException("boom") }),
)
val result = service.schedulePeriodicSearchBackfill(SearchBackfillStrategy.BACKGROUND)
assertThat(result.exceptionOrNull()).isInstanceOf(IllegalStateException::class.java)
}
@Test
fun `cancelPeriodicSearchBackfill cancels the search backfill work of the session`() {
val cancelLambda = lambdaRecorder<SessionId, WorkManagerRequestType?, Unit> { _, _ -> }
val service = createService(workManagerScheduler = FakeWorkManagerScheduler(cancelLambda = cancelLambda))
service.cancelPeriodicSearchBackfill()
cancelLambda.assertions().isCalledOnce().with(value(A_SESSION_ID), value(WorkManagerRequestType.SEARCH_BACKFILL))
}
@Test
fun `startSearchBackfill maps the strategy and returns a running task`() {
val strategyLambda = lambdaRecorder<RustSearchBackfillStrategy, TaskHandle> { FakeFfiTaskHandle(isFinishedResult = { false }) }
val service = createService(client = FakeFfiClient(runSearchBackfillResult = strategyLambda))
val task = service.startSearchBackfill(SearchBackfillStrategy.FOREGROUND).getOrThrow()
strategyLambda.assertions().isCalledOnce().with(value(RustSearchBackfillStrategy.FOREGROUND))
assertThat(task.isRunning()).isTrue()
assertThat(service.isSearchBackfillRunning()).isTrue()
}
@Test
fun `startSearchBackfill maps the background strategy`() {
val strategyLambda = lambdaRecorder<RustSearchBackfillStrategy, TaskHandle> { FakeFfiTaskHandle() }
val service = createService(client = FakeFfiClient(runSearchBackfillResult = strategyLambda))
service.startSearchBackfill(SearchBackfillStrategy.BACKGROUND)
strategyLambda.assertions().isCalledOnce().with(value(RustSearchBackfillStrategy.BACKGROUND))
}
@Test
fun `startSearchBackfill fails when a backfill is already running`() {
val strategyLambda = lambdaRecorder<RustSearchBackfillStrategy, TaskHandle> { FakeFfiTaskHandle(isFinishedResult = { false }) }
val service = createService(client = FakeFfiClient(runSearchBackfillResult = strategyLambda))
service.startSearchBackfill(SearchBackfillStrategy.FOREGROUND)
val result = service.startSearchBackfill(SearchBackfillStrategy.FOREGROUND)
assertThat(result.exceptionOrNull()).isInstanceOf(IllegalStateException::class.java)
strategyLambda.assertions().isCalledOnce()
}
@Test
fun `startSearchBackfill can start again once the previous task has finished`() {
val strategyLambda = lambdaRecorder<RustSearchBackfillStrategy, TaskHandle> { FakeFfiTaskHandle(isFinishedResult = { true }) }
val service = createService(client = FakeFfiClient(runSearchBackfillResult = strategyLambda))
assertThat(service.startSearchBackfill(SearchBackfillStrategy.FOREGROUND).isSuccess).isTrue()
assertThat(service.isSearchBackfillRunning()).isFalse()
assertThat(service.startSearchBackfill(SearchBackfillStrategy.FOREGROUND).isSuccess).isTrue()
strategyLambda.assertions().isCalledExactly(2)
}
@Test
fun `startSearchBackfill returns a failure when the SDK fails`() {
val service = createService(client = FakeFfiClient(runSearchBackfillResult = { error("boom") }))
val result = service.startSearchBackfill(SearchBackfillStrategy.FOREGROUND)
assertThat(result.isFailure).isTrue()
assertThat(service.isSearchBackfillRunning()).isFalse()
}
@Test
fun `isSearchBackfillRunning is false when nothing was started`() {
assertThat(createService().isSearchBackfillRunning()).isFalse()
}
@Test
fun `cancelSearchBackfill cancels the running task`() {
val cancelLambda = lambdaRecorder<Unit> { }
val service = createService(
client = FakeFfiClient(runSearchBackfillResult = { FakeFfiTaskHandle(cancelResult = cancelLambda) }),
)
service.startSearchBackfill(SearchBackfillStrategy.FOREGROUND)
val result = service.cancelSearchBackfill()
assertThat(result.isSuccess).isTrue()
cancelLambda.assertions().isCalledOnce()
assertThat(service.isSearchBackfillRunning()).isFalse()
}
@Test
fun `cancelSearchBackfill without a running task is a no-op`() {
assertThat(createService().cancelSearchBackfill().isSuccess).isTrue()
}
@Test
fun `startSearchBackfill works again after cancelSearchBackfill`() {
val strategyLambda = lambdaRecorder<RustSearchBackfillStrategy, TaskHandle> { FakeFfiTaskHandle() }
val service = createService(client = FakeFfiClient(runSearchBackfillResult = strategyLambda))
service.startSearchBackfill(SearchBackfillStrategy.FOREGROUND)
service.cancelSearchBackfill()
assertThat(service.startSearchBackfill(SearchBackfillStrategy.FOREGROUND).isSuccess).isTrue()
strategyLambda.assertions().isCalledExactly(2)
}
private fun aRequestBuilder() = object : SearchBackfillRequestBuilder {
override suspend fun build(): Result<List<WorkManagerRequestWrapper>> = Result.success(emptyList())
}
private fun createService(
client: FakeFfiClient = FakeFfiClient(),
requestBuilderFactory: SearchBackfillRequestBuilder.Factory = SearchBackfillRequestBuilder.Factory { aRequestBuilder() },
workManagerScheduler: FakeWorkManagerScheduler = FakeWorkManagerScheduler(),
) = RustSearchBackfillService(
innerClient = client,
searchBackfillRequestBuilderFactory = requestBuilderFactory,
workManagerScheduler = workManagerScheduler,
)
}
@@ -0,0 +1,78 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.impl.search.workmanager
import android.net.NetworkCapabilities
import androidx.work.PeriodicWorkRequest
import androidx.work.hasKeyWithValueOfType
import com.google.common.truth.Truth.assertThat
import io.element.android.features.networkmonitor.test.FakeNetworkMonitor
import io.element.android.libraries.matrix.test.A_SESSION_ID
import io.element.android.libraries.workmanager.api.WorkManagerRequestType
import io.element.android.libraries.workmanager.api.WorkManagerWorkerType
import io.element.android.libraries.workmanager.api.workManagerTag
import io.element.android.tests.testutils.robolectric.RobolectricTest
import kotlinx.coroutines.test.runTest
import org.junit.Test
import kotlin.time.Duration.Companion.hours
import kotlin.time.Duration.Companion.minutes
class DefaultSearchBackfillRequestBuilderTest : RobolectricTest() {
private val expectedTag = workManagerTag(A_SESSION_ID, WorkManagerRequestType.SEARCH_BACKFILL)
@Test
fun `build - creates a periodic request with session id and tag`() = runTest {
val result = createBuilder().build().getOrThrow().single()
assertThat(result.type).isEqualTo(WorkManagerWorkerType.Default)
result.request.run {
assertThat(this).isInstanceOf(PeriodicWorkRequest::class.java)
assertThat(workSpec.isPeriodic).isTrue()
assertThat(workSpec.intervalDuration).isEqualTo(1.hours.inWholeMilliseconds)
assertThat(workSpec.initialDelay).isEqualTo(5.minutes.inWholeMilliseconds)
assertThat(workSpec.input.hasKeyWithValueOfType<String>(SearchBackfillWorker.SESSION_ID_PARAM)).isTrue()
assertThat(workSpec.input.getString(SearchBackfillWorker.SESSION_ID_PARAM)).isEqualTo(A_SESSION_ID.value)
assertThat(tags).contains(expectedTag)
assertThat(workSpec.workerClassName).isEqualTo(SearchBackfillWorker::class.java.name)
}
}
@Test
fun `build - requires charging and an unmetered network`() = runTest {
val result = createBuilder().build().getOrThrow().single()
val constraints = result.request.workSpec.constraints
assertThat(constraints.requiresCharging()).isTrue()
val networkRequest = constraints.requiredNetworkRequest
assertThat(networkRequest).isNotNull()
assertThat(networkRequest!!.hasCapability(NetworkCapabilities.NET_CAPABILITY_NOT_METERED)).isTrue()
}
@Test
fun `build - validates the network when not in an air-gapped env`() = runTest {
val result = createBuilder(isInAirGapEnvironment = false).build().getOrThrow().single()
val networkRequest = result.request.workSpec.constraints.requiredNetworkRequest!!
assertThat(networkRequest.hasCapability(NetworkCapabilities.NET_CAPABILITY_VALIDATED)).isTrue()
}
@Test
fun `build - does not validate the network when in an air-gapped env`() = runTest {
val result = createBuilder(isInAirGapEnvironment = true).build().getOrThrow().single()
val networkRequest = result.request.workSpec.constraints.requiredNetworkRequest!!
assertThat(networkRequest.hasCapability(NetworkCapabilities.NET_CAPABILITY_VALIDATED)).isFalse()
}
private fun createBuilder(
isInAirGapEnvironment: Boolean = false,
) = DefaultSearchBackfillRequestBuilder(
sessionId = A_SESSION_ID,
networkMonitor = FakeNetworkMonitor().apply { givenIsInAirGappedEnvironment(isInAirGapEnvironment) },
)
}
@@ -0,0 +1,161 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.impl.search.workmanager
import androidx.test.platform.app.InstrumentationRegistry
import androidx.work.Data
import androidx.work.ListenableWorker
import androidx.work.WorkerParameters
import androidx.work.impl.utils.taskexecutor.WorkManagerTaskExecutor
import androidx.work.workDataOf
import com.google.common.truth.Truth.assertThat
import com.google.common.util.concurrent.ListenableFuture
import io.element.android.libraries.matrix.api.MatrixClient
import io.element.android.libraries.matrix.api.SdkPendingTask
import io.element.android.libraries.matrix.api.search.SearchBackfillStrategy
import io.element.android.libraries.matrix.test.A_SESSION_ID
import io.element.android.libraries.matrix.test.FakeMatrixClient
import io.element.android.libraries.matrix.test.FakeMatrixClientProvider
import io.element.android.libraries.matrix.test.FakeSdkPendingTask
import io.element.android.libraries.matrix.test.search.FakeSearchBackfillService
import io.element.android.libraries.workmanager.api.di.MetroWorkerFactory
import io.element.android.tests.testutils.lambda.lambdaRecorder
import io.element.android.tests.testutils.lambda.value
import io.element.android.tests.testutils.robolectric.RobolectricTest
import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.async
import kotlinx.coroutines.test.TestScope
import kotlinx.coroutines.test.advanceTimeBy
import kotlinx.coroutines.test.runTest
import org.junit.Test
import java.util.UUID
import java.util.concurrent.Executor
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
import kotlin.time.Duration.Companion.minutes
import kotlin.time.Duration.Companion.seconds
@OptIn(ExperimentalCoroutinesApi::class)
class SearchBackfillWorkerTest : RobolectricTest() {
@Test
fun `missing session id - fails`() = runTest {
val worker = createWorker(input = null, client = FakeMatrixClient())
assertThat(worker.doWork()).isEqualTo(ListenableWorker.Result.failure())
}
@Test
fun `client cannot be restored - fails`() = runTest {
val worker = createWorker(client = null)
assertThat(worker.doWork()).isEqualTo(ListenableWorker.Result.failure())
}
@Test
fun `backfill already running - does not start another one`() = runTest {
val startLambda = lambdaRecorder<SearchBackfillStrategy, Result<SdkPendingTask>> { error("Should not be called") }
val client = FakeMatrixClient(
searchBackfillService = FakeSearchBackfillService(
isSearchBackfillRunningLambda = { true },
startSearchBackfillLambda = startLambda,
),
)
val result = createWorker(client = client).doWork()
assertThat(result).isEqualTo(ListenableWorker.Result.success())
startLambda.assertions().isNeverCalled()
}
@Test
fun `start fails - fails`() = runTest {
val client = FakeMatrixClient(
searchBackfillService = FakeSearchBackfillService(
startSearchBackfillLambda = { Result.failure(IllegalStateException("boom")) },
),
)
assertThat(createWorker(client = client).doWork()).isEqualTo(ListenableWorker.Result.failure())
}
@Test
fun `backfill finishes - succeeds and closes the task`() = runTest {
var running = true
val closeLambda = lambdaRecorder<Unit> { }
val startLambda = lambdaRecorder<SearchBackfillStrategy, Result<SdkPendingTask>> {
Result.success(FakeSdkPendingTask(isRunningLambda = { running }, closeLambda = closeLambda))
}
val client = FakeMatrixClient(searchBackfillService = FakeSearchBackfillService(startSearchBackfillLambda = startLambda))
val deferred = async { createWorker(client = client).doWork() }
advanceTimeBy(10.seconds)
running = false
advanceTimeBy(10.seconds)
assertThat(deferred.await()).isEqualTo(ListenableWorker.Result.success())
startLambda.assertions().isCalledOnce().with(value(SearchBackfillStrategy.BACKGROUND))
closeLambda.assertions().isCalledOnce()
}
@Test
fun `backfill times out after 10 minutes - closes the task and retries`() = runTest {
val closeLambda = lambdaRecorder<Unit> { }
val client = FakeMatrixClient(
searchBackfillService = FakeSearchBackfillService(
startSearchBackfillLambda = { Result.success(FakeSdkPendingTask(isRunningLambda = { true }, closeLambda = closeLambda)) },
),
)
val deferred = async { createWorker(client = client).doWork() }
advanceTimeBy(9.minutes)
assertThat(deferred.isCompleted).isFalse()
advanceTimeBy(2.minutes)
assertThat(deferred.await()).isEqualTo(ListenableWorker.Result.retry())
closeLambda.assertions().isCalledOnce()
}
private fun TestScope.createWorker(
input: String? = A_SESSION_ID.value,
client: MatrixClient?,
) = SearchBackfillWorker(
params = createWorkerParams(
inputData = input?.let { workDataOf(SearchBackfillWorker.SESSION_ID_PARAM to it) } ?: Data.EMPTY,
),
context = InstrumentationRegistry.getInstrumentation().context,
matrixClientProvider = FakeMatrixClientProvider(
getClient = { client?.let { Result.success(it) } ?: Result.failure(IllegalStateException("No client")) }
),
)
private fun TestScope.createWorkerParams(
inputData: Data = Data.EMPTY,
): WorkerParameters = WorkerParameters(
UUID.randomUUID(),
inputData,
emptySet(),
WorkerParameters.RuntimeExtras(),
0,
0,
Executors.newSingleThreadExecutor(),
backgroundScope.coroutineContext,
WorkManagerTaskExecutor(Executors.newSingleThreadExecutor()),
MetroWorkerFactory(emptyMap()),
{ _, _, _ -> FakeListenableFuture() },
{ _, _, _ -> FakeListenableFuture() },
)
}
private class FakeListenableFuture<T> : ListenableFuture<T> {
override fun addListener(listener: Runnable, executor: Executor) = Unit
override fun cancel(mayInterruptIfRunning: Boolean): Boolean = true
override fun get(): T? = null
override fun get(timeout: Long, unit: TimeUnit?): T? = null
override fun isCancelled(): Boolean = false
override fun isDone(): Boolean = false
}
@@ -40,6 +40,7 @@ import io.element.android.libraries.matrix.api.roomdirectory.RoomDirectoryServic
import io.element.android.libraries.matrix.api.roomlist.RoomListService
import io.element.android.libraries.matrix.api.scanner.ContentScanner
import io.element.android.libraries.matrix.api.search.MessageSearchService
import io.element.android.libraries.matrix.api.search.SearchBackfillService
import io.element.android.libraries.matrix.api.spaces.SpaceService
import io.element.android.libraries.matrix.api.sync.SlidingSyncVersion
import io.element.android.libraries.matrix.api.sync.SyncService
@@ -57,6 +58,7 @@ import io.element.android.libraries.matrix.test.pushers.FakePushersService
import io.element.android.libraries.matrix.test.roomdirectory.FakeRoomDirectoryService
import io.element.android.libraries.matrix.test.roomlist.FakeRoomListService
import io.element.android.libraries.matrix.test.search.FakeMessageSearchService
import io.element.android.libraries.matrix.test.search.FakeSearchBackfillService
import io.element.android.libraries.matrix.test.spaces.FakeSpaceService
import io.element.android.libraries.matrix.test.sync.FakeSyncService
import io.element.android.libraries.matrix.test.verification.FakeSessionVerificationService
@@ -131,6 +133,7 @@ class FakeMatrixClient(
private val markRoomAsFullyReadResult: (RoomId, EventId) -> Result<Unit> = { _, _ -> lambdaError() },
private val markAllRoomsAsReadResult: () -> Result<Unit> = { Result.success(Unit) },
private val performDatabaseVacuumLambda: () -> Result<Unit> = { lambdaError() },
override val searchBackfillService: SearchBackfillService = FakeSearchBackfillService(),
private val getDatabaseSizesLambda: () -> Result<SdkStoreSizes> = { lambdaError() },
private val resetWellKnownConfigLambda: () -> Result<Unit> = { lambdaError() },
private val enableAutomaticCallStatusLambda: (Boolean) -> Unit = { },
@@ -0,0 +1,20 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.test
import io.element.android.libraries.matrix.api.SdkPendingTask
import io.element.android.tests.testutils.lambda.lambdaError
class FakeSdkPendingTask(
private val isRunningLambda: () -> Boolean = { lambdaError() },
private val closeLambda: () -> Unit = { lambdaError() },
) : SdkPendingTask {
override fun isRunning(): Boolean = isRunningLambda()
override fun close() = closeLambda()
}
@@ -0,0 +1,41 @@
/*
* Copyright (c) 2026 Element Creations Ltd.
*
* SPDX-License-Identifier: AGPL-3.0-only OR LicenseRef-Element-Commercial.
* Please see LICENSE files in the repository root for full details.
*/
package io.element.android.libraries.matrix.test.search
import io.element.android.libraries.matrix.api.SdkPendingTask
import io.element.android.libraries.matrix.api.search.SearchBackfillService
import io.element.android.libraries.matrix.api.search.SearchBackfillStrategy
import io.element.android.tests.testutils.lambda.lambdaError
class FakeSearchBackfillService(
private val schedulePeriodicSearchBackfillLambda: (SearchBackfillStrategy) -> Result<Unit> = { lambdaError() },
private val cancelPeriodicSearchBackfillLambda: () -> Unit = { lambdaError() },
private val startSearchBackfillLambda: (SearchBackfillStrategy) -> Result<SdkPendingTask> = { lambdaError() },
private val isSearchBackfillRunningLambda: () -> Boolean = { false },
private val cancelSearchBackfillLambda: () -> Result<Unit> = { lambdaError() },
) : SearchBackfillService {
override suspend fun schedulePeriodicSearchBackfill(strategy: SearchBackfillStrategy): Result<Unit> {
return schedulePeriodicSearchBackfillLambda(strategy)
}
override fun cancelPeriodicSearchBackfill() {
cancelPeriodicSearchBackfillLambda()
}
override fun startSearchBackfill(strategy: SearchBackfillStrategy): Result<SdkPendingTask> {
return startSearchBackfillLambda(strategy)
}
override fun isSearchBackfillRunning(): Boolean {
return isSearchBackfillRunningLambda()
}
override fun cancelSearchBackfill(): Result<Unit> {
return cancelSearchBackfillLambda()
}
}
@@ -43,6 +43,7 @@ fun workManagerTag(sessionId: SessionId, requestType: WorkManagerRequestType): S
val prefix = when (requestType) {
WorkManagerRequestType.NOTIFICATION_SYNC -> "notifications"
WorkManagerRequestType.DB_VACUUM -> "db_vacuum"
WorkManagerRequestType.SEARCH_BACKFILL -> "search_backfill"
}
return "$prefix-$sessionId"
}
@@ -50,4 +51,5 @@ fun workManagerTag(sessionId: SessionId, requestType: WorkManagerRequestType): S
enum class WorkManagerRequestType {
NOTIFICATION_SYNC,
DB_VACUUM,
SEARCH_BACKFILL,
}