musikr: simplify pipeline

app/src/main/java/org/oxycblt/auxio/music/MusicRepository.kt

@@ -29,6 +29,7 @@ import kotlinx.coroutines.withContext
 import kotlinx.coroutines.yield
 import org.oxycblt.auxio.image.covers.SettingCovers
 import org.oxycblt.auxio.music.MusicRepository.IndexingWorker
+import org.oxycblt.auxio.music.shim.WriteOnlyMutableCache
 import org.oxycblt.musikr.IndexingProgress
 import org.oxycblt.musikr.Interpretation
 import org.oxycblt.musikr.Library
@@ -38,7 +39,7 @@ import org.oxycblt.musikr.MutableLibrary
 import org.oxycblt.musikr.Playlist
 import org.oxycblt.musikr.Song
 import org.oxycblt.musikr.Storage
-import org.oxycblt.musikr.cache.StoredCache
+import org.oxycblt.musikr.cache.db.MutableDBCache
 import org.oxycblt.musikr.playlist.db.StoredPlaylists
 import org.oxycblt.musikr.tag.interpret.Naming
 import org.oxycblt.musikr.tag.interpret.Separators
@@ -236,7 +237,7 @@ class MusicRepositoryImpl
 @Inject
 constructor(
     @ApplicationContext private val context: Context,
-    private val storedCache: StoredCache,
+    private val dbCache: MutableDBCache,
     private val storedPlaylists: StoredPlaylists,
     private val settingCovers: SettingCovers,
     private val musicSettings: MusicSettings
@@ -388,11 +389,10 @@ constructor(
 
         val currentRevision = musicSettings.revision
         val newRevision = currentRevision?.takeIf { withCache } ?: UUID.randomUUID()
-        val cache = if (withCache) storedCache.visible() else storedCache.invisible()
+        val cache = if (withCache) dbCache else WriteOnlyMutableCache(dbCache)
         val covers = settingCovers.mutate(context, newRevision)
         val storage = Storage(cache, covers, storedPlaylists)
         val interpretation = Interpretation(nameFactory, separators, ignoreHidden)
-
         val result =
             Musikr.new(context, storage, interpretation).run(locations, ::emitIndexingProgress)
         // Music loading completed, update the revision right now so we re-use this work

app/src/main/java/org/oxycblt/auxio/music/shim/MusikrShimModule.kt

@@ -25,7 +25,7 @@ import dagger.hilt.InstallIn
 import dagger.hilt.android.qualifiers.ApplicationContext
 import dagger.hilt.components.SingletonComponent
 import javax.inject.Singleton
-import org.oxycblt.musikr.cache.StoredCache
+import org.oxycblt.musikr.cache.db.MutableDBCache
 import org.oxycblt.musikr.playlist.db.StoredPlaylists
 
 @Module
@@ -33,7 +33,7 @@ import org.oxycblt.musikr.playlist.db.StoredPlaylists
 class MusikrShimModule {
     @Singleton
     @Provides
-    fun storedCache(@ApplicationContext context: Context) = StoredCache.from(context)
+    fun cache(@ApplicationContext context: Context) = MutableDBCache.from(context)
 
     @Singleton
     @Provides

app/src/main/java/org/oxycblt/auxio/music/shim/WriteOnlyMutableCache.kt

@@ -0,0 +1,41 @@
+/*
+ * Copyright (c) 2025 Auxio Project
+ * WriteOnlyMutableCache.kt is part of Auxio.
+ *
+ * This program 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.
+ *
+ * This program 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 this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+ 
+package org.oxycblt.auxio.music.shim
+
+import org.oxycblt.musikr.cache.CacheResult
+import org.oxycblt.musikr.cache.CachedSong
+import org.oxycblt.musikr.cache.MutableCache
+import org.oxycblt.musikr.fs.device.DeviceFile
+
+class WriteOnlyMutableCache(private val inner: MutableCache) : MutableCache {
+    override suspend fun read(file: DeviceFile): CacheResult {
+        return when (val result = inner.read(file)) {
+            is CacheResult.Hit -> CacheResult.Stale(file, result.song.addedMs)
+            else -> result
+        }
+    }
+
+    override suspend fun write(cachedSong: CachedSong) {
+        inner.write(cachedSong)
+    }
+
+    override suspend fun cleanup(excluding: List<CachedSong>) {
+        inner.cleanup(excluding)
+    }
+}

musikr/src/main/java/org/oxycblt/musikr/Config.kt

@@ -18,7 +18,7 @@
  
 package org.oxycblt.musikr
 
-import org.oxycblt.musikr.cache.Cache
+import org.oxycblt.musikr.cache.MutableCache
 import org.oxycblt.musikr.cover.Cover
 import org.oxycblt.musikr.cover.MutableCovers
 import org.oxycblt.musikr.playlist.db.StoredPlaylists
@@ -31,14 +31,14 @@ data class Storage(
      * A factory producing a repository of cached metadata to read and write from over the course of
      * music loading. This will only be used during music loading.
      */
-    val cache: Cache.Factory,
+    val cache: MutableCache,
 
     /**
      * A repository of cover images to for re-use during music loading. Should be kept in lock-step
      * with the cache for best performance. This will be used during music loading and when
      * retrieving cover information from the library.
      */
-    val storedCovers: MutableCovers<out Cover>,
+    val covers: MutableCovers<out Cover>,
 
     /**
      * A repository of user-created playlists that should also be loaded into the library. This will

musikr/src/main/java/org/oxycblt/musikr/Musikr.kt

@@ -143,6 +143,6 @@ private class LibraryResultImpl(
     override val library: MutableLibrary
 ) : LibraryResult {
     override suspend fun cleanup() {
-        storage.storedCovers.cleanup(library.songs.mapNotNull { it.cover })
+        storage.covers.cleanup(library.songs.mapNotNull { it.cover })
     }
 }

musikr/src/main/java/org/oxycblt/musikr/cache/Cache.kt

@@ -18,25 +18,32 @@
  
 package org.oxycblt.musikr.cache
 
-import org.oxycblt.musikr.cover.Cover
-import org.oxycblt.musikr.cover.Covers
 import org.oxycblt.musikr.fs.device.DeviceFile
-import org.oxycblt.musikr.pipeline.RawSong
+import org.oxycblt.musikr.metadata.Properties
+import org.oxycblt.musikr.tag.parse.ParsedTags
 
-abstract class Cache {
+interface Cache {
-    internal abstract suspend fun read(file: DeviceFile, covers: Covers<out Cover>): CacheResult
+    suspend fun read(file: DeviceFile): CacheResult
-
-    internal abstract suspend fun write(song: RawSong)
-
-    internal abstract suspend fun finalize()
-
-    abstract class Factory {
-        internal abstract fun open(): Cache
-    }
 }
 
-internal sealed interface CacheResult {
+interface MutableCache : Cache {
-    data class Hit(val song: RawSong) : CacheResult
+    suspend fun write(cachedSong: CachedSong)
 
-    data class Miss(val file: DeviceFile, val addedMs: Long?) : CacheResult
+    suspend fun cleanup(excluding: List<CachedSong>)
+}
+
+data class CachedSong(
+    val file: DeviceFile,
+    val properties: Properties,
+    val tags: ParsedTags,
+    val coverId: String?,
+    val addedMs: Long
+)
+
+sealed interface CacheResult {
+    data class Hit(val song: CachedSong) : CacheResult
+
+    data class Miss(val file: DeviceFile) : CacheResult
+
+    data class Stale(val file: DeviceFile, val addedMs: Long) : CacheResult
 }

musikr/src/main/java/org/oxycblt/musikr/cache/CacheDatabase.kt

@@ -1,209 +0,0 @@
-/*
- * Copyright (c) 2023 Auxio Project
- * CacheDatabase.kt is part of Auxio.
- *
- * This program 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.
- *
- * This program 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 this program.  If not, see <https://www.gnu.org/licenses/>.
- */
- 
-package org.oxycblt.musikr.cache
-
-import android.content.Context
-import androidx.room.Dao
-import androidx.room.Database
-import androidx.room.Entity
-import androidx.room.Insert
-import androidx.room.OnConflictStrategy
-import androidx.room.PrimaryKey
-import androidx.room.Query
-import androidx.room.Room
-import androidx.room.RoomDatabase
-import androidx.room.Transaction
-import androidx.room.TypeConverter
-import androidx.room.TypeConverters
-import org.oxycblt.musikr.cover.Cover
-import org.oxycblt.musikr.cover.CoverResult
-import org.oxycblt.musikr.cover.Covers
-import org.oxycblt.musikr.fs.device.DeviceFile
-import org.oxycblt.musikr.metadata.Properties
-import org.oxycblt.musikr.pipeline.RawSong
-import org.oxycblt.musikr.tag.Date
-import org.oxycblt.musikr.tag.parse.ParsedTags
-import org.oxycblt.musikr.util.correctWhitespace
-import org.oxycblt.musikr.util.splitEscaped
-
-@Database(entities = [CachedSong::class], version = 60, exportSchema = false)
-internal abstract class CacheDatabase : RoomDatabase() {
-    abstract fun visibleDao(): VisibleCacheDao
-
-    abstract fun invisibleDao(): InvisibleCacheDao
-
-    abstract fun writeDao(): CacheWriteDao
-
-    companion object {
-        fun from(context: Context) =
-            Room.databaseBuilder(
-                    context.applicationContext, CacheDatabase::class.java, "music_cache.db")
-                .fallbackToDestructiveMigration()
-                .build()
-    }
-}
-
-@Dao
-internal interface VisibleCacheDao {
-    @Query("SELECT * FROM CachedSong WHERE uri = :uri")
-    suspend fun selectSong(uri: String): CachedSong?
-
-    @Query("SELECT addedMs FROM CachedSong WHERE uri = :uri")
-    suspend fun selectAddedMs(uri: String): Long?
-
-    @Transaction suspend fun touch(uri: String) = updateTouchedNs(uri, System.nanoTime())
-
-    @Query("UPDATE CachedSong SET touchedNs = :nowNs WHERE uri = :uri")
-    suspend fun updateTouchedNs(uri: String, nowNs: Long)
-}
-
-@Dao
-internal interface InvisibleCacheDao {
-    @Query("SELECT addedMs FROM CachedSong WHERE uri = :uri")
-    suspend fun selectAddedMs(uri: String): Long?
-}
-
-@Dao
-internal interface CacheWriteDao {
-    @Insert(onConflict = OnConflictStrategy.REPLACE) suspend fun updateSong(cachedSong: CachedSong)
-
-    @Query("DELETE FROM CachedSong WHERE touchedNs < :now") suspend fun pruneOlderThan(now: Long)
-}
-
-@Entity
-@TypeConverters(CachedSong.Converters::class)
-internal data class CachedSong(
-    @PrimaryKey val uri: String,
-    val modifiedMs: Long,
-    val addedMs: Long,
-    val touchedNs: Long,
-    val mimeType: String,
-    val durationMs: Long,
-    val bitrateHz: Int,
-    val sampleRateHz: Int,
-    val musicBrainzId: String?,
-    val name: String?,
-    val sortName: String?,
-    val track: Int?,
-    val disc: Int?,
-    val subtitle: String?,
-    val date: Date?,
-    val albumMusicBrainzId: String?,
-    val albumName: String?,
-    val albumSortName: String?,
-    val releaseTypes: List<String>,
-    val artistMusicBrainzIds: List<String>,
-    val artistNames: List<String>,
-    val artistSortNames: List<String>,
-    val albumArtistMusicBrainzIds: List<String>,
-    val albumArtistNames: List<String>,
-    val albumArtistSortNames: List<String>,
-    val genreNames: List<String>,
-    val replayGainTrackAdjustment: Float?,
-    val replayGainAlbumAdjustment: Float?,
-    val coverId: String?,
-) {
-    suspend fun intoRawSong(file: DeviceFile, covers: Covers<out Cover>): RawSong? {
-        val cover =
-            when (val result = coverId?.let { covers.obtain(it) }) {
-                // We found the cover.
-                is CoverResult.Hit<out Cover> -> result.cover
-                // We actually didn't find the cover, can't safely convert.
-                is CoverResult.Miss<out Cover> -> return null
-                // No cover in the first place, can ignore.
-                null -> null
-            }
-        return RawSong(
-            file,
-            Properties(mimeType, durationMs, bitrateHz, sampleRateHz),
-            ParsedTags(
-                musicBrainzId = musicBrainzId,
-                name = name,
-                sortName = sortName,
-                durationMs = durationMs,
-                track = track,
-                disc = disc,
-                subtitle = subtitle,
-                date = date,
-                albumMusicBrainzId = albumMusicBrainzId,
-                albumName = albumName,
-                albumSortName = albumSortName,
-                releaseTypes = releaseTypes,
-                artistMusicBrainzIds = artistMusicBrainzIds,
-                artistNames = artistNames,
-                artistSortNames = artistSortNames,
-                albumArtistMusicBrainzIds = albumArtistMusicBrainzIds,
-                albumArtistNames = albumArtistNames,
-                albumArtistSortNames = albumArtistSortNames,
-                genreNames = genreNames,
-                replayGainTrackAdjustment = replayGainTrackAdjustment,
-                replayGainAlbumAdjustment = replayGainAlbumAdjustment),
-            cover = cover,
-            addedMs = addedMs)
-    }
-
-    object Converters {
-        @TypeConverter
-        fun fromMultiValue(values: List<String>) =
-            values.joinToString(";") { it.replace(";", "\\;") }
-
-        @TypeConverter
-        fun toMultiValue(string: String) = string.splitEscaped { it == ';' }.correctWhitespace()
-
-        @TypeConverter fun fromDate(date: Date?) = date?.toString()
-
-        @TypeConverter fun toDate(string: String?) = string?.let(Date::from)
-    }
-
-    companion object {
-        fun fromRawSong(rawSong: RawSong) =
-            CachedSong(
-                uri = rawSong.file.uri.toString(),
-                modifiedMs = rawSong.file.modifiedMs,
-                addedMs = rawSong.addedMs,
-                // Should be strictly monotonic so we don't prune this
-                // by accident later.
-                touchedNs = System.nanoTime(),
-                musicBrainzId = rawSong.tags.musicBrainzId,
-                name = rawSong.tags.name,
-                sortName = rawSong.tags.sortName,
-                durationMs = rawSong.tags.durationMs,
-                track = rawSong.tags.track,
-                disc = rawSong.tags.disc,
-                subtitle = rawSong.tags.subtitle,
-                date = rawSong.tags.date,
-                albumMusicBrainzId = rawSong.tags.albumMusicBrainzId,
-                albumName = rawSong.tags.albumName,
-                albumSortName = rawSong.tags.albumSortName,
-                releaseTypes = rawSong.tags.releaseTypes,
-                artistMusicBrainzIds = rawSong.tags.artistMusicBrainzIds,
-                artistNames = rawSong.tags.artistNames,
-                artistSortNames = rawSong.tags.artistSortNames,
-                albumArtistMusicBrainzIds = rawSong.tags.albumArtistMusicBrainzIds,
-                albumArtistNames = rawSong.tags.albumArtistNames,
-                albumArtistSortNames = rawSong.tags.albumArtistSortNames,
-                genreNames = rawSong.tags.genreNames,
-                replayGainTrackAdjustment = rawSong.tags.replayGainTrackAdjustment,
-                replayGainAlbumAdjustment = rawSong.tags.replayGainAlbumAdjustment,
-                coverId = rawSong.cover?.id,
-                mimeType = rawSong.properties.mimeType,
-                bitrateHz = rawSong.properties.bitrateKbps,
-                sampleRateHz = rawSong.properties.sampleRateHz)
-    }
-}

musikr/src/main/java/org/oxycblt/musikr/cache/StoredCache.kt

@@ -1,88 +0,0 @@
-/*
- * Copyright (c) 2024 Auxio Project
- * StoredCache.kt is part of Auxio.
- *
- * This program 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.
- *
- * This program 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 this program.  If not, see <https://www.gnu.org/licenses/>.
- */
- 
-package org.oxycblt.musikr.cache
-
-import android.content.Context
-import org.oxycblt.musikr.cover.Cover
-import org.oxycblt.musikr.cover.Covers
-import org.oxycblt.musikr.fs.device.DeviceFile
-import org.oxycblt.musikr.pipeline.RawSong
-
-interface StoredCache {
-    fun visible(): Cache.Factory
-
-    fun invisible(): Cache.Factory
-
-    companion object {
-        fun from(context: Context): StoredCache = StoredCacheImpl(CacheDatabase.from(context))
-    }
-}
-
-private class StoredCacheImpl(private val cacheDatabase: CacheDatabase) : StoredCache {
-    override fun visible(): Cache.Factory = VisibleStoredCache.Factory(cacheDatabase)
-
-    override fun invisible(): Cache.Factory = InvisibleStoredCache.Factory(cacheDatabase)
-}
-
-private abstract class BaseStoredCache(protected val writeDao: CacheWriteDao) : Cache() {
-    private val created = System.nanoTime()
-
-    override suspend fun write(song: RawSong) = writeDao.updateSong(CachedSong.fromRawSong(song))
-
-    override suspend fun finalize() {
-        // Anything not create during this cache's use implies that it has not been
-        // access during this run and should be pruned.
-        writeDao.pruneOlderThan(created)
-    }
-}
-
-private class VisibleStoredCache(private val visibleDao: VisibleCacheDao, writeDao: CacheWriteDao) :
-    BaseStoredCache(writeDao) {
-    override suspend fun read(file: DeviceFile, covers: Covers<out Cover>): CacheResult {
-        val song = visibleDao.selectSong(file.uri.toString()) ?: return CacheResult.Miss(file, null)
-        if (song.modifiedMs != file.modifiedMs) {
-            // We *found* this file earlier, but it's out of date.
-            // Send back it with the timestamp so it will be re-used.
-            // The touch timestamp will be updated on write.
-            return CacheResult.Miss(file, song.addedMs)
-        }
-        // Valid file, update the touch time.
-        visibleDao.touch(file.uri.toString())
-        val rawSong = song.intoRawSong(file, covers) ?: return CacheResult.Miss(file, song.addedMs)
-        return CacheResult.Hit(rawSong)
-    }
-
-    class Factory(private val cacheDatabase: CacheDatabase) : Cache.Factory() {
-        override fun open() =
-            VisibleStoredCache(cacheDatabase.visibleDao(), cacheDatabase.writeDao())
-    }
-}
-
-private class InvisibleStoredCache(
-    private val invisibleCacheDao: InvisibleCacheDao,
-    writeDao: CacheWriteDao
-) : BaseStoredCache(writeDao) {
-    override suspend fun read(file: DeviceFile, covers: Covers<out Cover>) =
-        CacheResult.Miss(file, invisibleCacheDao.selectAddedMs(file.uri.toString()))
-
-    class Factory(private val cacheDatabase: CacheDatabase) : Cache.Factory() {
-        override fun open() =
-            InvisibleStoredCache(cacheDatabase.invisibleDao(), cacheDatabase.writeDao())
-    }
-}

musikr/src/main/java/org/oxycblt/musikr/cache/db/CacheDatabase.kt

@@ -0,0 +1,127 @@
+/*
+ * Copyright (c) 2023 Auxio Project
+ * CacheDatabase.kt is part of Auxio.
+ *
+ * This program 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.
+ *
+ * This program 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 this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+ 
+package org.oxycblt.musikr.cache.db
+
+import android.content.Context
+import android.net.Uri
+import androidx.room.Dao
+import androidx.room.Database
+import androidx.room.Entity
+import androidx.room.Insert
+import androidx.room.OnConflictStrategy
+import androidx.room.PrimaryKey
+import androidx.room.Query
+import androidx.room.Room
+import androidx.room.RoomDatabase
+import androidx.room.Transaction
+import androidx.room.TypeConverter
+import androidx.room.TypeConverters
+import org.oxycblt.musikr.tag.Date
+import org.oxycblt.musikr.util.correctWhitespace
+import org.oxycblt.musikr.util.splitEscaped
+
+@Database(entities = [CachedSongData::class], version = 61, exportSchema = false)
+internal abstract class CacheDatabase : RoomDatabase() {
+    abstract fun readDao(): CacheReadDao
+
+    abstract fun writeDao(): CacheWriteDao
+
+    companion object {
+        fun from(context: Context) =
+            Room.databaseBuilder(
+                    context.applicationContext, CacheDatabase::class.java, "music_cache.db")
+                .fallbackToDestructiveMigration()
+                .build()
+    }
+}
+
+@Dao
+internal interface CacheReadDao {
+    @Query("SELECT * FROM CachedSongData WHERE uri = :uri")
+    suspend fun selectSong(uri: String): CachedSongData?
+}
+
+@Dao
+internal interface CacheWriteDao {
+    @Insert(onConflict = OnConflictStrategy.REPLACE)
+    suspend fun updateSong(CachedSongData: CachedSongData)
+
+    @Transaction
+    suspend fun deleteExcludingUris(uris: Set<String>) {
+        val delete = selectAllUris().toSet() - uris
+        for (chunk in delete.chunked(999)) {
+            deleteExcludingUriChunk(chunk)
+        }
+    }
+
+    @Query("SELECT uri FROM CachedSongData") suspend fun selectAllUris(): List<String>
+
+    @Query("DELETE FROM CachedSongData WHERE uri IN (:uris)")
+    suspend fun deleteExcludingUriChunk(uris: List<String>)
+}
+
+@Entity
+@TypeConverters(CachedSongData.Converters::class)
+internal data class CachedSongData(
+    @PrimaryKey val uri: Uri,
+    val modifiedMs: Long,
+    val addedMs: Long,
+    val mimeType: String,
+    val durationMs: Long,
+    val bitrateKbps: Int,
+    val sampleRateHz: Int,
+    val musicBrainzId: String?,
+    val name: String?,
+    val sortName: String?,
+    val track: Int?,
+    val disc: Int?,
+    val subtitle: String?,
+    val date: Date?,
+    val albumMusicBrainzId: String?,
+    val albumName: String?,
+    val albumSortName: String?,
+    val releaseTypes: List<String>,
+    val artistMusicBrainzIds: List<String>,
+    val artistNames: List<String>,
+    val artistSortNames: List<String>,
+    val albumArtistMusicBrainzIds: List<String>,
+    val albumArtistNames: List<String>,
+    val albumArtistSortNames: List<String>,
+    val genreNames: List<String>,
+    val replayGainTrackAdjustment: Float?,
+    val replayGainAlbumAdjustment: Float?,
+    val coverId: String?,
+) {
+    object Converters {
+        @TypeConverter
+        fun fromMultiValue(values: List<String>) =
+            values.joinToString(";") { it.replace(";", "\\;") }
+
+        @TypeConverter
+        fun toMultiValue(string: String) = string.splitEscaped { it == ';' }.correctWhitespace()
+
+        @TypeConverter fun fromDate(date: Date?) = date?.toString()
+
+        @TypeConverter fun toDate(string: String?) = string?.let(Date::from)
+
+        @TypeConverter fun toUri(string: String) = Uri.parse(string)
+
+        @TypeConverter fun fromUri(uri: Uri) = uri.toString()
+    }
+}

musikr/src/main/java/org/oxycblt/musikr/cache/db/DBCache.kt

@@ -0,0 +1,120 @@
+/*
+ * Copyright (c) 2025 Auxio Project
+ * DBCache.kt is part of Auxio.
+ *
+ * This program 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.
+ *
+ * This program 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 this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+ 
+package org.oxycblt.musikr.cache.db
+
+import android.content.Context
+import org.oxycblt.musikr.cache.Cache
+import org.oxycblt.musikr.cache.CacheResult
+import org.oxycblt.musikr.cache.CachedSong
+import org.oxycblt.musikr.cache.MutableCache
+import org.oxycblt.musikr.fs.device.DeviceFile
+import org.oxycblt.musikr.metadata.Properties
+import org.oxycblt.musikr.tag.parse.ParsedTags
+
+open class DBCache internal constructor(private val readDao: CacheReadDao) : Cache {
+    override suspend fun read(file: DeviceFile): CacheResult {
+        val dbSong = readDao.selectSong(file.uri.toString()) ?: return CacheResult.Miss(file)
+        if (dbSong.modifiedMs != file.modifiedMs) {
+            return CacheResult.Stale(file, dbSong.addedMs)
+        }
+        val song =
+            CachedSong(
+                file,
+                Properties(
+                    dbSong.mimeType, dbSong.durationMs, dbSong.bitrateKbps, dbSong.sampleRateHz),
+                ParsedTags(
+                    musicBrainzId = dbSong.musicBrainzId,
+                    name = dbSong.name,
+                    sortName = dbSong.sortName,
+                    durationMs = dbSong.durationMs,
+                    track = dbSong.track,
+                    disc = dbSong.disc,
+                    subtitle = dbSong.subtitle,
+                    date = dbSong.date,
+                    albumMusicBrainzId = dbSong.albumMusicBrainzId,
+                    albumName = dbSong.albumName,
+                    albumSortName = dbSong.albumSortName,
+                    releaseTypes = dbSong.releaseTypes,
+                    artistMusicBrainzIds = dbSong.artistMusicBrainzIds,
+                    artistNames = dbSong.artistNames,
+                    artistSortNames = dbSong.artistSortNames,
+                    albumArtistMusicBrainzIds = dbSong.albumArtistMusicBrainzIds,
+                    albumArtistNames = dbSong.albumArtistNames,
+                    albumArtistSortNames = dbSong.albumArtistSortNames,
+                    genreNames = dbSong.genreNames,
+                    replayGainTrackAdjustment = dbSong.replayGainTrackAdjustment,
+                    replayGainAlbumAdjustment = dbSong.replayGainAlbumAdjustment),
+                coverId = dbSong.coverId,
+                addedMs = dbSong.addedMs)
+        return CacheResult.Hit(song)
+    }
+
+    companion object {
+        fun from(context: Context) = DBCache(CacheDatabase.from(context).readDao())
+    }
+}
+
+class MutableDBCache
+private constructor(readDao: CacheReadDao, private val writeDao: CacheWriteDao) :
+    MutableCache, DBCache(readDao) {
+    override suspend fun write(cachedSong: CachedSong) {
+        val dbSong =
+            CachedSongData(
+                uri = cachedSong.file.uri,
+                modifiedMs = cachedSong.file.modifiedMs,
+                addedMs = cachedSong.addedMs,
+                mimeType = cachedSong.properties.mimeType,
+                durationMs = cachedSong.properties.durationMs,
+                bitrateKbps = cachedSong.properties.bitrateKbps,
+                sampleRateHz = cachedSong.properties.sampleRateHz,
+                musicBrainzId = cachedSong.tags.musicBrainzId,
+                name = cachedSong.tags.name,
+                sortName = cachedSong.tags.sortName,
+                track = cachedSong.tags.track,
+                disc = cachedSong.tags.disc,
+                subtitle = cachedSong.tags.subtitle,
+                date = cachedSong.tags.date,
+                albumMusicBrainzId = cachedSong.tags.albumMusicBrainzId,
+                albumName = cachedSong.tags.albumName,
+                albumSortName = cachedSong.tags.albumSortName,
+                releaseTypes = cachedSong.tags.releaseTypes,
+                artistMusicBrainzIds = cachedSong.tags.artistMusicBrainzIds,
+                artistNames = cachedSong.tags.artistNames,
+                artistSortNames = cachedSong.tags.artistSortNames,
+                albumArtistMusicBrainzIds = cachedSong.tags.albumArtistMusicBrainzIds,
+                albumArtistNames = cachedSong.tags.albumArtistNames,
+                albumArtistSortNames = cachedSong.tags.albumArtistSortNames,
+                genreNames = cachedSong.tags.genreNames,
+                replayGainTrackAdjustment = cachedSong.tags.replayGainTrackAdjustment,
+                replayGainAlbumAdjustment = cachedSong.tags.replayGainAlbumAdjustment,
+                coverId = cachedSong.coverId)
+        writeDao.updateSong(dbSong)
+    }
+
+    override suspend fun cleanup(excluding: List<CachedSong>) {
+        writeDao.deleteExcludingUris(excluding.mapTo(mutableSetOf()) { it.file.uri.toString() })
+    }
+
+    companion object {
+        fun from(context: Context): MutableDBCache {
+            val db = CacheDatabase.from(context)
+            return MutableDBCache(db.readDao(), db.writeDao())
+        }
+    }
+}

musikr/src/main/java/org/oxycblt/musikr/metadata/MetadataExtractor.kt

@@ -18,24 +18,29 @@
  
 package org.oxycblt.musikr.metadata
 
-import android.os.ParcelFileDescriptor
+import android.content.ContentResolver
+import android.content.Context
 import java.io.FileInputStream
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
 import org.oxycblt.musikr.fs.device.DeviceFile
 
 internal interface MetadataExtractor {
-    suspend fun extract(deviceFile: DeviceFile, fd: ParcelFileDescriptor): Metadata?
+    suspend fun extract(deviceFile: DeviceFile): Metadata?
 
     companion object {
-        fun new(): MetadataExtractor = MetadataExtractorImpl
+        fun from(context: Context): MetadataExtractor =
+            MetadataExtractorImpl(context.contentResolver)
     }
 }
 
-private object MetadataExtractorImpl : MetadataExtractor {
+private class MetadataExtractorImpl(private val contentResolver: ContentResolver) :
-    override suspend fun extract(deviceFile: DeviceFile, fd: ParcelFileDescriptor) =
+    MetadataExtractor {
+    override suspend fun extract(deviceFile: DeviceFile): Metadata? =
         withContext(Dispatchers.IO) {
-            val fis = FileInputStream(fd.fileDescriptor)
+            contentResolver.openFileDescriptor(deviceFile.uri, "r")?.use { fd ->
-            TagLibJNI.open(deviceFile, fis).also { fis.close() }
+                val fis = FileInputStream(fd.fileDescriptor)
+                TagLibJNI.open(deviceFile, fis).also { fis.close() }
+            }
         }
 }

musikr/src/main/java/org/oxycblt/musikr/pipeline/EvaluateStep.kt

@@ -18,16 +18,9 @@
  
 package org.oxycblt.musikr.pipeline
 
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.buffer
-import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.filterIsInstance
-import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.fold
-import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.flow.merge
-import kotlinx.coroutines.flow.onEach
 import org.oxycblt.musikr.Interpretation
 import org.oxycblt.musikr.MutableLibrary
 import org.oxycblt.musikr.Storage
@@ -38,7 +31,7 @@ import org.oxycblt.musikr.playlist.interpret.PlaylistInterpreter
 import org.oxycblt.musikr.tag.interpret.TagInterpreter
 
 internal interface EvaluateStep {
-    suspend fun evaluate(extractedMusic: Flow<ExtractedMusic>): MutableLibrary
+    suspend fun evaluate(extractedMusic: Flow<Extracted>): MutableLibrary
 
     companion object {
         fun new(storage: Storage, interpretation: Interpretation): EvaluateStep =
@@ -56,33 +49,16 @@ private class EvaluateStepImpl(
     private val storedPlaylists: StoredPlaylists,
     private val libraryFactory: LibraryFactory
 ) : EvaluateStep {
-    override suspend fun evaluate(extractedMusic: Flow<ExtractedMusic>): MutableLibrary {
+    override suspend fun evaluate(extractedMusic: Flow<Extracted>): MutableLibrary =
-        val filterFlow =
+        extractedMusic
-            extractedMusic.filterIsInstance<ExtractedMusic.Valid>().divert {
+            .filterIsInstance<Extracted.Valid>()
-                when (it) {
+            .fold(MusicGraph.builder()) { graphBuilder, extracted ->
-                    is ExtractedMusic.Valid.Song -> Divert.Right(it.song)
+                when (extracted) {
-                    is ExtractedMusic.Valid.Playlist -> Divert.Left(it.file)
+                    is RawSong -> graphBuilder.add(tagInterpreter.interpret(extracted))
+                    is RawPlaylist ->
+                        graphBuilder.add(playlistInterpreter.interpret(extracted.file))
                 }
+                graphBuilder
             }
-        val rawSongs = filterFlow.right
+            .let { libraryFactory.create(it.build(), storedPlaylists, playlistInterpreter) }
-        val preSongs =
-            rawSongs
-                .map { wrap(it, tagInterpreter::interpret) }
-                .flowOn(Dispatchers.Default)
-                .buffer(Channel.UNLIMITED)
-        val prePlaylists =
-            filterFlow.left
-                .map { wrap(it, playlistInterpreter::interpret) }
-                .flowOn(Dispatchers.Default)
-                .buffer(Channel.UNLIMITED)
-        val graphBuilder = MusicGraph.builder()
-        val graphBuild =
-            merge(
-                filterFlow.manager,
-                preSongs.onEach { wrap(it, graphBuilder::add) },
-                prePlaylists.onEach { wrap(it, graphBuilder::add) })
-        graphBuild.collect()
-        val graph = graphBuilder.build()
-        return libraryFactory.create(graph, storedPlaylists, playlistInterpreter)
-    }
 }

musikr/src/main/java/org/oxycblt/musikr/pipeline/ExploreStep.kt

@@ -31,50 +31,84 @@ import kotlinx.coroutines.flow.flowOn
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.merge
 import org.oxycblt.musikr.Storage
+import org.oxycblt.musikr.cache.Cache
+import org.oxycblt.musikr.cache.CacheResult
+import org.oxycblt.musikr.cover.Cover
+import org.oxycblt.musikr.cover.CoverResult
+import org.oxycblt.musikr.cover.Covers
 import org.oxycblt.musikr.fs.MusicLocation
 import org.oxycblt.musikr.fs.device.DeviceDirectory
 import org.oxycblt.musikr.fs.device.DeviceFile
 import org.oxycblt.musikr.fs.device.DeviceFiles
 import org.oxycblt.musikr.fs.device.DeviceNode
-import org.oxycblt.musikr.playlist.PlaylistFile
 import org.oxycblt.musikr.playlist.db.StoredPlaylists
 import org.oxycblt.musikr.playlist.m3u.M3U
 
 internal interface ExploreStep {
-    fun explore(locations: List<MusicLocation>): Flow<ExploreNode>
+    fun explore(locations: List<MusicLocation>): Flow<Explored>
 
     companion object {
         fun from(context: Context, storage: Storage): ExploreStep =
-            ExploreStepImpl(DeviceFiles.from(context), storage.storedPlaylists)
+            ExploreStepImpl(
+                DeviceFiles.from(context), storage.cache, storage.covers, storage.storedPlaylists)
     }
 }
 
 private class ExploreStepImpl(
     private val deviceFiles: DeviceFiles,
+    private val cache: Cache,
+    private val covers: Covers<out Cover>,
     private val storedPlaylists: StoredPlaylists
 ) : ExploreStep {
-    override fun explore(locations: List<MusicLocation>): Flow<ExploreNode> {
+    @OptIn(ExperimentalCoroutinesApi::class)
-        val audios =
+    override fun explore(locations: List<MusicLocation>): Flow<Explored> {
+        val addingMs = System.currentTimeMillis()
+        return merge(
             deviceFiles
                 .explore(locations.asFlow())
                 .flattenFilter { it.mimeType.startsWith("audio/") || it.mimeType == M3U.MIME_TYPE }
+                .distribute(8)
+                .distributedMap { file ->
+                    val cachedSong =
+                        when (val cacheResult = cache.read(file)) {
+                            is CacheResult.Hit -> cacheResult.song
+                            is CacheResult.Stale ->
+                                return@distributedMap NewSong(cacheResult.file, cacheResult.addedMs)
+                            is CacheResult.Miss ->
+                                return@distributedMap NewSong(cacheResult.file, addingMs)
+                        }
+                    val cover =
+                        cachedSong.coverId?.let { coverId ->
+                            when (val coverResult = covers.obtain(coverId)) {
+                                is CoverResult.Hit -> coverResult.cover
+                                else ->
+                                    return@distributedMap NewSong(
+                                        cachedSong.file, cachedSong.addedMs)
+                            }
+                        }
+                    RawSong(
+                        cachedSong.file,
+                        cachedSong.properties,
+                        cachedSong.tags,
+                        cover,
+                        cachedSong.addedMs)
+                }
+                .flattenMerge()
                 .flowOn(Dispatchers.IO)
-                .buffer()
+                .buffer(),
-        val playlists =
             flow { emitAll(storedPlaylists.read().asFlow()) }
-                .map { ExploreNode.Playlist(it) }
+                .map { RawPlaylist(it) }
                 .flowOn(Dispatchers.IO)
-                .buffer()
+                .buffer())
-        return merge(audios, playlists)
     }
 
     @OptIn(ExperimentalCoroutinesApi::class)
-    private fun Flow<DeviceNode>.flattenFilter(block: (DeviceFile) -> Boolean): Flow<ExploreNode> =
+    private fun Flow<DeviceNode>.flattenFilter(block: (DeviceFile) -> Boolean): Flow<DeviceFile> =
         flow {
             collect {
-                val recurse = mutableListOf<Flow<ExploreNode>>()
+                val recurse = mutableListOf<Flow<DeviceFile>>()
                 when {
-                    it is DeviceFile && block(it) -> emit(ExploreNode.Audio(it))
+                    it is DeviceFile && block(it) -> emit(it)
                     it is DeviceDirectory -> recurse.add(it.children.flattenFilter(block))
                     else -> {}
                 }
@@ -82,9 +116,3 @@ private class ExploreStepImpl(
             }
         }
 }
-
-internal sealed interface ExploreNode {
-    data class Audio(val file: DeviceFile) : ExploreNode
-
-    data class Playlist(val file: PlaylistFile) : ExploreNode
-}

musikr/src/main/java/org/oxycblt/musikr/pipeline/ExtractStep.kt

@@ -19,181 +19,63 @@
 package org.oxycblt.musikr.pipeline
 
 import android.content.Context
-import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.ExperimentalCoroutinesApi
-import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.flattenMerge
-import kotlinx.coroutines.flow.flowOn
-import kotlinx.coroutines.flow.map
-import kotlinx.coroutines.flow.mapNotNull
-import kotlinx.coroutines.flow.merge
 import kotlinx.coroutines.flow.onCompletion
-import kotlinx.coroutines.withContext
 import org.oxycblt.musikr.Storage
-import org.oxycblt.musikr.cache.Cache
+import org.oxycblt.musikr.cache.CachedSong
-import org.oxycblt.musikr.cache.CacheResult
+import org.oxycblt.musikr.cache.MutableCache
 import org.oxycblt.musikr.cover.Cover
 import org.oxycblt.musikr.cover.CoverResult
 import org.oxycblt.musikr.cover.MutableCovers
-import org.oxycblt.musikr.fs.device.DeviceFile
 import org.oxycblt.musikr.metadata.MetadataExtractor
-import org.oxycblt.musikr.metadata.Properties
-import org.oxycblt.musikr.playlist.PlaylistFile
-import org.oxycblt.musikr.tag.parse.ParsedTags
 import org.oxycblt.musikr.tag.parse.TagParser
 
 internal interface ExtractStep {
-    fun extract(nodes: Flow<ExploreNode>): Flow<ExtractedMusic>
+    fun extract(nodes: Flow<Explored>): Flow<Extracted>
 
     companion object {
         fun from(context: Context, storage: Storage): ExtractStep =
             ExtractStepImpl(
-                context,
+                MetadataExtractor.from(context), TagParser.new(), storage.cache, storage.covers)
-                MetadataExtractor.new(),
-                TagParser.new(),
-                storage.cache,
-                storage.storedCovers)
     }
 }
 
 private class ExtractStepImpl(
-    private val context: Context,
     private val metadataExtractor: MetadataExtractor,
     private val tagParser: TagParser,
-    private val cacheFactory: Cache.Factory,
+    private val cache: MutableCache,
     private val covers: MutableCovers<out Cover>
 ) : ExtractStep {
     @OptIn(ExperimentalCoroutinesApi::class)
-    override fun extract(nodes: Flow<ExploreNode>): Flow<ExtractedMusic> {
+    override fun extract(nodes: Flow<Explored>): Flow<Extracted> {
-        val cache = cacheFactory.open()
+        val exclude = mutableListOf<CachedSong>()
-        val addingMs = System.currentTimeMillis()
+        return nodes
-        val filterFlow =
+            .distribute(8)
-            nodes.divert {
+            .distributedMap {
                 when (it) {
-                    is ExploreNode.Audio -> Divert.Right(it.file)
+                    is RawSong -> it
-                    is ExploreNode.Playlist -> Divert.Left(it.file)
+                    is RawPlaylist -> it
-                }
+                    is NewSong -> {
-            }
+                        val metadata =
-        val audioNodes = filterFlow.right
+                            metadataExtractor.extract(it.file) ?: return@distributedMap InvalidSong
-        val playlistNodes = filterFlow.left.map { ExtractedMusic.Valid.Playlist(it) }
+                        val tags = tagParser.parse(metadata)
-
+                        val cover =
-        // First distribute audio nodes for parallel cache reading
+                            when (val result = covers.create(it.file, metadata)) {
-        val readDistributedFlow = audioNodes.distribute(8)
+                                is CoverResult.Hit -> result.cover
-        val cacheResults =
+                                else -> null
-            readDistributedFlow.flows
-                .map { flow ->
-                    flow
-                        .map { wrap(it) { file -> cache.read(file, covers) } }
-                        .flowOn(Dispatchers.IO)
-                        .buffer(Channel.UNLIMITED)
-                }
-                .flattenMerge()
-                .buffer(Channel.UNLIMITED)
-
-        // Divert cache hits and misses
-        val cacheFlow =
-            cacheResults.divert {
-                when (it) {
-                    is CacheResult.Hit -> Divert.Left(it.song)
-                    is CacheResult.Miss -> Divert.Right(it.file)
-                }
-            }
-
-        // Cache hits can be directly converted to valid songs
-        val cachedSongs = cacheFlow.left.map { ExtractedMusic.Valid.Song(it) }
-
-        // Process uncached files in parallel
-        val uncachedFiles = cacheFlow.right
-        val processingDistributedFlow = uncachedFiles.distribute(8)
-
-        // Process each uncached file in parallel flows
-        val processedSongs =
-            processingDistributedFlow.flows
-                .map { flow ->
-                    flow
-                        .mapNotNull { file ->
-                            wrap(file) { f ->
-                                withContext(Dispatchers.IO) {
-                                        context.contentResolver.openFileDescriptor(f.uri, "r")
-                                    }
-                                    ?.use {
-                                        val extractedMetadata = metadataExtractor.extract(file, it)
-
-                                        if (extractedMetadata != null) {
-                                            val tags = tagParser.parse(extractedMetadata)
-                                            val cover =
-                                                when (val result =
-                                                    covers.create(f, extractedMetadata)) {
-                                                    is CoverResult.Hit -> result.cover
-                                                    else -> null
-                                                }
-                                            val rawSong =
-                                                RawSong(
-                                                    f,
-                                                    extractedMetadata.properties,
-                                                    tags,
-                                                    cover,
-                                                    addingMs)
-                                            cache.write(rawSong)
-
-                                            ExtractedMusic.Valid.Song(rawSong)
-                                        } else {
-                                            ExtractedMusic.Invalid
-                                        }
-                                    }
                             }
-                        }
+                        val cachedSong =
-                        .flowOn(Dispatchers.IO)
+                            CachedSong(it.file, metadata.properties, tags, cover?.id, it.addedMs)
-                        .buffer(Channel.UNLIMITED)
+                        cache.write(cachedSong)
-                }
+                        exclude.add(cachedSong)
-                .flattenMerge()
+                        val rawSong = RawSong(it.file, metadata.properties, tags, cover, it.addedMs)
-                .buffer(Channel.UNLIMITED)
+                        rawSong
-
+                    }
-        // Separate valid processed songs from invalid ones
-        val processedFlow =
-            processedSongs.divert {
-                when (it) {
-                    is ExtractedMusic.Valid.Song -> Divert.Left(it)
-                    is ExtractedMusic.Invalid -> Divert.Right(it)
-                    else -> Divert.Right(ExtractedMusic.Invalid)
                 }
             }
-
+            .flattenMerge()
-        val processedValidSongs = processedFlow.left
+            .onCompletion { cache.cleanup(exclude) }
-        val invalidSongs = processedFlow.right
-
-        val merged =
-            merge(
-                filterFlow.manager,
-                readDistributedFlow.manager,
-                cacheFlow.manager,
-                processingDistributedFlow.manager,
-                processedFlow.manager,
-                cachedSongs,
-                processedValidSongs,
-                invalidSongs,
-                playlistNodes)
-
-        return merged.onCompletion { cache.finalize() }
     }
 }
-
-internal data class RawSong(
-    val file: DeviceFile,
-    val properties: Properties,
-    val tags: ParsedTags,
-    val cover: Cover?,
-    val addedMs: Long
-)
-
-internal sealed interface ExtractedMusic {
-    sealed interface Valid : ExtractedMusic {
-        data class Song(val song: RawSong) : Valid
-
-        data class Playlist(val file: PlaylistFile) : Valid
-    }
-
-    data object Invalid : ExtractedMusic
-}

musikr/src/main/java/org/oxycblt/musikr/pipeline/FlowUtil.kt

@@ -65,7 +65,7 @@ internal class DistributedFlow<T>(val manager: Flow<Nothing>, val flows: Flow<Fl
  * Note that this function requires the "manager" flow to be consumed alongside the split flows in
  * order to function. Without this, all of the newly split flows will simply block.
  */
-internal fun <T> Flow<T>.distribute(n: Int): DistributedFlow<T> {
+internal fun <T> Flow<T>.distribute(n: Int): Flow<Flow<T>> {
     val posChannels = List(n) { Channel<T>(Channel.UNLIMITED) }
     val managerFlow =
         flow<Nothing> {
@@ -77,6 +77,42 @@ internal fun <T> Flow<T>.distribute(n: Int): DistributedFlow<T> {
                 channel.close()
             }
         }
-    val hotFlows = posChannels.asFlow().map { it.receiveAsFlow() }
+    return (posChannels.map { it.receiveAsFlow() } + managerFlow).asFlow()
-    return DistributedFlow(managerFlow, hotFlows)
+}
+
+internal fun <T, R> Flow<Flow<T>>.distributedMap(transform: suspend (T) -> R): Flow<Flow<R>> =
+    flow {
+        collect { innerFlow -> emit(innerFlow.tryMap(transform)) }
+    }
+
+internal fun <T, R> Flow<T>.tryMap(transform: suspend (T) -> R): Flow<R> = flow {
+    collect { value ->
+        try {
+            emit(transform(value))
+        } catch (e: Exception) {
+            throw PipelineException(value, e)
+        }
+    }
+}
+
+internal fun <T, R> Flow<T>.tryMapNotNull(transform: suspend (T) -> R?): Flow<R> = flow {
+    collect { value ->
+        try {
+            transform(value)?.let { emit(it) }
+        } catch (e: Exception) {
+            throw PipelineException(value, e)
+        }
+    }
+}
+
+internal fun <A, T> Flow<T>.tryFold(initial: A, operation: suspend (A, T) -> A): Flow<A> = flow {
+    var accumulator = initial
+    collect { value ->
+        try {
+            accumulator = operation(accumulator, value)
+            emit(accumulator)
+        } catch (e: Exception) {
+            throw PipelineException(value, e)
+        }
+    }
 }

musikr/src/main/java/org/oxycblt/musikr/pipeline/PipelineException.kt

@@ -18,71 +18,9 @@
  
 package org.oxycblt.musikr.pipeline
 
-import org.oxycblt.musikr.fs.device.DeviceFile
+class PipelineException(whileProcessing: Any?, val error: Exception) : Exception() {
-import org.oxycblt.musikr.playlist.PlaylistFile
-import org.oxycblt.musikr.playlist.interpret.PrePlaylist
-import org.oxycblt.musikr.tag.interpret.PreSong
-
-class PipelineException(val processing: WhileProcessing, val error: Exception) : Exception() {
     override val cause = error
 
-    override val message = "Error while processing ${processing}: ${error.stackTraceToString()}"
+    override val message =
+        "Error while processing ${whileProcessing}: ${error.stackTraceToString()}"
 }
-
-sealed interface WhileProcessing {
-    class AFile internal constructor(private val file: DeviceFile) : WhileProcessing {
-        override fun toString() = "File @ ${file.path}"
-    }
-
-    class ARawSong internal constructor(private val rawSong: RawSong) : WhileProcessing {
-        override fun toString() = "Raw Song @ ${rawSong.file.path}"
-    }
-
-    class APlaylistFile internal constructor(private val playlist: PlaylistFile) : WhileProcessing {
-        override fun toString() = "Playlist File @ ${playlist.name}"
-    }
-
-    class APreSong internal constructor(private val preSong: PreSong) : WhileProcessing {
-        override fun toString() = "Pre Song @ ${preSong.path}"
-    }
-
-    class APrePlaylist internal constructor(private val prePlaylist: PrePlaylist) :
-        WhileProcessing {
-        override fun toString() = "Pre Playlist @ ${prePlaylist.name}"
-    }
-}
-
-internal suspend fun <R> wrap(file: DeviceFile, block: suspend (DeviceFile) -> R): R =
-    try {
-        block(file)
-    } catch (e: Exception) {
-        throw PipelineException(WhileProcessing.AFile(file), e)
-    }
-
-internal suspend fun <R> wrap(song: RawSong, block: suspend (RawSong) -> R): R =
-    try {
-        block(song)
-    } catch (e: Exception) {
-        throw PipelineException(WhileProcessing.ARawSong(song), e)
-    }
-
-internal suspend fun <R> wrap(file: PlaylistFile, block: suspend (PlaylistFile) -> R): R =
-    try {
-        block(file)
-    } catch (e: Exception) {
-        throw PipelineException(WhileProcessing.APlaylistFile(file), e)
-    }
-
-internal suspend fun <R> wrap(song: PreSong, block: suspend (PreSong) -> R): R =
-    try {
-        block(song)
-    } catch (e: Exception) {
-        throw PipelineException(WhileProcessing.APreSong(song), e)
-    }
-
-internal suspend fun <R> wrap(playlist: PrePlaylist, block: suspend (PrePlaylist) -> R): R =
-    try {
-        block(playlist)
-    } catch (e: Exception) {
-        throw PipelineException(WhileProcessing.APrePlaylist(playlist), e)
-    }

musikr/src/main/java/org/oxycblt/musikr/pipeline/PipelineItem.kt

@@ -0,0 +1,57 @@
+/*
+ * Copyright (c) 2025 Auxio Project
+ * PipelineItem.kt is part of Auxio.
+ *
+ * This program 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.
+ *
+ * This program 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 this program.  If not, see <https://www.gnu.org/licenses/>.
+ */
+ 
+package org.oxycblt.musikr.pipeline
+
+import org.oxycblt.musikr.cover.Cover
+import org.oxycblt.musikr.fs.device.DeviceFile
+import org.oxycblt.musikr.metadata.Properties
+import org.oxycblt.musikr.playlist.PlaylistFile
+import org.oxycblt.musikr.tag.parse.ParsedTags
+
+internal sealed interface PipelineItem
+
+internal sealed interface Incomplete : PipelineItem
+
+internal sealed interface Complete : PipelineItem
+
+internal sealed interface Explored : PipelineItem {
+    sealed interface New : Explored, Incomplete
+
+    sealed interface Known : Explored, Complete
+}
+
+internal data class NewSong(val file: DeviceFile, val addedMs: Long) : Explored.New
+
+internal sealed interface Extracted : PipelineItem {
+    sealed interface Valid : Complete, Extracted
+
+    sealed interface Invalid : Extracted
+}
+
+data object InvalidSong : Extracted.Invalid
+
+internal data class RawPlaylist(val file: PlaylistFile) : Explored.Known, Extracted.Valid
+
+internal data class RawSong(
+    val file: DeviceFile,
+    val properties: Properties,
+    val tags: ParsedTags,
+    val cover: Cover?,
+    val addedMs: Long
+) : Explored.Known, Extracted.Valid

musikr/src/main/java/org/oxycblt/musikr/tag/parse/ParsedTags.kt

@@ -20,7 +20,7 @@ package org.oxycblt.musikr.tag.parse
 
 import org.oxycblt.musikr.tag.Date
 
-internal data class ParsedTags(
+data class ParsedTags(
     val durationMs: Long,
     val replayGainTrackAdjustment: Float? = null,
     val replayGainAlbumAdjustment: Float? = null,