Migrate watch history to Jetpack Compose

app/build.gradle

@@ -232,6 +232,7 @@ dependencies {
     implementation libs.androidx.recyclerview
     implementation libs.androidx.room.runtime
     implementation libs.androidx.room.rxjava3
+    implementation libs.androidx.room.paging
     kapt libs.androidx.room.compiler
     implementation libs.androidx.swiperefreshlayout
     implementation libs.androidx.work.runtime

app/src/androidTest/java/org/schabi/newpipe/database/DatabaseMigrationTest.kt

@@ -129,7 +129,7 @@ class DatabaseMigrationTest {
         )
 
         val migratedDatabaseV3 = getMigratedDatabase()
-        val listFromDB = migratedDatabaseV3.streamDAO().all.blockingFirst()
+        val listFromDB = migratedDatabaseV3.streamDAO().getAll().blockingFirst()
 
         // Only expect 2, the one with the null url will be ignored
         assertEquals(2, listFromDB.size)

app/src/androidTest/java/org/schabi/newpipe/database/FeedDAOTest.kt

@@ -3,7 +3,9 @@ package org.schabi.newpipe.database
 import android.content.Context
 import androidx.room.Room
 import androidx.test.core.app.ApplicationProvider
-import io.reactivex.rxjava3.core.Single
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.rx3.await
 import org.junit.After
 import org.junit.Assert.assertEquals
 import org.junit.Assert.assertNotNull
@@ -22,7 +24,6 @@ import org.schabi.newpipe.extractor.channel.ChannelInfo
 import org.schabi.newpipe.extractor.stream.StreamType
 import java.io.IOException
 import java.time.OffsetDateTime
-import kotlin.streams.toList
 
 class FeedDAOTest {
     private lateinit var db: AppDatabase
@@ -94,14 +95,10 @@ class FeedDAOTest {
         )
     }
 
-    private fun setupUnlinkDelete(time: String) {
+    private fun setupUnlinkDelete(time: String) = runBlocking(Dispatchers.IO) {
         clearAndFillTables()
-        Single.fromCallable {
-            feedDAO.unlinkStreamsOlderThan(OffsetDateTime.parse(time))
-        }.blockingSubscribe()
-        Single.fromCallable {
-            streamDAO.deleteOrphans()
-        }.blockingSubscribe()
+        feedDAO.unlinkStreamsOlderThan(OffsetDateTime.parse(time))
+        streamDAO.deleteOrphans().await()
     }
 
     private fun clearAndFillTables() {

app/src/androidTest/java/org/schabi/newpipe/local/playlist/LocalPlaylistManagerTest.kt

@@ -72,6 +72,6 @@ class LocalPlaylistManagerTest {
         val result = manager.createPlaylist("name", listOf(stream, upserted))
 
         result.test().await().assertComplete()
-        database.streamDAO().all.test().awaitCount(1).assertValue(listOf(stream, upserted))
+        database.streamDAO().getAll().test().awaitCount(1).assertValue(listOf(stream, upserted))
     }
 }

app/src/main/java/org/schabi/newpipe/database/feed/dao/FeedDAO.kt

@@ -119,7 +119,7 @@ abstract class FeedDAO {
               AND   s.upload_date <> max_upload_date))
         """
     )
-    abstract fun unlinkStreamsOlderThan(offsetDateTime: OffsetDateTime)
+    abstract suspend fun unlinkStreamsOlderThan(offsetDateTime: OffsetDateTime)
 
     @Query(
         """

app/src/main/java/org/schabi/newpipe/database/history/dao/StreamHistoryDAO.java

@@ -1,18 +1,5 @@
 package org.schabi.newpipe.database.history.dao;
 
-import androidx.annotation.Nullable;
-import androidx.room.Dao;
-import androidx.room.Query;
-import androidx.room.RewriteQueriesToDropUnusedColumns;
-
-import org.schabi.newpipe.database.history.model.StreamHistoryEntity;
-import org.schabi.newpipe.database.history.model.StreamHistoryEntry;
-import org.schabi.newpipe.database.stream.StreamStatisticsEntry;
-
-import java.util.List;
-
-import io.reactivex.rxjava3.core.Flowable;
-
 import static org.schabi.newpipe.database.history.model.StreamHistoryEntity.JOIN_STREAM_ID;
 import static org.schabi.newpipe.database.history.model.StreamHistoryEntity.STREAM_ACCESS_DATE;
 import static org.schabi.newpipe.database.history.model.StreamHistoryEntity.STREAM_HISTORY_TABLE;
@@ -25,65 +12,95 @@
 import static org.schabi.newpipe.database.stream.model.StreamStateEntity.STREAM_PROGRESS_MILLIS;
 import static org.schabi.newpipe.database.stream.model.StreamStateEntity.STREAM_STATE_TABLE;
 
-@Dao
-public abstract class StreamHistoryDAO implements HistoryDAO<StreamHistoryEntity> {
-    @Query("SELECT * FROM " + STREAM_HISTORY_TABLE
-            + " WHERE " + STREAM_ACCESS_DATE + " = "
-            + "(SELECT MAX(" + STREAM_ACCESS_DATE + ") FROM " + STREAM_HISTORY_TABLE + ")")
-    @Override
-    @Nullable
-    public abstract StreamHistoryEntity getLatestEntry();
+import androidx.annotation.Nullable;
+import androidx.paging.PagingSource;
+import androidx.room.Dao;
+import androidx.room.Delete;
+import androidx.room.Insert;
+import androidx.room.Query;
+import androidx.room.RewriteQueriesToDropUnusedColumns;
 
-    @Override
-    @Query("SELECT * FROM " + STREAM_HISTORY_TABLE)
-    public abstract Flowable<List<StreamHistoryEntity>> getAll();
+import org.schabi.newpipe.database.history.model.StreamHistoryEntity;
+import org.schabi.newpipe.database.history.model.StreamHistoryEntry;
+import org.schabi.newpipe.database.stream.StreamStatisticsEntry;
 
-    @Override
-    @Query("DELETE FROM " + STREAM_HISTORY_TABLE)
-    public abstract int deleteAll();
+import java.util.List;
 
-    @Override
-    public Flowable<List<StreamHistoryEntity>> listByService(final int serviceId) {
-        throw new UnsupportedOperationException();
-    }
+import io.reactivex.rxjava3.core.Completable;
+import io.reactivex.rxjava3.core.Flowable;
 
-    @Query("SELECT * FROM " + STREAM_TABLE
-            + " INNER JOIN " + STREAM_HISTORY_TABLE
-            + " ON " + STREAM_ID + " = " + JOIN_STREAM_ID
-            + " ORDER BY " + STREAM_ACCESS_DATE + " DESC")
-    public abstract Flowable<List<StreamHistoryEntry>> getHistory();
+@Dao
+public interface StreamHistoryDAO {
+    @Insert
+    long insert(StreamHistoryEntity entity);
 
+    @Delete
+    void delete(StreamHistoryEntity entity);
+
+    @Query("DELETE FROM " + STREAM_HISTORY_TABLE)
+    Completable deleteAll();
 
     @Query("SELECT * FROM " + STREAM_TABLE
             + " INNER JOIN " + STREAM_HISTORY_TABLE
             + " ON " + STREAM_ID + " = " + JOIN_STREAM_ID
             + " ORDER BY " + STREAM_ID + " ASC")
-    public abstract Flowable<List<StreamHistoryEntry>> getHistorySortedById();
+    Flowable<List<StreamHistoryEntry>> getHistorySortedById();
 
     @Query("SELECT * FROM " + STREAM_HISTORY_TABLE + " WHERE " + JOIN_STREAM_ID
             + " = :streamId ORDER BY " + STREAM_ACCESS_DATE + " DESC LIMIT 1")
     @Nullable
-    public abstract StreamHistoryEntity getLatestEntry(long streamId);
+    StreamHistoryEntity getLatestEntry(long streamId);
 
     @Query("DELETE FROM " + STREAM_HISTORY_TABLE + " WHERE " + JOIN_STREAM_ID + " = :streamId")
-    public abstract int deleteStreamHistory(long streamId);
+    Completable deleteStreamHistory(long streamId);
+
+    @Query("SELECT * FROM " + STREAM_TABLE
+            + " INNER JOIN " + STREAM_HISTORY_TABLE
+            + " ON " + STREAM_ID + " = " + JOIN_STREAM_ID
+            + " ORDER BY " + STREAM_ACCESS_DATE + " DESC")
+    Flowable<List<StreamHistoryEntry>> getHistory();
 
     @RewriteQueriesToDropUnusedColumns
     @Query("SELECT * FROM " + STREAM_TABLE
+            // Select the latest entry and watch count for each stream id on history table
+            + " INNER JOIN "
+            + "(SELECT " + JOIN_STREAM_ID + ", "
+            + "  MAX(" + STREAM_ACCESS_DATE + ") AS " + STREAM_LATEST_DATE + ", "
+            + "  SUM(" + STREAM_REPEAT_COUNT + ") AS " + STREAM_WATCH_COUNT
+            + " FROM " + STREAM_HISTORY_TABLE
+            + " GROUP BY " + JOIN_STREAM_ID + ")"
+
+            + " ON " + STREAM_ID + " = " + JOIN_STREAM_ID
 
+            + " LEFT JOIN "
+            + "(SELECT " + JOIN_STREAM_ID + " AS " + JOIN_STREAM_ID_ALIAS + ", "
+            + STREAM_PROGRESS_MILLIS
+            + " FROM " + STREAM_STATE_TABLE + " )"
+            + " ON " + STREAM_ID + " = " + JOIN_STREAM_ID_ALIAS
+
+            + " ORDER BY " + STREAM_LATEST_DATE + " DESC"
+    )
+    PagingSource<Integer, StreamStatisticsEntry> getHistoryOrderedByLastWatched();
+
+    @RewriteQueriesToDropUnusedColumns
+    @Query("SELECT * FROM " + STREAM_TABLE
             // Select the latest entry and watch count for each stream id on history table
             + " INNER JOIN "
             + "(SELECT " + JOIN_STREAM_ID + ", "
             + "  MAX(" + STREAM_ACCESS_DATE + ") AS " + STREAM_LATEST_DATE + ", "
             + "  SUM(" + STREAM_REPEAT_COUNT + ") AS " + STREAM_WATCH_COUNT
-            + " FROM " + STREAM_HISTORY_TABLE + " GROUP BY " + JOIN_STREAM_ID + ")"
+            + " FROM " + STREAM_HISTORY_TABLE
+            + " GROUP BY " + JOIN_STREAM_ID + ")"
 
             + " ON " + STREAM_ID + " = " + JOIN_STREAM_ID
 
             + " LEFT JOIN "
             + "(SELECT " + JOIN_STREAM_ID + " AS " + JOIN_STREAM_ID_ALIAS + ", "
             + STREAM_PROGRESS_MILLIS
             + " FROM " + STREAM_STATE_TABLE + " )"
-            + " ON " + STREAM_ID + " = " + JOIN_STREAM_ID_ALIAS)
-    public abstract Flowable<List<StreamStatisticsEntry>> getStatistics();
+            + " ON " + STREAM_ID + " = " + JOIN_STREAM_ID_ALIAS
+
+            + " ORDER BY " + STREAM_WATCH_COUNT + " DESC"
+    )
+    PagingSource<Integer, StreamStatisticsEntry> getHistoryOrderedByViewCount();
 }

app/src/main/java/org/schabi/newpipe/database/stream/StreamStatisticsEntry.kt

@@ -6,8 +6,6 @@ import org.schabi.newpipe.database.LocalItem
 import org.schabi.newpipe.database.history.model.StreamHistoryEntity
 import org.schabi.newpipe.database.stream.model.StreamEntity
 import org.schabi.newpipe.database.stream.model.StreamStateEntity.STREAM_PROGRESS_MILLIS
-import org.schabi.newpipe.extractor.stream.StreamInfoItem
-import org.schabi.newpipe.util.image.ImageStrategy
 import java.time.OffsetDateTime
 
 class StreamStatisticsEntry(
@@ -26,16 +24,6 @@ class StreamStatisticsEntry(
     @ColumnInfo(name = STREAM_WATCH_COUNT)
     val watchCount: Long
 ) : LocalItem {
-    fun toStreamInfoItem(): StreamInfoItem {
-        val item = StreamInfoItem(streamEntity.serviceId, streamEntity.url, streamEntity.title, streamEntity.streamType)
-        item.duration = streamEntity.duration
-        item.uploaderName = streamEntity.uploader
-        item.uploaderUrl = streamEntity.uploaderUrl
-        item.thumbnails = ImageStrategy.dbUrlToImageList(streamEntity.thumbnailUrl)
-
-        return item
-    }
-
     override fun getLocalItemType(): LocalItem.LocalItemType {
         return LocalItem.LocalItemType.STATISTIC_STREAM_ITEM
     }

app/src/main/java/org/schabi/newpipe/database/stream/dao/StreamDAO.kt

@@ -30,9 +30,6 @@ abstract class StreamDAO : BasicDAO<StreamEntity> {
     @Query("SELECT * FROM streams WHERE url = :url AND service_id = :serviceId")
     abstract fun getStream(serviceId: Long, url: String): Maybe<StreamEntity>
 
-    @Query("UPDATE streams SET uploader_url = :uploaderUrl WHERE url = :url AND service_id = :serviceId")
-    abstract fun setUploaderUrl(serviceId: Long, url: String, uploaderUrl: String): Completable
-
     @Insert(onConflict = OnConflictStrategy.IGNORE)
     internal abstract fun silentInsertInternal(stream: StreamEntity): Long
 
@@ -123,7 +120,7 @@ abstract class StreamDAO : BasicDAO<StreamEntity> {
         WHERE f.stream_id = streams.uid)
         """
     )
-    abstract fun deleteOrphans(): Int
+    abstract fun deleteOrphans(): Completable
 
     /**
      * Minimal entry class used when comparing/updating an existent stream.

app/src/main/java/org/schabi/newpipe/database/stream/dao/StreamStateDAO.kt

@@ -0,0 +1,23 @@
+package org.schabi.newpipe.database.stream.dao
+
+import androidx.room.Dao
+import androidx.room.Query
+import androidx.room.Upsert
+import io.reactivex.rxjava3.core.Completable
+import io.reactivex.rxjava3.core.Maybe
+import org.schabi.newpipe.database.stream.model.StreamStateEntity
+
+@Dao
+interface StreamStateDAO {
+    @Query("DELETE FROM " + StreamStateEntity.STREAM_STATE_TABLE)
+    fun deleteAll(): Completable
+
+    @Query("SELECT * FROM " + StreamStateEntity.STREAM_STATE_TABLE + " WHERE " + StreamStateEntity.JOIN_STREAM_ID + " = :streamId")
+    fun getState(streamId: Long): Maybe<StreamStateEntity>
+
+    @Query("DELETE FROM " + StreamStateEntity.STREAM_STATE_TABLE + " WHERE " + StreamStateEntity.JOIN_STREAM_ID + " = :streamId")
+    fun deleteState(streamId: Long): Completable
+
+    @Upsert
+    fun upsert(stream: StreamStateEntity)
+}