Rewrite *Manager classes to Kotlin for new room API

Seems to be unavailable in Java, rewrite and simplify it a bit as well Signed-off-by: Aayush Gupta <contact@aayush.io>
2026-10-27 17:54:34 +00:00 · 2026-10-22 11:21:47 +02:00
1 parent b6a7bfcf97
commit 9f6ff2fd37
@@ -13,6 +13,7 @@ import androidx.room3.Transaction
 import androidx.room3.rxjava3.RxDaoReturnTypeConverters
 import io.reactivex.rxjava3.core.Flowable
 import org.schabi.newpipe.database.BasicDAO
+import org.schabi.newpipe.database.playlist.PlaylistMetadataEntry
 import org.schabi.newpipe.database.playlist.model.PlaylistEntity

@Dao
@@ -48,4 +49,10 @@ interface PlaylistDAO : BasicDAO<PlaylistEntity> {
            return playlist.uid
        }
    }
+
+    @Transaction
+    fun updatePlaylists(updateItems: List<PlaylistMetadataEntry>, deletedItems: List<Long>) {
+        deletedItems.forEach { uid -> deletePlaylist(uid) }
+        updateItems.forEach { item -> upsertPlaylist(PlaylistEntity(item)) }
+    }
 }
@@ -136,4 +136,10 @@ interface PlaylistStreamDAO : BasicDAO<PlaylistStreamEntity> {
        """
    )
    fun getPlaylistDuplicatesMetadata(streamUrl: String): Flowable<MutableList<PlaylistDuplicatesEntry>>
+
+    @Transaction
+    fun updatePlaylistStreams(playlistId: Long, joinEntities: List<PlaylistStreamEntity>) {
+        deleteBatch(playlistId)
+        insertAll(joinEntities)
+    }
 }
@@ -65,7 +65,7 @@ public class StreamMiniInfoItemHolder extends InfoItemHolder {
            if (DependentPreferenceHelper
                    .getPositionsInListsEnabled(itemProgressView.getContext())) {
                state2 = historyRecordManager.loadStreamState(infoItem)
-                        .blockingGet()[0];
+                        .blockingGet();
            }
            if (state2 != null) {
                itemProgressView.setVisibility(View.VISIBLE);
@@ -120,7 +120,7 @@ public class StreamMiniInfoItemHolder extends InfoItemHolder {
        if (DependentPreferenceHelper.getPositionsInListsEnabled(itemProgressView.getContext())) {
            state = historyRecordManager
                    .loadStreamState(infoItem)
-                    .blockingGet()[0];
+                    .blockingGet();
        }
        if (state != null && item.getDuration() > 0
                && !StreamTypeUtil.isLiveStream(item.getStreamType())) {
@@ -1,318 +0,0 @@
-package org.schabi.newpipe.local.history;
-
-/*
- * Copyright (C) Mauricio Colli 2018
- * HistoryRecordManager.java is part of NewPipe.
- *
- * NewPipe is free software: you can redistribute it and/or modify
- * it under the terms of the GNU General Public License as published by
- * the Free Software Foundation, either version 3 of the License, or
- * (at your option) any later version.
- *
- * NewPipe is distributed in the hope that it will be useful,
- * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- * GNU General Public License for more details.
- *
- * You should have received a copy of the GNU General Public License
- * along with NewPipe.  If not, see <http://www.gnu.org/licenses/>.
- */
-
-import android.content.Context;
-import android.content.SharedPreferences;
-
-import androidx.annotation.NonNull;
-import androidx.preference.PreferenceManager;
-
-import org.schabi.newpipe.NewPipeDatabase;
-import org.schabi.newpipe.R;
-import org.schabi.newpipe.database.AppDatabase;
-import org.schabi.newpipe.database.LocalItem;
-import org.schabi.newpipe.database.history.dao.SearchHistoryDAO;
-import org.schabi.newpipe.database.history.dao.StreamHistoryDAO;
-import org.schabi.newpipe.database.history.model.SearchHistoryEntry;
-import org.schabi.newpipe.database.history.model.StreamHistoryEntity;
-import org.schabi.newpipe.database.history.model.StreamHistoryEntry;
-import org.schabi.newpipe.database.playlist.PlaylistStreamEntry;
-import org.schabi.newpipe.database.playlist.model.PlaylistStreamEntity;
-import org.schabi.newpipe.database.stream.StreamStatisticsEntry;
-import org.schabi.newpipe.database.stream.dao.StreamDAO;
-import org.schabi.newpipe.database.stream.dao.StreamStateDAO;
-import org.schabi.newpipe.database.stream.model.StreamEntity;
-import org.schabi.newpipe.database.stream.model.StreamStateEntity;
-import org.schabi.newpipe.extractor.InfoItem;
-import org.schabi.newpipe.extractor.stream.StreamInfo;
-import org.schabi.newpipe.extractor.stream.StreamInfoItem;
-import org.schabi.newpipe.local.feed.FeedViewModel;
-import org.schabi.newpipe.player.playqueue.PlayQueueItem;
-import org.schabi.newpipe.util.ExtractorHelper;
-
-import java.time.OffsetDateTime;
-import java.time.ZoneOffset;
-import java.util.ArrayList;
-import java.util.List;
-
-import io.reactivex.rxjava3.core.Completable;
-import io.reactivex.rxjava3.core.Flowable;
-import io.reactivex.rxjava3.core.Maybe;
-import io.reactivex.rxjava3.core.Single;
-import io.reactivex.rxjava3.schedulers.Schedulers;
-
-public class HistoryRecordManager {
-    private final AppDatabase database;
-    private final StreamDAO streamTable;
-    private final StreamHistoryDAO streamHistoryTable;
-    private final SearchHistoryDAO searchHistoryTable;
-    private final StreamStateDAO streamStateTable;
-    private final SharedPreferences sharedPreferences;
-    private final String searchHistoryKey;
-    private final String streamHistoryKey;
-
-    public HistoryRecordManager(final Context context) {
-        database = NewPipeDatabase.getInstance(context);
-        streamTable = database.streamDAO();
-        streamHistoryTable = database.streamHistoryDAO();
-        searchHistoryTable = database.searchHistoryDAO();
-        streamStateTable = database.streamStateDAO();
-        sharedPreferences = PreferenceManager.getDefaultSharedPreferences(context);
-        searchHistoryKey = context.getString(R.string.enable_search_history_key);
-        streamHistoryKey = context.getString(R.string.enable_watch_history_key);
-    }
-
-    ///////////////////////////////////////////////////////
-    // Watch History
-    ///////////////////////////////////////////////////////
-
-    /**
-     * Marks a stream item as watched such that it is hidden from the feed if watched videos are
-     * hidden. Adds a history entry and updates the stream progress to 100%.
-     *
-     * @see FeedViewModel#setSaveShowPlayedItems
-     * @param info the item to mark as watched
-     * @return a Maybe containing the ID of the item if successful
-     */
-    public Maybe<Long> markAsWatched(final StreamInfoItem info) {
-        if (!isStreamHistoryEnabled()) {
-            return Maybe.empty();
-        }
-
-        final OffsetDateTime currentTime = OffsetDateTime.now(ZoneOffset.UTC);
-        return Maybe.fromCallable(() -> database.runInTransaction(() -> {
-            final long streamId;
-            final long duration;
-            // Duration will not exist if the item was loaded with fast mode, so fetch it if empty
-            if (info.getDuration() < 0) {
-                final StreamInfo completeInfo = ExtractorHelper.getStreamInfo(
-                        info.getServiceId(),
-                        info.getUrl(),
-                        false
-                )
-                        .subscribeOn(Schedulers.io())
-                        .blockingGet();
-                duration = completeInfo.getDuration();
-                streamId = streamTable.upsert(new StreamEntity(completeInfo));
-            } else {
-                duration = info.getDuration();
-                streamId = streamTable.upsert(new StreamEntity(info));
-            }
-
-            // Update the stream progress to the full duration of the video
-            final StreamStateEntity entity = new StreamStateEntity(
-                    streamId,
-                    duration * 1000
-            );
-            streamStateTable.upsert(entity);
-
-            // Add a history entry
-            final StreamHistoryEntity latestEntry = streamHistoryTable.getLatestEntry(streamId);
-            if (latestEntry == null) {
-                // never actually viewed: add history entry but with 0 views
-                return streamHistoryTable.insert(new StreamHistoryEntity(streamId, currentTime, 0));
-            } else {
-                return 0L;
-            }
-        })).subscribeOn(Schedulers.io());
-    }
-
-    public Maybe<Long> onViewed(final StreamInfo info) {
-        if (!isStreamHistoryEnabled()) {
-            return Maybe.empty();
-        }
-
-        final OffsetDateTime currentTime = OffsetDateTime.now(ZoneOffset.UTC);
-        return Maybe.fromCallable(() -> database.runInTransaction(() -> {
-            final long streamId = streamTable.upsert(new StreamEntity(info));
-            final StreamHistoryEntity latestEntry = streamHistoryTable.getLatestEntry(streamId);
-
-            if (latestEntry != null) {
-                streamHistoryTable.delete(latestEntry);
-                latestEntry.setAccessDate(currentTime);
-                latestEntry.setRepeatCount(latestEntry.getRepeatCount() + 1);
-                return streamHistoryTable.insert(latestEntry);
-            } else {
-                // just viewed for the first time: set 1 view
-                return streamHistoryTable.insert(new StreamHistoryEntity(streamId, currentTime, 1));
-            }
-        })).subscribeOn(Schedulers.io());
-    }
-
-    public Completable deleteStreamHistoryAndState(final long streamId) {
-        return Completable.fromAction(() -> {
-            streamStateTable.deleteState(streamId);
-            streamHistoryTable.deleteStreamHistory(streamId);
-        }).subscribeOn(Schedulers.io());
-    }
-
-    public Single<Integer> deleteWholeStreamHistory() {
-        return Single.fromCallable(streamHistoryTable::deleteAll)
-                .subscribeOn(Schedulers.io());
-    }
-
-    public Single<Integer> deleteCompleteStreamStateHistory() {
-        return Single.fromCallable(streamStateTable::deleteAll)
-                .subscribeOn(Schedulers.io());
-    }
-
-    public Flowable<List<StreamHistoryEntry>> getStreamHistorySortedById() {
-        return streamHistoryTable.getHistorySortedById().subscribeOn(Schedulers.io());
-    }
-
-    public Flowable<List<StreamStatisticsEntry>> getStreamStatistics() {
-        return streamHistoryTable.getStatistics().subscribeOn(Schedulers.io());
-    }
-
-    private boolean isStreamHistoryEnabled() {
-        return sharedPreferences.getBoolean(streamHistoryKey, false);
-    }
-
-    ///////////////////////////////////////////////////////
-    // Search History
-    ///////////////////////////////////////////////////////
-
-    public Maybe<Long> onSearched(final int serviceId, final String search) {
-        if (!isSearchHistoryEnabled()) {
-            return Maybe.empty();
-        }
-
-        final OffsetDateTime currentTime = OffsetDateTime.now(ZoneOffset.UTC);
-        final SearchHistoryEntry newEntry = new SearchHistoryEntry(currentTime, serviceId, search);
-
-        return Maybe.fromCallable(() -> database.runInTransaction(() -> {
-            final SearchHistoryEntry latestEntry = searchHistoryTable.getLatestEntry();
-            if (latestEntry != null && latestEntry.hasEqualValues(newEntry)) {
-                latestEntry.setCreationDate(currentTime);
-                return (long) searchHistoryTable.update(latestEntry);
-            } else {
-                return searchHistoryTable.insert(newEntry);
-            }
-        })).subscribeOn(Schedulers.io());
-    }
-
-    public Single<Integer> deleteSearchHistory(final String search) {
-        return Single.fromCallable(() -> searchHistoryTable.deleteAllWhereQuery(search))
-                .subscribeOn(Schedulers.io());
-    }
-
-    public Single<Integer> deleteCompleteSearchHistory() {
-        return Single.fromCallable(searchHistoryTable::deleteAll)
-                .subscribeOn(Schedulers.io());
-    }
-
-    public Flowable<List<String>> getRelatedSearches(final String query,
-                                                     final int similarQueryLimit,
-                                                     final int uniqueQueryLimit) {
-        return query.length() > 0
-                ? searchHistoryTable.getSimilarEntries(query, similarQueryLimit)
-                : searchHistoryTable.getUniqueEntries(uniqueQueryLimit);
-    }
-
-    private boolean isSearchHistoryEnabled() {
-        return sharedPreferences.getBoolean(searchHistoryKey, false);
-    }
-
-    ///////////////////////////////////////////////////////
-    // Stream State History
-    ///////////////////////////////////////////////////////
-
-    public Maybe<StreamStateEntity> loadStreamState(final PlayQueueItem queueItem) {
-        return queueItem.getStream()
-                .map(info -> streamTable.upsert(new StreamEntity(info)))
-                .flatMapPublisher(streamStateTable::getState)
-                .firstElement()
-                .flatMap(list -> list.isEmpty() ? Maybe.empty() : Maybe.just(list.get(0)))
-                .filter(state -> state.isValid(queueItem.getDuration()))
-                .subscribeOn(Schedulers.io());
-    }
-
-    public Maybe<StreamStateEntity> loadStreamState(final StreamInfo info) {
-        return Single.fromCallable(() -> streamTable.upsert(new StreamEntity(info)))
-                .flatMapPublisher(streamStateTable::getState)
-                .firstElement()
-                .flatMap(list -> list.isEmpty() ? Maybe.empty() : Maybe.just(list.get(0)))
-                .filter(state -> state.isValid(info.getDuration()))
-                .subscribeOn(Schedulers.io());
-    }
-
-    public Completable saveStreamState(@NonNull final StreamInfo info, final long progressMillis) {
-        return Completable.fromAction(() -> database.runInTransaction(() -> {
-            final long streamId = streamTable.upsert(new StreamEntity(info));
-            final StreamStateEntity state = new StreamStateEntity(streamId, progressMillis);
-            if (state.isValid(info.getDuration())) {
-                streamStateTable.upsert(state);
-            }
-        })).subscribeOn(Schedulers.io());
-    }
-
-    public Single<StreamStateEntity[]> loadStreamState(final InfoItem info) {
-        return Single.fromCallable(() -> {
-            final List<StreamEntity> entities = streamTable
-                    .getStream(info.getServiceId(), info.getUrl()).blockingFirst();
-            if (entities.isEmpty()) {
-                return new StreamStateEntity[]{null};
-            }
-            final List<StreamStateEntity> states = streamStateTable
-                    .getState(entities.get(0).getUid()).blockingFirst();
-            if (states.isEmpty()) {
-                return new StreamStateEntity[]{null};
-            }
-            return new StreamStateEntity[]{states.get(0)};
-        }).subscribeOn(Schedulers.io());
-    }
-
-    public Single<List<StreamStateEntity>> loadLocalStreamStateBatch(
-            final List<? extends LocalItem> items) {
-        return Single.fromCallable(() -> {
-            final List<StreamStateEntity> result = new ArrayList<>(items.size());
-            for (final LocalItem item : items) {
-                final long streamId;
-                if (item instanceof StreamStatisticsEntry) {
-                    streamId = ((StreamStatisticsEntry) item).getStreamId();
-                } else if (item instanceof PlaylistStreamEntity) {
-                    streamId = ((PlaylistStreamEntity) item).getStreamUid();
-                } else if (item instanceof PlaylistStreamEntry) {
-                    streamId = ((PlaylistStreamEntry) item).getStreamId();
-                } else {
-                    result.add(null);
-                    continue;
-                }
-                final List<StreamStateEntity> states = streamStateTable.getState(streamId)
-                        .blockingFirst();
-                if (states.isEmpty()) {
-                    result.add(null);
-                } else {
-                    result.add(states.get(0));
-                }
-            }
-            return result;
-        }).subscribeOn(Schedulers.io());
-    }
-
-    ///////////////////////////////////////////////////////
-    // Utility
-    ///////////////////////////////////////////////////////
-
-    public Single<Integer> removeOrphanedRecords() {
-        return Single.fromCallable(streamTable::deleteOrphans).subscribeOn(Schedulers.io());
-    }
-
-}
@@ -0,0 +1,292 @@
+/*
+ * SPDX-FileCopyrightText: 2018-2023 NewPipe contributors <https://newpipe.net>
+ * SPDX-FileCopyrightText: 2026 NewPipe e.V. <https://newpipe-ev.de>
+ * SPDX-License-Identifier: GPL-3.0-or-later
+ */
+
+package org.schabi.newpipe.local.history
+
+import android.content.Context
+import androidx.preference.PreferenceManager
+import androidx.room3.withWriteTransaction
+import io.reactivex.rxjava3.core.Completable
+import io.reactivex.rxjava3.core.Flowable
+import io.reactivex.rxjava3.core.Maybe
+import io.reactivex.rxjava3.core.Single
+import io.reactivex.rxjava3.schedulers.Schedulers
+import java.time.OffsetDateTime
+import java.time.ZoneOffset
+import kotlinx.coroutines.runBlocking
+import org.schabi.newpipe.NewPipeDatabase
+import org.schabi.newpipe.R
+import org.schabi.newpipe.database.LocalItem
+import org.schabi.newpipe.database.history.model.SearchHistoryEntry
+import org.schabi.newpipe.database.history.model.StreamHistoryEntity
+import org.schabi.newpipe.database.history.model.StreamHistoryEntry
+import org.schabi.newpipe.database.playlist.PlaylistStreamEntry
+import org.schabi.newpipe.database.playlist.model.PlaylistStreamEntity
+import org.schabi.newpipe.database.stream.StreamStatisticsEntry
+import org.schabi.newpipe.database.stream.model.StreamEntity
+import org.schabi.newpipe.database.stream.model.StreamStateEntity
+import org.schabi.newpipe.extractor.InfoItem
+import org.schabi.newpipe.extractor.stream.StreamInfo
+import org.schabi.newpipe.extractor.stream.StreamInfoItem
+import org.schabi.newpipe.local.feed.FeedViewModel
+import org.schabi.newpipe.player.playqueue.PlayQueueItem
+import org.schabi.newpipe.util.ExtractorHelper
+
+class HistoryRecordManager(context: Context) {
+
+    private val database = NewPipeDatabase.getInstance(context)
+    private val streamTable = database.streamDAO()
+    private val streamHistoryTable = database.streamHistoryDAO()
+    private val searchHistoryTable = database.searchHistoryDAO()
+    private val streamStateTable = database.streamStateDAO()
+
+    private val sharedPreferences = PreferenceManager.getDefaultSharedPreferences(context)
+    private val searchHistoryKey = context.getString(R.string.enable_search_history_key)
+    private val streamHistoryKey = context.getString(R.string.enable_watch_history_key)
+
+    private val isSearchHistoryEnabled: Boolean
+        get() = sharedPreferences.getBoolean(searchHistoryKey, false)
+
+    private val isStreamHistoryEnabled: Boolean
+        get() = sharedPreferences.getBoolean(streamHistoryKey, false)
+
+    /////////////////////////////////////////////////////////
+    // Watch History
+    /////////////////////////////////////////////////////////
+
+    /**
+     * Marks a stream item as watched such that it is hidden from the feed if watched videos are
+     * hidden. Adds a history entry and updates the stream progress to 100%.
+     *
+     * @see FeedViewModel.setSaveShowPlayedItems
+     * @param info the item to mark as watched
+     * @return a Maybe containing the ID of the item if successful
+     */
+    fun markAsWatched(info: StreamInfoItem): Maybe<Long> {
+        if (!isStreamHistoryEnabled) {
+            return Maybe.empty()
+        }
+
+        val currentTime = OffsetDateTime.now(ZoneOffset.UTC)
+
+        return Maybe.fromCallable {
+            runBlocking {
+                database.withWriteTransaction {
+                    val streamId: Long
+                    val duration: Long
+
+                    // Duration will not exist if the item was loaded with fast mode, so fetch it
+                    // if empty
+                    if (info.duration < 0) {
+                        val completeInfo = ExtractorHelper.getStreamInfo(
+                            info.serviceId,
+                            info.url,
+                            false
+                        ).subscribeOn(Schedulers.io()).blockingGet()
+
+                        duration = completeInfo.duration
+                        streamId = streamTable.upsert(StreamEntity(completeInfo))
+                    } else {
+                        duration = info.duration
+                        streamId = streamTable.upsert(StreamEntity(info))
+                    }
+
+                    // Update the stream progress to the full duration of the video
+                    val entity = StreamStateEntity(
+                        streamId,
+                        duration * 1000
+                    )
+                    streamStateTable.upsert(entity)
+
+                    // Add a history entry
+                    val latestEntry = streamHistoryTable.getLatestEntry(streamId)
+                    if (latestEntry == null) {
+                        // never actually viewed: add history entry but with 0 views
+                        streamHistoryTable.insert(StreamHistoryEntity(streamId, currentTime, 0))
+                    } else {
+                        0L
+                    }
+                }
+            }
+        }.subscribeOn(Schedulers.io())
+    }
+
+    fun onViewed(info: StreamInfo): Maybe<Long> {
+        if (!isStreamHistoryEnabled) {
+            return Maybe.empty()
+        }
+
+        val currentTime = OffsetDateTime.now(ZoneOffset.UTC)
+        return Maybe.fromCallable {
+            runBlocking {
+                database.withWriteTransaction {
+                    val streamId = streamTable.upsert(StreamEntity(info))
+                    val latestEntry = streamHistoryTable.getLatestEntry(streamId)
+
+                    if (latestEntry != null) {
+                        streamHistoryTable.delete(latestEntry)
+                        latestEntry.accessDate = currentTime
+                        latestEntry.repeatCount = latestEntry.repeatCount + 1
+                        streamHistoryTable.insert(latestEntry)
+                    } else {
+                        // just viewed for the first time: set 1 view
+                        streamHistoryTable.insert(StreamHistoryEntity(streamId, currentTime, 1))
+                    }
+                }
+            }
+        }.subscribeOn(Schedulers.io())
+    }
+
+    fun deleteStreamHistoryAndState(streamId: Long): Completable {
+        return Completable.fromAction {
+            streamStateTable.deleteState(streamId)
+            streamHistoryTable.deleteStreamHistory(streamId)
+        }.subscribeOn(Schedulers.io())
+    }
+
+    fun deleteWholeStreamHistory(): Single<Int> {
+        return Single.fromCallable { streamHistoryTable.deleteAll() }
+            .subscribeOn(Schedulers.io())
+    }
+
+    fun deleteCompleteStreamStateHistory(): Single<Int> {
+        return Single.fromCallable { streamStateTable.deleteAll() }
+            .subscribeOn(Schedulers.io())
+    }
+
+    fun getStreamHistorySortedById(): Flowable<MutableList<StreamHistoryEntry>> {
+        return streamHistoryTable.historySortedById.subscribeOn(Schedulers.io())
+    }
+
+    fun getStreamStatistics(): Flowable<MutableList<StreamStatisticsEntry>> {
+        return streamHistoryTable.getStatistics().subscribeOn(Schedulers.io())
+    }
+
+    /////////////////////////////////////////////////////////
+    // Search History
+    /////////////////////////////////////////////////////////
+
+    fun onSearched(serviceId: Int, search: String): Maybe<Long> {
+        if (!isSearchHistoryEnabled) {
+            return Maybe.empty()
+        }
+
+        val currentTime = OffsetDateTime.now(ZoneOffset.UTC)
+        val newEntry = SearchHistoryEntry(currentTime, serviceId, search)
+
+        return Maybe.fromCallable {
+            runBlocking {
+                database.withWriteTransaction {
+                    val latestEntry = searchHistoryTable.latestEntry
+                    if (latestEntry != null && latestEntry.hasEqualValues(newEntry)) {
+                        latestEntry.creationDate = currentTime
+                        searchHistoryTable.update(latestEntry).toLong()
+                    } else {
+                        searchHistoryTable.insert(newEntry)
+                    }
+                }
+            }
+        }.subscribeOn(Schedulers.io())
+    }
+
+    fun deleteSearchHistory(search: String): Single<Int> {
+        return Single.fromCallable { searchHistoryTable.deleteAllWhereQuery(search) }
+            .subscribeOn(Schedulers.io())
+    }
+
+    fun deleteCompleteSearchHistory(): Single<Int> {
+        return Single.fromCallable { searchHistoryTable.deleteAll() }
+            .subscribeOn(Schedulers.io())
+    }
+
+    fun getRelatedSearches(
+        query: String,
+        similarQueryLimit: Int,
+        uniqueQueryLimit: Int
+    ): Flowable<MutableList<String>> {
+        return if (query.isNotEmpty()) {
+            searchHistoryTable.getSimilarEntries(query, similarQueryLimit)
+        } else {
+            searchHistoryTable.getUniqueEntries(uniqueQueryLimit)
+        }
+    }
+
+    /////////////////////////////////////////////////////////
+    // Stream State History
+    /////////////////////////////////////////////////////////
+
+    fun loadStreamState(queueItem: PlayQueueItem): Maybe<StreamStateEntity> {
+        return queueItem.stream
+            .map { info -> streamTable.upsert(StreamEntity(info)) }
+            .flatMapPublisher { streamId -> streamStateTable.getState(streamId) }
+            .firstElement()
+            .flatMap { list -> if (list.isEmpty()) Maybe.empty() else Maybe.just(list[0]) }
+            .filter { state -> state.isValid(queueItem.duration) }
+            .subscribeOn(Schedulers.io())
+    }
+
+    fun loadStreamState(info: StreamInfo): Maybe<StreamStateEntity> {
+        return Single.fromCallable { streamTable.upsert(StreamEntity(info)) }
+            .flatMapPublisher { streamId -> streamStateTable.getState(streamId) }
+            .firstElement()
+            .flatMap { list -> if (list.isEmpty()) Maybe.empty() else Maybe.just(list[0]) }
+            .filter { state -> state.isValid(info.duration) }
+            .subscribeOn(Schedulers.io())
+    }
+
+    fun saveStreamState(info: StreamInfo, progressMillis: Long): Completable {
+        return Completable.fromAction {
+            runBlocking {
+                database.withWriteTransaction {
+                    val streamId = streamTable.upsert(StreamEntity(info))
+                    val state = StreamStateEntity(streamId, progressMillis)
+                    if (state.isValid(info.duration)) {
+                        streamStateTable.upsert(state)
+                    }
+                }
+            }
+        }.subscribeOn(Schedulers.io())
+    }
+
+    fun loadStreamState(info: InfoItem): Maybe<StreamStateEntity> {
+        return Maybe.fromCallable {
+            streamTable.getStream(info.serviceId.toLong(), info.url)
+                .blockingFirst()
+                .firstOrNull()
+                ?.let { entity ->
+                    streamStateTable.getState(entity.uid)
+                        .blockingFirst()
+                        .firstOrNull()
+                }
+        }.subscribeOn(Schedulers.io())
+    }
+
+    fun loadLocalStreamStateBatch(items: List<LocalItem>): Single<List<StreamStateEntity?>> {
+        return Single.fromCallable {
+            items.map { item ->
+                val streamId = when (item) {
+                    is StreamStatisticsEntry -> item.streamId
+                    is PlaylistStreamEntity -> item.streamUid
+                    is PlaylistStreamEntry -> item.streamId
+                    else -> return@map null
+                }
+
+                streamStateTable.getState(streamId)
+                    .blockingFirst()
+                    .firstOrNull()
+            }
+        }.subscribeOn(Schedulers.io())
+    }
+
+    /////////////////////////////////////////////////////////
+    // Utility
+    /////////////////////////////////////////////////////////
+
+    fun removeOrphanedRecords(): Single<Int> {
+        return Single.fromCallable { streamTable.deleteOrphans() }
+            .subscribeOn(Schedulers.io())
+    }
+}
@@ -1,189 +0,0 @@
-package org.schabi.newpipe.local.playlist;
-
-import androidx.annotation.Nullable;
-
-import org.schabi.newpipe.database.AppDatabase;
-import org.schabi.newpipe.database.playlist.PlaylistDuplicatesEntry;
-import org.schabi.newpipe.database.playlist.PlaylistMetadataEntry;
-import org.schabi.newpipe.database.playlist.PlaylistStreamEntry;
-import org.schabi.newpipe.database.playlist.dao.PlaylistDAO;
-import org.schabi.newpipe.database.playlist.dao.PlaylistStreamDAO;
-import org.schabi.newpipe.database.playlist.model.PlaylistEntity;
-import org.schabi.newpipe.database.playlist.model.PlaylistStreamEntity;
-import org.schabi.newpipe.database.stream.dao.StreamDAO;
-import org.schabi.newpipe.database.stream.model.StreamEntity;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import io.reactivex.rxjava3.core.Completable;
-import io.reactivex.rxjava3.core.Flowable;
-import io.reactivex.rxjava3.core.Maybe;
-import io.reactivex.rxjava3.schedulers.Schedulers;
-
-public class LocalPlaylistManager {
-    private static final long THUMBNAIL_ID_LEAVE_UNCHANGED = -2;
-
-    private final AppDatabase database;
-    private final StreamDAO streamTable;
-    private final PlaylistDAO playlistTable;
-    private final PlaylistStreamDAO playlistStreamTable;
-
-    public LocalPlaylistManager(final AppDatabase db) {
-        database = db;
-        streamTable = db.streamDAO();
-        playlistTable = db.playlistDAO();
-        playlistStreamTable = db.playlistStreamDAO();
-    }
-
-    public Maybe<List<Long>> createPlaylist(final String name, final List<StreamEntity> streams) {
-        // Disallow creation of empty playlists
-        if (streams.isEmpty()) {
-            return Maybe.empty();
-        }
-
-        // Save to the database directly.
-        // Make sure the new playlist is always on the top of bookmark.
-        // The index will be reassigned to non-negative number in BookmarkFragment.
-        return Maybe.fromCallable(() -> database.runInTransaction(() -> {
-                    final List<Long> streamIds = streamTable.upsertAll(streams);
-                    final PlaylistEntity newPlaylist = new PlaylistEntity(name, false,
-                            streamIds.get(0), -1);
-
-                    return insertJoinEntities(playlistTable.insert(newPlaylist),
-                            streamIds, 0);
-                }
-        )).subscribeOn(Schedulers.io());
-    }
-
-    public Maybe<List<Long>> appendToPlaylist(final long playlistId,
-                                              final List<StreamEntity> streams) {
-        return playlistStreamTable.getMaximumIndexOf(playlistId)
-                .firstElement()
-                .map(maxJoinIndex -> database.runInTransaction(() -> {
-                            final List<Long> streamIds = streamTable.upsertAll(streams);
-                            return insertJoinEntities(playlistId, streamIds, maxJoinIndex + 1);
-                        }
-                )).subscribeOn(Schedulers.io());
-    }
-
-    private List<Long> insertJoinEntities(final long playlistId, final List<Long> streamIds,
-                                          final int indexOffset) {
-
-        final List<PlaylistStreamEntity> joinEntities = new ArrayList<>(streamIds.size());
-
-        for (int index = 0; index < streamIds.size(); index++) {
-            joinEntities.add(new PlaylistStreamEntity(playlistId, streamIds.get(index),
-                    index + indexOffset));
-        }
-        return playlistStreamTable.insertAll(joinEntities);
-    }
-
-    public Completable updateJoin(final long playlistId, final List<Long> streamIds) {
-        final List<PlaylistStreamEntity> joinEntities = new ArrayList<>(streamIds.size());
-        for (int i = 0; i < streamIds.size(); i++) {
-            joinEntities.add(new PlaylistStreamEntity(playlistId, streamIds.get(i), i));
-        }
-
-        return Completable.fromRunnable(() -> database.runInTransaction(() -> {
-            playlistStreamTable.deleteBatch(playlistId);
-            playlistStreamTable.insertAll(joinEntities);
-        })).subscribeOn(Schedulers.io());
-    }
-
-    public Completable updatePlaylists(final List<PlaylistMetadataEntry> updateItems,
-                                       final List<Long> deletedItems) {
-        final List<PlaylistEntity> items = new ArrayList<>(updateItems.size());
-        for (final PlaylistMetadataEntry item : updateItems) {
-            items.add(new PlaylistEntity(item));
-        }
-        return Completable.fromRunnable(() -> database.runInTransaction(() -> {
-            for (final Long uid : deletedItems) {
-                playlistTable.deletePlaylist(uid);
-            }
-            for (final PlaylistEntity item : items) {
-                playlistTable.upsertPlaylist(item);
-            }
-        })).subscribeOn(Schedulers.io());
-    }
-
-    public Flowable<List<PlaylistStreamEntry>> getDistinctPlaylistStreams(final long playlistId) {
-        return playlistStreamTable
-                .getStreamsWithoutDuplicates(playlistId).subscribeOn(Schedulers.io());
-    }
-
-    /**
-     * Get playlists with attached information about how many times the provided stream is already
-     * contained in each playlist.
-     *
-     * @param streamUrl the stream url for which to check for duplicates
-     * @return a list of {@link PlaylistDuplicatesEntry}
-     */
-    public Flowable<List<PlaylistDuplicatesEntry>> getPlaylistDuplicates(final String streamUrl) {
-        return playlistStreamTable.getPlaylistDuplicatesMetadata(streamUrl)
-                .subscribeOn(Schedulers.io());
-    }
-
-    public Flowable<List<PlaylistMetadataEntry>> getPlaylists() {
-        return playlistStreamTable.getPlaylistMetadata().subscribeOn(Schedulers.io());
-    }
-
-    public Flowable<List<PlaylistStreamEntry>> getPlaylistStreams(final long playlistId) {
-        return playlistStreamTable.getOrderedStreamsOf(playlistId).subscribeOn(Schedulers.io());
-    }
-
-    public Maybe<Integer> renamePlaylist(final long playlistId, final String name) {
-        return modifyPlaylist(playlistId, name, THUMBNAIL_ID_LEAVE_UNCHANGED, false);
-    }
-
-    public Maybe<Integer> changePlaylistThumbnail(final long playlistId,
-                                                  final long thumbnailStreamId,
-                                                  final boolean isPermanent) {
-        return modifyPlaylist(playlistId, null, thumbnailStreamId, isPermanent);
-    }
-
-    public long getPlaylistThumbnailStreamId(final long playlistId) {
-        return playlistTable.getPlaylist(playlistId).blockingFirst().get(0).getThumbnailStreamId();
-    }
-
-    public boolean getIsPlaylistThumbnailPermanent(final long playlistId) {
-        return playlistTable.getPlaylist(playlistId).blockingFirst().get(0)
-                .isThumbnailPermanent();
-    }
-
-    public long getAutomaticPlaylistThumbnailStreamId(final long playlistId) {
-        final long streamId = playlistStreamTable.getAutomaticThumbnailStreamId(playlistId)
-                .blockingFirst();
-        if (streamId < 0) {
-            return PlaylistEntity.DEFAULT_THUMBNAIL_ID;
-        }
-        return streamId;
-    }
-
-    private Maybe<Integer> modifyPlaylist(final long playlistId,
-                                          @Nullable final String name,
-                                          final long thumbnailStreamId,
-                                          final boolean isPermanent) {
-        return playlistTable.getPlaylist(playlistId)
-                .firstElement()
-                .filter(playlistEntities -> !playlistEntities.isEmpty())
-                .map(playlistEntities -> {
-                    final PlaylistEntity playlist = playlistEntities.get(0);
-                    if (name != null) {
-                        playlist.setName(name);
-                    }
-                    if (thumbnailStreamId != THUMBNAIL_ID_LEAVE_UNCHANGED) {
-                        playlist.setThumbnailStreamId(thumbnailStreamId);
-                        playlist.setThumbnailPermanent(isPermanent);
-                    }
-                    return playlistTable.update(playlist);
-                }).subscribeOn(Schedulers.io());
-    }
-
-    public Maybe<Boolean> hasPlaylists() {
-        return playlistTable.getCount()
-                .firstElement()
-                .map(count -> count > 0)
-                .subscribeOn(Schedulers.io());
-    }
-}
@@ -0,0 +1,181 @@
+/*
+ * SPDX-FileCopyrightText: 2018-2024 NewPipe contributors <https://newpipe.net>
+ * SPDX-FileCopyrightText: 2025-2026 NewPipe e.V. <https://newpipe-ev.de>
+ * SPDX-License-Identifier: GPL-3.0-or-later
+ */
+
+package org.schabi.newpipe.local.playlist
+
+import androidx.room3.withWriteTransaction
+import io.reactivex.rxjava3.core.Completable
+import io.reactivex.rxjava3.core.Flowable
+import io.reactivex.rxjava3.core.Maybe
+import io.reactivex.rxjava3.schedulers.Schedulers
+import kotlinx.coroutines.runBlocking
+import org.schabi.newpipe.database.AppDatabase
+import org.schabi.newpipe.database.playlist.PlaylistDuplicatesEntry
+import org.schabi.newpipe.database.playlist.PlaylistMetadataEntry
+import org.schabi.newpipe.database.playlist.PlaylistStreamEntry
+import org.schabi.newpipe.database.playlist.model.PlaylistEntity
+import org.schabi.newpipe.database.playlist.model.PlaylistStreamEntity
+import org.schabi.newpipe.database.stream.model.StreamEntity
+
+class LocalPlaylistManager(private val database: AppDatabase) {
+
+    private val streamTable = database.streamDAO()
+    private val playlistTable = database.playlistDAO()
+    private val playlistStreamTable = database.playlistStreamDAO()
+
+    fun createPlaylist(name: String, streams: List<StreamEntity>): Maybe<List<Long>> {
+        // Disallow creation of empty playlists
+        if (streams.isEmpty()) {
+            return Maybe.empty()
+        }
+
+        // Save to the database directly.
+        // Make sure the new playlist is always on the top of bookmark.
+        // The index will be reassigned to non-negative number in BookmarkFragment.
+        return Maybe.fromCallable {
+            runBlocking {
+                database.withWriteTransaction {
+                    val streamIds = streamTable.upsertAll(streams)
+                    val newPlaylist = PlaylistEntity(
+                        name = name,
+                        isThumbnailPermanent = false,
+                        thumbnailStreamId = streamIds[0],
+                        displayIndex = -1
+                    )
+                    insertJoinEntities(playlistTable.insert(newPlaylist), streamIds, 0)
+                }
+            }
+        }.subscribeOn(Schedulers.io())
+    }
+
+    fun appendToPlaylist(playlistId: Long, streams: List<StreamEntity>): Maybe<List<Long>> {
+        return playlistStreamTable.getMaximumIndexOf(playlistId)
+            .firstElement()
+            .map { maxJoinIndex ->
+                runBlocking {
+                    database.withWriteTransaction<List<Long>> {
+                        val streamIds = streamTable.upsertAll(streams)
+                        insertJoinEntities(playlistId, streamIds, maxJoinIndex + 1)
+                    }
+                }
+            }.subscribeOn(Schedulers.io())
+    }
+
+    private fun insertJoinEntities(
+        playlistId: Long,
+        streamIds: List<Long>,
+        indexOffset: Int
+    ): List<Long> {
+        val entities = streamIds.mapIndexed { index, streamId ->
+            PlaylistStreamEntity(playlistId, streamId, index + indexOffset)
+        }
+        return playlistStreamTable.insertAll(entities)
+    }
+
+    fun updateJoin(playlistId: Long, streamIds: List<Long>): Completable {
+        val joinEntities = streamIds.mapIndexed { index, streamId ->
+            PlaylistStreamEntity(playlistId, streamId, index)
+        }
+
+        return Completable.fromAction {
+            playlistStreamTable.updatePlaylistStreams(playlistId, joinEntities)
+        }.subscribeOn(Schedulers.io())
+    }
+
+    fun updatePlaylists(
+        updateItems: List<PlaylistMetadataEntry>,
+        deletedItems: List<Long>
+    ): Completable {
+        return Completable.fromRunnable {
+            playlistTable.updatePlaylists(updateItems = updateItems, deletedItems = deletedItems)
+        }.subscribeOn(Schedulers.io())
+    }
+
+    fun getDistinctPlaylistStreams(playlistId: Long): Flowable<MutableList<PlaylistStreamEntry>> {
+        return playlistStreamTable
+            .getStreamsWithoutDuplicates(playlistId)
+            .subscribeOn(Schedulers.io())
+    }
+
+    /**
+     * Get playlists with attached information about how many times the provided stream is already
+     * contained in each playlist.
+     *
+     * @param streamUrl the stream url for which to check for duplicates
+     * @return a list of [PlaylistDuplicatesEntry]
+     */
+    fun getPlaylistDuplicates(streamUrl: String): Flowable<MutableList<PlaylistDuplicatesEntry>> {
+        return playlistStreamTable.getPlaylistDuplicatesMetadata(streamUrl)
+            .subscribeOn(Schedulers.io())
+    }
+
+    fun getPlaylists(): Flowable<MutableList<PlaylistMetadataEntry>> {
+        return playlistStreamTable.getPlaylistMetadata().subscribeOn(Schedulers.io())
+    }
+
+    fun getPlaylistStreams(playlistId: Long): Flowable<MutableList<PlaylistStreamEntry>> {
+        return playlistStreamTable.getOrderedStreamsOf(playlistId).subscribeOn(Schedulers.io())
+    }
+
+    fun renamePlaylist(playlistId: Long, name: String): Maybe<Int> {
+        return modifyPlaylist(playlistId, name, THUMBNAIL_ID_LEAVE_UNCHANGED, false)
+    }
+
+    fun changePlaylistThumbnail(
+        playlistId: Long,
+        thumbnailStreamId: Long,
+        isPermanent: Boolean
+    ): Maybe<Int> {
+        return modifyPlaylist(playlistId, null, thumbnailStreamId, isPermanent)
+    }
+
+    fun getPlaylistThumbnailStreamId(playlistId: Long): Long {
+        return playlistTable.getPlaylist(playlistId).blockingFirst()[0].thumbnailStreamId
+    }
+
+    fun getIsPlaylistThumbnailPermanent(playlistId: Long): Boolean {
+        return playlistTable.getPlaylist(playlistId).blockingFirst()[0].isThumbnailPermanent
+    }
+
+    fun getAutomaticPlaylistThumbnailStreamId(playlistId: Long): Long {
+        val streamId = playlistStreamTable.getAutomaticThumbnailStreamId(playlistId)
+            .blockingFirst()
+        return if (streamId < 0) PlaylistEntity.DEFAULT_THUMBNAIL_ID else streamId
+    }
+
+    private fun modifyPlaylist(
+        playlistId: Long,
+        name: String?,
+        thumbnailStreamId: Long,
+        isPermanent: Boolean
+    ): Maybe<Int> {
+        return playlistTable.getPlaylist(playlistId)
+            .firstElement()
+            .filter { playlistEntities -> playlistEntities.isNotEmpty() }
+            .map { playlistEntities ->
+                val playlist = playlistEntities[0]
+                if (name != null) {
+                    playlist.name = name
+                }
+                if (thumbnailStreamId != THUMBNAIL_ID_LEAVE_UNCHANGED) {
+                    playlist.thumbnailStreamId = thumbnailStreamId
+                    playlist.isThumbnailPermanent = isPermanent
+                }
+                playlistTable.update(playlist)
+            }.subscribeOn(Schedulers.io())
+    }
+
+    fun hasPlaylists(): Maybe<Boolean> {
+        return playlistTable.count
+            .firstElement()
+            .map { count -> count > 0 }
+            .subscribeOn(Schedulers.io())
+    }
+
+    companion object {
+        private const val THUMBNAIL_ID_LEAVE_UNCHANGED = -2L
+    }
+}
@@ -43,7 +43,7 @@ class LocalPlaylistViewModel(application: Application) : AndroidViewModel(applic
    fun removeWatchedStreams(playlistId: Long, removePartiallyWatched: Boolean) {
        _workState.value = WorkState.Loading

-        val historyIdsMaybe = recordManager.streamHistorySortedById
+        val historyIdsMaybe = recordManager.getStreamHistorySortedById()
            .firstElement()
            .map { historyList ->
                historyList.map { it.streamId }