Download anything that joins the queue, straight away

Peter asked for it, and it is what makes the YouTube fix comfortable: those
episodes have to be fetched before they play at all, so having them already on
disk by the time he taps Play turns a thirty-second wait into none.

The notification lives on QueueRepository rather than at each call site because
the queue is the one choke point every route goes through -- a rule, a drip
release, a tap on Play, the share sheet, a drag. It reports only ids that were
not in the queue a moment ago, so reordering, removing and advancing say
nothing, and re-adding something already queued (which moves it) says nothing
either.

QueuedDownloads decides what actually needs fetching. Skipping what is already
downloaded or already in flight is the part worth testing: a second download of
bytes on disk costs real data for nothing. A FAILED episode is deliberately
retried, since re-queueing something is a reasonable way to ask for that.

The one-off job carries no network constraint, per Peter's choice that this
happens immediately on any network. The periodic sweep keeps its wifi
constraint. Worth being clear about the implication he accepted: a rule that
catches several episodes at once will now pull them over mobile data, capped
only by DownloadDrainer's ten per run.

Verified on a device: queueing an episode from the episode screen took it from
NONE to DOWNLOADING within seconds and to DOWNLOADED with a file on disk,
rather than waiting for the two-hourly wifi job.

No permission changes, no schema change. 403 unit tests green, detekt and lint
clean, coverage 65.16 -> 65.22.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018AwnALZL6bD4VBtATDVv3B

5 files changed, 250 insertions, 8 deletions

diff --git a/app/src/main/java/org/terst/listen/data/Repositories.kt b/app/src/main/java/org/terst/listen/data/Repositories.kt
index 2a2d74e..da3c8d6 100644
--- a/app/src/main/java/org/terst/listen/data/Repositories.kt
+++ b/app/src/main/java/org/terst/listen/data/Repositories.kt
@@ -15,6 +15,8 @@ import org.terst.listen.queue.DripReleaser
 import org.terst.listen.queue.QueueOrdering
 import org.terst.listen.queue.RetentionPolicy
 import org.terst.listen.queue.RuleApplier
+import org.terst.listen.work.QueuedDownloads
+import org.terst.listen.work.Scheduling
 import java.io.File
 import java.util.concurrent.TimeUnit
 
@@ -88,7 +90,15 @@ class AppGraph(private val context: Context) {
         RetentionPolicy(gracePeriodMs = TimeUnit.DAYS.toMillis(1))
     }
 
-    val queue: QueueRepository by lazy { QueueRepository(queueDao) }
+    private val queuedDownloads: QueuedDownloads by lazy { QueuedDownloads(episodes) }
+
+    val queue: QueueRepository by lazy {
+        QueueRepository(queueDao) { joined ->
+            // Anything queued is meant to be listened to, so it is fetched for offline
+            // straight away rather than at the next periodic sweep.
+            if (queuedDownloads.request(joined) > 0) Scheduling.downloadNow(context)
+        }
+    }
 
     // Reuses the app's own OkHttpClient rather than Coil's default, so artwork fetches share
     // connection pooling and timeouts with feed refreshes and downloads.
@@ -105,7 +115,17 @@ class AppGraph(private val context: Context) {
 }
 
 /** Applies QueueOrdering's decisions to storage, keeping positions contiguous. */
