mirror of
https://github.com/tapframe/NuvioStreaming.git
synced 2026-08-16 12:24:40 +00:00
Merge branch 'fix/sentrybatch' into cmp-rewrite
This commit is contained in:
tapframe
commit
31547fcd6a
20 changed files with 1295 additions and 316 deletions
|
|
@ -6,11 +6,14 @@ import android.content.SharedPreferences
|
|||
internal object PluginStorage {
|
||||
private const val preferencesName = "nuvio_plugins"
|
||||
private const val pluginsStateKey = "plugins_state"
|
||||
private const val scraperCodeDirectoryName = "nuvio_plugin_scrapers"
|
||||
|
||||
private var preferences: SharedPreferences? = null
|
||||
private var scraperCodeStore: PluginScraperCodeFileStore? = null
|
||||
|
||||
fun initialize(context: Context) {
|
||||
preferences = context.getSharedPreferences(preferencesName, Context.MODE_PRIVATE)
|
||||
scraperCodeStore = PluginScraperCodeFileStore(context.filesDir.resolve(scraperCodeDirectoryName))
|
||||
}
|
||||
|
||||
fun loadState(profileId: Int): String? =
|
||||
|
|
@ -23,6 +26,19 @@ internal object PluginStorage {
|
|||
?.apply()
|
||||
}
|
||||
|
||||
fun hasScraperCode(profileId: Int, scraperId: String): Boolean =
|
||||
scraperCodeStore?.contains(profileId, scraperId) == true
|
||||
|
||||
fun loadScraperCode(profileId: Int, scraperId: String): String? =
|
||||
scraperCodeStore?.load(profileId, scraperId)
|
||||
|
||||
fun saveScraperCode(
|
||||
profileId: Int,
|
||||
scraperId: String,
|
||||
code: String,
|
||||
overwrite: Boolean,
|
||||
): Boolean = scraperCodeStore?.save(profileId, scraperId, code, overwrite) == true
|
||||
|
||||
fun loadScraperSettings(scraperId: String): String? =
|
||||
preferences?.getString("settings_${scraperId}", null)
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,64 @@
|
|||
package com.nuvio.app.features.plugins
|
||||
|
||||
import kotlinx.atomicfu.locks.SynchronizedObject
|
||||
import kotlinx.atomicfu.locks.synchronized
|
||||
import java.io.File
|
||||
|
||||
internal class PluginScraperCodeFileStore(
|
||||
private val root: File,
|
||||
) {
|
||||
private val lock = SynchronizedObject()
|
||||
|
||||
fun contains(profileId: Int, scraperId: String): Boolean =
|
||||
scraperCodeFile(profileId, scraperId).isFile
|
||||
|
||||
fun load(profileId: Int, scraperId: String): String? = synchronized(lock) {
|
||||
val file = scraperCodeFile(profileId, scraperId).takeIf(File::isFile)
|
||||
?: return@synchronized null
|
||||
runCatching { file.readText() }.getOrNull()
|
||||
}
|
||||
|
||||
fun save(
|
||||
profileId: Int,
|
||||
scraperId: String,
|
||||
code: String,
|
||||
overwrite: Boolean,
|
||||
): Boolean {
|
||||
val target = scraperCodeFile(profileId, scraperId)
|
||||
if (!overwrite && target.isFile) return true
|
||||
return synchronized(lock) {
|
||||
if (!overwrite && target.isFile) return@synchronized true
|
||||
val directory = target.parentFile ?: return@synchronized false
|
||||
if (!directory.exists() && !directory.mkdirs()) return@synchronized false
|
||||
|
||||
val temporary = runCatching {
|
||||
File.createTempFile("scraper-", ".tmp", directory)
|
||||
}.getOrNull() ?: return@synchronized false
|
||||
val backup = directory.resolve("${target.name}.backup")
|
||||
|
||||
try {
|
||||
temporary.bufferedWriter().use { writer -> writer.write(code) }
|
||||
if (!overwrite && target.isFile) return@synchronized true
|
||||
if (backup.exists() && !backup.delete()) return@synchronized false
|
||||
val hadTarget = target.exists()
|
||||
if (hadTarget && !target.renameTo(backup)) return@synchronized false
|
||||
if (!temporary.renameTo(target)) {
|
||||
if (hadTarget) backup.renameTo(target)
|
||||
return@synchronized false
|
||||
}
|
||||
if (backup.exists()) backup.delete()
|
||||
true
|
||||
} catch (_: Throwable) {
|
||||
false
|
||||
} finally {
|
||||
if (temporary.exists()) temporary.delete()
|
||||
if (backup.exists() && !target.exists()) backup.renameTo(target)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private fun scraperCodeFile(profileId: Int, scraperId: String): File {
|
||||
val fileName = "${pluginDigestHex("SHA256", scraperId)}.js"
|
||||
return root.resolve(profileId.toString()).resolve(fileName)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,33 @@
|
|||
package com.nuvio.app.features.plugins
|
||||
|
||||
import java.nio.file.Files
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFalse
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class PluginScraperCodeFileStoreTest {
|
||||
@Test
|
||||
fun scraperCodeRoundTripsWithoutRewritingUnlessRequested() {
|
||||
val root = Files.createTempDirectory("nuvio-plugin-code").toFile()
|
||||
try {
|
||||
val store = PluginScraperCodeFileStore(root)
|
||||
val scraperId = "https://plugins.example/manifest.json:scraper"
|
||||
val original = "x".repeat(4 * 1024 * 1024)
|
||||
val refreshed = "updated scraper"
|
||||
|
||||
assertFalse(store.contains(1, scraperId))
|
||||
assertTrue(store.save(1, scraperId, original, overwrite = false))
|
||||
assertEquals(original, store.load(1, scraperId))
|
||||
|
||||
assertTrue(store.save(1, scraperId, refreshed, overwrite = false))
|
||||
assertEquals(original, store.load(1, scraperId))
|
||||
|
||||
assertTrue(store.save(1, scraperId, refreshed, overwrite = true))
|
||||
assertEquals(refreshed, store.load(1, scraperId))
|
||||
assertFalse(store.contains(2, scraperId))
|
||||
} finally {
|
||||
root.deleteRecursively()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -50,5 +50,6 @@ internal actual object PlatformLocalAccountDataCleaner {
|
|||
.clear()
|
||||
.apply()
|
||||
}
|
||||
context.filesDir.resolve("nuvio_plugin_scrapers").deleteRecursively()
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -4,7 +4,6 @@ import android.app.Notification
|
|||
import android.app.NotificationChannel
|
||||
import android.app.NotificationManager
|
||||
import android.app.PendingIntent
|
||||
import android.app.Service
|
||||
import android.content.BroadcastReceiver
|
||||
import android.content.Context
|
||||
import android.content.Intent
|
||||
|
|
@ -15,7 +14,6 @@ import android.media.session.MediaSession
|
|||
import android.media.session.PlaybackState
|
||||
import android.os.Build
|
||||
import android.os.Handler
|
||||
import android.os.IBinder
|
||||
import android.os.Looper
|
||||
import android.os.SystemClock
|
||||
import android.util.Log
|
||||
|
|
@ -29,9 +27,6 @@ import java.util.concurrent.Executors
|
|||
import java.util.concurrent.atomic.AtomicInteger
|
||||
import kotlin.math.abs
|
||||
|
||||
private const val NOW_PLAYING_TAG = "NuvioNowPlaying"
|
||||
private const val NOW_PLAYING_CHANNEL_ID = "nuvio_playback"
|
||||
private const val NOW_PLAYING_NOTIFICATION_ID = 0x4E55
|
||||
private const val SEEK_INTERVAL_MS = 10_000L
|
||||
private const val MAX_ARTWORK_DOWNLOAD_BYTES = 12 * 1024 * 1024
|
||||
private const val MAX_ARTWORK_EDGE_PX = 1_024
|
||||
|
|
@ -40,7 +35,6 @@ private const val ACTION_PLAY = "com.nuvio.app.nowplaying.PLAY"
|
|||
private const val ACTION_PAUSE = "com.nuvio.app.nowplaying.PAUSE"
|
||||
private const val ACTION_REWIND = "com.nuvio.app.nowplaying.REWIND"
|
||||
private const val ACTION_FAST_FORWARD = "com.nuvio.app.nowplaying.FAST_FORWARD"
|
||||
private const val ACTION_START_FOREGROUND = "com.nuvio.app.nowplaying.START_FOREGROUND"
|
||||
|
||||
private data class AndroidNowPlayingMetadata(
|
||||
val title: String,
|
||||
|
|
@ -350,69 +344,6 @@ class PlayerNowPlayingActionReceiver : BroadcastReceiver() {
|
|||
}
|
||||
}
|
||||
|
||||
class PlayerNowPlayingService : Service() {
|
||||
override fun onBind(intent: Intent?): IBinder? = null
|
||||
|
||||
override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
|
||||
if (intent?.action != ACTION_START_FOREGROUND) {
|
||||
stopSelf(startId)
|
||||
return START_NOT_STICKY
|
||||
}
|
||||
|
||||
val notification = PlayerNowPlayingServiceState.notification
|
||||
if (notification == null) {
|
||||
stopSelf(startId)
|
||||
return START_NOT_STICKY
|
||||
}
|
||||
|
||||
startForeground(NOW_PLAYING_NOTIFICATION_ID, notification)
|
||||
return START_NOT_STICKY
|
||||
}
|
||||
|
||||
override fun onDestroy() {
|
||||
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.N) {
|
||||
stopForeground(STOP_FOREGROUND_REMOVE)
|
||||
} else {
|
||||
@Suppress("DEPRECATION")
|
||||
stopForeground(true)
|
||||
}
|
||||
super.onDestroy()
|
||||
}
|
||||
|
||||
companion object {
|
||||
internal fun publish(context: Context, notification: Notification) {
|
||||
if (!AppFeaturePolicy.mediaPlaybackForegroundServiceEnabled) return
|
||||
PlayerNowPlayingServiceState.notification = notification
|
||||
val intent = Intent(context, PlayerNowPlayingService::class.java)
|
||||
.setAction(ACTION_START_FOREGROUND)
|
||||
runCatching {
|
||||
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
|
||||
context.startForegroundService(intent)
|
||||
} else {
|
||||
context.startService(intent)
|
||||
}
|
||||
context.getSystemService(NotificationManager::class.java)
|
||||
?.notify(NOW_PLAYING_NOTIFICATION_ID, notification)
|
||||
}.onFailure { error ->
|
||||
Log.w(NOW_PLAYING_TAG, "Unable to publish playback notification", error)
|
||||
}
|
||||
}
|
||||
|
||||
internal fun hide(context: Context) {
|
||||
if (!AppFeaturePolicy.mediaPlaybackForegroundServiceEnabled) return
|
||||
PlayerNowPlayingServiceState.notification = null
|
||||
runCatching { context.stopService(Intent(context, PlayerNowPlayingService::class.java)) }
|
||||
context.getSystemService(NotificationManager::class.java)
|
||||
?.cancel(NOW_PLAYING_NOTIFICATION_ID)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private object PlayerNowPlayingServiceState {
|
||||
@Volatile
|
||||
var notification: Notification? = null
|
||||
}
|
||||
|
||||
private object AndroidNowPlayingActionDispatcher {
|
||||
@Volatile
|
||||
private var controllerRef: WeakReference<AndroidPlayerNowPlayingController>? = null
|
||||
|
|
|
|||
|
|
@ -0,0 +1,154 @@
|
|||
package com.nuvio.app.features.player
|
||||
|
||||
import android.app.Notification
|
||||
import android.app.NotificationManager
|
||||
import android.app.Service
|
||||
import android.content.Context
|
||||
import android.content.Intent
|
||||
import android.os.Build
|
||||
import android.os.IBinder
|
||||
import android.util.Log
|
||||
import com.nuvio.app.core.build.AppFeaturePolicy
|
||||
import com.nuvio.app.core.concurrent.ConflatedTaskDispatcher
|
||||
import java.util.concurrent.Executors
|
||||
import java.util.concurrent.atomic.AtomicBoolean
|
||||
|
||||
internal const val NOW_PLAYING_TAG = "NuvioNowPlaying"
|
||||
internal const val NOW_PLAYING_CHANNEL_ID = "nuvio_playback"
|
||||
internal const val NOW_PLAYING_NOTIFICATION_ID = 0x4E55
|
||||
private const val ACTION_START_FOREGROUND = "com.nuvio.app.nowplaying.START_FOREGROUND"
|
||||
|
||||
class PlayerNowPlayingService : Service() {
|
||||
override fun onBind(intent: Intent?): IBinder? = null
|
||||
|
||||
override fun onStartCommand(intent: Intent?, flags: Int, startId: Int): Int {
|
||||
if (intent?.action != ACTION_START_FOREGROUND) {
|
||||
stopSelf(startId)
|
||||
return START_NOT_STICKY
|
||||
}
|
||||
|
||||
val notification = PlayerNowPlayingServiceState.notification
|
||||
if (notification == null) {
|
||||
stopSelf(startId)
|
||||
return START_NOT_STICKY
|
||||
}
|
||||
|
||||
startForeground(NOW_PLAYING_NOTIFICATION_ID, notification)
|
||||
return START_NOT_STICKY
|
||||
}
|
||||
|
||||
override fun onDestroy() {
|
||||
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.N) {
|
||||
stopForeground(STOP_FOREGROUND_REMOVE)
|
||||
} else {
|
||||
@Suppress("DEPRECATION")
|
||||
stopForeground(true)
|
||||
}
|
||||
PlayerNowPlayingServiceController.onServiceDestroyed(applicationContext)
|
||||
super.onDestroy()
|
||||
}
|
||||
|
||||
companion object {
|
||||
internal fun publish(context: Context, notification: Notification) {
|
||||
if (!AppFeaturePolicy.mediaPlaybackForegroundServiceEnabled) return
|
||||
PlayerNowPlayingServiceController.publish(context.applicationContext, notification)
|
||||
}
|
||||
|
||||
internal fun hide(context: Context) {
|
||||
if (!AppFeaturePolicy.mediaPlaybackForegroundServiceEnabled) return
|
||||
PlayerNowPlayingServiceController.hide(context.applicationContext)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private sealed interface PlayerNowPlayingServiceCommand {
|
||||
data class Publish(
|
||||
val context: Context,
|
||||
val notification: Notification,
|
||||
) : PlayerNowPlayingServiceCommand
|
||||
|
||||
data class Hide(val context: Context) : PlayerNowPlayingServiceCommand
|
||||
}
|
||||
|
||||
private object PlayerNowPlayingServiceController {
|
||||
private val executor = Executors.newSingleThreadExecutor { runnable ->
|
||||
Thread(runnable, "NuvioNowPlayingService").apply { isDaemon = true }
|
||||
}
|
||||
private val startRequested = AtomicBoolean(false)
|
||||
private val commands = ConflatedTaskDispatcher<PlayerNowPlayingServiceCommand>(
|
||||
schedule = { task -> executor.execute(task) },
|
||||
consume = ::execute,
|
||||
)
|
||||
|
||||
fun publish(context: Context, notification: Notification) {
|
||||
PlayerNowPlayingServiceState.notification = notification
|
||||
commands.dispatch(PlayerNowPlayingServiceCommand.Publish(context, notification))
|
||||
}
|
||||
|
||||
fun hide(context: Context) {
|
||||
PlayerNowPlayingServiceState.notification = null
|
||||
commands.dispatch(PlayerNowPlayingServiceCommand.Hide(context))
|
||||
}
|
||||
|
||||
fun onServiceDestroyed(context: Context) {
|
||||
startRequested.set(false)
|
||||
val notification = PlayerNowPlayingServiceState.notification ?: return
|
||||
commands.dispatch(PlayerNowPlayingServiceCommand.Publish(context, notification))
|
||||
}
|
||||
|
||||
private fun execute(command: PlayerNowPlayingServiceCommand) {
|
||||
when (command) {
|
||||
is PlayerNowPlayingServiceCommand.Publish -> publishNow(command)
|
||||
is PlayerNowPlayingServiceCommand.Hide -> hideNow(command)
|
||||
}
|
||||
}
|
||||
|
||||
private fun publishNow(command: PlayerNowPlayingServiceCommand.Publish) {
|
||||
if (PlayerNowPlayingServiceState.notification !== command.notification) return
|
||||
if (startRequested.compareAndSet(false, true)) {
|
||||
val intent = Intent(command.context, PlayerNowPlayingService::class.java)
|
||||
.setAction(ACTION_START_FOREGROUND)
|
||||
val started = runCatching {
|
||||
if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.O) {
|
||||
command.context.startForegroundService(intent)
|
||||
} else {
|
||||
command.context.startService(intent)
|
||||
}
|
||||
}.onFailure { error ->
|
||||
Log.w(NOW_PLAYING_TAG, "Unable to start playback service", error)
|
||||
}.isSuccess
|
||||
if (!started) {
|
||||
startRequested.set(false)
|
||||
return
|
||||
}
|
||||
}
|
||||
if (PlayerNowPlayingServiceState.notification !== command.notification) return
|
||||
runCatching {
|
||||
command.context.getSystemService(NotificationManager::class.java)
|
||||
?.notify(NOW_PLAYING_NOTIFICATION_ID, command.notification)
|
||||
}.onFailure { error ->
|
||||
Log.w(NOW_PLAYING_TAG, "Unable to publish playback notification", error)
|
||||
}
|
||||
}
|
||||
|
||||
private fun hideNow(command: PlayerNowPlayingServiceCommand.Hide) {
|
||||
if (PlayerNowPlayingServiceState.notification != null) return
|
||||
startRequested.set(false)
|
||||
runCatching {
|
||||
command.context.stopService(Intent(command.context, PlayerNowPlayingService::class.java))
|
||||
}.onFailure { error ->
|
||||
Log.w(NOW_PLAYING_TAG, "Unable to stop playback service", error)
|
||||
}
|
||||
runCatching {
|
||||
command.context.getSystemService(NotificationManager::class.java)
|
||||
?.cancel(NOW_PLAYING_NOTIFICATION_ID)
|
||||
}.onFailure { error ->
|
||||
Log.w(NOW_PLAYING_TAG, "Unable to hide playback notification", error)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private object PlayerNowPlayingServiceState {
|
||||
@Volatile
|
||||
var notification: Notification? = null
|
||||
}
|
||||
|
|
@ -0,0 +1,60 @@
|
|||
package com.nuvio.app.core.concurrent
|
||||
|
||||
import kotlinx.atomicfu.locks.SynchronizedObject
|
||||
import kotlinx.atomicfu.locks.synchronized
|
||||
|
||||
internal class ConflatedTaskDispatcher<T>(
|
||||
private val schedule: (() -> Unit) -> Unit,
|
||||
private val consume: (T) -> Unit,
|
||||
) {
|
||||
private data class Pending<T>(val value: T)
|
||||
|
||||
private val lock = SynchronizedObject()
|
||||
private var pending: Pending<T>? = null
|
||||
private var drainScheduled = false
|
||||
|
||||
fun dispatch(value: T) {
|
||||
synchronized(lock) {
|
||||
pending = Pending(value)
|
||||
}
|
||||
scheduleDrain()
|
||||
}
|
||||
|
||||
private fun scheduleDrain() {
|
||||
val shouldSchedule = synchronized(lock) {
|
||||
if (drainScheduled) {
|
||||
false
|
||||
} else {
|
||||
drainScheduled = true
|
||||
true
|
||||
}
|
||||
}
|
||||
if (!shouldSchedule) return
|
||||
|
||||
try {
|
||||
schedule(::drain)
|
||||
} catch (error: Throwable) {
|
||||
synchronized(lock) {
|
||||
drainScheduled = false
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
|
||||
private fun drain() {
|
||||
try {
|
||||
while (true) {
|
||||
val next = synchronized(lock) {
|
||||
pending.also { pending = null }
|
||||
} ?: return
|
||||
consume(next.value)
|
||||
}
|
||||
} finally {
|
||||
val hasPending = synchronized(lock) {
|
||||
drainScheduled = false
|
||||
pending != null
|
||||
}
|
||||
if (hasPending) scheduleDrain()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -134,9 +134,75 @@ internal data class StoredPluginScraper(
|
|||
val logo: String? = null,
|
||||
val contentLanguage: List<String> = emptyList(),
|
||||
val formats: List<String>? = null,
|
||||
val code: String,
|
||||
val code: String? = null,
|
||||
)
|
||||
|
||||
internal data class RestoredPluginScraper(
|
||||
val scraper: PluginScraper,
|
||||
val requiresMigration: Boolean,
|
||||
)
|
||||
|
||||
internal fun PluginsUiState.toStoredPluginsState(): StoredPluginsState =
|
||||
StoredPluginsState(
|
||||
pluginsEnabled = pluginsEnabled,
|
||||
groupStreamsByRepository = groupStreamsByRepository,
|
||||
repositories = repositories.map { repository ->
|
||||
StoredPluginRepository(
|
||||
manifestUrl = repository.manifestUrl,
|
||||
name = repository.name,
|
||||
description = repository.description,
|
||||
version = repository.version,
|
||||
scraperCount = repository.scraperCount,
|
||||
lastUpdated = repository.lastUpdated,
|
||||
)
|
||||
},
|
||||
scrapers = scrapers.map(PluginScraper::toStoredPluginScraper),
|
||||
)
|
||||
|
||||
internal fun PluginScraper.toStoredPluginScraper(): StoredPluginScraper =
|
||||
StoredPluginScraper(
|
||||
id = id,
|
||||
repositoryUrl = repositoryUrl,
|
||||
name = name,
|
||||
description = description,
|
||||
version = version,
|
||||
filename = filename,
|
||||
supportedTypes = supportedTypes,
|
||||
enabled = enabled,
|
||||
manifestEnabled = manifestEnabled,
|
||||
hasSettings = hasSettings,
|
||||
logo = logo,
|
||||
contentLanguage = contentLanguage,
|
||||
formats = formats,
|
||||
code = null,
|
||||
)
|
||||
|
||||
internal fun StoredPluginScraper.restorePluginScraper(
|
||||
loadCachedCode: (String) -> String?,
|
||||
): RestoredPluginScraper? {
|
||||
val cachedCode = loadCachedCode(id)
|
||||
val resolvedCode = cachedCode ?: code ?: return null
|
||||
return RestoredPluginScraper(
|
||||
scraper = PluginScraper(
|
||||
id = id,
|
||||
repositoryUrl = repositoryUrl,
|
||||
name = name,
|
||||
description = description,
|
||||
version = version,
|
||||
filename = filename,
|
||||
supportedTypes = supportedTypes,
|
||||
enabled = enabled,
|
||||
manifestEnabled = manifestEnabled,
|
||||
hasSettings = hasSettings,
|
||||
logo = logo,
|
||||
contentLanguage = contentLanguage,
|
||||
formats = formats,
|
||||
code = resolvedCode,
|
||||
),
|
||||
requiresMigration = code != null,
|
||||
)
|
||||
}
|
||||
|
||||
internal fun normalizePluginType(value: String): String =
|
||||
when (value.lowercase()) {
|
||||
"series", "show", "other" -> "tv"
|
||||
|
|
|
|||
|
|
@ -0,0 +1,32 @@
|
|||
package com.nuvio.app.features.watched
|
||||
|
||||
import com.nuvio.app.features.tracking.TrackingProviderId
|
||||
import kotlinx.atomicfu.locks.SynchronizedObject
|
||||
import kotlinx.atomicfu.locks.synchronized
|
||||
|
||||
internal class WatchedItemsStore {
|
||||
private val lock = SynchronizedObject()
|
||||
private val nuvioItems = mutableMapOf<String, WatchedItem>()
|
||||
private val providerItems = mutableMapOf<TrackingProviderId, MutableMap<String, WatchedItem>>()
|
||||
private val dirtyNuvioKeys = mutableSetOf<String>()
|
||||
|
||||
fun <T> read(
|
||||
block: (
|
||||
nuvioItems: Map<String, WatchedItem>,
|
||||
providerItems: Map<TrackingProviderId, Map<String, WatchedItem>>,
|
||||
dirtyNuvioKeys: Set<String>,
|
||||
) -> T,
|
||||
): T = synchronized(lock) {
|
||||
block(nuvioItems, providerItems, dirtyNuvioKeys)
|
||||
}
|
||||
|
||||
fun <T> update(
|
||||
block: (
|
||||
nuvioItems: MutableMap<String, WatchedItem>,
|
||||
providerItems: MutableMap<TrackingProviderId, MutableMap<String, WatchedItem>>,
|
||||
dirtyNuvioKeys: MutableSet<String>,
|
||||
) -> T,
|
||||
): T = synchronized(lock) {
|
||||
block(nuvioItems, providerItems, dirtyNuvioKeys)
|
||||
}
|
||||
}
|
||||
|
|
@ -101,6 +101,23 @@ internal fun replaceWatchedItemsForSource(
|
|||
target.putAll(replacement)
|
||||
}
|
||||
|
||||
internal suspend fun <T> watchedProviderRefreshOrNull(
|
||||
refresh: suspend () -> T,
|
||||
onFailure: (Throwable) -> Unit,
|
||||
): T? = try {
|
||||
refresh()
|
||||
} catch (error: CancellationException) {
|
||||
throw error
|
||||
} catch (error: Throwable) {
|
||||
onFailure(error)
|
||||
null
|
||||
}
|
||||
|
||||
internal fun extraWatchedKeysChanged(
|
||||
previous: Set<String>?,
|
||||
current: Set<String>,
|
||||
): Boolean = previous.orEmpty() != current
|
||||
|
||||
object WatchedRepository {
|
||||
private data class WatchedRefreshOperation(
|
||||
val profileId: Int,
|
||||
|
|
@ -133,8 +150,7 @@ object WatchedRepository {
|
|||
private var profileGeneration: Long = 0L
|
||||
private var activeSource: WatchProgressSource = WatchProgressSource.NUVIO_SYNC
|
||||
private var sourceGeneration: Long = 0L
|
||||
private var nuvioItemsByKey: MutableMap<String, WatchedItem> = mutableMapOf()
|
||||
private var providerItemsByKey: MutableMap<TrackingProviderId, MutableMap<String, WatchedItem>> = mutableMapOf()
|
||||
private val itemsStore = WatchedItemsStore()
|
||||
private var nuvioFullyWatchedSeriesKeys: Set<String> = emptySet()
|
||||
private var providerFullyWatchedSeriesKeys: MutableMap<TrackingProviderId, Set<String>> = mutableMapOf()
|
||||
private var expandedSiblingKeys: Set<String> = emptySet()
|
||||
|
|
@ -143,7 +159,6 @@ object WatchedRepository {
|
|||
private var loadedProviders: MutableSet<TrackingProviderId> = mutableSetOf()
|
||||
private var nuvioHasLoadedRemote: Boolean = false
|
||||
private var providersLoadedFromRemote: MutableSet<TrackingProviderId> = mutableSetOf()
|
||||
private var nuvioDirtyWatchedKeys: MutableSet<String> = mutableSetOf()
|
||||
private var lastSuccessfulPushEpochMs: Long = 0L
|
||||
private var deltaCursorEventId: Long = 0L
|
||||
private var deltaInitialized: Boolean = false
|
||||
|
|
@ -185,8 +200,11 @@ object WatchedRepository {
|
|||
profileGeneration += 1L
|
||||
activeSource = WatchProgressSource.NUVIO_SYNC
|
||||
sourceGeneration += 1L
|
||||
nuvioItemsByKey.clear()
|
||||
providerItemsByKey.clear()
|
||||
itemsStore.update { nuvioItems, providerItems, dirtyNuvioKeys ->
|
||||
nuvioItems.clear()
|
||||
providerItems.clear()
|
||||
dirtyNuvioKeys.clear()
|
||||
}
|
||||
nuvioFullyWatchedSeriesKeys = emptySet()
|
||||
providerFullyWatchedSeriesKeys.clear()
|
||||
expandedSiblingKeys = emptySet()
|
||||
|
|
@ -195,7 +213,6 @@ object WatchedRepository {
|
|||
loadedProviders.clear()
|
||||
nuvioHasLoadedRemote = false
|
||||
providersLoadedFromRemote.clear()
|
||||
nuvioDirtyWatchedKeys.clear()
|
||||
lastSuccessfulPushEpochMs = 0L
|
||||
deltaCursorEventId = 0L
|
||||
deltaInitialized = false
|
||||
|
|
@ -209,8 +226,11 @@ object WatchedRepository {
|
|||
activeSource = WatchProgressSource.NUVIO_SYNC
|
||||
sourceGeneration += 1L
|
||||
hasLoaded = true
|
||||
nuvioItemsByKey.clear()
|
||||
providerItemsByKey.clear()
|
||||
itemsStore.update { nuvioItems, providerItems, dirtyNuvioKeys ->
|
||||
nuvioItems.clear()
|
||||
providerItems.clear()
|
||||
dirtyNuvioKeys.clear()
|
||||
}
|
||||
nuvioFullyWatchedSeriesKeys = emptySet()
|
||||
providerFullyWatchedSeriesKeys.clear()
|
||||
expandedSiblingKeys = emptySet()
|
||||
|
|
@ -219,7 +239,6 @@ object WatchedRepository {
|
|||
loadedProviders.clear()
|
||||
nuvioHasLoadedRemote = false
|
||||
providersLoadedFromRemote.clear()
|
||||
nuvioDirtyWatchedKeys.clear()
|
||||
|
||||
val payload = WatchedStorage.loadPayload(profileId).orEmpty().trim()
|
||||
if (payload.isNotEmpty()) {
|
||||
|
|
@ -229,19 +248,19 @@ object WatchedRepository {
|
|||
lastSuccessfulPushEpochMs = storedPayload.lastSuccessfulPushEpochMs
|
||||
deltaCursorEventId = storedPayload.deltaCursorEventId
|
||||
deltaInitialized = storedPayload.deltaInitialized
|
||||
nuvioItemsByKey = storedPayload.items
|
||||
val restoredItems = storedPayload.items
|
||||
.map(WatchedItem::normalizedMarkedAt)
|
||||
.associateBy { watchedItemKey(it.type, it.id, it.season, it.episode) }
|
||||
.toMutableMap()
|
||||
nuvioDirtyWatchedKeys = storedPayload.dirtyWatchedKeys
|
||||
.filterTo(mutableSetOf()) { key -> key in nuvioItemsByKey }
|
||||
itemsStore.update { nuvioItems, _, dirtyNuvioKeys ->
|
||||
nuvioItems.putAll(restoredItems)
|
||||
dirtyNuvioKeys += storedPayload.dirtyWatchedKeys.filter { key -> key in restoredItems }
|
||||
}
|
||||
nuvioFullyWatchedSeriesKeys = storedPayload.fullyWatchedSeriesKeys
|
||||
expandedSiblingKeys = storedPayload.expandedSiblingKeys
|
||||
} else {
|
||||
lastSuccessfulPushEpochMs = 0L
|
||||
deltaCursorEventId = 0L
|
||||
deltaInitialized = false
|
||||
nuvioDirtyWatchedKeys.clear()
|
||||
nuvioFullyWatchedSeriesKeys = emptySet()
|
||||
}
|
||||
|
||||
|
|
@ -265,7 +284,9 @@ object WatchedRepository {
|
|||
}
|
||||
val previousSource = activeSource
|
||||
source.providerId?.let { providerId ->
|
||||
providerItemsByKey.getOrPut(providerId, ::mutableMapOf).clear()
|
||||
itemsStore.update { _, providerItems, _ ->
|
||||
providerItems.getOrPut(providerId, ::mutableMapOf).clear()
|
||||
}
|
||||
providerFullyWatchedSeriesKeys[providerId] = emptySet()
|
||||
providerExtraWatchedKeys.remove(providerId)
|
||||
loadedProviders -= providerId
|
||||
|
|
@ -437,6 +458,7 @@ object WatchedRepository {
|
|||
pageSize = watchedItemsPageSize,
|
||||
)
|
||||
val fullyWatchedSeriesKeys = adapter.pullFullyWatchedSeriesKeys(profileId)
|
||||
val extraWatchedKeys = adapter.pullExtraWatchedKeys(profileId)
|
||||
val source = operation.sourceOperation.source
|
||||
log.i {
|
||||
"Watched adapter result source=$source provider=${source.providerId?.storageId ?: "nuvio"} " +
|
||||
|
|
@ -457,35 +479,36 @@ object WatchedRepository {
|
|||
}
|
||||
return false
|
||||
}
|
||||
val localAtApply = itemsForSource(operation.sourceOperation.source).values.toList()
|
||||
|
||||
val mergedSnapshot = mergeWatchedSnapshot(
|
||||
serverItems = serverItems,
|
||||
localItems = localAtApply,
|
||||
dirtyKeys = if (operation.sourceOperation.source == WatchProgressSource.NUVIO_SYNC) {
|
||||
nuvioDirtyWatchedKeys
|
||||
} else {
|
||||
emptySet()
|
||||
},
|
||||
)
|
||||
replaceWatchedItemsForSource(
|
||||
source = operation.sourceOperation.source,
|
||||
nuvioItems = nuvioItemsByKey,
|
||||
providerItems = providerItemsByKey,
|
||||
replacement = mergedSnapshot.items,
|
||||
)
|
||||
fullyWatchedSeriesKeys?.let { keys ->
|
||||
setFullyWatchedSeriesKeysForSource(operation.sourceOperation.source, keys)
|
||||
itemsStore.update { nuvioItems, providerItems, dirtyNuvioKeys ->
|
||||
val items = source.providerId
|
||||
?.let { providerId -> providerItems[providerId]?.values.orEmpty() }
|
||||
?: nuvioItems.values
|
||||
val merged = mergeWatchedSnapshot(
|
||||
serverItems = serverItems,
|
||||
localItems = items.toList(),
|
||||
dirtyKeys = if (source.providerId == null) dirtyNuvioKeys else emptySet(),
|
||||
)
|
||||
replaceWatchedItemsForSource(
|
||||
source = source,
|
||||
nuvioItems = nuvioItems,
|
||||
providerItems = providerItems,
|
||||
replacement = merged.items,
|
||||
)
|
||||
if (source.providerId == null) {
|
||||
dirtyNuvioKeys.clear()
|
||||
dirtyNuvioKeys += merged.dirtyKeys
|
||||
}
|
||||
}
|
||||
val extraWatchedKeys = adapter.pullExtraWatchedKeys(profileId)
|
||||
operation.sourceOperation.source.providerId?.let { providerId ->
|
||||
fullyWatchedSeriesKeys?.let { keys ->
|
||||
setFullyWatchedSeriesKeysForSource(source, keys)
|
||||
}
|
||||
source.providerId?.let { providerId ->
|
||||
if (extraWatchedKeys.isNotEmpty()) {
|
||||
providerExtraWatchedKeys[providerId] = extraWatchedKeys
|
||||
}
|
||||
loadedProviders += providerId
|
||||
providersLoadedFromRemote += providerId
|
||||
} ?: run {
|
||||
nuvioDirtyWatchedKeys = mergedSnapshot.dirtyKeys.toMutableSet()
|
||||
nuvioHasLoaded = true
|
||||
nuvioHasLoadedRemote = true
|
||||
if (resetDeltaState) {
|
||||
|
|
@ -547,11 +570,13 @@ object WatchedRepository {
|
|||
if (!isActiveOperation(operation)) return false
|
||||
if (events.isEmpty()) break
|
||||
|
||||
applyWatchedDeltaEvents(
|
||||
targetItems = nuvioItemsByKey,
|
||||
dirtyKeys = nuvioDirtyWatchedKeys,
|
||||
events = events,
|
||||
)
|
||||
itemsStore.update { nuvioItems, _, dirtyNuvioKeys ->
|
||||
applyWatchedDeltaEvents(
|
||||
targetItems = nuvioItems,
|
||||
dirtyKeys = dirtyNuvioKeys,
|
||||
events = events,
|
||||
)
|
||||
}
|
||||
cursor = maxOf(cursor, events.maxOf { it.eventId })
|
||||
deltaCursorEventId = cursor
|
||||
deltaInitialized = true
|
||||
|
|
@ -673,10 +698,13 @@ object WatchedRepository {
|
|||
return remoteType.isSeriesLikeWatchedType() && localType.isSeriesLikeWatchedType()
|
||||
}
|
||||
|
||||
private fun itemsForSource(source: WatchProgressSource): MutableMap<String, WatchedItem> =
|
||||
source.providerId
|
||||
?.let { providerId -> providerItemsByKey.getOrPut(providerId, ::mutableMapOf) }
|
||||
?: nuvioItemsByKey
|
||||
private fun itemsForSourceSnapshot(source: WatchProgressSource): List<WatchedItem> =
|
||||
itemsStore.read { nuvioItems, providerItems, _ ->
|
||||
val items = source.providerId
|
||||
?.let { providerId -> providerItems[providerId]?.values.orEmpty() }
|
||||
?: nuvioItems.values
|
||||
items.toList()
|
||||
}
|
||||
|
||||
private fun fullyWatchedSeriesKeysForSource(source: WatchProgressSource): Set<String> =
|
||||
source.providerId
|
||||
|
|
@ -697,16 +725,23 @@ object WatchedRepository {
|
|||
private fun hasLoadedSource(source: WatchProgressSource): Boolean =
|
||||
source.providerId?.let(loadedProviders::contains) ?: nuvioHasLoaded
|
||||
|
||||
private fun itemCountForSource(source: WatchProgressSource): Int = source.providerId
|
||||
?.let { providerId -> providerItemsByKey[providerId]?.size ?: 0 }
|
||||
?: nuvioItemsByKey.size
|
||||
private fun itemCountForSource(source: WatchProgressSource): Int =
|
||||
itemsStore.read { nuvioItems, providerItems, _ ->
|
||||
source.providerId
|
||||
?.let { providerId -> providerItems[providerId]?.size ?: 0 }
|
||||
?: nuvioItems.size
|
||||
}
|
||||
|
||||
fun toggleWatched(item: WatchedItem) {
|
||||
ensureLoaded()
|
||||
val source = activeSource
|
||||
val targetItems = itemsForSource(source)
|
||||
val key = watchedItemKey(item.type, item.id, item.season, item.episode)
|
||||
if (targetItems.containsKey(key)) {
|
||||
val isMarked = itemsStore.read { nuvioItems, providerItems, _ ->
|
||||
source.providerId
|
||||
?.let { providerId -> providerItems[providerId]?.containsKey(key) == true }
|
||||
?: nuvioItems.containsKey(key)
|
||||
}
|
||||
if (isMarked) {
|
||||
unmarkWatched(item)
|
||||
} else {
|
||||
markWatched(item)
|
||||
|
|
@ -737,16 +772,20 @@ object WatchedRepository {
|
|||
ensureLoaded()
|
||||
if (items.isEmpty()) return
|
||||
val source = activeSource
|
||||
val targetItems = itemsForSource(source)
|
||||
val markedAt = WatchedClock.nowEpochMs()
|
||||
val timestampedItems = items.map { watchedItem ->
|
||||
watchedItem.copy(markedAtEpochMs = markedAt)
|
||||
}
|
||||
timestampedItems.forEach { watchedItem ->
|
||||
val key = watchedItemKey(watchedItem.type, watchedItem.id, watchedItem.season, watchedItem.episode)
|
||||
targetItems[key] = watchedItem
|
||||
if (source.providerId == null) {
|
||||
nuvioDirtyWatchedKeys += key
|
||||
itemsStore.update { nuvioItems, providerItems, dirtyNuvioKeys ->
|
||||
val targetItems = source.providerId
|
||||
?.let { providerId -> providerItems.getOrPut(providerId, ::mutableMapOf) }
|
||||
?: nuvioItems
|
||||
timestampedItems.forEach { watchedItem ->
|
||||
val key = watchedItemKey(watchedItem.type, watchedItem.id, watchedItem.season, watchedItem.episode)
|
||||
targetItems[key] = watchedItem
|
||||
if (source.providerId == null) {
|
||||
dirtyNuvioKeys += key
|
||||
}
|
||||
}
|
||||
}
|
||||
publish()
|
||||
|
|
@ -790,25 +829,27 @@ object WatchedRepository {
|
|||
ensureLoaded()
|
||||
if (items.isEmpty()) return
|
||||
val source = activeSource
|
||||
val targetItems = itemsForSource(source)
|
||||
val removedItems = items.mapNotNull { watchedItem ->
|
||||
val key = watchedItemKey(watchedItem.type, watchedItem.id, watchedItem.season, watchedItem.episode)
|
||||
targetItems.remove(key)?.let { storeItem ->
|
||||
// Preserve videoId from the original request (store items don't have it)
|
||||
if (watchedItem.videoId != null && storeItem.videoId == null) {
|
||||
storeItem.copy(videoId = watchedItem.videoId)
|
||||
} else {
|
||||
storeItem
|
||||
}
|
||||
}?.also {
|
||||
if (source.providerId == null) {
|
||||
nuvioDirtyWatchedKeys -= key
|
||||
}
|
||||
// Optimistically remove from extra keys so publish() doesn't re-add it
|
||||
source.providerId?.let { providerId ->
|
||||
providerExtraWatchedKeys[providerId]?.let { extraKeys ->
|
||||
if (key in extraKeys) {
|
||||
providerExtraWatchedKeys[providerId] = extraKeys - key
|
||||
val removedItems = itemsStore.update { nuvioItems, providerItems, dirtyNuvioKeys ->
|
||||
val targetItems = source.providerId
|
||||
?.let { providerId -> providerItems.getOrPut(providerId, ::mutableMapOf) }
|
||||
?: nuvioItems
|
||||
items.mapNotNull { watchedItem ->
|
||||
val key = watchedItemKey(watchedItem.type, watchedItem.id, watchedItem.season, watchedItem.episode)
|
||||
targetItems.remove(key)?.let { storeItem ->
|
||||
if (watchedItem.videoId != null && storeItem.videoId == null) {
|
||||
storeItem.copy(videoId = watchedItem.videoId)
|
||||
} else {
|
||||
storeItem
|
||||
}
|
||||
}?.also {
|
||||
if (source.providerId == null) {
|
||||
dirtyNuvioKeys -= key
|
||||
}
|
||||
source.providerId?.let { providerId ->
|
||||
providerExtraWatchedKeys[providerId]?.let { extraKeys ->
|
||||
if (key in extraKeys) {
|
||||
providerExtraWatchedKeys[providerId] = extraKeys - key
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -835,7 +876,13 @@ object WatchedRepository {
|
|||
episode: Int? = null,
|
||||
): Boolean {
|
||||
ensureLoaded()
|
||||
return itemsForSource(activeSource).containsKey(watchedItemKey(type, id, season, episode))
|
||||
val source = activeSource
|
||||
val key = watchedItemKey(type, id, season, episode)
|
||||
return itemsStore.read { nuvioItems, providerItems, _ ->
|
||||
source.providerId
|
||||
?.let { providerId -> providerItems[providerId]?.containsKey(key) == true }
|
||||
?: nuvioItems.containsKey(key)
|
||||
}
|
||||
}
|
||||
|
||||
fun isFullyWatchedSeries(id: String, type: String): Boolean {
|
||||
|
|
@ -862,7 +909,7 @@ object WatchedRepository {
|
|||
"Watched series reconciliation source=$activeSource content=${meta.type}:${meta.id} " +
|
||||
"episodes=${meta.videos.size} shouldMarkSeries=$shouldMarkSeriesWatched " +
|
||||
"hasSeriesMarker=$hasSeriesWatchedMarker " +
|
||||
"matchingItems=${itemsForSource(activeSource).values.count { it.id == meta.id }}"
|
||||
"matchingItems=${itemsForSourceSnapshot(activeSource).count { it.id == meta.id }}"
|
||||
}
|
||||
if (shouldMarkSeriesWatched) {
|
||||
if (!hasSeriesWatchedMarker) {
|
||||
|
|
@ -995,10 +1042,15 @@ object WatchedRepository {
|
|||
}
|
||||
|
||||
private fun publish() {
|
||||
val (nuvioItems, providerItems) = itemsStore.read { storedNuvioItems, storedProviderItems, _ ->
|
||||
storedNuvioItems.values.toList() to storedProviderItems.mapValues { (_, itemsByKey) ->
|
||||
itemsByKey.values.toList()
|
||||
}
|
||||
}
|
||||
val items = watchedItemsForSource(
|
||||
source = activeSource,
|
||||
nuvioItems = nuvioItemsByKey.values,
|
||||
providerItems = providerItemsByKey.mapValues { (_, itemsByKey) -> itemsByKey.values },
|
||||
nuvioItems = nuvioItems,
|
||||
providerItems = providerItems,
|
||||
)
|
||||
.map(WatchedItem::normalizedMarkedAt)
|
||||
.sortedByDescending { it.markedAtEpochMs }
|
||||
|
|
@ -1052,19 +1104,29 @@ object WatchedRepository {
|
|||
adapter.observeExtraWatchedKeys(currentProfileId)
|
||||
.distinctUntilChanged()
|
||||
.collectLatest { extraKeys ->
|
||||
val keysChanged = providerExtraWatchedKeys[providerId] != extraKeys
|
||||
val keysChanged = extraWatchedKeysChanged(
|
||||
previous = providerExtraWatchedKeys[providerId],
|
||||
current = extraKeys,
|
||||
)
|
||||
if (keysChanged) {
|
||||
providerExtraWatchedKeys[providerId] = extraKeys
|
||||
// Re-pull items from provider to reflect snapshot changes
|
||||
// (e.g. after remote episode removal)
|
||||
val freshItems = adapter.pull(
|
||||
profileId = currentProfileId,
|
||||
pageSize = watchedItemsPageSize,
|
||||
)
|
||||
val freshItems = watchedProviderRefreshOrNull(
|
||||
refresh = {
|
||||
adapter.pull(
|
||||
profileId = currentProfileId,
|
||||
pageSize = watchedItemsPageSize,
|
||||
)
|
||||
},
|
||||
onFailure = { error ->
|
||||
log.w(error) { "Failed to refresh watched items from ${providerId.storageId}" }
|
||||
},
|
||||
) ?: return@collectLatest
|
||||
val itemsByKey = freshItems.associateBy { item ->
|
||||
watchedItemKey(item.type, item.id, item.season, item.episode)
|
||||
}.toMutableMap()
|
||||
providerItemsByKey[providerId] = itemsByKey
|
||||
providerExtraWatchedKeys[providerId] = extraKeys
|
||||
itemsStore.update { _, providerItems, _ ->
|
||||
providerItems[providerId] = itemsByKey
|
||||
}
|
||||
publish()
|
||||
}
|
||||
}
|
||||
|
|
@ -1077,19 +1139,22 @@ object WatchedRepository {
|
|||
}
|
||||
|
||||
private fun persistNuvio() {
|
||||
val (items, dirtyKeys) = itemsStore.read { nuvioItems, _, dirtyNuvioKeys ->
|
||||
nuvioItems.values
|
||||
.map(WatchedItem::normalizedMarkedAt)
|
||||
.sortedByDescending { it.markedAtEpochMs } to dirtyNuvioKeys.toSet()
|
||||
}
|
||||
WatchedStorage.savePayload(
|
||||
currentProfileId,
|
||||
json.encodeToString(
|
||||
StoredWatchedPayload(
|
||||
items = nuvioItemsByKey.values
|
||||
.map(WatchedItem::normalizedMarkedAt)
|
||||
.sortedByDescending { it.markedAtEpochMs },
|
||||
items = items,
|
||||
fullyWatchedSeriesKeys = nuvioFullyWatchedSeriesKeys,
|
||||
expandedSiblingKeys = expandedSiblingKeys,
|
||||
lastSuccessfulPushEpochMs = lastSuccessfulPushEpochMs,
|
||||
deltaCursorEventId = deltaCursorEventId,
|
||||
deltaInitialized = deltaInitialized,
|
||||
dirtyWatchedKeys = nuvioDirtyWatchedKeys.toSet(),
|
||||
dirtyWatchedKeys = dirtyKeys,
|
||||
),
|
||||
),
|
||||
)
|
||||
|
|
@ -1101,26 +1166,31 @@ object WatchedRepository {
|
|||
items: Collection<WatchedItem>,
|
||||
) {
|
||||
if (profileId != currentProfileId || operationGeneration != profileGeneration) return
|
||||
val acknowledgedDirtyKeys = acknowledgeSuccessfulWatchedPush(
|
||||
currentItems = nuvioItemsByKey,
|
||||
dirtyKeys = nuvioDirtyWatchedKeys,
|
||||
pushedItems = items,
|
||||
)
|
||||
val latestPushed = items
|
||||
.asSequence()
|
||||
.map { item -> normalizeWatchedMarkedAtEpochMs(item.markedAtEpochMs) }
|
||||
.maxOrNull()
|
||||
?: return
|
||||
val updatedLastSuccessfulPushEpochMs = maxOf(lastSuccessfulPushEpochMs, latestPushed)
|
||||
if (
|
||||
acknowledgedDirtyKeys == nuvioDirtyWatchedKeys &&
|
||||
updatedLastSuccessfulPushEpochMs == lastSuccessfulPushEpochMs
|
||||
) {
|
||||
return
|
||||
val changed = itemsStore.update { nuvioItems, _, dirtyNuvioKeys ->
|
||||
val acknowledgedDirtyKeys = acknowledgeSuccessfulWatchedPush(
|
||||
currentItems = nuvioItems,
|
||||
dirtyKeys = dirtyNuvioKeys,
|
||||
pushedItems = items,
|
||||
)
|
||||
val updatedLastSuccessfulPushEpochMs = maxOf(lastSuccessfulPushEpochMs, latestPushed)
|
||||
if (
|
||||
acknowledgedDirtyKeys == dirtyNuvioKeys &&
|
||||
updatedLastSuccessfulPushEpochMs == lastSuccessfulPushEpochMs
|
||||
) {
|
||||
false
|
||||
} else {
|
||||
dirtyNuvioKeys.clear()
|
||||
dirtyNuvioKeys += acknowledgedDirtyKeys
|
||||
lastSuccessfulPushEpochMs = updatedLastSuccessfulPushEpochMs
|
||||
true
|
||||
}
|
||||
}
|
||||
nuvioDirtyWatchedKeys = acknowledgedDirtyKeys.toMutableSet()
|
||||
lastSuccessfulPushEpochMs = updatedLastSuccessfulPushEpochMs
|
||||
persistNuvio()
|
||||
if (changed) persistNuvio()
|
||||
}
|
||||
|
||||
private suspend fun pushToTargetsForSource(
|
||||
|
|
|
|||
|
|
@ -14,6 +14,7 @@ import com.nuvio.app.features.watched.normalizeWatchedMarkedAtEpochMs
|
|||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.async
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.delay
|
||||
import kotlinx.serialization.SerialName
|
||||
import kotlinx.serialization.Serializable
|
||||
import kotlinx.serialization.encodeToString
|
||||
|
|
@ -23,6 +24,71 @@ private const val BASE_URL = "https://api.trakt.tv"
|
|||
private const val WATCHED_PAGE_LIMIT = 250
|
||||
private const val WATCHED_MAX_PAGES = 1_000
|
||||
private const val WATCHED_SHOWS_EXTENDED = "progress"
|
||||
internal const val TRAKT_WATCHED_MAX_RESPONSE_BODY_BYTES = 8 * 1024 * 1024
|
||||
private const val TRAKT_WATCHED_MAX_ATTEMPTS = 2
|
||||
private const val TRAKT_WATCHED_MAX_RETRY_DELAY_MS = 60_000L
|
||||
|
||||
internal fun interface TraktWatchedHttpEngine {
|
||||
suspend fun get(
|
||||
url: String,
|
||||
headers: Map<String, String>,
|
||||
maxResponseBodyBytes: Int,
|
||||
): RawHttpResponse
|
||||
}
|
||||
|
||||
internal class TraktWatchedPageClient(
|
||||
private val engine: TraktWatchedHttpEngine,
|
||||
private val sleep: suspend (Long) -> Unit = { delayMs -> delay(delayMs) },
|
||||
) {
|
||||
suspend fun get(url: String, headers: Map<String, String>): RawHttpResponse {
|
||||
repeat(TRAKT_WATCHED_MAX_ATTEMPTS) { attempt ->
|
||||
val response = engine.get(
|
||||
url = url,
|
||||
headers = headers,
|
||||
maxResponseBodyBytes = TRAKT_WATCHED_MAX_RESPONSE_BODY_BYTES,
|
||||
)
|
||||
if (response.status in 200..299) return response
|
||||
if (!isTransientTraktWatchedStatus(response.status) || attempt == TRAKT_WATCHED_MAX_ATTEMPTS - 1) {
|
||||
throw TraktWatchedHttpException(response.status)
|
||||
}
|
||||
sleep(
|
||||
traktWatchedRetryDelayMs(
|
||||
attempt = attempt,
|
||||
retryAfterSeconds = response.headers.entries
|
||||
.firstOrNull { (name, _) -> name.equals("retry-after", ignoreCase = true) }
|
||||
?.value,
|
||||
),
|
||||
)
|
||||
}
|
||||
error("Trakt watched request exhausted without a response")
|
||||
}
|
||||
}
|
||||
|
||||
internal class TraktWatchedHttpException(
|
||||
val status: Int,
|
||||
) : Exception("Trakt watched request failed: $status")
|
||||
|
||||
internal fun isTransientTraktWatchedStatus(status: Int): Boolean =
|
||||
status == 429 || status in 500..599
|
||||
|
||||
internal fun traktWatchedRetryDelayMs(attempt: Int, retryAfterSeconds: String?): Long =
|
||||
retryAfterSeconds
|
||||
?.trim()
|
||||
?.toLongOrNull()
|
||||
?.coerceAtLeast(0L)
|
||||
?.times(1_000L)
|
||||
?.coerceAtMost(TRAKT_WATCHED_MAX_RETRY_DELAY_MS)
|
||||
?: (1_000L shl attempt.coerceIn(0, 5)).coerceAtMost(TRAKT_WATCHED_MAX_RETRY_DELAY_MS)
|
||||
|
||||
private val platformTraktWatchedHttpEngine = TraktWatchedHttpEngine { url, headers, maxResponseBodyBytes ->
|
||||
httpRequestRaw(
|
||||
method = "GET",
|
||||
url = url,
|
||||
headers = headers,
|
||||
body = "",
|
||||
maxResponseBodyBytes = maxResponseBodyBytes,
|
||||
)
|
||||
}
|
||||
|
||||
|
||||
object TraktWatchedSyncAdapter : TrackingWatchedProvider {
|
||||
|
|
@ -33,6 +99,7 @@ object TraktWatchedSyncAdapter : TrackingWatchedProvider {
|
|||
encodeDefaults = false
|
||||
explicitNulls = false
|
||||
}
|
||||
private val pageClient = TraktWatchedPageClient(platformTraktWatchedHttpEngine)
|
||||
|
||||
// ── pull ────────────────────────────────────────────────────────────
|
||||
override suspend fun pull(
|
||||
|
|
@ -118,15 +185,10 @@ object TraktWatchedSyncAdapter : TrackingWatchedProvider {
|
|||
val items = mutableListOf<TraktWatchedMovieDto>()
|
||||
var page = 1
|
||||
while (page <= WATCHED_MAX_PAGES) {
|
||||
val response = httpRequestRaw(
|
||||
method = "GET",
|
||||
val response = pageClient.get(
|
||||
url = "$BASE_URL/sync/watched/movies?page=$page&limit=$WATCHED_PAGE_LIMIT",
|
||||
headers = headers,
|
||||
body = "",
|
||||
)
|
||||
if (response.status !in 200..299) {
|
||||
error("Trakt watched movies request failed: ${response.status}")
|
||||
}
|
||||
val pageItems = json.decodeFromString<List<TraktWatchedMovieDto>>(response.body)
|
||||
if (pageItems.isEmpty()) break
|
||||
items.addAll(pageItems)
|
||||
|
|
@ -144,15 +206,10 @@ object TraktWatchedSyncAdapter : TrackingWatchedProvider {
|
|||
val items = mutableListOf<TraktWatchedShowDto>()
|
||||
var page = 1
|
||||
while (page <= WATCHED_MAX_PAGES) {
|
||||
val response = httpRequestRaw(
|
||||
method = "GET",
|
||||
val response = pageClient.get(
|
||||
url = "$BASE_URL/sync/watched/shows?page=$page&limit=$WATCHED_PAGE_LIMIT&extended=$WATCHED_SHOWS_EXTENDED",
|
||||
headers = headers,
|
||||
body = "",
|
||||
)
|
||||
if (response.status !in 200..299) {
|
||||
error("Trakt watched shows request failed: ${response.status}")
|
||||
}
|
||||
val pageItems = json.decodeFromString<List<TraktWatchedShowDto>>(response.body)
|
||||
if (pageItems.isEmpty()) break
|
||||
items.addAll(pageItems)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,71 @@
|
|||
package com.nuvio.app.core.concurrent
|
||||
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFailsWith
|
||||
|
||||
class ConflatedTaskDispatcherTest {
|
||||
@Test
|
||||
fun dispatchConsumesOnlyLatestPendingValue() {
|
||||
val scheduled = ArrayDeque<() -> Unit>()
|
||||
val consumed = mutableListOf<Int>()
|
||||
val dispatcher = ConflatedTaskDispatcher<Int>(
|
||||
schedule = scheduled::addLast,
|
||||
consume = consumed::add,
|
||||
)
|
||||
|
||||
dispatcher.dispatch(1)
|
||||
dispatcher.dispatch(2)
|
||||
dispatcher.dispatch(3)
|
||||
|
||||
assertEquals(1, scheduled.size)
|
||||
assertEquals(emptyList(), consumed)
|
||||
scheduled.removeFirst().invoke()
|
||||
assertEquals(listOf(3), consumed)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun valuesSubmittedWhileConsumingAreConflated() {
|
||||
val scheduled = ArrayDeque<() -> Unit>()
|
||||
val consumed = mutableListOf<Int>()
|
||||
lateinit var dispatcher: ConflatedTaskDispatcher<Int>
|
||||
dispatcher = ConflatedTaskDispatcher(
|
||||
schedule = scheduled::addLast,
|
||||
consume = { value ->
|
||||
consumed += value
|
||||
if (value == 1) {
|
||||
dispatcher.dispatch(2)
|
||||
dispatcher.dispatch(3)
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
dispatcher.dispatch(1)
|
||||
scheduled.removeFirst().invoke()
|
||||
|
||||
assertEquals(listOf(1, 3), consumed)
|
||||
assertEquals(0, scheduled.size)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun consumerFailureDoesNotStallFutureDispatches() {
|
||||
val scheduled = ArrayDeque<() -> Unit>()
|
||||
val consumed = mutableListOf<Int>()
|
||||
val dispatcher = ConflatedTaskDispatcher<Int>(
|
||||
schedule = scheduled::addLast,
|
||||
consume = { value ->
|
||||
if (value == 1) error("failed")
|
||||
consumed += value
|
||||
},
|
||||
)
|
||||
|
||||
dispatcher.dispatch(1)
|
||||
assertFailsWith<IllegalStateException> {
|
||||
scheduled.removeFirst().invoke()
|
||||
}
|
||||
dispatcher.dispatch(2)
|
||||
scheduled.removeFirst().invoke()
|
||||
|
||||
assertEquals(listOf(2), consumed)
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,66 @@
|
|||
package com.nuvio.app.features.plugins
|
||||
|
||||
import kotlinx.serialization.encodeToString
|
||||
import kotlinx.serialization.json.Json
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFalse
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class PluginPersistenceTest {
|
||||
private val json = Json { encodeDefaults = true }
|
||||
|
||||
@Test
|
||||
fun metadataStateDoesNotSerializeScraperSourceCode() {
|
||||
val sourceCode = "module.exports = " + "x".repeat(4 * 1024 * 1024)
|
||||
val stored = PluginsUiState(
|
||||
scrapers = listOf(pluginScraper(sourceCode)),
|
||||
).toStoredPluginsState()
|
||||
|
||||
val encoded = json.encodeToString(stored)
|
||||
|
||||
assertNull(stored.scrapers.single().code)
|
||||
assertTrue(encoded.length < 2_048)
|
||||
assertFalse(sourceCode in encoded)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun cachedScraperCodeRestoresOfflineWithoutMigration() {
|
||||
val sourceCode = "offline scraper source"
|
||||
val stored = pluginScraper(sourceCode).toStoredPluginScraper()
|
||||
|
||||
val restored = stored.restorePluginScraper { scraperId ->
|
||||
if (scraperId == stored.id) sourceCode else null
|
||||
}
|
||||
|
||||
assertEquals(sourceCode, restored?.scraper?.code)
|
||||
assertFalse(restored?.requiresMigration ?: true)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun legacyEmbeddedScraperCodeIsPreservedForMigration() {
|
||||
val sourceCode = "legacy scraper source"
|
||||
val stored = pluginScraper(sourceCode)
|
||||
.toStoredPluginScraper()
|
||||
.copy(code = sourceCode)
|
||||
|
||||
val restored = stored.restorePluginScraper { null }
|
||||
|
||||
assertEquals(sourceCode, restored?.scraper?.code)
|
||||
assertTrue(restored?.requiresMigration == true)
|
||||
}
|
||||
|
||||
private fun pluginScraper(code: String): PluginScraper = PluginScraper(
|
||||
id = "https://plugins.example/manifest.json:scraper",
|
||||
repositoryUrl = "https://plugins.example/manifest.json",
|
||||
name = "Scraper",
|
||||
description = "",
|
||||
version = "1.0.0",
|
||||
filename = "scraper.js",
|
||||
supportedTypes = listOf("movie", "tv"),
|
||||
enabled = true,
|
||||
manifestEnabled = true,
|
||||
code = code,
|
||||
)
|
||||
}
|
||||
|
|
@ -275,7 +275,7 @@ class SimklAnimeWatchedResolutionTest {
|
|||
}
|
||||
|
||||
@Test
|
||||
fun `resolveAnimeEpisodeForSimkl returns unchanged for IMDB videoId prefix`() {
|
||||
fun `resolveAnimeEpisodeForSimkl strips anime ids for IMDB videoId prefix`() {
|
||||
val reference = TrackingMediaReference(
|
||||
kind = TrackingMediaKind.ANIME,
|
||||
title = "Some Anime",
|
||||
|
|
@ -290,12 +290,14 @@ class SimklAnimeWatchedResolutionTest {
|
|||
|
||||
val resolved = reference.resolveAnimeEpisodeForSimkl()
|
||||
|
||||
// "tt2560140" prefix is not an anime prefix → unchanged
|
||||
assertEquals(reference, resolved)
|
||||
assertEquals("tt2560140", resolved.ids.imdb)
|
||||
assertNull(resolved.ids.mal)
|
||||
assertEquals(reference.episode, resolved.episode)
|
||||
assertEquals(reference.catalog, resolved.catalog)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `resolveAnimeEpisodeForSimkl returns unchanged without catalog videoId`() {
|
||||
fun `resolveAnimeEpisodeForSimkl strips anime ids without catalog videoId`() {
|
||||
val reference = TrackingMediaReference(
|
||||
kind = TrackingMediaKind.ANIME,
|
||||
title = "Some Anime",
|
||||
|
|
@ -306,7 +308,9 @@ class SimklAnimeWatchedResolutionTest {
|
|||
|
||||
val resolved = reference.resolveAnimeEpisodeForSimkl()
|
||||
|
||||
assertEquals(reference, resolved)
|
||||
assertNull(resolved.ids.mal)
|
||||
assertEquals(reference.episode, resolved.episode)
|
||||
assertNull(resolved.catalog)
|
||||
}
|
||||
|
||||
// ──────────────────────────────────────────────────────────────────────────
|
||||
|
|
@ -432,14 +436,14 @@ class SimklAnimeWatchedResolutionTest {
|
|||
// ──────────────────────────────────────────────────────────────────────────
|
||||
|
||||
@Test
|
||||
fun `library projection uses anime type for anime entries`() {
|
||||
fun `library projection uses series type for episodic anime entries`() {
|
||||
val entry = animeEntry(simklId = 39687, imdb = "tt2560140", mal = 16498)
|
||||
val snapshot = SimklSyncSnapshot(entries = listOf(entry))
|
||||
val projection = snapshot.toSimklLibraryProjection()
|
||||
|
||||
val item = projection.items.singleOrNull { it.id == "tt2560140" }
|
||||
assertNotNull(item)
|
||||
assertEquals("anime", item.type)
|
||||
assertEquals("series", item.type)
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
|
|||
|
|
@ -0,0 +1,55 @@
|
|||
package com.nuvio.app.features.watched
|
||||
|
||||
import com.nuvio.app.features.tracking.TrackingProviderId
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.coroutineScope
|
||||
import kotlinx.coroutines.launch
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
|
||||
class WatchedItemsStoreTest {
|
||||
@Test
|
||||
fun `concurrent updates publish coherent item snapshots`() = runBlocking {
|
||||
val store = WatchedItemsStore()
|
||||
|
||||
coroutineScope {
|
||||
repeat(4) { writer ->
|
||||
launch(Dispatchers.Default) {
|
||||
repeat(500) { index ->
|
||||
val key = "$writer:$index"
|
||||
val item = WatchedItem(
|
||||
id = key,
|
||||
type = "movie",
|
||||
name = key,
|
||||
markedAtEpochMs = index.toLong(),
|
||||
)
|
||||
store.update { nuvioItems, providerItems, dirtyNuvioKeys ->
|
||||
nuvioItems[key] = item
|
||||
providerItems
|
||||
.getOrPut(TrackingProviderId.TRAKT, ::mutableMapOf)[key] = item
|
||||
dirtyNuvioKeys += key
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
repeat(4) {
|
||||
launch(Dispatchers.Default) {
|
||||
repeat(500) {
|
||||
store.read { nuvioItems, providerItems, dirtyNuvioKeys ->
|
||||
val nuvioKeys = nuvioItems.keys.toSet()
|
||||
assertEquals(nuvioKeys, providerItems[TrackingProviderId.TRAKT].orEmpty().keys)
|
||||
assertEquals(nuvioKeys, dirtyNuvioKeys)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
store.read { nuvioItems, providerItems, dirtyNuvioKeys ->
|
||||
assertEquals(2_000, nuvioItems.size)
|
||||
assertEquals(nuvioItems.keys, providerItems[TrackingProviderId.TRAKT].orEmpty().keys)
|
||||
assertEquals(nuvioItems.keys, dirtyNuvioKeys)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -4,12 +4,61 @@ import com.nuvio.app.features.details.MetaDetails
|
|||
import com.nuvio.app.features.details.MetaVideo
|
||||
import com.nuvio.app.features.tracking.TrackingProviderId
|
||||
import com.nuvio.app.features.tracking.WatchProgressSource
|
||||
import kotlinx.coroutines.CancellationException
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFailsWith
|
||||
import kotlin.test.assertFalse
|
||||
import kotlin.test.assertNull
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class WatchedRepositoryTest {
|
||||
@Test
|
||||
fun emptyProviderExtraKeys_doNotTriggerInitialRefresh() {
|
||||
assertFalse(extraWatchedKeysChanged(previous = null, current = emptySet()))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun populatedProviderExtraKeys_triggerRefreshFromEmptyState() {
|
||||
assertTrue(extraWatchedKeysChanged(previous = null, current = setOf("series:tt1:-1:-1")))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun changedProviderExtraKeys_triggerRefresh() {
|
||||
assertTrue(
|
||||
extraWatchedKeysChanged(
|
||||
previous = setOf("series:tt1:-1:-1"),
|
||||
current = setOf("series:tt2:-1:-1"),
|
||||
),
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun providerRefreshFailure_isContainedWithoutReplacingState() = runBlocking {
|
||||
val failure = IllegalStateException("rate limited")
|
||||
var observedFailure: Throwable? = null
|
||||
|
||||
val result = watchedProviderRefreshOrNull(
|
||||
refresh = { throw failure },
|
||||
onFailure = { observedFailure = it },
|
||||
)
|
||||
|
||||
assertNull(result)
|
||||
assertEquals(failure, observedFailure)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun providerRefreshCancellation_isNotContained() = runBlocking {
|
||||
assertFailsWith<CancellationException> {
|
||||
watchedProviderRefreshOrNull(
|
||||
refresh = { throw CancellationException("cancelled") },
|
||||
onFailure = {},
|
||||
)
|
||||
}
|
||||
Unit
|
||||
}
|
||||
|
||||
@Test
|
||||
fun watchedItemKey_isTypeAware() {
|
||||
assertEquals("movie:tt1:-1:-1", watchedItemKey(type = "movie", id = "tt1"))
|
||||
|
|
|
|||
|
|
@ -0,0 +1,91 @@
|
|||
package com.nuvio.app.features.watching.sync
|
||||
|
||||
import com.nuvio.app.features.addons.DefaultRawHttpResponseMaxBytes
|
||||
import com.nuvio.app.features.addons.RawHttpResponse
|
||||
import kotlinx.coroutines.runBlocking
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFailsWith
|
||||
import kotlin.test.assertFalse
|
||||
|
||||
class TraktWatchedSyncAdapterTest {
|
||||
@Test
|
||||
fun `watched history responses larger than generic limit remain complete`() = runBlocking {
|
||||
val body = "x".repeat(DefaultRawHttpResponseMaxBytes + 1)
|
||||
var requestedLimit = 0
|
||||
val client = TraktWatchedPageClient(
|
||||
TraktWatchedHttpEngine { _, _, maxResponseBodyBytes ->
|
||||
requestedLimit = maxResponseBodyBytes
|
||||
val truncated = body.length > maxResponseBodyBytes
|
||||
RawHttpResponse(
|
||||
status = 200,
|
||||
statusText = "OK",
|
||||
url = "https://api.trakt.tv/sync/watched/shows",
|
||||
body = if (truncated) body.take(maxResponseBodyBytes) else body,
|
||||
headers = emptyMap(),
|
||||
)
|
||||
},
|
||||
)
|
||||
|
||||
val response = client.get(
|
||||
url = "https://api.trakt.tv/sync/watched/shows",
|
||||
headers = emptyMap(),
|
||||
)
|
||||
|
||||
assertEquals(TRAKT_WATCHED_MAX_RESPONSE_BODY_BYTES, requestedLimit)
|
||||
assertFalse(response.body.length < body.length)
|
||||
assertEquals(body, response.body)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `rate limited watched request retries once after retry after`() = runBlocking {
|
||||
var attempts = 0
|
||||
val delays = mutableListOf<Long>()
|
||||
val client = TraktWatchedPageClient(
|
||||
engine = TraktWatchedHttpEngine { _, _, _ ->
|
||||
attempts += 1
|
||||
response(
|
||||
status = if (attempts == 1) 429 else 200,
|
||||
headers = if (attempts == 1) mapOf("Retry-After" to "2") else emptyMap(),
|
||||
)
|
||||
},
|
||||
sleep = delays::add,
|
||||
)
|
||||
|
||||
val result = client.get("https://api.trakt.tv/sync/watched/shows", emptyMap())
|
||||
|
||||
assertEquals(200, result.status)
|
||||
assertEquals(2, attempts)
|
||||
assertEquals(listOf(2_000L), delays)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `repeated rate limit becomes a bounded typed failure`() = runBlocking {
|
||||
var attempts = 0
|
||||
val client = TraktWatchedPageClient(
|
||||
engine = TraktWatchedHttpEngine { _, _, _ ->
|
||||
attempts += 1
|
||||
response(status = 429)
|
||||
},
|
||||
sleep = {},
|
||||
)
|
||||
|
||||
val error = assertFailsWith<TraktWatchedHttpException> {
|
||||
client.get("https://api.trakt.tv/sync/watched/shows", emptyMap())
|
||||
}
|
||||
|
||||
assertEquals(429, error.status)
|
||||
assertEquals(2, attempts)
|
||||
}
|
||||
|
||||
private fun response(
|
||||
status: Int,
|
||||
headers: Map<String, String> = emptyMap(),
|
||||
): RawHttpResponse = RawHttpResponse(
|
||||
status = status,
|
||||
statusText = "",
|
||||
url = "https://api.trakt.tv/sync/watched/shows",
|
||||
body = "[]",
|
||||
headers = headers,
|
||||
)
|
||||
}
|
||||
|
|
@ -9,6 +9,9 @@ import com.nuvio.app.features.plugins.runtime.PluginRuntime
|
|||
import io.github.jan.supabase.postgrest.postgrest
|
||||
import io.github.jan.supabase.postgrest.query.Order
|
||||
import io.github.jan.supabase.postgrest.rpc
|
||||
import kotlinx.atomicfu.atomic
|
||||
import kotlinx.atomicfu.locks.SynchronizedObject
|
||||
import kotlinx.atomicfu.locks.synchronized
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.Job
|
||||
|
|
@ -52,6 +55,18 @@ private data class PluginPushItem(
|
|||
@SerialName("sort_order") val sortOrder: Int = 0,
|
||||
)
|
||||
|
||||
private data class PluginPersistenceSnapshot(
|
||||
val profileId: Int,
|
||||
val generation: Long,
|
||||
val revision: Long,
|
||||
val state: PluginsUiState,
|
||||
)
|
||||
|
||||
private data class LoadedPluginState(
|
||||
val state: PluginsUiState,
|
||||
val requiresMigration: Boolean,
|
||||
)
|
||||
|
||||
actual object PluginRepository {
|
||||
private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
|
||||
private val log = Logger.withTag("PluginRepository")
|
||||
|
|
@ -64,6 +79,10 @@ actual object PluginRepository {
|
|||
private var pulledFromServer = false
|
||||
private var currentProfileId = 1
|
||||
private val activeRefreshJobs = mutableMapOf<String, Job>()
|
||||
private val persistenceGeneration = atomic(0L)
|
||||
private val persistenceRevision = atomic(0L)
|
||||
private val persistenceLock = SynchronizedObject()
|
||||
private val persistedRevisionByProfile = mutableMapOf<Int, Long>()
|
||||
|
||||
actual fun initialize() {
|
||||
val effectiveProfileId = resolveEffectiveProfileId(ProfileRepository.activeProfileId)
|
||||
|
|
@ -89,6 +108,7 @@ actual object PluginRepository {
|
|||
|
||||
actual fun clearLocalState() {
|
||||
cancelActiveRefreshes()
|
||||
persistenceGeneration.incrementAndGet()
|
||||
currentProfileId = 1
|
||||
initialized = false
|
||||
pulledFromServer = false
|
||||
|
|
@ -358,59 +378,71 @@ actual object PluginRepository {
|
|||
private suspend fun fetchRepositoryData(
|
||||
manifestUrl: String,
|
||||
previousScrapers: Map<String, PluginScraper>,
|
||||
): Pair<PluginRepositoryItem, List<PluginScraper>> = withContext(Dispatchers.Default) {
|
||||
val payload = httpGetText(manifestUrl)
|
||||
val manifest = PluginManifestParser.parse(payload)
|
||||
val baseUrl = manifestUrl.substringBefore("?").removeSuffix("/manifest.json")
|
||||
): Pair<PluginRepositoryItem, List<PluginScraper>> {
|
||||
val storageProfileId = currentProfileId
|
||||
return withContext(Dispatchers.Default) {
|
||||
val payload = httpGetText(manifestUrl)
|
||||
val manifest = PluginManifestParser.parse(payload)
|
||||
val baseUrl = manifestUrl.substringBefore("?").removeSuffix("/manifest.json")
|
||||
|
||||
val scrapers = manifest.scrapers
|
||||
.filter { scraper -> scraper.isSupportedOnCurrentPlatform() }
|
||||
.mapNotNull { info ->
|
||||
val codeUrl = if (info.filename.startsWith("http://") || info.filename.startsWith("https://")) {
|
||||
info.filename
|
||||
} else {
|
||||
"$baseUrl/${info.filename.trimStart('/')}"
|
||||
}
|
||||
runCatching {
|
||||
val code = httpGetText(codeUrl)
|
||||
val scraperId = "${manifestUrl.lowercase()}:${info.id}"
|
||||
val previous = previousScrapers[scraperId]
|
||||
val enabled = when {
|
||||
!info.enabled -> false
|
||||
previous != null -> previous.enabled
|
||||
else -> info.enabled
|
||||
val scrapers = manifest.scrapers
|
||||
.filter { scraper -> scraper.isSupportedOnCurrentPlatform() }
|
||||
.mapNotNull { info ->
|
||||
val codeUrl = if (info.filename.startsWith("http://") || info.filename.startsWith("https://")) {
|
||||
info.filename
|
||||
} else {
|
||||
"$baseUrl/${info.filename.trimStart('/')}"
|
||||
}
|
||||
runCatching {
|
||||
val code = httpGetText(codeUrl)
|
||||
val scraperId = "${manifestUrl.lowercase()}:${info.id}"
|
||||
val cached = PluginStorage.saveScraperCode(
|
||||
profileId = storageProfileId,
|
||||
scraperId = scraperId,
|
||||
code = code,
|
||||
overwrite = true,
|
||||
)
|
||||
if (!cached) {
|
||||
log.w { "Failed to cache plugin scraper $scraperId" }
|
||||
}
|
||||
val previous = previousScrapers[scraperId]
|
||||
val enabled = when {
|
||||
!info.enabled -> false
|
||||
previous != null -> previous.enabled
|
||||
else -> info.enabled
|
||||
}
|
||||
|
||||
PluginScraper(
|
||||
id = scraperId,
|
||||
repositoryUrl = manifestUrl,
|
||||
name = info.name,
|
||||
description = info.description.orEmpty(),
|
||||
version = info.version,
|
||||
filename = info.filename,
|
||||
supportedTypes = info.supportedTypes,
|
||||
enabled = enabled,
|
||||
manifestEnabled = info.enabled,
|
||||
hasSettings = info.hasSettings,
|
||||
logo = info.logo,
|
||||
contentLanguage = info.contentLanguage ?: emptyList(),
|
||||
formats = info.formats ?: info.supportedFormats,
|
||||
code = code,
|
||||
)
|
||||
}.getOrNull()
|
||||
}
|
||||
PluginScraper(
|
||||
id = scraperId,
|
||||
repositoryUrl = manifestUrl,
|
||||
name = info.name,
|
||||
description = info.description.orEmpty(),
|
||||
version = info.version,
|
||||
filename = info.filename,
|
||||
supportedTypes = info.supportedTypes,
|
||||
enabled = enabled,
|
||||
manifestEnabled = info.enabled,
|
||||
hasSettings = info.hasSettings,
|
||||
logo = info.logo,
|
||||
contentLanguage = info.contentLanguage ?: emptyList(),
|
||||
formats = info.formats ?: info.supportedFormats,
|
||||
code = code,
|
||||
)
|
||||
}.getOrNull()
|
||||
}
|
||||
|
||||
val repo = PluginRepositoryItem(
|
||||
manifestUrl = manifestUrl,
|
||||
name = manifest.name,
|
||||
description = manifest.description,
|
||||
version = manifest.version,
|
||||
scraperCount = scrapers.size,
|
||||
lastUpdated = currentEpochMillis(),
|
||||
isRefreshing = false,
|
||||
errorMessage = null,
|
||||
)
|
||||
repo to scrapers
|
||||
val repo = PluginRepositoryItem(
|
||||
manifestUrl = manifestUrl,
|
||||
name = manifest.name,
|
||||
description = manifest.description,
|
||||
version = manifest.version,
|
||||
scraperCount = scrapers.size,
|
||||
lastUpdated = currentEpochMillis(),
|
||||
isRefreshing = false,
|
||||
errorMessage = null,
|
||||
)
|
||||
repo to scrapers
|
||||
}
|
||||
}
|
||||
|
||||
private fun PluginManifestScraper.isSupportedOnCurrentPlatform(): Boolean {
|
||||
|
|
@ -460,39 +492,49 @@ actual object PluginRepository {
|
|||
}
|
||||
|
||||
private fun persist() {
|
||||
val state = _uiState.value
|
||||
val payload = StoredPluginsState(
|
||||
pluginsEnabled = state.pluginsEnabled,
|
||||
groupStreamsByRepository = state.groupStreamsByRepository,
|
||||
repositories = state.repositories.map { repo ->
|
||||
StoredPluginRepository(
|
||||
manifestUrl = repo.manifestUrl,
|
||||
name = repo.name,
|
||||
description = repo.description,
|
||||
version = repo.version,
|
||||
scraperCount = repo.scraperCount,
|
||||
lastUpdated = repo.lastUpdated,
|
||||
)
|
||||
},
|
||||
scrapers = state.scrapers.map { scraper ->
|
||||
StoredPluginScraper(
|
||||
id = scraper.id,
|
||||
repositoryUrl = scraper.repositoryUrl,
|
||||
name = scraper.name,
|
||||
description = scraper.description,
|
||||
version = scraper.version,
|
||||
filename = scraper.filename,
|
||||
supportedTypes = scraper.supportedTypes,
|
||||
enabled = scraper.enabled,
|
||||
manifestEnabled = scraper.manifestEnabled,
|
||||
hasSettings = scraper.hasSettings,
|
||||
logo = scraper.logo,
|
||||
contentLanguage = scraper.contentLanguage,
|
||||
formats = scraper.formats,
|
||||
code = scraper.code,
|
||||
) },
|
||||
val snapshot = PluginPersistenceSnapshot(
|
||||
profileId = currentProfileId,
|
||||
generation = persistenceGeneration.value,
|
||||
revision = persistenceRevision.incrementAndGet(),
|
||||
state = _uiState.value,
|
||||
)
|
||||
PluginStorage.saveState(currentProfileId, json.encodeToString(payload))
|
||||
val requiresCodeWrite = snapshot.state.scrapers.any { scraper ->
|
||||
!PluginStorage.hasScraperCode(snapshot.profileId, scraper.id)
|
||||
}
|
||||
if (requiresCodeWrite) {
|
||||
scope.launch { persist(snapshot) }
|
||||
} else {
|
||||
persist(snapshot)
|
||||
}
|
||||
}
|
||||
|
||||
private fun persist(snapshot: PluginPersistenceSnapshot) {
|
||||
if (snapshot.generation != persistenceGeneration.value) return
|
||||
var cached = true
|
||||
snapshot.state.scrapers.forEach { scraper ->
|
||||
val scraperCached = PluginStorage.saveScraperCode(
|
||||
profileId = snapshot.profileId,
|
||||
scraperId = scraper.id,
|
||||
code = scraper.code,
|
||||
overwrite = false,
|
||||
)
|
||||
cached = scraperCached && cached
|
||||
}
|
||||
if (!cached || snapshot.generation != persistenceGeneration.value) {
|
||||
if (!cached) {
|
||||
log.w { "Failed to persist plugin scraper cache for profile ${snapshot.profileId}" }
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
synchronized(persistenceLock) {
|
||||
if (snapshot.generation != persistenceGeneration.value) return@synchronized
|
||||
val persistedRevision = persistedRevisionByProfile[snapshot.profileId] ?: Long.MIN_VALUE
|
||||
if (snapshot.revision < persistedRevision) return@synchronized
|
||||
val payload = snapshot.state.toStoredPluginsState()
|
||||
PluginStorage.saveState(snapshot.profileId, json.encodeToString(payload))
|
||||
persistedRevisionByProfile[snapshot.profileId] = snapshot.revision
|
||||
}
|
||||
}
|
||||
|
||||
private fun loadStoredState(profileId: Int): StoredPluginsState? {
|
||||
|
|
@ -517,49 +559,45 @@ actual object PluginRepository {
|
|||
}
|
||||
|
||||
currentProfileId = profileId
|
||||
_uiState.value = loadStateAsUiState(profileId)
|
||||
val loadedState = loadStateAsUiState(profileId)
|
||||
_uiState.value = loadedState.state
|
||||
initialized = true
|
||||
if (loadedState.requiresMigration) persist()
|
||||
}
|
||||
|
||||
private fun loadStateAsUiState(profileId: Int): PluginsUiState {
|
||||
private fun loadStateAsUiState(profileId: Int): LoadedPluginState {
|
||||
val stored = loadStoredState(profileId)
|
||||
return PluginsUiState(
|
||||
pluginsEnabled = stored?.pluginsEnabled ?: true,
|
||||
groupStreamsByRepository = stored?.groupStreamsByRepository ?: false,
|
||||
repositories = stored?.repositories
|
||||
?.map {
|
||||
PluginRepositoryItem(
|
||||
manifestUrl = it.manifestUrl,
|
||||
name = it.name,
|
||||
description = it.description,
|
||||
version = it.version,
|
||||
scraperCount = it.scraperCount,
|
||||
lastUpdated = it.lastUpdated,
|
||||
isRefreshing = false,
|
||||
errorMessage = null,
|
||||
)
|
||||
}
|
||||
?: emptyList(),
|
||||
scrapers = stored?.scrapers
|
||||
?.map {
|
||||
PluginScraper(
|
||||
id = it.id,
|
||||
repositoryUrl = it.repositoryUrl,
|
||||
name = it.name,
|
||||
description = it.description,
|
||||
version = it.version,
|
||||
filename = it.filename,
|
||||
supportedTypes = it.supportedTypes,
|
||||
enabled = it.enabled,
|
||||
manifestEnabled = it.manifestEnabled,
|
||||
hasSettings = it.hasSettings,
|
||||
logo = it.logo,
|
||||
contentLanguage = it.contentLanguage,
|
||||
formats = it.formats,
|
||||
code = it.code,
|
||||
)
|
||||
}
|
||||
?: emptyList(),
|
||||
var requiresMigration = false
|
||||
val scrapers = stored?.scrapers
|
||||
?.mapNotNull { storedScraper ->
|
||||
storedScraper.restorePluginScraper { scraperId ->
|
||||
PluginStorage.loadScraperCode(profileId, scraperId)
|
||||
}?.also { restored ->
|
||||
requiresMigration = requiresMigration || restored.requiresMigration
|
||||
}?.scraper
|
||||
}
|
||||
?: emptyList()
|
||||
return LoadedPluginState(
|
||||
state = PluginsUiState(
|
||||
pluginsEnabled = stored?.pluginsEnabled ?: true,
|
||||
groupStreamsByRepository = stored?.groupStreamsByRepository ?: false,
|
||||
repositories = stored?.repositories
|
||||
?.map {
|
||||
PluginRepositoryItem(
|
||||
manifestUrl = it.manifestUrl,
|
||||
name = it.name,
|
||||
description = it.description,
|
||||
version = it.version,
|
||||
scraperCount = it.scraperCount,
|
||||
lastUpdated = it.lastUpdated,
|
||||
isRefreshing = false,
|
||||
errorMessage = null,
|
||||
)
|
||||
}
|
||||
?: emptyList(),
|
||||
scrapers = scrapers,
|
||||
),
|
||||
requiresMigration = requiresMigration,
|
||||
)
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,10 +1,30 @@
|
|||
@file:OptIn(kotlinx.cinterop.ExperimentalForeignApi::class)
|
||||
|
||||
package com.nuvio.app.features.plugins
|
||||
|
||||
import kotlinx.atomicfu.locks.SynchronizedObject
|
||||
import kotlinx.atomicfu.locks.synchronized
|
||||
import kotlinx.cinterop.ExperimentalForeignApi
|
||||
import kotlinx.cinterop.addressOf
|
||||
import kotlinx.cinterop.convert
|
||||
import kotlinx.cinterop.usePinned
|
||||
import platform.Foundation.NSFileManager
|
||||
import platform.Foundation.NSHomeDirectory
|
||||
import platform.Foundation.NSUserDefaults
|
||||
import platform.Foundation.timeIntervalSince1970
|
||||
import platform.posix.SEEK_END
|
||||
import platform.posix.fclose
|
||||
import platform.posix.fopen
|
||||
import platform.posix.fread
|
||||
import platform.posix.fseek
|
||||
import platform.posix.ftell
|
||||
import platform.posix.fwrite
|
||||
import platform.posix.rewind
|
||||
|
||||
internal object PluginStorage {
|
||||
private const val pluginsStateKey = "plugins_state"
|
||||
private const val scraperCodeDirectoryName = "nuvio_plugin_scrapers"
|
||||
private val scraperCodeLock = SynchronizedObject()
|
||||
|
||||
fun loadState(profileId: Int): String? =
|
||||
NSUserDefaults.standardUserDefaults.stringForKey("${pluginsStateKey}_$profileId")
|
||||
|
|
@ -16,6 +36,53 @@ internal object PluginStorage {
|
|||
)
|
||||
}
|
||||
|
||||
fun hasScraperCode(profileId: Int, scraperId: String): Boolean =
|
||||
NSFileManager.defaultManager.fileExistsAtPath(scraperCodePath(profileId, scraperId))
|
||||
|
||||
fun loadScraperCode(profileId: Int, scraperId: String): String? = synchronized(scraperCodeLock) {
|
||||
readUtf8File(scraperCodePath(profileId, scraperId))
|
||||
}
|
||||
|
||||
fun saveScraperCode(
|
||||
profileId: Int,
|
||||
scraperId: String,
|
||||
code: String,
|
||||
overwrite: Boolean,
|
||||
): Boolean {
|
||||
val manager = NSFileManager.defaultManager
|
||||
val target = scraperCodePath(profileId, scraperId)
|
||||
if (!overwrite && manager.fileExistsAtPath(target)) return true
|
||||
return synchronized(scraperCodeLock) {
|
||||
val directory = scraperCodeDirectory(profileId)
|
||||
if (!manager.createDirectoryAtPath(directory, true, null, null)) return@synchronized false
|
||||
if (!overwrite && manager.fileExistsAtPath(target)) return@synchronized true
|
||||
|
||||
val temporary = "$target.tmp"
|
||||
val backup = "$target.backup"
|
||||
if (!writeUtf8File(temporary, code)) return@synchronized false
|
||||
|
||||
try {
|
||||
if (!overwrite && manager.fileExistsAtPath(target)) return@synchronized true
|
||||
if (manager.fileExistsAtPath(backup) && !manager.removeItemAtPath(backup, null)) {
|
||||
return@synchronized false
|
||||
}
|
||||
val hadTarget = manager.fileExistsAtPath(target)
|
||||
if (hadTarget && !manager.moveItemAtPath(target, backup, null)) return@synchronized false
|
||||
if (!manager.moveItemAtPath(temporary, target, null)) {
|
||||
if (hadTarget) manager.moveItemAtPath(backup, target, null)
|
||||
return@synchronized false
|
||||
}
|
||||
if (manager.fileExistsAtPath(backup)) manager.removeItemAtPath(backup, null)
|
||||
true
|
||||
} finally {
|
||||
if (manager.fileExistsAtPath(temporary)) manager.removeItemAtPath(temporary, null)
|
||||
if (manager.fileExistsAtPath(backup) && !manager.fileExistsAtPath(target)) {
|
||||
manager.moveItemAtPath(backup, target, null)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fun loadScraperSettings(scraperId: String): String? =
|
||||
NSUserDefaults.standardUserDefaults.stringForKey("settings_${scraperId}")
|
||||
|
||||
|
|
@ -25,6 +92,51 @@ internal object PluginStorage {
|
|||
forKey = "settings_${scraperId}",
|
||||
)
|
||||
}
|
||||
|
||||
private fun scraperCodeDirectory(profileId: Int): String =
|
||||
"${NSHomeDirectory()}/Library/Application Support/$scraperCodeDirectoryName/$profileId"
|
||||
|
||||
private fun scraperCodePath(profileId: Int, scraperId: String): String =
|
||||
"${scraperCodeDirectory(profileId)}/${pluginDigestHex("SHA256", scraperId)}.js"
|
||||
}
|
||||
|
||||
@OptIn(ExperimentalForeignApi::class)
|
||||
private fun readUtf8File(path: String): String? {
|
||||
val file = fopen(path, "rb") ?: return null
|
||||
return try {
|
||||
if (fseek(file, 0L, SEEK_END) != 0) return null
|
||||
val size = ftell(file)
|
||||
if (size < 0L || size > Int.MAX_VALUE.toLong()) return null
|
||||
rewind(file)
|
||||
val bytes = ByteArray(size.toInt())
|
||||
if (bytes.isNotEmpty()) {
|
||||
val read = bytes.usePinned { pinned ->
|
||||
fread(pinned.addressOf(0), 1.convert(), bytes.size.convert(), file)
|
||||
}
|
||||
if (read.toLong() != bytes.size.toLong()) return null
|
||||
}
|
||||
bytes.decodeToString()
|
||||
} finally {
|
||||
fclose(file)
|
||||
}
|
||||
}
|
||||
|
||||
@OptIn(ExperimentalForeignApi::class)
|
||||
private fun writeUtf8File(path: String, value: String): Boolean {
|
||||
val bytes = value.encodeToByteArray()
|
||||
val file = fopen(path, "wb") ?: return false
|
||||
return try {
|
||||
if (bytes.isEmpty()) {
|
||||
true
|
||||
} else {
|
||||
val written = bytes.usePinned { pinned ->
|
||||
fwrite(pinned.addressOf(0), 1.convert(), bytes.size.convert(), file)
|
||||
}
|
||||
written.toLong() == bytes.size.toLong()
|
||||
}
|
||||
} finally {
|
||||
fclose(file)
|
||||
}
|
||||
}
|
||||
|
||||
internal fun currentPluginPlatform(): String = "ios"
|
||||
|
|
|
|||
|
|
@ -1,6 +1,10 @@
|
|||
@file:OptIn(kotlinx.cinterop.ExperimentalForeignApi::class)
|
||||
|
||||
package com.nuvio.app.core.storage
|
||||
|
||||
import platform.Foundation.NSUserDefaults
|
||||
import platform.Foundation.NSFileManager
|
||||
import platform.Foundation.NSHomeDirectory
|
||||
import com.nuvio.app.features.profiles.MAX_PROFILES
|
||||
|
||||
internal actual object PlatformLocalAccountDataCleaner {
|
||||
|
|
@ -90,5 +94,10 @@ internal actual object PlatformLocalAccountDataCleaner {
|
|||
defaults.removeObjectForKey(keyString)
|
||||
}
|
||||
}
|
||||
|
||||
val scraperCodePath = "${NSHomeDirectory()}/Library/Application Support/nuvio_plugin_scrapers"
|
||||
if (NSFileManager.defaultManager.fileExistsAtPath(scraperCodePath)) {
|
||||
NSFileManager.defaultManager.removeItemAtPath(scraperCodePath, null)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue