-Added schema for local playlist and stream statistics.

-Added normalized schema for stream history.
-Added managers for specialized database access for stream and local playlist.

app/src/main/java/org/schabi/newpipe/fragments/playlist/LocalPlaylistManager.java

@@ -0,0 +1,78 @@
+package org.schabi.newpipe.fragments.playlist;
+
+import org.schabi.newpipe.database.AppDatabase;
+import org.schabi.newpipe.database.playlist.PlaylistMetadataEntry;
+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.Completable;
+import io.reactivex.Maybe;
+
+public class LocalPlaylistManager {
+
+    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 until user is able to select thumbnail
+        if (streams.isEmpty()) return Maybe.empty();
+        final StreamEntity defaultStream = streams.get(0);
+        final PlaylistEntity newPlaylist = new PlaylistEntity(name, defaultStream.getThumbnailUrl());
+
+        return Maybe.fromCallable(() -> database.runInTransaction(() -> {
+            final long playlistId = playlistTable.insert(newPlaylist);
+
+            List<PlaylistStreamEntity> joinEntities = new ArrayList<>(streams.size());
+            for (int index = 0; index < streams.size(); index++) {
+                // Upsert streams and get their ids
+                final long streamId = streamTable.upsert(streams.get(index));
+                joinEntities.add(new PlaylistStreamEntity(playlistId, streamId, index));
+            }
+
+            return playlistStreamTable.insertAll(joinEntities);
+        }));
+    }
+
+    public Maybe<Long> appendToPlaylist(final long playlistId, final StreamEntity stream) {
+        final Maybe<Long> streamIdFuture = Maybe.fromCallable(() -> streamTable.upsert(stream));
+        final Maybe<Integer> joinIndexFuture =
+                playlistStreamTable.getMaximumIndexOf(playlistId).firstElement();
+
+        return Maybe.zip(streamIdFuture, joinIndexFuture, (streamId, currentMaxJoinIndex) ->
+                playlistStreamTable.insert(new PlaylistStreamEntity(playlistId,
+                        streamId, currentMaxJoinIndex + 1))
+        );
+    }
+
+    public Completable updateJoin(final long playlistId, final List<Long> streamIds) {
+        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);
+        }));
+    }
+
+    public Maybe<List<PlaylistMetadataEntry>> getPlaylists() {
+        return playlistStreamTable.getPlaylistMetadata().firstElement();
+    }
+}

app/src/main/java/org/schabi/newpipe/fragments/playlist/StreamRecordManager.java

@@ -0,0 +1,47 @@
+package org.schabi.newpipe.fragments.playlist;
+
+import org.schabi.newpipe.database.AppDatabase;
+import org.schabi.newpipe.database.stream.StreamStatisticsEntry;
+import org.schabi.newpipe.database.stream.dao.StreamDAO;
+import org.schabi.newpipe.database.stream.dao.StreamHistoryDAO;
+import org.schabi.newpipe.database.stream.model.StreamEntity;
+import org.schabi.newpipe.database.stream.model.StreamHistoryEntity;
+import org.schabi.newpipe.extractor.stream.StreamInfo;
+import org.schabi.newpipe.extractor.stream.StreamInfoItem;
+
+import java.util.Date;
+import java.util.List;
+
+import io.reactivex.MaybeObserver;
+import io.reactivex.Single;
+import io.reactivex.disposables.Disposable;
+import io.reactivex.schedulers.Schedulers;
+
+public class StreamRecordManager {
+
+    private final AppDatabase database;
+    private final StreamDAO streamTable;
+    private final StreamHistoryDAO historyTable;
+
+    public StreamRecordManager(final AppDatabase db) {
+        database = db;
+        streamTable = db.streamDAO();
+        historyTable = db.streamHistoryDAO();
+    }
+
+    public int onChanged(final StreamInfoItem infoItem) {
+        // Only existing streams are updated
+        return streamTable.update(new StreamEntity(infoItem));
+    }
+
+    public Single<Long> onViewed(final StreamInfo info) {
+        return Single.fromCallable(() -> database.runInTransaction(() -> {
+            final long streamId = streamTable.upsert(new StreamEntity(info));
+            return historyTable.insert(new StreamHistoryEntity(streamId, new Date()));
+        })).subscribeOn(Schedulers.io());
+    }
+
+    public int removeHistory(final long streamId) {
+        return historyTable.deleteHistory(streamId);
+    }
+}