diff options
| author | Peter Stone <thepeterstone@gmail.com> | 2026-10-05 09:54:59 +0000 |
|---|---|---|
| committer | Peter Stone <thepeterstone@gmail.com> | 2026-10-05 09:54:59 +0000 |
| commit | 217c982403a105e986fb7ce13b922d9f08583632 (patch) | |
| tree | 38f2d3a9556b3dfa1b835b9b2c1f7efcb547e719 | |
| parent | 7cc88394458e9d03e3a6a6e58cdff9c8dd4ad7a3 (diff) | |
Move skip detection to a server, and find that it detects ads
The server is Peter's design and it works: listend polls the phone's feed
list, mirrors the newest 8 episodes per show, detects across all of them at
once, and serves the offsets *and the audio they belong to*. Deployed on
tungsten behind Apache at https://listen.terst.org. FingerprintWorker and
MidrollWorker are deleted -- the battery drain is gone because the work is.
Serving the audio is load-bearing. Dynamic ad insertion turned out to be on
every episode: Grass Apes declares 22:05 and the server was served 28:18. An
offset measured on one copy is minutes wrong for another, so the phone has to
play the bytes the detector examined. The offsets land in the same columns the
on-device detector wrote, so playback needed no changes and there is no
migration.
**The headline finding is a failure, and it is proved rather than suspected.**
Detection reported the intro confirmed on 8 of 8 Radiolab episodes and passed
its self-check. It had locked onto an advertisement. The chosen segment was
searched for in an ad-free copy of the same episode fetched hours earlier:
r = 0.0621. It is in the ad break, not the programme.
The confirmation threshold assumes three *independently fetched* episodes.
Mirroring 8 within one minute from the same ad server makes them maximally
dependent -- whatever campaign is running is in all of them, identical -- and
SelfCheck cannot catch it, because the segments genuinely are the same audio.
Moving to a server fixed the cost problem and broke the independence the
safety mechanism rests on. The published offsets are head-window matches, so
they happen to skip pre-roll advertising, which is useful by accident and will
move whenever the campaign rotates. It is not intro detection and is not
described as such anywhere.
The fix, not made: record when each episode was fetched and require the
agreeing episodes to span more than one fetch occasion. Reasoning, and an
argument for detecting the ads directly instead, in docs/detection-findings.md.
Also recorded there: two tooling traps that each cost real time. `android-ui
tap` can land on the soft keyboard, because the dump reports layout
coordinates the IME does not move -- a Save button and a text field both
appeared tapped and neither fired. And the server's token is alphanumeric now
because `adb shell input text` ate the leading "-" of a base64url one and
produced a silent 401 that looked like broken wiring.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_018AwnALZL6bD4VBtATDVv3B
30 files changed, 3341 insertions, 116 deletions
diff --git a/.coverage-baseline b/.coverage-baseline index 423c6ac..6e6545f 100644 --- a/.coverage-baseline +++ b/.coverage-baseline @@ -1 +1 @@ -93.41 +93.64 @@ -266,6 +266,46 @@ by default and correct; it is simply waiting for a publisher who marks breaks. seeks approximate on a VBR file, and the playback position is the crown jewel. Full numbers in `docs/detection-findings.md`. +- **Detection runs on a server now, and `server/` is part of this repo.** Go, deployed on + tungsten as `listend.service`, behind Apache at https://listen.terst.org with a Let's Encrypt + cert. Data in `/site/listen.terst.org/data` (`state.json`, `audio/`, `token`). It polls the + phone's feed list, mirrors the newest 8 episodes per show, detects across all of them at once, + and serves the offsets *and the audio they belong to*. `FingerprintWorker` and `MidrollWorker` + are gone from the app. + + **Serving the audio is load-bearing, not a convenience.** Dynamic ad insertion is on every + episode: Radiolab's Grass Apes declares 22:05 and the server was served 28:18. An offset + measured on one copy is minutes wrong for another, so the phone has to play the bytes the + detector examined. `EpisodeDownloader` prefers the server and falls back to the publisher -- + the fallback matters, because skips improve playback rather than being a precondition for it. + + **The offsets go into the same columns the on-device detector wrote**, which is why playback + needed no changes: `effectiveIntroMs`, `effectiveOutroMs` and `PlayheadPolicy` already read + them. No migration either. + +- **The confirmation threshold does not survive simultaneous mirroring, and currently detects + ads rather than intros.** Proved, not suspected: the segment confirmed on 8 of 8 Radiolab + episodes was searched for in an ad-free copy of the same episode fetched hours earlier and + scored **r = 0.0621**. It is in the ad break, not the programme. Mirroring 8 episodes within + one minute from the same ad server makes them maximally dependent, so whatever campaign is + running is in all of them, identical -- and `SelfCheck` cannot catch it, because the segments + genuinely *are* the same audio. Do not describe the current offsets as intro detection. The + fix is to record fetch times and require the agreeing episodes to span more than one fetch + occasion; full reasoning in `docs/detection-findings.md`. + +- **`android-ui tap` can land on the soft keyboard.** The dump reports the layout's own + coordinates, which the IME does not move, so an element listed at y=1643 can be underneath the + keyboard -- taps on it silently do nothing, which reads exactly like a broken handler. Found + the hard way: a Save button and a text field both appeared tapped and neither fired. Check + `adb shell dumpsys input_method | grep mInputShown` and dismiss with `input keyevent + KEYCODE_BACK` (not `KEYCODE_ESCAPE`, which does not close it) before tapping anything low on a + screen that has a text field. + +- **The server's API token is alphanumeric on purpose.** It is typed into a phone by hand, and + the first generated token was base64url -- whose leading `-` was eaten by `adb shell input + text` treating it as a flag, producing a silent 401 that looked like broken wiring. The + alphabet also omits l, I, 1, O and 0. + - **Android Auto is verified in a real car**, as of 2026-09-28. It appears, and connects. The notification Peter saw on first connect was Android Auto's own new-media-app announcement, not Listen: the only notification this app can raise is the Media3 transport control, which diff --git a/app/src/main/java/org/terst/listen/data/Preferences.kt b/app/src/main/java/org/terst/listen/data/Preferences.kt index 1f7e14f..0c170a1 100644 --- a/app/src/main/java/org/terst/listen/data/Preferences.kt +++ b/app/src/main/java/org/terst/listen/data/Preferences.kt @@ -2,6 +2,7 @@ package org.terst.listen.data import android.content.Context import androidx.datastore.preferences.core.booleanPreferencesKey +import androidx.datastore.preferences.core.stringPreferencesKey import androidx.datastore.preferences.core.edit import androidx.datastore.preferences.preferencesDataStore import kotlinx.coroutines.flow.Flow @@ -31,7 +32,24 @@ class Preferences(private val context: Context) { context.store.edit { it[SHOW_PLAYED] = value } } + /** + * Where the skip server is, and the token for it. Blank means not configured, which is a + * working state: the app downloads from publishers and plays without detected skips, + * exactly as it did before the server existed. + */ + val serverUrl: Flow<String> = context.store.data.map { it[SERVER_URL] ?: "" } + val serverToken: Flow<String> = context.store.data.map { it[SERVER_TOKEN] ?: "" } + + suspend fun setServer(url: String, token: String) { + context.store.edit { + it[SERVER_URL] = url.trim() + it[SERVER_TOKEN] = token.trim() + } + } + private companion object { val SHOW_PLAYED = booleanPreferencesKey("show_played") + val SERVER_URL = stringPreferencesKey("server_url") + val SERVER_TOKEN = stringPreferencesKey("server_token") } } 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 1b174cf..e425ccf 100644 --- a/app/src/main/java/org/terst/listen/data/Repositories.kt +++ b/app/src/main/java/org/terst/listen/data/Repositories.kt @@ -3,7 +3,12 @@ package org.terst.listen.data import android.content.Context import coil3.ImageLoader import coil3.network.okhttp.OkHttpNetworkFetcherFactory +import kotlinx.coroutines.flow.first import okhttp3.OkHttpClient +import org.terst.listen.server.OkHttpSkipTransport +import org.terst.listen.server.ServerSkips +import org.terst.listen.server.SkipServer +import org.terst.listen.server.SkipServerConfig import org.terst.listen.download.ChaquopyAudioSource import org.terst.listen.download.EpisodeDownloader import org.terst.listen.feed.ChapterFetcher @@ -96,8 +101,30 @@ class AppGraph(private val context: Context) { subscriptions, episodes, FeedFetcher(client), - onDiscovered = { ruleApplier.apply(it) }, - probe = { url -> ChaquopyAudioSource(context).probe(url) }, + SubscriptionCollaborators( + onDiscovered = { ruleApplier.apply(it) }, + probe = { url -> ChaquopyAudioSource(context).probe(url) }, + onRefreshed = { feedUrls -> skipServer.pushSubscriptions(feedUrls) }, + ), + ) + } + + /** + * The skip server, which is where detection happens now. + * + * It moved off the phone because it could never work here: comparing several episodes' + * audio at once needs more memory than an Android process gets and more time than + * WorkManager's ten-minute ceiling allows, and the attempt burned the battery for weeks + * while confirming nothing. + */ + val skipServer: SkipServer by lazy { + SkipServer( + config = { + val url = preferences.serverUrl.first() + val token = preferences.serverToken.first() + if (url.isBlank() || token.isBlank()) null else SkipServerConfig(url, token) + }, + transport = OkHttpSkipTransport(client), ) } @@ -107,10 +134,7 @@ class AppGraph(private val context: Context) { episodes, libraryDir, ChaquopyAudioSource(context), - // Fingerprinting needs the file on disk to decode, so it can only start once a - // download actually lands -- one job per episode rather than a periodic sweep, - // since the file is right there and there is nothing to gain by waiting. - onDownloaded = { episode -> Scheduling.fingerprintNow(context, episode.id) }, + serverAudio = ServerSkips(skipServer, subscriptions, fingerprints, playbackState), ) } diff --git a/app/src/main/java/org/terst/listen/data/SubscriptionRepository.kt b/app/src/main/java/org/terst/listen/data/SubscriptionRepository.kt index 9b94c75..a66441a 100644 --- a/app/src/main/java/org/terst/listen/data/SubscriptionRepository.kt +++ b/app/src/main/java/org/terst/listen/data/SubscriptionRepository.kt @@ -30,26 +30,43 @@ private const val SAVED_TITLE = "Saved videos" private const val SAVED_FACEBOOK_FEED_URL = "listen:saved:facebook" private const val SAVED_FACEBOOK_TITLE = "Saved Facebook videos" -class SubscriptionRepository( - private val subscriptions: SubscriptionDao, - private val episodes: EpisodeDao, - private val fetcher: FeedFetcher, +/** + * Everything [SubscriptionRepository] tells the rest of the app about, bundled. + * + * Grouped rather than passed one by one so the constructor stays a readable, fixed size as more + * of the app comes to care what a refresh found -- the same reasoning as [RuleEngine]'s and + * [org.terst.listen.fingerprint.MidrollUpdater]'s own tuning values. + */ +data class SubscriptionCollaborators( /** * Told about episodes a refresh has just discovered, so the auto-queue rules can act on them. * - * A collaborator rather than a call into the queue, so this class never learns what a queue - * is. Deliberately not invoked by [subscribe] or [addSingleEpisode]: see their comments. + * Deliberately not invoked by `subscribe` or `addSingleEpisode`: see their comments. */ - private val onDiscovered: suspend (List<EpisodeEntity>) -> Unit = {}, + val onDiscovered: suspend (List<EpisodeEntity>) -> Unit = {}, /** * Asks the extractor what a video is, for the cases where its page will not say. * * Facebook serves no `og:` tags to a non-browser at all, so a shared video would otherwise * be filed as "Video 10153231379946729". yt-dlp knows, and is going to be run against the - * URL anyway. A collaborator rather than a direct call because it reaches the embedded - * Python, which cannot run in a unit test. + * URL anyway. + */ + val probe: suspend (String) -> ProbedVideo? = { null }, + /** + * Told the subscription list on every refresh, so the skip server knows which feeds to + * mirror and analyse. + * + * Fired from `refreshAll`, the choke point every refresh route passes through -- including + * the library's own refresh button, which does not go via a worker. */ - private val probe: suspend (String) -> ProbedVideo? = { null }, + val onRefreshed: suspend (List<String>) -> Unit = {}, +) + +class SubscriptionRepository( + private val subscriptions: SubscriptionDao, + private val episodes: EpisodeDao, + private val fetcher: FeedFetcher, + private val collaborators: SubscriptionCollaborators = SubscriptionCollaborators(), private val now: () -> Long = System::currentTimeMillis, ) { @@ -108,8 +125,15 @@ class SubscriptionRepository( } } - suspend fun refreshAll(): Map<String, RefreshOutcome> = - subscriptions.getAll().associate { it.id to refresh(it) } + suspend fun refreshAll(): Map<String, RefreshOutcome> { + val shows = subscriptions.getAll() + // The one place every refresh passes through, which is why the server is told here + // rather than in a worker: the refresh button in the library calls this directly and + // would otherwise never announce a new subscription. Failure is ignored on purpose -- + // the skip server improves playback rather than being a precondition for it. + collaborators.onRefreshed(shows.map { it.feedUrl }) + return shows.associate { it.id to refresh(it) } + } /** * Pulls a show's back catalogue by walking its RFC 5005 `prev-archive` chain. @@ -156,7 +180,7 @@ class SubscriptionRepository( val result = EpisodeMatcher.match(subscriptionId, stored, feed.episodes, now()) if (result.inserts.isNotEmpty()) episodes.insertAll(result.inserts) if (result.updates.isNotEmpty()) episodes.updateAll(result.updates) - if (report && result.inserts.isNotEmpty()) onDiscovered(result.inserts) + if (report && result.inserts.isNotEmpty()) collaborators.onDiscovered(result.inserts) return RefreshOutcome.Updated(result.inserts.size, result.updates.size) } @@ -192,7 +216,7 @@ class SubscriptionRepository( // Only when the page gave nothing: a title that was shared is the one that was meant, // and this costs a network round trip. - val probed = if (video.title.isNullOrBlank()) probe(video.watchUrl) else null + val probed = if (video.title.isNullOrBlank()) collaborators.probe(video.watchUrl) else null val guid = if (facebook) "fb:video:${video.videoId}" else "yt:video:${video.videoId}" val existing = episodes.getByGuid(container.id, guid) diff --git a/app/src/main/java/org/terst/listen/download/EpisodeDownloader.kt b/app/src/main/java/org/terst/listen/download/EpisodeDownloader.kt index ed8bf70..32180c4 100644 --- a/app/src/main/java/org/terst/listen/download/EpisodeDownloader.kt +++ b/app/src/main/java/org/terst/listen/download/EpisodeDownloader.kt @@ -9,6 +9,7 @@ import org.terst.listen.data.EpisodeDao import org.terst.listen.data.EpisodeEntity import org.terst.listen.data.ProbedVideo import org.terst.listen.data.SubscriptionKind +import org.terst.listen.server.ServerEpisode import java.io.File import java.io.IOException @@ -20,18 +21,35 @@ import java.io.IOException * play with no indication why -- so the kind decides the route, and there is no fallback * between them. */ +/** + * The skip server's part in a download: where the audio should come from, and what to write + * down once it has arrived. + * + * Behind an interface because the decision belongs to the download path while the HTTP and the + * database writes do not. + */ +interface ServerAudio { + /** What the server has for this episode, or null to download from the publisher. */ + suspend fun resolve(episode: EpisodeEntity): ServerEpisode? + + /** Records the server's offsets, once its audio is actually the file on disk. */ + suspend fun record(episode: EpisodeEntity, info: ServerEpisode) +} + class EpisodeDownloader( private val client: OkHttpClient, private val episodes: EpisodeDao, private val libraryDir: File, private val extractor: ExtractedAudioSource? = null, /** - * Fired once a download actually lands on disk -- the trigger for Tier 3 fingerprint - * matching, which needs the file present to decode. Every download route ends up here - * regardless of whether it was a manual tap or a queued/rule-triggered batch, so this is - * the one place that guarantees every newly downloaded episode gets checked. + * The skip server, preferred over the publisher when it has the episode. + * + * Not an optimisation. A show using dynamic ad insertion serves a different file on every + * request -- one real episode measured 22:05 on one fetch and 28:18 on another -- so an + * offset computed on the server's copy points at the wrong moment in any other copy. + * Playing the bytes the detector examined is the only thing that makes the offsets true. */ - private val onDownloaded: suspend (EpisodeEntity) -> Unit = {}, + private val serverAudio: ServerAudio? = null, private val now: () -> Long = System::currentTimeMillis, ) { @@ -47,13 +65,15 @@ class EpisodeDownloader( SubscriptionKind.FACEBOOK_VIDEO, -> downloadViaExtractor(episode) - SubscriptionKind.PODCAST -> downloadOverHttp(episode) + SubscriptionKind.PODCAST -> downloadPodcast(episode) } result.fold( + // Nothing is triggered by a completed download any more. It used to start Tier 3 + // fingerprinting, which has moved to the server -- where comparing several + // episodes at once is possible at all. onSuccess = { episodes.updateDownload(episode.id, DownloadState.DOWNLOADED, it.absolutePath, now()) - onDownloaded(episode.copy(downloadState = DownloadState.DOWNLOADED, localPath = it.absolutePath)) }, // localPath is cleared: a FAILED row pointing at a path invites something downstream // to try playing a file that is not there. @@ -72,13 +92,35 @@ class EpisodeDownloader( return File(path).takeIf { it.isFile && it.length() > 0 } } + /** + * Prefers the skip server, falling back to the publisher. + * + * The fallback is deliberate and matters: skips improve playback rather than being a + * precondition for it, so a server that is down or mid-fetch must not turn a reachable + * episode into a failed download. Offsets are recorded only when the server's audio is + * actually what landed on disk. + */ + private suspend fun downloadPodcast(episode: EpisodeEntity): Result<File> { + val fromServer = serverAudio?.resolve(episode) + val serverUrl = fromServer?.audioUrl?.takeIf { fromServer.ready } + + if (serverUrl != null) { + val result = downloadOverHttp(episode, serverUrl) + if (result.isSuccess) { + serverAudio?.record(episode, fromServer) + return result + } + } + return downloadOverHttp(episode, episode.mediaUrl) + } + private suspend fun downloadViaExtractor(episode: EpisodeEntity): Result<File> { val source = extractor ?: return Result.failure(IllegalStateException("No audio extractor configured")) return runCatching { source.download(episode.mediaUrl, libraryDir) } } - private suspend fun downloadOverHttp(episode: EpisodeEntity): Result<File> = + private suspend fun downloadOverHttp(episode: EpisodeEntity, url: String): Result<File> = withContext(Dispatchers.IO) { val target = targetFile(episode) // Written to a partial name and renamed on success. A truncated file recorded as @@ -87,7 +129,7 @@ class EpisodeDownloader( partial.parentFile?.mkdirs() try { - val request = Request.Builder().url(episode.mediaUrl).header("User-Agent", USER_AGENT).build() + val request = Request.Builder().url(url).header("User-Agent", USER_AGENT).build() client.newCall(request).execute().use { response -> if (!response.isSuccessful) { partial.delete() diff --git a/app/src/main/java/org/terst/listen/server/OkHttpSkipTransport.kt b/app/src/main/java/org/terst/listen/server/OkHttpSkipTransport.kt new file mode 100644 index 0000000..1541dfb --- /dev/null +++ b/app/src/main/java/org/terst/listen/server/OkHttpSkipTransport.kt @@ -0,0 +1,50 @@ +package org.terst.listen.server + +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.withContext +import okhttp3.MediaType.Companion.toMediaType +import okhttp3.OkHttpClient +import okhttp3.Request +import okhttp3.RequestBody.Companion.toRequestBody + +/** + * [SkipTransport] over the app's own OkHttp client. + * + * On the framework-adapter exclusion list: there is no decision in here, only the HTTP that + * [SkipServer]'s logic is deliberately kept free of. + * + * Every failure is a null or a false rather than an exception. The server is an improvement the + * app prefers, not something it depends on -- so unreachable, slow and broken all have to mean + * the same thing to the caller as not configured: carry on without it. + */ +class OkHttpSkipTransport(private val client: OkHttpClient) : SkipTransport { + + override suspend fun get(url: String, token: String): String? = withContext(Dispatchers.IO) { + runCatching { + val request = Request.Builder() + .url(url) + .header("Authorization", "Bearer $token") + .build() + client.newCall(request).execute().use { response -> + if (!response.isSuccessful) return@use null + response.body?.string() + } + }.getOrNull() + } + + override suspend fun post(url: String, token: String, body: String): Boolean = + withContext(Dispatchers.IO) { + runCatching { + val request = Request.Builder() + .url(url) + .header("Authorization", "Bearer $token") + .post(body.toRequestBody(JSON)) + .build() + client.newCall(request).execute().use { it.isSuccessful } + }.getOrDefault(false) + } + + private companion object { + val JSON = "application/json; charset=utf-8".toMediaType() + } +} diff --git a/app/src/main/java/org/terst/listen/server/ServerSkips.kt b/app/src/main/java/org/terst/listen/server/ServerSkips.kt new file mode 100644 index 0000000..edff77b --- /dev/null +++ b/app/src/main/java/org/terst/listen/server/ServerSkips.kt @@ -0,0 +1,56 @@ +package org.terst.listen.server + +import org.terst.listen.data.EpisodeEntity +import org.terst.listen.data.PlaybackStateDao +import org.terst.listen.data.ShowFingerprintDao +import org.terst.listen.data.SubscriptionDao +import org.terst.listen.download.ServerAudio + +/** + * Connects the skip server to the download path. + * + * What the server returns goes into exactly the columns the on-device detector used to write, + * which is why playback needed no changes at all: `effectiveIntroMs`, `effectiveOutroMs` and + * `PlayheadPolicy` already read them. Only the origin of the numbers has moved. + */ +class ServerSkips( + private val server: SkipServer, + private val subscriptions: SubscriptionDao, + private val fingerprints: ShowFingerprintDao, + private val playbackState: PlaybackStateDao, +) : ServerAudio { + + /** + * Asks the server about an episode, identifying it the way the server keys them. + * + * Three identifiers are sent, not one, because publishers rewrite feeds: a GUID that has + * moved still matches on its enclosure URL, and failing that on its title. An episode that + * loses its identity loses its offsets. + */ + override suspend fun resolve(episode: EpisodeEntity): ServerEpisode? { + val feedUrl = subscriptions.getById(episode.subscriptionId)?.feedUrl ?: return null + return server.episode( + feedUrl = feedUrl, + guid = episode.guid, + enclosureUrl = episode.mediaUrl, + title = episode.title, + ) + } + + /** + * Writes down what the server knew, once its audio is the file on disk. + * + * Each value is written only if the server actually reported it. A missing offset is not a + * zero: a detected zero means "this show has no intro", which playback treats differently + * from having nothing detected, and collapsing the two would make a real finding + * indistinguishable from an absent one. + */ + override suspend fun record(episode: EpisodeEntity, info: ServerEpisode) { + info.introMs?.let { fingerprints.setDetectedIntroMs(episode.id, it) } + info.outroMs?.let { fingerprints.setDetectedOutroMs(episode.id, it) } + // The length the file really is. The feed's figure is minutes short on any show using + // dynamic ad insertion, and this is the number that places the outro offset and decides + // when an episode counts as finished. + info.durationMs?.let { playbackState.updateDuration(episode.id, it) } + } +} diff --git a/app/src/main/java/org/terst/listen/server/SkipServer.kt b/app/src/main/java/org/terst/listen/server/SkipServer.kt new file mode 100644 index 0000000..a480fb5 --- /dev/null +++ b/app/src/main/java/org/terst/listen/server/SkipServer.kt @@ -0,0 +1,144 @@ +package org.terst.listen.server + +import kotlinx.serialization.SerialName +import kotlinx.serialization.Serializable +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonPrimitive + +/** + * Where the skip server is and how to authenticate to it. + * + * Null everywhere it appears means "not configured", which is a working state: the app falls + * back to downloading from publishers with no detected skips, exactly as it did before the + * server existed. + */ +data class SkipServerConfig(val baseUrl: String, val token: String) + +/** The HTTP the client needs, behind an interface so its logic is testable off-device. */ +interface SkipTransport { + /** Returns the response body, or null for any failure. */ + suspend fun get(url: String, token: String): String? + + /** Returns whether the post succeeded. */ + suspend fun post(url: String, token: String, body: String): Boolean +} + +/** What the server knows about one episode. */ +data class ServerEpisode( + /** Whether the audio can be played from the server now. */ + val ready: Boolean, + /** An absolute, token-carrying URL, usable by the player as given. */ + val audioUrl: String?, + /** The file's real length, which is routinely not what the feed declared. */ + val durationMs: Long?, + val introMs: Long?, + val outroMs: Long?, + /** + * Whether detection has run over this episode's show. Distinct from having offsets: a show + * can be examined and genuinely have no recurring intro, and reporting that as "nothing + * found yet" is the mistake the app's own readout used to make. + */ + val detected: Boolean, +) + +/** + * The phone's half of the skip server. + * + * Detection itself moved off the phone. The intro search compares several episodes' audio at + * once, which needs more memory than an Android process is given and more time than + * WorkManager's ten-minute ceiling allows -- the attempt burned the battery for weeks and + * confirmed nothing. What is left here is asking. + * + * The server serves the audio as well as the offsets, and that is load-bearing rather than + * convenient: a show using dynamic ad insertion returns a different file on every request, so + * an offset measured on the server's copy points at the wrong moment in a copy the phone + * fetched for itself. Measured on one real episode: 22:05 on one fetch, 28:18 on another. + * Playing the server's bytes is what makes its numbers true. + */ +class SkipServer( + private val config: suspend () -> SkipServerConfig?, + private val transport: SkipTransport, +) { + + /** + * Tells the server what the phone subscribes to. + * + * The phone owns this list: everything already built around it -- the share sheet, OPML + * import, the queue rules, Android Auto's browse tree -- keeps working untouched, and the + * server stays something the app prefers rather than depends on. + */ + suspend fun pushSubscriptions(feedUrls: List<String>): Boolean { + val target = config() ?: return false + val body = buildString { + append("""{"feeds":[""") + feedUrls.forEachIndexed { index, url -> + if (index > 0) append(",") + append(JsonPrimitive(url)) + } + append("]}") + } + return transport.post(base(target) + "/api/subscriptions", target.token, body) + } + + /** + * Asks what to skip in one episode, and where its audio is. + * + * Null whenever there is no answer -- not configured, unreachable, or a response that is not + * what this expects. Every one of those means the same thing to the caller: carry on the way + * the app worked before the server existed. + */ + suspend fun episode( + feedUrl: String, + guid: String, + enclosureUrl: String, + title: String, + ): ServerEpisode? { + val target = config() ?: return null + val query = buildString { + append("?feed=").append(encode(feedUrl)) + append("&guid=").append(encode(guid)) + append("&enclosure=").append(encode(enclosureUrl)) + append("&title=").append(encode(title)) + } + + val body = transport.get(base(target) + "/api/episode" + query, target.token) ?: return null + val parsed = runCatching { lenient.decodeFromString<EpisodeResponse>(body) }.getOrNull() ?: return null + + return ServerEpisode( + ready = parsed.status == "ready" && parsed.audioUrl != null, + audioUrl = parsed.audioUrl?.let { absolute(base(target), it) }, + durationMs = parsed.durationMs?.takeIf { it > 0 }, + introMs = parsed.introMs, + outroMs = parsed.outroMs, + detected = parsed.detected, + ) + } + + private fun base(target: SkipServerConfig) = target.baseUrl.trimEnd('/') + + /** + * The server returns a root-relative audio path; a player handed that unresolved would try + * to open a local file. + */ + private fun absolute(base: String, path: String): String = + if (path.startsWith("http://") || path.startsWith("https://")) path else base + path + + @Serializable + private data class EpisodeResponse( + val status: String = "", + @SerialName("audioUrl") val audioUrl: String? = null, + @SerialName("durationMs") val durationMs: Long? = null, + @SerialName("introMs") val introMs: Long? = null, + @SerialName("outroMs") val outroMs: Long? = null, + val detected: Boolean = false, + ) + + private companion object { + /** Tolerant of fields this app does not know about, so the server can grow. */ + val lenient = Json { ignoreUnknownKeys = true; isLenient = true } + } +} + +/** Percent-encoding for a query value. A feed URL and a GUID both contain characters that + * would otherwise end the parameter early. */ +private fun encode(value: String): String = java.net.URLEncoder.encode(value, "UTF-8") diff --git a/app/src/main/java/org/terst/listen/ui/MainActivity.kt b/app/src/main/java/org/terst/listen/ui/MainActivity.kt index a4cc048..771e844 100644 --- a/app/src/main/java/org/terst/listen/ui/MainActivity.kt +++ b/app/src/main/java/org/terst/listen/ui/MainActivity.kt @@ -16,6 +16,8 @@ import androidx.compose.runtime.collectAsState import androidx.compose.runtime.getValue import androidx.compose.ui.Modifier import androidx.navigation.compose.NavHost +import androidx.navigation.NavGraphBuilder +import androidx.navigation.NavHostController import androidx.navigation.compose.composable import androidx.navigation.compose.rememberNavController @@ -108,10 +110,7 @@ private fun ListenApp( PlayerScreen(player = player, onBack = { nav.popBackStack() }) } episodeRoute(nav, viewModel, player) - composable("settings") { - SettingsScreen(opml = opmlViewModel, player = player, - onOpenRules = { nav.navigate("rules") }, onBack = { nav.popBackStack() }) - } + settingsRoute(nav, opmlViewModel, player) rulesRoutes(nav, rulesViewModel) composable("queue") { QueueScreen( @@ -216,3 +215,23 @@ private fun androidx.navigation.NavGraphBuilder.rulesRoutes( } } } + +/** + * The settings route, extracted so the app's own navigation graph stays readable as screens are + * added to it. + */ +private fun NavGraphBuilder.settingsRoute( + nav: NavHostController, + opml: OpmlViewModel, + player: PlayerViewModel, +) { + composable("settings") { + SettingsScreen( + opml = opml, + player = player, + server = rememberSkipServerSettings(), + onOpenRules = { nav.navigate("rules") }, + onBack = { nav.popBackStack() }, + ) + } +} diff --git a/app/src/main/java/org/terst/listen/ui/SettingsScreen.kt b/app/src/main/java/org/terst/listen/ui/SettingsScreen.kt index 1d8da1b..9afa3e9 100644 --- a/app/src/main/java/org/terst/listen/ui/SettingsScreen.kt +++ b/app/src/main/java/org/terst/listen/ui/SettingsScreen.kt @@ -15,11 +15,12 @@ import androidx.compose.material.icons.Icons import androidx.compose.material.icons.automirrored.filled.ArrowBack import androidx.compose.material3.Button import androidx.compose.material3.ExperimentalMaterial3Api +import androidx.compose.material3.HorizontalDivider import androidx.compose.material3.Icon import androidx.compose.material3.IconButton -import androidx.compose.material3.HorizontalDivider import androidx.compose.material3.MaterialTheme import androidx.compose.material3.OutlinedButton +import androidx.compose.material3.OutlinedTextField import androidx.compose.material3.Scaffold import androidx.compose.material3.Snackbar import androidx.compose.material3.SnackbarHost @@ -30,7 +31,9 @@ import androidx.compose.runtime.Composable import androidx.compose.runtime.LaunchedEffect import androidx.compose.runtime.collectAsState import androidx.compose.runtime.getValue +import androidx.compose.runtime.mutableStateOf import androidx.compose.runtime.remember +import androidx.compose.runtime.setValue import androidx.compose.ui.Modifier import androidx.compose.ui.unit.dp @@ -47,6 +50,7 @@ private val OPML_MIME_TYPES = arrayOf("text/x-opml", "text/xml", "application/xm fun SettingsScreen( opml: OpmlViewModel, player: PlayerViewModel, + server: SkipServerSettings, onOpenRules: () -> Unit, onBack: () -> Unit, ) { @@ -103,10 +107,64 @@ fun SettingsScreen( onImport = { importLauncher.launch(OPML_MIME_TYPES) }, onExport = { exportLauncher.launch("listen-subscriptions.opml") }, ) + HorizontalDivider() + SkipServerSection(server) } } } +/** The skip server's address and token, as the settings screen reads and writes them. */ +data class SkipServerSettings( + val url: String, + val token: String, + val onSave: (url: String, token: String) -> Unit, +) + +/** + * Where the skip detection comes from. + * + * Detection used to run on the phone and could not work there: comparing several episodes' + * audio at once needs more memory than an Android process gets and more time than WorkManager + * allows, so it burned the battery for weeks and confirmed nothing. The server does that work + * and serves the audio it examined -- which is the part that matters, because a show using + * dynamic ad insertion returns a different file on every request and an offset measured on one + * copy is wrong for another. + * + * Left blank the app behaves as it did before: downloads from publishers, no detected skips. + */ +@Composable +private fun SkipServerSection(settings: SkipServerSettings) { + var url by remember(settings.url) { mutableStateOf(settings.url) } + var token by remember(settings.token) { mutableStateOf(settings.token) } + + Section("Skip server") { + OutlinedTextField( + value = url, + onValueChange = { url = it }, + label = { Text("Address") }, + placeholder = { Text("https://listen.terst.org") }, + singleLine = true, + modifier = Modifier.fillMaxWidth(), + ) + OutlinedTextField( + value = token, + onValueChange = { token = it }, + label = { Text("Token") }, + singleLine = true, + modifier = Modifier.fillMaxWidth(), + ) + Button(onClick = { settings.onSave(url, token) }) { Text("Save") } + Text( + if (settings.url.isBlank()) { + "Not set. Episodes download from publishers and play without detected skips." + } else { + "Episodes download from the server, which is also what analysed them." + }, + style = MaterialTheme.typography.bodySmall, + ) + } +} + @Composable private fun SpeedSection(current: Float, onSelect: (Float) -> Unit) { Section("Playback speed") { diff --git a/app/src/main/java/org/terst/listen/ui/SkipServerSettingsState.kt b/app/src/main/java/org/terst/listen/ui/SkipServerSettingsState.kt new file mode 100644 index 0000000..fc0e49d --- /dev/null +++ b/app/src/main/java/org/terst/listen/ui/SkipServerSettingsState.kt @@ -0,0 +1,35 @@ +package org.terst.listen.ui + +import androidx.compose.runtime.Composable +import androidx.compose.runtime.collectAsState +import androidx.compose.runtime.getValue +import androidx.compose.runtime.remember +import androidx.compose.runtime.rememberCoroutineScope +import androidx.compose.ui.platform.LocalContext +import kotlinx.coroutines.launch +import org.terst.listen.data.Graph + +/** + * Reads and writes the skip server's address and token. + * + * Trivial glue, kept out of the composable that renders them so the screen stays a screen. In + * DataStore rather than Room, like the other settings that describe how the app is being used + * rather than what is in the library. + */ +@Composable +internal fun rememberSkipServerSettings(): SkipServerSettings { + val context = LocalContext.current + val preferences = remember(context) { Graph.get(context).preferences } + val scope = rememberCoroutineScope() + + val url by preferences.serverUrl.collectAsState(initial = "") + val token by preferences.serverToken.collectAsState(initial = "") + + return SkipServerSettings( + url = url, + token = token, + onSave = { newUrl, newToken -> + scope.launch { preferences.setServer(newUrl, newToken) } + }, + ) +} 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 9c48db0..9809e9d 100644 --- a/app/src/main/java/org/terst/listen/work/Workers.kt +++ b/app/src/main/java/org/terst/listen/work/Workers.kt @@ -12,9 +12,6 @@ import androidx.work.WorkManager import androidx.work.WorkerParameters import org.terst.listen.data.Graph import org.terst.listen.data.RefreshOutcome -import org.terst.listen.fingerprint.FingerprintUpdater -import org.terst.listen.fingerprint.MediaCodecClipSource -import org.terst.listen.fingerprint.MidrollUpdater import java.util.concurrent.TimeUnit /** @@ -28,6 +25,9 @@ class RefreshWorker(context: Context, params: WorkerParameters) : CoroutineWorke override suspend fun doWork(): Result { val graph = Graph.get(applicationContext) + // The subscription list reaches the skip server from SubscriptionRepository.refreshAll, + // which every refresh route goes through -- including the library's refresh button, + // which does not come via this worker. val outcomes = graph.repository.refreshAll() val failed = outcomes.values.count { it is RefreshOutcome.Failed } // After the refresh, so a drip release can pick up anything that just arrived. @@ -77,62 +77,15 @@ class RetentionWorker(context: Context, params: WorkerParameters) : CoroutineWor } } -/** - * Checks one newly downloaded episode against its show's Tier 3 fingerprint state. - * - * The decode is [org.terst.listen.fingerprint.MediaCodecClipSource] (no JVM equivalent); the - * decision is [org.terst.listen.fingerprint.FingerprintUpdater], which is tested. - */ -class FingerprintWorker(context: Context, params: WorkerParameters) : CoroutineWorker(context, params) { - - override suspend fun doWork(): Result { - val episodeId = inputData.getString(KEY_EPISODE_ID) ?: return Result.failure() - val graph = Graph.get(applicationContext) - val episode = graph.episodes.getById(episodeId) ?: return Result.success() - - FingerprintUpdater(graph.fingerprints, graph.episodes, MediaCodecClipSource()).process(episode) - - return Result.success() - } - - companion object { - const val KEY_EPISODE_ID = "episodeId" - } -} - -/** - * Runs [MidrollUpdater] for every show, nightly rather than per-download like [FingerprintWorker]: - * a full-body scan is expensive enough (see the desktop prototype findings) that it is worth - * batching, and unlike a download it never touches the network, so there is nothing per-episode - * to react to promptly. - */ -class MidrollWorker(context: Context, params: WorkerParameters) : CoroutineWorker(context, params) { - - override suspend fun doWork(): Result { - val graph = Graph.get(applicationContext) - val updater = MidrollUpdater( - graph.subscriptions, - graph.episodes, - graph.midrolls, - MediaCodecClipSource(), - graph.midrollClips, - ) - for (show in graph.subscriptions.getAll()) updater.processShow(show.id) - return Result.success() - } -} - 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 MIDROLL_WORK = "listen-midroll" private const val REFRESH_HOURS = 6L private const val DOWNLOAD_HOURS = 2L private const val RETENTION_HOURS = 12L - private const val MIDROLL_HOURS = 24L /** * Fetches what is in the queue now, rather than at the next periodic sweep. @@ -150,20 +103,6 @@ object Scheduling { } /** - * Checks one freshly downloaded episode against its show's fingerprint state. - * - * Not a unique work name: unlike a burst of queueing, a run of downloads is one episode - * each, and there is no "latest one wins" -- every episode needs its own check. - */ - fun fingerprintNow(context: Context, episodeId: String) { - WorkManager.getInstance(context).enqueue( - OneTimeWorkRequestBuilder<FingerprintWorker>() - .setInputData(androidx.work.Data.Builder().putString(FingerprintWorker.KEY_EPISODE_ID, episodeId).build()) - .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. @@ -208,18 +147,5 @@ object Scheduling { .build(), ) - manager.enqueueUniquePeriodicWork( - MIDROLL_WORK, - ExistingPeriodicWorkPolicy.KEEP, - PeriodicWorkRequestBuilder<MidrollWorker>(MIDROLL_HOURS, TimeUnit.HOURS) - .setConstraints( - Constraints.Builder() - // Charging, not wifi: everything it touches is already on disk, so - // unlike DownloadWorker there is no data cost to gate on network for. - .setRequiresCharging(true) - .build() - ) - .build(), - ) } } diff --git a/app/src/test/java/org/terst/listen/data/SubscriptionRepositoryTest.kt b/app/src/test/java/org/terst/listen/data/SubscriptionRepositoryTest.kt index ccede65..2474f8a 100644 --- a/app/src/test/java/org/terst/listen/data/SubscriptionRepositoryTest.kt +++ b/app/src/test/java/org/terst/listen/data/SubscriptionRepositoryTest.kt @@ -349,7 +349,7 @@ class SubscriptionRepositoryTest { subs, eps, FeedFetcher(OkHttpClient()), - probe = { ProbedVideo(title = "Sheep dog at work", durationMs = 92_000) }, + SubscriptionCollaborators(probe = { ProbedVideo(title = "Sheep dog at work", durationMs = 92_000) }), ) { clock } probing.addSingleEpisode(video("111", null, kind = SubscriptionKind.FACEBOOK_VIDEO)) @@ -366,7 +366,7 @@ class SubscriptionRepositoryTest { subs, eps, FeedFetcher(OkHttpClient()), - probe = { ProbedVideo(title = "Something else", durationMs = 1) }, + SubscriptionCollaborators(probe = { ProbedVideo(title = "Something else", durationMs = 1) }), ) { clock } probing.addSingleEpisode(video("aaa", "From the page")) @@ -381,7 +381,7 @@ class SubscriptionRepositoryTest { subs, eps, FeedFetcher(OkHttpClient()), - probe = { null }, + SubscriptionCollaborators(probe = { null }), ) { clock } probing.addSingleEpisode(video("111", null, kind = SubscriptionKind.FACEBOOK_VIDEO)) @@ -403,7 +403,7 @@ class SubscriptionRepositoryTest { subs, eps, FeedFetcher(OkHttpClient()), - onDiscovered = { discovered += it }, + SubscriptionCollaborators(onDiscovered = { discovered += it }), ) { clock } @Test diff --git a/app/src/test/java/org/terst/listen/download/EpisodeDownloaderTest.kt b/app/src/test/java/org/terst/listen/download/EpisodeDownloaderTest.kt index d5e971b..b35983a 100644 --- a/app/src/test/java/org/terst/listen/download/EpisodeDownloaderTest.kt +++ b/app/src/test/java/org/terst/listen/download/EpisodeDownloaderTest.kt @@ -17,6 +17,7 @@ import org.terst.listen.data.DownloadState import org.terst.listen.data.EpisodeEntity import org.terst.listen.data.FakeEpisodeDao import org.terst.listen.data.SubscriptionKind +import org.terst.listen.server.ServerEpisode import java.io.File class EpisodeDownloaderTest { @@ -161,4 +162,126 @@ class EpisodeDownloaderTest { downloader.download(ep, SubscriptionKind.YOUTUBE_CHANNEL) assertEquals("no plain HTTP request should have been made", 0, server.requestCount) } + + // ---- the skip server as the preferred source ---- + // + // The server has to serve the audio, not just the offsets. A show using dynamic ad + // insertion returns a different file on every request -- one real episode measured 22:05 on + // one fetch and 28:18 on another -- so an offset computed on the server's copy is simply + // wrong for a copy the phone fetched itself. Downloading from the server is what makes its + // numbers apply to the bytes being played. + + private class FakeServerAudio(private val info: ServerEpisode?) : ServerAudio { + var recorded: Pair<String, ServerEpisode>? = null + override suspend fun resolve(episode: EpisodeEntity) = info + override suspend fun record(episode: EpisodeEntity, info: ServerEpisode) { + recorded = episode.id to info + } + } + + private fun withServer(audio: ServerAudio?) = + EpisodeDownloader(OkHttpClient(), dao, library, null, audio) { 1_000L } + + @Test + fun `downloads from the skip server when it has the episode`() = runTest { + server.enqueue(MockResponse().setBody("server-bytes")) + val audio = FakeServerAudio( + ServerEpisode( + ready = true, + audioUrl = server.url("/audio/abc?token=t").toString(), + durationMs = 1_698_717, + introMs = 142_000, + outroMs = 11_000, + detected = true, + ) + ) + val ep = episode(url = "https://publisher.example/original.mp3").also { dao.rows += it } + + val result = withServer(audio).download(ep, SubscriptionKind.PODCAST) + + assertTrue(result.isSuccess) + assertEquals("server-bytes", result.getOrThrow().readText()) + assertEquals("/audio/abc?token=t", server.takeRequest().path) + } + + @Test + fun `records what the server knew once the file is on disk`() = runTest { + server.enqueue(MockResponse().setBody("server-bytes")) + val info = ServerEpisode( + ready = true, + audioUrl = server.url("/audio/abc").toString(), + durationMs = 1_698_717, + introMs = 142_000, + outroMs = 11_000, + detected = true, + ) + val audio = FakeServerAudio(info) + val ep = episode(url = "https://publisher.example/original.mp3").also { dao.rows += it } + + withServer(audio).download(ep, SubscriptionKind.PODCAST) + + assertEquals(ep.id to info, audio.recorded) + } + + @Test + fun `downloads from the publisher when the server has nothing yet`() = runTest { + server.enqueue(MockResponse().setBody("publisher-bytes")) + val ep = episode(url = server.url("/original.mp3").toString()).also { dao.rows += it } + + val result = withServer(FakeServerAudio(null)).download(ep, SubscriptionKind.PODCAST) + + assertTrue(result.isSuccess) + assertEquals("/original.mp3", server.takeRequest().path) + } + + @Test + fun `records nothing when the server had nothing`() = runTest { + server.enqueue(MockResponse().setBody("publisher-bytes")) + val audio = FakeServerAudio(null) + val ep = episode(url = server.url("/original.mp3").toString()).also { dao.rows += it } + + withServer(audio).download(ep, SubscriptionKind.PODCAST) + + assertNull(audio.recorded) + } + + /** + * The server going down must not cost Peter the episode. Skips are an improvement on + * playback, not a precondition for it -- and a dependency that turns a reachable publisher + * into a failed download would make the app worse than before the server existed. + */ + @Test + fun `falls back to the publisher when the server's own audio fails`() = runTest { + server.enqueue(MockResponse().setResponseCode(502)) + server.enqueue(MockResponse().setBody("publisher-bytes")) + val audio = FakeServerAudio( + ServerEpisode(ready = true, audioUrl = server.url("/audio/broken").toString(), + durationMs = null, introMs = 1, outroMs = 2, detected = true) + ) + val ep = episode(url = server.url("/original.mp3").toString()).also { dao.rows += it } + + val result = withServer(audio).download(ep, SubscriptionKind.PODCAST) + + assertTrue("expected the publisher fallback to succeed", result.isSuccess) + assertEquals("publisher-bytes", result.getOrThrow().readText()) + assertEquals("/audio/broken", server.takeRequest().path) + assertEquals("/original.mp3", server.takeRequest().path) + } + + /** Offsets describe the server's copy, so they must not be recorded against a file that + * came from the publisher instead. */ + @Test + fun `records nothing when the audio came from the publisher after all`() = runTest { + server.enqueue(MockResponse().setResponseCode(502)) + server.enqueue(MockResponse().setBody("publisher-bytes")) + val audio = FakeServerAudio( + ServerEpisode(ready = true, audioUrl = server.url("/audio/broken").toString(), + durationMs = null, introMs = 1, outroMs = 2, detected = true) + ) + val ep = episode(url = server.url("/original.mp3").toString()).also { dao.rows += it } + + withServer(audio).download(ep, SubscriptionKind.PODCAST) + + assertNull("recorded offsets for audio it did not download", audio.recorded) + } } diff --git a/app/src/test/java/org/terst/listen/server/OkHttpSkipTransportTest.kt b/app/src/test/java/org/terst/listen/server/OkHttpSkipTransportTest.kt new file mode 100644 index 0000000..335b68d --- /dev/null +++ b/app/src/test/java/org/terst/listen/server/OkHttpSkipTransportTest.kt @@ -0,0 +1,113 @@ +package org.terst.listen.server + +import kotlinx.coroutines.test.runTest +import okhttp3.OkHttpClient +import okhttp3.mockwebserver.MockResponse +import okhttp3.mockwebserver.MockWebServer +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Before +import org.junit.Test + +/** + * The transport's whole contract is that nothing it does can break the app. + * + * The skip server is an improvement on playback, not a precondition for it: unreachable, slow, + * misconfigured and returning nonsense all have to mean the same thing to the caller as "not + * configured" -- carry on downloading from publishers. An exception escaping from here would + * turn a reachable episode into a failed download, which is worse than never having had a + * server at all. + */ +class OkHttpSkipTransportTest { + + private lateinit var server: MockWebServer + private lateinit var transport: OkHttpSkipTransport + + @Before + fun setUp() { + server = MockWebServer() + server.start() + transport = OkHttpSkipTransport(OkHttpClient()) + } + + @After + fun tearDown() = server.shutdown() + + @Test + fun `returns the body of a successful get`() = runTest { + server.enqueue(MockResponse().setBody("""{"status":"ready"}""")) + + val got = transport.get(server.url("/api/episode").toString(), "tok") + + assertEquals("""{"status":"ready"}""", got) + } + + @Test + fun `sends the token as a bearer header`() = runTest { + server.enqueue(MockResponse().setBody("{}")) + + transport.get(server.url("/api/episode").toString(), "hunter2") + + assertEquals("Bearer hunter2", server.takeRequest().getHeader("Authorization")) + } + + @Test + fun `an unauthorised response is nothing rather than an error`() = runTest { + server.enqueue(MockResponse().setResponseCode(401).setBody("""{"error":"unauthorized"}""")) + + assertNull(transport.get(server.url("/api/episode").toString(), "wrong")) + } + + @Test + fun `a server error is nothing rather than an error`() = runTest { + server.enqueue(MockResponse().setResponseCode(502)) + + assertNull(transport.get(server.url("/api/episode").toString(), "tok")) + } + + /** A misconfigured address must not throw out of the download path. */ + @Test + fun `an unreachable host is nothing rather than an error`() = runTest { + assertNull(transport.get("http://127.0.0.1:1/api/episode", "tok")) + } + + @Test + fun `a malformed url is nothing rather than an error`() = runTest { + assertNull(transport.get("not a url at all", "tok")) + } + + @Test + fun `posts the body as json`() = runTest { + server.enqueue(MockResponse().setResponseCode(200)) + + val ok = transport.post( + server.url("/api/subscriptions").toString(), + "tok", + """{"feeds":["https://a"]}""", + ) + + assertTrue(ok) + val request = server.takeRequest() + assertEquals("POST", request.method) + assertEquals("""{"feeds":["https://a"]}""", request.body.readUtf8()) + assertTrue( + "content type was ${request.getHeader("Content-Type")}", + request.getHeader("Content-Type")?.startsWith("application/json") == true, + ) + } + + @Test + fun `a rejected post reports false`() = runTest { + server.enqueue(MockResponse().setResponseCode(400)) + + assertFalse(transport.post(server.url("/api/subscriptions").toString(), "tok", "{}")) + } + + @Test + fun `a post to an unreachable host reports false`() = runTest { + assertFalse(transport.post("http://127.0.0.1:1/api/subscriptions", "tok", "{}")) + } +} diff --git a/app/src/test/java/org/terst/listen/server/ServerSkipsTest.kt b/app/src/test/java/org/terst/listen/server/ServerSkipsTest.kt new file mode 100644 index 0000000..57c03f3 --- /dev/null +++ b/app/src/test/java/org/terst/listen/server/ServerSkipsTest.kt @@ -0,0 +1,155 @@ +package org.terst.listen.server + +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test +import org.terst.listen.data.EpisodeEntity +import org.terst.listen.data.FakeEpisodeDao +import org.terst.listen.data.FakePlaybackStateDao +import org.terst.listen.data.FakeShowFingerprintDao +import org.terst.listen.data.FakeSubscriptionDao +import org.terst.listen.data.SubscriptionEntity + +/** + * What the phone does with the server's answer. + * + * The offsets land in the same columns the on-device detector used to write, which is why + * playback needed no changes at all: `effectiveIntroMs`, `MidrollSkip` and `PlayheadPolicy` + * already read exactly these. Only where the numbers come from has changed. + */ +class ServerSkipsTest { + + private class StubTransport(private val body: String?) : SkipTransport { + val gets = mutableListOf<String>() + override suspend fun get(url: String, token: String): String? { + gets += url + return body + } + + override suspend fun post(url: String, token: String, body: String) = true + } + + private val show = SubscriptionEntity( + id = "s1", + feedUrl = "https://feeds.simplecast.com/EmVW7VGp", + title = "Radiolab", + addedAtMs = 0, + ) + + private val episode = EpisodeEntity( + id = "e1", + subscriptionId = "s1", + guid = "guid-1", + title = "Grass Apes", + mediaUrl = "https://cdn.example/grass.mp3", + mediaType = "audio/mpeg", + discoveredAtMs = 0, + ) + + private fun subject( + transport: SkipTransport, + subscriptions: FakeSubscriptionDao = FakeSubscriptionDao().apply { rows += show }, + episodes: FakeEpisodeDao = FakeEpisodeDao().apply { rows += episode }, + fingerprints: FakeShowFingerprintDao = FakeShowFingerprintDao(), + ): ServerSkips { + // Both fakes write through to the same episode rows, the way the real foreign table + // does, so a test sees the value where production would put it. + fingerprints.episodeRows = episodes.rows + return ServerSkips( + SkipServer({ SkipServerConfig("https://listen.terst.org", "tok") }, transport), + subscriptions, + fingerprints, + FakePlaybackStateDao(episodes.rows), + ) + } + + /** The server keys episodes by feed, so the show's feed URL has to be looked up and sent. */ + @Test + fun `asks the server using the show's feed url`() = runTest { + val transport = StubTransport("""{"status":"pending"}""") + + subject(transport).resolve(episode) + + val url = transport.gets.single() + assertTrue("feed url not sent in $url", url.contains("EmVW7VGp")) + assertTrue("guid not sent in $url", url.contains("guid-1")) + } + + /** An episode whose show has been unsubscribed has no feed to ask about. */ + @Test + fun `does not ask when the show is gone`() = runTest { + val transport = StubTransport("""{"status":"pending"}""") + + val got = subject(transport, subscriptions = FakeSubscriptionDao()).resolve(episode) + + assertNull(got) + assertTrue("asked about a show that is not subscribed", transport.gets.isEmpty()) + } + + @Test + fun `records the detected offsets against the episode`() = runTest { + val episodes = FakeEpisodeDao().apply { rows += episode } + + subject(StubTransport(null), episodes = episodes).record( + episode, + ServerEpisode(ready = true, audioUrl = "x", durationMs = 1_698_717, + introMs = 142_000, outroMs = 11_000, detected = true), + ) + + val stored = episodes.rows.single { it.id == "e1" } + assertEquals(142_000L, stored.detectedIntroMs) + assertEquals(11_000L, stored.detectedOutroMs) + } + + /** + * The measured length too: the feed's claim is minutes short on any show using dynamic ad + * insertion, and this length is the one the file actually has. + */ + @Test + fun `records the measured duration`() = runTest { + val episodes = FakeEpisodeDao().apply { rows += episode } + + subject(StubTransport(null), episodes = episodes).record( + episode, + ServerEpisode(ready = true, audioUrl = "x", durationMs = 1_698_717, + introMs = null, outroMs = null, detected = true), + ) + + assertEquals(1_698_717L, episodes.rows.single { it.id == "e1" }.actualDurationMs) + } + + /** + * An absent offset must not be written as zero. Zero is a real detected value meaning "this + * show has no intro", and playback distinguishes the two. + */ + @Test + fun `leaves an offset the server did not report alone`() = runTest { + val episodes = FakeEpisodeDao().apply { rows += episode } + + subject(StubTransport(null), episodes = episodes).record( + episode, + ServerEpisode(ready = true, audioUrl = "x", durationMs = null, + introMs = 142_000, outroMs = null, detected = true), + ) + + val stored = episodes.rows.single { it.id == "e1" } + assertEquals(142_000L, stored.detectedIntroMs) + assertNull("wrote an outro the server never reported", stored.detectedOutroMs) + } + + /** A detected zero is a finding, not an absence, and has to be stored as one. */ + @Test + fun `records a detected zero offset`() = runTest { + val episodes = FakeEpisodeDao().apply { rows += episode } + + subject(StubTransport(null), episodes = episodes).record( + episode, + ServerEpisode(ready = true, audioUrl = "x", durationMs = null, + introMs = 0, outroMs = null, detected = true), + ) + + assertEquals(0L, episodes.rows.single { it.id == "e1" }.detectedIntroMs) + } +} diff --git a/app/src/test/java/org/terst/listen/server/SkipServerTest.kt b/app/src/test/java/org/terst/listen/server/SkipServerTest.kt new file mode 100644 index 0000000..bdf9d6a --- /dev/null +++ b/app/src/test/java/org/terst/listen/server/SkipServerTest.kt @@ -0,0 +1,194 @@ +package org.terst.listen.server + +import kotlinx.coroutines.test.runTest +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test + +/** + * The phone's half of the skip server. + * + * Detection does not happen here any more, and for a reason worth keeping written down: the + * intro search needs several episodes' audio compared at once, which needs more memory than an + * Android process gets and more time than WorkManager allows. The phone's job is now to say + * what it subscribes to, ask what to skip, and play the audio the answer belongs to. + * + * That last part is not optional. A show using dynamic ad insertion serves a different file on + * every request, so an offset measured on the server's copy is wrong for a copy the phone + * fetched itself -- Grass Apes came back at 22:05 on one fetch and 28:18 on another. Playing + * the server's bytes is what makes the numbers apply. + */ +class SkipServerTest { + + private class FakeTransport : SkipTransport { + var response: String? = null + var failPost = false + val gets = mutableListOf<String>() + val posts = mutableListOf<Pair<String, String>>() + var seenToken: String? = null + + override suspend fun get(url: String, token: String): String? { + gets += url + seenToken = token + return response + } + + override suspend fun post(url: String, token: String, body: String): Boolean { + posts += url to body + seenToken = token + return !failPost + } + } + + private fun server( + transport: SkipTransport, + baseUrl: String? = "https://listen.terst.org", + token: String? = "tok", + ) = SkipServer({ if (baseUrl == null || token == null) null else SkipServerConfig(baseUrl, token) }, transport) + + private val ready = """ + {"status":"ready","audioUrl":"/audio/abc123?token=tok","durationMs":1698717, + "introMs":142000,"outroMs":11000,"detected":true} + """.trimIndent() + + @Test + fun `reads what the server says about an episode`() = runTest { + val transport = FakeTransport().apply { response = ready } + + val got = server(transport).episode("https://f", "g1", "https://cdn/1.mp3", "One") + + assertTrue("expected a ready episode, got $got", got?.ready == true) + assertEquals("https://listen.terst.org/audio/abc123?token=tok", got?.audioUrl) + assertEquals(1698717L, got?.durationMs) + assertEquals(142000L, got?.introMs) + assertEquals(11000L, got?.outroMs) + assertTrue(got?.detected == true) + } + + /** + * The server returns a root-relative audio path. Handing that to a player unresolved would + * have it try to open a local file. + */ + @Test + fun `resolves a relative audio url against the server`() = runTest { + val transport = FakeTransport().apply { + response = """{"status":"ready","audioUrl":"/audio/abc?token=tok"}""" + } + + val got = server(transport, baseUrl = "https://listen.terst.org/").episode("https://f", "g1", "", "") + + assertEquals("https://listen.terst.org/audio/abc?token=tok", got?.audioUrl) + } + + @Test + fun `an absolute audio url is left alone`() = runTest { + val transport = FakeTransport().apply { + response = """{"status":"ready","audioUrl":"https://cdn.example/abc.mp3"}""" + } + + val got = server(transport).episode("https://f", "g1", "", "") + + assertEquals("https://cdn.example/abc.mp3", got?.audioUrl) + } + + /** Pending is a real answer: the server is fetching it, so the phone should not treat it + * as a failure, but it has nothing to play yet either. */ + @Test + fun `a pending episode is reported as not ready`() = runTest { + val transport = FakeTransport().apply { response = """{"status":"pending","detected":false}""" } + + val got = server(transport).episode("https://f", "g1", "", "") + + assertEquals(false, got?.ready) + assertNull(got?.audioUrl) + } + + /** Missing offsets must come back as absent, not as zero: zero is a real detected value + * meaning "this show has no intro", and playback treats the two differently. */ + @Test + fun `absent offsets are null rather than zero`() = runTest { + val transport = FakeTransport().apply { response = """{"status":"ready","audioUrl":"/audio/a"}""" } + + val got = server(transport).episode("https://f", "g1", "", "") + + assertNull(got?.introMs) + assertNull(got?.outroMs) + } + + /** A feed URL and a GUID both contain characters that would otherwise break the query. */ + @Test + fun `encodes the feed url and guid into the query`() = runTest { + val transport = FakeTransport() + + server(transport).episode( + feedUrl = "https://feeds.simplecast.com/EmVW7VGp?x=1&y=2", + guid = "tag:example,2026:1 odd", + enclosureUrl = "https://cdn/a b.mp3", + title = "Grass Apes & Co", + ) + + val url = transport.gets.single() + assertTrue("feed not encoded in $url", url.contains("feed=https%3A%2F%2Ffeeds.simplecast.com%2FEmVW7VGp%3Fx%3D1%26y%3D2")) + assertTrue("guid not encoded in $url", url.contains("guid=tag%3Aexample%2C2026%3A1+odd")) + assertTrue("ampersand in the title not encoded in $url", url.contains("%26+Co")) + } + + @Test + fun `sends the token`() = runTest { + val transport = FakeTransport().apply { response = ready } + server(transport, token = "hunter2").episode("https://f", "g1", "", "") + assertEquals("hunter2", transport.seenToken) + } + + /** With nothing configured the server is simply not used, and nothing throws. */ + @Test + fun `an unconfigured server is a no-op rather than an error`() = runTest { + val transport = FakeTransport() + + assertNull(server(transport, baseUrl = null).episode("https://f", "g1", "", "")) + assertEquals(false, server(transport, token = null).pushSubscriptions(listOf("https://f"))) + assertTrue("talked to a server that is not configured", transport.gets.isEmpty() && transport.posts.isEmpty()) + } + + @Test + fun `rubbish from the server is reported as nothing rather than crashing`() = runTest { + val transport = FakeTransport().apply { response = "<html>502 Bad Gateway</html>" } + + assertNull(server(transport).episode("https://f", "g1", "", "")) + } + + @Test + fun `no response at all is reported as nothing`() = runTest { + assertNull(server(FakeTransport()).episode("https://f", "g1", "", "")) + } + + @Test + fun `pushes the subscription list as json`() = runTest { + val transport = FakeTransport() + + val ok = server(transport).pushSubscriptions(listOf("https://a", "https://b")) + + assertTrue(ok) + val (url, body) = transport.posts.single() + assertEquals("https://listen.terst.org/api/subscriptions", url) + assertEquals("""{"feeds":["https://a","https://b"]}""", body) + } + + @Test + fun `a failed push is reported rather than thrown`() = runTest { + val transport = FakeTransport().apply { failPost = true } + + assertEquals(false, server(transport).pushSubscriptions(listOf("https://a"))) + } + + /** A trailing slash on the configured URL must not produce a double slash in the path. */ + @Test + fun `tolerates a trailing slash on the base url`() = runTest { + val transport = FakeTransport() + + server(transport, baseUrl = "https://listen.terst.org/").pushSubscriptions(listOf("https://a")) + + assertEquals("https://listen.terst.org/api/subscriptions", transport.posts.single().first) + } +} diff --git a/docs/detection-findings.md b/docs/detection-findings.md index 1bb7880..26b31b3 100644 --- a/docs/detection-findings.md +++ b/docs/detection-findings.md @@ -495,3 +495,77 @@ milliseconds, and `SelfCheck` tolerates half a second of position error for the one host with curl is not enough to conclude either way, and it should be measured properly before the design leans on it. - **Midroll detection is not ported.** Only intro/outro. + +--- + +## 2026-10-05, later: the server ships, and simultaneous mirroring defeats the threshold + +`server/` now runs on tungsten as `listend.service`, behind Apache at https://listen.terst.org +with a Let's Encrypt certificate. It polls the phone's subscriptions, mirrors the newest 8 +episodes per show, detects, and serves both the offsets and the audio they belong to. The phone +pushes its feed list from `SubscriptionRepository.refreshAll`, prefers the server's audio when it +has it, and writes the offsets into the same `detectedIntroMs`/`detectedOutroMs` columns the +on-device detector used -- so playback needed no changes at all. `FingerprintWorker` and +`MidrollWorker` are deleted. + +Verified live, phone to server: `POST /api/subscriptions` returned 200 from the app, the server +woke, mirrored 8 Radiolab episodes, and detected across all of them in 3m59s. + +### Dynamic ad insertion is on every episode, and it is not subtle + +Every one of the 8 came back longer than declared: + +| episode | declared | served | inserted | +| --- | --- | --- | --- | +| Grass Apes | 22:05 | 28:18 | 6:13 | +| The Sweetest Thing | 31:02 | 37:15 | 6:13 | +| 23 Weeks 6 Days | 63:27 | 72:26 | 8:59 | +| Nigel and the Real Gannet | 25:54 | 33:01 | 7:07 | +| The Shoe Thrower | 77:29 | 87:02 | 9:33 | + +This also corrects the earlier pass in this document, which found declared == served on four +fetches made with plain `curl` and doubted whether insertion was happening at all. It is, on +every episode, when fetched by the server. Why `curl` got clean files and the server does not is +unexplained and worth knowing -- the only obvious difference is the User-Agent. + +### The confirmation threshold does not survive simultaneous fetching + +Detection reported the intro confirmed on 8 of 8, self-check passed. **It had locked onto an +advertisement.** + +Proved, rather than suspected: the 8-second segment the detector chose was extracted from the +server's copy of Grass Apes and searched for in the `curl` copy of the same episode fetched +hours earlier with no inserted advertising. **r = 0.0621.** It is not in the programme. It is in +the ad break. + +The mechanism is the Darknet Diaries trap at a new scale. The threshold's whole premise is that +three *independently fetched* episodes are unlikely to share an inserted ad. Mirroring 8 +episodes within the same minute from the same ad server makes them maximally dependent: whatever +campaign is running is in all 8, identical, and the detector finds it with scores of 0.93 to +1.00. The self-check cannot catch this -- it only proves the located segments are the same +audio, and they genuinely are. + +This is *worse* than the phone's original design, where episodes arrived days apart and so were +genuinely independent. Moving to a server fixed the cost problem and broke the independence +assumption the safety mechanism rests on. + +### What the published offsets currently are + +Head-window matches only, so every offset lands inside the first four minutes, and skipping to +the end of a matched segment skips pre-roll advertising. That is useful by accident, not by +design, and it will move unpredictably whenever the ad campaign rotates. **It is not intro +detection and must not be described as such.** + +### The fix, not yet made + +Record when each episode's audio was fetched, and require the agreeing episodes to span more +than one fetch occasion before a segment is trusted as the show's own. The mirror already rolls +forward one episode at a time as feeds update, so independence arrives naturally -- it is only +the initial batch that is simultaneous. Expect no offsets at all for the first day or two after +a show is added, which is the correct answer rather than a confident wrong one. + +There is a reframing worth considering instead, which the data argues for: what this reliably +detects, with very high confidence, is **the inserted advertising itself** -- because +simultaneous fetches share it exactly. Skipping ads is the actual goal, and the declared-versus- +served gap is a free, independent check on whether a scan has accounted for all of it. Detecting +ads directly may be a better target than detecting intros and hoping ads do not get in the way. diff --git a/server/cmd/listend/main.go b/server/cmd/listend/main.go new file mode 100644 index 0000000..588af0d --- /dev/null +++ b/server/cmd/listend/main.go @@ -0,0 +1,411 @@ +// Command listend mirrors a phone's podcast subscriptions, works out what to skip in them, and +// serves both to the phone. +// +// It exists because detection cannot work on the phone. The intro/outro search needs several +// episodes' audio compared at once, which needs more memory than an Android process gets and +// more time than WorkManager's ten-minute ceiling allows -- and the attempt burned the battery +// while confirming nothing. Here the same work is unremarkable. +// +// It serves the audio as well as the offsets, which is not a convenience. A show using dynamic +// ad insertion returns a different file on every request, so an offset measured on one copy +// points at the wrong moment in another. The phone playing the very bytes the detector examined +// is what makes the numbers correct. +package main + +import ( + "context" + "errors" + "flag" + "fmt" + "io" + "log" + "net/http" + "os" + "os/signal" + "path/filepath" + "sync" + "syscall" + "time" + + "github.com/terst/listen/server/internal/api" + "github.com/terst/listen/server/internal/decode" + "github.com/terst/listen/server/internal/feed" + "github.com/terst/listen/server/internal/mirror" + "github.com/terst/listen/server/internal/skips" + "github.com/terst/listen/server/internal/store" +) + +func main() { + var ( + addr = flag.String("addr", "127.0.0.1:38090", "address to listen on") + dataDir = flag.String("data", "/site/listen.terst.org/data", "where state and audio live") + keep = flag.Int("keep", 8, "episodes to mirror per show") + pollEvery = flag.Duration("poll", 2*time.Hour, "how often to refresh every feed") + tokenFile = flag.String("token-file", "", "file holding the API token (defaults to <data>/token)") + once = flag.Bool("once", false, "run one poll cycle and exit, for checking a deployment") + ) + flag.Parse() + + token, err := readToken(*tokenFile, *dataDir) + if err != nil { + log.Fatalf("token: %v", err) + } + + statePath := filepath.Join(*dataDir, "state.json") + audioDir := filepath.Join(*dataDir, "audio") + state, err := store.Load(statePath) + if err != nil { + log.Fatalf("state: %v", err) + } + + decoder := decode.Decoder{} + worker := &worker{ + state: state, + statePath: statePath, + keep: *keep, + mirror: &mirror.Mirror{AudioDir: audioDir, Probe: decoder}, + decoder: decoder, + requests: make(chan request, 64), + wakeups: make(chan struct{}, 1), + } + + server := api.New(state, worker, api.Options{ + Token: token, + AudioDir: audioDir, + // Act on a new subscription list straight away: the phone sends it on refresh, and + // waiting for the next interval would leave a new show with nothing for hours. + OnSubscriptions: func([]string) { worker.wake() }, + }) + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + if *once { + worker.pollAll(ctx) + if err := state.Save(statePath); err != nil { + log.Fatalf("save: %v", err) + } + log.Printf("one cycle complete") + return + } + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + worker.run(ctx, *pollEvery) + }() + + httpServer := &http.Server{ + Addr: *addr, + Handler: server.Handler(), + ReadHeaderTimeout: 20 * time.Second, + } + go func() { + <-ctx.Done() + shutdownCtx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + _ = httpServer.Shutdown(shutdownCtx) + }() + + log.Printf("listening on %s, data in %s, keeping %d episodes per show", *addr, *dataDir, *keep) + if err := httpServer.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) { + log.Fatalf("http: %v", err) + } + wg.Wait() + if err := state.Save(statePath); err != nil { + log.Printf("save on exit: %v", err) + } +} + +// request is an episode the phone asked for that the server did not have. +type request struct { + feedURL, guid, enclosureURL, title string +} + +type worker struct { + state *store.State + statePath string + keep int + mirror *mirror.Mirror + decoder decode.Decoder + requests chan request + wakeups chan struct{} +} + +// wake asks for a poll now. Coalescing, so a burst of subscription updates is one poll. +func (w *worker) wake() { + select { + case w.wakeups <- struct{}{}: + default: + } +} + +// Request implements api.Fetcher. Non-blocking: a full queue drops the request rather than +// stalling the phone, and the next poll will pick the episode up anyway. +func (w *worker) Request(feedURL, guid, enclosureURL, title string) { + select { + case w.requests <- request{feedURL, guid, enclosureURL, title}: + default: + log.Printf("request queue full, dropping %s %s", feedURL, guid) + } +} + +func (w *worker) run(ctx context.Context, every time.Duration) { + w.pollAll(ctx) + w.save() + + ticker := time.NewTicker(every) + defer ticker.Stop() + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + w.pollAll(ctx) + w.save() + case <-w.wakeups: + w.pollAll(ctx) + w.save() + case req := <-w.requests: + w.serveRequest(ctx, req) + w.save() + } + } +} + +func (w *worker) save() { + if err := w.state.Save(w.statePath); err != nil { + log.Printf("save: %v", err) + } +} + +// pollAll refreshes every subscribed feed, then mirrors and detects what changed. +func (w *worker) pollAll(ctx context.Context) { + feeds := w.state.Feeds() + if len(feeds) == 0 { + log.Printf("no subscriptions yet; waiting for the phone to send its list") + return + } + for _, feedURL := range feeds { + if ctx.Err() != nil { + return + } + w.pollOne(ctx, feedURL) + } + // A feed no longer subscribed stops being polled; reclaim its audio too. + for _, url := range w.state.SetFeeds(feeds) { + if show := w.state.Forget(url); show != nil { + for _, e := range show.Episodes { + _ = w.mirror.Prune(e) + } + log.Printf("%s: unsubscribed, reclaimed %d episodes", url, len(show.Episodes)) + } + } +} + +func (w *worker) pollOne(ctx context.Context, feedURL string) { + raw, err := fetchFeed(ctx, feedURL) + if err != nil { + w.noteFeedError(feedURL, err) + return + } + channel, err := feed.Parse(raw) + if err != nil { + w.noteFeedError(feedURL, err) + return + } + + show := w.state.Merge(feedURL, channel.Title, channel.Episodes) + show.LastPolledUnix = time.Now().Unix() + show.LastError = "" + + kept, pruned := show.Partition(w.keep) + for _, e := range pruned { + if e.Mirrored() { + _ = w.mirror.Prune(e) + } + } + + fetched := 0 + for _, e := range kept { + if ctx.Err() != nil { + return + } + if e.Mirrored() { + continue + } + if err := w.mirror.Fetch(ctx, e); err != nil { + log.Printf("%s: %v", channel.Title, err) + continue + } + fetched++ + } + + if fetched > 0 { + log.Printf("%s: mirrored %d new episode(s)", channel.Title, fetched) + } + w.detect(ctx, show, kept) +} + +// detect runs the intro/outro search over a show's mirrored episodes and records the result. +// +// Every mirrored episode at once, which is the whole reason this is a server: the phone could +// only hold one episode's clip, so it needed a rolling state machine that accumulated agreement +// over weeks and lost it whenever retention reclaimed the file it was holding. +func (w *worker) detect(ctx context.Context, show *store.Show, candidates []*store.Episode) { + var episodes []skips.Episode + byID := map[string]*store.Episode{} + for _, e := range candidates { + if !e.Mirrored() { + continue + } + episodes = append(episodes, skips.Episode{ + ID: e.ID, + Path: filepath.Join(w.mirror.AudioDir, e.Path), + }) + byID[e.ID] = e + } + if len(episodes) < skips.DefaultConfirmThreshold { + return + } + + started := time.Now() + result, err := skips.Detect(ctx, w.decoder, episodes, skips.DefaultConfig()) + if err != nil { + log.Printf("%s: detection failed: %v", show.Title, err) + return + } + + now := time.Now().Unix() + applied := 0 + for _, o := range result.Offsets { + e := byID[o.EpisodeID] + if e == nil { + continue + } + e.DetectedAtUnix = now + e.IntroMs, e.OutroMs = o.IntroMs, o.OutroMs + if o.IntroMs != nil || o.OutroMs != nil { + applied++ + } + } + + log.Printf("%s: intro on %d/%d%s, outro on %d/%d%s, %d episode(s) carry offsets, took %s", + show.Title, + result.IntroConfirmedOn, len(episodes), checkNote(result.IntroCheck), + result.OutroConfirmedOn, len(episodes), checkNote(result.OutroCheck), + applied, time.Since(started).Round(time.Second)) +} + +// checkNote reports a withheld result loudly. A confirmation that could not verify itself is +// the case where publishing offsets would cut the wrong audio. +func checkNote(r skips.CheckReport) string { + if r.Pairs == 0 || r.OK { + return "" + } + return fmt.Sprintf(" (SELF-CHECK FAILED %s vs %s at r=%.3f, offsets withheld)", r.WorstA, r.WorstB, r.WorstScore) +} + +// serveRequest handles an episode the phone wanted that was not mirrored. +func (w *worker) serveRequest(ctx context.Context, req request) { + e := w.state.Match(req.feedURL, req.guid, req.enclosureURL, req.title) + if e == nil { + // Not in the feed as last polled. Register it from what the phone told us, so an + // episode older than the mirrored window is still playable with offsets. + if req.enclosureURL == "" { + return + } + w.state.Merge(req.feedURL, "", []feed.Episode{{ + GUID: req.guid, + Title: req.title, + EnclosureURL: req.enclosureURL, + PublishedUnix: time.Now().Unix(), + }}) + e = w.state.Match(req.feedURL, req.guid, req.enclosureURL, req.title) + if e == nil { + return + } + } + if e.Mirrored() { + return + } + if err := w.mirror.Fetch(ctx, e); err != nil { + log.Printf("on-demand fetch %s: %v", e.ID, err) + return + } + log.Printf("on-demand: mirrored %s", e.Title) + + if show := w.state.Show(req.feedURL); show != nil { + kept, _ := show.Partition(w.keep) + w.detect(ctx, show, kept) + } +} + +func (w *worker) noteFeedError(feedURL string, err error) { + log.Printf("%s: %v", feedURL, err) + if show := w.state.Show(feedURL); show != nil { + show.LastError = err.Error() + show.LastPolledUnix = time.Now().Unix() + } +} + +func fetchFeed(ctx context.Context, url string) ([]byte, error) { + ctx, cancel := context.WithTimeout(ctx, 2*time.Minute) + defer cancel() + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + req.Header.Set("User-Agent", "listen-server/1.0 (+https://terst.org)") + resp, err := http.DefaultClient.Do(req) + if err != nil { + return nil, err + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("%s returned %s", url, resp.Status) + } + // Feeds are large -- a real subscription produced 2753 episodes -- but not unbounded. + return io.ReadAll(io.LimitReader(resp.Body, 64<<20)) +} + +// readToken loads the API token, generating one on first run. +func readToken(tokenFile, dataDir string) (string, error) { + if tokenFile == "" { + tokenFile = filepath.Join(dataDir, "token") + } + if raw, err := os.ReadFile(tokenFile); err == nil { + token := trimSpace(string(raw)) + if token == "" { + return "", fmt.Errorf("%s is empty", tokenFile) + } + return token, nil + } else if !os.IsNotExist(err) { + return "", err + } + + token, err := generateToken() + if err != nil { + return "", err + } + if err := os.MkdirAll(filepath.Dir(tokenFile), 0o755); err != nil { + return "", err + } + if err := os.WriteFile(tokenFile, []byte(token+"\n"), 0o600); err != nil { + return "", err + } + log.Printf("generated a new API token in %s", tokenFile) + return token, nil +} + +func trimSpace(s string) string { + start, end := 0, len(s) + for start < end && (s[start] == ' ' || s[start] == '\n' || s[start] == '\r' || s[start] == '\t') { + start++ + } + for end > start && (s[end-1] == ' ' || s[end-1] == '\n' || s[end-1] == '\r' || s[end-1] == '\t') { + end-- + } + return s[start:end] +} diff --git a/server/cmd/listend/token.go b/server/cmd/listend/token.go new file mode 100644 index 0000000..c6962a7 --- /dev/null +++ b/server/cmd/listend/token.go @@ -0,0 +1,36 @@ +package main + +import ( + "crypto/rand" + "math/big" + "strings" +) + +// tokenAlphabet is deliberately letters and digits only, with no punctuation. +// +// This token is typed by hand into a phone. A base64url token looks stronger and is worse in +// practice: the leading "-" of one generated token was eaten by `adb shell input text` treating +// it as a flag, and "-" and "_" are both buried behind a modifier on a phone keyboard. 32 +// characters of alphanumeric is ~165 bits, which is not the weak link here. +const tokenAlphabet = "abcdefghijkmnopqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789" + +// tokenLength is 32 characters, grouped into fours by the caller when it prints them. +const tokenLength = 32 + +// generateToken makes a fresh API token. +// +// From crypto/rand: this is the only thing between the open internet and a private library. +// The alphabet omits l, I, 1, O and 0 so a token read off a screen cannot be mistyped into a +// different valid-looking one. +func generateToken() (string, error) { + limit := big.NewInt(int64(len(tokenAlphabet))) + var out strings.Builder + for i := 0; i < tokenLength; i++ { + n, err := rand.Int(rand.Reader, limit) + if err != nil { + return "", err + } + out.WriteByte(tokenAlphabet[n.Int64()]) + } + return out.String(), nil +} diff --git a/server/internal/api/api.go b/server/internal/api/api.go new file mode 100644 index 0000000..b6f14ff --- /dev/null +++ b/server/internal/api/api.go @@ -0,0 +1,198 @@ +// Package api serves the phone: what to skip, and the audio those offsets belong to. +package api + +import ( + "encoding/json" + "net/http" + "net/url" + "os" + "path/filepath" + "strings" + + "github.com/terst/listen/server/internal/store" +) + +// Fetcher is asked to mirror an episode the phone wants and the server does not have. +// +// Asynchronous on purpose: the phone gets an immediate "pending" rather than holding a request +// open for the length of a download. +type Fetcher interface { + Request(feedURL, guid, enclosureURL, title string) +} + +// Options configures the server. +type Options struct { + // Token guards everything but health. This serves a private library over the open + // internet. + Token string + // AudioDir is where mirrored episodes live. + AudioDir string + // OnSubscriptions is called when the phone sends its list, so a newly-subscribed show is + // polled at once rather than at the next interval. Optional. + OnSubscriptions func(feeds []string) +} + +// Server is the HTTP surface. +type Server struct { + state *store.State + fetcher Fetcher + opts Options +} + +// New builds the server. +func New(state *store.State, fetcher Fetcher, opts Options) *Server { + return &Server{state: state, fetcher: fetcher, opts: opts} +} + +type episodeResponse struct { + // Status is "ready" when the audio can be played now, or "pending" when the server has + // been asked to fetch it. There is deliberately no "unknown": an episode the phone knows + // about is one the server can be told to go and get. + Status string `json:"status"` + AudioURL string `json:"audioUrl,omitempty"` + DurationMs int64 `json:"durationMs,omitempty"` + IntroMs *int64 `json:"introMs,omitempty"` + OutroMs *int64 `json:"outroMs,omitempty"` + // Detected says whether detection has run over this episode's show, so the phone can tell + // "looked and found nothing" apart from "not looked at yet" -- a distinction the app got + // wrong when the readout said "None" for both. + Detected bool `json:"detected"` +} + +// Handler returns the routed, authenticated handler. +func (s *Server) Handler() http.Handler { + mux := http.NewServeMux() + mux.HandleFunc("/api/health", s.health) + mux.HandleFunc("/api/subscriptions", s.authed(s.subscriptions)) + mux.HandleFunc("/api/episode", s.authed(s.episode)) + mux.HandleFunc("/audio/", s.authed(s.audio)) + return mux +} + +func (s *Server) health(w http.ResponseWriter, _ *http.Request) { + writeJSON(w, http.StatusOK, map[string]string{"status": "ok"}) +} + +// authed accepts the token in the Authorization header or the query string. +// +// The query string is not laziness: a media player fetching audio cannot always be made to set +// a header, and an audio URL the phone cannot use as given would be useless. +func (s *Server) authed(next http.HandlerFunc) http.HandlerFunc { + return func(w http.ResponseWriter, r *http.Request) { + if s.opts.Token == "" || !s.tokenOK(r) { + w.Header().Set("WWW-Authenticate", `Bearer realm="listen"`) + writeJSON(w, http.StatusUnauthorized, map[string]string{"error": "unauthorized"}) + return + } + next(w, r) + } +} + +func (s *Server) tokenOK(r *http.Request) bool { + if header := r.Header.Get("Authorization"); strings.HasPrefix(header, "Bearer ") { + if strings.TrimPrefix(header, "Bearer ") == s.opts.Token { + return true + } + } + return r.URL.Query().Get("token") == s.opts.Token +} + +func (s *Server) subscriptions(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + writeJSON(w, http.StatusMethodNotAllowed, map[string]string{"error": "post only"}) + return + } + var body struct { + Feeds []string `json:"feeds"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "expected {\"feeds\":[...]}"}) + return + } + removed := s.state.SetFeeds(body.Feeds) + if s.opts.OnSubscriptions != nil { + s.opts.OnSubscriptions(s.state.Feeds()) + } + writeJSON(w, http.StatusOK, map[string]any{ + "feeds": len(s.state.Feeds()), + "removed": removed, + }) +} + +func (s *Server) episode(w http.ResponseWriter, r *http.Request) { + q := r.URL.Query() + feedURL, guid := q.Get("feed"), q.Get("guid") + enclosure, title := q.Get("enclosure"), q.Get("title") + + if feedURL == "" || (guid == "" && enclosure == "" && title == "") { + writeJSON(w, http.StatusBadRequest, + map[string]string{"error": "feed and at least one of guid, enclosure or title are required"}) + return + } + + found := s.state.Match(feedURL, guid, enclosure, title) + if found == nil || !found.Mirrored() { + // Ask for it rather than refusing. Everything Peter actually listens to should end up + // with offsets, including an episode older than the ones kept by default. + s.fetcher.Request(feedURL, guid, enclosure, title) + writeJSON(w, http.StatusOK, episodeResponse{Status: "pending"}) + return + } + if _, err := os.Stat(filepath.Join(s.opts.AudioDir, found.Path)); err != nil { + // The record claims a file that has gone. Treat it as not mirrored and fetch again. + s.fetcher.Request(feedURL, found.GUID, found.EnclosureURL, found.Title) + writeJSON(w, http.StatusOK, episodeResponse{Status: "pending"}) + return + } + + writeJSON(w, http.StatusOK, episodeResponse{ + Status: "ready", + AudioURL: "/audio/" + found.ID + "?token=" + url.QueryEscape(s.opts.Token), + DurationMs: found.DurationMs, + IntroMs: found.IntroMs, + OutroMs: found.OutroMs, + Detected: found.DetectedAtUnix != 0, + }) +} + +func (s *Server) audio(w http.ResponseWriter, r *http.Request) { + id := strings.TrimPrefix(r.URL.Path, "/audio/") + // No separators: an id is a flat hex string, and anything else is an attempt to read + // outside the audio directory. + if id == "" || strings.ContainsAny(id, "/\\") || strings.Contains(id, "..") { + http.NotFound(w, r) + return + } + + found := s.episodeByID(id) + if found == nil { + http.NotFound(w, r) + return + } + path := filepath.Join(s.opts.AudioDir, found.Path) + info, err := os.Stat(path) + if err != nil || info.IsDir() { + http.NotFound(w, r) + return + } + // ServeFile handles Range, If-Modified-Since and the content type, which a podcast player + // needs for seeking. + http.ServeFile(w, r, path) +} + +func (s *Server) episodeByID(id string) *store.Episode { + for _, show := range s.state.Shows() { + for _, e := range show.Episodes { + if e.ID == id && e.Mirrored() { + return e + } + } + } + return nil +} + +func writeJSON(w http.ResponseWriter, status int, body any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(body) +} diff --git a/server/internal/api/api_test.go b/server/internal/api/api_test.go new file mode 100644 index 0000000..21ef1de --- /dev/null +++ b/server/internal/api/api_test.go @@ -0,0 +1,325 @@ +package api + +import ( + "encoding/json" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/terst/listen/server/internal/feed" + "github.com/terst/listen/server/internal/store" +) + +const token = "s3cret" + +type fakeFetcher struct{ asked []string } + +func (f *fakeFetcher) Request(feedURL, guid, enclosureURL, title string) { + f.asked = append(f.asked, feedURL+" "+guid+" "+enclosureURL) +} + +func newServer(t *testing.T) (*Server, *store.State, *fakeFetcher, string) { + t.Helper() + audioDir := t.TempDir() + st := store.New() + fetcher := &fakeFetcher{} + return New(st, fetcher, Options{Token: token, AudioDir: audioDir}), st, fetcher, audioDir +} + +func get(t *testing.T, s *Server, path string) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(http.MethodGet, path, nil) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, req) + return rec +} + +func post(t *testing.T, s *Server, path, body string) *httptest.ResponseRecorder { + t.Helper() + req := httptest.NewRequest(http.MethodPost, path, strings.NewReader(body)) + req.Header.Set("Authorization", "Bearer "+token) + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, req) + return rec +} + +// This serves a private library over the public internet, so nothing but health is reachable +// without the token. + +func TestRejectsAnUnauthenticatedRequest(t *testing.T) { + s, _, _, _ := newServer(t) + for _, path := range []string{"/api/episode?feed=f&guid=g", "/api/subscriptions", "/audio/abc"} { + req := httptest.NewRequest(http.MethodGet, path, nil) + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusUnauthorized { + t.Errorf("%s: status %d, want 401", path, rec.Code) + } + } +} + +func TestRejectsAWrongToken(t *testing.T) { + s, _, _, _ := newServer(t) + req := httptest.NewRequest(http.MethodGet, "/api/episode?feed=f&guid=g", nil) + req.Header.Set("Authorization", "Bearer wrong") + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusUnauthorized { + t.Errorf("status %d, want 401", rec.Code) + } +} + +// A media player cannot always set headers on the request it makes for audio, so the token is +// accepted in the query string too -- which is what makes the returned audio URL usable as-is. +func TestAcceptsTheTokenAsAQueryParameter(t *testing.T) { + s, st, _, audioDir := newServer(t) + seedMirrored(t, st, audioDir, "hello") + + req := httptest.NewRequest(http.MethodGet, "/audio/"+episodeID(st)+"?token="+token, nil) + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status %d, want 200", rec.Code) + } + if rec.Body.String() != "hello" { + t.Errorf("body %q", rec.Body.String()) + } +} + +func TestHealthNeedsNoToken(t *testing.T) { + s, _, _, _ := newServer(t) + req := httptest.NewRequest(http.MethodGet, "/api/health", nil) + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Errorf("status %d, want 200", rec.Code) + } +} + +func TestSubscriptionsReplacesTheFeedList(t *testing.T) { + s, st, _, _ := newServer(t) + rec := post(t, s, "/api/subscriptions", `{"feeds":["https://a","https://b"]}`) + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + if got := st.Feeds(); len(got) != 2 { + t.Errorf("feeds %v, want 2", got) + } +} + +// The phone sends its list when it refreshes, and waiting up to a poll interval before acting +// on it would mean a newly-subscribed show has nothing for hours. +func TestSubscriptionsTriggersAPollImmediately(t *testing.T) { + audioDir := t.TempDir() + st := store.New() + notified := make(chan []string, 1) + s := New(st, &fakeFetcher{}, Options{ + Token: token, + AudioDir: audioDir, + OnSubscriptions: func(feeds []string) { + notified <- feeds + }, + }) + + req := httptest.NewRequest(http.MethodPost, "/api/subscriptions", strings.NewReader(`{"feeds":["https://a"]}`)) + req.Header.Set("Authorization", "Bearer "+token) + s.Handler().ServeHTTP(httptest.NewRecorder(), req) + + select { + case feeds := <-notified: + if len(feeds) != 1 || feeds[0] != "https://a" { + t.Errorf("notified with %v", feeds) + } + default: + t.Error("subscriptions did not trigger a poll") + } +} + +func TestSubscriptionsRejectsRubbish(t *testing.T) { + s, _, _, _ := newServer(t) + if rec := post(t, s, "/api/subscriptions", `not json`); rec.Code != http.StatusBadRequest { + t.Errorf("status %d, want 400", rec.Code) + } +} + +// An episode nobody has heard of still gets a useful answer: the server is told enough to go +// and fetch it, and says so, rather than returning a bare 404 the phone can only give up on. +func TestAnUnknownEpisodeIsRequestedAndReportedPending(t *testing.T) { + s, _, fetcher, _ := newServer(t) + + rec := get(t, s, "/api/episode?feed=https://f&guid=g1&enclosure=https://cdn/1.mp3&title=One") + if rec.Code != http.StatusOK { + t.Fatalf("status %d: %s", rec.Code, rec.Body.String()) + } + var got episodeResponse + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if got.Status != "pending" { + t.Errorf("status %q, want pending", got.Status) + } + if len(fetcher.asked) != 1 { + t.Errorf("fetcher asked %d times, want 1", len(fetcher.asked)) + } +} + +func TestAKnownButUnmirroredEpisodeIsRequestedAndReportedPending(t *testing.T) { + s, st, fetcher, _ := newServer(t) + st.Merge("https://f", "A Show", []feed.Episode{ + {GUID: "g1", Title: "One", EnclosureURL: "https://cdn/1.mp3", PublishedUnix: 100}, + }) + + rec := get(t, s, "/api/episode?feed=https://f&guid=g1") + var got episodeResponse + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if got.Status != "pending" { + t.Errorf("status %q, want pending", got.Status) + } + if len(fetcher.asked) != 1 { + t.Errorf("fetcher asked %d times, want 1", len(fetcher.asked)) + } +} + +// Audio becomes usable as soon as it is mirrored, whether or not detection has run. Waiting for +// offsets before handing over playable audio would make the server strictly worse than +// downloading from the publisher. +func TestAMirroredEpisodeIsReadyEvenWithoutOffsets(t *testing.T) { + s, st, _, audioDir := newServer(t) + seedMirrored(t, st, audioDir, "audio-bytes") + + rec := get(t, s, "/api/episode?feed=https://f&guid=g1") + var got episodeResponse + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if got.Status != "ready" { + t.Fatalf("status %q, want ready", got.Status) + } + if got.AudioURL == "" { + t.Error("no audio url") + } + if got.IntroMs != nil || got.OutroMs != nil { + t.Errorf("reported offsets that were never detected: %v %v", got.IntroMs, got.OutroMs) + } +} + +func TestAMirroredEpisodeReportsItsDetectedOffsets(t *testing.T) { + s, st, _, audioDir := newServer(t) + seedMirrored(t, st, audioDir, "audio-bytes") + e := st.Match("https://f", "g1", "", "") + intro, outro := int64(16652), int64(11000) + e.IntroMs, e.OutroMs, e.DurationMs = &intro, &outro, 1325113 + + rec := get(t, s, "/api/episode?feed=https://f&guid=g1") + var got episodeResponse + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + if got.IntroMs == nil || *got.IntroMs != 16652 { + t.Errorf("introMs %v, want 16652", got.IntroMs) + } + if got.OutroMs == nil || *got.OutroMs != 11000 { + t.Errorf("outroMs %v, want 11000", got.OutroMs) + } + if got.DurationMs != 1325113 { + t.Errorf("durationMs %d, want the measured length", got.DurationMs) + } +} + +// The audio URL has to carry the token, or the phone gets a link it cannot use. +func TestTheAudioUrlIsUsableAsGiven(t *testing.T) { + s, st, _, audioDir := newServer(t) + seedMirrored(t, st, audioDir, "audio-bytes") + + rec := get(t, s, "/api/episode?feed=https://f&guid=g1") + var got episodeResponse + if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil { + t.Fatal(err) + } + + req := httptest.NewRequest(http.MethodGet, got.AudioURL, nil) + rec2 := httptest.NewRecorder() + s.Handler().ServeHTTP(rec2, req) + if rec2.Code != http.StatusOK { + t.Fatalf("fetching the url it gave returned %d", rec2.Code) + } + if rec2.Body.String() != "audio-bytes" { + t.Errorf("body %q", rec2.Body.String()) + } +} + +// Seeking is the normal case for a podcast, so byte ranges have to work. +func TestAudioSupportsRangeRequests(t *testing.T) { + s, st, _, audioDir := newServer(t) + seedMirrored(t, st, audioDir, "0123456789") + + req := httptest.NewRequest(http.MethodGet, "/audio/"+episodeID(st)+"?token="+token, nil) + req.Header.Set("Range", "bytes=3-5") + rec := httptest.NewRecorder() + s.Handler().ServeHTTP(rec, req) + + if rec.Code != http.StatusPartialContent { + t.Fatalf("status %d, want 206", rec.Code) + } + if rec.Body.String() != "345" { + t.Errorf("body %q, want %q", rec.Body.String(), "345") + } +} + +func TestAudioForAnUnknownIdIs404(t *testing.T) { + s, _, _, _ := newServer(t) + if rec := get(t, s, "/audio/nope"); rec.Code != http.StatusNotFound { + t.Errorf("status %d, want 404", rec.Code) + } +} + +// A record that claims a file which is no longer there must not 500. +func TestAudioForAMissingFileIs404(t *testing.T) { + s, st, _, audioDir := newServer(t) + seedMirrored(t, st, audioDir, "gone") + os.Remove(filepath.Join(audioDir, st.Match("https://f", "g1", "", "").Path)) + + if rec := get(t, s, "/audio/"+episodeID(st)); rec.Code != http.StatusNotFound { + t.Errorf("status %d, want 404", rec.Code) + } +} + +// An id containing path separators must not be able to read outside the audio directory. +func TestAudioIdCannotEscapeTheAudioDirectory(t *testing.T) { + s, _, _, _ := newServer(t) + for _, id := range []string{"..%2f..%2fetc%2fpasswd", "../../etc/passwd"} { + rec := get(t, s, "/audio/"+id) + if rec.Code == http.StatusOK { + t.Errorf("%q was served", id) + } + } +} + +func TestEpisodeNeedsAFeedAndSomeIdentity(t *testing.T) { + s, _, _, _ := newServer(t) + if rec := get(t, s, "/api/episode"); rec.Code != http.StatusBadRequest { + t.Errorf("status %d, want 400", rec.Code) + } +} + +func seedMirrored(t *testing.T, st *store.State, audioDir, contents string) { + t.Helper() + st.Merge("https://f", "A Show", []feed.Episode{ + {GUID: "g1", Title: "One", EnclosureURL: "https://cdn/1.mp3", PublishedUnix: 100}, + }) + e := st.Match("https://f", "g1", "", "") + e.Path = e.ID + ".mp3" + if err := os.WriteFile(filepath.Join(audioDir, e.Path), []byte(contents), 0o644); err != nil { + t.Fatal(err) + } +} + +func episodeID(st *store.State) string { + return st.Match("https://f", "g1", "", "").ID +} diff --git a/server/internal/feed/feed.go b/server/internal/feed/feed.go new file mode 100644 index 0000000..edc58be --- /dev/null +++ b/server/internal/feed/feed.go @@ -0,0 +1,143 @@ +// Package feed parses a podcast RSS feed down to the few fields the server needs. +package feed + +import ( + "encoding/xml" + "fmt" + "sort" + "strconv" + "strings" + "time" +) + +// Episode is one audio episode as its feed describes it. +type Episode struct { + GUID string + Title string + // EnclosureURL is where the audio is, and the fallback identity when a GUID moves. + EnclosureURL string + EnclosureType string + PublishedUnix int64 + // DeclaredDurationMs is the feed's claim, which is routinely wrong -- a show using dynamic + // ad insertion declares the ad-free length. Useful for ordering, not for arithmetic. + DeclaredDurationMs int64 +} + +// Channel is a show and its audio episodes, newest first. +type Channel struct { + Title string + Episodes []Episode +} + +type rss struct { + Channel struct { + Title string `xml:"title"` + Items []struct { + Title string `xml:"title"` + GUID string `xml:"guid"` + PubDate string `xml:"pubDate"` + Duration string `xml:"http://www.itunes.com/dtds/podcast-1.0.dtd duration"` + Enclosure struct { + URL string `xml:"url,attr"` + Type string `xml:"type,attr"` + } `xml:"enclosure"` + } `xml:"item"` + } `xml:"channel"` +} + +// Parse reads a feed and returns its audio episodes, newest first. +// +// Audio only: video was cut from the app's scope, and a YouTube or Facebook episode has no +// enclosure the server could fetch regardless -- its audio only exists once yt-dlp has resolved +// it, which stays the phone's job. +func Parse(raw []byte) (*Channel, error) { + var doc rss + if err := xml.Unmarshal(raw, &doc); err != nil { + return nil, fmt.Errorf("parse feed: %w", err) + } + if doc.Channel.Title == "" && len(doc.Channel.Items) == 0 { + return nil, fmt.Errorf("parse feed: no channel found") + } + + out := &Channel{Title: strings.TrimSpace(doc.Channel.Title)} + for _, item := range doc.Channel.Items { + if !isAudio(item.Enclosure.Type, item.Enclosure.URL) { + continue + } + out.Episodes = append(out.Episodes, Episode{ + GUID: strings.TrimSpace(item.GUID), + Title: strings.TrimSpace(item.Title), + EnclosureURL: strings.TrimSpace(item.Enclosure.URL), + EnclosureType: strings.TrimSpace(item.Enclosure.Type), + PublishedUnix: parseDate(item.PubDate), + DeclaredDurationMs: parseDuration(item.Duration), + }) + } + + // Newest first. Feeds are not reliably ordered, and the retention rule depends on this. + sort.SliceStable(out.Episodes, func(i, j int) bool { + return out.Episodes[i].PublishedUnix > out.Episodes[j].PublishedUnix + }) + return out, nil +} + +func isAudio(mediaType, url string) bool { + if url == "" { + return false + } + if strings.HasPrefix(strings.ToLower(mediaType), "audio/") { + return true + } + // Some feeds omit or mistype the enclosure type; fall back to the extension rather than + // dropping a real episode over a publisher's metadata slip. + if mediaType == "" { + lower := strings.ToLower(url) + for _, ext := range []string{".mp3", ".m4a", ".aac", ".ogg", ".opus", ".flac", ".wav"} { + if strings.Contains(lower, ext) { + return true + } + } + } + return false +} + +// dateLayouts covers what podcast feeds actually emit, which is RFC 1123 with and without a +// named zone plus a few near-misses. +var dateLayouts = []string{ + time.RFC1123Z, time.RFC1123, time.RFC822Z, time.RFC822, time.RFC3339, + "Mon, 2 Jan 2006 15:04:05 -0700", + "Mon, 2 Jan 2006 15:04:05 MST", + "Mon, 02 Jan 2006 15:04 -0700", + "2 Jan 2006 15:04:05 -0700", +} + +func parseDate(raw string) int64 { + raw = strings.TrimSpace(raw) + if raw == "" { + return 0 + } + for _, layout := range dateLayouts { + if t, err := time.Parse(layout, raw); err == nil { + return t.Unix() + } + } + return 0 +} + +// parseDuration reads itunes:duration, which appears as bare seconds, mm:ss, or hh:mm:ss. +func parseDuration(raw string) int64 { + raw = strings.TrimSpace(raw) + if raw == "" { + return 0 + } + parts := strings.Split(raw, ":") + var total int64 + for _, p := range parts { + n, err := strconv.ParseInt(strings.TrimSpace(p), 10, 64) + if err != nil { + return 0 + } + total = total*60 + n + } + return total * 1000 +} diff --git a/server/internal/feed/feed_test.go b/server/internal/feed/feed_test.go new file mode 100644 index 0000000..e62558f --- /dev/null +++ b/server/internal/feed/feed_test.go @@ -0,0 +1,145 @@ +package feed + +import "testing" + +// Real podcast feeds, trimmed. The shapes here are the ones that actually caused trouble in +// the Android parser and are worth not relearning: a GUID that is not a URL, an enclosure whose +// type is video, CDATA titles, and an itunes:duration in three different formats. + +const sample = `<?xml version="1.0" encoding="UTF-8"?> +<rss version="2.0" xmlns:itunes="http://www.itunes.com/dtds/podcast-1.0.dtd"> +<channel> + <title>A Show</title> + <item> + <title><![CDATA[Newest Episode]]></title> + <guid isPermaLink="false">tag:example,2026:1</guid> + <pubDate>Fri, 03 Oct 2026 10:00:00 -0000</pubDate> + <itunes:duration>00:22:05</itunes:duration> + <enclosure url="https://cdn.example/1.mp3" type="audio/mpeg" length="21000000"/> + </item> + <item> + <title>Second Episode</title> + <guid>https://example.com/2</guid> + <pubDate>Fri, 26 Sep 2026 10:00:00 -0000</pubDate> + <itunes:duration>1325</itunes:duration> + <enclosure url="https://cdn.example/2.mp3" type="audio/mpeg"/> + </item> + <item> + <title>A Video One</title> + <guid>https://example.com/3</guid> + <pubDate>Fri, 19 Sep 2026 10:00:00 -0000</pubDate> + <enclosure url="https://cdn.example/3.mp4" type="video/mp4"/> + </item> + <item> + <title>No Enclosure At All</title> + <guid>https://example.com/4</guid> + </item> + <item> + <title>Minutes And Seconds</title> + <guid>https://example.com/5</guid> + <pubDate>Fri, 12 Sep 2026 10:00:00 -0000</pubDate> + <itunes:duration>31:02</itunes:duration> + <enclosure url="https://cdn.example/5.mp3" type="audio/mpeg"/> + </item> +</channel> +</rss>` + +func TestParsesTheChannelTitle(t *testing.T) { + ch, err := Parse([]byte(sample)) + if err != nil { + t.Fatal(err) + } + if ch.Title != "A Show" { + t.Errorf("title %q, want %q", ch.Title, "A Show") + } +} + +// Audio only. Video was cut from the app's scope, and an episode whose audio comes from yt-dlp +// has no enclosure the server could fetch anyway. +func TestKeepsOnlyAudioEnclosures(t *testing.T) { + ch, err := Parse([]byte(sample)) + if err != nil { + t.Fatal(err) + } + if len(ch.Episodes) != 3 { + var got []string + for _, e := range ch.Episodes { + got = append(got, e.Title) + } + t.Fatalf("kept %d episodes (%v), want 3 audio ones", len(ch.Episodes), got) + } + for _, e := range ch.Episodes { + if e.EnclosureURL == "https://cdn.example/3.mp4" { + t.Error("kept a video enclosure") + } + } +} + +func TestReadsTheFieldsNeededToIdentifyAnEpisode(t *testing.T) { + ch, err := Parse([]byte(sample)) + if err != nil { + t.Fatal(err) + } + first := ch.Episodes[0] + if first.Title != "Newest Episode" { + t.Errorf("title %q, want CDATA content unwrapped", first.Title) + } + if first.GUID != "tag:example,2026:1" { + t.Errorf("guid %q, want the non-URL guid verbatim", first.GUID) + } + if first.EnclosureURL != "https://cdn.example/1.mp3" { + t.Errorf("enclosure %q", first.EnclosureURL) + } + if first.PublishedUnix == 0 { + t.Error("published date not parsed") + } +} + +// A declared duration is the feed's claim and routinely wrong, but it is still what decides +// which episodes look newest before anything has been downloaded. +func TestParsesEveryDurationFormatFeedsActuallyUse(t *testing.T) { + ch, err := Parse([]byte(sample)) + if err != nil { + t.Fatal(err) + } + want := map[string]int64{ + "Newest Episode": 1325000, // 00:22:05 + "Second Episode": 1325000, // bare seconds + "Minutes And Seconds": 1862000, // 31:02 + } + for _, e := range ch.Episodes { + if w, ok := want[e.Title]; ok && e.DeclaredDurationMs != w { + t.Errorf("%s: duration %dms, want %dms", e.Title, e.DeclaredDurationMs, w) + } + } +} + +// Newest first, because that is the order the retention rule needs and feeds are not reliably +// sorted. +func TestOrdersEpisodesNewestFirst(t *testing.T) { + ch, err := Parse([]byte(sample)) + if err != nil { + t.Fatal(err) + } + for i := 1; i < len(ch.Episodes); i++ { + if ch.Episodes[i-1].PublishedUnix < ch.Episodes[i].PublishedUnix { + t.Errorf("episode %d is newer than %d", i, i-1) + } + } +} + +func TestRejectsSomethingThatIsNotAFeed(t *testing.T) { + if _, err := Parse([]byte("not xml at all")); err == nil { + t.Error("expected an error") + } +} + +func TestAFeedWithNoItemsIsNotAnError(t *testing.T) { + ch, err := Parse([]byte(`<rss><channel><title>Empty</title></channel></rss>`)) + if err != nil { + t.Fatal(err) + } + if len(ch.Episodes) != 0 { + t.Errorf("got %d episodes", len(ch.Episodes)) + } +} diff --git a/server/internal/mirror/mirror.go b/server/internal/mirror/mirror.go new file mode 100644 index 0000000..9e91041 --- /dev/null +++ b/server/internal/mirror/mirror.go @@ -0,0 +1,156 @@ +// Package mirror keeps a local copy of the audio the server detects against and the phone +// plays. +package mirror + +import ( + "context" + "fmt" + "io" + "net/http" + "net/url" + "os" + "path" + "path/filepath" + "strings" + "time" + + "github.com/terst/listen/server/internal/store" +) + +// Probe measures a downloaded file's real length. +type Probe interface { + DurationMs(ctx context.Context, path string) (int64, error) +} + +// Mirror downloads episodes and reclaims them. +// +// The reason the server holds the audio at all, rather than only timestamps: an episode fetched +// twice is not the same file when a show uses dynamic ad insertion, so an offset measured on one +// copy points at the wrong moment in another. The phone playing the same bytes the detector +// examined is what makes the offsets correct. +type Mirror struct { + AudioDir string + Probe Probe + // Client defaults to a client with a generous timeout: episodes are tens of megabytes. + Client *http.Client + // UserAgent identifies this server to publishers. + UserAgent string +} + +// DefaultTimeout bounds one episode download. +const DefaultTimeout = 15 * time.Minute + +func (m *Mirror) client() *http.Client { + if m.Client != nil { + return m.Client + } + return &http.Client{Timeout: DefaultTimeout} +} + +func (m *Mirror) userAgent() string { + if m.UserAgent != "" { + return m.UserAgent + } + return "listen-server/1.0 (+https://terst.org)" +} + +// Fetch downloads an episode's audio and records where it went and how long it really is. +// +// Writes to a temporary file and renames, so an interrupted download cannot leave a truncated +// file that looks mirrored -- which would have the phone play a cut-off episode and the +// detector correlate against noise. +func (m *Mirror) Fetch(ctx context.Context, e *store.Episode) error { + if e.EnclosureURL == "" { + return fmt.Errorf("episode %s: nothing to download", e.ID) + } + if err := os.MkdirAll(m.AudioDir, 0o755); err != nil { + return fmt.Errorf("episode %s: %w", e.ID, err) + } + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, e.EnclosureURL, nil) + if err != nil { + return fmt.Errorf("episode %s: %w", e.ID, err) + } + req.Header.Set("User-Agent", m.userAgent()) + + resp, err := m.client().Do(req) + if err != nil { + return fmt.Errorf("episode %s: %w", e.ID, err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("episode %s: %s returned %s", e.ID, e.EnclosureURL, resp.Status) + } + + name := e.ID + extensionFor(e.EnclosureURL, resp.Header.Get("Content-Type")) + tmp, err := os.CreateTemp(m.AudioDir, ".download-*") + if err != nil { + return fmt.Errorf("episode %s: %w", e.ID, err) + } + tmpName := tmp.Name() + defer os.Remove(tmpName) + + if _, err := io.Copy(tmp, resp.Body); err != nil { + tmp.Close() + return fmt.Errorf("episode %s: %w", e.ID, err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("episode %s: %w", e.ID, err) + } + + final := filepath.Join(m.AudioDir, name) + if err := os.Rename(tmpName, final); err != nil { + return fmt.Errorf("episode %s: %w", e.ID, err) + } + + // The real length, which is routinely not what the feed declared. + if ms, err := m.Probe.DurationMs(ctx, final); err == nil { + e.DurationMs = ms + } + e.Path = name + return nil +} + +// Prune deletes an episode's audio, leaving everything detection decided about it in place. +// +// The offsets describe the episode, not the file, and re-deriving them would mean downloading +// it again -- so they outlive the audio. +func (m *Mirror) Prune(e *store.Episode) error { + if e.Path == "" { + return nil + } + err := os.Remove(filepath.Join(m.AudioDir, e.Path)) + e.Path = "" + if err != nil && !os.IsNotExist(err) { + return fmt.Errorf("episode %s: %w", e.ID, err) + } + return nil +} + +// extensionFor picks a file extension, preferring the URL's since a player chooses its decoder +// from it. +func extensionFor(rawURL, contentType string) string { + if u, err := url.Parse(rawURL); err == nil { + if ext := path.Ext(u.Path); isAudioExt(ext) { + return ext + } + } + switch { + case strings.Contains(contentType, "mp4"), strings.Contains(contentType, "m4a"): + return ".m4a" + case strings.Contains(contentType, "aac"): + return ".aac" + case strings.Contains(contentType, "ogg"), strings.Contains(contentType, "opus"): + return ".ogg" + default: + return ".mp3" + } +} + +func isAudioExt(ext string) bool { + switch strings.ToLower(ext) { + case ".mp3", ".m4a", ".m4b", ".aac", ".ogg", ".opus", ".flac", ".wav": + return true + } + return false +} diff --git a/server/internal/mirror/mirror_test.go b/server/internal/mirror/mirror_test.go new file mode 100644 index 0000000..b510451 --- /dev/null +++ b/server/internal/mirror/mirror_test.go @@ -0,0 +1,162 @@ +package mirror + +import ( + "context" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/terst/listen/server/internal/store" +) + +type stubProbe struct{ ms int64 } + +func (s stubProbe) DurationMs(context.Context, string) (int64, error) { return s.ms, nil } + +func episode(id, url string) *store.Episode { + return &store.Episode{ID: id, GUID: "g-" + id, EnclosureURL: url, Title: "An Episode"} +} + +func TestFetchWritesTheAudioAndRecordsWhereItWent(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + w.Header().Set("Content-Type", "audio/mpeg") + _, _ = w.Write([]byte("fake-mp3-bytes")) + })) + defer srv.Close() + + dir := t.TempDir() + m := &Mirror{AudioDir: dir, Probe: stubProbe{ms: 1325113}} + e := episode("abc", srv.URL+"/1.mp3") + + if err := m.Fetch(context.Background(), e); err != nil { + t.Fatal(err) + } + if e.Path == "" { + t.Fatal("Path not recorded") + } + raw, err := os.ReadFile(filepath.Join(dir, e.Path)) + if err != nil { + t.Fatal(err) + } + if string(raw) != "fake-mp3-bytes" { + t.Errorf("contents %q", raw) + } + // The measured length, not the feed's claim: everything downstream reads this. + if e.DurationMs != 1325113 { + t.Errorf("DurationMs %d, want 1325113", e.DurationMs) + } +} + +// The file is named by the episode id, which is stable and is what the audio URL already uses. +func TestFetchNamesTheFileByEpisodeId(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + _, _ = w.Write([]byte("x")) + })) + defer srv.Close() + + m := &Mirror{AudioDir: t.TempDir(), Probe: stubProbe{ms: 1}} + e := episode("deadbeef", srv.URL+"/whatever.m4a") + if err := m.Fetch(context.Background(), e); err != nil { + t.Fatal(err) + } + if !strings.HasPrefix(e.Path, "deadbeef") { + t.Errorf("Path %q does not start with the episode id", e.Path) + } + if filepath.Ext(e.Path) != ".m4a" { + t.Errorf("Path %q lost the extension, which the player needs to pick a decoder", e.Path) + } +} + +// A failed or interrupted download must not leave a half file that looks mirrored. That is the +// state that would have the phone play a truncated episode and the detector read noise. +func TestAFailedFetchLeavesNothingBehind(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + http.Error(w, "gone", http.StatusNotFound) + })) + defer srv.Close() + + dir := t.TempDir() + m := &Mirror{AudioDir: dir, Probe: stubProbe{ms: 1}} + e := episode("abc", srv.URL+"/1.mp3") + + if err := m.Fetch(context.Background(), e); err == nil { + t.Fatal("expected an error") + } + if e.Path != "" { + t.Errorf("recorded a path for a download that failed: %q", e.Path) + } + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + if len(entries) != 0 { + t.Errorf("left %d files behind", len(entries)) + } +} + +func TestFetchRefusesAnEpisodeWithNoEnclosure(t *testing.T) { + m := &Mirror{AudioDir: t.TempDir(), Probe: stubProbe{ms: 1}} + if err := m.Fetch(context.Background(), episode("abc", "")); err == nil { + t.Error("expected an error for an episode with nothing to download") + } +} + +func TestPruneRemovesTheFileAndForgetsThePath(t *testing.T) { + dir := t.TempDir() + m := &Mirror{AudioDir: dir, Probe: stubProbe{ms: 1}} + e := episode("abc", "https://cdn/1.mp3") + e.Path = "abc.mp3" + if err := os.WriteFile(filepath.Join(dir, e.Path), []byte("bytes"), 0o644); err != nil { + t.Fatal(err) + } + + if err := m.Prune(e); err != nil { + t.Fatal(err) + } + if e.Path != "" { + t.Errorf("Path still set: %q", e.Path) + } + if _, err := os.Stat(filepath.Join(dir, "abc.mp3")); !os.IsNotExist(err) { + t.Error("file still on disk") + } +} + +// Detected offsets outlive the audio: they describe the episode, and re-deriving them would +// mean downloading it again. +func TestPruneKeepsWhatDetectionDecided(t *testing.T) { + dir := t.TempDir() + m := &Mirror{AudioDir: dir, Probe: stubProbe{ms: 1}} + e := episode("abc", "https://cdn/1.mp3") + e.Path = "abc.mp3" + intro := int64(16652) + e.IntroMs = &intro + e.DurationMs = 1325113 + if err := os.WriteFile(filepath.Join(dir, e.Path), []byte("bytes"), 0o644); err != nil { + t.Fatal(err) + } + + if err := m.Prune(e); err != nil { + t.Fatal(err) + } + if e.IntroMs == nil || *e.IntroMs != 16652 { + t.Errorf("lost the detected intro: %v", e.IntroMs) + } + if e.DurationMs != 1325113 { + t.Errorf("lost the measured duration: %d", e.DurationMs) + } +} + +func TestPruningSomethingAlreadyGoneIsFine(t *testing.T) { + m := &Mirror{AudioDir: t.TempDir(), Probe: stubProbe{ms: 1}} + e := episode("abc", "https://cdn/1.mp3") + e.Path = "not-there.mp3" + if err := m.Prune(e); err != nil { + t.Errorf("pruning a missing file errored: %v", err) + } + if e.Path != "" { + t.Error("Path not cleared") + } +} diff --git a/server/internal/store/id.go b/server/internal/store/id.go new file mode 100644 index 0000000..52bab46 --- /dev/null +++ b/server/internal/store/id.go @@ -0,0 +1,20 @@ +package store + +import ( + "crypto/sha256" + "encoding/hex" +) + +// newID is a stable identifier for an episode, derived from what identifies it rather than +// from a counter or a random value. +// +// Stable matters because it names the audio file on disk and appears in the URL the phone +// fetches: a regenerated id would orphan the file and break a link the phone already holds. +func newID(feedURL, guid, enclosureURL string) string { + key := guid + if key == "" { + key = enclosureURL + } + sum := sha256.Sum256([]byte(feedURL + "\x00" + key)) + return hex.EncodeToString(sum[:12]) +} diff --git a/server/internal/store/store.go b/server/internal/store/store.go new file mode 100644 index 0000000..cfe6f40 --- /dev/null +++ b/server/internal/store/store.go @@ -0,0 +1,295 @@ +// Package store holds what the server knows: which feeds to poll, which episodes exist, where +// their audio is, and what detection decided about them. +package store + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + "sync" + + "github.com/terst/listen/server/internal/feed" +) + +// Episode is one episode's full server-side record. +type Episode struct { + ID string `json:"id"` + GUID string `json:"guid"` + Title string `json:"title"` + EnclosureURL string `json:"enclosureUrl"` + PublishedUnix int64 `json:"publishedUnix"` + // DeclaredDurationMs is the feed's claim; DurationMs is what the file turned out to be. + DeclaredDurationMs int64 `json:"declaredDurationMs"` + DurationMs int64 `json:"durationMs,omitempty"` + + // Path is where the mirrored audio sits, relative to the audio directory. Empty until + // the episode has been downloaded. + Path string `json:"path,omitempty"` + + // IntroMs and OutroMs are what the phone needs, in this episode's own timeline. Nil until + // detection has confirmed something that passed its self-check. + IntroMs *int64 `json:"introMs,omitempty"` + OutroMs *int64 `json:"outroMs,omitempty"` + + // DetectedAtUnix records when detection last ran over the show this belongs to, so a + // result can be told apart from never having looked. + DetectedAtUnix int64 `json:"detectedAtUnix,omitempty"` +} + +// Mirrored reports whether this episode's audio is on disk. +func (e *Episode) Mirrored() bool { return e.Path != "" } + +// Show is a subscribed feed and everything known about its episodes. +type Show struct { + FeedURL string `json:"feedUrl"` + Title string `json:"title"` + Episodes []*Episode `json:"episodes"` + + // LastPolledUnix and LastError record the most recent refresh, so a feed that is failing + // says so instead of silently having nothing. + LastPolledUnix int64 `json:"lastPolledUnix,omitempty"` + LastError string `json:"lastError,omitempty"` +} + +// Partition splits a show's episodes into the newest keep and everything else, newest first. +// +// By publish date rather than feed order, because feeds are not reliably sorted and the whole +// point is to mirror what is current. +func (s *Show) Partition(keep int) (kept, pruned []*Episode) { + ordered := make([]*Episode, len(s.Episodes)) + copy(ordered, s.Episodes) + sort.SliceStable(ordered, func(i, j int) bool { + return ordered[i].PublishedUnix > ordered[j].PublishedUnix + }) + if keep >= len(ordered) { + return ordered, nil + } + return ordered[:keep], ordered[keep:] +} + +// State is the server's whole persistent world. +type State struct { + mu sync.RWMutex + feeds []string + shows map[string]*Show +} + +type persisted struct { + Feeds []string `json:"feeds"` + Shows map[string]*Show `json:"shows"` +} + +// New returns an empty state. +func New() *State { + return &State{shows: map[string]*Show{}} +} + +// Load reads state from disk. A missing file is a first run, not an error. +func Load(path string) (*State, error) { + raw, err := os.ReadFile(path) + if os.IsNotExist(err) { + return New(), nil + } + if err != nil { + return nil, fmt.Errorf("load state: %w", err) + } + var p persisted + if err := json.Unmarshal(raw, &p); err != nil { + return nil, fmt.Errorf("load state: %w", err) + } + s := New() + s.feeds = p.Feeds + if p.Shows != nil { + s.shows = p.Shows + } + return s, nil +} + +// Save writes state to disk atomically. +// +// Via a temporary file and a rename, because this holds the only record of where mirrored audio +// is and what detection decided: a truncated write would orphan every file on disk and throw +// away hours of detection. +func (s *State) Save(path string) error { + s.mu.RLock() + raw, err := json.MarshalIndent(persisted{Feeds: s.feeds, Shows: s.shows}, "", " ") + s.mu.RUnlock() + if err != nil { + return fmt.Errorf("save state: %w", err) + } + + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return fmt.Errorf("save state: %w", err) + } + tmp, err := os.CreateTemp(filepath.Dir(path), ".state-*.tmp") + if err != nil { + return fmt.Errorf("save state: %w", err) + } + tmpName := tmp.Name() + defer os.Remove(tmpName) + + if _, err := tmp.Write(raw); err != nil { + tmp.Close() + return fmt.Errorf("save state: %w", err) + } + if err := tmp.Sync(); err != nil { + tmp.Close() + return fmt.Errorf("save state: %w", err) + } + if err := tmp.Close(); err != nil { + return fmt.Errorf("save state: %w", err) + } + if err := os.Rename(tmpName, path); err != nil { + return fmt.Errorf("save state: %w", err) + } + return nil +} + +// SetFeeds replaces the subscription list and reports the feeds that are no longer wanted. +// +// Wholesale replacement because the phone owns this list; reconciling two editable copies of +// something changed this rarely would cost more in sync bugs than it saves. +func (s *State) SetFeeds(feeds []string) (removed []string) { + s.mu.Lock() + defer s.mu.Unlock() + + wanted := make(map[string]bool, len(feeds)) + clean := make([]string, 0, len(feeds)) + for _, f := range feeds { + f = strings.TrimSpace(f) + if f == "" || wanted[f] { + continue + } + wanted[f] = true + clean = append(clean, f) + } + for url := range s.shows { + if !wanted[url] { + removed = append(removed, url) + } + } + sort.Strings(removed) + s.feeds = clean + return removed +} + +// Feeds is the current subscription list. +func (s *State) Feeds() []string { + s.mu.RLock() + defer s.mu.RUnlock() + return append([]string(nil), s.feeds...) +} + +// Forget drops a show and returns it, so its files can be reclaimed. +func (s *State) Forget(feedURL string) *Show { + s.mu.Lock() + defer s.mu.Unlock() + show := s.shows[feedURL] + delete(s.shows, feedURL) + return show +} + +// Show returns a feed's show, or nil. +func (s *State) Show(feedURL string) *Show { + s.mu.RLock() + defer s.mu.RUnlock() + return s.shows[feedURL] +} + +// Shows returns every tracked show. +func (s *State) Shows() []*Show { + s.mu.RLock() + defer s.mu.RUnlock() + out := make([]*Show, 0, len(s.shows)) + for _, sh := range s.shows { + out = append(out, sh) + } + sort.Slice(out, func(i, j int) bool { return out[i].FeedURL < out[j].FeedURL }) + return out +} + +// Merge folds a parsed feed into stored state, adding episodes and refreshing feed-owned +// fields while leaving everything the server learned alone. +// +// The distinction matters and has bitten the Android side twice: feed metadata is rewritten on +// every refresh, so anything measured or detected has to live in fields the merge does not +// touch, or it silently reverts. +func (s *State) Merge(feedURL, title string, episodes []feed.Episode) *Show { + s.mu.Lock() + defer s.mu.Unlock() + + show := s.shows[feedURL] + if show == nil { + show = &Show{FeedURL: feedURL} + s.shows[feedURL] = show + } + if title != "" { + show.Title = title + } + + for _, parsed := range episodes { + existing := matchIn(show, parsed.GUID, parsed.EnclosureURL, parsed.Title) + if existing == nil { + show.Episodes = append(show.Episodes, &Episode{ + ID: newID(feedURL, parsed.GUID, parsed.EnclosureURL), + GUID: parsed.GUID, + Title: parsed.Title, + EnclosureURL: parsed.EnclosureURL, + PublishedUnix: parsed.PublishedUnix, + DeclaredDurationMs: parsed.DeclaredDurationMs, + }) + continue + } + // Feed-owned fields only. Path, DurationMs, IntroMs, OutroMs are the server's. + existing.GUID = parsed.GUID + existing.Title = parsed.Title + existing.EnclosureURL = parsed.EnclosureURL + if parsed.PublishedUnix != 0 { + existing.PublishedUnix = parsed.PublishedUnix + } + existing.DeclaredDurationMs = parsed.DeclaredDurationMs + } + return show +} + +// Match finds an episode within a show, on GUID, then enclosure URL, then title. +// +// That cascade exists because publishers rewrite feeds: an episode that loses its identity +// loses its playback position on the phone and its detected offsets here. +func (s *State) Match(feedURL, guid, enclosureURL, title string) *Episode { + s.mu.RLock() + defer s.mu.RUnlock() + show := s.shows[feedURL] + if show == nil { + return nil + } + return matchIn(show, guid, enclosureURL, title) +} + +func matchIn(show *Show, guid, enclosureURL, title string) *Episode { + if guid != "" { + for _, e := range show.Episodes { + if e.GUID == guid { + return e + } + } + } + if enclosureURL != "" { + for _, e := range show.Episodes { + if e.EnclosureURL == enclosureURL { + return e + } + } + } + if title != "" { + for _, e := range show.Episodes { + if e.Title == title { + return e + } + } + } + return nil +} diff --git a/server/internal/store/store_test.go b/server/internal/store/store_test.go new file mode 100644 index 0000000..cd5cf6f --- /dev/null +++ b/server/internal/store/store_test.go @@ -0,0 +1,239 @@ +package store + +import ( + "os" + "path/filepath" + "testing" + + "github.com/terst/listen/server/internal/feed" +) + +func ep(guid, title, enclosure string, published int64) feed.Episode { + return feed.Episode{GUID: guid, Title: title, EnclosureURL: enclosure, PublishedUnix: published} +} + +// Merging a refreshed feed into stored state must not lose what the server has learned. The +// Android side has already had this bug twice in different forms -- a corrected duration +// overwritten on every refresh, and episode rows cascading away carrying the only record of +// where their files were. + +func TestMergeKeepsWhatTheServerLearnedAboutAnEpisode(t *testing.T) { + s := New() + s.Merge("https://f", "A Show", []feed.Episode{ep("g1", "One", "https://cdn/1.mp3", 100)}) + + got := s.Match("https://f", "g1", "", "") + if got == nil { + t.Fatal("episode not stored") + } + got.Path = "audio/abc.mp3" + got.DurationMs = 1325113 + intro := int64(16652) + got.IntroMs = &intro + + // The same feed again, as a refresh would deliver it. + s.Merge("https://f", "A Show", []feed.Episode{ep("g1", "One", "https://cdn/1.mp3", 100)}) + + after := s.Match("https://f", "g1", "", "") + if after == nil { + t.Fatal("episode lost on refresh") + } + if after.Path != "audio/abc.mp3" { + t.Errorf("mirrored path lost: %q", after.Path) + } + if after.DurationMs != 1325113 { + t.Errorf("measured duration lost: %d", after.DurationMs) + } + if after.IntroMs == nil || *after.IntroMs != 16652 { + t.Errorf("detected intro lost: %v", after.IntroMs) + } +} + +func TestMergeAddsNewEpisodesAndKeepsOldOnes(t *testing.T) { + s := New() + s.Merge("https://f", "A Show", []feed.Episode{ep("g1", "One", "https://cdn/1.mp3", 100)}) + s.Merge("https://f", "A Show", []feed.Episode{ + ep("g2", "Two", "https://cdn/2.mp3", 200), + ep("g1", "One", "https://cdn/1.mp3", 100), + }) + + if n := len(s.Show("https://f").Episodes); n != 2 { + t.Errorf("%d episodes, want 2", n) + } +} + +// GUIDs move: publishers rewrite feeds. Matching on the GUID alone loses an episode's +// identity, and with it its playback position on the phone and its detected offsets here. +func TestMatchFallsBackToTheEnclosureUrlWhenTheGuidChanged(t *testing.T) { + s := New() + s.Merge("https://f", "A Show", []feed.Episode{ep("old-guid", "One", "https://cdn/1.mp3", 100)}) + + got := s.Match("https://f", "new-guid", "https://cdn/1.mp3", "") + if got == nil { + t.Fatal("not matched by enclosure url") + } + if got.GUID != "old-guid" { + t.Errorf("matched the wrong episode: %q", got.GUID) + } +} + +func TestMatchFallsBackToTheTitleWhenBothGuidAndUrlChanged(t *testing.T) { + s := New() + s.Merge("https://f", "A Show", []feed.Episode{ep("old-guid", "Grass Apes", "https://cdn/old.mp3", 100)}) + + got := s.Match("https://f", "new-guid", "https://cdn/new.mp3", "Grass Apes") + if got == nil { + t.Fatal("not matched by title") + } + if got.GUID != "old-guid" { + t.Errorf("matched the wrong episode: %q", got.GUID) + } +} + +// The cascade has to be in that order: a GUID match must win over an enclosure match, or a +// re-uploaded file would steal another episode's identity. +func TestMatchPrefersTheGuidOverTheEnclosureUrl(t *testing.T) { + s := New() + s.Merge("https://f", "A Show", []feed.Episode{ + ep("g1", "One", "https://cdn/shared.mp3", 100), + ep("g2", "Two", "https://cdn/shared.mp3", 200), + }) + + got := s.Match("https://f", "g2", "https://cdn/shared.mp3", "") + if got == nil || got.GUID != "g2" { + t.Fatalf("matched %v, want g2", got) + } +} + +func TestMatchReturnsNothingForAnUnknownEpisode(t *testing.T) { + s := New() + s.Merge("https://f", "A Show", []feed.Episode{ep("g1", "One", "https://cdn/1.mp3", 100)}) + + if got := s.Match("https://f", "nope", "https://cdn/nope.mp3", "Nope"); got != nil { + t.Errorf("matched %q for an unknown episode", got.GUID) + } +} + +func TestMatchIsScopedToTheShow(t *testing.T) { + s := New() + s.Merge("https://a", "Show A", []feed.Episode{ep("g1", "One", "https://cdn/1.mp3", 100)}) + s.Merge("https://b", "Show B", []feed.Episode{}) + + if got := s.Match("https://b", "g1", "", ""); got != nil { + t.Error("matched an episode from a different show") + } +} + +// Retention: newest N by publish date. Anything the server is mid-way through using has to +// survive, which is the lesson from the phone's own retention reclaiming the file detection +// was holding. +func TestKeepNewestSelectsByPublishDateNotFeedOrder(t *testing.T) { + s := New() + s.Merge("https://f", "A Show", []feed.Episode{ + ep("g1", "Oldest", "https://cdn/1.mp3", 100), + ep("g3", "Newest", "https://cdn/3.mp3", 300), + ep("g2", "Middle", "https://cdn/2.mp3", 200), + }) + + keep, prune := s.Show("https://f").Partition(2) + if len(keep) != 2 || len(prune) != 1 { + t.Fatalf("keep %d prune %d, want 2 and 1", len(keep), len(prune)) + } + if keep[0].GUID != "g3" || keep[1].GUID != "g2" { + t.Errorf("kept %s and %s, want g3 and g2", keep[0].GUID, keep[1].GUID) + } + if prune[0].GUID != "g1" { + t.Errorf("pruned %s, want g1", prune[0].GUID) + } +} + +func TestKeepNewestKeepsEverythingWhenThereAreFewerThanTheLimit(t *testing.T) { + s := New() + s.Merge("https://f", "A Show", []feed.Episode{ep("g1", "One", "https://cdn/1.mp3", 100)}) + + keep, prune := s.Show("https://f").Partition(8) + if len(keep) != 1 || len(prune) != 0 { + t.Errorf("keep %d prune %d, want 1 and 0", len(keep), len(prune)) + } +} + +// Subscriptions are replaced wholesale by the phone, which owns the list. A show that is gone +// from the list stops being polled, but its files are the mirror's business to clean up. +func TestSetFeedsReplacesTheSubscriptionList(t *testing.T) { + s := New() + s.Merge("https://a", "A", nil) + s.Merge("https://b", "B", nil) + + removed := s.SetFeeds([]string{"https://b", "https://c"}) + + if len(removed) != 1 || removed[0] != "https://a" { + t.Errorf("removed %v, want [https://a]", removed) + } + got := s.Feeds() + if len(got) != 2 { + t.Errorf("feeds %v, want 2", got) + } +} + +func TestSaveAndLoadRoundTripsEverything(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "state.json") + + s := New() + s.SetFeeds([]string{"https://f"}) + s.Merge("https://f", "A Show", []feed.Episode{ep("g1", "One", "https://cdn/1.mp3", 100)}) + e := s.Match("https://f", "g1", "", "") + e.Path = "audio/abc.mp3" + intro := int64(16652) + e.IntroMs = &intro + + if err := s.Save(path); err != nil { + t.Fatal(err) + } + loaded, err := Load(path) + if err != nil { + t.Fatal(err) + } + + got := loaded.Match("https://f", "g1", "", "") + if got == nil { + t.Fatal("episode did not survive the round trip") + } + if got.Path != "audio/abc.mp3" || got.IntroMs == nil || *got.IntroMs != 16652 { + t.Errorf("round trip lost data: %+v", got) + } + if len(loaded.Feeds()) != 1 { + t.Errorf("feeds did not survive: %v", loaded.Feeds()) + } +} + +// A missing state file is a first run, not a failure. +func TestLoadingAMissingFileGivesAnEmptyState(t *testing.T) { + s, err := Load(filepath.Join(t.TempDir(), "absent.json")) + if err != nil { + t.Fatal(err) + } + if len(s.Feeds()) != 0 { + t.Errorf("expected no feeds, got %v", s.Feeds()) + } +} + +// Saving must not be able to truncate a good state file into a broken one. +func TestSaveIsAtomic(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "state.json") + s := New() + s.SetFeeds([]string{"https://f"}) + if err := s.Save(path); err != nil { + t.Fatal(err) + } + + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + for _, entry := range entries { + if entry.Name() != "state.json" { + t.Errorf("left a stray file behind: %s", entry.Name()) + } + } +} |