diff --git a/composeApp/build.gradle.kts b/composeApp/build.gradle.kts index 02d3a58d..a4e0e4db 100644 --- a/composeApp/build.gradle.kts +++ b/composeApp/build.gradle.kts @@ -226,6 +226,7 @@ val iosDistributionSourceDir = if (iosDistribution == "full") { "src/iosAppStore/kotlin" } val iosFrameworkBundleId = "com.nuvio.media" +val nuvioEngineAppleFramework = rootProject.file("../nuvio-engine/platform/apple/NuvioEngine.xcframework") val fullCommonSourceDir = project.file("src/fullCommonMain/kotlin") val generatedRuntimeConfigDir = layout.buildDirectory.dir("generated/runtime-config/kotlin") val requestedGradleTasks = gradle.startParameter.taskNames.map { taskName -> @@ -329,12 +330,28 @@ kotlin { ) iosTargets.forEach { iosTarget -> + val nuvioEngineSlice = if (iosTarget.name == "iosArm64") { + "ios-arm64" + } else { + "ios-arm64_x86_64-simulator" + } + val nuvioEngineSliceDirectory = nuvioEngineAppleFramework.resolve(nuvioEngineSlice) iosTarget.compilations.getByName("main") { cinterops { create("commoncrypto") { defFile(project.file("src/nativeInterop/cinterop/commoncrypto.def")) compilerOpts("-I${project.projectDir}/src/nativeInterop/cinterop") } + if (iosDistribution == "full") { + check(nuvioEngineSliceDirectory.resolve("libCNuvioEngine.a").isFile) { + "Build the local Nuvio Engine Apple XCFramework before compiling iOS Full." + } + create("nuvioengine") { + defFile(project.file("src/nativeInterop/cinterop/nuvioengine.def")) + compilerOpts("-I${nuvioEngineSliceDirectory.resolve("Headers").absolutePath}") + extraOpts("-libraryPath", nuvioEngineSliceDirectory.absolutePath) + } + } } if (iosDistribution == "full") { @@ -354,6 +371,14 @@ kotlin { baseName = "ComposeApp" isStatic = true freeCompilerArgs += listOf("-Xbinary=bundleId=$iosFrameworkBundleId") + if (iosDistribution == "full") { + linkerOpts( + "-lc++", + "-framework", "Security", + "-framework", "SystemConfiguration", + "-framework", "CoreFoundation", + ) + } } } @@ -392,6 +417,7 @@ kotlin { implementation(libs.androidx.media3.container) implementation(libs.androidx.media3.extractor) implementation(libs.mpv.android.lib) + implementation("com.nuvio:nuvio-engine-android:0.1.0-SNAPSHOT") implementation(fileTree(mapOf("dir" to "libs", "include" to listOf("lib-*.aar")))) if (androidDistribution == "full") { implementation(files("libs/quickjs-kt-android-1.0.5-nuvio.aar")) diff --git a/composeApp/src/androidFull/kotlin/com/nuvio/app/features/player/PlatformPlaybackDataSourceFactory.android.kt b/composeApp/src/androidFull/kotlin/com/nuvio/app/features/player/PlatformPlaybackDataSourceFactory.android.kt index c39674bd..c488bb4b 100644 --- a/composeApp/src/androidFull/kotlin/com/nuvio/app/features/player/PlatformPlaybackDataSourceFactory.android.kt +++ b/composeApp/src/androidFull/kotlin/com/nuvio/app/features/player/PlatformPlaybackDataSourceFactory.android.kt @@ -11,12 +11,16 @@ internal object PlatformPlaybackDataSourceFactory { defaultRequestHeaders: Map, defaultResponseHeaders: Map, useYoutubeChunkedPlayback: Boolean, + useLongReadTimeout: Boolean = false, externalSubtitles: List = emptyList(), ): DataSource.Factory { val networkFactory: DataSource.Factory = if (useYoutubeChunkedPlayback) { YoutubeChunkedDataSourceFactory(defaultRequestHeaders = defaultRequestHeaders) } else { - PlayerPlaybackNetworking.createHttpDataSourceFactory(defaultRequestHeaders) + PlayerPlaybackNetworking.createHttpDataSourceFactory( + defaultRequestHeaders, + useLongReadTimeout, + ) } val subtitleHeaderFactory = SubtitleRequestHeaderDataSourceFactory( upstreamFactory = networkFactory, @@ -32,4 +36,4 @@ internal object PlatformPlaybackDataSourceFactory { ) } } -} \ No newline at end of file +} diff --git a/composeApp/src/androidHostTest/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngineAndroidTest.kt b/composeApp/src/androidHostTest/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngineAndroidTest.kt new file mode 100644 index 00000000..393e18ea --- /dev/null +++ b/composeApp/src/androidHostTest/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngineAndroidTest.kt @@ -0,0 +1,92 @@ +package com.nuvio.app.features.p2p + +import com.nuvio.engine.NuvioUploadMode +import com.nuvio.engine.NuvioTorrentProfile +import java.io.File +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertNull + +class P2pStreamingEngineAndroidTest { + @Test + fun appOwnedRoutesDisableAutomaticInactivityExpiry() { + val stateDirectory = File("state") + val cacheDirectory = File("cache") + + val uploading = buildNuvioEngineConfig( + stateDirectory = stateDirectory, + cacheDirectory = cacheDirectory, + uploadEnabled = true, + torrentProfile = P2pTorrentProfile.FAST, + diskCacheCapacityBytes = P2pCacheSize.GB_5.bytes, + ) + val downloadOnly = buildNuvioEngineConfig( + stateDirectory = stateDirectory, + cacheDirectory = cacheDirectory, + uploadEnabled = false, + torrentProfile = P2pTorrentProfile.SOFT, + diskCacheCapacityBytes = P2pCacheSize.NONE.bytes, + ) + + assertEquals(0, uploading.streamInactivityTimeoutMilliseconds) + assertEquals(NuvioUploadMode.Unlimited, uploading.uploadMode) + assertEquals(NuvioTorrentProfile.Fast, uploading.torrentProfile) + assertEquals(P2pCacheSize.GB_5.bytes, uploading.diskCacheCapacityBytes) + assertEquals(0, downloadOnly.streamInactivityTimeoutMilliseconds) + assertEquals(NuvioUploadMode.Disabled, downloadOnly.uploadMode) + assertEquals(NuvioTorrentProfile.Soft, downloadOnly.torrentProfile) + assertEquals(0L, downloadOnly.diskCacheCapacityBytes) + } + + @Test + fun matchingUnsolicitedStopBecomesTerminalError() { + val error = unexpectedStreamStopError( + requestId = 0L, + eventStreamId = "stream", + currentStreamId = "stream", + message = "stream expired after inactivity", + fallbackMessage = "unknown", + ) + + assertEquals( + P2pStreamingState.Error("stream expired after inactivity"), + error, + ) + } + + @Test + fun explicitAndStaleStopsAreIgnored() { + assertNull( + unexpectedStreamStopError( + requestId = 7L, + eventStreamId = "stream", + currentStreamId = "stream", + message = "stopped", + fallbackMessage = "unknown", + ) + ) + assertNull( + unexpectedStreamStopError( + requestId = 0L, + eventStreamId = "old-stream", + currentStreamId = "stream", + message = "stopped", + fallbackMessage = "unknown", + ) + ) + } + + @Test + fun blankStopMessageUsesFallback() { + assertEquals( + P2pStreamingState.Error("unknown"), + unexpectedStreamStopError( + requestId = 0L, + eventStreamId = "stream", + currentStreamId = "stream", + message = " ", + fallbackMessage = "unknown", + ), + ) + } +} diff --git a/composeApp/src/androidMain/jniLibs/arm64-v8a/libtorrserver.so b/composeApp/src/androidMain/jniLibs/arm64-v8a/libtorrserver.so deleted file mode 100755 index 12994d13..00000000 Binary files a/composeApp/src/androidMain/jniLibs/arm64-v8a/libtorrserver.so and /dev/null differ diff --git a/composeApp/src/androidMain/jniLibs/armeabi-v7a/libtorrserver.so b/composeApp/src/androidMain/jniLibs/armeabi-v7a/libtorrserver.so deleted file mode 100755 index 6e7bc6bc..00000000 Binary files a/composeApp/src/androidMain/jniLibs/armeabi-v7a/libtorrserver.so and /dev/null differ diff --git a/composeApp/src/androidMain/jniLibs/x86/libtorrserver.so b/composeApp/src/androidMain/jniLibs/x86/libtorrserver.so deleted file mode 100755 index 6309221c..00000000 Binary files a/composeApp/src/androidMain/jniLibs/x86/libtorrserver.so and /dev/null differ diff --git a/composeApp/src/androidMain/jniLibs/x86_64/libtorrserver.so b/composeApp/src/androidMain/jniLibs/x86_64/libtorrserver.so deleted file mode 100755 index 039eba7d..00000000 Binary files a/composeApp/src/androidMain/jniLibs/x86_64/libtorrserver.so and /dev/null differ diff --git a/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pSettingsStorage.android.kt b/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pSettingsStorage.android.kt index 0e3a39a7..7f83e2b3 100644 --- a/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pSettingsStorage.android.kt +++ b/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pSettingsStorage.android.kt @@ -9,6 +9,8 @@ internal actual object P2pSettingsStorage { private const val p2pEnabledKey = "p2p_enabled" private const val enableUploadKey = "enable_upload" private const val hideTorrentStatsKey = "hide_torrent_stats" + private const val torrentProfileKey = "torrent_profile" + private const val cacheSizeKey = "cache_size" private var preferences: SharedPreferences? = null @@ -37,6 +39,18 @@ internal actual object P2pSettingsStorage { saveBoolean(hideTorrentStatsKey, enabled) } + actual fun loadTorrentProfile(): String? = loadString(torrentProfileKey) + + actual fun saveTorrentProfile(profile: String) { + saveString(torrentProfileKey, profile) + } + + actual fun loadCacheSize(): String? = loadString(cacheSizeKey) + + actual fun saveCacheSize(size: String) { + saveString(cacheSizeKey, size) + } + private fun loadBoolean(keyBase: String): Boolean? = preferences?.let { sharedPreferences -> val key = ProfileScopedKey.of(keyBase) @@ -53,4 +67,14 @@ internal actual object P2pSettingsStorage { ?.putBoolean(ProfileScopedKey.of(keyBase), value) ?.apply() } + + private fun loadString(keyBase: String): String? = + preferences?.getString(ProfileScopedKey.of(keyBase), null) + + private fun saveString(keyBase: String, value: String) { + preferences + ?.edit() + ?.putString(ProfileScopedKey.of(keyBase), value) + ?.apply() + } } diff --git a/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.android.kt b/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.android.kt index 93aecad7..b7523e66 100644 --- a/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.android.kt +++ b/composeApp/src/androidMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.android.kt @@ -1,172 +1,775 @@ package com.nuvio.app.features.p2p import android.content.Context +import android.os.SystemClock import android.util.Log import com.nuvio.app.core.i18n.localizedP2pUnknownTorrentError +import com.nuvio.engine.NuvioEngine +import com.nuvio.engine.NuvioEngineConfig +import com.nuvio.engine.NuvioEventType +import com.nuvio.engine.NuvioStream +import com.nuvio.engine.NuvioTorrentProfile +import com.nuvio.engine.NuvioUploadMode +import java.io.File +import java.util.concurrent.atomic.AtomicReference import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.currentCoroutineContext import kotlinx.coroutines.delay +import kotlinx.coroutines.ensureActive import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.flow.collect import kotlinx.coroutines.isActive import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock import kotlinx.coroutines.withContext -import okhttp3.MediaType.Companion.toMediaType -import okhttp3.OkHttpClient -import okhttp3.Request -import okhttp3.RequestBody.Companion.toRequestBody -import org.json.JSONArray -import org.json.JSONObject -import java.io.File -import java.net.URLEncoder -import java.util.concurrent.TimeUnit private const val TAG = "P2pStreamingEngine" -private val VIDEO_EXTENSIONS = setOf("mkv", "mp4", "avi", "webm", "ts", "m4v", "mov", "wmv", "flv") +private const val DIAGNOSTIC_TAG = "NuvioP2PDiag" +private const val DIAGNOSTIC_SAMPLE_INTERVAL_MS = 1_000L + +internal fun buildNuvioEngineConfig( + stateDirectory: File, + cacheDirectory: File, + uploadEnabled: Boolean, + torrentProfile: P2pTorrentProfile, + diskCacheCapacityBytes: Long, +): NuvioEngineConfig = NuvioEngineConfig( + dataDirectory = stateDirectory, + cacheDirectory = cacheDirectory, + diskCacheCapacityBytes = diskCacheCapacityBytes, + torrentProfile = when (torrentProfile) { + P2pTorrentProfile.SOFT -> NuvioTorrentProfile.Soft + P2pTorrentProfile.BALANCED -> NuvioTorrentProfile.Balanced + P2pTorrentProfile.FAST -> NuvioTorrentProfile.Fast + }, + uploadMode = if (uploadEnabled) { + NuvioUploadMode.Unlimited + } else { + NuvioUploadMode.Disabled + }, + streamInactivityTimeoutMilliseconds = 0, +) + +internal fun unexpectedStreamStopError( + requestId: Long, + eventStreamId: String?, + currentStreamId: String?, + message: String?, + fallbackMessage: String, +): P2pStreamingState.Error? { + if (requestId != 0L || currentStreamId == null || eventStreamId != currentStreamId) { + return null + } + return P2pStreamingState.Error( + message?.trim()?.takeIf(String::isNotEmpty) ?: fallbackMessage, + ) +} actual object P2pStreamingEngine { + private data class EngineConfigurationKey( + val uploadEnabled: Boolean, + val torrentProfile: P2pTorrentProfile, + val diskCacheCapacityBytes: Long, + ) + + private data class DetachedStream( + val engine: NuvioEngine?, + val streamId: String?, + ) + private val _state = MutableStateFlow(P2pStreamingState.Idle) actual val state: StateFlow = _state.asStateFlow() + private val _cacheState = MutableStateFlow(P2pCacheUiState()) + actual val cacheState: StateFlow = _cacheState.asStateFlow() private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) private val lifecycleLock = Any() + private val startMutex = Mutex() private var statsJob: Job? = null private var cleanupJob: Job? = null - private var currentHash: String? = null + private var engineEventsJob: Job? = null private var streamGeneration = 0L + @Volatile + private var currentTorrentId: String? = null + @Volatile + private var currentStreamId: String? = null private var appContext: Context? = null - private val binary = TorrServerBinary() - private val api = TorrServerApi(binary) + @Volatile + private var engine: NuvioEngine? = null + private var engineConfigurationKey: EngineConfigurationKey? = null + private val knownTorrentIds = mutableSetOf() + private var diagnosticRequestSequence = 0L fun initialize(context: Context) { appContext = context.applicationContext - binary.initialize(context.applicationContext) } actual suspend fun startStream(request: P2pStreamRequest): String = withContext(Dispatchers.IO) { - stopStreamNow(stopBinary = false) - val generation = nextStreamGeneration() - _state.value = P2pStreamingState.Connecting + startMutex.withLock { startStreamLocked(request) } + } - try { - binary.start() + actual suspend fun clearCache(): P2pCacheClearResult = withContext(Dispatchers.IO) { + startMutex.withLock { + check(_state.value !is P2pStreamingState.Streaming && + _state.value !is P2pStreamingState.Connecting) { + "Torrent cache cannot be cleared during active playback" + } + _cacheState.value = _cacheState.value.copy(isClearing = true) + try { + val activeEngine = ensureEngine() + if (!_cacheState.value.hasMeasurement) { + delay(DIAGNOSTIC_SAMPLE_INTERVAL_MS + 100L) + val initial = activeEngine.stats.value + updateCacheState( + initial.diskCacheUsedBytes, + initial.diskCacheProtectedBytes, + ) + } + val before = activeEngine.stats.value + activeEngine.reclaimDiskCache(0L) + delay(DIAGNOSTIC_SAMPLE_INTERVAL_MS + 100L) + val after = activeEngine.stats.value + updateCacheState(after.diskCacheUsedBytes, after.diskCacheProtectedBytes) + P2pCacheClearResult( + reclaimedBytes = (before.diskCacheUsedBytes - after.diskCacheUsedBytes) + .coerceAtLeast(0L), + remainingBytes = after.diskCacheUsedBytes, + protectedBytes = after.diskCacheProtectedBytes, + ) + } finally { + _cacheState.value = _cacheState.value.copy(isClearing = false) + } + } + } + + private suspend fun startStreamLocked(request: P2pStreamRequest): String { + val requestSequence = nextDiagnosticRequestSequence() + val startedAtMs = SystemClock.elapsedRealtime() + val phase = AtomicReference("stop_previous") + Log.i( + DIAGNOSTIC_TAG, + "start request=$requestSequence phase=accepted hash=${diagnosticId(request.infoHash)} " + + "fileIndex=${request.fileIdx ?: -1} filenameHint=${!request.filename.isNullOrBlank()} " + + "requestTrackers=${request.trackers.size}", + ) + stopStreamNow(shutdownEngine = false) + logPhase(requestSequence, startedAtMs, phase.get()) + val generation = beginStreamGeneration() + + var activeEngine: NuvioEngine? = null + var payloadDownloadBaseline = 0L + var preparedStream: NuvioStream? = null + var attached = false + var startupStatsJob: Job? = null + return try { + phase.set("build_magnet") + val magnetUri = buildP2pMagnetUri( + request.infoHash, + (DEFAULT_TRACKERS + request.trackers).distinct(), + ) + logPhase(requestSequence, startedAtMs, phase.get()) + + phase.set("ensure_engine") + val resolvedEngine = ensureEngine() + activeEngine = resolvedEngine + payloadDownloadBaseline = resolvedEngine.stats.value.totalPayloadDownloadBytes ensureCurrentGeneration(generation) + logPhase(requestSequence, startedAtMs, phase.get()) + startupStatsJob = startStartupStatsPolling( + activeEngine = resolvedEngine, + generation = generation, + phase = phase, + requestSequence = requestSequence, + startedAtMs = startedAtMs, + ) - val magnetLink = buildMagnetUri(request.infoHash, request.trackers) - Log.d(TAG, "Starting stream: $magnetLink") + phase.set("add_magnet") + val canonicalHash = canonicalP2pInfoHash(request.infoHash) + val reusedTorrent = canonicalHash in knownTorrentIds + Log.i( + DIAGNOSTIC_TAG, + "start request=$requestSequence phase=${phase.get()} begin elapsedMs=${elapsedSince(startedAtMs)} " + + "cached=$reusedTorrent hash=${diagnosticId(canonicalHash)}", + ) + val torrentId = if (reusedTorrent) { + canonicalHash + } else { + resolvedEngine.addMagnet(magnetUri).also { knownTorrentIds += it } + } + ensureCurrentGeneration(generation) + Log.i( + DIAGNOSTIC_TAG, + "start request=$requestSequence phase=${phase.get()} complete elapsedMs=${elapsedSince(startedAtMs)} " + + "torrent=${diagnosticId(torrentId)} cached=$reusedTorrent", + ) - val hash = api.addTorrent(magnetLink) - ?: throw P2pStreamingException("Failed to add torrent") - if (!attachTorrentIfCurrent(generation, hash)) { - api.dropTorrent(hash) + phase.set("prepare_stream") + Log.i( + DIAGNOSTIC_TAG, + "start request=$requestSequence phase=${phase.get()} begin elapsedMs=${elapsedSince(startedAtMs)} " + + "fileIndex=${request.fileIdx ?: -1}", + ) + val stream = resolvedEngine.prepareStream( + torrentId = torrentId, + fileIndex = request.fileIdx, + filenameHint = request.filename, + ) + preparedStream = stream + Log.i( + DIAGNOSTIC_TAG, + "start request=$requestSequence phase=${phase.get()} complete elapsedMs=${elapsedSince(startedAtMs)} " + + "stream=${diagnosticId(stream.id)} selectedFile=${stream.fileIndex} fileBytes=${stream.fileSize} " + + "url=${diagnosticLoopbackUrl(stream.url)}", + ) + currentCoroutineContext().ensureActive() + phase.set("attach_route") + if (!attachStreamIfCurrent(generation, torrentId, stream.id)) { + withContext(NonCancellable) { + stopPreparedStream(resolvedEngine, stream.id) + } + preparedStream = null throw CancellationException("P2P stream start was cancelled") } + attached = true + logPhase(requestSequence, startedAtMs, phase.get()) - val resolvedIdx = resolveFileIndex( - hash = hash, - requestedIdx = request.fileIdx, - filename = request.filename, + startStatsPolling( + activeEngine = resolvedEngine, + stream = stream, + generation = generation, + requestSequence = requestSequence, + startedAtMs = startedAtMs, + payloadDownloadBaseline = payloadDownloadBaseline, ) - ensureCurrentGeneration(generation) - - val streamUrl = api.getStreamUrl(magnetLink, resolvedIdx) - Log.d(TAG, "Stream URL: $streamUrl") - - startStatsPolling(hash, generation) - - ensureCurrentGeneration(generation) - _state.value = P2pStreamingState.Streaming( - localUrl = streamUrl, - downloadSpeed = 0, - uploadSpeed = 0, - peers = 0, - seeds = 0, - bufferProgress = 0f, - totalProgress = 0f, - ) - - streamUrl - } catch (e: CancellationException) { - throw e - } catch (e: Exception) { - if (isCurrentGeneration(generation)) { - _state.value = P2pStreamingState.Error(e.message ?: localizedP2pUnknownTorrentError()) + val initialAggregate = resolvedEngine.stats.value + if (!publishStreamingIfCurrent( + generation = generation, + state = P2pStreamingState.Streaming( + localUrl = stream.url, + downloadSpeed = initialAggregate.downloadRateBytesPerSecond, + uploadSpeed = initialAggregate.uploadRateBytesPerSecond, + peers = initialAggregate.connectedPeers, + seeds = initialAggregate.connectedSeeds, + bufferProgress = 0f, + totalProgress = 0f, + downloadedBytes = (initialAggregate.totalPayloadDownloadBytes - + payloadDownloadBaseline).coerceAtLeast(0L), + ), + )) { + throw CancellationException("P2P stream start was cancelled") } - throw e + Log.i(TAG, "Nuvio Engine stream ready: ${stream.url}") + Log.i( + DIAGNOSTIC_TAG, + "start request=$requestSequence phase=route_ready elapsedMs=${elapsedSince(startedAtMs)} " + + "generation=$generation stream=${diagnosticId(stream.id)}", + ) + stream.url + } catch (cancellation: CancellationException) { + Log.w( + DIAGNOSTIC_TAG, + "start request=$requestSequence phase=${phase.get()} cancelled elapsedMs=${elapsedSince(startedAtMs)} " + + "generation=$generation", + ) + withContext(NonCancellable) { + cleanupFailedStart( + generation = generation, + activeEngine = activeEngine, + preparedStream = preparedStream, + attached = attached, + terminalState = P2pStreamingState.Idle, + ) + } + throw cancellation + } catch (error: Exception) { + Log.e( + DIAGNOSTIC_TAG, + "start request=$requestSequence phase=${phase.get()} failed elapsedMs=${elapsedSince(startedAtMs)} " + + "generation=$generation error=${diagnosticMessage(error.message)}", + error, + ) + val terminalState = P2pStreamingState.Error( + error.message ?: localizedP2pUnknownTorrentError() + ) + withContext(NonCancellable) { + cleanupFailedStart( + generation = generation, + activeEngine = activeEngine, + preparedStream = preparedStream, + attached = attached, + terminalState = terminalState, + ) + } + throw error + } finally { + startupStatsJob?.cancel() } } actual fun stopStream() { - scheduleStop(stopBinary = false) + scheduleStop(shutdownEngine = false) } actual fun shutdown() { - scheduleStop(stopBinary = true) + scheduleStop(shutdownEngine = true) } - private fun scheduleStop(stopBinary: Boolean) { - val hash = detachActiveStream() + private fun scheduleStop(shutdownEngine: Boolean) { + val startedAtMs = SystemClock.elapsedRealtime() + val detached = detachActiveStream() + Log.i( + DIAGNOSTIC_TAG, + "cleanup scheduled shutdownEngine=$shutdownEngine stream=${diagnosticId(detached.streamId)}", + ) val previousCleanup = cleanupJob cleanupJob = scope.launch { previousCleanup?.join() - cleanupDetachedStream(hash, stopBinary) + cleanupDetachedStream(detached, shutdownEngine) + Log.i( + DIAGNOSTIC_TAG, + "cleanup complete shutdownEngine=$shutdownEngine stream=${diagnosticId(detached.streamId)} " + + "elapsedMs=${elapsedSince(startedAtMs)}", + ) } } - private suspend fun stopStreamNow(stopBinary: Boolean) { + private suspend fun stopStreamNow(shutdownEngine: Boolean) { + val startedAtMs = SystemClock.elapsedRealtime() cleanupJob?.join() - val hash = detachActiveStream() - cleanupDetachedStream(hash, stopBinary) + val detached = detachActiveStream() + cleanupDetachedStream(detached, shutdownEngine) + Log.i( + DIAGNOSTIC_TAG, + "cleanup synchronous shutdownEngine=$shutdownEngine stream=${diagnosticId(detached.streamId)} " + + "elapsedMs=${elapsedSince(startedAtMs)}", + ) } - private fun detachActiveStream(): String? { - val detached = synchronized(lifecycleLock) { + private fun detachActiveStream(): DetachedStream { + val detached: Pair = synchronized(lifecycleLock) { streamGeneration += 1 - val hash = currentHash + val value = DetachedStream(engine = engine, streamId = currentStreamId) val job = statsJob - currentHash = null + currentTorrentId = null + currentStreamId = null statsJob = null - hash to job + _state.value = P2pStreamingState.Idle + value to job } detached.second?.cancel() - _state.value = P2pStreamingState.Idle return detached.first } - private suspend fun cleanupDetachedStream(hash: String?, stopBinary: Boolean) { - hash?.let { - try { - api.dropTorrent(it) - } catch (e: Exception) { - Log.w(TAG, "Error dropping torrent", e) + private fun detachGenerationIfCurrent( + generation: Long, + terminalState: P2pStreamingState, + ): DetachedStream? { + val detached: Pair? = synchronized(lifecycleLock) { + if (streamGeneration != generation) return@synchronized null + streamGeneration += 1 + val value = DetachedStream(engine = engine, streamId = currentStreamId) + val job = statsJob + currentTorrentId = null + currentStreamId = null + statsJob = null + _state.value = terminalState + value to job + } + detached?.second?.cancel() + return detached?.first + } + + private suspend fun cleanupDetachedStream(detached: DetachedStream, shutdownEngine: Boolean) { + detached.streamId?.let { streamId -> + stopPreparedStream(detached.engine, streamId) + } + if (shutdownEngine) { + closeEngine(detached.engine) + } + } + + private suspend fun cleanupFailedStart( + generation: Long, + activeEngine: NuvioEngine?, + preparedStream: NuvioStream?, + attached: Boolean, + terminalState: P2pStreamingState, + ) { + if (attached) { + detachGenerationIfCurrent(generation, terminalState)?.let { detached -> + cleanupDetachedStream(detached, shutdownEngine = false) } + } else { + preparedStream?.let { stream -> stopPreparedStream(activeEngine, stream.id) } + detachGenerationIfCurrent(generation, terminalState) + } + } + + private suspend fun stopPreparedStream(activeEngine: NuvioEngine?, streamId: String) { + try { + activeEngine?.stopStream(streamId) + } catch (cancellation: CancellationException) { + throw cancellation + } catch (error: Exception) { + Log.w(TAG, "Error stopping Nuvio Engine stream route", error) + } + } + + private suspend fun ensureEngine(): NuvioEngine { + val startedAtMs = SystemClock.elapsedRealtime() + P2pSettingsRepository.ensureLoaded() + val settings = P2pSettingsRepository.uiState.value + val configurationKey = EngineConfigurationKey( + uploadEnabled = settings.enableUpload, + torrentProfile = settings.torrentProfile, + diskCacheCapacityBytes = settings.cacheSize.bytes, + ) + engine?.takeIf { engineConfigurationKey == configurationKey }?.let { + Log.i( + DIAGNOSTIC_TAG, + "engine reuse configuration=$configurationKey elapsedMs=${elapsedSince(startedAtMs)}", + ) + return it } - if (stopBinary) { + Log.i( + DIAGNOSTIC_TAG, + "engine create begin configuration=$configurationKey replacing=${engine != null}", + ) + closeEngine(engine) + currentCoroutineContext().ensureActive() + val context = requireContext() + val stateDirectory = File(context.noBackupFilesDir, "nuvio-engine/state") + val cacheDirectory = File(context.cacheDir, "nuvio-engine/payload") + check(stateDirectory.mkdirs() || stateDirectory.isDirectory) { + "Could not create the Nuvio Engine state directory" + } + check(cacheDirectory.mkdirs() || cacheDirectory.isDirectory) { + "Could not create the Nuvio Engine cache directory" + } + return NuvioEngine.create( + buildNuvioEngineConfig( + stateDirectory = stateDirectory, + cacheDirectory = cacheDirectory, + uploadEnabled = configurationKey.uploadEnabled, + torrentProfile = configurationKey.torrentProfile, + diskCacheCapacityBytes = configurationKey.diskCacheCapacityBytes, + ) + ).also { created -> + engine = created + engineConfigurationKey = configurationKey + observeEngineEvents(created) + Log.i( + TAG, + "Using Nuvio Engine ${NuvioEngine.version} (${NuvioEngine.protocolBackendVersion})" + ) + Log.i( + DIAGNOSTIC_TAG, + "engine create complete configuration=$configurationKey elapsedMs=${elapsedSince(startedAtMs)} " + + "version=${NuvioEngine.version} backend=${NuvioEngine.protocolBackendVersion}", + ) + } + } + + private suspend fun closeEngine(target: NuvioEngine?) { + if (target == null) return + val startedAtMs = SystemClock.elapsedRealtime() + Log.i(DIAGNOSTIC_TAG, "engine shutdown begin active=${engine === target}") + if (engine === target) { + engineEventsJob?.cancel() + engineEventsJob = null + engine = null + engineConfigurationKey = null + knownTorrentIds.clear() + } + withContext(NonCancellable) { try { - binary.stop() - } catch (e: Exception) { - Log.w(TAG, "Error stopping TorrServer", e) + target.shutdown() + } catch (error: Exception) { + Log.w(TAG, "Error shutting down Nuvio Engine", error) + } + } + Log.i( + DIAGNOSTIC_TAG, + "engine shutdown complete elapsedMs=${elapsedSince(startedAtMs)}", + ) + } + + private fun observeEngineEvents(activeEngine: NuvioEngine) { + engineEventsJob?.cancel() + engineEventsJob = scope.launch { + activeEngine.events.collect { event -> + Log.i( + DIAGNOSTIC_TAG, + "event type=${event.type} sequence=${event.sequence} requestId=${event.requestId} " + + "dropped=${event.droppedEvents} torrent=${diagnosticId(event.torrentId)} " + + "stream=${diagnosticId(event.streamId)} fileIndex=${event.fileIndex ?: -1} " + + "fileBytes=${event.fileSize} message=${diagnosticMessage(event.message)}", + ) + if (engine !== activeEngine) return@collect + when (event.type) { + NuvioEventType.TorrentError -> { + if (event.requestId != 0L) return@collect + val terminalError = P2pStreamingState.Error( + event.message ?: localizedP2pUnknownTorrentError() + ) + synchronized(lifecycleLock) { + if (engine !== activeEngine || + (event.torrentId != null && event.torrentId != currentTorrentId) + ) { + return@synchronized + } + streamGeneration += 1 + statsJob?.cancel() + statsJob = null + _state.value = terminalError + } + } + NuvioEventType.StreamStopped -> { + if (event.requestId != 0L) return@collect + val fallbackMessage = localizedP2pUnknownTorrentError() + synchronized(lifecycleLock) { + if (engine !== activeEngine) return@synchronized + val error = unexpectedStreamStopError( + requestId = event.requestId, + eventStreamId = event.streamId, + currentStreamId = currentStreamId, + message = event.message, + fallbackMessage = fallbackMessage, + ) ?: return@synchronized + streamGeneration += 1 + currentTorrentId = null + currentStreamId = null + statsJob?.cancel() + statsJob = null + _state.value = error + } + } + else -> Unit + } } } } - private fun nextStreamGeneration(): Long = - synchronized(lifecycleLock) { - streamGeneration += 1 - streamGeneration + private fun startStatsPolling( + activeEngine: NuvioEngine, + stream: NuvioStream, + generation: Long, + requestSequence: Long, + startedAtMs: Long, + payloadDownloadBaseline: Long, + ) { + statsJob?.cancel() + statsJob = scope.launch { + var nextDiagnosticSampleAtMs = 0L + while (isActive) { + if (!isCurrentGeneration(generation)) return@launch + val currentState = _state.value + if (currentState is P2pStreamingState.Streaming) { + val route = try { + activeEngine.currentStreamStats(stream.id) + } catch (cancellation: CancellationException) { + throw cancellation + } catch (error: Exception) { + Log.w(TAG, "Error sampling Nuvio Engine stream progress", error) + null + } + val aggregate = activeEngine.stats.value + updateCacheState( + aggregate.diskCacheUsedBytes, + aggregate.diskCacheProtectedBytes, + ) + val nowMs = SystemClock.elapsedRealtime() + if (nowMs >= nextDiagnosticSampleAtMs) { + nextDiagnosticSampleAtMs = nowMs + DIAGNOSTIC_SAMPLE_INTERVAL_MS + Log.i( + DIAGNOSTIC_TAG, + "sample request=$requestSequence elapsedMs=${elapsedSince(startedAtMs)} " + + "generation=$generation stream=${diagnosticId(stream.id)} " + + "http=${aggregate.activeHttpRequests} pendingReads=${aggregate.pendingPieceReads} " + + "peers=${aggregate.connectedPeers} seeds=${aggregate.connectedSeeds} " + + "known=${aggregate.knownPeers} candidates=${aggregate.connectCandidates} " + + "interested=${aggregate.interestedPeers} unchoked=${aggregate.unchokedPeers} " + + "downloading=${aggregate.downloadingPeers} snubbed=${aggregate.snubbedPeers} " + + "connecting=${aggregate.connectingPeers} handshaking=${aggregate.handshakingPeers} " + + "targetPeers=${aggregate.targetPiecePeers} " + + "targetUnchoked=${aggregate.targetPieceUnchokedPeers} " + + "targetDownloading=${aggregate.targetPieceDownloadingPeers} " + + "offTargetDownloading=${aggregate.offTargetDownloadingPeers} " + + "pendingBlocks=${aggregate.pendingBlockRequests} " + + "targetBlocks=${aggregate.targetBlockRequests} " + + "timedOutBlocks=${aggregate.timedOutBlockRequests} " + + "downBps=${aggregate.downloadRateBytesPerSecond} upBps=${aggregate.uploadRateBytesPerSecond} " + + "payloadDown=${aggregate.totalPayloadDownloadBytes} " + + "trackerReplies=${aggregate.trackerReplyEvents} " + + "trackerErrors=${aggregate.trackerErrorEvents} " + + "trackerPeers=${aggregate.trackerPeersReturned} " + + "dhtReplies=${aggregate.dhtReplyEvents} dhtPeers=${aggregate.dhtPeersReturned} " + + "peerConnects=${aggregate.peerConnectEvents} " + + "peerDisconnects=${aggregate.peerDisconnectEvents} " + + "disconnectTimeouts=${aggregate.peerDisconnectTimeouts} " + + "disconnectConnectFailures=${aggregate.peerDisconnectConnectFailures} " + + "disconnectRedundant=${aggregate.peerDisconnectRedundant} " + + "disconnectTurnover=${aggregate.peerDisconnectTurnover} " + + "disconnectOther=${aggregate.peerDisconnectOther} " + + "finishedTransitions=${aggregate.torrentFinishedEvents} " + + "contiguous=${route?.contiguousReadyBytes ?: -1L} " + + "verified=${route?.verifiedFileBytes ?: -1L} delivered=${route?.deliveredBytes ?: -1L} " + + "demands=${route?.activeDemands ?: -1} scheduledPieces=${route?.scheduledPieces ?: -1} " + + "blockingPieces=${route?.blockingPieces ?: -1} " + + "blockingPrimary=${route?.primaryBlockingPiece ?: -1} " + + "blockingSecondary=${route?.secondaryBlockingPiece ?: -1} " + + "demandPrimary=${route?.primaryDemandStart ?: -1L}-${route?.primaryDemandEnd ?: -1L} " + + "demandSecondary=${route?.secondaryDemandStart ?: -1L}-${route?.secondaryDemandEnd ?: -1L} " + + "lastReadyPiece=${route?.lastReadyPiece ?: -1} " + + "scheduleRevision=${route?.scheduleRevision ?: -1L} " + + "fileBytes=${route?.fileSize ?: stream.fileSize} activeTorrents=${aggregate.activeTorrents} " + + "activeStreams=${aggregate.activeStreams} warm=${aggregate.warmTorrents} " + + "quiesced=${aggregate.quiescedTorrents} memUsed=${aggregate.memoryCacheUsedBytes} " + + "memHits=${aggregate.memoryCacheHits} memMisses=${aggregate.memoryCacheMisses} " + + "diskUsed=${aggregate.diskCacheUsedBytes} diskProtected=${aggregate.diskCacheProtectedBytes} " + + "diskOverBudget=${aggregate.diskCacheOverBudget}", + ) + } + updateStreamingIfCurrent(generation) { latestState -> + latestState.copy( + downloadSpeed = aggregate.downloadRateBytesPerSecond, + uploadSpeed = aggregate.uploadRateBytesPerSecond, + peers = aggregate.connectedPeers, + seeds = aggregate.connectedSeeds, + bufferProgress = route?.bufferProgress ?: latestState.bufferProgress, + totalProgress = route?.fileProgress ?: latestState.totalProgress, + downloadedBytes = (aggregate.totalPayloadDownloadBytes - + payloadDownloadBaseline).coerceAtLeast(0L), + verifiedBytes = route?.verifiedFileBytes ?: latestState.verifiedBytes, + deliveredBytes = route?.deliveredBytes ?: latestState.deliveredBytes, + ) + } + } + delay(250L) + } } + } - private fun attachTorrentIfCurrent(generation: Long, hash: String): Boolean = - synchronized(lifecycleLock) { - if (streamGeneration != generation) return@synchronized false - currentHash = hash - true + private fun updateCacheState(usedBytes: Long, protectedBytes: Long) { + _cacheState.value = _cacheState.value.copy( + usedBytes = usedBytes, + protectedBytes = protectedBytes, + hasMeasurement = true, + ) + } + + private fun startStartupStatsPolling( + activeEngine: NuvioEngine, + generation: Long, + phase: AtomicReference, + requestSequence: Long, + startedAtMs: Long, + ): Job = scope.launch { + while (isActive) { + val aggregate = activeEngine.stats.value + updateConnectingIfCurrent( + generation = generation, + phase = phase.get(), + downloadSpeed = aggregate.downloadRateBytesPerSecond, + uploadSpeed = aggregate.uploadRateBytesPerSecond, + peers = aggregate.connectedPeers, + seeds = aggregate.connectedSeeds, + ) + Log.i( + DIAGNOSTIC_TAG, + "startupSample request=$requestSequence phase=${phase.get()} elapsedMs=${elapsedSince(startedAtMs)} " + + "http=${aggregate.activeHttpRequests} pendingReads=${aggregate.pendingPieceReads} " + + "peers=${aggregate.connectedPeers} seeds=${aggregate.connectedSeeds} " + + "known=${aggregate.knownPeers} candidates=${aggregate.connectCandidates} " + + "interested=${aggregate.interestedPeers} unchoked=${aggregate.unchokedPeers} " + + "downloading=${aggregate.downloadingPeers} snubbed=${aggregate.snubbedPeers} " + + "connecting=${aggregate.connectingPeers} handshaking=${aggregate.handshakingPeers} " + + "pendingBlocks=${aggregate.pendingBlockRequests} " + + "targetBlocks=${aggregate.targetBlockRequests} " + + "timedOutBlocks=${aggregate.timedOutBlockRequests} " + + "downBps=${aggregate.downloadRateBytesPerSecond} upBps=${aggregate.uploadRateBytesPerSecond} " + + "payloadDown=${aggregate.totalPayloadDownloadBytes} activeTorrents=${aggregate.activeTorrents} " + + "trackerReplies=${aggregate.trackerReplyEvents} trackerErrors=${aggregate.trackerErrorEvents} " + + "trackerPeers=${aggregate.trackerPeersReturned} dhtReplies=${aggregate.dhtReplyEvents} " + + "dhtPeers=${aggregate.dhtPeersReturned} peerConnects=${aggregate.peerConnectEvents} " + + "peerDisconnects=${aggregate.peerDisconnectEvents} " + + "activeStreams=${aggregate.activeStreams} warm=${aggregate.warmTorrents} " + + "quiesced=${aggregate.quiescedTorrents}", + ) + delay(DIAGNOSTIC_SAMPLE_INTERVAL_MS) } + } + + private fun beginStreamGeneration(): Long = synchronized(lifecycleLock) { + streamGeneration += 1 + _state.value = P2pStreamingState.Connecting() + streamGeneration + } + + private fun updateConnectingIfCurrent( + generation: Long, + phase: String, + downloadSpeed: Long, + uploadSpeed: Long, + peers: Int, + seeds: Int, + ) = synchronized(lifecycleLock) { + if (streamGeneration != generation || _state.value !is P2pStreamingState.Connecting) { + return@synchronized + } + _state.value = P2pStreamingState.Connecting( + phase = phase, + downloadSpeed = downloadSpeed, + uploadSpeed = uploadSpeed, + peers = peers, + seeds = seeds, + ) + } + + private fun nextDiagnosticRequestSequence(): Long = synchronized(lifecycleLock) { + diagnosticRequestSequence += 1 + diagnosticRequestSequence + } + + private fun attachStreamIfCurrent( + generation: Long, + torrentId: String, + streamId: String, + ): Boolean = synchronized(lifecycleLock) { + if (streamGeneration != generation) return@synchronized false + currentTorrentId = torrentId + currentStreamId = streamId + true + } + + private fun publishStreamingIfCurrent( + generation: Long, + state: P2pStreamingState.Streaming, + ): Boolean = synchronized(lifecycleLock) { + if (streamGeneration != generation || currentStreamId == null) { + return@synchronized false + } + _state.value = state + true + } + + private fun updateStreamingIfCurrent( + generation: Long, + update: (P2pStreamingState.Streaming) -> P2pStreamingState.Streaming, + ) = synchronized(lifecycleLock) { + if (streamGeneration != generation) return@synchronized + val current = _state.value as? P2pStreamingState.Streaming ?: return@synchronized + _state.value = update(current) + } private fun isCurrentGeneration(generation: Long): Boolean = synchronized(lifecycleLock) { streamGeneration == generation } @@ -177,381 +780,50 @@ actual object P2pStreamingEngine { } } - private fun buildMagnetUri(infoHash: String, extraTrackers: List): String { - val trackers = (DEFAULT_TRACKERS + extraTrackers).distinct() - val trackerParams = trackers.joinToString("") { "&tr=$it" } - return "magnet:?xt=urn:btih:$infoHash$trackerParams" - } - - private suspend fun resolveFileIndex(hash: String, requestedIdx: Int?, filename: String?): Int { - val deadline = System.currentTimeMillis() + 15_000L - var files: List = emptyList() - - while (System.currentTimeMillis() < deadline) { - files = api.getTorrentStats(hash)?.files ?: emptyList() - if (files.isNotEmpty()) break - Log.d(TAG, "Waiting for torrent metadata...") - delay(1_000L) - } - - if (files.isEmpty()) { - Log.w(TAG, "No files after metadata timeout, guessing index ${requestedIdx?.plus(1) ?: 1}") - return requestedIdx?.plus(1) ?: 1 - } - - if (!filename.isNullOrBlank()) { - val name = filename.trim() - val exact = files.firstOrNull { file -> - file.path.substringAfterLast('/').equals(name, ignoreCase = true) - } - if (exact != null) { - Log.d(TAG, "File resolved by exact filename match: ${exact.path} -> id=${exact.id}") - return exact.id - } - - val contains = files.firstOrNull { file -> - file.path.contains(name, ignoreCase = true) - } - if (contains != null) { - Log.d(TAG, "File resolved by filename contains match: ${contains.path} -> id=${contains.id}") - return contains.id - } - } - - if (requestedIdx != null) { - val torrServerIndex = requestedIdx + 1 - if (files.any { it.id == torrServerIndex }) { - Log.d(TAG, "File resolved by ID offset: id=$torrServerIndex") - return torrServerIndex - } - } - - if (requestedIdx != null && requestedIdx in files.indices) { - val positionalFile = files[requestedIdx] - Log.d(TAG, "File resolved by positional index: [$requestedIdx] -> ${positionalFile.path} (id=${positionalFile.id})") - return positionalFile.id - } - - val videoFile = files - .filter { file -> - val ext = file.path.substringAfterLast('.', "").lowercase() - ext in VIDEO_EXTENSIONS - } - .maxByOrNull { it.length } - - val result = videoFile?.id ?: files.maxByOrNull { it.length }?.id ?: 1 - Log.d(TAG, "File resolved by largest video fallback: id=$result") - return result - } - - private fun startStatsPolling(hash: String, generation: Long) { - statsJob?.cancel() - statsJob = scope.launch { - while (isActive) { - if (!isCurrentGeneration(generation)) return@launch - try { - val stats = api.getTorrentStats(hash) - val currentState = _state.value - if ( - stats != null && - currentState is P2pStreamingState.Streaming && - isCurrentGeneration(generation) - ) { - _state.value = currentState.copy( - downloadSpeed = stats.downloadSpeed, - uploadSpeed = stats.uploadSpeed, - peers = stats.peers, - seeds = stats.seeds, - preloadedBytes = stats.preloadedBytes, - ) - } - } catch (e: CancellationException) { - throw e - } catch (e: Exception) { - Log.w(TAG, "Stats polling error", e) - } - delay(1_000L) - } - } - } - private fun requireContext(): Context = appContext ?: throw P2pStreamingException("P2P streaming engine is not initialized") + private fun logPhase(requestSequence: Long, startedAtMs: Long, phase: String) { + Log.i( + DIAGNOSTIC_TAG, + "start request=$requestSequence phase=$phase complete elapsedMs=${elapsedSince(startedAtMs)}", + ) + } + + private fun elapsedSince(startedAtMs: Long): Long = + (SystemClock.elapsedRealtime() - startedAtMs).coerceAtLeast(0L) + + private fun diagnosticId(value: String?): String = + value?.trim()?.take(12)?.ifBlank { "none" } ?: "none" + + private fun diagnosticLoopbackUrl(value: String): String = runCatching { + val uri = android.net.Uri.parse(value) + "${uri.scheme}://${uri.host}:${uri.port}/…" + }.getOrDefault("unparseable") + + private fun diagnosticMessage(value: String?): String = + value?.replace('\n', ' ')?.replace('\r', ' ')?.take(160) ?: "none" + private val DEFAULT_TRACKERS = listOf( + "udp://zer0day.ch:1337/announce", + "udp://tracker.publictracker.xyz:6969/announce", "udp://tracker.opentrackr.org:1337/announce", + "udp://open.demonii.com:1337/announce", "udp://open.stealth.si:80/announce", - "udp://tracker.openbittorrent.com:6969/announce", - "udp://exodus.desync.com:6969/announce", + "http://tracker.renfei.net:8080/announce", + "udp://udp.tracker.projectk.org:23333/announce", + "udp://tracker.tryhackx.org:6969/announce", "udp://tracker.torrent.eu.org:451/announce", + "udp://tracker.theoks.net:6969/announce", + "udp://tracker.startwork.cv:1337/announce", + "udp://tracker.qu.ax:6969/announce", + "udp://tracker.plx.im:6969/announce", + "udp://tracker.nyaa.vc:6969/announce", + "udp://tracker.iperson.xyz:6969/announce", + "udp://tracker.gmi.gd:6969/announce", + "udp://tracker.fnix.net:6969/announce", + "udp://tracker.flatuslifir.is:6969/announce", + "udp://tracker.ducks.party:1984/announce", + "udp://tracker.bluefrog.pw:2710/announce", ) - - private class TorrServerBinary { - private var context: Context? = null - private var process: Process? = null - private val healthClient = OkHttpClient.Builder() - .connectTimeout(2, TimeUnit.SECONDS) - .readTimeout(5, TimeUnit.SECONDS) - .build() - - val baseUrl: String get() = "http://127.0.0.1:$PORT" - - fun initialize(context: Context) { - this.context = context.applicationContext - } - - suspend fun start() = withContext(Dispatchers.IO) { - if (isRunning()) { - Log.d(TAG, "TorrServer already running") - return@withContext - } - - killOrphanedProcess() - - val ctx = requireContext() - val binaryFile = File(ctx.applicationInfo.nativeLibraryDir, "libtorrserver.so") - if (!binaryFile.exists()) { - throw P2pStreamingException("TorrServer binary not found at ${binaryFile.absolutePath}") - } - - if (!binaryFile.canExecute()) { - binaryFile.setExecutable(true) - } - - val configDir = File(ctx.filesDir, "torrserver").also { it.mkdirs() } - val processBuilder = ProcessBuilder( - binaryFile.absolutePath, - "--port", - PORT.toString(), - "--path", - configDir.absolutePath, - ) - processBuilder.directory(configDir) - processBuilder.redirectErrorStream(true) - - Log.d(TAG, "Starting TorrServer on port $PORT from ${binaryFile.absolutePath}") - process = processBuilder.start() - - val proc = process!! - Thread { - try { - proc.inputStream.bufferedReader().forEachLine { line -> - Log.d(TAG, "[server] $line") - } - } catch (_: Exception) { - } - }.apply { - isDaemon = true - start() - } - - val deadline = System.currentTimeMillis() + STARTUP_TIMEOUT_MS - while (System.currentTimeMillis() < deadline) { - if (isRunning()) { - Log.d(TAG, "TorrServer started successfully") - return@withContext - } - if (!isProcessAlive(process)) { - val exitCode = process?.exitValue() ?: -1 - process = null - throw P2pStreamingException("TorrServer process died on startup (exit code $exitCode)") - } - delay(HEALTH_CHECK_INTERVAL_MS) - } - - stop() - throw P2pStreamingException("TorrServer failed to start within ${STARTUP_TIMEOUT_MS / 1000}s") - } - - fun isRunning(): Boolean { - return try { - val request = Request.Builder().url("$baseUrl/echo").build() - healthClient.newCall(request).execute().use { it.isSuccessful } - } catch (e: Exception) { - false - } - } - - fun stop() { - try { - val request = Request.Builder().url("$baseUrl/shutdown").build() - healthClient.newCall(request).execute().close() - } catch (_: Exception) { - } - - process?.let { proc -> - try { - Thread.sleep(3_000L) - if (isProcessAlive(proc)) { - proc.destroyForcibly() - } - } catch (_: Exception) { - proc.destroyForcibly() - } - } - process = null - Log.d(TAG, "TorrServer stopped") - } - - private fun killOrphanedProcess() { - try { - val request = Request.Builder().url("$baseUrl/shutdown").build() - healthClient.newCall(request).execute().close() - Thread.sleep(1_000L) - Log.d(TAG, "Shut down orphaned TorrServer instance") - } catch (_: Exception) { - } - } - - private fun isProcessAlive(proc: Process?): Boolean { - if (proc == null) return false - return try { - proc.exitValue() - false - } catch (_: IllegalThreadStateException) { - true - } catch (_: Exception) { - false - } - } - - private fun requireContext(): Context = - context ?: throw P2pStreamingException("P2P streaming engine is not initialized") - - companion object { - const val PORT = 8091 - private const val STARTUP_TIMEOUT_MS = 15_000L - private const val HEALTH_CHECK_INTERVAL_MS = 200L - } - } - - private data class TorrServerFile( - val id: Int, - val path: String, - val length: Long, - ) - - private data class TorrServerStats( - val downloadSpeed: Long, - val uploadSpeed: Long, - val peers: Int, - val seeds: Int, - val preloadedBytes: Long, - val loadedSize: Long, - val torrentSize: Long, - val files: List, - ) - - private class TorrServerApi( - private val binary: TorrServerBinary, - ) { - private val client = OkHttpClient.Builder() - .connectTimeout(10, TimeUnit.SECONDS) - .readTimeout(30, TimeUnit.SECONDS) - .build() - - private val baseUrl: String get() = binary.baseUrl - - suspend fun addTorrent(magnetLink: String, title: String? = null): String? = withContext(Dispatchers.IO) { - val body = JSONObject().apply { - put("action", "add") - put("link", magnetLink) - put("save_to_db", false) - if (title != null) put("title", title) - } - - val request = Request.Builder() - .url("$baseUrl/torrents") - .post(body.toString().toRequestBody(JSON_TYPE)) - .build() - - try { - client.newCall(request).execute().use { response -> - if (!response.isSuccessful) { - Log.e(TAG, "addTorrent failed: ${response.code}") - return@withContext null - } - val json = JSONObject(response.body?.string() ?: "{}") - val hash = json.optString("hash", "") - Log.d(TAG, "Torrent added: $hash") - hash.ifEmpty { null } - } - } catch (e: Exception) { - Log.e(TAG, "addTorrent error", e) - null - } - } - - suspend fun getTorrentStats(hash: String): TorrServerStats? = withContext(Dispatchers.IO) { - val body = JSONObject().apply { - put("action", "get") - put("hash", hash) - } - - val request = Request.Builder() - .url("$baseUrl/torrents") - .post(body.toString().toRequestBody(JSON_TYPE)) - .build() - - try { - client.newCall(request).execute().use { response -> - if (!response.isSuccessful) return@withContext null - val json = JSONObject(response.body?.string() ?: "{}") - - val files = mutableListOf() - val fileList = json.optJSONArray("file_stats") ?: JSONArray() - for (i in 0 until fileList.length()) { - val file = fileList.getJSONObject(i) - files.add( - TorrServerFile( - id = file.optInt("id", i + 1), - path = file.optString("path", ""), - length = file.optLong("length", 0), - ), - ) - } - - TorrServerStats( - downloadSpeed = json.optLong("download_speed", 0), - uploadSpeed = json.optLong("upload_speed", 0), - peers = json.optInt("active_peers", 0), - seeds = json.optInt("connected_seeders", 0), - preloadedBytes = json.optLong("preloaded_bytes", 0), - loadedSize = json.optLong("loaded_size", 0), - torrentSize = json.optLong("torrent_size", 0), - files = files, - ) - } - } catch (e: Exception) { - Log.w(TAG, "getTorrentStats error", e) - null - } - } - - suspend fun dropTorrent(hash: String) = withContext(Dispatchers.IO) { - val body = JSONObject().apply { - put("action", "drop") - put("hash", hash) - } - - val request = Request.Builder() - .url("$baseUrl/torrents") - .post(body.toString().toRequestBody(JSON_TYPE)) - .build() - - try { - client.newCall(request).execute().close() - Log.d(TAG, "Torrent dropped: $hash") - } catch (e: Exception) { - Log.w(TAG, "dropTorrent error", e) - } - } - - fun getStreamUrl(magnetLink: String, fileIdx: Int): String { - val encodedLink = URLEncoder.encode(magnetLink, "UTF-8") - return "$baseUrl/stream?link=$encodedLink&index=$fileIdx&play" - } - } - - private val JSON_TYPE = "application/json".toMediaType() } diff --git a/composeApp/src/androidMain/kotlin/com/nuvio/app/features/player/PlayerEngine.android.kt b/composeApp/src/androidMain/kotlin/com/nuvio/app/features/player/PlayerEngine.android.kt index 85fe5fdd..5303fe9e 100644 --- a/composeApp/src/androidMain/kotlin/com/nuvio/app/features/player/PlayerEngine.android.kt +++ b/composeApp/src/androidMain/kotlin/com/nuvio/app/features/player/PlayerEngine.android.kt @@ -9,6 +9,7 @@ import android.util.Log import android.util.TypedValue import android.graphics.Typeface import android.os.Build +import android.os.SystemClock import android.view.ViewGroup.LayoutParams.MATCH_PARENT import android.util.AttributeSet import androidx.compose.runtime.getValue @@ -79,6 +80,12 @@ import java.net.HttpURLConnection import java.net.URL private const val TAG = "NuvioPlayer" +private const val PLAYER_DIAGNOSTIC_TAG = "NuvioPlayerDiag" + +private class PlaybackDiagnostics { + var prepareStartedAtMs: Long = 0L + var attempt: Int = 0 +} @androidx.annotation.OptIn(UnstableApi::class) @Composable @@ -92,8 +99,11 @@ actual fun PlatformPlayerSurface( useYoutubeChunkedPlayback: Boolean, modifier: Modifier, playWhenReady: Boolean, + initialPositionMs: Long?, + initialPositionRequestKey: String?, resizeMode: PlayerResizeMode, useNativeController: Boolean, + onInitialPositionHandled: (key: String, handled: Boolean) -> Unit, onControllerReady: (PlayerEngineController) -> Unit, onSnapshot: (PlayerPlaybackSnapshot) -> Unit, onError: (String?) -> Unit, @@ -109,6 +119,7 @@ actual fun PlatformPlayerSurface( sanitizePlaybackResponseHeaders(sourceResponseHeaders), normalizeStreamType(streamType).orEmpty(), useYoutubeChunkedPlayback, + initialPositionRequestKey.orEmpty(), ) var activeEngine by remember(playerSourceKey, playerSettings.androidPlaybackEngine) { mutableStateOf(playerSettings.androidPlaybackEngine.initialAndroidEngine()) @@ -125,13 +136,19 @@ actual fun PlatformPlayerSurface( useYoutubeChunkedPlayback = useYoutubeChunkedPlayback, modifier = modifier, playWhenReady = playWhenReady, + initialPositionMs = initialPositionMs, + initialPositionRequestKey = initialPositionRequestKey, resizeMode = resizeMode, useNativeController = useNativeController, + onInitialPositionHandled = onInitialPositionHandled, onControllerReady = onControllerReady, onSnapshot = onSnapshot, onError = { message -> if (message != null && playerSettings.androidPlaybackEngine == AndroidPlaybackEngine.Auto) { Log.w(TAG, "ExoPlayer failed; falling back to libmpv: $message") + initialPositionRequestKey?.let { key -> + onInitialPositionHandled(key, false) + } activeEngine = ResolvedAndroidPlaybackEngine.Libmpv onError(null) } else { @@ -139,21 +156,28 @@ actual fun PlatformPlayerSurface( } }, ) - ResolvedAndroidPlaybackEngine.Libmpv -> LibmpvPlayerSurface( - sourceUrl = sourceUrl, - sourceAudioUrl = sourceAudioUrl, - sourceHeaders = sourceHeaders, - externalSubtitles = externalSubtitles, - modifier = modifier, - playWhenReady = playWhenReady, - resizeMode = resizeMode, - videoOutput = playerSettings.androidLibmpvVideoOutput, - hardwareDecodingEnabled = playerSettings.androidLibmpvHardwareDecodingEnabled, - yuv420pEnabled = playerSettings.androidLibmpvYuv420pEnabled, - onControllerReady = onControllerReady, - onSnapshot = onSnapshot, - onError = onError, - ) + ResolvedAndroidPlaybackEngine.Libmpv -> { + LaunchedEffect(initialPositionRequestKey) { + initialPositionRequestKey?.let { key -> + onInitialPositionHandled(key, false) + } + } + LibmpvPlayerSurface( + sourceUrl = sourceUrl, + sourceAudioUrl = sourceAudioUrl, + sourceHeaders = sourceHeaders, + externalSubtitles = externalSubtitles, + modifier = modifier, + playWhenReady = playWhenReady, + resizeMode = resizeMode, + videoOutput = playerSettings.androidLibmpvVideoOutput, + hardwareDecodingEnabled = playerSettings.androidLibmpvHardwareDecodingEnabled, + yuv420pEnabled = playerSettings.androidLibmpvYuv420pEnabled, + onControllerReady = onControllerReady, + onSnapshot = onSnapshot, + onError = onError, + ) + } } } @@ -181,8 +205,11 @@ private fun ExoPlayerSurface( useYoutubeChunkedPlayback: Boolean, modifier: Modifier, playWhenReady: Boolean, + initialPositionMs: Long?, + initialPositionRequestKey: String?, resizeMode: PlayerResizeMode, useNativeController: Boolean, + onInitialPositionHandled: (key: String, handled: Boolean) -> Unit, onControllerReady: (PlayerEngineController) -> Unit, onSnapshot: (PlayerPlaybackSnapshot) -> Unit, onError: (String?) -> Unit, @@ -191,6 +218,7 @@ private fun ExoPlayerSurface( val lifecycleOwner = LocalLifecycleOwner.current val latestOnSnapshot = rememberUpdatedState(onSnapshot) val latestOnError = rememberUpdatedState(onError) + val latestOnInitialPositionHandled = rememberUpdatedState(onInitialPositionHandled) val latestPlayWhenReady = rememberUpdatedState(playWhenReady) val coroutineScope = rememberCoroutineScope() @@ -219,7 +247,9 @@ private fun ExoPlayerSurface( sanitizedSourceResponseHeaders, normalizedStreamType.orEmpty(), useYoutubeChunkedPlayback, + initialPositionRequestKey.orEmpty(), ) + val playbackDiagnostics = remember(playerSourceKey) { PlaybackDiagnostics() } var subtitleDelayMs by remember(playerSourceKey) { mutableStateOf(0) } var selectedExternalSubtitleMimeType by remember(playerSourceKey) { mutableStateOf(null) } val latestSubtitleDelayMs = rememberUpdatedState(subtitleDelayMs) @@ -262,6 +292,7 @@ private fun ExoPlayerSurface( } val dataSourceFactory = remember( context, + sourceUrl, sanitizedSourceHeaders, sanitizedSourceResponseHeaders, useYoutubeChunkedPlayback, @@ -272,6 +303,7 @@ private fun ExoPlayerSurface( defaultRequestHeaders = sanitizedSourceHeaders, defaultResponseHeaders = sanitizedSourceResponseHeaders, useYoutubeChunkedPlayback = useYoutubeChunkedPlayback, + useLongReadTimeout = isLoopbackPlaybackSource(sourceUrl), externalSubtitles = externalSubtitles, ) } @@ -302,6 +334,7 @@ private fun ExoPlayerSurface( normalizedStreamType, useYoutubeChunkedPlayback, effectiveDecoderPriority, + initialPositionRequestKey, ) { val renderersFactory = SubtitleOffsetRenderersFactory( context = context, @@ -389,9 +422,28 @@ private fun ExoPlayerSurface( onDispose { nowPlayingController.release() } } - LaunchedEffect(exoPlayer, resolvedMediaItem) { + LaunchedEffect(exoPlayer, resolvedMediaItem, initialPositionRequestKey) { val mediaItem = resolvedMediaItem ?: return@LaunchedEffect - exoPlayer.setPlaybackMediaItem(mediaItem, fallbackStartPositionMs) + val requestedStartPositionMs = fallbackStartPositionMs + ?: initialPositionMs?.takeIf { it > 0L } + playbackDiagnostics.attempt += 1 + playbackDiagnostics.prepareStartedAtMs = SystemClock.elapsedRealtime() + Log.i( + PLAYER_DIAGNOSTIC_TAG, + "prepare begin attempt=${playbackDiagnostics.attempt} " + + "source=${diagnosticPlaybackSource(sourceUrl)} audioSource=${!sourceAudioUrl.isNullOrBlank()} " + + "mime=${mediaItem.localConfiguration?.mimeType ?: "auto"} " + + "startPositionMs=${requestedStartPositionMs ?: 0L}", + ) + exoPlayer.setPlaybackMediaItem(mediaItem, requestedStartPositionMs) + if (fallbackStartPositionMs == null) { + initialPositionRequestKey?.let { key -> + latestOnInitialPositionHandled.value( + key, + requestedStartPositionMs != null, + ) + } + } exoPlayer.prepare() } @@ -451,6 +503,14 @@ private fun ExoPlayerSurface( val listener = object : Player.Listener { override fun onPlayerError(error: PlaybackException) { syncPlayerViewKeepScreenOn() + Log.e( + PLAYER_DIAGNOSTIC_TAG, + "error attempt=${playbackDiagnostics.attempt} " + + "elapsedMs=${diagnosticElapsedSince(playbackDiagnostics.prepareStartedAtMs)} " + + "code=${error.errorCodeName} cause=${error.cause?.javaClass?.simpleName ?: "none"} " + + "message=${diagnosticPlayerMessage(error.message)}", + error, + ) val isSourceError = error.errorCode == PlaybackException.ERROR_CODE_BEHIND_LIVE_WINDOW || error.errorCode == PlaybackException.ERROR_CODE_IO_UNSPECIFIED || @@ -503,6 +563,14 @@ private fun ExoPlayerSurface( else -> "UNKNOWN($playbackState)" } Log.d(TAG, "onPlaybackStateChanged: $stateName") + Log.i( + PLAYER_DIAGNOSTIC_TAG, + "state=$stateName attempt=${playbackDiagnostics.attempt} " + + "elapsedMs=${diagnosticElapsedSince(playbackDiagnostics.prepareStartedAtMs)} " + + "positionMs=${exoPlayer.currentPosition.coerceAtLeast(0L)} " + + "bufferedMs=${exoPlayer.bufferedPosition.coerceAtLeast(0L)} " + + "bufferedPercent=${exoPlayer.bufferedPercentage} playWhenReady=${exoPlayer.playWhenReady}", + ) if (playbackState == Player.STATE_READY) { fallbackStartPositionMs = null latestOnError.value(null) @@ -517,10 +585,25 @@ private fun ExoPlayerSurface( } override fun onIsPlayingChanged(isPlaying: Boolean) { + Log.i( + PLAYER_DIAGNOSTIC_TAG, + "isPlaying=$isPlaying attempt=${playbackDiagnostics.attempt} " + + "elapsedMs=${diagnosticElapsedSince(playbackDiagnostics.prepareStartedAtMs)} " + + "positionMs=${exoPlayer.currentPosition.coerceAtLeast(0L)}", + ) syncPlayerViewKeepScreenOn() dispatchExoPlayerSnapshot() } + override fun onRenderedFirstFrame() { + Log.i( + PLAYER_DIAGNOSTIC_TAG, + "firstFrame attempt=${playbackDiagnostics.attempt} " + + "elapsedMs=${diagnosticElapsedSince(playbackDiagnostics.prepareStartedAtMs)} " + + "positionMs=${exoPlayer.currentPosition.coerceAtLeast(0L)}", + ) + } + override fun onPlaybackParametersChanged(playbackParameters: androidx.media3.common.PlaybackParameters) { dispatchExoPlayerSnapshot() } @@ -1895,6 +1978,26 @@ private fun guessSubtitleMime(url: String): String { } } +private fun diagnosticElapsedSince(startedAtMs: Long): Long = + if (startedAtMs <= 0L) -1L else (SystemClock.elapsedRealtime() - startedAtMs).coerceAtLeast(0L) + +private fun diagnosticPlaybackSource(value: String): String = runCatching { + val uri = Uri.parse(value) + val host = uri.host.orEmpty() + val isLoopback = host == "127.0.0.1" || host == "localhost" || host == "::1" + "scheme=${uri.scheme ?: "none"},host=${host.ifBlank { "none" }},port=${uri.port},loopback=$isLoopback" +}.getOrDefault("unparseable") + +private fun isLoopbackPlaybackSource(value: String): Boolean = runCatching { + when (Uri.parse(value).host.orEmpty().lowercase()) { + "127.0.0.1", "localhost", "::1" -> true + else -> false + } +}.getOrDefault(false) + +private fun diagnosticPlayerMessage(value: String?): String = + value?.replace('\n', ' ')?.replace('\r', ' ')?.take(160) ?: "none" + internal class SubtitleRequestHeaderDataSourceFactory( private val upstreamFactory: DataSource.Factory, private val externalSubtitles: List, diff --git a/composeApp/src/androidMain/kotlin/com/nuvio/app/features/player/PlayerPlaybackNetworking.kt b/composeApp/src/androidMain/kotlin/com/nuvio/app/features/player/PlayerPlaybackNetworking.kt index 075ce0e4..3d3e2f19 100644 --- a/composeApp/src/androidMain/kotlin/com/nuvio/app/features/player/PlayerPlaybackNetworking.kt +++ b/composeApp/src/androidMain/kotlin/com/nuvio/app/features/player/PlayerPlaybackNetworking.kt @@ -60,10 +60,28 @@ internal object PlayerPlaybackNetworking { .build() } - fun createHttpDataSourceFactory(defaultHeaders: Map = emptyMap()): DataSource.Factory { + private val loopbackPlaybackHttpClient: OkHttpClient by lazy { + playbackHttpClient.newBuilder() + .addInterceptor { chain -> + val request = chain.request() + val requestChain = if (isLoopbackHost(request.url.host)) { + chain.withReadTimeout(65, TimeUnit.SECONDS) + } else { + chain + } + requestChain.proceed(request) + } + .build() + } + + fun createHttpDataSourceFactory( + defaultHeaders: Map = emptyMap(), + useLongReadTimeout: Boolean = false, + ): DataSource.Factory { val requestHeaders = sanitizeHeaders(defaultHeaders) + val baseClient = if (useLongReadTimeout) loopbackPlaybackHttpClient else playbackHttpClient val client = requestHeaders.headerValue("Authorization")?.let { authorization -> - playbackHttpClient.newBuilder() + baseClient.newBuilder() .addNetworkInterceptor { chain -> val request = chain.request() if (request.header("Authorization") == null) { @@ -77,7 +95,7 @@ internal object PlayerPlaybackNetworking { } } .build() - } ?: playbackHttpClient + } ?: baseClient return OkHttpDataSource.Factory(client).apply { setDefaultRequestProperties(requestHeaders) @@ -90,8 +108,12 @@ internal object PlayerPlaybackNetworking { fun createDataSourceFactory( context: Context, defaultHeaders: Map = emptyMap(), + useLongReadTimeout: Boolean = false, ): DataSource.Factory { - return DefaultDataSource.Factory(context, createHttpDataSourceFactory(defaultHeaders)) + return DefaultDataSource.Factory( + context, + createHttpDataSourceFactory(defaultHeaders, useLongReadTimeout), + ) } fun openConnection( @@ -133,6 +155,11 @@ internal object PlayerPlaybackNetworking { } }.toMap() + private fun isLoopbackHost(host: String): Boolean = when (host.lowercase()) { + "127.0.0.1", "localhost", "::1" -> true + else -> false + } + private fun withDefaultUserAgent(headers: Map): Map { val sanitized = sanitizeHeaders(headers) if (sanitized.headerValue("User-Agent") != null) return sanitized diff --git a/composeApp/src/androidPlaystore/kotlin/com/nuvio/app/features/player/PlatformPlaybackDataSourceFactory.android.kt b/composeApp/src/androidPlaystore/kotlin/com/nuvio/app/features/player/PlatformPlaybackDataSourceFactory.android.kt index f2d30fa7..73f15959 100644 --- a/composeApp/src/androidPlaystore/kotlin/com/nuvio/app/features/player/PlatformPlaybackDataSourceFactory.android.kt +++ b/composeApp/src/androidPlaystore/kotlin/com/nuvio/app/features/player/PlatformPlaybackDataSourceFactory.android.kt @@ -10,9 +10,13 @@ internal object PlatformPlaybackDataSourceFactory { defaultRequestHeaders: Map, defaultResponseHeaders: Map, useYoutubeChunkedPlayback: Boolean, + useLongReadTimeout: Boolean = false, externalSubtitles: List = emptyList(), ): DataSource.Factory { - val httpFactory = PlayerPlaybackNetworking.createHttpDataSourceFactory(defaultRequestHeaders) + val httpFactory = PlayerPlaybackNetworking.createHttpDataSourceFactory( + defaultRequestHeaders, + useLongReadTimeout, + ) val subtitleHeaderFactory = SubtitleRequestHeaderDataSourceFactory( upstreamFactory = httpFactory, externalSubtitles = externalSubtitles @@ -27,4 +31,4 @@ internal object PlatformPlaybackDataSourceFactory { ) } } -} \ No newline at end of file +} diff --git a/composeApp/src/commonMain/composeResources/values/strings.xml b/composeApp/src/commonMain/composeResources/values/strings.xml index bd607aab..1e20233a 100644 --- a/composeApp/src/commonMain/composeResources/values/strings.xml +++ b/composeApp/src/commonMain/composeResources/values/strings.xml @@ -2011,12 +2011,35 @@ Allow peer-to-peer (torrent) streams Hide torrent stats Hide buffer, seeds, peers and download speed during loading and playback + Torrent profile + Soft + Balanced + Fast + Uses fewer peer connections to reduce battery and network load + Recommended balance of startup speed and connection stability + Uses more peer connections for difficult or high-bitrate torrents + Torrent cache size + No persistent cache + 2 GB + 5 GB + 10 GB + Clear torrent cache + %1$s currently used + Usage is measured after P2P starts; tap to clear cached torrent data + Clearing inactive torrent data… + Stop P2P playback before clearing the cache + Cleared %1$s of inactive torrent data + Could not clear the torrent cache %1$d seeds · %2$d peers %1$s buffered · %2$s · %3$s %1$s · %2$s %1$d peers · %2$d seeds · %3$d%% Connecting to peers… Starting P2P engine… + Fetching torrent metadata… + Preparing torrent stream… + %1$s · %2$s · %3$s + %1$s downloaded · %2$s · %3$s Failed to start torrent: %1$s Torrent error: %1$s diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/p2p/P2pStreaming.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/p2p/P2pStreaming.kt index 092b855f..198ba645 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/p2p/P2pStreaming.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/p2p/P2pStreaming.kt @@ -8,7 +8,35 @@ import kotlinx.coroutines.flow.asStateFlow data class P2pSettingsUiState( val p2pEnabled: Boolean = false, val enableUpload: Boolean = true, - val hideTorrentStats: Boolean = true, + val hideTorrentStats: Boolean = false, + val torrentProfile: P2pTorrentProfile = P2pTorrentProfile.BALANCED, + val cacheSize: P2pCacheSize = P2pCacheSize.GB_2, +) + +enum class P2pTorrentProfile { + SOFT, + BALANCED, + FAST, +} + +enum class P2pCacheSize(val bytes: Long) { + NONE(0L), + GB_2(2L * 1024L * 1024L * 1024L), + GB_5(5L * 1024L * 1024L * 1024L), + GB_10(10L * 1024L * 1024L * 1024L), +} + +data class P2pCacheUiState( + val usedBytes: Long = 0L, + val protectedBytes: Long = 0L, + val isClearing: Boolean = false, + val hasMeasurement: Boolean = false, +) + +data class P2pCacheClearResult( + val reclaimedBytes: Long, + val remainingBytes: Long, + val protectedBytes: Long, ) object P2pSettingsRepository { @@ -21,7 +49,9 @@ object P2pSettingsRepository { private var hasLoaded = false private var p2pEnabled = false private var enableUpload = true - private var hideTorrentStats = true + private var hideTorrentStats = false + private var torrentProfile = P2pTorrentProfile.BALANCED + private var cacheSize = P2pCacheSize.GB_2 fun ensureLoaded() { if (hasLoaded) return @@ -36,7 +66,9 @@ object P2pSettingsRepository { hasLoaded = false p2pEnabled = false enableUpload = true - hideTorrentStats = true + hideTorrentStats = false + torrentProfile = P2pTorrentProfile.BALANCED + cacheSize = P2pCacheSize.GB_2 publish() } @@ -64,11 +96,33 @@ object P2pSettingsRepository { publish() } + fun setTorrentProfile(profile: P2pTorrentProfile) { + ensureLoaded() + if (torrentProfile == profile) return + torrentProfile = profile + P2pSettingsStorage.saveTorrentProfile(profile.name) + publish() + } + + fun setCacheSize(size: P2pCacheSize) { + ensureLoaded() + if (cacheSize == size) return + cacheSize = size + P2pSettingsStorage.saveCacheSize(size.name) + publish() + } + private fun loadFromDisk() { hasLoaded = true p2pEnabled = P2pSettingsStorage.loadP2pEnabled() ?: false enableUpload = P2pSettingsStorage.loadEnableUpload() ?: true - hideTorrentStats = P2pSettingsStorage.loadHideTorrentStats() ?: true + hideTorrentStats = P2pSettingsStorage.loadHideTorrentStats() ?: false + torrentProfile = P2pSettingsStorage.loadTorrentProfile() + ?.let { stored -> P2pTorrentProfile.entries.firstOrNull { it.name == stored } } + ?: P2pTorrentProfile.BALANCED + cacheSize = P2pSettingsStorage.loadCacheSize() + ?.let { stored -> P2pCacheSize.entries.firstOrNull { it.name == stored } } + ?: P2pCacheSize.GB_2 publish() } @@ -77,6 +131,8 @@ object P2pSettingsRepository { p2pEnabled = p2pEnabled, enableUpload = enableUpload, hideTorrentStats = hideTorrentStats, + torrentProfile = torrentProfile, + cacheSize = cacheSize, ) } } @@ -88,6 +144,10 @@ internal expect object P2pSettingsStorage { fun saveEnableUpload(enabled: Boolean) fun loadHideTorrentStats(): Boolean? fun saveHideTorrentStats(enabled: Boolean) + fun loadTorrentProfile(): String? + fun saveTorrentProfile(profile: String) + fun loadCacheSize(): String? + fun saveCacheSize(size: String) } data class P2pStreamRequest( @@ -97,9 +157,56 @@ data class P2pStreamRequest( val trackers: List = emptyList(), ) +internal fun canonicalP2pInfoHash(infoHash: String): String { + val canonical = infoHash.trim().lowercase() + require((canonical.length == 40 || canonical.length == 64) && + canonical.all { it in '0'..'9' || it in 'a'..'f' }) { + "Torrent info hash must be 40 or 64 hexadecimal characters" + } + return canonical +} + +internal fun buildP2pMagnetUri(infoHash: String, trackers: List): String { + val canonicalHash = canonicalP2pInfoHash(infoHash) + val topic = if (canonicalHash.length == 40) { + "urn:btih:$canonicalHash" + } else { + "urn:btmh:1220$canonicalHash" + } + val trackerParameters = trackers.filter(String::isNotBlank).distinct().joinToString("") { tracker -> + "&tr=${tracker.encodeP2pQueryValue()}" + } + return "magnet:?xt=$topic$trackerParameters" +} + +private fun String.encodeP2pQueryValue(): String = buildString { + for (byte in this@encodeP2pQueryValue.encodeToByteArray()) { + val value = byte.toInt() and 0xff + if ((value in 'a'.code..'z'.code) || + (value in 'A'.code..'Z'.code) || + (value in '0'.code..'9'.code) || + value == '-'.code || value == '.'.code || value == '_'.code || value == '~'.code) { + append(value.toChar()) + } else { + append('%') + append(HEX_DIGITS[value ushr 4]) + append(HEX_DIGITS[value and 0x0f]) + } + } +} + +private const val HEX_DIGITS = "0123456789ABCDEF" + sealed class P2pStreamingState { data object Idle : P2pStreamingState() - data object Connecting : P2pStreamingState() + + data class Connecting( + val phase: String = "starting_engine", + val downloadSpeed: Long = 0L, + val uploadSpeed: Long = 0L, + val peers: Int = 0, + val seeds: Int = 0, + ) : P2pStreamingState() data class Streaming( val localUrl: String, @@ -109,7 +216,9 @@ sealed class P2pStreamingState { val seeds: Int, val bufferProgress: Float, val totalProgress: Float, - val preloadedBytes: Long = 0L, + val downloadedBytes: Long = 0L, + val verifiedBytes: Long = 0L, + val deliveredBytes: Long = 0L, ) : P2pStreamingState() data class Error(val message: String) : P2pStreamingState() @@ -119,7 +228,9 @@ class P2pStreamingException(message: String) : Exception(message) expect object P2pStreamingEngine { val state: StateFlow + val cacheState: StateFlow suspend fun startStream(request: P2pStreamRequest): String + suspend fun clearCache(): P2pCacheClearResult fun stopStream() fun shutdown() } diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerEngine.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerEngine.kt index 846c1020..f8bb4905 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerEngine.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerEngine.kt @@ -65,8 +65,11 @@ expect fun PlatformPlayerSurface( useYoutubeChunkedPlayback: Boolean = false, modifier: Modifier = Modifier, playWhenReady: Boolean = true, + initialPositionMs: Long? = null, + initialPositionRequestKey: String? = null, resizeMode: PlayerResizeMode = PlayerResizeMode.Fit, useNativeController: Boolean = false, + onInitialPositionHandled: (key: String, handled: Boolean) -> Unit = { _, _ -> }, onControllerReady: (PlayerEngineController) -> Unit, onSnapshot: (PlayerPlaybackSnapshot) -> Unit, onError: (String?) -> Unit, diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerOverlays.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerOverlays.kt index d3ae9b4b..c9ba1115 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerOverlays.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerOverlays.kt @@ -1,6 +1,5 @@ package com.nuvio.app.features.player -import androidx.compose.animation.Crossfade import androidx.compose.animation.core.FastOutSlowInEasing import androidx.compose.animation.core.LinearEasing import androidx.compose.animation.core.RepeatMode @@ -247,30 +246,23 @@ internal fun OpeningOverlay( val showHorizontalProgress = progressActive && logo == null if (!message.isNullOrBlank() || showHorizontalProgress) { Spacer(modifier = Modifier.height(16.dp)) - Crossfade( - targetState = message?.takeIf { it.isNotBlank() }, - animationSpec = tween(durationMillis = 260), - label = "openingOverlayMessageCrossfade", + Box( modifier = Modifier .fillMaxWidth() .height(40.dp), - ) { loadingMessage -> - Box( - modifier = Modifier.fillMaxSize(), - contentAlignment = Alignment.Center, - ) { - if (loadingMessage != null) { - Text( - text = loadingMessage, - color = Color.White.copy(alpha = 0.72f), - textAlign = TextAlign.Center, - maxLines = 2, - style = MaterialTheme.typography.labelMedium, - modifier = Modifier - .fillMaxWidth() - .padding(horizontal = 24.dp), - ) - } + contentAlignment = Alignment.Center, + ) { + message?.takeIf { it.isNotBlank() }?.let { loadingMessage -> + Text( + text = loadingMessage, + color = Color.White.copy(alpha = 0.72f), + textAlign = TextAlign.Center, + maxLines = 2, + style = MaterialTheme.typography.labelMedium, + modifier = Modifier + .fillMaxWidth() + .padding(horizontal = 24.dp), + ) } } if (showHorizontalProgress) { diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenRuntimeEffects.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenRuntimeEffects.kt index 81e1cf1b..02df155a 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenRuntimeEffects.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenRuntimeEffects.kt @@ -99,6 +99,8 @@ internal fun PlayerScreenRuntime.BindPlayerRuntimeEffects() { return@LaunchedEffect } if (!P2pSettingsRepository.isVisible || !p2pSettingsUiState.p2pEnabled) { + p2pResolvedSourceUrl = null + P2pStreamingEngine.stopStream() return@LaunchedEffect } @@ -142,6 +144,11 @@ internal fun PlayerScreenRuntime.BindPlayerRuntimeEffects() { LaunchedEffect(p2pStreamingState, activeTorrentInfoHash) { val state = p2pStreamingState if (activeTorrentInfoHash != null && state is P2pStreamingState.Error) { + p2pResolvedSourceUrl = null + playerController = null + playerControllerSourceUrl = null + playbackSnapshot = PlayerPlaybackSnapshot() + initialLoadCompleted = true errorMessage = getString(Res.string.player_error_torrent, state.message) controlsVisible = !playerControlsLocked } diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenRuntimeState.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenRuntimeState.kt index b6f4e42e..43b4977a 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenRuntimeState.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenRuntimeState.kt @@ -64,8 +64,8 @@ internal class PlayerScreenRuntime( lateinit var hapticFeedback: HapticFeedback var playerSettingsUiState: PlayerSettingsUiState = PlayerSettingsUiState() - var p2pSettingsUiState: P2pSettingsUiState = P2pSettingsUiState() - var p2pStreamingState: P2pStreamingState = P2pStreamingState.Idle + var p2pSettingsUiState by mutableStateOf(P2pSettingsUiState()) + var p2pStreamingState by mutableStateOf(P2pStreamingState.Idle) var metaScreenSettingsUiState: MetaScreenSettingsUiState = MetaScreenSettingsUiState() var watchedUiState: WatchedUiState = WatchedUiState() var watchProgressUiState: WatchProgressUiState = WatchProgressUiState() diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenRuntimeUi.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenRuntimeUi.kt index 227a9d59..d83c22fd 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenRuntimeUi.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenRuntimeUi.kt @@ -25,6 +25,7 @@ internal fun PlayerScreenRuntime.RenderPlayerRuntimeUi() { val isEpisode = activeSeasonNumber != null && activeEpisodeNumber != null val currentGestureFeedback = liveGestureFeedback ?: gestureFeedback val isP2pPlaybackActive = activeTorrentInfoHash != null + val p2pConnecting = p2pStreamingState as? P2pStreamingState.Connecting val p2pStats = p2pStreamingState as? P2pStreamingState.Streaming val p2pPeerInfo = p2pStats?.let { stats -> org.jetbrains.compose.resources.stringResource( @@ -34,20 +35,35 @@ internal fun PlayerScreenRuntime.RenderPlayerRuntimeUi() { ) } val p2pDownloadSpeed = p2pStats?.let { formatP2pSpeed(it.downloadSpeed) } + val p2pLoadingBytes = p2pStats?.let { maxOf(it.downloadedBytes, it.deliveredBytes) } ?: 0L + val connectingPeerInfo = p2pConnecting?.let { state -> + org.jetbrains.compose.resources.stringResource( + nuvio.composeapp.generated.resources.Res.string.player_torrent_peer_info, + state.seeds, + state.peers, + ) + } val p2pInitialLoadingMessage = when { !isP2pPlaybackActive || initialLoadCompleted -> null - p2pStreamingState is P2pStreamingState.Connecting -> { - org.jetbrains.compose.resources.stringResource( - nuvio.composeapp.generated.resources.Res.string.player_torrent_connecting_peers, - ) + p2pConnecting != null -> { + if (p2pSettingsUiState.hideTorrentStats) { + p2pConnectingPhaseLabel(p2pConnecting.phase) + } else { + org.jetbrains.compose.resources.stringResource( + nuvio.composeapp.generated.resources.Res.string.player_torrent_connecting_status, + p2pConnectingPhaseLabel(p2pConnecting.phase), + connectingPeerInfo.orEmpty(), + formatP2pSpeed(p2pConnecting.downloadSpeed), + ) + } } p2pStats != null -> { if (p2pSettingsUiState.hideTorrentStats) { null } else { org.jetbrains.compose.resources.stringResource( - nuvio.composeapp.generated.resources.Res.string.player_torrent_buffered_status, - formatP2pMegabytes(p2pStats.preloadedBytes), + nuvio.composeapp.generated.resources.Res.string.player_torrent_loading_status, + formatP2pMegabytes(p2pLoadingBytes), p2pPeerInfo.orEmpty(), p2pDownloadSpeed.orEmpty(), ) @@ -57,9 +73,15 @@ internal fun PlayerScreenRuntime.RenderPlayerRuntimeUi() { nuvio.composeapp.generated.resources.Res.string.player_torrent_starting_engine, ) } + val bufferedAheadMs = (playbackSnapshot.bufferedPositionMs - playbackSnapshot.positionMs) + .coerceAtLeast(0L) val p2pInitialLoadingProgress = when { !isP2pPlaybackActive || initialLoadCompleted || p2pStats == null -> null - else -> (p2pStats.preloadedBytes.toFloat() / P2pInitialPreloadTargetBytes.toFloat()).coerceIn(0f, 1f) + else -> p2pInitialLoadingProgress( + bufferedAheadMs = bufferedAheadMs, + downloadedBytes = p2pStats.downloadedBytes, + deliveredBytes = p2pStats.deliveredBytes, + ) } val showP2pRebufferStats = isP2pPlaybackActive && initialLoadCompleted && @@ -116,6 +138,7 @@ internal fun PlayerScreenRuntime.RenderPlayerRuntimeUi() { ), ) { val playerSurfaceSourceUrl = if (isP2pPlaybackActive) p2pResolvedSourceUrl else activeSourceUrl + val initialPositionRequestKey = currentInitialPositionRequestKey() if (playerSurfaceSourceUrl != null) { PlatformPlayerSurface( sourceUrl = playerSurfaceSourceUrl, @@ -126,7 +149,14 @@ internal fun PlayerScreenRuntime.RenderPlayerRuntimeUi() { streamType = activeStreamType, modifier = Modifier.fillMaxSize(), playWhenReady = shouldPlay, + initialPositionMs = activeInitialPositionMs.takeIf { it > 0L }, + initialPositionRequestKey = initialPositionRequestKey, resizeMode = resizeMode, + onInitialPositionHandled = { key, handled -> + if (key == currentInitialPositionRequestKey()) { + initialSeekApplied = handled + } + }, onControllerReady = { controller -> playerController = controller playerControllerSourceUrl = activeSourceUrl @@ -187,6 +217,24 @@ internal fun PlayerScreenRuntime.RenderPlayerRuntimeUi() { } } +@Composable +private fun p2pConnectingPhaseLabel(phase: String): String = when (phase) { + "add_magnet" -> org.jetbrains.compose.resources.stringResource( + nuvio.composeapp.generated.resources.Res.string.player_torrent_fetching_metadata, + ) + "prepare_stream", "attach_route" -> org.jetbrains.compose.resources.stringResource( + nuvio.composeapp.generated.resources.Res.string.player_torrent_preparing_stream, + ) + else -> org.jetbrains.compose.resources.stringResource( + nuvio.composeapp.generated.resources.Res.string.player_torrent_starting_engine, + ) +} + +private fun PlayerScreenRuntime.currentInitialPositionRequestKey(): String? { + val positionMs = activeInitialPositionMs.takeIf { it > 0L } ?: return null + return "$activePlaybackIdentity:${activeVideoId.orEmpty()}:$positionMs" +} + @Composable private fun PlayerScreenRuntime.RenderPlayerControls(displayedPositionMs: Long, isEpisode: Boolean) { val isInPip = rememberIsInPictureInPicture() diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenSupport.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenSupport.kt index 83c7944a..20970103 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenSupport.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/player/PlayerScreenSupport.kt @@ -15,9 +15,39 @@ internal const val PlayerVerticalGestureTouchSlopMultiplier = 3f internal const val PlayerVerticalGestureMinHeightFraction = 0.06f internal const val PlayerVerticalGestureDominanceRatio = 1.2f internal const val PlayerSeekProgressSyncDebounceMs = 700L -internal const val P2pInitialPreloadTargetBytes = 5_242_880L +internal const val P2pInitialByteProgressMidpoint = 5_242_880L +internal const val P2pInitialBufferTargetMs = 10_000L +internal const val P2pInitialNetworkStageWeight = 0.45f +internal const val P2pInitialDeliveryStageWeight = 0.30f +internal const val P2pInitialPlayerStageStart = 0.75f +internal const val P2pInitialLoadingMaximum = 0.95f internal const val NEXT_EPISODE_HARD_TIMEOUT_MS = 120_000L +internal fun p2pInitialLoadingProgress( + bufferedAheadMs: Long, + downloadedBytes: Long, + deliveredBytes: Long, +): Float { + val networkProgress = saturatingProgress(downloadedBytes, P2pInitialByteProgressMidpoint) * + P2pInitialNetworkStageWeight + val deliveryProgress = saturatingProgress(deliveredBytes, P2pInitialByteProgressMidpoint) * + P2pInitialDeliveryStageWeight + val engineProgress = networkProgress + deliveryProgress + val playerProgress = if (bufferedAheadMs > 0L) { + P2pInitialPlayerStageStart + + (bufferedAheadMs.toFloat() / P2pInitialBufferTargetMs.toFloat()).coerceIn(0f, 1f) * + (P2pInitialLoadingMaximum - P2pInitialPlayerStageStart) + } else { + 0f + } + return maxOf(engineProgress, playerProgress).coerceIn(0f, P2pInitialLoadingMaximum) +} + +private fun saturatingProgress(value: Long, midpoint: Long): Float { + val safeValue = value.coerceAtLeast(0L).toDouble() + return (safeValue / (safeValue + midpoint.toDouble())).toFloat() +} + internal val PlayerSideGestureSystemEdgeExclusion = 72.dp internal val PlayerSliderOverlayGap = 12.dp internal val PlayerTimeRowHeight = 36.dp diff --git a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/settings/PlaybackSettingsPage.kt b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/settings/PlaybackSettingsPage.kt index 706a928e..3dcf5a16 100644 --- a/composeApp/src/commonMain/kotlin/com/nuvio/app/features/settings/PlaybackSettingsPage.kt +++ b/composeApp/src/commonMain/kotlin/com/nuvio/app/features/settings/PlaybackSettingsPage.kt @@ -75,7 +75,12 @@ import com.nuvio.app.features.player.formatPlaybackSpeedLabel import com.nuvio.app.features.player.languageLabelForCode import com.nuvio.app.features.player.toStorageHexString import com.nuvio.app.features.p2p.P2pConsentDialog +import com.nuvio.app.features.p2p.P2pCacheClearResult +import com.nuvio.app.features.p2p.P2pCacheSize import com.nuvio.app.features.p2p.P2pSettingsRepository +import com.nuvio.app.features.p2p.P2pStreamingEngine +import com.nuvio.app.features.p2p.P2pStreamingState +import com.nuvio.app.features.p2p.P2pTorrentProfile import com.nuvio.app.features.plugins.PluginsUiState import com.nuvio.app.features.plugins.PluginRepository import com.nuvio.app.features.streams.StreamAutoPlayMode @@ -144,6 +149,31 @@ private fun formatStep(value: Float): String { } } +@Composable +private fun p2pProfileLabel(profile: P2pTorrentProfile): String = when (profile) { + P2pTorrentProfile.SOFT -> stringResource(Res.string.settings_p2p_profile_soft) + P2pTorrentProfile.BALANCED -> stringResource(Res.string.settings_p2p_profile_balanced) + P2pTorrentProfile.FAST -> stringResource(Res.string.settings_p2p_profile_fast) +} + +@Composable +private fun p2pCacheSizeLabel(size: P2pCacheSize): String = when (size) { + P2pCacheSize.NONE -> stringResource(Res.string.settings_p2p_cache_none) + P2pCacheSize.GB_2 -> stringResource(Res.string.settings_p2p_cache_2_gb) + P2pCacheSize.GB_5 -> stringResource(Res.string.settings_p2p_cache_5_gb) + P2pCacheSize.GB_10 -> stringResource(Res.string.settings_p2p_cache_10_gb) +} + +private fun formatP2pCacheBytes(bytes: Long): String { + val gibibyte = 1024.0 * 1024.0 * 1024.0 + val mebibyte = 1024.0 * 1024.0 + return if (bytes >= gibibyte) { + "${kotlin.math.round(bytes / gibibyte * 10.0) / 10.0} GB" + } else { + "${kotlin.math.round(bytes / mebibyte * 10.0) / 10.0} MB" + } +} + @Composable private fun addonSubtitleStartupModeLabel(mode: AddonSubtitleStartupMode): String = when (mode) { @@ -298,12 +328,19 @@ private fun PlaybackSettingsSection( var showAutoPlayPluginSelectionDialog by remember { mutableStateOf(false) } var showAutoPlayRegexDialog by remember { mutableStateOf(false) } var showP2pConsentDialog by remember { mutableStateOf(false) } + var showP2pProfileDialog by remember { mutableStateOf(false) } + var showP2pCacheSizeDialog by remember { mutableStateOf(false) } + var p2pCacheClearResult by remember { mutableStateOf(null) } + var p2pCacheClearFailed by remember { mutableStateOf(false) } val pluginsEnabled = AppFeaturePolicy.pluginsEnabled val autoPlayPlayerSettings by PlayerSettingsRepository.uiState.collectAsStateWithLifecycle() val p2pSettings by remember { P2pSettingsRepository.ensureLoaded() P2pSettingsRepository.uiState }.collectAsStateWithLifecycle() + val p2pCacheState by P2pStreamingEngine.cacheState.collectAsStateWithLifecycle() + val p2pStreamingState by P2pStreamingEngine.state.collectAsStateWithLifecycle() + val coroutineScope = rememberCoroutineScope() val availableExternalPlayers = ExternalPlayerPlatform.availablePlayers() val selectedExternalPlayer = availableExternalPlayers.firstOrNull { it.id == autoPlayPlayerSettings.externalPlayerId @@ -641,6 +678,56 @@ private fun PlaybackSettingsSection( isTablet = isTablet, onCheckedChange = P2pSettingsRepository::setHideTorrentStats, ) + SettingsGroupDivider(isTablet = isTablet) + SettingsNavigationRow( + title = stringResource(Res.string.settings_p2p_profile_title), + description = p2pProfileLabel(p2pSettings.torrentProfile), + isTablet = isTablet, + onClick = { showP2pProfileDialog = true }, + ) + SettingsGroupDivider(isTablet = isTablet) + SettingsNavigationRow( + title = stringResource(Res.string.settings_p2p_cache_size_title), + description = p2pCacheSizeLabel(p2pSettings.cacheSize), + isTablet = isTablet, + onClick = { showP2pCacheSizeDialog = true }, + ) + SettingsGroupDivider(isTablet = isTablet) + val cacheClearAvailable = p2pStreamingState !is P2pStreamingState.Connecting && + p2pStreamingState !is P2pStreamingState.Streaming && + !p2pCacheState.isClearing + SettingsNavigationRow( + title = stringResource(Res.string.settings_p2p_clear_cache_title), + description = when { + p2pCacheState.isClearing -> + stringResource(Res.string.settings_p2p_clear_cache_clearing) + !cacheClearAvailable -> + stringResource(Res.string.settings_p2p_clear_cache_playback_active) + p2pCacheClearFailed -> + stringResource(Res.string.settings_p2p_clear_cache_failed) + p2pCacheClearResult != null -> stringResource( + Res.string.settings_p2p_clear_cache_done, + formatP2pCacheBytes(p2pCacheClearResult!!.reclaimedBytes), + ) + !p2pCacheState.hasMeasurement -> + stringResource(Res.string.settings_p2p_clear_cache_usage_pending) + else -> stringResource( + Res.string.settings_p2p_clear_cache_usage, + formatP2pCacheBytes(p2pCacheState.usedBytes), + ) + }, + enabled = cacheClearAvailable, + isTablet = isTablet, + onClick = { + p2pCacheClearResult = null + p2pCacheClearFailed = false + coroutineScope.launch { + runCatching { P2pStreamingEngine.clearCache() } + .onSuccess { p2pCacheClearResult = it } + .onFailure { p2pCacheClearFailed = true } + } + }, + ) } } } @@ -1182,6 +1269,45 @@ private fun PlaybackSettingsSection( ) } + if (showP2pProfileDialog) { + IosEnumSelectionDialog( + title = stringResource(Res.string.settings_p2p_profile_title), + options = P2pTorrentProfile.entries, + selected = p2pSettings.torrentProfile, + label = { p2pProfileLabel(it) }, + description = { profile -> + when (profile) { + P2pTorrentProfile.SOFT -> + stringResource(Res.string.settings_p2p_profile_soft_description) + P2pTorrentProfile.BALANCED -> + stringResource(Res.string.settings_p2p_profile_balanced_description) + P2pTorrentProfile.FAST -> + stringResource(Res.string.settings_p2p_profile_fast_description) + } + }, + onSelect = { profile -> + P2pSettingsRepository.setTorrentProfile(profile) + showP2pProfileDialog = false + }, + onDismiss = { showP2pProfileDialog = false }, + ) + } + + if (showP2pCacheSizeDialog) { + IosEnumSelectionDialog( + title = stringResource(Res.string.settings_p2p_cache_size_title), + options = P2pCacheSize.entries, + selected = p2pSettings.cacheSize, + label = { p2pCacheSizeLabel(it) }, + onSelect = { size -> + P2pSettingsRepository.setCacheSize(size) + p2pCacheClearResult = null + showP2pCacheSizeDialog = false + }, + onDismiss = { showP2pCacheSizeDialog = false }, + ) + } + if (showSecondaryAudioDialog) { val originalHint = stringResource(Res.string.settings_playback_option_original_hint) LanguageSelectionDialog( @@ -2143,7 +2269,7 @@ private fun IosEnumSelectionDialog( title: String, options: List, selected: T, - label: (T) -> String, + label: @Composable (T) -> String, description: @Composable (T) -> String? = { null }, onSelect: (T) -> Unit, onDismiss: () -> Unit, diff --git a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/p2p/P2pMagnetTest.kt b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/p2p/P2pMagnetTest.kt new file mode 100644 index 00000000..fba6f0d3 --- /dev/null +++ b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/p2p/P2pMagnetTest.kt @@ -0,0 +1,32 @@ +package com.nuvio.app.features.p2p + +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFailsWith +import kotlin.test.assertTrue + +class P2pMagnetTest { + @Test + fun v1HashAndTrackersAreCanonicalEncodedAndDeduplicated() { + val hash = "ABCDEF0123456789ABCDEF0123456789ABCDEF01" + val tracker = "udp://tracker.example:80/announce?key=hello world" + val magnet = buildP2pMagnetUri(hash, listOf("", tracker, " ", tracker)) + + assertTrue(magnet.startsWith("magnet:?xt=urn:btih:${hash.lowercase()}")) + assertEquals(1, magnet.windowed("&tr=".length).count { it == "&tr=" }) + assertTrue(magnet.endsWith("udp%3A%2F%2Ftracker.example%3A80%2Fannounce%3Fkey%3Dhello%20world")) + } + + @Test + fun v2HashUsesMultihashTopic() { + val hash = "a".repeat(64) + assertTrue(buildP2pMagnetUri(hash, emptyList()).contains("xt=urn:btmh:1220$hash")) + } + + @Test + fun invalidHashIsRejectedBeforeNativeWork() { + assertFailsWith { + buildP2pMagnetUri("not-a-hash", emptyList()) + } + } +} diff --git a/composeApp/src/commonTest/kotlin/com/nuvio/app/features/p2p/P2pTelemetryTest.kt b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/p2p/P2pTelemetryTest.kt new file mode 100644 index 00000000..42fe41fe --- /dev/null +++ b/composeApp/src/commonTest/kotlin/com/nuvio/app/features/p2p/P2pTelemetryTest.kt @@ -0,0 +1,39 @@ +package com.nuvio.app.features.p2p + +import com.nuvio.app.features.player.p2pInitialLoadingProgress +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertFalse + +class P2pTelemetryTest { + @Test + fun torrentStatsAreVisibleByDefault() { + assertFalse(P2pSettingsUiState().hideTorrentStats) + } + + @Test + fun connectingStateCarriesStartupTelemetry() { + val state = P2pStreamingState.Connecting( + phase = "add_magnet", + downloadSpeed = 2_000_000L, + peers = 12, + seeds = 7, + ) + + assertEquals("add_magnet", state.phase) + assertEquals(2_000_000L, state.downloadSpeed) + assertEquals(12, state.peers) + assertEquals(7, state.seeds) + } + + @Test + fun initialProgressUsesReadinessStagesWithoutCompletingDuringBuffering() { + val target = 5_242_880L + + assertEquals(0f, p2pInitialLoadingProgress(0L, 0L, 0L)) + assertEquals(0.225f, p2pInitialLoadingProgress(0L, target, 0L)) + assertEquals(0.375f, p2pInitialLoadingProgress(0L, target, target)) + assertEquals(0.85f, p2pInitialLoadingProgress(5_000L, target, target)) + assertEquals(0.95f, p2pInitialLoadingProgress(15_000L, Long.MAX_VALUE, Long.MAX_VALUE)) + } +} diff --git a/composeApp/src/iosMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt b/composeApp/src/iosAppStore/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt similarity index 72% rename from composeApp/src/iosMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt rename to composeApp/src/iosAppStore/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt index df670f29..e267bc88 100644 --- a/composeApp/src/iosMain/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt +++ b/composeApp/src/iosAppStore/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt @@ -7,6 +7,8 @@ import kotlinx.coroutines.flow.asStateFlow actual object P2pStreamingEngine { private val _state = MutableStateFlow(P2pStreamingState.Idle) actual val state: StateFlow = _state.asStateFlow() + private val _cacheState = MutableStateFlow(P2pCacheUiState()) + actual val cacheState: StateFlow = _cacheState.asStateFlow() actual suspend fun startStream(request: P2pStreamRequest): String { val message = "P2P streaming is not available on this platform" @@ -14,6 +16,9 @@ actual object P2pStreamingEngine { throw P2pStreamingException(message) } + actual suspend fun clearCache(): P2pCacheClearResult = + P2pCacheClearResult(reclaimedBytes = 0L, remainingBytes = 0L, protectedBytes = 0L) + actual fun stopStream() { _state.value = P2pStreamingState.Idle } diff --git a/composeApp/src/iosFull/kotlin/com/nuvio/app/core/build/AppFeaturePolicy.ios.kt b/composeApp/src/iosFull/kotlin/com/nuvio/app/core/build/AppFeaturePolicy.ios.kt index b86fa9f4..526130b9 100644 --- a/composeApp/src/iosFull/kotlin/com/nuvio/app/core/build/AppFeaturePolicy.ios.kt +++ b/composeApp/src/iosFull/kotlin/com/nuvio/app/core/build/AppFeaturePolicy.ios.kt @@ -5,7 +5,7 @@ actual object AppFeaturePolicy { actual val supportersContributorsPageEnabled: Boolean = true actual val accountDeletionEnabled: Boolean = false actual val personalMediaAddonCopyEnabled: Boolean = false - actual val p2pEnabled: Boolean = false + actual val p2pEnabled: Boolean = true actual val trailerPlaybackMode: TrailerPlaybackMode = TrailerPlaybackMode.IN_APP actual val heroTrailerPlaybackSupported: Boolean = false actual val inAppUpdaterEnabled: Boolean = false diff --git a/composeApp/src/iosFull/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt b/composeApp/src/iosFull/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt new file mode 100644 index 00000000..29363c63 --- /dev/null +++ b/composeApp/src/iosFull/kotlin/com/nuvio/app/features/p2p/P2pStreamingEngine.ios.kt @@ -0,0 +1,563 @@ +@file:OptIn(kotlinx.cinterop.ExperimentalForeignApi::class) + +package com.nuvio.app.features.p2p + +import cnames.structs.nuvio_engine +import com.nuvio.app.features.p2p.native.NUVIO_ENGINE_EVENT_DISK_CACHE_RECLAIMED +import com.nuvio.app.features.p2p.native.NUVIO_ENGINE_EVENT_STREAM_PREPARED +import com.nuvio.app.features.p2p.native.NUVIO_ENGINE_EVENT_STREAM_STOPPED +import com.nuvio.app.features.p2p.native.NUVIO_ENGINE_EVENT_TORRENT_ERROR +import com.nuvio.app.features.p2p.native.NUVIO_ENGINE_EVENT_TORRENT_METADATA_READY +import com.nuvio.app.features.p2p.native.NUVIO_ENGINE_STATUS_NO_EVENT +import com.nuvio.app.features.p2p.native.NUVIO_ENGINE_STATUS_OK +import com.nuvio.app.features.p2p.native.NUVIO_ENGINE_TORRENT_PROFILE_BALANCED +import com.nuvio.app.features.p2p.native.NUVIO_ENGINE_TORRENT_PROFILE_FAST +import com.nuvio.app.features.p2p.native.NUVIO_ENGINE_TORRENT_PROFILE_SOFT +import com.nuvio.app.features.p2p.native.NUVIO_ENGINE_UPLOAD_DISABLED +import com.nuvio.app.features.p2p.native.NUVIO_ENGINE_UPLOAD_UNLIMITED +import com.nuvio.app.features.p2p.native.nuvio_engine_add_torrent +import com.nuvio.app.features.p2p.native.nuvio_engine_config +import com.nuvio.app.features.p2p.native.nuvio_engine_config_init_sized +import com.nuvio.app.features.p2p.native.nuvio_engine_create +import com.nuvio.app.features.p2p.native.nuvio_engine_destroy +import com.nuvio.app.features.p2p.native.nuvio_engine_event +import com.nuvio.app.features.p2p.native.nuvio_engine_event_init_sized +import com.nuvio.app.features.p2p.native.nuvio_engine_get_stats +import com.nuvio.app.features.p2p.native.nuvio_engine_get_stream_stats +import com.nuvio.app.features.p2p.native.nuvio_engine_poll_event +import com.nuvio.app.features.p2p.native.nuvio_engine_prepare_stream +import com.nuvio.app.features.p2p.native.nuvio_engine_reclaim_disk_cache +import com.nuvio.app.features.p2p.native.nuvio_engine_stats +import com.nuvio.app.features.p2p.native.nuvio_engine_stats_init_sized +import com.nuvio.app.features.p2p.native.nuvio_engine_status_message +import com.nuvio.app.features.p2p.native.nuvio_engine_stop_stream +import com.nuvio.app.features.p2p.native.nuvio_engine_stream_request +import com.nuvio.app.features.p2p.native.nuvio_engine_stream_request_init_sized +import com.nuvio.app.features.p2p.native.nuvio_engine_stream_stats +import com.nuvio.app.features.p2p.native.nuvio_engine_stream_stats_init_sized +import com.nuvio.app.features.p2p.native.nuvio_engine_torrent_request +import com.nuvio.app.features.p2p.native.nuvio_engine_torrent_request_init_sized +import kotlinx.cinterop.CPointer +import kotlinx.cinterop.CPointerVar +import kotlinx.cinterop.ULongVar +import kotlinx.cinterop.alloc +import kotlinx.cinterop.cstr +import kotlinx.cinterop.memScoped +import kotlinx.cinterop.ptr +import kotlinx.cinterop.sizeOf +import kotlinx.cinterop.toKString +import kotlinx.cinterop.value +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.NonCancellable +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.currentCoroutineContext +import kotlinx.coroutines.delay +import kotlinx.coroutines.ensureActive +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.isActive +import kotlinx.coroutines.launch +import kotlinx.coroutines.sync.Mutex +import kotlinx.coroutines.sync.withLock +import kotlinx.coroutines.withContext +import platform.Foundation.NSFileManager +import platform.Foundation.NSHomeDirectory +import platform.Foundation.NSLog + +private const val StatsPollIntervalMs = 250L +private const val StartupStatsPollIntervalMs = 1_000L +private const val MemoryCacheCapacityBytes = 64L * 1024L * 1024L + +actual object P2pStreamingEngine { + private data class EngineConfigurationKey( + val uploadEnabled: Boolean, + val torrentProfile: P2pTorrentProfile, + val diskCacheCapacityBytes: Long, + ) + + private data class NativeEvent( + val type: UInt, + val requestId: ULong, + val torrentId: String?, + val message: String?, + val fileIndex: UInt?, + val fileSize: ULong, + val streamId: String?, + val streamUrl: String?, + ) + + private data class NativeStream( + val id: String, + val url: String, + val torrentId: String, + val fileIndex: Int, + val fileSize: Long, + ) + + private data class AggregateStats( + val peers: Int, + val seeds: Int, + val downloadSpeed: Long, + val uploadSpeed: Long, + val payloadDownloaded: Long, + val diskUsed: Long, + val diskProtected: Long, + ) + + private data class StreamStats( + val fileSize: Long, + val contiguousReady: Long, + val verified: Long, + val delivered: Long, + ) + + private val _state = MutableStateFlow(P2pStreamingState.Idle) + actual val state: StateFlow = _state.asStateFlow() + private val _cacheState = MutableStateFlow(P2pCacheUiState()) + actual val cacheState: StateFlow = _cacheState.asStateFlow() + + private val scope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + private val lifecycleMutex = Mutex() + private var engine: CPointer? = null + private var engineConfigurationKey: EngineConfigurationKey? = null + private var currentTorrentId: String? = null + private var currentStream: NativeStream? = null + private var statsJob: Job? = null + private var generation = 0L + private val knownTorrentIds = mutableSetOf() + + actual suspend fun startStream(request: P2pStreamRequest): String = lifecycleMutex.withLock { + stopLocked(shutdownEngine = false) + generation += 1 + val streamGeneration = generation + _state.value = P2pStreamingState.Connecting() + var phase = "build_magnet" + var startupStatsJob: Job? = null + var preparedStream: NativeStream? = null + try { + val magnet = buildP2pMagnetUri( + request.infoHash, + (DefaultTrackers + request.trackers).distinct(), + ) + phase = "ensure_engine" + val activeEngine = ensureEngine() + val baseline = readAggregateStats(activeEngine).payloadDownloaded + startupStatsJob = startStartupStatsPolling(activeEngine, streamGeneration) { phase } + + phase = "add_magnet" + val canonicalHash = canonicalP2pInfoHash(request.infoHash) + val torrentId = if (canonicalHash in knownTorrentIds) { + canonicalHash + } else { + addMagnet(activeEngine, magnet).also(knownTorrentIds::add) + } + currentCoroutineContext().ensureActive() + + phase = "prepare_stream" + val stream = prepareStream( + activeEngine, + torrentId, + request.fileIdx, + request.filename, + ) + preparedStream = stream + currentCoroutineContext().ensureActive() + currentTorrentId = torrentId + currentStream = stream + phase = "attach_route" + + val aggregate = readAggregateStats(activeEngine) + publishStreaming(streamGeneration, stream, aggregate, baseline, null) + startStatsPolling(activeEngine, stream, streamGeneration, baseline) + log("stream ready fileIndex=${stream.fileIndex} fileBytes=${stream.fileSize}") + stream.url + } catch (cancellation: CancellationException) { + withContext(NonCancellable) { + preparedStream?.let { runCatching { stopNativeStream(engine, it.id) } } + } + currentTorrentId = null + currentStream = null + _state.value = P2pStreamingState.Idle + throw cancellation + } catch (error: Throwable) { + preparedStream?.let { runCatching { stopNativeStream(engine, it.id) } } + currentTorrentId = null + currentStream = null + val message = error.message ?: "Unable to start torrent stream" + _state.value = P2pStreamingState.Error(message) + log("stream failed phase=$phase error=$message") + throw P2pStreamingException(message) + } finally { + startupStatsJob?.cancel() + } + } + + actual suspend fun clearCache(): P2pCacheClearResult = lifecycleMutex.withLock { + check(_state.value !is P2pStreamingState.Connecting && + _state.value !is P2pStreamingState.Streaming) { + "Torrent cache cannot be cleared during active playback" + } + _cacheState.value = _cacheState.value.copy(isClearing = true) + try { + val activeEngine = ensureEngine() + val before = readAggregateStats(activeEngine) + reclaimDiskCache(activeEngine) + val after = readAggregateStats(activeEngine) + updateCacheState(after) + P2pCacheClearResult( + reclaimedBytes = (before.diskUsed - after.diskUsed).coerceAtLeast(0L), + remainingBytes = after.diskUsed, + protectedBytes = after.diskProtected, + ) + } finally { + _cacheState.value = _cacheState.value.copy(isClearing = false) + } + } + + actual fun stopStream() { + scope.launch { + lifecycleMutex.withLock { stopLocked(shutdownEngine = false) } + } + } + + actual fun shutdown() { + scope.launch { + lifecycleMutex.withLock { stopLocked(shutdownEngine = true) } + } + } + + private suspend fun stopLocked(shutdownEngine: Boolean) { + generation += 1 + statsJob?.cancel() + statsJob = null + val stream = currentStream + currentStream = null + currentTorrentId = null + _state.value = P2pStreamingState.Idle + stream?.let { runCatching { stopNativeStream(engine, it.id) } } + if (shutdownEngine) closeEngine() + } + + private fun startStartupStatsPolling( + activeEngine: CPointer, + streamGeneration: Long, + phase: () -> String, + ): Job = scope.launch { + while (isActive && generation == streamGeneration) { + runCatching { readAggregateStats(activeEngine) }.getOrNull()?.let { stats -> + if (_state.value is P2pStreamingState.Connecting) { + _state.value = P2pStreamingState.Connecting( + phase = phase(), + downloadSpeed = stats.downloadSpeed, + uploadSpeed = stats.uploadSpeed, + peers = stats.peers, + seeds = stats.seeds, + ) + updateCacheState(stats) + } + } + delay(StartupStatsPollIntervalMs) + } + } + + private fun startStatsPolling( + activeEngine: CPointer, + stream: NativeStream, + streamGeneration: Long, + payloadBaseline: Long, + ) { + statsJob?.cancel() + statsJob = scope.launch { + while (isActive && generation == streamGeneration) { + try { + val aggregate = readAggregateStats(activeEngine) + val route = readStreamStats(activeEngine, stream.id) + updateCacheState(aggregate) + publishStreaming(streamGeneration, stream, aggregate, payloadBaseline, route) + } catch (error: Throwable) { + if (error is CancellationException) throw error + if (generation == streamGeneration) { + _state.value = P2pStreamingState.Error( + error.message ?: "Torrent stream stopped unexpectedly" + ) + } + return@launch + } + delay(StatsPollIntervalMs) + } + } + } + + private fun publishStreaming( + streamGeneration: Long, + stream: NativeStream, + aggregate: AggregateStats, + payloadBaseline: Long, + route: StreamStats?, + ) { + if (generation != streamGeneration || currentStream?.id != stream.id) return + _state.value = P2pStreamingState.Streaming( + localUrl = stream.url, + downloadSpeed = aggregate.downloadSpeed, + uploadSpeed = aggregate.uploadSpeed, + peers = aggregate.peers, + seeds = aggregate.seeds, + bufferProgress = ratio(route?.contiguousReady ?: 0L, route?.fileSize ?: stream.fileSize), + totalProgress = ratio(route?.verified ?: 0L, route?.fileSize ?: stream.fileSize), + downloadedBytes = (aggregate.payloadDownloaded - payloadBaseline).coerceAtLeast(0L), + verifiedBytes = route?.verified ?: 0L, + deliveredBytes = route?.delivered ?: 0L, + ) + } + + private suspend fun ensureEngine(): CPointer { + P2pSettingsRepository.ensureLoaded() + val settings = P2pSettingsRepository.uiState.value + val configuration = EngineConfigurationKey( + uploadEnabled = settings.enableUpload, + torrentProfile = settings.torrentProfile, + diskCacheCapacityBytes = settings.cacheSize.bytes, + ) + engine?.takeIf { engineConfigurationKey == configuration }?.let { return it } + closeEngine() + + val stateDirectory = "${NSHomeDirectory()}/Library/Application Support/NuvioEngine/state" + val cacheDirectory = "${NSHomeDirectory()}/Library/Caches/NuvioEngine/payload" + createDirectory(stateDirectory) + createDirectory(cacheDirectory) + val created = memScoped { + val config = alloc() + nuvio_engine_config_init_sized(config.ptr, sizeOf().toUInt()) + config.data_directory = stateDirectory.cstr.getPointer(this) + config.cache_directory = cacheDirectory.cstr.getPointer(this) + config.memory_cache_capacity_bytes = MemoryCacheCapacityBytes.toULong() + config.disk_cache_capacity_bytes = configuration.diskCacheCapacityBytes.toULong() + config.upload_mode = if (configuration.uploadEnabled) { + NUVIO_ENGINE_UPLOAD_UNLIMITED + } else { + NUVIO_ENGINE_UPLOAD_DISABLED + } + config.upload_limit_bytes_per_second = 0uL + config.stream_inactivity_timeout_milliseconds = 0u + config.warm_torrent_timeout_milliseconds = 60_000u + config.tls_ca_bundle_path = null + config.torrent_profile = when (configuration.torrentProfile) { + P2pTorrentProfile.SOFT -> NUVIO_ENGINE_TORRENT_PROFILE_SOFT + P2pTorrentProfile.BALANCED -> NUVIO_ENGINE_TORRENT_PROFILE_BALANCED + P2pTorrentProfile.FAST -> NUVIO_ENGINE_TORRENT_PROFILE_FAST + } + val output = alloc>() + checkStatus(nuvio_engine_create(config.ptr, output.ptr)) + output.value ?: throw P2pStreamingException("Nuvio Engine returned an empty handle") + } + engine = created + engineConfigurationKey = configuration + log("engine created configuration=$configuration") + return created + } + + private fun closeEngine() { + val activeEngine = engine ?: return + engine = null + engineConfigurationKey = null + knownTorrentIds.clear() + nuvio_engine_destroy(activeEngine) + } + + private suspend fun addMagnet(activeEngine: CPointer, magnet: String): String { + val requestId = memScoped { + val request = alloc() + nuvio_engine_torrent_request_init_sized( + request.ptr, + sizeOf().toUInt(), + ) + request.magnet_uri = magnet.cstr.getPointer(this) + request.source_type = 0u + val output = alloc() + checkStatus(nuvio_engine_add_torrent(activeEngine, request.ptr, output.ptr)) + output.value + } + return awaitEvent(activeEngine, requestId, NUVIO_ENGINE_EVENT_TORRENT_METADATA_READY) + .torrentId + ?: throw P2pStreamingException("Torrent metadata event omitted its identifier") + } + + private suspend fun prepareStream( + activeEngine: CPointer, + torrentId: String, + fileIndex: Int?, + filename: String?, + ): NativeStream { + val requestId = memScoped { + val request = alloc() + nuvio_engine_stream_request_init_sized( + request.ptr, + sizeOf().toUInt(), + ) + request.torrent_id = torrentId.cstr.getPointer(this) + request.file_index = fileIndex?.toUInt() ?: UInt.MAX_VALUE + request.filename_hint = filename?.cstr?.getPointer(this) + val output = alloc() + checkStatus(nuvio_engine_prepare_stream(activeEngine, request.ptr, output.ptr)) + output.value + } + val event = awaitEvent(activeEngine, requestId, NUVIO_ENGINE_EVENT_STREAM_PREPARED) + return NativeStream( + id = event.streamId + ?: throw P2pStreamingException("Stream event omitted its identifier"), + url = event.streamUrl + ?: throw P2pStreamingException("Stream event omitted its URL"), + torrentId = event.torrentId ?: torrentId, + fileIndex = event.fileIndex?.toInt() ?: fileIndex ?: -1, + fileSize = event.fileSize.toLong(), + ) + } + + private suspend fun stopNativeStream(activeEngine: CPointer?, streamId: String) { + if (activeEngine == null) return + val requestId = memScoped { + val output = alloc() + checkStatus(nuvio_engine_stop_stream(activeEngine, streamId, output.ptr)) + output.value + } + awaitEvent(activeEngine, requestId, NUVIO_ENGINE_EVENT_STREAM_STOPPED) + } + + private suspend fun reclaimDiskCache(activeEngine: CPointer) { + val requestId = memScoped { + val output = alloc() + checkStatus(nuvio_engine_reclaim_disk_cache(activeEngine, 0uL, output.ptr)) + output.value + } + awaitEvent(activeEngine, requestId, NUVIO_ENGINE_EVENT_DISK_CACHE_RECLAIMED) + } + + private suspend fun awaitEvent( + activeEngine: CPointer, + requestId: ULong, + expectedType: UInt, + ): NativeEvent { + while (true) { + currentCoroutineContext().ensureActive() + when (val event = pollEvent(activeEngine)) { + null -> delay(20L) + else -> { + if (event.requestId == requestId && event.type == NUVIO_ENGINE_EVENT_TORRENT_ERROR) { + throw P2pStreamingException(event.message ?: "Torrent engine command failed") + } + if (event.requestId == requestId && event.type == expectedType) return event + } + } + } + } + + private fun pollEvent(activeEngine: CPointer): NativeEvent? = memScoped { + val event = alloc() + nuvio_engine_event_init_sized(event.ptr, sizeOf().toUInt()) + val status = nuvio_engine_poll_event(activeEngine, event.ptr) + if (status == NUVIO_ENGINE_STATUS_NO_EVENT) return@memScoped null + checkStatus(status) + NativeEvent( + type = event.type, + requestId = event.request_id, + torrentId = event.torrent_id.toKString().ifBlank { null }, + message = event.message.toKString().ifBlank { null }, + fileIndex = event.file_index.takeUnless { it == UInt.MAX_VALUE }, + fileSize = event.file_size, + streamId = event.stream_id.toKString().ifBlank { null }, + streamUrl = event.stream_url.toKString().ifBlank { null }, + ) + } + + private fun readAggregateStats(activeEngine: CPointer): AggregateStats = memScoped { + val stats = alloc() + nuvio_engine_stats_init_sized(stats.ptr, sizeOf().toUInt()) + checkStatus(nuvio_engine_get_stats(activeEngine, stats.ptr)) + AggregateStats( + peers = stats.connected_peers.toInt(), + seeds = stats.connected_seeds.toInt(), + downloadSpeed = stats.download_rate_bytes_per_second.toLong(), + uploadSpeed = stats.upload_rate_bytes_per_second.toLong(), + payloadDownloaded = stats.total_payload_download_bytes.toLong(), + diskUsed = stats.disk_cache_used_bytes.toLong(), + diskProtected = stats.disk_cache_protected_bytes.toLong(), + ) + } + + private fun readStreamStats( + activeEngine: CPointer, + streamId: String, + ): StreamStats = memScoped { + val stats = alloc() + nuvio_engine_stream_stats_init_sized( + stats.ptr, + sizeOf().toUInt(), + ) + checkStatus(nuvio_engine_get_stream_stats(activeEngine, streamId, stats.ptr)) + StreamStats( + fileSize = stats.file_size.toLong(), + contiguousReady = stats.contiguous_ready_bytes.toLong(), + verified = stats.verified_file_bytes.toLong(), + delivered = stats.delivered_bytes.toLong(), + ) + } + + private fun updateCacheState(stats: AggregateStats) { + _cacheState.value = _cacheState.value.copy( + usedBytes = stats.diskUsed, + protectedBytes = stats.diskProtected, + hasMeasurement = true, + ) + } + + private fun checkStatus(status: UInt) { + if (status == NUVIO_ENGINE_STATUS_OK) return + val message = nuvio_engine_status_message(status)?.toKString() + ?: "Nuvio Engine status $status" + throw P2pStreamingException(message) + } + + private fun createDirectory(path: String) { + val manager = NSFileManager.defaultManager + check(manager.createDirectoryAtPath( + path, + withIntermediateDirectories = true, + attributes = null, + error = null, + )) { "Could not create Nuvio Engine directory" } + } + + private fun ratio(value: Long, total: Long): Float = + if (total <= 0L) 0f else (value.toDouble() / total.toDouble()).toFloat().coerceIn(0f, 1f) + + private fun log(message: String) { + NSLog("NuvioP2PDiag: $message") + } + + private val DefaultTrackers = listOf( + "udp://zer0day.ch:1337/announce", + "udp://tracker.publictracker.xyz:6969/announce", + "udp://tracker.opentrackr.org:1337/announce", + "udp://open.demonii.com:1337/announce", + "udp://open.stealth.si:80/announce", + "http://tracker.renfei.net:8080/announce", + "udp://udp.tracker.projectk.org:23333/announce", + "udp://tracker.tryhackx.org:6969/announce", + "udp://tracker.torrent.eu.org:451/announce", + "udp://tracker.theoks.net:6969/announce", + "udp://tracker.startwork.cv:1337/announce", + "udp://tracker.qu.ax:6969/announce", + "udp://tracker.plx.im:6969/announce", + "udp://tracker.nyaa.vc:6969/announce", + "udp://tracker.iperson.xyz:6969/announce", + "udp://tracker.gmi.gd:6969/announce", + "udp://tracker.fnix.net:6969/announce", + "udp://tracker.flatuslifir.is:6969/announce", + "udp://tracker.ducks.party:1984/announce", + "udp://tracker.bluefrog.pw:2710/announce", + ) +} diff --git a/composeApp/src/iosMain/kotlin/com/nuvio/app/features/p2p/P2pSettingsStorage.ios.kt b/composeApp/src/iosMain/kotlin/com/nuvio/app/features/p2p/P2pSettingsStorage.ios.kt index a7b9d953..ed355c14 100644 --- a/composeApp/src/iosMain/kotlin/com/nuvio/app/features/p2p/P2pSettingsStorage.ios.kt +++ b/composeApp/src/iosMain/kotlin/com/nuvio/app/features/p2p/P2pSettingsStorage.ios.kt @@ -7,6 +7,8 @@ internal actual object P2pSettingsStorage { private const val p2pEnabledKey = "p2p_enabled" private const val enableUploadKey = "enable_upload" private const val hideTorrentStatsKey = "hide_torrent_stats" + private const val torrentProfileKey = "torrent_profile" + private const val cacheSizeKey = "cache_size" actual fun loadP2pEnabled(): Boolean? = loadBoolean(p2pEnabledKey) @@ -29,6 +31,18 @@ internal actual object P2pSettingsStorage { saveBoolean(hideTorrentStatsKey, enabled) } + actual fun loadTorrentProfile(): String? = loadString(torrentProfileKey) + + actual fun saveTorrentProfile(profile: String) { + saveString(torrentProfileKey, profile) + } + + actual fun loadCacheSize(): String? = loadString(cacheSizeKey) + + actual fun saveCacheSize(size: String) { + saveString(cacheSizeKey, size) + } + private fun loadBoolean(keyBase: String): Boolean? { val defaults = NSUserDefaults.standardUserDefaults val key = ProfileScopedKey.of(keyBase) @@ -38,4 +52,11 @@ internal actual object P2pSettingsStorage { private fun saveBoolean(keyBase: String, value: Boolean) { NSUserDefaults.standardUserDefaults.setBool(value, forKey = ProfileScopedKey.of(keyBase)) } + + private fun loadString(keyBase: String): String? = + NSUserDefaults.standardUserDefaults.stringForKey(ProfileScopedKey.of(keyBase)) + + private fun saveString(keyBase: String, value: String) { + NSUserDefaults.standardUserDefaults.setObject(value, forKey = ProfileScopedKey.of(keyBase)) + } } diff --git a/composeApp/src/iosMain/kotlin/com/nuvio/app/features/player/PlayerEngine.ios.kt b/composeApp/src/iosMain/kotlin/com/nuvio/app/features/player/PlayerEngine.ios.kt index 68697a34..9c0b0abe 100644 --- a/composeApp/src/iosMain/kotlin/com/nuvio/app/features/player/PlayerEngine.ios.kt +++ b/composeApp/src/iosMain/kotlin/com/nuvio/app/features/player/PlayerEngine.ios.kt @@ -51,8 +51,11 @@ actual fun PlatformPlayerSurface( useYoutubeChunkedPlayback: Boolean, modifier: Modifier, playWhenReady: Boolean, + initialPositionMs: Long?, + initialPositionRequestKey: String?, resizeMode: PlayerResizeMode, useNativeController: Boolean, + onInitialPositionHandled: (key: String, handled: Boolean) -> Unit, onControllerReady: (PlayerEngineController) -> Unit, onSnapshot: (PlayerPlaybackSnapshot) -> Unit, onError: (String?) -> Unit, diff --git a/composeApp/src/nativeInterop/cinterop/nuvioengine.def b/composeApp/src/nativeInterop/cinterop/nuvioengine.def new file mode 100644 index 00000000..096f3957 --- /dev/null +++ b/composeApp/src/nativeInterop/cinterop/nuvioengine.def @@ -0,0 +1,3 @@ +headers = nuvio_engine/nuvio_engine.h +package = com.nuvio.app.features.p2p.native +staticLibraries = libCNuvioEngine.a diff --git a/iosApp/iosApp.xcodeproj/project.pbxproj b/iosApp/iosApp.xcodeproj/project.pbxproj index 2a2c1e59..f15b3beb 100644 --- a/iosApp/iosApp.xcodeproj/project.pbxproj +++ b/iosApp/iosApp.xcodeproj/project.pbxproj @@ -405,6 +405,11 @@ "$(inherited)", "@executable_path/Frameworks", ); + OTHER_LDFLAGS = ( + "$(inherited)", + "-framework", + SystemConfiguration, + ); PRODUCT_BUNDLE_IDENTIFIER = com.nuvio.media; SWIFT_EMIT_LOC_STRINGS = YES; SWIFT_VERSION = 5.0; @@ -435,6 +440,11 @@ "$(inherited)", "@executable_path/Frameworks", ); + OTHER_LDFLAGS = ( + "$(inherited)", + "-framework", + SystemConfiguration, + ); PRODUCT_BUNDLE_IDENTIFIER = com.nuvio.app; SWIFT_EMIT_LOC_STRINGS = YES; SWIFT_VERSION = 5.0; diff --git a/scripts/debug_logs.sh b/scripts/debug_logs.sh index 7f2014be..4c262d41 100755 --- a/scripts/debug_logs.sh +++ b/scripts/debug_logs.sh @@ -12,6 +12,7 @@ # -s, --serial ADB device serial (optional, for multi-device) # -p, --package Android package/applicationId (default: debug build) # -t, --tag Additional logcat tag filter regex (default: all app) +# --p2p Focus on Nuvio Engine startup/playback diagnostics # -c, --clear Clear logcat buffer before streaming # -h, --help Show help # ───────────────────────────────────────────────────────────────────────────── @@ -21,6 +22,7 @@ PACKAGE="${NUVIO_LOG_PACKAGE:-com.nuviodebug.com}" SERIAL="" CLEAR_BUFFER=false TAG_FILTER="" +P2P_MODE=false STREAM_PID="" USER_QUIT=false @@ -33,6 +35,7 @@ LOG_FILE="${LOG_DIR}/debug_$(date +%Y%m%d_%H%M%S).log" # ── Noise suppression ─────────────────────────────────────────────────────── NOISE_TAGS='EGL_emulation|OpenGLRenderer|eglCodecCommon|goldfish|gralloc|hwcomposer|SurfaceFlinger|chatty|ConfigStore|libEGL|MediaCodec|AudioTrack|AudioFlinger|BufferQueueProducer|GraphicBufferSource|OMXClient' +P2P_TAGS='NuvioP2PDiag|NuvioPlayerDiag|P2pStreamingEngine|NuvioPlayer' # ── ANSI colour codes ─────────────────────────────────────────────────────── RST='\033[0m' @@ -62,6 +65,7 @@ Options: -s, --serial ADB device serial (optional) -p, --package Android package/applicationId (default: com.nuviodebug.com) -t, --tag Additional grep regex to filter log tags + --p2p Focus on engine phases, route telemetry, and player startup -c, --clear Clear logcat buffer before streaming -h, --help Show this help @@ -73,6 +77,8 @@ Examples: ./scripts/debug_logs.sh ./scripts/debug_logs.sh --serial emulator-5554 ./scripts/debug_logs.sh --package com.nuvio.app + ./scripts/debug_logs.sh --clear --p2p + ./scripts/debug_logs.sh --serial emulator-5554 --clear --p2p ./scripts/debug_logs.sh --tag 'MetaDetailsRepo|SeriesContent' ./scripts/debug_logs.sh --clear --tag 'Sync|Auth' EOF @@ -84,12 +90,21 @@ while [[ $# -gt 0 ]]; do -s|--serial) SERIAL="${2:-}"; shift 2 ;; -p|--package) PACKAGE="${2:-}"; shift 2 ;; -t|--tag) TAG_FILTER="${2:-}"; shift 2 ;; + --p2p) P2P_MODE=true; shift ;; -c|--clear) CLEAR_BUFFER=true; shift ;; -h|--help) usage; exit 0 ;; *) echo "Unknown option: $1" >&2; usage; exit 1 ;; esac done +if $P2P_MODE; then + if [[ -n "$TAG_FILTER" ]]; then + TAG_FILTER="${P2P_TAGS}|${TAG_FILTER}" + else + TAG_FILTER="$P2P_TAGS" + fi +fi + # ── ADB setup ──────────────────────────────────────────────────────────────── ADB=(adb) if [[ -n "$SERIAL" ]]; then @@ -104,6 +119,9 @@ fi DEVICE_MODEL=$("${ADB[@]}" shell getprop ro.product.model 2>/dev/null | tr -d '\r' || echo "unknown") ANDROID_VER=$("${ADB[@]}" shell getprop ro.build.version.release 2>/dev/null | tr -d '\r' || echo "?") +ANDROID_SDK=$("${ADB[@]}" shell getprop ro.build.version.sdk 2>/dev/null | tr -d '\r' || echo "?") +DEVICE_ABI=$("${ADB[@]}" shell getprop ro.product.cpu.abi 2>/dev/null | tr -d '\r' || echo "unknown") +DEVICE_SERIAL=$("${ADB[@]}" get-serialno 2>/dev/null | tr -d '\r' || echo "unknown") if $CLEAR_BUFFER; then "${ADB[@]}" logcat -c @@ -129,6 +147,22 @@ if [[ -z "$APP_UID" ]]; then APP_UID="$(uid_from_dumpsys || true)" fi +APP_VERSION=$("${ADB[@]}" shell dumpsys package "$PACKAGE" 2>/dev/null \ + | tr -d '\r' \ + | sed -nE 's/.*versionName=([^[:space:]]+).*/\1/p' \ + | head -n1) +APP_VERSION="${APP_VERSION:-unknown}" + +write_session_metadata() { + { + echo "# Nuvio Android diagnostic capture" + echo "# started=$(date '+%Y-%m-%dT%H:%M:%S%z')" + echo "# serial=${DEVICE_SERIAL} model=${DEVICE_MODEL} android=${ANDROID_VER} sdk=${ANDROID_SDK} abi=${DEVICE_ABI}" + echo "# package=${PACKAGE} version=${APP_VERSION} uid=${APP_UID:-unknown}" + echo "# p2pMode=${P2P_MODE} filter=${TAG_FILTER:-all-app-logs}" + } >> "$LOG_FILE" +} + # ── Colour-coding function ────────────────────────────────────────────────── colorize_line() { local line="$1" @@ -219,7 +253,9 @@ print_banner() { echo -e "${CLR_HEADER}${BOLD} ╚══════════════════════════════════════════════════════╝${RST}" echo -e "" echo -e " ${CLR_META}Device:${RST} ${CLR_ACCENT}${DEVICE_MODEL}${RST} ${CLR_META}(Android ${ANDROID_VER})${RST}" + echo -e " ${CLR_META}ABI / SDK:${RST} ${CLR_ACCENT}${DEVICE_ABI} / ${ANDROID_SDK}${RST}" echo -e " ${CLR_META}Package:${RST} ${CLR_ACCENT}${PACKAGE}${RST}" + echo -e " ${CLR_META}App version:${RST} ${CLR_ACCENT}${APP_VERSION}${RST}" echo -e " ${CLR_META}Log file:${RST} ${CLR_ACCENT}${LOG_FILE}${RST}" if [[ -n "$TAG_FILTER" ]]; then echo -e " ${CLR_META}Tag filter:${RST} ${CLR_ACCENT}${TAG_FILTER}${RST}" @@ -256,6 +292,7 @@ run_with_hotkeys() { c|C) "${ADB[@]}" logcat -c >/dev/null 2>&1 || true : > "$LOG_FILE" + write_session_metadata clear_terminal ;; q|Q) @@ -280,6 +317,7 @@ cleanup() { trap cleanup EXIT INT TERM # ── Main ───────────────────────────────────────────────────────────────────── +write_session_metadata print_banner if [[ -n "$APP_UID" ]]; then diff --git a/settings.gradle.kts b/settings.gradle.kts index 884dc8e9..9335014e 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -25,6 +25,11 @@ dependencyResolutionManagement { } } mavenCentral() + maven { + name = "localNuvioEngine" + url = uri("../nuvio-engine/platform/android/engine/build/repository") + content { includeGroup("com.nuvio") } + } } }