-class QueueRepository(private val dao: QueueDao) {
+class QueueRepository(
+    private val dao: QueueDao,
+    /**
+     * Told which episodes have just joined the queue, so they can be downloaded.
+     *
+     * Here rather than at each call site because the queue is the one choke point every route
+     * goes through -- a rule, a drip release, a tap on Play, the share sheet. Fired with newly
+     * present ids only, so reordering, removing and advancing say nothing.
+     */
+    private val onQueued: suspend (List<String>) -> Unit = {},
+) {
 
     fun observe() = dao.observeQueue()
 
@@ -126,19 +146,27 @@ class QueueRepository(private val dao: QueueDao) {
      * Separate from [move]: a drag gesture reports intermediate positions continuously, and
      * only the final order should ever reach storage -- one write, not one per drag step.
      */
-    suspend fun reorder(episodeIds: List<String>) = dao.replaceAll(episodeIds)
+    suspend fun reorder(episodeIds: List<String>) = write(episodeIds, ids())
 
     /** Drops the finished episode and reports what plays next. */
     suspend fun advance(): String? {
         val current = ids()
         val (next, remaining) = QueueOrdering.advance(current)
-        dao.replaceAll(remaining)
+        write(remaining, before = current)
         return next
     }
 
     private suspend fun ids() = dao.getEntries().map { it.episodeId }
 
     private suspend fun rewrite(transform: (List<String>) -> List<String>) {
-        dao.replaceAll(transform(ids()))
+        val before = ids()
+        write(transform(before), before)
+    }
+
+    /** Writes the order and reports whatever was not in the queue a moment ago. */
+    private suspend fun write(after: List<String>, before: List<String>) {
+        dao.replaceAll(after)
+        val joined = after - before.toSet()
+        if (joined.isNotEmpty()) onQueued(joined)
     }
 }
diff --git a/app/src/main/java/org/terst/listen/work/QueuedDownloads.kt b/app/src/main/java/org/terst/listen/work/QueuedDownloads.kt
new file mode 100644
index 0000000..f9642ff
--- /dev/null
+++ b/app/src/main/java/org/terst/listen/work/QueuedDownloads.kt
@@ -0,0 +1,36 @@
+package org.terst.listen.work
+
+import org.terst.listen.data.DownloadState
+import org.terst.listen.data.EpisodeDao
+
+/**
+ * Marks episodes that have just joined the queue as wanted, for [DownloadDrainer] to collect.
+ *
+ * Anything queued is something intended for listening, so it gets fetched for offline. The part
+ * worth testing is what to leave alone: a second download of bytes already on disk, or of one
+ * already in flight, costs real data for nothing.
+ */
+class QueuedDownloads(private val episodes: EpisodeDao) {
+
+    /** Returns how many episodes were newly marked as wanted. */
+    suspend fun request(episodeIds: List<String>): Int {
+        val wanted = episodeIds
+            .mapNotNull { episodes.getById(it) }
+            .filterNot { it.downloadState in SETTLED }
+
+        wanted.forEach { episodes.updateDownload(it.id, DownloadState.QUEUED, null, null) }
+        return wanted.size
+    }
+
+    private companion object {
+        /**
+         * States that need nothing doing. FAILED is deliberately absent: re-queueing something
+         * is a reasonable way to ask for a download that did not work to be tried again.
+         */
+        val SETTLED = setOf(
+            DownloadState.DOWNLOADED,
+            DownloadState.DOWNLOADING,
+            DownloadState.QUEUED,
+        )
+    }
+}
diff --git a/app/src/main/java/org/terst/listen/work/Workers.kt b/app/src/main/java/org/terst/listen/work/Workers.kt
index 3f829ab..db896a6 100644
--- a/app/src/main/java/org/terst/listen/work/Workers.kt
+++ b/app/src/main/java/org/terst/listen/work/Workers.kt
@@ -4,6 +4,8 @@ import android.content.Context
 import androidx.work.Constraints
 import androidx.work.CoroutineWorker
 import androidx.work.ExistingPeriodicWorkPolicy
+import androidx.work.ExistingWorkPolicy
+import androidx.work.OneTimeWorkRequestBuilder
 import androidx.work.NetworkType
 import androidx.work.PeriodicWorkRequestBuilder
 import androidx.work.WorkManager
@@ -70,15 +72,31 @@ object Scheduling {
 
     private const val REFRESH_WORK = "listen-refresh"
     private const val DOWNLOAD_WORK = "listen-download"
+    private const val DOWNLOAD_NOW_WORK = "listen-download-now"
     private const val RETENTION_WORK = "listen-retention"
     private const val REFRESH_HOURS = 6L
     private const val DOWNLOAD_HOURS = 2L
     private const val RETENTION_HOURS = 12L
 
     /**
-     * Both jobs are KEEP rather than UPDATE, so the schedule survives an app restart instead of
-     * resetting its interval every launch -- which on a frequently-opened app means the periodic
-     * work never actually fires.
+     * Fetches what is in the queue now, rather than at the next periodic sweep.
+     *
+     * No network constraint, unlike the periodic job: Peter asked for anything added to the
+     * queue to download immediately, mobile data included. REPLACE so a burst of queueing --
+     * a rule catching several episodes at once -- collapses into one run.
+     */
+    fun downloadNow(context: Context) {
+        WorkManager.getInstance(context).enqueueUniqueWork(
+            DOWNLOAD_NOW_WORK,
+            ExistingWorkPolicy.REPLACE,
+            OneTimeWorkRequestBuilder<DownloadWorker>().build(),
+        )
+    }
+
+    /**
+     * The periodic jobs are KEEP rather than UPDATE, so the schedule survives an app restart
+     * instead of resetting its interval every launch -- which on a frequently-opened app means
+     * the periodic work never actually fires.
      */
     fun schedule(context: Context) {
         val manager = WorkManager.getInstance(context)
diff --git a/app/src/test/java/org/terst/listen/data/QueueRepositoryTest.kt b/app/src/test/java/org/terst/listen/data/QueueRepositoryTest.kt
index d880e57..11b080e 100644
--- a/app/src/test/java/org/terst/listen/data/QueueRepositoryTest.kt
+++ b/app/src/test/java/org/terst/listen/data/QueueRepositoryTest.kt
@@ -153,4 +153,61 @@ class QueueRepositoryTest {
         assertEquals(listOf("c", "a", "b"), ids())
         assertContiguous()
     }
+
+    // --- telling someone what just joined the queue -------------------------------------------
+
+    /**
+     * Anything entering the queue is something Peter intends to listen to, so it gets
+     * downloaded. The queue is the one choke point every route goes through, which is why the
+     * notification lives here rather than being repeated at each call site.
+     */
+    @Test
+    fun `appending reports the new episode`() = runTest {
+        val queued = mutableListOf<List<String>>()
+        val repo = QueueRepository(dao) { queued += it }
+
+        repo.append("a")
+
+        assertEquals(listOf(listOf("a")), queued)
+    }
+
+    @Test
+    fun `playing next and playing now both report`() = runTest {
+        val queued = mutableListOf<List<String>>()
+        val repo = QueueRepository(dao) { queued += it }
+
+        repo.playNext("a")
+        repo.playNow("b")
+
+        assertEquals(listOf(listOf("a"), listOf("b")), queued)
+    }
+
+    /** Re-adding something already queued moves it. Nothing joined, so nothing to fetch. */
+    @Test
+    fun `re-adding a queued episode reports nothing`() = runTest {
+        val queued = mutableListOf<List<String>>()
+        val repo = QueueRepository(dao) { queued += it }
+        repo.append("a")
+        queued.clear()
+
+        repo.append("a")
+
+        assertEquals(emptyList<List<String>>(), queued)
+    }
+
+    @Test
+    fun `removing, reordering and advancing report nothing`() = runTest {
+        val queued = mutableListOf<List<String>>()
+        val repo = QueueRepository(dao) { queued += it }
+        repo.append("a")
+        repo.append("b")
+        queued.clear()
+
+        repo.move(0, 1)
+        repo.reorder(listOf("b", "a"))
+        repo.remove("a")
+        repo.advance()
+
+        assertEquals(emptyList<List<String>>(), queued)
+    }
 }
diff --git a/app/src/test/java/org/terst/listen/work/QueuedDownloadsTest.kt b/app/src/test/java/org/terst/listen/work/QueuedDownloadsTest.kt
new file mode 100644
index 0000000..244da4c
--- /dev/null
+++ b/app/src/test/java/org/terst/listen/work/QueuedDownloadsTest.kt
@@ -0,0 +1,103 @@
+package org.terst.listen.work
+
+import kotlinx.coroutines.test.runTest
+import org.junit.Assert.assertEquals
+import org.junit.Before
+import org.junit.Test
+import org.terst.listen.data.DownloadState
+import org.terst.listen.data.EpisodeEntity
+import org.terst.listen.data.FakeEpisodeDao
+
+/**
+ * Marking newly queued episodes as wanted.
+ *
+ * Anything Peter puts in the queue is something he intends to listen to, so it gets fetched for
+ * offline. The interesting part is what *not* to re-request: a second download of a file
+ * already on disk costs real data for nothing.
+ */
+class QueuedDownloadsTest {
+
+    private lateinit var episodes: FakeEpisodeDao
+    private lateinit var requests: QueuedDownloads
+
+    @Before
+    fun setUp() {
+        episodes = FakeEpisodeDao()
+        requests = QueuedDownloads(episodes)
+    }
+
+    private fun episode(id: String, state: DownloadState, localPath: String? = null) {
+        episodes.rows += EpisodeEntity(
+            id = id,
+            subscriptionId = "s1",
+            guid = id,
+            title = id,
+            mediaUrl = "https://cdn/$id.mp3",
+            mediaType = "audio/mpeg",
+            discoveredAtMs = 0,
+            downloadState = state,
+            localPath = localPath,
+        )
+    }
+
+    private fun stateOf(id: String) = episodes.rows.first { it.id == id }.downloadState
+
+    @Test
+    fun `an episode with nothing downloaded is wanted`() = runTest {
+        episode("a", DownloadState.NONE)
+
+        assertEquals(1, requests.request(listOf("a")))
+        assertEquals(DownloadState.QUEUED, stateOf("a"))
+    }
+
+    /** Already on disk. Asking again would spend the data twice for the same bytes. */
+    @Test
+    fun `an episode already downloaded is left alone`() = runTest {
+        episode("a", DownloadState.DOWNLOADED, localPath = "/audio/a.mp3")
+
+        assertEquals(0, requests.request(listOf("a")))
+        assertEquals(DownloadState.DOWNLOADED, stateOf("a"))
+    }
+
+    /** Already in flight -- re-marking it QUEUED would let the drainer start a second copy. */
+    @Test
+    fun `an episode being downloaded is left alone`() = runTest {
+        episode("a", DownloadState.DOWNLOADING)
+
+        assertEquals(0, requests.request(listOf("a")))
+        assertEquals(DownloadState.DOWNLOADING, stateOf("a"))
+    }
+
+    /** Re-queueing something is a reasonable way to ask for a failed download again. */
+    @Test
+    fun `a failed episode is wanted again`() = runTest {
+        episode("a", DownloadState.FAILED)
+
+        assertEquals(1, requests.request(listOf("a")))
+        assertEquals(DownloadState.QUEUED, stateOf("a"))
+    }
+
+    @Test
+    fun `an already queued episode is not counted twice`() = runTest {
+        episode("a", DownloadState.QUEUED)
+
+        assertEquals(0, requests.request(listOf("a")))
+    }
+
+    @Test
+    fun `an episode that has gone is skipped rather than throwing`() = runTest {
+        assertEquals(0, requests.request(listOf("gone")))
+    }
+
+    @Test
+    fun `a batch marks each one that needs it`() = runTest {
+        episode("a", DownloadState.NONE)
+        episode("b", DownloadState.DOWNLOADED, localPath = "/audio/b.mp3")
+        episode("c", DownloadState.FAILED)
+
+        assertEquals(2, requests.request(listOf("a", "b", "c")))
+        assertEquals(DownloadState.QUEUED, stateOf("a"))
+        assertEquals(DownloadState.DOWNLOADED, stateOf("b"))
+        assertEquals(DownloadState.QUEUED, stateOf("c"))
+    }
+}